fix(execution): retain scoped cleanup after uncertain outcomes
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
This commit is contained in:
codex
parent
3b6e647aa0
commit
59d1441cf8
7 changed files with 462 additions and 4 deletions
|
|
@ -57,6 +57,26 @@ _NEXT = {
|
|||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Cleanup:
|
||||
"""Owner-derived exact call commitment, never a handle or authority token.
|
||||
|
||||
The owner mediator must include these commitments in its approved plan before
|
||||
reservation, then compare the actual native cleanup call immediately before
|
||||
dispatch. This journal does not authenticate that plan or native adapter.
|
||||
"""
|
||||
kind: str
|
||||
call_digest: str
|
||||
|
||||
def validate(self):
|
||||
if type(self.kind) is not str or self.kind not in {
|
||||
"task-token-revoke", "identity-token-revoke", "grant-policy-delete"
|
||||
}:
|
||||
raise Refused("unsupported cleanup obligation")
|
||||
if type(self.call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", self.call_digest):
|
||||
raise Refused("invalid cleanup call digest")
|
||||
|
||||
|
||||
class Journal:
|
||||
def __init__(self, directory: Path):
|
||||
# Require an existing owner-private directory: deployment creates it.
|
||||
|
|
@ -82,8 +102,10 @@ class Journal:
|
|||
self.db.execute("PRAGMA journal_mode=DELETE")
|
||||
self.db.execute("CREATE TABLE IF NOT EXISTS operations (approval TEXT PRIMARY KEY, request TEXT NOT NULL, operation TEXT UNIQUE NOT NULL, plan TEXT NOT NULL, state TEXT NOT NULL, evidence TEXT)")
|
||||
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._held = False
|
||||
self._admission = None
|
||||
self._consumed_admission = None
|
||||
self._sync_directory()
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -123,11 +145,20 @@ class Journal:
|
|||
yield self
|
||||
finally:
|
||||
self._admission = None
|
||||
self._consumed_admission = None
|
||||
self._held = False
|
||||
fcntl.flock(self._lock_fd, fcntl.LOCK_UN)
|
||||
|
||||
def reserve(self, binding: Binding):
|
||||
def reserve(self, binding: Binding, cleanup: tuple[Cleanup, ...] = ()):
|
||||
binding.validate()
|
||||
if type(cleanup) is not tuple or len(cleanup) > 3:
|
||||
raise Refused("closed cleanup plan required")
|
||||
for item in cleanup:
|
||||
if type(item) is not Cleanup:
|
||||
raise Refused("typed cleanup obligation required")
|
||||
item.validate()
|
||||
if len({item.kind for item in cleanup}) != len(cleanup):
|
||||
raise Refused("duplicate cleanup obligation")
|
||||
if not self._held or self._admission is not None:
|
||||
raise Refused("reservation requires fresh exclusive invocation")
|
||||
try:
|
||||
|
|
@ -135,6 +166,9 @@ class Journal:
|
|||
self.db.execute("INSERT INTO operations VALUES (?,?,?,?,?,NULL)",
|
||||
(binding.approval_id, binding.request_digest,
|
||||
binding.operation_id, binding.plan_digest, "reserved"))
|
||||
self.db.executemany("INSERT INTO cleanup VALUES (?,?,?,?,'owed',NULL)",
|
||||
[(binding.operation_id, i, item.kind, item.call_digest)
|
||||
for i, item in enumerate(cleanup)])
|
||||
self.db.execute("COMMIT")
|
||||
self._sync_directory()
|
||||
except sqlite3.IntegrityError:
|
||||
|
|
@ -158,6 +192,13 @@ class Journal:
|
|||
try:
|
||||
self.db.execute("BEGIN IMMEDIATE")
|
||||
pending=self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'",(binding.operation_id,)).fetchone()[0]
|
||||
started = self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome!='owed'", (binding.operation_id,)).fetchone()[0]
|
||||
if started and target not in {'unknown', 'cleanup-confirmed'}:
|
||||
raise Refused("cleanup has closed business execution")
|
||||
if target == 'cleanup-confirmed':
|
||||
total, owed = self.db.execute("SELECT COUNT(*),COALESCE(SUM(outcome!='confirmed'),0) FROM cleanup WHERE operation=?", (binding.operation_id,)).fetchone()
|
||||
if not total or owed:
|
||||
raise Refused("explicit cleanup obligations are unconfirmed")
|
||||
if pending and target != 'unknown':
|
||||
raise Refused("unconfirmed external call requires unknown")
|
||||
changed = self.db.execute(
|
||||
|
|
@ -168,6 +209,8 @@ class Journal:
|
|||
raise Refused("journal state changed")
|
||||
self.db.execute("COMMIT")
|
||||
self._sync_directory()
|
||||
if target == "consume-confirmed":
|
||||
self._consumed_admission = binding
|
||||
except BaseException:
|
||||
if self.db.in_transaction:
|
||||
self.db.execute("ROLLBACK")
|
||||
|
|
@ -182,6 +225,7 @@ class Journal:
|
|||
raise Refused("journal binding mismatch")
|
||||
# Readback cannot create a live admission, even for an interrupted state.
|
||||
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]
|
||||
return {"state": row[3], "evidence_digest": row[4], "dispatch_allowed": False,
|
||||
"external_outcome_unknown": bool(pending)}
|
||||
|
||||
|
|
@ -201,7 +245,8 @@ class Journal:
|
|||
self.db.execute("BEGIN IMMEDIATE")
|
||||
count,pending = self.db.execute("SELECT COUNT(*), COALESCE(SUM(outcome='pending'),0) FROM calls WHERE operation=?",(binding.operation_id,)).fetchone()
|
||||
state=self.db.execute("SELECT state FROM operations WHERE operation=?",(binding.operation_id,)).fetchone()[0]
|
||||
if ordinal != count or pending or state in {'failed','unknown','cleanup-confirmed'}:
|
||||
cleanup_started = self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome!='owed'", (binding.operation_id,)).fetchone()[0]
|
||||
if cleanup_started or ordinal != count or pending or state in {'failed','unknown','cleanup-confirmed'}:
|
||||
raise Refused("call is replayed, uncertain or terminal")
|
||||
self.db.execute("INSERT INTO calls VALUES (?,?,?,'pending',NULL)",(binding.operation_id,ordinal,call_digest))
|
||||
self.db.execute("COMMIT")
|
||||
|
|
@ -229,6 +274,61 @@ class Journal:
|
|||
if self.db.in_transaction: self.db.execute("ROLLBACK")
|
||||
raise
|
||||
|
||||
def cleanup_status(self, binding: Binding):
|
||||
# Observation never restores live execution or cleanup authority.
|
||||
if self.observe(binding) is None:
|
||||
raise Refused("operation is not admitted")
|
||||
rows = self.db.execute("SELECT ordinal,kind,call_digest,outcome,evidence FROM cleanup WHERE operation=? ORDER BY ordinal", (binding.operation_id,)).fetchall()
|
||||
return {"confirmed": bool(rows) and all(row[3] == "confirmed" for row in rows),
|
||||
"obligations": [{"ordinal": row[0], "kind": row[1], "call_digest": row[2],
|
||||
"outcome": row[3], "evidence_digest": row[4]} for row in rows]}
|
||||
|
||||
def _cleanup_admission(self, binding, ordinal, obligation):
|
||||
if not self._held or self._admission != binding or self._consumed_admission != binding:
|
||||
raise Refused("cleanup requires same live consumed admission")
|
||||
if type(ordinal) is not int or not 0 <= ordinal < 3 or type(obligation) is not Cleanup:
|
||||
raise Refused("exact cleanup obligation required")
|
||||
obligation.validate()
|
||||
|
||||
def before_cleanup(self, binding: Binding, ordinal: int, obligation: Cleanup):
|
||||
"""Commit intent for one pre-admitted cleanup call, even after unknown.
|
||||
|
||||
Never resumes an interrupted invocation. Independent owner recovery after
|
||||
a crash still requires the owner's protected recovery channel/approval.
|
||||
Cleanup has no business deadline: its authority is the original separate
|
||||
explicit obligation, and cannot authorize a read, repair or business write.
|
||||
"""
|
||||
self._cleanup_admission(binding, ordinal, obligation)
|
||||
try:
|
||||
self.db.execute("BEGIN IMMEDIATE")
|
||||
changed = self.db.execute("UPDATE cleanup SET outcome='pending' WHERE operation=? AND ordinal=? AND kind=? AND call_digest=? AND outcome='owed'",
|
||||
(binding.operation_id, ordinal, obligation.kind, obligation.call_digest)).rowcount
|
||||
if changed != 1:
|
||||
raise Refused("cleanup is unadmitted, replayed or uncertain")
|
||||
self.db.execute("COMMIT")
|
||||
self._sync_directory()
|
||||
except BaseException:
|
||||
if self.db.in_transaction: self.db.execute("ROLLBACK")
|
||||
raise
|
||||
|
||||
def after_cleanup(self, binding: Binding, ordinal: int, obligation: Cleanup,
|
||||
*, confirmed: bool, evidence_digest: str):
|
||||
self._cleanup_admission(binding, ordinal, obligation)
|
||||
if type(confirmed) is not bool or type(evidence_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", evidence_digest):
|
||||
raise Refused("closed cleanup evidence required")
|
||||
try:
|
||||
self.db.execute("BEGIN IMMEDIATE")
|
||||
changed = self.db.execute("UPDATE cleanup SET outcome=?, evidence=? WHERE operation=? AND ordinal=? AND kind=? AND call_digest=? AND outcome='pending'",
|
||||
('confirmed' if confirmed else 'unconfirmed', evidence_digest,
|
||||
binding.operation_id, ordinal, obligation.kind, obligation.call_digest)).rowcount
|
||||
if changed != 1:
|
||||
raise Refused("cleanup confirmation mismatch")
|
||||
self.db.execute("COMMIT")
|
||||
self._sync_directory()
|
||||
except BaseException:
|
||||
if self.db.in_transaction: self.db.execute("ROLLBACK")
|
||||
raise
|
||||
|
||||
def close(self):
|
||||
if hasattr(self, "db"):
|
||||
self.db.close()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue