Operations: retain original distribution admission abort receipts

libraries/operations/README.md

@@ -168,3 +168,49 @@ do not detect erased rows or a complete valid schema recreated externally.
 Main and existing sidecars must be regular files without symlinks before SQLite
 is called, so retained FIFOs cannot block startup. Callers serialize storage
 lifecycle changes; this is not an adversarial VFS or path-swap guarantee.
+
+### Optional pre-retirement distribution admission abort
+
+`DurableIntakeFence.ADMISSION_ABORT_VERSION == 1` declares
+`abort_admission(context_with_proof, owner_id=trusted_owner_id, pending=0)` and
+`admission_abort_receipt(context, owner_id=trusted_owner_id)`. Consumers authenticate
+the coordinator, retain exclusive lifetime process ownership, and count every active
+call, background effect and pending effect. These synchronous methods neither stop
+processes nor authorize a distribution update.
+
+For `distribution-update`, `acquire` journals its original acquisition or busy
+refusal atomically in the existing `releases` table under the original fence ID.
+A typed `admission-v1` record retains context, acquisition, ordinary release and
+abort proof/receipt. No new database or required table is introduced; original
+legacy release rows remain readable. An identical busy-refused fence stays refused
+after work finishes; only a separately authorized new fence can acquire. Legacy
+held/refused attempts without the new journal cannot obtain abort authority.
+
+Abort requires exactly the Host admission-abort proof fields `kind`, `verified`,
+`purpose`, `receiptId`, `commandId`, `fenceId`, `instanceId`, `dataScope`, with kind
+`distribution-admission-abort`, literal verified true, distribution-update purpose
+and all four original IDs. Ordinary running/release proof is insufficient. The
+caller must authenticate pre-retirement applicability; the mechanism checks bindings.
+A missing fence never proves not-acquired: that result needs the exact durable busy
+refusal. An acquired fence returns released only when it is held or an exact
+pre-effect admission-refused rollback is retained. Other ordinary release outcomes
+refuse this path. Active/background/pending work refuses settlement.
+
+The exact seven-field receipt contains the four original IDs, configured `ownerId`,
+`status` released or not-acquired, and a durable `receiptId`. The full proof and
+receipt are committed atomically with fence deletion. Lost replies and restart read
+the same receipt; changed proof, owner or identity refuses. Reading an old completed
+abort never clears a newer fence, and an aborted input cannot acquire again.
+Aggregate owners retain every attempted subowner identity and require conclusive
+receipts from each; this mechanism alone cannot certify an aggregate.
+
+[Admission abort tests](tests/test_admission_abort.py) cover acquisition/refusal
+commit before lost reply, restart, changed proof, legacy evidence, active work,
+pre-effect rollback and newer-fence preservation. Production authenticated Updates
+proofs, owner adapters, full composition and live adoption require separate review.
+
+Original conclusive admission evidence can settle even after a distinct newer
+fence was acquired: an exact pre-effect rollback returns `released`, and an
+original durable refusal returns `not-acquired`. The original abort receipt
+never clears or changes that newer fence. Generic release evidence remains
+insufficient for first admission-abort settlement.

libraries/operations/src/amplifier_operations/quiescence.py

@@ -7,6 +7,7 @@
 import json
 import os
 import sqlite3
+from uuid import uuid4
 from pathlib import Path
 
 _SERVICE_KEYS = ('installationId', 'dataScope', 'ownerId', 'instanceId', 'releaseDigest')
@@ -60,6 +61,7 @@ def _service_proof(proof, context, outcome):
 class DurableIntakeFence:
     # Consumers advertise service-stop only when this public contract is present.
     SERVICE_STOP_VERSION = 1
+    ADMISSION_ABORT_VERSION = 1
     def __init__(self,path):
         path = Path(path).absolute()
         # Callers already hold exclusive lifetime ownership of this partition.
@@ -131,17 +133,81 @@ def context(self,value):
                 raise ValueError('Service identity differs from acquired host context')
             context['serviceIdentity'] = service
         return context
-    def acquire(self,value,*,pending=0):
+    def acquire(self,value,*,pending=0,refusal_reason=None):
         context=self.context(value)
         if type(pending) is not int or pending<0:raise ValueError('Nonnegative active work count required')
+        if refusal_reason is not None and (not isinstance(refusal_reason,str) or not 1<=len(refusal_reason)<=200 or any(ord(c)<32 for c in refusal_reason)):raise ValueError('Bounded original refusal reason required')
         if self.fence:
             if self.fence!=context:raise ValueError('Owner intake already belongs to another exact fence')
             return {'acquired':True,'intakeClosed':True,'fenceId':context['fenceId']}
