"""Closed local execution kernel v1; no installed adapters or live CLI dispatch. An owner must acknowledge each descriptor and provide an independent provenance verifier. Descriptive case types remain plan-only. Nothing executes shell prose. """ from __future__ import annotations from contextlib import contextmanager, redirect_stderr, redirect_stdout from dataclasses import asdict, dataclass, replace import fcntl import hashlib import json import os from pathlib import Path import re import sqlite3 import threading import uuid from typing import Callable import yaml from .case_inputs import BrowserInputs from .case_receipt_state import ( Outcome, Pins, Receipt, Refused, StepState, begin, finish, recover, begin_rollback, finish_rollback, ) IDENTIFIER = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9_.:/@-]{0,191}\Z") MAX_SOURCE_BYTES = 262144 MAX_HISTORY = 128 OUTPUT_LOCK = threading.RLock() def bounded_source(value: bytes): if type(value) is not bytes or len(value) > MAX_SOURCE_BYTES: raise Refused("source exceeds bounded input contract") @contextmanager def contained_output(): """Discard Python and inherited OS stdout/stderr; serialize process fd changes.""" with OUTPUT_LOCK: saved = [] null = os.open(os.devnull, os.O_WRONLY) try: for target in (1, 2): saved.append((target, os.dup(target))) os.dup2(null, target) with redirect_stdout(_Discard()), redirect_stderr(_Discard()): try: yield except Exception: raise Refused("owner callback failed") from None except BaseException: raise OwnerInterrupted() from None finally: for target, original in saved: os.dup2(original, target) os.close(original) os.close(null) DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z") def digest(value: bytes) -> str: return "sha256:" + hashlib.sha256(value).hexdigest() @dataclass(frozen=True, slots=True) class Descriptor: version: str step: str owner: str adapter: str adapter_source: str actor: str lane: str target_parameter: str effect: str allowed_created: tuple[str, ...] retry_safe: bool revocation_scopes: tuple[str, ...] @classmethod def decode(cls, source: bytes) -> Descriptor: bounded_source(source) try: obj = json.loads(source) if set(obj) != set(cls.__dataclass_fields__): raise ValueError if not isinstance(obj["allowed_created"], list): raise ValueError obj["allowed_created"] = tuple(obj["allowed_created"]) if not isinstance(obj["revocation_scopes"], list): raise ValueError obj["revocation_scopes"] = tuple(obj["revocation_scopes"]) result = cls(**obj) identifiers = ( result.step, result.owner, result.adapter, result.lane, result.target_parameter, *result.allowed_created, ) if result.version != "warden.case.execution.v1" or not all( isinstance(v, str) and IDENTIFIER.fullmatch(v) for v in identifiers ): raise ValueError if not DIGEST.fullmatch(result.adapter_source) or result.actor not in ( "human", "agent", ): raise ValueError if ( result.effect not in ("read", "change", "mint") or type(result.retry_safe) is not bool ): raise ValueError scopes = set(result.revocation_scopes) if ( len(scopes) != len(result.revocation_scopes) or not scopes <= {"parent-session", "broker-grant"} or "parent-session" not in scopes ): raise ValueError if result.effect == "mint" and "broker-grant" not in scopes: raise ValueError if len(result.allowed_created) > 32: raise ValueError if len(set(result.allowed_created)) != len(result.allowed_created): raise ValueError return result except (ValueError, TypeError, KeyError): raise Refused("invalid closed execution descriptor") from None @dataclass(frozen=True, slots=True) class OwnerResult: outcome: Outcome evidence: str created: tuple[str, ...] = () revoked: bool = False session_revocation_evidence: str | None = None grant_revocation_evidence: str | None = None attempt_id: str = "" operation: str = "" @dataclass(frozen=True, slots=True) class Invocation: step: str owner: str target: str actor: str principal: str lane: str pins: Pins attempt_id: str = "" operation: str = "admission" tenant: str = "" realm: str = "" INTENT_SCHEMA = "warden.case.original-intent.v1" @dataclass(frozen=True, slots=True) class OriginalIntent: """Durable record of the original mutating dispatch, written before dispatch. Binds the original Invocation (attempt, step, owner, target, actor, principal, lane), all pins and the admitted adapter source. Current admit leaves tenant and realm empty; their reserved fields are not admitted provenance. This is local dispatch correspondence, not owner proof of creation. Owners still authenticate original created-only provenance independently. """ binding: str invocation: Invocation source: str def validate(self) -> None: """Fixed bounded metadata only; tenant/realm are not authority claims. Current admit leaves tenant/realm empty. Nonempty future coordinates need separate owner provenance; serializability alone must never supply it. """ if ( type(self.invocation) is not Invocation or type(self.invocation.pins) is not Pins ): raise Refused("exact original invocation required") i = self.invocation fields = (i.step, i.owner, i.target, i.principal, i.lane, i.attempt_id) if ( not all( type(value) is str and IDENTIFIER.fullmatch(value) for value in fields ) or type(i.actor) is not str or i.actor not in ("human", "agent") or type(i.operation) is not str or i.operation not in ("run", "rollback") or not all( type(value) is str and (not value or IDENTIFIER.fullmatch(value)) for value in (i.tenant, i.realm) ) or type(self.source) is not str or not DIGEST.fullmatch(self.source) or type(self.binding) is not str or self.binding != f"{i.step}:{i.target}" ): raise Refused("invalid original intent metadata") if set(Invocation.__dataclass_fields__) != { "step", "owner", "target", "actor", "principal", "lane", "pins", "attempt_id", "operation", "tenant", "realm", }: raise Refused("original invocation field contract changed") if not all( type(v) is str and DIGEST.fullmatch(v) for v in ( i.pins.case, i.pins.case_type, i.pins.adapter, i.pins.catalog, i.pins.acceptance, i.pins.run_sheet, ) ): raise Refused("invalid original intent pins") def payload(self) -> str: self.validate() return json.dumps( {"schema": INTENT_SCHEMA, "binding": self.binding, "source": self.source} | asdict(self.invocation), separators=(",", ":"), sort_keys=True, ) @classmethod def decode(cls, raw: str) -> OriginalIntent: def unique(pairs): result = {} for key, value in pairs: if key in result: raise Refused("duplicate original intent field") result[key] = value return result try: if type(raw) is not str or len(raw.encode()) > MAX_SOURCE_BYTES: raise Refused("original intent input exceeds bound") obj = json.loads(raw, object_pairs_hook=unique) fields = { "step", "owner", "target", "actor", "principal", "lane", "pins", "attempt_id", "operation", "tenant", "realm", } if ( type(obj) is not dict or set(obj) != fields | {"schema", "binding", "source"} or obj["schema"] != INTENT_SCHEMA ): raise Refused("unknown original intent schema") inv = {k: obj[k] for k in fields} inv["pins"] = Pins(**inv["pins"]) result = cls(obj["binding"], Invocation(**inv), obj["source"]) result.validate() return result except (TypeError, ValueError, KeyError): raise Refused("invalid original intent record") from None @dataclass(frozen=True, slots=True) class ReconcileContext: """Original intent and the new reconciliation attempt, kept separate.""" original: OriginalIntent attempt: Invocation @dataclass(frozen=True, slots=True) class Adapter: reference: str source: str invoke: Callable[[Invocation], OwnerResult] verify: Callable[[Invocation, OwnerResult], bool] rollback: Callable[[Invocation, tuple[str, ...]], OwnerResult] | None = None reconcile: Callable[[Invocation], OwnerResult] | None = None reconcile_original: Callable[[ReconcileContext], OwnerResult] | None = None class Journal: """Private SQLite journal plus process lock; transaction commits before dispatch.""" def __init__(self, directory: Path): directory.mkdir(mode=0o700, parents=True, exist_ok=True) if ( directory.is_symlink() or directory.stat().st_uid != os.getuid() or directory.stat().st_mode & 0o077 ): raise Refused("journal directory is not private") self.directory = directory for name in ("journal.sqlite", "execution.lock"): path = directory / name fd = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: if os.fstat(fd).st_uid != os.getuid() or os.fstat(fd).st_mode & 0o077: raise Refused("journal file is not private") finally: os.close(fd) self.db = sqlite3.connect(directory / "journal.sqlite", isolation_level=None) self.db.execute("PRAGMA synchronous=FULL") self.db.execute( "CREATE TABLE IF NOT EXISTS receipts (seq INTEGER PRIMARY KEY, binding TEXT NOT NULL, payload TEXT NOT NULL)" ) self.db.execute( "CREATE TABLE IF NOT EXISTS intents (seq INTEGER PRIMARY KEY, attempt_id TEXT NOT NULL UNIQUE, binding TEXT NOT NULL, payload TEXT NOT NULL)" ) def original_intent( self, binding: str, current: Invocation, source: str ) -> OriginalIntent: """Latest original run intent for binding, refused unless it matches exactly. Missing or mismatched records refuse; the step stays uncertain. """ row = self.db.execute( "SELECT attempt_id,payload FROM intents WHERE binding=? ORDER BY seq DESC LIMIT 1", (binding,), ).fetchone() if row is None: raise Refused("original intent record missing; step stays uncertain") intent = OriginalIntent.decode(row[1]) o = intent.invocation if row[0] != o.attempt_id: raise Refused("original intent attempt identity differs") if intent.binding != binding or o.operation != "run": raise Refused("original intent binding mismatch") if o.pins != current.pins: raise Refused("original intent pins mismatch") if intent.source != source: raise Refused("original intent source mismatch") for name in ( "step", "owner", "target", "actor", "principal", "lane", "tenant", "realm", ): if getattr(o, name) != getattr(current, name): raise Refused(f"original intent {name} mismatch") # Match the persisted original running receipt. An orphan context or a # payload from another attempt cannot become the current lookup key. receipts = self.db.execute( "SELECT payload FROM receipts WHERE binding=? ORDER BY seq DESC LIMIT ?", (binding, MAX_HISTORY + 1), ).fetchall() if len(receipts) > MAX_HISTORY: raise Refused("original intent receipt history exceeds bound") matched = False try: for (raw,) in receipts: if type(raw) is not str or len(raw.encode()) > MAX_SOURCE_BYTES: raise Refused("original receipt input exceeds bound") payload = json.loads(raw) if ( payload.get("attempt_id") != o.attempt_id or payload.get("operation") != "run" ): continue last = payload.get("receipts", [])[-1] if last.get("outcome") != "running": continue if ( payload.get("journal_schema") != "warden.case.step-journal.v1" or payload.get("step_id") != o.step or payload.get("owner") != o.owner or payload.get("pins") != asdict(o.pins) ): raise Refused("original running receipt binding differs") matched = True break except (TypeError, ValueError, KeyError, IndexError, AttributeError): raise Refused("invalid original running receipt") from None if not matched: raise Refused("original running receipt missing; no backfill") return intent @contextmanager def locked(self): with (self.directory / "execution.lock").open("rb") as fh: try: fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise Refused("case execution already active") from None try: yield finally: fcntl.flock(fh, fcntl.LOCK_UN) def load(self, binding: str, initial: StepState) -> StepState: row = self.db.execute( "SELECT payload FROM receipts WHERE binding=? ORDER BY seq DESC LIMIT 1", (binding,), ).fetchone() if row is None: return initial obj = json.loads(row[0]) expected = { "step_id", "owner", "pins", "allowed_created", "retry_safe", "receipts", "journal_schema", "evidence_digest", "session_revoked", "session_revocation_digest", "grant_revocation_digest", "attempt_id", "operation", } if ( set(obj) != expected or obj["journal_schema"] != "warden.case.step-journal.v1" ): raise Refused("unknown journal receipt schema") state = StepState( obj["step_id"], obj["owner"], Pins(**obj["pins"]), tuple(obj["allowed_created"]), obj["retry_safe"], tuple( Receipt(r["attempt"], Outcome(r["outcome"]), tuple(r["created"])) for r in obj["receipts"] ), ) if ( state.step_id, state.owner, state.pins, state.allowed_created, state.retry_safe, ) != ( initial.step_id, initial.owner, initial.pins, initial.allowed_created, initial.retry_safe, ): raise Refused("journal pins changed") return state def save( self, binding: str, state: StepState, evidence: OwnerResult | None = None, invocation: Invocation | None = None, intent: OriginalIntent | None = None, ): if intent is not None: if type(intent) is not OriginalIntent: raise Refused("exact original intent required") intent.validate() expected = ( Outcome.RUNNING if intent.invocation.operation == "run" else Outcome.ROLLBACK_RUNNING ) if ( intent.binding != binding or intent.invocation != invocation or intent.invocation.pins != state.pins or not state.last or state.last.outcome is not expected ): raise Refused("original intent and running receipt differ") self.db.execute("BEGIN IMMEDIATE") try: if intent is not None: self.db.execute( "INSERT INTO intents(attempt_id,binding,payload) VALUES (?,?,?)", (intent.invocation.attempt_id, binding, intent.payload()), ) self.db.execute( "INSERT INTO receipts(binding,payload) VALUES (?,?)", ( binding, json.dumps( asdict(state) | { "journal_schema": "warden.case.step-journal.v1", "attempt_id": invocation.attempt_id if invocation else None, "operation": invocation.operation if invocation else None, "evidence_digest": digest(evidence.evidence.encode()) if evidence else None, "session_revoked": evidence.revoked if evidence else None, "session_revocation_digest": digest( evidence.session_revocation_evidence.encode() ) if evidence and evidence.session_revocation_evidence else None, "grant_revocation_digest": digest( evidence.grant_revocation_evidence.encode() ) if evidence and evidence.grant_revocation_evidence else None, }, separators=(",", ":"), ), ), ) self.db.execute("COMMIT") except BaseException: self.db.execute("ROLLBACK") raise def _catalog_lane(catalog_source: bytes, descriptor: Descriptor): """KERNEL-2: the descriptor's lane must be an active, fitting catalog entry. catalog-routing-acceptance still covers the whole catalog digest; this ties the one lane the descriptor names to an entry inside that digest. """ from .routing.catalog import CatalogError, _parse_entry try: raw = yaml.safe_load(catalog_source) items = raw["entries"] if not isinstance(items, list): raise TypeError entries = {} for index, item in enumerate(items): parsed = _parse_entry(item, index) if parsed.id in entries: raise ValueError entries[parsed.id] = parsed except (yaml.YAMLError, CatalogError, KeyError, TypeError, ValueError): raise Refused("invalid catalog source shape") from None entry = entries.get(descriptor.lane) if entry is None: raise Refused("descriptor lane is not in the catalog") if not entry.is_active: raise Refused("descriptor lane is not active in the catalog") if entry.delegation is None: raise Refused("descriptor lane has no catalog delegation record") if not entry.is_graded: raise Refused("descriptor lane carries no risk grade") if descriptor.owner not in entry.owner_repo.split("/"): raise Refused("descriptor owner does not fit catalog lane owner_repo") if descriptor.effect in ("mint", "change") and entry.risk != "high": raise Refused("catalog lane risk does not cover the descriptor effect") return entry def _step_mode(case_type: dict, case: dict, descriptor: Descriptor) -> str: """KERNEL-3: warden, attended or owner, from the type step's own text.""" steps = [ step for step in case_type.get("follow_through") or [] if isinstance(step, dict) and step.get("id") == descriptor.step ] step = steps[0] if len(steps) == 1 else {} executes, attended = step.get("warden_executes"), step.get("attended") if ( type(executes) is not bool or type(attended) is not bool or (executes and attended) or not isinstance(step.get("lane"), str) ): raise Refused("type step does not declare how the kernel dispatches it") owner = step.get("owner") if isinstance(owner, str) and re.fullmatch(r"\{[a-z_]+\}", owner): owner = (case.get("parameters") or {}).get(owner[1:-1]) if step["lane"] != descriptor.lane or owner != descriptor.owner: raise Refused("type step lane or owner disagrees with descriptor") return "warden" if executes else "attended" if attended else "owner" class Executor: """Injected fixed owner adapters and provenance verifier; none ship by default. evidence verifier must authenticate owner custody/provenance or explicitly attest attended manual evidence; a bare JSON reference is never sufficient. It receives purpose, exact source digest, and named evidence reference. """ def __init__( self, journal: Journal, adapters: dict[str, Adapter], evidence_verifier: Callable[[str, str, str], bool], *, browser_inputs: BrowserInputs | None = None, ): self.journal, self.adapters, self.evidence_verifier = ( journal, dict(adapters), evidence_verifier, ) self._admitted = {} self.browser_inputs = browser_inputs def admit( self, case_source: bytes, type_source: bytes, descriptor_source: bytes, catalog_source: bytes, *, principal: str, actor: str, lane: str, acceptance_evidence: str, materialization_evidence: str, descriptor_evidence: str, session_evidence: str, ) -> tuple[Descriptor, Invocation, StepState]: for source in (case_source, type_source, descriptor_source, catalog_source): bounded_source(source) descriptor = Descriptor.decode(descriptor_source) try: case = yaml.safe_load(case_source)["spec"] case_type = yaml.safe_load(type_source)["spec"] if not isinstance(case, dict) or not isinstance(case_type, dict): raise ValueError except (yaml.YAMLError, KeyError, TypeError, ValueError): raise Refused("invalid case or type source shape") from None if case.get("status") != "materialized" or not case.get("acceptance"): raise Refused("case is not accepted and structurally materialized") if case_type.get("status") not in ("acknowledged", "implemented"): raise Refused("type is not acknowledged") if case.get("type") != f"{case_type['id']}@{case_type['version']}": raise Refused("case type pin mismatch") if case_type.get("execution_descriptors", {}).get(descriptor.step) != digest( descriptor_source ): raise Refused( "type has no exact acknowledged machine descriptor; plan only" ) for purpose, source, evidence in ( ("case-acceptance", digest(case_source), acceptance_evidence), ( "ops-mason-materialization", digest(case_source), materialization_evidence, ), ( "owner-descriptor-acceptance:" + descriptor.owner, digest(descriptor_source), descriptor_evidence, ), ("type-machine-seam-acceptance", digest(type_source), descriptor_evidence), ("catalog-routing-acceptance", digest(catalog_source), descriptor_evidence), ( f"caller-session:{actor}:{principal}:{lane}", digest(case_source), session_evidence, ), ): if ( not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or self._verify_evidence(purpose, source, evidence) is not True ): raise Refused("independent owner evidence not verified") if ( not IDENTIFIER.fullmatch(principal) or actor != descriptor.actor or lane != descriptor.lane ): raise Refused("principal actor or lane mismatch") if descriptor.effect == "mint" and ( actor != "human" or os.getenv("WARDEN_AGENT_ID") ): raise Refused("agent mint is forbidden") entry = _catalog_lane(catalog_source, descriptor) mode = _step_mode(case_type, case, descriptor) if mode == "attended" and ( actor != "human" or os.getenv("WARDEN_AGENT_ID") or "parent-session" not in descriptor.revocation_scopes ): raise Refused("attended step requires a human parent session") if mode == "owner" and ( entry.effective_delegation.mode != "native" or entry.owner_repo != descriptor.owner or entry.warden_executes or descriptor.effect == "mint" ): raise Refused("owner step lane is not the owner's native path") parameters = case.get("parameters", {}) declared = { item["name"]: item for item in case_type.get("parameters", []) if isinstance(item, dict) and isinstance(item.get("name"), str) } if ( not isinstance(parameters, dict) or descriptor.target_parameter not in declared ): raise Refused("target parameter is not declared by acknowledged type") if case_type.get("id") == "oidc-browser-relying-party": if type(self.browser_inputs) is not BrowserInputs: raise Refused("browser metadata validator is not configured") target = self.browser_inputs.target( yaml.safe_load(case_source), yaml.safe_load(type_source), descriptor.target_parameter, ) else: for name in parameters: definition = declared.get(name) if ( definition is None or definition.get("sensitivity") not in ("public", "operational") or definition.get("case_settable") is False ): raise Refused("unknown, fixed or secret-sensitivity case parameter") target = parameters.get(descriptor.target_parameter) if not isinstance(target, str) or not IDENTIFIER.fullmatch(target): raise Refused("target must be exact case metadata") adapter = self.adapters.get(descriptor.adapter) if adapter is None or (adapter.reference, adapter.source) != ( descriptor.adapter, descriptor.adapter_source, ): raise Refused("unknown adapter or owner source pin") sheet = json.dumps( [ descriptor.step, descriptor.owner, target, actor, principal, lane, acceptance_evidence, materialization_evidence, descriptor_evidence, session_evidence, ], separators=(",", ":"), ).encode() pins = Pins( digest(case_source), digest(type_source), digest(descriptor_source), digest(catalog_source), digest(acceptance_evidence.encode()), digest(sheet), ) invocation = Invocation( descriptor.step, descriptor.owner, target, actor, principal, lane, pins ) initial = StepState( descriptor.step, descriptor.owner, pins, descriptor.allowed_created, descriptor.retry_safe, ) self._admitted[invocation] = (descriptor, initial, session_evidence) return descriptor, invocation, initial def _result( self, adapter: Adapter, invocation: Invocation, result: OwnerResult ) -> OwnerResult: if type(result) is not OwnerResult or not isinstance(result.outcome, Outcome): raise Refused("unknown owner result shape") if type(result.created) is not tuple or not all( isinstance(v, str) and IDENTIFIER.fullmatch(v) for v in result.created ): raise Refused("owner created-object result invalid") if not isinstance(result.evidence, str) or not IDENTIFIER.fullmatch( result.evidence ): raise Refused("owner evidence identifier invalid") if adapter.verify(invocation, result) is not True or result.revoked is not True: raise Refused("owner result or session revocation unverified") if (result.attempt_id, result.operation) != ( invocation.attempt_id, invocation.operation, ): raise Refused("owner receipt belongs to another attempt") descriptor = self._admitted[ replace(invocation, attempt_id="", operation="admission") ][0] proof_fields = { "parent-session": ( f"session-revocation:{invocation.lane}:{invocation.operation}:{invocation.attempt_id}", result.session_revocation_evidence, ), "broker-grant": ( f"grant-revocation:{invocation.lane}:{invocation.operation}:{invocation.attempt_id}", result.grant_revocation_evidence, ), } proofs = [proof_fields[scope] for scope in descriptor.revocation_scopes] for purpose, evidence in proofs: if ( not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or self._verify_evidence(purpose, invocation.pins.case, evidence) is not True ): raise Refused("independent grant or parent-session revocation absent") return result def _verify_evidence(self, purpose, source, evidence): try: with contained_output(): return self.evidence_verifier(purpose, source, evidence) is True except Exception: raise Refused("independent evidence verification failed") from None def _admission( self, descriptor: Descriptor, invocation: Invocation, initial: StepState ): if self._admitted.get(invocation, ())[:2] != (descriptor, initial): raise Refused("execution requires verified admission on this executor") adapter = self.adapters.get(descriptor.adapter) if adapter is None or (adapter.reference, adapter.source) != ( descriptor.adapter, descriptor.adapter_source, ): raise Refused("adapter changed after admission") session = self._admitted[invocation][2] if ( self._verify_evidence( f"caller-session:{invocation.actor}:{invocation.principal}:{invocation.lane}", invocation.pins.case, session, ) is not True ): raise Refused("caller session no longer verified") if descriptor.effect == "mint" and os.getenv("WARDEN_AGENT_ID"): raise Refused("agent mint is forbidden") return adapter def _attempt(self, invocation: Invocation, state: StepState, operation: str): identity = json.dumps( [asdict(invocation.pins), operation, state.last.attempt, uuid.uuid4().hex], separators=(",", ":"), ).encode() return replace( invocation, attempt_id="attempt-" + hashlib.sha256(identity).hexdigest(), operation=operation, ) def run( self, descriptor: Descriptor, invocation: Invocation, initial: StepState, *, retry=False, ) -> StepState: adapter = self._admission(descriptor, invocation, initial) with self.journal.locked(): binding = f"{invocation.step}:{invocation.target}" state = self.journal.load(binding, initial) recovered = recover(state) if recovered != state: self.journal.save(binding, recovered) state = recovered if len(state.receipts) > MAX_HISTORY - 4: raise Refused("bounded journal history requires owner archival") running = begin(state, invocation.pins, retry=retry) dispatched = self._attempt(invocation, running, "run") self.journal.save( binding, running, invocation=dispatched, intent=OriginalIntent(binding, dispatched, adapter.source), ) evidence = None try: with contained_output(): result = self._result( adapter, dispatched, adapter.invoke(dispatched) ) completed = finish( running, invocation.pins, result.outcome, created=result.created ) evidence = result except Exception: completed = finish(running, invocation.pins, Outcome.UNCERTAIN) self.journal.save(binding, completed, evidence, dispatched) return completed def rollback( self, descriptor: Descriptor, invocation: Invocation, initial: StepState ) -> StepState: adapter = self._admission(descriptor, invocation, initial) with self.journal.locked(): binding = f"{invocation.step}:{invocation.target}" state = self.journal.load(binding, initial) if recover(state) != state: self.journal.save(binding, recover(state)) raise Refused("lost rollback result requires owner reconciliation") if adapter.rollback is None: raise Refused("owner has no reviewed undo adapter") if len(state.receipts) > MAX_HISTORY - 4: raise Refused("bounded journal history requires owner archival") running = begin_rollback(state, invocation.pins) dispatched = self._attempt(invocation, running, "rollback") self.journal.save( binding, running, invocation=dispatched, intent=OriginalIntent(binding, dispatched, adapter.source), ) evidence = None try: with contained_output(): result = self._result( adapter, dispatched, adapter.rollback(dispatched, running.last.created), ) verified = result.outcome is Outcome.ROLLED_BACK and not result.created evidence = result if verified else None except Exception: verified = False completed = finish_rollback(running, invocation.pins, verified=verified) self.journal.save(binding, completed, evidence, dispatched) return completed def reconcile( self, descriptor: Descriptor, invocation: Invocation, initial: StepState ) -> StepState: """Explicit owner-native state inspection; never redispatches the mutation.""" adapter = self._admission(descriptor, invocation, initial) if adapter.reconcile is None and adapter.reconcile_original is None: raise Refused("owner reconciliation adapter absent") with self.journal.locked(): binding = f"{invocation.step}:{invocation.target}" state = recover(self.journal.load(binding, initial)) if not state.last or state.last.outcome not in ( Outcome.UNCERTAIN, Outcome.FAILED_AFTER, ): raise Refused("no uncertain mutation to reconcile") if len(state.receipts) > MAX_HISTORY - 4: raise Refused("bounded journal history requires owner archival") state = replace( state, receipts=( *state.receipts, Receipt(len(state.receipts) + 1, Outcome.UNCERTAIN), ), ) original = ( self.journal.original_intent(binding, invocation, adapter.source) if adapter.reconcile_original is not None else None ) dispatched = self._attempt(invocation, state, "reconcile") self.journal.save(binding, state, invocation=dispatched) with contained_output(): result = self._result( adapter, dispatched, adapter.reconcile_original(ReconcileContext(original, dispatched)) if original is not None else adapter.reconcile(dispatched), ) if result.outcome not in ( Outcome.NO_CHANGE, Outcome.SUCCESS, Outcome.FAILED_BEFORE, ): raise Refused("owner did not reconcile target state") # Validate created-object scope using the same closed result reducer. finish( begin( StepState( state.step_id, state.owner, state.pins, state.allowed_created, state.retry_safe, ), state.pins, ), state.pins, result.outcome, created=result.created, ) reconciled = StepState( state.step_id, state.owner, state.pins, state.allowed_created, state.retry_safe, ( *state.receipts, Receipt(len(state.receipts) + 1, result.outcome, result.created), ), ) self.journal.save(binding, reconciled, result, dispatched) return reconciled class OwnerInterrupted(KeyboardInterrupt): """Sanitized interruption preserves persisted running/crash recovery semantics.""" def __init__(self): super().__init__("owner execution interrupted") class _Discard: """Never retain or forward adapter output, including unbounded malicious text.""" def write(self, text): return len(text) def flush(self): pass