"""Modules 2-10: deterministic planners, checkpoints and synthetic human review. SQLite lives in RAM. Exports simulate transfer to a restarted process; no real record, model, account, credential, network or external action is involved. """ from dataclasses import asdict, dataclass import hashlib import json import sqlite3 from typing import Protocol from types import MappingProxyType from workflow import Request, RequestConflict, input_error def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() class TransitionError(ValueError): pass class StaleRevision(TransitionError): pass class ReviewError(TransitionError): pass class SimulatedCrash(RuntimeError): pass @dataclass(frozen=True) class Policy: max_steps: int = 16 max_tool_calls: int = 3 max_failures: int = 2 review_ttl: int = 3 def __post_init__(self): for name, value in asdict(self).items(): minimum = 0 if name == "max_tool_calls" else 1 if type(value) is not int or value < minimum: raise ValueError(f"invalid {name}") @dataclass(frozen=True) class Proposal: name: str exhibit_id: str | None = None @dataclass(frozen=True) class View: exhibit_id: str observed_year: int has_fact: bool tool_calls_left: int source_version: str # Raw note and review authority are intentionally absent. class Planner(Protocol): def propose(self, view: View) -> Proposal: ... class BaselinePlanner: """Fixed routing, not a model agent.""" def propose(self, view): return Proposal("draft") if view.has_fact else Proposal("lookup_catalogue", view.exhibit_id) class ScriptedPlanner: """A deterministic test double: emits one supplied proposal every time.""" def __init__(self, proposal): self.proposal = proposal def propose(self, view): return self.proposal TOOL_REGISTRY = MappingProxyType({ "lookup_catalogue": {"authority": "synthetic_read", "argument": "requested exhibit only"}, "draft": {"authority": "local_draft", "argument": "none"}, }) class SyntheticTool: def __init__(self, version="CAT-v1", failures=0, wrong_output=False): if version not in ("CAT-v1", "CAT-v2"): raise ValueError("unknown fixture version") if type(failures) is not int or failures < 0: raise ValueError("failures must be a nonnegative integer") self.version, self.failures, self.wrong_output = version, failures, wrong_output self.calls = 0 def lookup(self, exhibit_id): self.calls += 1 if self.failures: self.failures -= 1 return {"outcome": "transient"} years = {"EX-001": 1924 if self.version == "CAT-v1" else 1925, "EX-002": 1986} if exhibit_id not in years: return {"outcome": "unknown"} return {"outcome": "ok", "exhibit_id": "EX-999" if self.wrong_output else exhibit_id, "year": years[exhibit_id], "source_version": self.version, "source_id": f"{self.version}:{exhibit_id}"} class Checkpoints: """Transactional snapshot storage; one SQLite connection, single-thread lab. JSON export/import is a deliberate restart simulation, not a disk backup. Only import course-produced snapshots; this is not an untrusted-file parser. """ def __init__(self): self.connection = sqlite3.connect(":memory:") self.connection.execute("CREATE TABLE runs (id TEXT PRIMARY KEY, revision INTEGER, payload TEXT)") def close(self): self.connection.close() def get(self, run_id): row = self.connection.execute("SELECT payload FROM runs WHERE id=?", (run_id,)).fetchone() if row is None: raise KeyError(run_id) return json.loads(row[0]) # Each caller receives a detached copy. def insert(self, state): with self.connection: self.connection.execute("INSERT INTO runs VALUES (?, ?, ?)", (state["id"], state["revision"], json.dumps(state))) def save(self, state, expected_revision): if state["revision"] != expected_revision + 1: raise StaleRevision("revision must increase by exactly one") with self.connection: result = self.connection.execute( "UPDATE runs SET revision=?, payload=? WHERE id=? AND revision=?", (state["revision"], json.dumps(state), state["id"], expected_revision), ) if result.rowcount != 1: raise StaleRevision("checkpoint changed; reload before deciding") return self.get(state["id"]) def export(self): rows = self.connection.execute("SELECT payload FROM runs ORDER BY id").fetchall() return json.dumps({"schema": "museum-checkpoint-v1", "runs": [json.loads(r[0]) for r in rows]}) @classmethod def restore(cls, payload): data = json.loads(payload) if data.get("schema") != "museum-checkpoint-v1": raise ValueError("unsupported checkpoint schema") store = cls() try: for state in data["runs"]: store.insert(state) except Exception: store.close() raise return store PHASES = { "ACTIVE": {"ACTIVE", "RESERVED", "REVIEW", "STOPPED"}, "RESERVED": {"IN_FLIGHT", "STOPPED"}, "IN_FLIGHT": {"ACTIVE", "STOPPED"}, "REVIEW": {"APPROVED_FOR_SIMULATION", "REJECTED", "STOPPED"}, "APPROVED_FOR_SIMULATION": {"SIMULATED", "STOPPED"}, "REJECTED": set(), "SIMULATED": set(), "STOPPED": set(), } class Engine: def __init__(self, store, tool=None, policy=None): self.store = store self.tool = tool if tool is not None else SyntheticTool() self.policy = policy if policy is not None else Policy() def start(self, request): error = input_error(request) if error: raise ValueError(error) identity = digest({"request": asdict(request), "source": self.tool.version, "policy": asdict(self.policy), "schema": "museum-run-v1"}) try: state = self.store.get(request.request_id) except KeyError: state = {"id": request.request_id, "input": asdict(request), "identity": identity, "source_version": self.tool.version, "policy": asdict(self.policy), "phase": "ACTIVE", "revision": 0, "steps": 0, "tool_calls": 0, "failures": 0, "pending": None, "fact": None, "draft": None, "draft_digest": None, "draft_created_tick": None, "deadline": None, "review": None, "reason": "started", "events": [], "receipt": None} self.store.insert(state) return state if state["identity"] != identity: raise RequestConflict("existing ID has different input, source or policy") return state def _save(self, state, phase, reason, work=True): old_phase, revision = state["phase"], state["revision"] if phase not in PHASES[old_phase]: raise TransitionError(f"illegal transition {old_phase} -> {phase}") state["phase"], state["reason"] = phase, reason state["revision"] += 1 state["steps"] += int(work) state["events"].append({"revision": state["revision"], "from": old_phase, "to": phase, "reason": reason}) return self.store.save(state, revision) def _guard(self, state): if state["policy"] != asdict(self.policy): raise TransitionError("resume must use the saved policy") if state["steps"] >= self.policy.max_steps: return self._save(state, "STOPPED", "step_budget_exhausted", work=False) if state["source_version"] != self.tool.version: return self._save(state, "STOPPED", "source_changed", work=False) return None def plan(self, run_id, planner, now_tick=0): if type(now_tick) is not int or now_tick < 0: raise ValueError("logical time must be a nonnegative integer") state = self.store.get(run_id) if state["phase"] != "ACTIVE": raise TransitionError("planning requires ACTIVE") guarded = self._guard(state) if guarded: return guarded view = View(state["input"]["exhibit_id"], state["input"]["observed_year"], state["fact"] is not None, self.policy.max_tool_calls - state["tool_calls"], state["source_version"]) try: proposal = planner.propose(view) except Exception: return self._save(state, "STOPPED", "planner_failure") if (not isinstance(proposal, Proposal) or type(proposal.name) is not str or proposal.name not in TOOL_REGISTRY): return self._save(state, "STOPPED", "unsupported_proposal") if proposal.name == "lookup_catalogue": if proposal.exhibit_id != view.exhibit_id: return self._save(state, "STOPPED", "wrong_tool_target") if state["tool_calls"] >= self.policy.max_tool_calls: return self._save(state, "STOPPED", "tool_budget_exhausted") state["tool_calls"] += 1 # Charge before invoking; abandoned reservations still count. state["pending"] = {"operation_id": f"{run_id}:{state['tool_calls']}", "exhibit_id": proposal.exhibit_id} return self._save(state, "RESERVED", "read_attempt_reserved") if proposal.exhibit_id is not None: return self._save(state, "STOPPED", "unexpected_draft_argument") if state["fact"] is None: return self._save(state, "STOPPED", "draft_without_evidence") fact, observed = state["fact"], view.observed_year state["draft"] = {"exhibit_id": view.exhibit_id, "observed_year": observed, "catalogue_year": fact["year"], "source_id": fact["source_id"], "source_version": fact["source_version"], "finding": "MATCH" if observed == fact["year"] else "MISMATCH", "proposed_action": "record_comparison" if observed == fact["year"] else "review_possible_label_correction"} state["draft_digest"] = digest({"identity": state["identity"], "draft": state["draft"]}) state["draft_created_tick"] = now_tick state["deadline"] = now_tick + self.policy.review_ttl return self._save(state, "REVIEW", "human_review_required") def invoke(self, run_id, crash_after_read=False): state = self.store.get(run_id) if state["phase"] != "RESERVED": raise TransitionError("invoke only a RESERVED attempt; recover IN_FLIGHT explicitly") guarded = self._guard(state) if guarded: return guarded state = self._save(state, "IN_FLIGHT", "read_invocation_started") # The completed read also needs a checkpoint step. Stop before a call if absent. if state["steps"] >= self.policy.max_steps: return self._save(state, "STOPPED", "step_budget_exhausted", work=False) try: reply = self.tool.lookup(state["pending"]["exhibit_id"]) except Exception: return self._save(state, "STOPPED", "unclassified_tool_failure") if crash_after_read: raise SimulatedCrash("read returned, but result checkpoint was not recorded") if not isinstance(reply, dict): return self._save(state, "STOPPED", "invalid_tool_output") if reply.get("outcome") == "transient": state["failures"] += 1 state["pending"] = None if state["failures"] >= self.policy.max_failures: return self._save(state, "STOPPED", "retry_limit_exhausted") return self._save(state, "ACTIVE", "known_transient_read_failure") if reply.get("outcome") == "unknown": return self._save(state, "STOPPED", "unknown_exhibit") expected_id = state["input"]["exhibit_id"] if not (set(reply) == {"outcome", "exhibit_id", "year", "source_version", "source_id"} and reply.get("outcome") == "ok" and reply.get("exhibit_id") == expected_id and type(reply.get("year")) is int and 1000 <= reply["year"] <= 2100 and reply.get("source_version") == state["source_version"] and reply.get("source_id") == f"{state['source_version']}:{expected_id}"): return self._save(state, "STOPPED", "invalid_tool_output") state["fact"], state["pending"], state["failures"] = reply, None, 0 return self._save(state, "ACTIVE", "validated_fact_recorded") def recover(self, run_id, abandon_read=False): state = self.store.get(run_id) if state["phase"] != "IN_FLIGHT": raise TransitionError("recovery requires IN_FLIGHT") guarded = self._guard(state) if guarded: return guarded if not abandon_read: return self._save(state, "STOPPED", "uncertain_tool_outcome") state["pending"] = None return self._save(state, "ACTIVE", "read_reservation_abandoned") def run_until_pause(self, run_id, planner=None, now_tick=0): planner = planner if planner is not None else BaselinePlanner() for _ in range(self.policy.max_steps + 1): state = self.store.get(run_id) if state["phase"] == "ACTIVE": state = self.plan(run_id, planner, now_tick) elif state["phase"] == "RESERVED": state = self.invoke(run_id) else: return state if state["phase"] not in {"ACTIVE", "RESERVED"}: return state raise AssertionError("controller failed to reach a pause within the explicit bound") def review(self, run_id, expected_revision, draft_digest, reviewer, approve, now_tick): state = self.store.get(run_id) if state["revision"] != expected_revision: raise StaleRevision("review is stale") if state["phase"] != "REVIEW": raise ReviewError("review requires REVIEW") if reviewer not in {"synthetic-reviewer-a"} or type(approve) is not bool: raise ReviewError("invalid synthetic reviewer or decision") if draft_digest != state["draft_digest"]: raise ReviewError("decision is not bound to this draft") guarded = self._guard(state) if guarded: return guarded if type(now_tick) is not int or now_tick < 0: raise ReviewError("logical time must be a nonnegative integer") if now_tick < state["draft_created_tick"]: raise ReviewError("decision time precedes draft creation") if now_tick > state["deadline"]: return self._save(state, "STOPPED", "approval_expired") state["review"] = {"reviewer": reviewer, "digest": draft_digest, "tick": now_tick, "approved": approve} return self._save(state, "APPROVED_FOR_SIMULATION" if approve else "REJECTED", "synthetic_review_approved" if approve else "synthetic_review_rejected") def finalize_simulation(self, run_id, now_tick): state = self.store.get(run_id) if state["phase"] == "SIMULATED": return state # Saved receipt; no new event or action. if state["phase"] != "APPROVED_FOR_SIMULATION": raise ReviewError("a synthetic approval is required") if type(now_tick) is not int or now_tick < state["review"]["tick"]: raise ReviewError("finalization time precedes the decision") guarded = self._guard(state) if guarded: return guarded if now_tick > state["deadline"]: return self._save(state, "STOPPED", "approval_expired_before_simulation") current = digest({"identity": state["identity"], "draft": state["draft"]}) if current != state["review"]["digest"]: return self._save(state, "STOPPED", "draft_changed_after_review") state["receipt"] = {"kind": "SIMULATION_ONLY", "draft_digest": current, "would_change": state["draft"]["finding"] == "MISMATCH", "real_writes": 0} return self._save(state, "SIMULATED", "no_real_record_changed")