-        if self.db.execute('SELECT 1 FROM releases WHERE fence=?',(context['fenceId'],)).fetchone():raise ValueError('A released fence cannot acquire owner intake again')
-        if self.calls or self.background or pending:return {'acquired':False,'executed':False,'reason':'Owner has active admitted work'}
-        with self.db:self.db.execute('INSERT INTO fence VALUES(1,?)',(json.dumps(context),))
+        prior=self._record(context['fenceId'])
+        if prior:
+            if prior.get('kind')=='admission-v1' and prior.get('context')==context and prior.get('acquisition',{}).get('acquired') is False and not prior.get('abort'):
+                return dict(prior['acquisition'])
+            raise ValueError('A released fence cannot acquire owner intake again')
+        result={'acquired':False,'executed':False,'reason':refusal_reason or 'Owner has active admitted work'} if self.calls or self.background or pending else {'acquired':True,'intakeClosed':True,'fenceId':context['fenceId']}
+        with self.db:
+            # The existing releases table is the per-fence lifecycle journal.
+            # Legacy rows remain intact; absence is never an acquisition receipt.
+            if context['purpose']=='distribution-update':
+                self.db.execute('INSERT INTO releases VALUES(?,?)',(context['fenceId'],json.dumps({'kind':'admission-v1','context':context,'acquisition':result})))
+            if result['acquired']:self.db.execute('INSERT INTO fence VALUES(1,?)',(json.dumps(context),))
+        if not result['acquired']:return result
         self.fence=context;self._live_fence=context
-        return {'acquired':True,'intakeClosed':True,'fenceId':context['fenceId']}
+        return result
+    def _record(self,fence_id):
+        row=self.db.execute('SELECT value FROM releases WHERE fence=?',(fence_id,)).fetchone()
+        return json.loads(row[0]) if row else None
+    def admission_abort_receipt(self,value,*,owner_id):
+        context=self.context(value)
+        record=self._record(context['fenceId'])
+        if record and record.get('kind')=='admission-v1' and record.get('abort'):
+            if record['context']!=context or record['abort']['receipt']['ownerId']!=owner_id:raise ValueError('Exact retained admission abort identity required')
+            return dict(record['abort']['receipt'])
+        return None
+    def abort_admission(self,value,*,owner_id,pending=0):
+        """Settle a trusted pre-retirement distribution admission, never its effect.
+
+        Callers authenticate proof and count all active/pending work under their
+        exclusive process lock. Original acquisition/refusal must be journaled.
+        """
+        context=self.context(value)
+        if context['purpose']!='distribution-update':raise ValueError('Distribution admission abort required')
+        if not isinstance(owner_id,str) or not 1<=len(owner_id)<=200 or any(ord(c)<32 for c in owner_id):raise ValueError('Bounded trusted owner identity required')
+        if type(pending) is not int or pending<0:raise ValueError('Nonnegative active work count required')
+        if self.calls or self.background or pending:raise ValueError('Owner work is still in flight')
+        proof=value.get('proof')
+        expected={'kind','verified','purpose','receiptId','commandId','fenceId','instanceId','dataScope'}
+        if not isinstance(proof,dict) or set(proof)!=expected or proof.get('kind')!='distribution-admission-abort' or proof.get('verified') is not True or proof.get('purpose')!='distribution-update' or any(proof.get(k)!=context[k] for k in ('commandId','fenceId','instanceId','dataScope')):
+            raise ValueError('Exact distinct authenticated admission abort proof required')
+        if not isinstance(proof['receiptId'],str) or not 1<=len(proof['receiptId'])<=200 or any(ord(c)<32 for c in proof['receiptId']):raise ValueError('Bounded abort proof receipt identity required')
+        proof=json.loads(json.dumps(proof))
+        record=self._record(context['fenceId'])
+        if not record or record.get('kind')!='admission-v1' or record.get('context')!=context:raise ValueError('No exact original acquisition or refusal journal')
+        if record.get('abort'):
+            if record['abort']['proof']!=proof or record['abort']['receipt']['ownerId']!=owner_id:raise ValueError('Admission abort differs from its retained proof or owner')
+            return dict(record['abort']['receipt'])
+        acquired=record['acquisition'].get('acquired')
+        if acquired is True:
+            if self.fence!=context:
+                # A confirmed pre-effect rollback may already have removed the
+                # fence before its acknowledgement was lost. Other releases do
+                # not authenticate a pre-retirement abort.
+                prior=record.get('release')
+                if not prior or prior.get('outcome')!='unchanged' or prior.get('proof')!={'kind':'admission-refused'}:raise ValueError('Original acquired fence has no pre-effect settlement evidence')
+            status='released'
+        elif acquired is False:
+            if self.fence==context:raise ValueError('Refusal journal contradicts held intake')
+            status='not-acquired'
+        else:raise ValueError('Original acquisition remains unknown')
+        receipt={k:context[k] for k in ('commandId','fenceId','instanceId','dataScope')}
+        receipt.update(ownerId=owner_id,status=status,receiptId=str(uuid4()))
+        record['abort']={'proof':proof,'receipt':receipt}
+        with self.db:
+            self.db.execute('UPDATE releases SET value=? WHERE fence=?',(json.dumps(record),context['fenceId']))
+            if self.fence==context:self.db.execute('DELETE FROM fence WHERE id=1')
+        if self.fence==context:self.fence=None;self._live_fence=None
+        return dict(receipt)
     def release(self,value):
         context=self.context(value);outcome=value.get('outcome');proof=value.get('proof')
         if context['purpose'] == 'service-stop' and proof != {'kind':'admission-refused'} and outcome != 'unknown':
