Add reviewed one-shot task execution journal primitive

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

scripts/task_execution_journal.py

@@ -0,0 +1,234 @@
+"""Source-only admission primitive; never authorizes or dispatches native work.
+
+Trusted owner mediation must validate claim/PDP/consume before transitions.
+One approval admits one operation. A persisted reservation cannot be resumed.
+The process lock is held across the whole owner invocation, including cleanup.
+"""
+from __future__ import annotations
+
+import fcntl
+import os
+import re
+import sqlite3
+import stat
+import uuid
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class Refused(Exception):
+    pass
+
+
+@dataclass(frozen=True)
+class Binding:
+    approval_id: str
+    request_digest: str
+    operation_id: str
+    plan_digest: str
+
+    def validate(self):
+        if any(type(v) is not str for v in (self.approval_id, self.request_digest, self.operation_id, self.plan_digest)):
+            raise Refused("binding fields must be strings")
+        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", self.approval_id):
+            raise Refused("invalid approval identifier")
+        for digest in (self.request_digest, self.plan_digest):
+            if not re.fullmatch(r"[0-9a-f]{64}", digest):
+                raise Refused("invalid digest")
+        try:
+            if str(uuid.UUID(self.operation_id)) != self.operation_id:
+                raise ValueError()
+        except (ValueError, AttributeError):
+            raise Refused("invalid operation identifier") from None
+
+
+_NEXT = {
+    "reserved": {"consume-confirmed", "failed", "unknown"},
+    "consume-confirmed": {"authenticating", "failed", "unknown"},
+    "authenticating": {"grant-issuing", "failed", "unknown"},
+    "grant-issuing": {"grant-issued", "failed", "unknown"},
+    "grant-issued": {"applying", "failed", "unknown"},
+    "applying": {"readback", "failed", "unknown"},
+    "readback": {"verified", "failed", "unknown"},
+    "verified": {"cleanup-confirmed", "unknown"},
+}
+
+
+class Journal:
+    def __init__(self, directory: Path):
+        # Require an existing owner-private directory: deployment creates it.
+        if directory.is_symlink() or directory.resolve() != directory.absolute():
+            raise Refused("journal path is indirect")
+        self.directory = directory
+        self._private(directory.stat(), directory=True)
+        self._lock_fd = self._open("execution.lock")
+        try:
+            fd = self._open("journal.sqlite")
+        except BaseException:
+            os.close(self._lock_fd)
+            del self._lock_fd
+            raise
+        original = os.fstat(fd)
+        os.close(fd)
+        self.db = sqlite3.connect(directory / "journal.sqlite", isolation_level=None)
+        current = (directory / "journal.sqlite").stat()
+        if (original.st_dev, original.st_ino) != (current.st_dev, current.st_ino):
+            self.close()
+            raise Refused("journal changed during open")
+        self.db.execute("PRAGMA synchronous=FULL")
+        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._held = False
+        self._admission = None
+        self._sync_directory()
+
+    @staticmethod
+    def _private(s, directory=False):
+        kind = stat.S_ISDIR(s.st_mode) if directory else stat.S_ISREG(s.st_mode)
+        if not kind or s.st_uid != os.getuid() or s.st_mode & 0o077:
+            raise Refused("journal is not owner private")
+        if not directory and s.st_nlink != 1:
+            raise Refused("journal file has multiple links")
+
+    def _open(self, name):
+        fd = os.open(self.directory / name, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+        try:
+            self._private(os.fstat(fd))
+        except BaseException:
+            os.close(fd)
+            raise
+        return fd
+
+    def _sync_directory(self):
+        fd = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+
+    @contextmanager
+    def locked(self):
+        if self._held:
+            raise Refused("nested journal admission")
+        try:
+            fcntl.flock(self._lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            raise Refused("owner invocation already active") from None
+        self._held = True
+        try:
+            yield self
+        finally:
+            self._admission = None
+            self._held = False
+            fcntl.flock(self._lock_fd, fcntl.LOCK_UN)
+
+    def reserve(self, binding: Binding):
+        binding.validate()
+        if not self._held or self._admission is not None:
+            raise Refused("reservation requires fresh exclusive invocation")
+        try:
+            self.db.execute("BEGIN IMMEDIATE")
+            self.db.execute("INSERT INTO operations VALUES (?,?,?,?,?,NULL)",
+                            (binding.approval_id, binding.request_digest,
+                             binding.operation_id, binding.plan_digest, "reserved"))
+            self.db.execute("COMMIT")
+            self._sync_directory()
+        except sqlite3.IntegrityError:
+            self.db.execute("ROLLBACK")
+            raise Refused("approval or operation already admitted") from None
+        except BaseException:
+            if self.db.in_transaction:
+                self.db.execute("ROLLBACK")
+            raise
+        self._admission = binding
+
+    def transition(self, binding: Binding, expected: str, target: str, evidence_digest=None):
+        if not self._held or self._admission != binding:
+            raise Refused("no live admission")
+        if target not in _NEXT.get(expected, set()):
+            raise Refused("invalid state transition")
+        if target == "cleanup-confirmed" and evidence_digest is None:
+            raise Refused("cleanup requires evidence")
+        if evidence_digest is not None and not re.fullmatch(r"[0-9a-f]{64}", evidence_digest):
+            raise Refused("invalid evidence digest")
+        try:
+            self.db.execute("BEGIN IMMEDIATE")
+            pending=self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'",(binding.operation_id,)).fetchone()[0]
+            if pending and target != 'unknown':
+                raise Refused("unconfirmed external call requires unknown")
+            changed = self.db.execute(
+                "UPDATE operations SET state=?, evidence=? WHERE approval=? AND request=? AND operation=? AND plan=? AND state=?",
+                (target, evidence_digest, binding.approval_id, binding.request_digest,
+                 binding.operation_id, binding.plan_digest, expected)).rowcount
+            if changed != 1:
+                raise Refused("journal state changed")
+            self.db.execute("COMMIT")
+            self._sync_directory()
+        except BaseException:
+            if self.db.in_transaction:
+                self.db.execute("ROLLBACK")
+            raise
+
+    def observe(self, binding: Binding):
+        binding.validate()
+        row = self.db.execute("SELECT request,operation,plan,state,evidence FROM operations WHERE approval=?", (binding.approval_id,)).fetchone()
+        if row is None:
+            return None
+        if row[:3] != (binding.request_digest, binding.operation_id, binding.plan_digest):
+            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]
+        return {"state": row[3], "evidence_digest": row[4], "dispatch_allowed": False,
+                "external_outcome_unknown": bool(pending)}
+
+    def before_call(self, binding: Binding, ordinal: int, call_digest: str):
+        """Persist one exact plan-call intent before the mediator issues it.
+
+        Caller must derive ordinal/digest from the independently pinned owner plan.
+        This primitive doesn't validate method/body or decide whether a call is allowed.
+        """
+        if not self._held or self._admission != binding:
+            raise Refused("no live admission")
+        if type(ordinal) is not int or ordinal < 0 or ordinal > 63:
+            raise Refused("invalid fixed-plan ordinal")
+        if type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", call_digest):
+            raise Refused("invalid call digest")
+        try:
+            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'}:
+                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")
+            self._sync_directory()
+        except BaseException:
+            if self.db.in_transaction: self.db.execute("ROLLBACK")
+            raise
+
+    def after_call(self, binding: Binding, ordinal: int, call_digest: str, evidence_digest: str):
+        if not self._held or self._admission != binding:
+            raise Refused("no live admission")
+        if type(ordinal) is not int or ordinal < 0 or ordinal > 63:
+            raise Refused("invalid fixed-plan ordinal")
+        if type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", call_digest):
+            raise Refused("invalid call digest")
+        if type(evidence_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", evidence_digest):
+            raise Refused("invalid evidence digest")
+        try:
+            self.db.execute("BEGIN IMMEDIATE")
+            n=self.db.execute("UPDATE calls SET outcome='complete', evidence=? WHERE operation=? AND ordinal=? AND call_digest=? AND outcome='pending'",(evidence_digest,binding.operation_id,ordinal,call_digest)).rowcount
+            if n != 1: raise Refused("call 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()
+        if hasattr(self, "_lock_fd"):
+            os.close(self._lock_fd)