Recover exact task cleanup after executor process loss

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

scripts/task_execution_journal.py

@@ -104,6 +104,9 @@ class Journal:
         self.db.execute("CREATE TABLE IF NOT EXISTS calls (operation TEXT NOT NULL, ordinal INTEGER NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,ordinal))")
         self.db.execute("CREATE TABLE IF NOT EXISTS cleanup (operation TEXT NOT NULL, ordinal INTEGER NOT NULL, kind TEXT NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,ordinal), UNIQUE(operation,kind))")
         self.db.execute("CREATE TABLE IF NOT EXISTS cleanup_calls (operation TEXT NOT NULL, obligation INTEGER NOT NULL, step INTEGER NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,obligation,step))")
+        self.db.execute("CREATE TABLE IF NOT EXISTS recovery_attempts (attempt TEXT PRIMARY KEY, operation TEXT NOT NULL, original TEXT NOT NULL)")
+        self.db.execute("CREATE TABLE IF NOT EXISTS recovery_calls (attempt TEXT NOT NULL, obligation INTEGER NOT NULL, step INTEGER NOT NULL, digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(attempt,obligation,step))")
+        self.db.execute("CREATE TABLE IF NOT EXISTS recovery_results (attempt TEXT NOT NULL, obligation INTEGER NOT NULL, confirmed INTEGER NOT NULL, evidence TEXT NOT NULL, PRIMARY KEY(attempt,obligation))")
         self._held = False
         self._admission = None
         self._consumed_admission = None
