project/workflow.py

"""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")