"""A bounded, deterministic teaching workflow. No model, network or filesystem I/O. This is a single-process rehearsal, not a production agent or durable queue. """ from dataclasses import dataclass from enum import Enum import hashlib import json import re class State(str, Enum): VALIDATE = "VALIDATE" LOOKUP = "LOOKUP" DRAFT = "DRAFT" class Status(str, Enum): AWAITING_REVIEW = "AWAITING_REVIEW" STOPPED = "STOPPED" @dataclass(frozen=True) class Request: request_id: str exhibit_id: str observed_year: int note: str = "" @dataclass(frozen=True) class Limits: max_steps: int = 6 max_tool_calls: int = 2 max_attempts: int = 2 def __post_init__(self): for name in ("max_steps", "max_tool_calls", "max_attempts"): value = getattr(self, name) minimum = 0 if name == "max_tool_calls" else 1 if type(value) is not int or value < minimum: raise ValueError(f"{name} must be an integer >= {minimum}") @dataclass(frozen=True) class Fact: exhibit_id: str catalogue_year: int source_id: str @dataclass(frozen=True) class Draft: exhibit_id: str observed_year: int catalogue_year: int finding: str recommendation: str source_id: str @dataclass(frozen=True) class TraceEvent: step: int state: State action: str @dataclass(frozen=True) class Result: request_id: str status: Status reason: str draft: Draft | None tool_calls: int trace: tuple[TraceEvent, ...] @dataclass(frozen=True) class Receipt: result: Result replayed: bool new_tool_calls: int class TransientToolError(Exception): """Known failure before this read-only synthetic tool returns a result.""" class UnknownExhibit(Exception): pass class RequestConflict(ValueError): """An existing request ID was reused with changed material input.""" class SyntheticCatalogue: """An in-memory fixture with a deliberate, deterministic failure switch. No message, user text or model can select another tool or catalogue. """ version = "museum-fixture-v1" def __init__(self, transient_failures: int = 0): if type(transient_failures) is not int or transient_failures < 0: raise ValueError("transient_failures must be a non-negative integer") self.failures_remaining = transient_failures self.calls = 0 def lookup(self, exhibit_id: str) -> Fact: self.calls += 1 if self.failures_remaining: self.failures_remaining -= 1 raise TransientToolError("synthetic temporary lookup failure") facts = { "EX-001": Fact("EX-001", 1924, "CAT-v1:EX-001"), "EX-002": Fact("EX-002", 1986, "CAT-v1:EX-002"), } if exhibit_id not in facts: raise UnknownExhibit(exhibit_id) return facts[exhibit_id] def input_error(request: Request) -> str | None: if not isinstance(request, Request): return "invalid_request_type" if not isinstance(request.request_id, str) or not re.fullmatch( r"[A-Z][A-Z0-9-]{0,39}", request.request_id ): return "invalid_request_id" if not isinstance(request.exhibit_id, str) or not re.fullmatch( r"EX-[0-9]{3}", request.exhibit_id ): return "invalid_exhibit_id" # bool is an int subclass; a checkbox is not an observed year. if type(request.observed_year) is not int or not 1000 <= request.observed_year <= 2100: return "invalid_observed_year" if not isinstance(request.note, str) or len(request.note) > 500: return "invalid_note" return None def fingerprint(request: Request, catalogue_version: str) -> str: payload = { "schema": "label-review-v1", "catalogue": catalogue_version, "exhibit_id": request.exhibit_id, "observed_year": request.observed_year, "note": request.note, } canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def valid_fact(fact, exhibit_id: str) -> bool: return ( isinstance(fact, Fact) and fact.exhibit_id == exhibit_id and type(fact.catalogue_year) is int and 1000 <= fact.catalogue_year <= 2100 and isinstance(fact.source_id, str) and fact.source_id == f"CAT-v1:{exhibit_id}" ) class Workflow: """Serial controller; all paths end with a draft for review or a safe stop. The ledger caches terminal outcomes only while this instance is alive. It is deliberately not a persistent or concurrent idempotency service. """ def __init__(self, tool: SyntheticCatalogue | None = None, limits: Limits | None = None): self.tool = tool if tool is not None else SyntheticCatalogue() self.limits = limits if limits is not None else Limits() self._ledger: dict[str, tuple[str, Result]] = {} def run(self, request: Request) -> Receipt: error = input_error(request) if error: # Invalid input is never registered: it cannot reserve an ID. result = Result("INVALID", Status.STOPPED, error, None, 0, (TraceEvent(1, State.VALIDATE, error),)) return Receipt(result, False, 0) digest = fingerprint(request, self.tool.version) if request.request_id in self._ledger: saved_digest, result = self._ledger[request.request_id] if saved_digest != digest: raise RequestConflict("request ID already belongs to different input") return Receipt(result, True, 0) result = self._execute(request) self._ledger[request.request_id] = (digest, result) return Receipt(result, False, result.tool_calls) def _execute(self, request: Request) -> Result: state = State.VALIDATE trace: list[TraceEvent] = [] tool_calls = 0 fact = None def finish(reason: str, draft: Draft | None = None) -> Result: status = Status.AWAITING_REVIEW if draft is not None else Status.STOPPED return Result(request.request_id, status, reason, draft, tool_calls, tuple(trace)) # A step is one visited work state. Each retry consumes another step. for step in range(1, self.limits.max_steps + 1): if state == State.VALIDATE: trace.append(TraceEvent(step, state, "input accepted")) state = State.LOOKUP elif state == State.LOOKUP: if tool_calls >= self.limits.max_tool_calls: trace.append(TraceEvent(step, state, "tool budget exhausted")) return finish("tool_budget_exhausted") tool_calls += 1 # Charge attempts, including failures. try: fact = self.tool.lookup(request.exhibit_id) except TransientToolError: trace.append(TraceEvent(step, state, "known transient read failure")) if tool_calls >= self.limits.max_attempts: return finish("retry_limit_exhausted") continue except UnknownExhibit: trace.append(TraceEvent(step, state, "exhibit absent from fixture")) return finish("unknown_exhibit") except Exception: # No retry for an unclassified failure; no exception text in the trace. trace.append(TraceEvent(step, state, "unclassified tool failure")) return finish("unclassified_tool_failure") if not valid_fact(fact, request.exhibit_id): trace.append(TraceEvent(step, state, "tool output rejected")) return finish("invalid_tool_output") trace.append(TraceEvent(step, state, "catalogue fact accepted")) state = State.DRAFT elif state == State.DRAFT: mismatch = request.observed_year != fact.catalogue_year draft = Draft( request.exhibit_id, request.observed_year, fact.catalogue_year, "MISMATCH" if mismatch else "MATCH", "Review the source before changing the label." if mismatch else "Record the comparison; no change proposed.", fact.source_id, ) trace.append(TraceEvent(step, state, "draft prepared; human boundary reached")) return finish("review_required", draft) return finish("step_budget_exhausted")