@@ -152,8 +218,8 @@ def release(self,value):
             proof={key:proof.get(key) for key in ('verified','fenceId','commandId','outcome','instanceId','dataScope','receiptId')}
         receipt={'context':context,'outcome':outcome,'proof':proof}
         if self.fence is None:
-            row=self.db.execute('SELECT value FROM releases WHERE fence=?',(context['fenceId'],)).fetchone()
-            if row and json.loads(row[0])==receipt:return {'released':True,'intakeClosed':False}
+            prior=self._record(context['fenceId'])
+            if prior and (prior.get('release') if prior.get('kind')=='admission-v1' else prior)==receipt:return {'released':True,'intakeClosed':False}
             raise ValueError('No exact retained owner release receipt')
         if self.fence!=context:raise ValueError('Owner release does not identify the current fence')
         if outcome=='unknown':
@@ -167,7 +233,12 @@ def release(self,value):
             if not isinstance(proof.get('instanceId'),str) or (proof['instanceId']==context['instanceId'])!=(outcome=='unchanged'):raise ValueError('Release running instance does not match its outcome')
         if self.calls or self.background:raise ValueError('Owner work is still in flight')
         with self.db:
-            self.db.execute('INSERT OR IGNORE INTO releases VALUES(?,?)',(context['fenceId'],json.dumps(receipt)))
+            prior=self._record(context['fenceId'])
+            if prior and prior.get('kind')=='admission-v1':
+                if prior.get('abort'):raise ValueError('Original abort requires its distinct retained proof')
+                prior['release']=receipt
+                self.db.execute('UPDATE releases SET value=? WHERE fence=?',(json.dumps(prior),context['fenceId']))
+            else:self.db.execute('INSERT OR IGNORE INTO releases VALUES(?,?)',(context['fenceId'],json.dumps(receipt)))
             self.db.execute('DELETE FROM fence WHERE id=1')
         self.fence=None;self._live_fence=None;return {'released':True,'intakeClosed':False}
     def close(self):self.db.close()

libraries/operations/tests/test_admission_abort.py