@@ -228,6 +231,7 @@ class Journal:
         pending = self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0]
         pending += self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0]
         pending += self.db.execute("SELECT COUNT(*) FROM cleanup_calls WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0]
+        pending += self.db.execute("SELECT COUNT(*) FROM recovery_calls AS c JOIN recovery_attempts AS a ON c.attempt=a.attempt WHERE a.operation=? AND c.outcome='pending'", (binding.operation_id,)).fetchone()[0]
         return {"state": row[3], "evidence_digest": row[4], "dispatch_allowed": False,
                 "external_outcome_unknown": bool(pending)}
 

scripts/task_execution_native_fixture.py

@@ -3,7 +3,8 @@
 This adapter deliberately admits only a disposable loopback fixture. It supplies
 real native primitive coverage while production grant schema, identity/consume and
 origin/source admission are unfinished. It cannot open an estate backend. Owner
-credentials/handles stay in memory; no subprocess, redirects or ambient proxy.
+bearer credentials stay in memory; optional owner-private fixture custody persists
+accessors for independently admitted cleanup recovery; no redirects or ambient proxy.
 Python privacy is not credential isolation; production needs the owner runtime.
 """
 from __future__ import annotations
@@ -41,7 +42,7 @@ class FixtureAdmission:
 
 class NativeFixtureAdapter:
     def __init__(self, compiler: Compiler, plan: Plan, journal: Journal,
-                 binding: Binding, admission: FixtureAdmission):
+                 binding: Binding, admission: FixtureAdmission, *, recovery_custody=None):
         if type(admission) is not FixtureAdmission or type(admission.descriptor) is not bytes:
             raise Refused('independent fixture admission required')
         self.compiler,self.plan,self.journal,self.binding=compiler,plan,journal,binding
@@ -100,6 +101,12 @@ class NativeFixtureAdapter:
         if self.meta['cleanup'] != cleanup_specs(binding.operation_id): raise Refused('cleanup plan mismatch')
         self.task=None; self._mint_attempted=False; self._policy_attempted=False; self._ordinal=0; self._next_metadata=0
         self._cleanup_steps={item.kind:0 for item in self.cleanup}
+        self.recovery_custody=recovery_custody
+        if recovery_custody is not None:
+            from task_execution_recovery import FixtureRecoveryCustody
+            if type(recovery_custody) is not FixtureRecoveryCustody or journal.observe(binding) is not None:
+                raise Refused('fresh independent fixture recovery custody required')
+            recovery_custody.prepare(binding,self.cleanup,self.port,self.hcl,admission.identity_session.accessor)
 
     def _live(self):
         if not self.journal._held or self.journal._admission!=self.binding or self.journal._consumed_admission!=self.binding:
@@ -177,6 +184,8 @@ class NativeFixtureAdapter:
               'type':'service','display_name':'task-exec-'+self.binding.operation_id,'no_parent':False}
         if body!=self.meta['execution_grant']['openbao']['token_create_body']:
             raise Refused('native token body differs from bound owner projection')
+        if self.recovery_custody is not None:
+            self.recovery_custody.checkpoint(self.binding,'mint-intent')
         self._mint_attempted=True
         status,data=self._business('POST','auth/token/create',root,encoded(body));self._ok(status,data,{200})
         auth=data.get('auth',{})
@@ -185,6 +194,8 @@ class NativeFixtureAdapter:
         # Retain handles before validation so finally cleanup can revoke even a
         # malformed overbroad issue response. Never return token-bearing metadata.
         self.task=PrivateSession(auth['client_token'],auth['accessor'])
+        if self.recovery_custody is not None:
+            self.recovery_custody.checkpoint(self.binding,'issued',self.task.accessor)
         status,data=self._business('POST','auth/token/lookup',root,encoded({'token':self.task.token}));self._ok(status,data,{200})
         info=data.get('data',{})
         if (info.get('policies')!=[self.policy_name] or info.get('identity_policies',[]) or info.get('renewable') is not False

scripts/task_execution_recovery.py

@@ -0,0 +1,336 @@
+"""Owner-only cleanup recovery for the disposable native fixture.
+
+Private, separately installed custody persists accessors, never bearer tokens.
+This is a fixture implementation, not production custody/isolation admission.
+The recovery owner supplies a fresh capability and credential independently of
+public request/journal data. No business adapter or resumed admission is built.
+"""
+from __future__ import annotations
+from dataclasses import asdict, dataclass, field
+import hashlib
+import hmac
+import http.client
+import json
+import os
+from pathlib import Path
+import secrets
+import subprocess
+import time
+import uuid
+
+from task_execution_journal import Binding, Cleanup, Journal, Refused
+from task_execution_plan import cleanup_specs
+
+
+def _encoded(value):
+    return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()
+
+
+def _digest(value):
+    return hashlib.sha256(_encoded(value)).hexdigest()
+
+
+def _json_unique(pairs):
+    result={}
+    for key,value in pairs:
+        if key in result: raise Refused('duplicate private custody field')
+        result[key]=value
+    return result
+
+
+def _invalid_constant(value):
+    raise Refused('nonfinite private custody value')
+
+
+def _policy_absent(status, response):
+    # Exactly the native OpenBao empty-errors 404. Denial/malformed/error-shaped
+    # bodies never prove absence, including after a successful delete response.
+    return (status==404 and type(response) is dict and set(response)=={'errors'}
+            and type(response['errors']) is list and not response['errors'])
+
+
+def _same_policy(actual, expected):
+    # OpenBao may omit one terminal LF from HCL source. Only that equivalence
+    # is admitted: spaces, tabs, CR and additional blank lines remain meaningful.
+    if type(actual) is not str or type(expected) is not str: return False
+    strip_one=lambda value:value[:-1] if value.endswith('\n') else value
+    return strip_one(actual)==strip_one(expected)
+
+
+class FixtureRecoveryCustody:
+    """Independent owner-private persistent fixture custody; no public API.
+
+    Mode/ownership and no-follow checks protect against accidental exposure.
+    Same-UID hostile code is NOT isolated. Production must supply independently
+    enforced owner storage/executor authority; this implementation refuses estate
+    backends and is never evidence of a production grant.
+    """
+    def __init__(self, directory: Path, *, integrity_key: bytes):
+        if type(integrity_key) is not bytes or len(integrity_key)!=32:
+            raise Refused('independent owner capsule integrity key required')
+        self._integrity_key=integrity_key
+        self.directory = directory
+        if directory.is_symlink() or directory.absolute() != directory.resolve():
+            raise Refused('indirect recovery custody directory')
+        Journal._private(directory.stat(), directory=True)
+
+    def _path(self, binding):
+        if type(binding) is not Binding: raise Refused('exact recovery binding required')
+        binding.validate()
+        return self.directory / (binding.operation_id + '.private.json')
+
+    def _sync(self):
+        fd = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        try: os.fsync(fd)
+        finally: os.close(fd)
+
+    def read(self, binding):
+        try: fd = os.open(self._path(binding), os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        except OSError: raise Refused('private custody unavailable') from None
+        try:
+            before=os.fstat(fd)
+            Journal._private(before)
+            with os.fdopen(fd, 'rb', closefd=False) as stream: raw = stream.read(65537)
+            after=os.fstat(fd)
+            identity=lambda value:(value.st_dev,value.st_ino,value.st_size,value.st_mtime_ns,value.st_ctime_ns,value.st_mode,value.st_nlink)
+            if identity(before)!=identity(after) or len(raw)!=before.st_size:
+                raise Refused('private custody changed during read')
+        finally: os.close(fd)
+        if len(raw) > 65536: raise Refused('oversize private custody record')
+        try: capsule = json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant)
+        except Exception: raise Refused('invalid private custody record') from None
+        if (type(capsule) is not dict or set(capsule)!={'payload','mac'} or type(capsule['mac']) is not str
+                or len(capsule['mac'])!=64 or any(x not in '0123456789abcdef' for x in capsule['mac'])):
+            raise Refused('invalid authenticated private capsule')
+        expected_mac=hmac.new(self._integrity_key,_encoded(capsule['payload']),hashlib.sha256).hexdigest()
+        if not hmac.compare_digest(capsule['mac'],expected_mac):
+            raise Refused('private capsule integrity mismatch')
+        value=capsule['payload']
+        expected={'binding','cleanup','port','policy_hcl','identity_accessor','task_accessor','phase','revision'}
+        if type(value) is not dict or set(value)!=expected:
+            raise Refused('invalid private custody schema')
+        stored=value['binding']
+        if (type(stored) is not dict or set(stored)!=set(asdict(binding))
+                or any(type(v) is not str for v in stored.values())):
+            raise Refused('invalid private custody binding schema')
+        restored=Binding(**stored)
+        restored.validate()
+        if asdict(restored)!=asdict(binding): raise Refused('private custody binding mismatch')
+        if (type(value['port']) is not int or not 1<=value['port']<=65535
+                or type(value['policy_hcl']) is not str or not value['policy_hcl']
+                or type(value['identity_accessor']) is not str or not value['identity_accessor']
+                or type(value['phase']) is not str or value['phase'] not in {'prepared','mint-intent','issued'}
+                or type(value['revision']) is not int or value['revision']!={'prepared':0,'mint-intent':1,'issued':2}[value['phase']]
+                or (value['phase']=='issued' and (type(value['task_accessor']) is not str or not value['task_accessor']))
+                or (value['phase']!='issued' and value['task_accessor'] is not None)
+                or type(value['cleanup']) is not list or len(value['cleanup'])!=3):
+            raise Refused('invalid private custody schema')
+        for item in value['cleanup']:
+            if type(item) is not dict or set(item)!={'kind','call_digest'}:
+                raise Refused('invalid private cleanup schema')
+            Cleanup(**item).validate()
+        expected_cleanup=[{'kind':item['kind'],'call_digest':item['call_digest']}
+                          for item in cleanup_specs(binding.operation_id)]
+        if value['cleanup']!=expected_cleanup:
+            raise Refused('private cleanup differs from deterministic owner commitment')
+        return value
+
+    def _write(self, binding, value, initial=False):
+        capsule={'payload':value,'mac':hmac.new(self._integrity_key,_encoded(value),hashlib.sha256).hexdigest()}
+        raw = _encoded(capsule)
+        if len(raw) > 65536: raise Refused('oversize private custody record')
+        path = self._path(binding)
+        if initial:
+            fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        else:
+            # Existing owner-private record checked before replacement. Unique
+            # temp + fsync + replace + directory fsync never truncates old state.
+            self.read(binding)
+            tmp = self.directory / ('.pending-' + secrets.token_hex(16))
+            fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        try:
+            with os.fdopen(fd, 'wb', closefd=False) as stream:
+                stream.write(raw); stream.flush(); os.fsync(fd)
+        finally: os.close(fd)
+        if not initial: os.replace(tmp, path)
+        self._sync()
+
+    def prepare(self, binding, cleanup, port, policy_hcl, identity_accessor):
+        value = {'binding': asdict(binding), 'cleanup': [asdict(x) for x in cleanup],
+                 'port': port, 'policy_hcl': policy_hcl,
+                 'identity_accessor': identity_accessor, 'task_accessor': None,
+                 'phase': 'prepared', 'revision': 0}
+        self._write(binding, value, initial=True)
+
+    def checkpoint(self, binding, phase, task_accessor=None):
+        value = self.read(binding)
+        if (value['phase'], phase) not in {('prepared', 'mint-intent'), ('mint-intent', 'issued')}:
+            raise Refused('invalid protected recovery checkpoint')
+        if phase == 'issued' and (type(task_accessor) is not str or not task_accessor):
+            raise Refused('protected issued accessor required')
+        value.update(phase=phase, task_accessor=task_accessor, revision=value['revision'] + 1)
+        self._write(binding, value)
+
+    def revoke(self, permit):
+        if type(permit) is not FixtureRecoveryPermit or permit.custody is not self:
+            raise Refused('exact owner recovery permit required')
+        uuid.UUID(permit.attempt_id)
+        path=self.directory / (permit.attempt_id+'.revoked')
+        fd=os.open(path,os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,0o600)
+        try:os.fsync(fd)
+        finally:os.close(fd)
+        self._sync()
+
+    def admit(self, binding, *, owner_token, server_process, valid_seconds=30):
+        """Trusted recovery-owner action, not deserialization of caller flags.
+
+        Fresh admission seals the current private custody revision. The owner
+        supplies its credential and retained backend process separately. Neither
+        token nor private capsule digest is emitted to the public journal.
+        """
+        value = self.read(binding)
+        if type(valid_seconds) is not int or not 1 <= valid_seconds <= 60:
+            raise Refused('bounded recovery admission required')
+        if type(owner_token) is not str or not owner_token:
+            raise Refused('independent recovery owner credential required')
+        return FixtureRecoveryPermit(str(uuid.uuid4()), binding, self, _digest(value),
+                                     owner_token, server_process, time.monotonic()+valid_seconds)
+
+
+@dataclass(frozen=True, repr=False)
+class FixtureRecoveryPermit:
+    attempt_id: str
+    binding: Binding
+    custody: FixtureRecoveryCustody
+    private_revision_digest: str = field(repr=False)
+    owner_token: str = field(repr=False)
+    server_process: subprocess.Popen = field(repr=False)
+    deadline: float
+
+
+class _Recorder:
+    def __init__(self, journal, permit, value):
+        self.journal, self.permit = journal, permit
+        b = permit.binding
+        if journal._admission is not None or journal._consumed_admission is not None:
+            raise Refused('recovery cannot share a business admission')
+        if journal.observe(b) is None: raise Refused('original operation absent')
+        current = journal.cleanup_status(b)['obligations']
+        if [(x['kind'], x['call_digest']) for x in current] != [(x['kind'], x['call_digest']) for x in value['cleanup']]:
+            raise Refused('original cleanup commitment mismatch')
+        if all(x['outcome'] == 'confirmed' for x in current):
+            raise Refused('cleanup already confirmed')
+        self.current = current
+        # An authenticated older prepared capsule must not erase later issuance
+        # uncertainty. Without an issued accessor, any native call or state at
+        # the grant boundary makes absence unprovable. Recovery is conservative.
+        self.no_mint_proved=(journal.observe(b)['state'] in {'reserved','consume-confirmed','authenticating'}
+                            and journal.db.execute('SELECT COUNT(*) FROM calls WHERE operation=?',(b.operation_id,)).fetchone()[0]==0)
+        db = journal.db
+        try: db.execute('INSERT INTO recovery_attempts VALUES (?,?,?)', (permit.attempt_id,b.operation_id,json.dumps(current,sort_keys=True)))
+        except Exception: raise Refused('recovery admission already used') from None
+        journal._sync_directory()
+
+    def before(self, ordinal, step, method, path):
+        digest = _digest({'kind': self.current[ordinal]['kind'], 'commitment':self.current[ordinal]['call_digest'], 'method':method,'path':path,'step':step})
+        self.journal.db.execute("INSERT INTO recovery_calls VALUES (?,?,?,?,'pending',NULL)", (self.permit.attempt_id,ordinal,step,digest))
+        self.journal._sync_directory()
+        return digest
+
+    def after(self, ordinal, step, status):
+        self.journal.db.execute("UPDATE recovery_calls SET outcome='complete', evidence=? WHERE attempt=? AND obligation=? AND step=? AND outcome='pending'", (_digest({'http_status':status}),self.permit.attempt_id,ordinal,step))
+        self.journal._sync_directory()
+
+    def finish(self, ordinal, confirmed):
+        evidence = _digest({'attempt':self.permit.attempt_id,'obligation':ordinal,'confirmed':confirmed})
+        db = self.journal.db
+        try:
+            db.execute('BEGIN IMMEDIATE')
+            pending = db.execute("SELECT COUNT(*) FROM recovery_calls WHERE attempt=? AND obligation=? AND outcome='pending'",(self.permit.attempt_id,ordinal)).fetchone()[0]
+            if confirmed and pending: raise Refused('pending recovery cannot confirm')
+            db.execute('INSERT INTO recovery_results VALUES (?,?,?,?)',(self.permit.attempt_id,ordinal,int(confirmed),evidence))
+            if confirmed:
+                item=self.current[ordinal]
+                db.execute("UPDATE cleanup SET outcome='confirmed', evidence=? WHERE operation=? AND ordinal=? AND kind=? AND call_digest=?",(evidence,self.permit.binding.operation_id,ordinal,item['kind'],item['call_digest']))
+            db.execute('COMMIT');self.journal._sync_directory()
+        except BaseException:
+            if db.in_transaction: db.execute('ROLLBACK')
+            raise
+
+
+def recover_fixture_cleanup(journal: Journal, permit: FixtureRecoveryPermit):
+    """Fresh independent owner admission: fixed cleanup only, no business retry.
+
+    Each attempt preserves original pending/unconfirmed rows in recovery_attempts
+    and native pending call rows unchanged. A fresh owner permit is required for
+    another attempt after failure; never autonomously retry uncertain mutations.
+    """
+    if type(permit) is not FixtureRecoveryPermit or type(permit.custody) is not FixtureRecoveryCustody:
+        raise Refused('independent typed recovery admission required')
+    if str(uuid.UUID(permit.attempt_id)) != permit.attempt_id:
+        raise Refused('invalid recovery attempt identity')
+    value = permit.custody.read(permit.binding)
+    if _digest(value) != permit.private_revision_digest:
+        raise Refused('private custody revision changed')
+    proc=permit.server_process
+    if (type(proc) is not subprocess.Popen or proc.poll() is not None or type(proc.args) is not list
+            or '-dev' not in proc.args or '-dev-no-store-token' not in proc.args
+            or '-dev-listen-address=127.0.0.1:'+str(value['port']) not in proc.args):
+        raise Refused('same live disposable backend required')
+    def fresh():
+        if (permit.custody.directory / (permit.attempt_id+'.revoked')).exists():
+            raise Refused('recovery permit revoked')
+        if time.monotonic() >= permit.deadline: raise Refused('recovery admission expired')
+        if proc.poll() is not None: raise Refused('disposable backend lost')
+        if _digest(permit.custody.read(permit.binding)) != permit.private_revision_digest:
+            raise Refused('private custody changed during recovery')
+    fresh()
+    with journal.locked():
+        record = _Recorder(journal,permit,value)
+        for ordinal,item in enumerate(record.current):
+            if item['outcome']=='confirmed': continue
+            step=0
+            def call(method,path,body=None):
+                nonlocal step
+                fresh(); index=step; step+=1
+                record.before(ordinal,index,method,path)
+                conn=http.client.HTTPConnection('127.0.0.1',value['port'],timeout=2)
+                try:
+                    conn.request(method,'/v1/'+path,body=_encoded(body) if body is not None else None,
+                                 headers={'X-Vault-Token':permit.owner_token,'Content-Type':'application/json'})
+                    response=conn.getresponse();raw=response.read(1048577)
+                    if len(raw)>1048576: raise Refused('oversize recovery response')
+                    data=json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant) if raw else {}
+                    if type(data) is not dict: raise Refused('invalid recovery response')
+                    status=response.status
+                except Exception: raise Refused('recovery transport or response unconfirmed') from None
+                finally:conn.close()
+                record.after(ordinal,index,status)
+                return status,data
+            confirmed=False
+            try:
+                if item['kind'] in {'task-token-revoke','identity-token-revoke'}:
+                    accessor=value['task_accessor' if ordinal==0 else 'identity_accessor']
+                    if accessor is None:
+                        confirmed=ordinal==0 and value['phase']=='prepared' and record.no_mint_proved
+                    else:
+                        status,data=call('POST','auth/token/revoke-accessor',{'accessor':accessor})
+                        # Revoke is idempotent for an already absent accessor;
+                        # actual absence must still be independently read back.
+                        if status in {200,204,400}:
+                            status,data=call('POST','auth/token/lookup-accessor',{'accessor':accessor})
+                            confirmed=status==400 and data.get('errors') in (['invalid accessor'],['1 error occurred:\n\t* invalid accessor\n\n'])
+                elif item['kind']=='grant-policy-delete':
+                    path='sys/policies/acl/task-exec-'+permit.binding.operation_id
+                    status,data=call('GET',path)
+                    if _policy_absent(status,data):confirmed=True
+                    elif (status==200 and not data.get('errors') and type(data.get('data')) is dict
+                          and _same_policy(data['data'].get('policy'),value['policy_hcl'])):
+                        status,data=call('DELETE',path)
+                        if status in {200,204}:
+                            status,data=call('GET',path);confirmed=_policy_absent(status,data)
+            except Exception:
+                # No backend errors/response/handle hashes enter public evidence.
+                confirmed=False
+            record.finish(ordinal,confirmed)
+        return journal.cleanup_status(permit.binding)