diff --git a/libraries/operations/README.md b/libraries/operations/README.md index ad5c0341..1800b579 100644 --- a/libraries/operations/README.md +++ b/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. diff --git a/libraries/operations/src/amplifier_operations/quiescence.py b/libraries/operations/src/amplifier_operations/quiescence.py index 81b7c65f..f27e74a1 100644 --- a/libraries/operations/src/amplifier_operations/quiescence.py +++ b/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() diff --git a/libraries/operations/tests/test_admission_abort.py b/libraries/operations/tests/test_admission_abort.py new file mode 100644 index 00000000..62a80cba --- /dev/null +++ b/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()