@@ -0,0 +1,90 @@
+import json
+import pytest
+from amplifier_operations.quiescence import DurableIntakeFence
+C=dict(fenceId='update-fence',commandId='update-command',purpose='distribution-update',instanceId='original-launch',dataScope='owned-data')
+def proof(**changes):return dict(kind='distribution-admission-abort',verified=True,purpose='distribution-update',receiptId='authenticated-abort',**{k:C[k] for k in ('fenceId','commandId','instanceId','dataScope')},**changes)
+def abort(g,**changes):return g.abort_admission({**C,'proof':proof(),**changes},owner_id='fixture-owner')
+def test_acquired_commit_lost_reply_restart_has_exact_abort_receipt(tmp_path):
+ p=tmp_path/'intake.sqlite';g=DurableIntakeFence(p);g.acquire(C);g.close()
+ g=DurableIntakeFence(p);r=abort(g);assert r['status']=='released' and g.fence is None
+ assert set(r)=={'commandId','fenceId','instanceId','dataScope','ownerId','status','receiptId'}
+ g.close();g=DurableIntakeFence(p)
+ assert abort(g)==r and g.admission_abort_receipt(C,owner_id='fixture-owner')==r
+ with pytest.raises(ValueError):g.acquire(C)
+ with pytest.raises(ValueError):g.abort_admission({**C,'proof':{**proof(),'receiptId':'changed'}},owner_id='fixture-owner')
+ with pytest.raises(ValueError):g.abort_admission({**C,'proof':proof()},owner_id='other')
+ g.close()
+def test_busy_refusal_is_durable_not_acquired_and_cannot_later_acquire(tmp_path):
+ p=tmp_path/'intake.sqlite';g=DurableIntakeFence(p);g.calls=1;r=g.acquire(C);g.calls=0
+ assert g.acquire(C)==r and g.fence is None;g.close();g=DurableIntakeFence(p)
+ assert g.acquire(C)==r and abort(g)['status']=='not-acquired';g.close()
+def test_missing_or_legacy_held_evidence_never_means_not_acquired(tmp_path):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite')
+ with pytest.raises(ValueError,match='original'):abort(g)
+ g.db.execute('INSERT INTO fence VALUES(1,?)',(json.dumps(C),));g.db.commit();g.close()
+ g=DurableIntakeFence(tmp_path/'intake.sqlite')
+ with pytest.raises(ValueError,match='original'):abort(g)
+ assert g.fence==C;g.close()
+@pytest.mark.parametrize('change',[{'kind':'service-lifecycle'},{'verified':1},{'purpose':'recovery'},{'fenceId':'foreign'},{'commandId':'foreign'},{'instanceId':'replacement'},{'dataScope':'foreign'},{'outcome':'unchanged'},{'receiptId':''}])
+def test_generic_changed_or_misbound_proof_refuses(tmp_path,change):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite');g.acquire(C)
+ with pytest.raises(ValueError):g.abort_admission({**C,'proof':{**proof(),**change}},owner_id='fixture-owner')
+ assert g.fence==C;g.close()
+@pytest.mark.parametrize('counter',['calls','background','pending'])
+def test_active_effects_block_without_settlement(tmp_path,counter):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite');g.acquire(C)
+ args={}
+ if counter=='pending':args['pending']=1
+ else:setattr(g,counter,1)
+ with pytest.raises(ValueError,match='in flight'):g.abort_admission({**C,'proof':proof()},owner_id='fixture-owner',**args)
+ assert g.fence==C and g.admission_abort_receipt(C,owner_id='fixture-owner') is None
+ g.close()
+def test_pre_effect_rollback_receipt_can_settle_but_running_release_cannot(tmp_path):
+ g=DurableIntakeFence(tmp_path/'before.sqlite');g.acquire(C)
+ g.release({**C,'outcome':'unchanged','proof':{'kind':'admission-refused'}})
+ assert abort(g)['status']=='released';g.close()
+ g=DurableIntakeFence(tmp_path/'after.sqlite');g.acquire(C)
+ generic={**{k:C[k] for k in ('commandId','fenceId','instanceId','dataScope')},'outcome':'unchanged','verified':True,'receiptId':'generic-release'}
+ g.release({**C,'outcome':'unchanged','proof':generic})
+ with pytest.raises(ValueError,match='pre-effect'):abort(g)
+ g.close()
+def test_original_abort_retry_never_clears_a_new_fence(tmp_path):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite');g.acquire(C);r=abort(g)
+ new={**C,'fenceId':'new-fence','commandId':'new-command'};g.acquire(new)
+ assert abort(g)==r and g.fence==new;g.close()
+def test_refusal_cannot_release_a_different_current_fence(tmp_path):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite');g.acquire(C,pending=1)
+ new={**C,'fenceId':'new-fence','commandId':'new-command'};g.acquire(new)
+ assert abort(g)['status']=='not-acquired'
+ assert g.fence==new;g.close()
+
+
+def test_original_custom_refusal_reason_is_durable_and_detached(tmp_path):
+    gate = DurableIntakeFence(tmp_path / 'intake.sqlite3')
+    try:
+        context = dict(fenceId='refusal', commandId='command', purpose='distribution-update', instanceId='launch', dataScope='scope')
+        first = gate.acquire(context, pending=1, refusal_reason='Local listeners are still serving')
+        first['reason'] = 'caller mutation'
+        assert gate.acquire(context)['reason'] == 'Local listeners are still serving'
+    finally:
+        gate.close()
+
+
+def test_first_abort_after_exact_pre_effect_rollback_keeps_newer_hold(tmp_path):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite');g.acquire(C)
+ g.release({**C,'outcome':'unchanged','proof':{'kind':'admission-refused'}})
+ new={**C,'fenceId':'new-fence','commandId':'new-command'};g.acquire(new)
+ receipt=abort(g)
+ assert receipt['status']=='released' and g.fence==new
+ g.close();g=DurableIntakeFence(tmp_path/'intake.sqlite')
+ assert g.fence==new and abort(g)==receipt
+ g.close()
+
+def test_first_abort_after_generic_release_keeps_newer_hold_unknown(tmp_path):
+ g=DurableIntakeFence(tmp_path/'intake.sqlite');g.acquire(C)
+ generic={**{k:C[k] for k in ('commandId','fenceId','instanceId','dataScope')},'outcome':'unchanged','verified':True,'receiptId':'generic'}
+ g.release({**C,'outcome':'unchanged','proof':generic})
+ new={**C,'fenceId':'new-fence','commandId':'new-command'};g.acquire(new)
+ with pytest.raises(ValueError,match='pre-effect'):abort(g)
+ assert g.fence==new and g.admission_abort_receipt(C,owner_id='fixture-owner') is None
+ g.close()