"""SQLite persistence layer for the sibling Tool Result v2 reference contract. This standard-library-only project demonstrates a durable idempotency ledger, transactional outbox, expiring/reclaimable worker lease, downstream receipt stand-in, and explicit status reconciliation. It intentionally omits a continuously scheduled outbox poller and does not claim delivery liveness or distributed exactly-once execution. """ from __future__ import annotations import argparse import copy from dataclasses import dataclass from datetime import datetime, timedelta, timezone import json from pathlib import Path import re import sqlite3 import sys from typing import Any, Callable EXAMPLES_DIR = Path(__file__).resolve().parent.parent if str(EXAMPLES_DIR) not in sys.path: sys.path.insert(0, str(EXAMPLES_DIR)) import tool_dispatcher as tool_v2 # noqa: E402 DATABASE_SCHEMA_VERSION = "persistent-tool-runtime-v3" FIXTURE_SCHEMA_VERSION = "persistent-tool-case-v1" OUTBOX_SCHEMA_VERSION = "tool-outbox-event-v1" MINIMUM_SQLITE = (3, 37, 0) MAX_PORTABLE_UNIX_SECONDS = 253_402_300_799 # 9999-12-31T23:59:59Z DEFAULT_LEASE_SECONDS = 30 MAX_FIXTURE_BYTES = 65_536 MAX_STORED_JSON_BYTES = 65_536 UTC_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) VALID_FAILURES = { "none", "after_intent_commit", "after_claim", "after_downstream_commit", } HEX64 = re.compile(r"^[0-9a-f]{64}$") STATUS_REF = re.compile(r"^status_[0-9a-f]{32}$") FIXTURE_FIELDS = { "schema_version", "principal", "proposal", "execution_context", "approval_mode", "now", } class PersistenceContractError(RuntimeError): """Raised when stored state violates the persistence contract.""" class PersistenceFixtureError(ValueError): """Raised when a CLI fixture is outside its exact JSON contract.""" class PersistenceFixtureIOError(PersistenceFixtureError): """Raised when the CLI fixture cannot be read without exposing its path.""" class ClosingConnection(sqlite3.Connection): """Preserve sqlite3 transaction context semantics and close on exit.""" def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> bool: try: return bool(super().__exit__(exc_type, exc_value, traceback)) finally: self.close() @dataclass(frozen=True) class OutboxClaim: operation_pk: int status_ref: str worker_id: str request_sha256: str tool_name: str attempt_count: int @dataclass(frozen=True) class Reservation: disposition: str operation_pk: int | None status_ref: str | None SCHEMA_SQL = """ CREATE TABLE IF NOT EXISTS runtime_metadata ( metadata_key TEXT PRIMARY KEY, metadata_value TEXT NOT NULL ) STRICT; CREATE TABLE IF NOT EXISTS call_bindings ( tenant_id TEXT NOT NULL, subject_id TEXT NOT NULL, provider TEXT NOT NULL, api_family TEXT NOT NULL, response_id TEXT NOT NULL, call_id TEXT NOT NULL, fingerprint_sha256 TEXT NOT NULL CHECK(length(fingerprint_sha256) = 64 AND fingerprint_sha256 NOT GLOB '*[^0-9a-f]*'), created_at INTEGER NOT NULL CHECK(created_at >= 0), PRIMARY KEY ( tenant_id, subject_id, provider, api_family, response_id, call_id ) ) STRICT; CREATE TABLE IF NOT EXISTS operations ( operation_pk INTEGER PRIMARY KEY, tenant_id TEXT NOT NULL, subject_id TEXT NOT NULL, tool_name TEXT NOT NULL, idempotency_key TEXT NOT NULL CHECK(length(idempotency_key) BETWEEN 1 AND 255), request_sha256 TEXT NOT NULL CHECK(length(request_sha256) = 64 AND request_sha256 NOT GLOB '*[^0-9a-f]*'), arguments_json TEXT NOT NULL, input_schema_revision TEXT NOT NULL, output_schema_revision TEXT NOT NULL, effect_revision TEXT NOT NULL, handler_revision TEXT NOT NULL, producer_revision TEXT NOT NULL, policy_revision TEXT NOT NULL, approval_id TEXT NOT NULL, approver_id TEXT NOT NULL, approval_digest TEXT NOT NULL CHECK(length(approval_digest) = 64 AND approval_digest NOT GLOB '*[^0-9a-f]*'), approval_expires_at INTEGER NOT NULL CHECK(approval_expires_at >= 0), approved_at INTEGER NOT NULL CHECK(approved_at >= 0), first_operation_id TEXT NOT NULL, first_call_id TEXT NOT NULL, first_response_id TEXT NOT NULL, first_provider TEXT NOT NULL, first_api_family TEXT NOT NULL, first_adapter_revision TEXT NOT NULL, status_ref TEXT NOT NULL UNIQUE CHECK(length(status_ref) = 39 AND substr(status_ref, 1, 7) = 'status_' AND substr(status_ref, 8) NOT GLOB '*[^0-9a-f]*'), state TEXT NOT NULL CHECK(state IN ('pending', 'processing', 'succeeded')), created_at INTEGER NOT NULL CHECK(created_at >= 0), updated_at INTEGER NOT NULL CHECK(updated_at >= created_at), completed_at INTEGER CHECK(completed_at IS NULL OR completed_at >= created_at), UNIQUE(tenant_id, subject_id, tool_name, idempotency_key) ) STRICT; CREATE TABLE IF NOT EXISTS outbox ( event_id TEXT PRIMARY KEY, operation_pk INTEGER NOT NULL UNIQUE REFERENCES operations(operation_pk) ON DELETE RESTRICT, payload_json TEXT NOT NULL, state TEXT NOT NULL CHECK(state IN ('pending', 'processing', 'delivered')), lease_owner TEXT, lease_until INTEGER, attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0), created_at INTEGER NOT NULL CHECK(created_at >= 0), updated_at INTEGER NOT NULL CHECK(updated_at >= created_at), CHECK( (state = 'pending' AND lease_owner IS NULL AND lease_until IS NULL) OR (state = 'processing' AND lease_owner IS NOT NULL AND lease_until IS NOT NULL) OR (state = 'delivered' AND lease_owner IS NULL AND lease_until IS NULL) ) ) STRICT; -- This is an offline stand-in for an independently committed downstream -- system. Production code must query the real dependency using its supported -- idempotency/status contract rather than sharing this database. CREATE TABLE IF NOT EXISTS downstream_receipts ( downstream_pk INTEGER PRIMARY KEY, tenant_id TEXT NOT NULL, subject_id TEXT NOT NULL, tool_name TEXT NOT NULL, idempotency_key TEXT NOT NULL, request_sha256 TEXT NOT NULL CHECK(length(request_sha256) = 64 AND request_sha256 NOT GLOB '*[^0-9a-f]*'), receipt_id TEXT NOT NULL UNIQUE, downstream_request_id TEXT NOT NULL UNIQUE, result_json TEXT NOT NULL, producer_revision TEXT NOT NULL, resource_revision TEXT NOT NULL, observed_at TEXT NOT NULL, effect_revision TEXT NOT NULL, committed_at INTEGER NOT NULL CHECK(committed_at >= 0), UNIQUE(tenant_id, subject_id, tool_name, idempotency_key) ) STRICT; CREATE TABLE IF NOT EXISTS local_receipts ( operation_pk INTEGER PRIMARY KEY REFERENCES operations(operation_pk) ON DELETE RESTRICT, receipt_id TEXT NOT NULL UNIQUE, downstream_request_id TEXT NOT NULL UNIQUE, request_sha256 TEXT NOT NULL CHECK(length(request_sha256) = 64 AND request_sha256 NOT GLOB '*[^0-9a-f]*'), result_json TEXT NOT NULL, producer_revision TEXT NOT NULL, resource_revision TEXT NOT NULL, observed_at TEXT NOT NULL, effect_revision TEXT NOT NULL, reconciled_at INTEGER NOT NULL CHECK(reconciled_at >= 0) ) STRICT; """ def _reject_constant(value: str) -> None: raise PersistenceFixtureError(f"JSON disallows non-finite constants: {value}") def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise PersistenceFixtureError(f"JSON object contains a duplicate key: {key}") result[key] = value return result def _strict_json_loads(text: str, *, label: str) -> Any: if not isinstance(text, str): raise PersistenceContractError(f"{label} must be TEXT") try: raw = text.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise PersistenceContractError(f"{label} is not valid UTF-8 text") from exc if len(raw) > MAX_STORED_JSON_BYTES: raise PersistenceContractError( f"{label} exceeds the {MAX_STORED_JSON_BYTES}-byte UTF-8 limit" ) try: tool_v2._reject_excessive_json_nesting(text) value = json.loads( text, object_pairs_hook=_unique_object, parse_constant=_reject_constant, ) tool_v2._validate_json_domain(value) except (json.JSONDecodeError, RecursionError, TypeError, ValueError, UnicodeError) as exc: raise PersistenceContractError(f"{label} is not strict JSON: {exc}") from exc return value def _canonical_json(value: Any) -> str: return tool_v2._canonical_json(value) def _decode_canonical_object(text: Any, *, label: str, fields: set[str] | None = None) -> dict[str, Any]: if not isinstance(text, str): raise PersistenceContractError(f"{label} must be TEXT") value = _strict_json_loads(text, label=label) if not isinstance(value, dict): raise PersistenceContractError(f"{label} must parse to an object") if _canonical_json(value) != text: raise PersistenceContractError(f"{label} is not current canonical JSON encoding") if fields is not None and set(value) != fields: raise PersistenceContractError(f"{label} fields are not exact") return value def _nonempty(value: Any) -> bool: return isinstance(value, str) and bool(value.strip()) def validate_fixture(value: Any) -> list[str]: errors: list[str] = [] if not isinstance(value, dict): return ["fixture root must be an object"] if set(value) != FIXTURE_FIELDS: errors.append( f"fixture root fields do not match: missing={sorted(FIXTURE_FIELDS - set(value))}, " f"extra={sorted(set(value) - FIXTURE_FIELDS)}" ) return errors if value["schema_version"] != FIXTURE_SCHEMA_VERSION: errors.append("schema_version does not match") principal = value["principal"] if not isinstance(principal, dict) or set(principal) != tool_v2.PRINCIPAL_FIELDS: errors.append("principal fields are not exact") else: if not _nonempty(principal["tenant_id"]) or not _nonempty(principal["subject_id"]): errors.append("principal tenant/subject must be non-empty strings") roles = principal["roles"] if not isinstance(roles, list) or not all(_nonempty(role) for role in roles): errors.append("principal.roles must be an array of non-empty strings") elif roles != sorted(set(roles)): errors.append("principal.roles must be sorted and contain no duplicates") proposal = value["proposal"] if not isinstance(proposal, dict) or set(proposal) != tool_v2.PROPOSAL_FIELDS: errors.append("proposal fields are not exact") else: if proposal["name"] != "create_refund_draft": errors.append("the persistence project accepts only create_refund_draft") if not isinstance(proposal["arguments"], dict): errors.append("proposal.arguments must be an object") else: try: tool_v2._validate_json_domain(proposal["arguments"]) except (TypeError, ValueError, UnicodeError) as exc: errors.append(f"proposal.arguments is outside the JSON domain: {exc}") context = value["execution_context"] if not isinstance(context, dict) or set(context) != tool_v2.CONTEXT_FIELDS: errors.append("execution_context fields are not exact") else: string_fields = set(tool_v2.CONTEXT_FIELDS) - {"idempotency_key"} if not all(_nonempty(context[field]) for field in string_fields): errors.append("execution_context correlation fields must be non-empty strings") key = context["idempotency_key"] if not isinstance(key, str) or not 1 <= len(key) <= 255: errors.append("idempotency_key must contain 1..255 characters") profile = (context.get("provider"), context.get("api_family")) if tool_v2.PROVIDER_PROFILES.get(profile) != context.get("adapter_revision"): errors.append("provider profile/adapter revision is not registered") if value["approval_mode"] not in tool_v2.VALID_APPROVAL_MODES: errors.append("approval_mode is invalid") if ( type(value["now"]) is not int or value["now"] < 0 or value["now"] > MAX_PORTABLE_UNIX_SECONDS - DEFAULT_LEASE_SECONDS ): errors.append("now must be a non-negative timestamp representable by the portable UTC datetime range and the default lease") return errors def load_fixture(path: Path) -> dict[str, Any]: try: with path.open("rb") as handle: raw = handle.read(MAX_FIXTURE_BYTES + 1) except OSError as exc: raise PersistenceFixtureIOError("Cannot read fixture") from exc if len(raw) > MAX_FIXTURE_BYTES: raise PersistenceFixtureError(f"fixture exceeds {MAX_FIXTURE_BYTES} bytes") try: text = raw.decode("utf-8") tool_v2._reject_excessive_json_nesting(text) value = json.loads( text, object_pairs_hook=_unique_object, parse_constant=_reject_constant, ) tool_v2._validate_json_domain(value) except ( UnicodeDecodeError, json.JSONDecodeError, RecursionError, TypeError, ValueError, ) as exc: raise PersistenceFixtureError(f"fixture is not strict UTF-8 JSON: {exc}") from exc errors = validate_fixture(value) if errors: raise PersistenceFixtureError("fixture validation failed: " + "; ".join(errors)) return value def fixture_principal_call(fixture: dict[str, Any]) -> tuple[tool_v2.Principal, tool_v2.ToolCall]: principal_value = fixture["principal"] context = fixture["execution_context"] principal = tool_v2.Principal( principal_value["tenant_id"], principal_value["subject_id"], tuple(principal_value["roles"]), ) call = tool_v2.bind_tool_call( tool_v2.ModelProposal( fixture["proposal"]["name"], copy.deepcopy(fixture["proposal"]["arguments"]), ), tool_v2.ExecutionContext( call_id=context["call_id"], operation_id=context["operation_id"], idempotency_key=context["idempotency_key"], provider=context["provider"], api_family=context["api_family"], response_id=context["response_id"], adapter_revision=context["adapter_revision"], ), ) return principal, call def status_query_call( original_call: tool_v2.ToolCall, status_ref: str ) -> tool_v2.ToolCall: """Derive a separate host-owned provider identity for explicit status lookup.""" if not isinstance(status_ref, str) or STATUS_REF.fullmatch(status_ref) is None: raise ValueError("status_ref format is invalid") suffix = tool_v2._digest( { "provider": original_call.provider, "api_family": original_call.api_family, "response_id": original_call.response_id, "call_id": original_call.call_id, "status_ref": status_ref, } )[:24] return tool_v2.ToolCall( call_id=f"status-call-{suffix}", operation_id=original_call.operation_id, name=original_call.name, arguments=copy.deepcopy(original_call.arguments), idempotency_key=original_call.idempotency_key, provider=original_call.provider, api_family=original_call.api_family, response_id=f"status-response-{suffix}", adapter_revision=original_call.adapter_revision, ) class PersistentToolRuntime: """Durable adapter around the existing Tool Result v2 write contract.""" def __init__( self, database: Path, *, authorization_resolver: Callable[[tool_v2.Principal, str], bool] | None = None, principal_resolver: Callable[[str, str], tool_v2.Principal | None] | None = None, busy_timeout_ms: int = 5_000, ) -> None: self.database = Path(database) if str(self.database) == ":memory:": raise ValueError("the multi-connection project does not accept an :memory: database") if not self.database.parent.exists(): raise ValueError("the database parent directory does not exist") if type(busy_timeout_ms) is not int or busy_timeout_ms < 1: raise ValueError("busy_timeout_ms must be a positive integer") self.busy_timeout_ms = busy_timeout_ms self.contract = tool_v2.Dispatcher() self.authorization_resolver = authorization_resolver self.principal_resolver = principal_resolver self._initialize_database() def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect( self.database, timeout=self.busy_timeout_ms / 1000, isolation_level=None, factory=ClosingConnection, ) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute(f"PRAGMA busy_timeout = {self.busy_timeout_ms}") connection.execute("PRAGMA synchronous = FULL") return connection @staticmethod def _begin_immediate(connection: sqlite3.Connection) -> None: connection.execute("BEGIN IMMEDIATE") def _initialize_database(self) -> None: if sqlite3.sqlite_version_info < MINIMUM_SQLITE: minimum = ".".join(str(item) for item in MINIMUM_SQLITE) raise PersistenceContractError(f"SQLite {minimum}+ is required for the STRICT tables in this project") with self._connect() as connection: mode = connection.execute("PRAGMA journal_mode = WAL").fetchone()[0] if str(mode).lower() != "wal": raise PersistenceContractError("SQLite did not enter WAL mode") connection.executescript(SCHEMA_SQL) connection.execute( """ INSERT INTO runtime_metadata(metadata_key, metadata_value) VALUES ('schema_version', ?) ON CONFLICT(metadata_key) DO NOTHING """, (DATABASE_SCHEMA_VERSION,), ) stored = connection.execute( "SELECT metadata_value FROM runtime_metadata WHERE metadata_key = 'schema_version'" ).fetchone() if stored is None or stored["metadata_value"] != DATABASE_SCHEMA_VERSION: raise PersistenceContractError("database schema_version does not match") def _authorized(self, principal: tool_v2.Principal, order_ref: str) -> bool: if self.authorization_resolver is not None: return self.authorization_resolver(principal, order_ref) return self.contract._authorized_order(principal, order_ref) def _resolve_current_principal( self, tenant_id: str, subject_id: str ) -> tool_v2.Principal | None: """Resolve current claims; durable intents never retain a roles snapshot.""" if self.principal_resolver is None: # The teaching default is deliberately owner-only. A deployment # with role-based policy must inject current claims from IAM rather # than trusting the roles present when the intent was created. return tool_v2.Principal(tenant_id, subject_id, ()) try: principal = self.principal_resolver(tenant_id, subject_id) except Exception as exc: raise PersistenceContractError("current principal resolver failed") from exc if principal is None: return None if not isinstance(principal, tool_v2.Principal): raise PersistenceContractError("current principal resolver returned an invalid type") if (principal.tenant_id, principal.subject_id) != (tenant_id, subject_id): raise PersistenceContractError("current principal resolver changed tenant/subject") if ( not isinstance(principal.roles, tuple) or len(principal.roles) > tool_v2.MAX_ROLES or any( not tool_v2._bounded_identifier(role) for role in principal.roles ) or tuple(sorted(set(principal.roles))) != principal.roles ): raise PersistenceContractError("current principal resolver returned invalid roles") return principal def _current_authorized_principal( self, principal: tool_v2.Principal, order_ref: str ) -> tool_v2.Principal | None: """Resolve current claims and check the resource with those claims only. The presented ``Principal`` establishes the immutable tenant/subject scope of this request. It is deliberately not a durable role snapshot: both reservation and worker delivery obtain fresh roles through the same resolver. The owner-only default is therefore fail closed when a deployment has not installed a current-claims resolver. """ current = self._resolve_current_principal( principal.tenant_id, principal.subject_id ) if current is None or not self._authorized(current, order_ref): return None return current def _preflight( self, principal: tool_v2.Principal, call: tool_v2.ToolCall, *, now: int, ) -> tuple[tool_v2.ToolSpec | None, str, dict[str, Any] | None]: tool_v2.validate_host_context(principal, call, now=now) spec = self.contract.registry.get(call.name) if spec is None or call.name != "create_refund_draft": digest = tool_v2.request_digest(principal, call, spec) return spec, digest, tool_v2._error_package( principal, call, spec, "UNKNOWN_TOOL", digest, now=now ) if not self.contract._validate_arguments(spec, call.arguments): digest = tool_v2.request_digest(principal, call, spec) return spec, digest, tool_v2._error_package( principal, call, spec, "INVALID_ARGUMENTS", digest, now=now ) digest = tool_v2.request_digest(principal, call, spec) current_principal = self._current_authorized_principal( principal, call.arguments["order_ref"] ) if current_principal is None: return spec, digest, tool_v2._error_package( principal, call, spec, "NOT_FOUND", digest, now=now ) if not call.idempotency_key or len(call.idempotency_key) > 255: return spec, digest, tool_v2._error_package( principal, call, spec, "IDEMPOTENCY_KEY_REQUIRED", digest, now=now ) return spec, digest, None @staticmethod def _validate_timepoint(now: int) -> None: if ( type(now) is not int or now < 0 or now > MAX_PORTABLE_UNIX_SECONDS ): raise ValueError("now is outside the portable UTC datetime range") @classmethod def _validate_lease_window(cls, now: int, lease_seconds: int) -> None: cls._validate_timepoint(now) if type(lease_seconds) is not int or lease_seconds < 1: raise ValueError("lease_seconds must be a positive integer") if lease_seconds > MAX_PORTABLE_UNIX_SECONDS - now: raise ValueError("now + lease_seconds is outside the portable UTC datetime range") @staticmethod def _scope(principal: tool_v2.Principal, call: tool_v2.ToolCall) -> tuple[str, str, str, str]: if not call.idempotency_key: raise ValueError("write action is missing an idempotency key") return principal.tenant_id, principal.subject_id, call.name, call.idempotency_key @staticmethod def _status_ref(scope: tuple[str, str, str, str], request_sha256: str) -> str: return "status_" + tool_v2._digest( {"scope": list(scope), "request_sha256": request_sha256} )[:32] @staticmethod def _event_payload(operation_pk: int, request_sha256: str, tool_name: str) -> dict[str, Any]: return { "schema_version": OUTBOX_SCHEMA_VERSION, "operation_pk": operation_pk, "request_sha256": request_sha256, "tool": tool_name, } @staticmethod def _contract_matches(row: sqlite3.Row, spec: tool_v2.ToolSpec, digest: str) -> bool: return ( row["request_sha256"] == digest and row["input_schema_revision"] == spec.schema_version and row["output_schema_revision"] == spec.output_schema_revision and row["effect_revision"] == spec.effect_revision and row["handler_revision"] == spec.handler_revision and row["producer_revision"] == spec.producer_revision and row["policy_revision"] == spec.approval_revision ) @staticmethod def _approval_provider_context_matches( row: sqlite3.Row, call: tool_v2.ToolCall ) -> bool: """Keep an accepted approval inside its provider/API/adapter context.""" return ( row["first_provider"], row["first_api_family"], row["first_adapter_revision"], ) == (call.provider, call.api_family, call.adapter_revision) @staticmethod def _call_key( principal: tool_v2.Principal, call: tool_v2.ToolCall ) -> tuple[str, str, str, str, str, str]: return ( principal.tenant_id, principal.subject_id, call.provider, call.api_family, call.response_id, call.call_id, ) def _bind_call_identity( self, connection: sqlite3.Connection, principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, *, purpose: str, now: int, ) -> bool: fingerprint = tool_v2.call_fingerprint( principal, call, spec, purpose=purpose ) call_key = self._call_key(principal, call) connection.execute( """ INSERT INTO call_bindings( tenant_id, subject_id, provider, api_family, response_id, call_id, fingerprint_sha256, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT( tenant_id, subject_id, provider, api_family, response_id, call_id ) DO NOTHING """, (*call_key, fingerprint, now), ) call_row = connection.execute( """ SELECT fingerprint_sha256 FROM call_bindings WHERE tenant_id = ? AND subject_id = ? AND provider = ? AND api_family = ? AND response_id = ? AND call_id = ? """, call_key, ).fetchone() if call_row is None: raise PersistenceContractError("call binding is not visible after insertion") return call_row["fingerprint_sha256"] == fingerprint def _validate_operation_row( self, row: sqlite3.Row, principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, digest: str, ) -> dict[str, Any]: scope = self._scope(principal, call) if tuple(row[field] for field in ("tenant_id", "subject_id", "tool_name", "idempotency_key")) != scope: raise PersistenceContractError("operation scope is inconsistent with the current request") if not self._contract_matches(row, spec, digest): raise PersistenceContractError("operation digest/contract revision is inconsistent with the current request") arguments = _decode_canonical_object( row["arguments_json"], label="operations.arguments_json" ) if arguments != call.arguments: raise PersistenceContractError("operation arguments are inconsistent with the current request") expected_status_ref = self._status_ref(scope, digest) if row["status_ref"] != expected_status_ref or STATUS_REF.fullmatch(row["status_ref"]) is None: raise PersistenceContractError("operation status_ref binding is invalid") approval_context_fields = ( "first_operation_id", "first_call_id", "first_response_id", "first_provider", "first_api_family", "first_adapter_revision", ) if any(not tool_v2._bounded_identifier(row[field]) for field in approval_context_fields): raise PersistenceContractError("operation provider/call evidence fields are invalid") approval_call = tool_v2.ToolCall( row["first_call_id"], row["first_operation_id"], row["tool_name"], arguments, row["idempotency_key"], provider=row["first_provider"], api_family=row["first_api_family"], response_id=row["first_response_id"], adapter_revision=row["first_adapter_revision"], ) if ( tool_v2.PROVIDER_PROFILES.get( (approval_call.provider, approval_call.api_family) ) != approval_call.adapter_revision ): raise PersistenceContractError("operation provider-profile evidence is invalid") approver_id = row["approver_id"] if ( not tool_v2._bounded_identifier(approver_id) or approver_id not in tool_v2.AUTHORIZED_APPROVER_IDS ): raise PersistenceContractError("operation approver evidence is invalid") expected_approval = tool_v2.approval_digest( principal, approval_call, spec, approver_id=approver_id ) if ( not _nonempty(row["approval_id"]) or row["approval_digest"] != expected_approval or type(row["approved_at"]) is not int or type(row["approval_expires_at"]) is not int or row["approved_at"] != row["created_at"] or row["approval_expires_at"] <= row["approved_at"] or row["approval_expires_at"] > tool_v2.MAX_PORTABLE_UNIX_SECONDS ): raise PersistenceContractError("operation approval evidence binding is invalid") return arguments def _reserve( self, principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, digest: str, approvals: tuple[tool_v2.Approval, ...], now: int, ) -> Reservation: # Re-resolve immediately before the durable intent transaction. The # earlier preflight gives a cheap reject path; this second lookup closes # the gap in which a role or resource grant may have changed. current_principal = self._current_authorized_principal( principal, call.arguments["order_ref"] ) if current_principal is None: return Reservation("authorization_denied", None, None) scope = self._scope(current_principal, call) status_ref = self._status_ref(scope, digest) connection = self._connect() try: self._begin_immediate(connection) if not self._bind_call_identity( connection, current_principal, call, spec, purpose="dispatch", now=now, ): connection.commit() return Reservation("call_conflict", None, None) row = connection.execute( """ SELECT * FROM operations WHERE tenant_id = ? AND subject_id = ? AND tool_name = ? AND idempotency_key = ? """, scope, ).fetchone() if row is not None: if ( not self._contract_matches(row, spec, digest) or not self._approval_provider_context_matches(row, call) ): connection.commit() return Reservation("idempotency_conflict", row["operation_pk"], row["status_ref"]) self._validate_operation_row( row, current_principal, call, spec, digest ) disposition = "succeeded" if row["state"] == "succeeded" else "unknown" connection.commit() return Reservation(disposition, row["operation_pk"], row["status_ref"]) if not spec.business_validator(current_principal, call.arguments): connection.commit() return Reservation("business_rule_violation", None, None) if not approvals: connection.commit() return Reservation("approval_required", None, None) if not self.contract._approval_valid( approvals, current_principal, call, spec, now ): connection.commit() return Reservation("approval_invalid", None, None) matching_approvals = [ approval for approval in approvals if isinstance(approval, tool_v2.Approval) and tool_v2._bounded_identifier(approval.approver_id) and approval.approver_id in tool_v2.AUTHORIZED_APPROVER_IDS and approval.operation_id == call.operation_id and approval.call_id == call.call_id and approval.provider == call.provider and approval.api_family == call.api_family and approval.adapter_revision == call.adapter_revision and approval.idempotency_key == call.idempotency_key and approval.tool_name == call.name and approval.subject_id == current_principal.subject_id and approval.schema_version == spec.schema_version and approval.approval_revision == spec.approval_revision and approval.approval_digest == tool_v2.approval_digest( current_principal, call, spec, approver_id=approval.approver_id, ) and type(approval.expires_at) is int and approval.expires_at > now and approval.expires_at <= tool_v2.MAX_PORTABLE_UNIX_SECONDS ] if not matching_approvals: raise PersistenceContractError("approval validation is inconsistent with the persisted selection") selected_approval = sorted( matching_approvals, key=lambda item: (item.approval_id, item.expires_at) )[0] contract = tool_v2._tool_contract(spec, call.name) cursor = connection.execute( """ INSERT INTO operations( tenant_id, subject_id, tool_name, idempotency_key, request_sha256, arguments_json, input_schema_revision, output_schema_revision, effect_revision, handler_revision, producer_revision, policy_revision, approval_id, approver_id, approval_digest, approval_expires_at, approved_at, first_operation_id, first_call_id, first_response_id, first_provider, first_api_family, first_adapter_revision, status_ref, state, created_at, updated_at, completed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, NULL) ON CONFLICT(tenant_id, subject_id, tool_name, idempotency_key) DO NOTHING """, ( *scope, digest, _canonical_json(call.arguments), contract["input_schema_revision"], contract["output_schema_revision"], contract["effect_revision"], contract["handler_revision"], contract["producer_revision"], contract["policy_revision"], selected_approval.approval_id, selected_approval.approver_id, selected_approval.approval_digest, selected_approval.expires_at, now, call.operation_id, call.call_id, call.response_id, call.provider, call.api_family, call.adapter_revision, status_ref, now, now, ), ) if cursor.rowcount != 1: raise PersistenceContractError("an unexpected operation race occurred in BEGIN IMMEDIATE") operation_pk = int(cursor.lastrowid) payload = self._event_payload(operation_pk, digest, call.name) event_id = "outbox_" + tool_v2._digest(payload)[:32] connection.execute( """ INSERT INTO outbox( event_id, operation_pk, payload_json, state, lease_owner, lease_until, attempt_count, created_at, updated_at ) VALUES (?, ?, ?, 'pending', NULL, NULL, 0, ?, ?) """, (event_id, operation_pk, _canonical_json(payload), now, now), ) connection.commit() return Reservation("new", operation_pk, status_ref) except BaseException: if connection.in_transaction: connection.rollback() raise finally: connection.close() def _read_operation_by_status(self, status_ref: str) -> sqlite3.Row | None: with self._connect() as connection: return connection.execute( "SELECT * FROM operations WHERE status_ref = ?", (status_ref,) ).fetchone() def _unknown_package( self, principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, digest: str, *, now: int, status_ref: str, ) -> dict[str, Any]: return tool_v2._error_package( principal, call, spec, "OUTCOME_UNKNOWN", digest, now=now, status_ref=status_ref, ) def _load_local_record( self, operation_pk: int, principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, digest: str, ) -> tool_v2.IdempotencyRecord: with self._connect() as connection: operation = connection.execute( "SELECT * FROM operations WHERE operation_pk = ?", (operation_pk,) ).fetchone() receipt = connection.execute( "SELECT * FROM local_receipts WHERE operation_pk = ?", (operation_pk,) ).fetchone() if operation is None or receipt is None or operation["state"] != "succeeded": raise PersistenceContractError("a succeeded operation is missing its local receipt") self._validate_operation_row(operation, principal, call, spec, digest) if ( receipt["request_sha256"] != digest or receipt["producer_revision"] != spec.producer_revision or receipt["effect_revision"] != spec.effect_revision ): raise PersistenceContractError("local receipt is inconsistent with the operation/current contract") data = _decode_canonical_object(receipt["result_json"], label="local_receipts.result_json") raw = tool_v2.HandlerResult( data=data, producer_revision=receipt["producer_revision"], resource_revision=receipt["resource_revision"], observed_at=receipt["observed_at"], downstream_request_id=receipt["downstream_request_id"], receipt_id=receipt["receipt_id"], ) errors = tool_v2.validate_handler_result(spec, call, raw) if errors: raise PersistenceContractError("local receipt failed the output contract: " + "; ".join(errors)) return tool_v2.IdempotencyRecord( request_sha256=digest, provider=operation["first_provider"], api_family=operation["first_api_family"], adapter_revision=operation["first_adapter_revision"], data=copy.deepcopy(data), source_label=spec.producer_label, producer_revision=receipt["producer_revision"], resource_revision=receipt["resource_revision"], observed_at=receipt["observed_at"], output_schema_revision=spec.output_schema_revision, effect_revision=receipt["effect_revision"], handler_revision=spec.handler_revision, downstream_request_id=receipt["downstream_request_id"], receipt_id=receipt["receipt_id"], status_ref=operation["status_ref"], ) def dispatch( self, principal: tool_v2.Principal, call: tool_v2.ToolCall, *, approvals: tuple[tool_v2.Approval, ...] = (), now: int, failure: str = "none", worker_id: str = "inline-worker", lease_seconds: int = DEFAULT_LEASE_SECONDS, ) -> dict[str, Any]: """Reserve intent atomically, then optionally run the outbox worker.""" if failure not in VALID_FAILURES: raise ValueError(f"unknown failure: {failure}") spec, digest, preflight_error = self._preflight(principal, call, now=now) if preflight_error is not None: return preflight_error if spec is None: raise PersistenceContractError("internal error: preflight did not return a spec") # Validate all worker controls before the intent/outbox transaction. # Otherwise an invalid lease could strand a newly reserved operation. if failure != "after_intent_commit": if not _nonempty(worker_id) or len(worker_id) > 128: raise ValueError("worker_id must contain 1..128 characters") self._validate_lease_window(now, lease_seconds) reservation = self._reserve(principal, call, spec, digest, approvals, now) code_for_disposition = { "call_conflict": "CALL_ID_CONFLICT", "idempotency_conflict": "IDEMPOTENCY_CONFLICT", "authorization_denied": "NOT_FOUND", "business_rule_violation": "BUSINESS_RULE_VIOLATION", "approval_required": "APPROVAL_REQUIRED", "approval_invalid": "APPROVAL_INVALID", } if reservation.disposition in code_for_disposition: return tool_v2._error_package( principal, call, spec, code_for_disposition[reservation.disposition], digest, now=now, ) if reservation.operation_pk is None or reservation.status_ref is None: raise PersistenceContractError("operation reservation is missing identity") if reservation.disposition == "succeeded": record = self._load_local_record( reservation.operation_pk, principal, call, spec, digest ) return tool_v2._success_package( principal, call, spec, record, delivery="local_replay" ) if reservation.disposition == "unknown": return self._unknown_package( principal, call, spec, digest, now=now, status_ref=reservation.status_ref, ) if failure == "after_intent_commit": return self._unknown_package( principal, call, spec, digest, now=now, status_ref=reservation.status_ref, ) outcome = self.process_operation( reservation.status_ref, worker_id=worker_id, now=now, lease_seconds=lease_seconds, failure=failure, ) if outcome != "succeeded": return self._unknown_package( principal, call, spec, digest, now=now, status_ref=reservation.status_ref, ) record = self._load_local_record( reservation.operation_pk, principal, call, spec, digest ) return tool_v2._success_package( principal, call, spec, record, delivery="fresh" ) def claim_operation( self, status_ref: str, *, worker_id: str, now: int, lease_seconds: int = DEFAULT_LEASE_SECONDS, ) -> OutboxClaim | None: """Claim one pending or expired outbox row using a write transaction.""" if STATUS_REF.fullmatch(status_ref) is None: raise ValueError("status_ref format is invalid") if not _nonempty(worker_id) or len(worker_id) > 128: raise ValueError("worker_id must contain 1..128 characters") self._validate_lease_window(now, lease_seconds) connection = self._connect() try: self._begin_immediate(connection) row = connection.execute( """ SELECT o.operation_pk, o.status_ref, o.request_sha256, o.tool_name, b.payload_json, b.state AS outbox_state, b.lease_until, b.attempt_count FROM operations AS o JOIN outbox AS b ON b.operation_pk = o.operation_pk WHERE o.status_ref = ? """, (status_ref,), ).fetchone() if row is None or row["outbox_state"] == "delivered": connection.commit() return None claimable = row["outbox_state"] == "pending" or ( row["outbox_state"] == "processing" and isinstance(row["lease_until"], int) and row["lease_until"] <= now ) if not claimable: connection.commit() return None payload = _decode_canonical_object( row["payload_json"], label="outbox.payload_json", fields={"schema_version", "operation_pk", "request_sha256", "tool"}, ) expected = self._event_payload( row["operation_pk"], row["request_sha256"], row["tool_name"] ) if payload != expected: raise PersistenceContractError("outbox payload is inconsistent with the operation") changed = connection.execute( """ UPDATE outbox SET state = 'processing', lease_owner = ?, lease_until = ?, attempt_count = attempt_count + 1, updated_at = ? WHERE operation_pk = ? AND (state = 'pending' OR (state = 'processing' AND lease_until <= ?)) """, (worker_id, now + lease_seconds, now, row["operation_pk"], now), ).rowcount if changed != 1: connection.commit() return None connection.execute( """ UPDATE operations SET state = 'processing', updated_at = ? WHERE operation_pk = ? AND state != 'succeeded' """, (now, row["operation_pk"]), ) connection.commit() return OutboxClaim( operation_pk=row["operation_pk"], status_ref=row["status_ref"], worker_id=worker_id, request_sha256=row["request_sha256"], tool_name=row["tool_name"], attempt_count=row["attempt_count"] + 1, ) except BaseException: if connection.in_transaction: connection.rollback() raise finally: connection.close() def _commit_downstream(self, claim: OutboxClaim, *, now: int) -> None: """Commit a deterministic downstream receipt with its own unique key.""" with self._connect() as connection: operation = connection.execute( "SELECT * FROM operations WHERE operation_pk = ?", (claim.operation_pk,) ).fetchone() if operation is None or operation["status_ref"] != claim.status_ref: raise PersistenceContractError("the operation referenced by the claim does not exist") if operation["request_sha256"] != claim.request_sha256: raise PersistenceContractError("claim request digest does not match") arguments = _decode_canonical_object( operation["arguments_json"], label="operations.arguments_json" ) scope = tuple( operation[field] for field in ("tenant_id", "subject_id", "tool_name", "idempotency_key") ) effect_identity = tool_v2._digest( {"scope": list(scope), "request_sha256": claim.request_sha256} ) draft_number = int(effect_identity[:12], 16) + 1 data = { "draft_id": f"DRAFT-{draft_number}", "order_ref": arguments.get("order_ref"), "reason": arguments.get("reason"), } result_json = _canonical_json(data) receipt_id = "refund-receipt-" + effect_identity[:24] downstream_request_id = "refund-request-" + effect_identity[24:48] # Avoid the platform C runtime range used by datetime.fromtimestamp(). # The validated arithmetic is deterministic on Windows and POSIX. observed_at = (UTC_EPOCH + timedelta(seconds=now)).strftime( "%Y-%m-%dT%H:%M:%SZ" ) connection = self._connect() try: self._begin_immediate(connection) connection.execute( """ INSERT INTO downstream_receipts( tenant_id, subject_id, tool_name, idempotency_key, request_sha256, receipt_id, downstream_request_id, result_json, producer_revision, resource_revision, observed_at, effect_revision, committed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(tenant_id, subject_id, tool_name, idempotency_key) DO NOTHING """, ( *scope, claim.request_sha256, receipt_id, downstream_request_id, result_json, operation["producer_revision"], f"{data['draft_id'].lower()}-r1", observed_at, operation["effect_revision"], now, ), ) receipt = connection.execute( """ SELECT * FROM downstream_receipts WHERE tenant_id = ? AND subject_id = ? AND tool_name = ? AND idempotency_key = ? """, scope, ).fetchone() if receipt is None: raise PersistenceContractError("downstream receipt is missing after UPSERT") expected = { "request_sha256": claim.request_sha256, "result_json": result_json, "producer_revision": operation["producer_revision"], "effect_revision": operation["effect_revision"], } if any(receipt[field] != value for field, value in expected.items()): raise PersistenceContractError("the same downstream key has a different intent or receipt drift") connection.commit() except BaseException: if connection.in_transaction: connection.rollback() raise finally: connection.close() def _reconcile( self, operation_pk: int, principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, digest: str, *, now: int, ) -> tool_v2.IdempotencyRecord | None: connection = self._connect() try: self._begin_immediate(connection) operation = connection.execute( "SELECT * FROM operations WHERE operation_pk = ?", (operation_pk,) ).fetchone() if operation is None: connection.commit() return None self._validate_operation_row(operation, principal, call, spec, digest) downstream = connection.execute( """ SELECT * FROM downstream_receipts WHERE tenant_id = ? AND subject_id = ? AND tool_name = ? AND idempotency_key = ? """, self._scope(principal, call), ).fetchone() if downstream is None: connection.commit() return None if ( downstream["request_sha256"] != digest or downstream["producer_revision"] != spec.producer_revision or downstream["effect_revision"] != spec.effect_revision ): raise PersistenceContractError("downstream receipt is inconsistent with the operation/contract") data = _decode_canonical_object( downstream["result_json"], label="downstream_receipts.result_json" ) raw = tool_v2.HandlerResult( data=data, producer_revision=downstream["producer_revision"], resource_revision=downstream["resource_revision"], observed_at=downstream["observed_at"], downstream_request_id=downstream["downstream_request_id"], receipt_id=downstream["receipt_id"], ) errors = tool_v2.validate_handler_result(spec, call, raw) if errors: raise PersistenceContractError("downstream receipt failed the output contract: " + "; ".join(errors)) connection.execute( """ INSERT INTO local_receipts( operation_pk, receipt_id, downstream_request_id, request_sha256, result_json, producer_revision, resource_revision, observed_at, effect_revision, reconciled_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(operation_pk) DO NOTHING """, ( operation_pk, downstream["receipt_id"], downstream["downstream_request_id"], digest, downstream["result_json"], downstream["producer_revision"], downstream["resource_revision"], downstream["observed_at"], downstream["effect_revision"], now, ), ) local = connection.execute( "SELECT * FROM local_receipts WHERE operation_pk = ?", (operation_pk,) ).fetchone() if local is None: raise PersistenceContractError("local receipt is not visible after UPSERT") for field in ( "receipt_id", "downstream_request_id", "request_sha256", "result_json", "producer_revision", "resource_revision", "observed_at", "effect_revision", ): if local[field] != downstream[field]: raise PersistenceContractError("local/downstream receipts are inconsistent") connection.execute( """ UPDATE operations SET state = 'succeeded', updated_at = ?, completed_at = ? WHERE operation_pk = ? """, (now, now, operation_pk), ) connection.execute( """ UPDATE outbox SET state = 'delivered', lease_owner = NULL, lease_until = NULL, updated_at = ? WHERE operation_pk = ? """, (now, operation_pk), ) connection.commit() except BaseException: if connection.in_transaction: connection.rollback() raise finally: connection.close() return self._load_local_record(operation_pk, principal, call, spec, digest) def process_operation( self, status_ref: str, *, worker_id: str, now: int, lease_seconds: int = DEFAULT_LEASE_SECONDS, failure: str = "none", ) -> str: """Claim, perform the downstream stand-in, and reconcile the receipt.""" if failure not in {"none", "after_claim", "after_downstream_commit"}: raise ValueError("worker failure is invalid") claim = self.claim_operation( status_ref, worker_id=worker_id, now=now, lease_seconds=lease_seconds, ) if claim is None: row = self._read_operation_by_status(status_ref) if row is not None and row["state"] == "succeeded": return "succeeded" return "not_claimed" if failure == "after_claim": return "crashed_after_claim" with self._connect() as connection: operation = connection.execute( "SELECT * FROM operations WHERE operation_pk = ?", (claim.operation_pk,) ).fetchone() if operation is None: raise PersistenceContractError("worker lost the operation") arguments = _decode_canonical_object( operation["arguments_json"], label="operations.arguments_json" ) order_ref = arguments.get("order_ref") if not isinstance(order_ref, str): raise PersistenceContractError("operation arguments are missing order_ref") principal = self._current_authorized_principal( tool_v2.Principal(operation["tenant_id"], operation["subject_id"], ()), order_ref, ) if principal is None: return "authorization_denied" spec = self.contract.registry[operation["tool_name"]] call = tool_v2.ToolCall( operation["first_call_id"], operation["first_operation_id"], operation["tool_name"], arguments, operation["idempotency_key"], provider=operation["first_provider"], api_family=operation["first_api_family"], response_id=operation["first_response_id"], adapter_revision=operation["first_adapter_revision"], ) current_digest = tool_v2.request_digest(principal, call, spec) self._validate_operation_row( operation, principal, call, spec, current_digest ) if not spec.business_validator(principal, arguments): # The approved intent is durable, but a current business-state change # must fail closed before the downstream effect. return "business_rule_denied" self._commit_downstream(claim, now=now) if failure == "after_downstream_commit": return "outcome_unknown" record = self._reconcile( claim.operation_pk, principal, call, spec, current_digest, now=now, ) if record is None: raise PersistenceContractError("could not reconcile after the downstream commit") return "succeeded" def query_operation_status( self, principal: tool_v2.Principal, call: tool_v2.ToolCall, *, status_ref: str, expected_request_sha256: str, now: int, ) -> dict[str, Any]: """Reauthorize and reconcile one opaque status reference explicitly.""" spec, digest, preflight_error = self._preflight(principal, call, now=now) if preflight_error is not None: return preflight_error if spec is None: raise PersistenceContractError("internal error: status query is missing its spec") if STATUS_REF.fullmatch(status_ref) is None: return tool_v2._error_package( principal, call, spec, "NOT_FOUND", digest, now=now ) operation = self._read_operation_by_status(status_ref) if operation is None or ( operation["tenant_id"], operation["subject_id"] ) != (principal.tenant_id, principal.subject_id): return tool_v2._error_package( principal, call, spec, "NOT_FOUND", digest, now=now ) connection = self._connect() try: self._begin_immediate(connection) call_bound = self._bind_call_identity( connection, principal, call, spec, purpose="query_status", now=now, ) connection.commit() except BaseException: if connection.in_transaction: connection.rollback() raise finally: connection.close() if not call_bound: return tool_v2._error_package( principal, call, spec, "CALL_ID_CONFLICT", digest, now=now ) if ( not isinstance(expected_request_sha256, str) or HEX64.fullmatch(expected_request_sha256) is None or expected_request_sha256 != digest or operation["status_ref"] != status_ref or tuple( operation[field] for field in ("tenant_id", "subject_id", "tool_name", "idempotency_key") ) != self._scope(principal, call) or not self._contract_matches(operation, spec, digest) or not self._approval_provider_context_matches(operation, call) ): return tool_v2._error_package( principal, call, spec, "STATUS_CONFLICT", digest, now=now, status_ref=status_ref, ) self._validate_operation_row(operation, principal, call, spec, digest) if operation["state"] == "succeeded": record = self._load_local_record( operation["operation_pk"], principal, call, spec, digest ) return tool_v2._success_package( principal, call, spec, record, delivery="receipt_reconciled" ) record = self._reconcile( operation["operation_pk"], principal, call, spec, digest, now=now ) if record is None: return self._unknown_package( principal, call, spec, digest, now=now, status_ref=status_ref, ) return tool_v2._success_package( principal, call, spec, record, delivery="receipt_reconciled" ) @staticmethod def _counts_from_connection(connection: sqlite3.Connection) -> dict[str, int]: tables = ( "call_bindings", "operations", "outbox", "downstream_receipts", "local_receipts", ) return { table: int(connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]) for table in tables } def counts(self) -> dict[str, int]: with self._connect() as connection: connection.execute("BEGIN") result = self._counts_from_connection(connection) connection.commit() return result @staticmethod def _semantic_rows( connection: sqlite3.Connection, ) -> tuple[ list[sqlite3.Row], list[sqlite3.Row], list[sqlite3.Row], list[sqlite3.Row], ]: return ( list(connection.execute("SELECT * FROM operations ORDER BY operation_pk")), list(connection.execute("SELECT * FROM outbox ORDER BY operation_pk")), list( connection.execute( """ SELECT * FROM downstream_receipts ORDER BY tenant_id, subject_id, tool_name, idempotency_key """ ) ), list(connection.execute("SELECT * FROM local_receipts ORDER BY operation_pk")), ) def _semantic_audit( self, connection: sqlite3.Connection | None = None ) -> list[str]: """Recompute stored identities and receipt bindings, not just page health.""" if connection is None: with self._connect() as owned_connection: owned_connection.execute("BEGIN") rows = self._semantic_rows(owned_connection) owned_connection.commit() else: rows = self._semantic_rows(connection) operations, outbox_rows, downstream_rows, local_rows = rows errors: list[str] = [] outbox_by_operation = {row["operation_pk"]: row for row in outbox_rows} local_by_operation = {row["operation_pk"]: row for row in local_rows} downstream_by_scope = { tuple( row[field] for field in ("tenant_id", "subject_id", "tool_name", "idempotency_key") ): row for row in downstream_rows } operation_ids = {row["operation_pk"] for row in operations} operation_scopes: set[tuple[str, str, str, str]] = set() for operation in operations: operation_pk = operation["operation_pk"] label = f"operation[{operation_pk}]" scope = tuple( operation[field] for field in ("tenant_id", "subject_id", "tool_name", "idempotency_key") ) operation_scopes.add(scope) try: arguments = _decode_canonical_object( operation["arguments_json"], label=f"{label}.arguments_json" ) spec = self.contract.registry.get(operation["tool_name"]) if spec is None or operation["tool_name"] != "create_refund_draft": raise PersistenceContractError(f"{label} tool is not registered") actor = tool_v2.Principal(operation["tenant_id"], operation["subject_id"], ()) call = tool_v2.ToolCall( operation["first_call_id"], operation["first_operation_id"], operation["tool_name"], arguments, operation["idempotency_key"], provider=operation["first_provider"], api_family=operation["first_api_family"], response_id=operation["first_response_id"], adapter_revision=operation["first_adapter_revision"], ) digest = tool_v2.request_digest(actor, call, spec) if not self._contract_matches(operation, spec, digest): raise PersistenceContractError(f"{label} request/contract binding does not match") if operation["status_ref"] != self._status_ref(scope, digest): raise PersistenceContractError(f"{label} status_ref does not match") self._validate_operation_row(operation, actor, call, spec, digest) outbox = outbox_by_operation.get(operation_pk) if outbox is None: raise PersistenceContractError(f"{label} is missing an outbox record") payload = _decode_canonical_object( outbox["payload_json"], label=f"outbox[{operation_pk}].payload_json", fields={"schema_version", "operation_pk", "request_sha256", "tool"}, ) if payload != self._event_payload(operation_pk, digest, operation["tool_name"]): raise PersistenceContractError(f"outbox[{operation_pk}] is inconsistent with the operation") local = local_by_operation.get(operation_pk) downstream = downstream_by_scope.get(scope) if operation["state"] == "succeeded": if outbox["state"] != "delivered" or local is None or downstream is None: raise PersistenceContractError(f"{label} succeeded-state evidence is incomplete") elif local is not None or outbox["state"] == "delivered": raise PersistenceContractError(f"{label} has delivered/local receipt despite not succeeding") if downstream is not None: if ( downstream["request_sha256"] != digest or downstream["producer_revision"] != spec.producer_revision or downstream["effect_revision"] != spec.effect_revision ): raise PersistenceContractError(f"{label} downstream receipt binding does not match") downstream_data = _decode_canonical_object( downstream["result_json"], label=f"downstream_receipt[{operation_pk}].result_json", ) raw = tool_v2.HandlerResult( data=downstream_data, producer_revision=downstream["producer_revision"], resource_revision=downstream["resource_revision"], observed_at=downstream["observed_at"], downstream_request_id=downstream["downstream_request_id"], receipt_id=downstream["receipt_id"], ) output_errors = tool_v2.validate_handler_result(spec, call, raw) if output_errors: raise PersistenceContractError( f"{label} downstream output contract failed: " + "; ".join(output_errors) ) if local is not None: if downstream is None: raise PersistenceContractError(f"{label} local receipt lacks downstream evidence") for field in ( "receipt_id", "downstream_request_id", "request_sha256", "result_json", "producer_revision", "resource_revision", "observed_at", "effect_revision", ): if local[field] != downstream[field]: raise PersistenceContractError(f"{label} local/downstream receipts are inconsistent") except (KeyError, TypeError, ValueError, PersistenceContractError) as exc: errors.append(str(exc)) for operation_pk in sorted(set(outbox_by_operation) - operation_ids): errors.append(f"outbox[{operation_pk}] has no operation") for operation_pk in sorted(set(local_by_operation) - operation_ids): errors.append(f"local_receipt[{operation_pk}] has no operation") for scope in sorted(set(downstream_by_scope) - operation_scopes): # Audit output can cross an operator/CLI boundary. Keep the # orphan diagnosable without exposing tenant, subject, tool or the # idempotency key stored in the downstream scope. receipt = downstream_by_scope[scope] audit_ref = self._status_ref(scope, receipt["request_sha256"]) errors.append(f"AUDIT_ORPHAN_DOWNSTREAM_RECEIPT[{audit_ref}]") return errors def audit_database(self) -> dict[str, Any]: with self._connect() as connection: # One explicit read transaction gives integrity/FK, semantic rows, # and counts the same SQLite snapshot. Without it, a legal # reconcile commit between SELECTs could look like corruption. connection.execute("BEGIN") integrity = [row[0] for row in connection.execute("PRAGMA integrity_check")] foreign_keys = [list(row) for row in connection.execute("PRAGMA foreign_key_check")] journal_mode = str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower() synchronous = int(connection.execute("PRAGMA synchronous").fetchone()[0]) semantic_errors = self._semantic_audit(connection) counts = self._counts_from_connection(connection) connection.commit() return { "schema_version": DATABASE_SCHEMA_VERSION, "sqlite_version": sqlite3.sqlite_version, "journal_mode": journal_mode, "synchronous": synchronous, "integrity_check": integrity, "foreign_key_violations": foreign_keys, "semantic_errors": semantic_errors, "counts": counts, "passed": ( integrity == ["ok"] and not foreign_keys and not semantic_errors and journal_mode == "wal" and synchronous == 2 ), } def _result_code(package: dict[str, Any]) -> str: model = package["model_result"] if model["status"] == "succeeded": return "OK" return model["error"]["code"] def _cli_package_summary( package: dict[str, Any], principal: tool_v2.Principal, call: tool_v2.ToolCall, spec: tool_v2.ToolSpec, runtime: PersistentToolRuntime, ) -> dict[str, Any]: errors = tool_v2.validate_result(principal, call, spec, package) model = package["model_result"] status_ref = package["protected_audit"]["downstream"]["status_ref"] gate = "PASS" if model["status"] == "succeeded" and not errors else "BLOCK" return { "gate": gate, "code": _result_code(package), "status": model["status"], "delivery": model["execution"]["delivery"], "status_ref": status_ref, "validation_errors": errors, "database": runtime.audit_database(), } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--db", type=Path, required=True, help="SQLite database path") parser.add_argument( "--fixture", type=Path, default=Path(__file__).with_name("persistence-case.json"), help="strict JSON teaching scenario", ) subparsers = parser.add_subparsers(dest="command", required=True) dispatch = subparsers.add_parser("dispatch", help="reserve intent and optionally process the outbox") dispatch.add_argument("--failure", choices=sorted(VALID_FAILURES), default="none") dispatch.add_argument("--worker-id", default="cli-worker") status = subparsers.add_parser("status", help="explicitly query/reconcile an unknown operation") status.add_argument("--status-ref", required=True) subparsers.add_parser("audit", help="run SQLite integrity and foreign-key checks") return parser def _cli_error_code(exc: BaseException) -> str: """Map expected local failures to stable, path-free CLI diagnostics.""" if isinstance(exc, PersistenceFixtureIOError): return "FIXTURE_IO_ERROR" if isinstance(exc, PersistenceFixtureError): return "FIXTURE_CONTRACT_ERROR" if isinstance(exc, sqlite3.Error): return "SQLITE_ERROR" if isinstance(exc, PersistenceContractError): return "PERSISTENCE_CONTRACT_ERROR" if isinstance(exc, OSError): return "LOCAL_IO_ERROR" return "INVALID_VALUE" def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: runtime = PersistentToolRuntime(args.db) if args.command == "audit": summary = runtime.audit_database() summary["gate"] = "PASS" if summary["passed"] else "BLOCK" print(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True)) return 0 if summary["passed"] else 1 fixture = load_fixture(args.fixture) principal, call = fixture_principal_call(fixture) spec = runtime.contract.registry[call.name] now = fixture["now"] if args.command == "dispatch": approvals = tool_v2.make_approval( principal, call, spec, fixture["approval_mode"], now ) package = runtime.dispatch( principal, call, approvals=approvals, now=now, failure=args.failure, worker_id=args.worker_id, ) else: call = status_query_call(call, args.status_ref) package = runtime.query_operation_status( principal, call, status_ref=args.status_ref, expected_request_sha256=tool_v2.request_digest(principal, call, spec), now=now + 1, ) summary = _cli_package_summary(package, principal, call, spec, runtime) print(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True)) return 0 if summary["gate"] == "PASS" else 1 except ( OSError, sqlite3.Error, PersistenceContractError, PersistenceFixtureError, ValueError, ) as exc: print( json.dumps( {"gate": "BLOCK", "error": {"code": _cli_error_code(exc)}}, ensure_ascii=False, sort_keys=True, ), file=sys.stderr, ) return 2 if __name__ == "__main__": raise SystemExit(main())