All courses · Course overview · Lesson sequence
Skip to lesson textproject/engine.py
"""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")