"""Durable, fail-closed AINA Petri-dish process state machine.

The controller in this module is an orchestration boundary, not a motion API.
It accepts one injected, typed operations object and serializes every physical
step through that object.  It never imports the xArm SDK, camera drivers, the
sampling-tool SDK boundary, or a network client.

The SQLite ledger deliberately separates the physical result from upload:
``LOCAL_COMMIT`` must return a durable evidence receipt before ``QUEUE_UPLOAD``
may enqueue that receipt.  A network worker can later consume the queue, but it
cannot change the already committed physical result.
"""

from __future__ import annotations

import hashlib
import json
import math
import os
import re
import sqlite3
import stat
import threading
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from types import MappingProxyType
from typing import Any, Callable, Mapping, Optional, Sequence, Tuple

IDLE = "IDLE"
PREFLIGHT = "PREFLIGHT"
AUTHORIZED = "AUTHORIZED"
HOME = "HOME"
LOCATE_TRAY = "LOCATE_TRAY"
IDENTIFY_PLATES = "IDENTIFY_PLATES"
APPROACH = "APPROACH"
PICK_LID = "PICK_LID"
VERIFY_GRASP = "VERIFY_GRASP"
CAPTURE = "CAPTURE"
RETURN_LID = "RETURN_LID"
VERIFY_RELEASE = "VERIFY_RELEASE"
NEXT_PLATE = "NEXT_PLATE"
RETURN_HOME = "RETURN_HOME"
LOCAL_COMMIT = "LOCAL_COMMIT"
QUEUE_UPLOAD = "QUEUE_UPLOAD"
COMPLETE = "COMPLETE"
STOPPING = "STOPPING"
CANCELLED = "CANCELLED"
FAULT_LOCKED = "FAULT_LOCKED"
RECOVERY_REQUIRED = "RECOVERY_REQUIRED"

PROCESS_STATES = frozenset(
    {
        IDLE,
        PREFLIGHT,
        AUTHORIZED,
        HOME,
        LOCATE_TRAY,
        IDENTIFY_PLATES,
        APPROACH,
        PICK_LID,
        VERIFY_GRASP,
        CAPTURE,
        RETURN_LID,
        VERIFY_RELEASE,
        NEXT_PLATE,
        RETURN_HOME,
        LOCAL_COMMIT,
        QUEUE_UPLOAD,
        COMPLETE,
        STOPPING,
        CANCELLED,
        FAULT_LOCKED,
        RECOVERY_REQUIRED,
    }
)

_MOTION_OR_TOOL_STATES = frozenset(
    {
        AUTHORIZED,
        HOME,
        APPROACH,
        PICK_LID,
        VERIFY_GRASP,
        CAPTURE,
        RETURN_LID,
        VERIFY_RELEASE,
        NEXT_PLATE,
        RETURN_HOME,
    }
)
_OPERATION_STATES = (
    PREFLIGHT,
    AUTHORIZED,
    HOME,
    LOCATE_TRAY,
    IDENTIFY_PLATES,
    APPROACH,
    PICK_LID,
    VERIFY_GRASP,
    CAPTURE,
    RETURN_LID,
    VERIFY_RELEASE,
    NEXT_PLATE,
    RETURN_HOME,
    LOCAL_COMMIT,
    QUEUE_UPLOAD,
)
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:+-]{0,95}$")
_PLATE_ID = re.compile(r"^[A-Z0-9][A-Z0-9._-]{2,63}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_HEX32 = re.compile(r"^[0-9a-f]{32}$")
_MAX_EVIDENCE_BYTES = 64 * 1024


class PetriProcessError(RuntimeError):
    """Base class for bounded process-state rejections."""


class PetriProcessBusy(PetriProcessError):
    """Raised when another physical or recovery-owned run is active."""


class PetriProcessFault(PetriProcessError):
    """One state operation violated its deterministic contract."""

    def __init__(self, code: str, state: str) -> None:
        self.code = _error_code(code)
        self.state = state
        super().__init__("{}: {}".format(self.code, state))


class PetriProcessCancelled(PetriProcessError):
    """Raised internally after a STOP/cancel fence is observed."""


@dataclass(frozen=True)
class PetriProcessRecipe:
    """Immutable recipe/profile binding and per-state timeout policy."""

    recipe_id: str
    recipe_version: str
    process_profile_sha256: str
    state_timeouts_seconds: Mapping[str, float]

    @classmethod
    def create(
        cls,
        *,
        recipe_id: str,
        recipe_version: str,
        process_profile_sha256: str,
        state_timeouts_seconds: Mapping[str, Any],
    ) -> "PetriProcessRecipe":
        identifiers = []
        for value, name in (
            (recipe_id, "recipe_id"),
            (recipe_version, "recipe_version"),
        ):
            if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value):
                raise ValueError("{} is invalid".format(name))
            identifiers.append(value)
        if not isinstance(process_profile_sha256, str) or not _SHA256.fullmatch(
            process_profile_sha256
        ):
            raise ValueError("process_profile_sha256 is invalid")
        if not isinstance(state_timeouts_seconds, Mapping) or set(
            state_timeouts_seconds
        ) != set(_OPERATION_STATES):
            raise ValueError("timeouts must cover every operation state exactly")
        timeouts = {}
        for state in _OPERATION_STATES:
            value = state_timeouts_seconds[state]
            if (
                isinstance(value, bool)
                or not isinstance(value, (int, float))
                or not math.isfinite(float(value))
                or not 0.05 <= float(value) <= 3_600.0
            ):
                raise ValueError("{} timeout is invalid".format(state))
            timeouts[state] = float(value)
        return cls(
            recipe_id=identifiers[0],
            recipe_version=identifiers[1],
            process_profile_sha256=process_profile_sha256,
            state_timeouts_seconds=MappingProxyType(timeouts),
        )


@dataclass(frozen=True)
class PetriProcessSnapshot:
    run_id: str
    start_request_id: str
    state: str
    active: bool
    recipe_id: str
    recipe_version: str
    process_profile_sha256: str
    expected_plate_ids: Tuple[str, ...]
    identified_plate_ids: Tuple[str, ...]
    current_plate_index: int
    current_plate_id: str
    transition_sequence: int
    local_evidence_committed: bool
    local_commit_id: str
    local_evidence_sha256: str
    upload_queued: bool
    upload_queue_id: str
    recovery_required: bool
    error_code: str
    started_at: str
    updated_at: str


class DurablePetriProcessStore:
    """SQLite ledger with one active physical/recovery run per workcell."""

    def __init__(self, path: os.PathLike[str] | str) -> None:
        self.path = Path(path)
        if not self.path.is_absolute():
            raise ValueError("Petri process ledger path must be absolute")
        if self.path.exists():
            metadata = self.path.lstat()
            if (
                stat.S_ISLNK(metadata.st_mode)
                or not stat.S_ISREG(metadata.st_mode)
                or metadata.st_nlink != 1
            ):
                raise ValueError("Petri process ledger path is unsafe")
        self.path.parent.mkdir(parents=True, exist_ok=True)
        if self.path.parent.is_symlink() or not self.path.parent.is_dir():
            raise ValueError("Petri process ledger directory is unsafe")
        self._lock = threading.RLock()
        self._initialize()
        metadata = self.path.lstat()
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
            raise ValueError("Petri process ledger is unsafe")
        self._reconcile_interrupted_run()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(str(self.path), timeout=10.0)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON")
        connection.execute("PRAGMA busy_timeout = 10000")
        connection.execute("PRAGMA synchronous = FULL")
        return connection

    def _initialize(self) -> None:
        with self._lock, self._connect() as connection:
            connection.executescript("""
                PRAGMA journal_mode = WAL;
                CREATE TABLE IF NOT EXISTS petri_process_runs (
                    run_id TEXT PRIMARY KEY,
                    start_request_id TEXT NOT NULL UNIQUE,
                    state TEXT NOT NULL CHECK (state IN (
                        'IDLE', 'PREFLIGHT', 'AUTHORIZED', 'HOME',
                        'LOCATE_TRAY', 'IDENTIFY_PLATES', 'APPROACH',
                        'PICK_LID', 'VERIFY_GRASP', 'CAPTURE', 'RETURN_LID',
                        'VERIFY_RELEASE', 'NEXT_PLATE', 'RETURN_HOME',
                        'LOCAL_COMMIT', 'QUEUE_UPLOAD', 'COMPLETE', 'STOPPING',
                        'CANCELLED', 'FAULT_LOCKED', 'RECOVERY_REQUIRED'
                    )),
                    active INTEGER NOT NULL CHECK (active IN (0, 1)),
                    recipe_id TEXT NOT NULL,
                    recipe_version TEXT NOT NULL,
                    process_profile_sha256 TEXT NOT NULL,
                    expected_plate_ids_json TEXT NOT NULL,
                    identified_plate_ids_json TEXT NOT NULL DEFAULT '[]',
                    current_plate_index INTEGER NOT NULL DEFAULT 0
                        CHECK (current_plate_index >= 0),
                    current_plate_id TEXT NOT NULL DEFAULT '',
                    transition_sequence INTEGER NOT NULL DEFAULT 0
                        CHECK (transition_sequence >= 0),
                    local_commit_id TEXT NOT NULL DEFAULT '',
                    local_evidence_sha256 TEXT NOT NULL DEFAULT '',
                    upload_queue_id TEXT NOT NULL DEFAULT '',
                    recovery_required INTEGER NOT NULL DEFAULT 0
                        CHECK (recovery_required IN (0, 1)),
                    error_code TEXT NOT NULL DEFAULT '',
                    context_json TEXT NOT NULL DEFAULT '{}',
                    started_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );
                CREATE UNIQUE INDEX IF NOT EXISTS petri_one_active_run
                    ON petri_process_runs (active) WHERE active = 1;
                CREATE TABLE IF NOT EXISTS petri_process_transitions (
                    event_id TEXT PRIMARY KEY,
                    run_id TEXT NOT NULL,
                    sequence INTEGER NOT NULL CHECK (sequence > 0),
                    from_state TEXT NOT NULL,
                    to_state TEXT NOT NULL,
                    reason TEXT NOT NULL,
                    evidence_json TEXT NOT NULL,
                    recorded_at TEXT NOT NULL,
                    UNIQUE (run_id, sequence),
                    FOREIGN KEY (run_id) REFERENCES petri_process_runs (run_id)
                        ON DELETE RESTRICT
                );
                CREATE INDEX IF NOT EXISTS petri_transition_run_time
                    ON petri_process_transitions (run_id, recorded_at);
                CREATE TRIGGER IF NOT EXISTS petri_transition_no_update
                    BEFORE UPDATE ON petri_process_transitions
                    BEGIN
                        SELECT RAISE(ABORT, 'Petri transition ledger is append-only');
                    END;
                CREATE TRIGGER IF NOT EXISTS petri_transition_no_delete
                    BEFORE DELETE ON petri_process_transitions
                    BEGIN
                        SELECT RAISE(ABORT, 'Petri transition ledger is append-only');
                    END;
                CREATE TABLE IF NOT EXISTS petri_process_upload_outbox (
                    queue_id TEXT PRIMARY KEY,
                    run_id TEXT NOT NULL UNIQUE,
                    local_commit_id TEXT NOT NULL,
                    local_evidence_sha256 TEXT NOT NULL,
                    status TEXT NOT NULL DEFAULT 'pending'
                        CHECK (status IN ('pending', 'uploading', 'uploaded', 'failed')),
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY (run_id) REFERENCES petri_process_runs (run_id)
                        ON DELETE RESTRICT
                );
                CREATE INDEX IF NOT EXISTS petri_upload_outbox_status
                    ON petri_process_upload_outbox (status, created_at);
                """)
        try:
            os.chmod(self.path, 0o600)
        except OSError:
            pass

    def _reconcile_interrupted_run(self) -> None:
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT * FROM petri_process_runs WHERE active = 1"
            ).fetchone()
            if row is None:
                return
            if row["state"] == RECOVERY_REQUIRED:
                return
            self._transition_locked(
                connection,
                row,
                RECOVERY_REQUIRED,
                reason="runtime_restart",
                evidence={"interrupted_state": row["state"]},
                updates={
                    "active": 1,
                    "recovery_required": 1,
                    "error_code": "process_interrupted_by_restart",
                },
            )

    def create_or_replay(
        self,
        *,
        start_request_id: str,
        recipe: PetriProcessRecipe,
        expected_plate_ids: Sequence[str],
    ) -> tuple[PetriProcessSnapshot, bool]:
        request_id = _hex32(start_request_id, "start_request_id")
        plates = _plate_ids(expected_plate_ids)
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            replay = connection.execute(
                "SELECT * FROM petri_process_runs WHERE start_request_id = ?",
                (request_id,),
            ).fetchone()
            if replay is not None:
                if (
                    replay["recipe_id"] != recipe.recipe_id
                    or replay["recipe_version"] != recipe.recipe_version
                    or replay["process_profile_sha256"] != recipe.process_profile_sha256
                    or json.loads(replay["expected_plate_ids_json"]) != list(plates)
                ):
                    raise PetriProcessFault("start_request_conflict", IDLE)
                return _snapshot(replay), False
            active = connection.execute(
                "SELECT run_id FROM petri_process_runs WHERE active = 1"
            ).fetchone()
            if active is not None:
                raise PetriProcessBusy("another Petri process requires ownership")
            run_id = uuid.uuid4().hex
            now = _utc_now()
            connection.execute(
                """
                INSERT INTO petri_process_runs (
                    run_id, start_request_id, state, active, recipe_id,
                    recipe_version, process_profile_sha256,
                    expected_plate_ids_json, started_at, updated_at
                ) VALUES (?, ?, ?, 1, ?, ?, ?, ?, ?, ?)
                """,
                (
                    run_id,
                    request_id,
                    IDLE,
                    recipe.recipe_id,
                    recipe.recipe_version,
                    recipe.process_profile_sha256,
                    _canonical_json(list(plates)),
                    now,
                    now,
                ),
            )
            row = connection.execute(
                "SELECT * FROM petri_process_runs WHERE run_id = ?", (run_id,)
            ).fetchone()
            return _snapshot(row), True

    def transition(
        self,
        run_id: str,
        to_state: str,
        *,
        reason: str,
        evidence: Mapping[str, Any],
        updates: Optional[Mapping[str, Any]] = None,
    ) -> PetriProcessSnapshot:
        if to_state not in PROCESS_STATES:
            raise ValueError("unknown Petri process state")
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT * FROM petri_process_runs WHERE run_id = ?", (run_id,)
            ).fetchone()
            if row is None:
                raise PetriProcessFault("run_not_found", to_state)
            return self._transition_locked(
                connection,
                row,
                to_state,
                reason=reason,
                evidence=evidence,
                updates=updates or {},
            )

    def _transition_locked(
        self,
        connection: sqlite3.Connection,
        row: sqlite3.Row,
        to_state: str,
        *,
        reason: str,
        evidence: Mapping[str, Any],
        updates: Mapping[str, Any],
    ) -> PetriProcessSnapshot:
        sequence = int(row["transition_sequence"]) + 1
        now = _utc_now()
        evidence_json = _canonical_evidence(evidence)
        allowed_updates = {
            "active",
            "identified_plate_ids_json",
            "current_plate_index",
            "current_plate_id",
            "local_commit_id",
            "local_evidence_sha256",
            "upload_queue_id",
            "recovery_required",
            "error_code",
            "context_json",
        }
        if not set(updates) <= allowed_updates:
            raise ValueError("process transition update is invalid")
        assignments = ["state = ?", "transition_sequence = ?", "updated_at = ?"]
        values: list[Any] = [to_state, sequence, now]
        for name, value in updates.items():
            assignments.append("{} = ?".format(name))
            values.append(value)
        values.append(row["run_id"])
        connection.execute(
            "UPDATE petri_process_runs SET {} WHERE run_id = ?".format(
                ", ".join(assignments)
            ),
            values,
        )
        connection.execute(
            """
            INSERT INTO petri_process_transitions (
                event_id, run_id, sequence, from_state, to_state,
                reason, evidence_json, recorded_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                uuid.uuid4().hex,
                row["run_id"],
                sequence,
                row["state"],
                to_state,
                _bounded_text(reason, 96),
                evidence_json,
                now,
            ),
        )
        updated = connection.execute(
            "SELECT * FROM petri_process_runs WHERE run_id = ?", (row["run_id"],)
        ).fetchone()
        return _snapshot(updated)

    def snapshot(self, run_id: Optional[str] = None) -> Optional[PetriProcessSnapshot]:
        with self._lock, self._connect() as connection:
            if run_id is None:
                row = connection.execute("""
                    SELECT * FROM petri_process_runs
                    ORDER BY updated_at DESC, run_id DESC LIMIT 1
                    """).fetchone()
            else:
                row = connection.execute(
                    "SELECT * FROM petri_process_runs WHERE run_id = ?", (run_id,)
                ).fetchone()
        return _snapshot(row) if row is not None else None

    def transitions(self, run_id: str) -> Tuple[Mapping[str, Any], ...]:
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT sequence, from_state, to_state, reason, evidence_json,
                       recorded_at
                FROM petri_process_transitions
                WHERE run_id = ? ORDER BY sequence ASC
                """,
                (run_id,),
            ).fetchall()
        return tuple(
            MappingProxyType(
                {
                    "sequence": int(row["sequence"]),
                    "from_state": row["from_state"],
                    "to_state": row["to_state"],
                    "reason": row["reason"],
                    "evidence": json.loads(row["evidence_json"]),
                    "recorded_at": row["recorded_at"],
                }
            )
            for row in rows
        )

    def enqueue_upload(self, run_id: str, queue_id: str) -> None:
        """Atomically retain a local upload item without performing network I/O."""

        normalized_queue_id = _identifier(queue_id, "upload_queue_id")
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT * FROM petri_process_runs WHERE run_id = ?", (run_id,)
            ).fetchone()
            if row is None:
                raise PetriProcessFault("run_not_found", QUEUE_UPLOAD)
            if not row["local_commit_id"] or not _SHA256.fullmatch(
                row["local_evidence_sha256"]
            ):
                raise PetriProcessFault("local_evidence_not_committed", QUEUE_UPLOAD)
            existing = connection.execute(
                "SELECT * FROM petri_process_upload_outbox WHERE run_id = ?",
                (run_id,),
            ).fetchone()
            if existing is not None:
                if existing["queue_id"] != normalized_queue_id:
                    raise PetriProcessFault("upload_queue_conflict", QUEUE_UPLOAD)
                return
            connection.execute(
                """
                INSERT INTO petri_process_upload_outbox (
                    queue_id, run_id, local_commit_id, local_evidence_sha256,
                    created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?)
                """,
                (
                    normalized_queue_id,
                    run_id,
                    row["local_commit_id"],
                    row["local_evidence_sha256"],
                    _utc_now(),
                    _utc_now(),
                ),
            )

    def pending_uploads(self, *, limit: int = 20) -> Tuple[Mapping[str, Any], ...]:
        if isinstance(limit, bool) or not 1 <= int(limit) <= 100:
            raise ValueError("upload outbox limit is invalid")
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT queue_id, run_id, local_commit_id,
                       local_evidence_sha256, created_at
                FROM petri_process_upload_outbox
                WHERE status IN ('pending', 'failed')
                ORDER BY created_at ASC, queue_id ASC LIMIT ?
                """,
                (int(limit),),
            ).fetchall()
        return tuple(MappingProxyType(dict(row)) for row in rows)


class DurablePetriProcessController:
    """Run one deterministic Petri recipe through a durable state ledger."""

    def __init__(
        self,
        *,
        store: DurablePetriProcessStore,
        recipe: PetriProcessRecipe,
        expected_plate_ids: Sequence[str],
        operations: Any,
        monotonic: Callable[[], float] = time.monotonic,
    ) -> None:
        if not callable(getattr(operations, "execute_state", None)):
            raise ValueError("Petri operations require execute_state")
        if not callable(getattr(operations, "safe_stop", None)):
            raise ValueError("Petri operations require safe_stop")
        if not callable(getattr(operations, "supervised_recovery", None)):
            raise ValueError("Petri operations require supervised_recovery")
        self.store = store
        self.recipe = recipe
        self.expected_plate_ids = _plate_ids(expected_plate_ids)
        self.operations = operations
        self._monotonic = monotonic
        self._cancel_event = threading.Event()
        self._run_lock = threading.Lock()
        self._binding_lock = threading.Lock()
        self._sampling_tool_control: Any = None

    def bind_sampling_tool_control(self, control: Any) -> None:
        """Bind the one-owner typed #323 capability during trusted assembly."""

        bind = getattr(self.operations, "bind_sampling_tool_control", None)
        if not callable(bind):
            raise RuntimeError("Petri operations cannot accept sampling-tool control")
        with self._binding_lock:
            if self._sampling_tool_control is not None:
                raise RuntimeError("Sampling-tool process control is already bound")
            bind(control)
            self._sampling_tool_control = control

    def estimate(self) -> Mapping[str, Any]:
        per_plate = 7
        total = 5 + per_plate * len(self.expected_plate_ids) + 4
        seconds = sum(
            self.recipe.state_timeouts_seconds[state] for state in _OPERATION_STATES
        )
        seconds += (len(self.expected_plate_ids) - 1) * sum(
            self.recipe.state_timeouts_seconds[state]
            for state in (
                APPROACH,
                PICK_LID,
                VERIFY_GRASP,
                CAPTURE,
                RETURN_LID,
                VERIFY_RELEASE,
                NEXT_PLATE,
            )
        )
        return {"total_steps": total, "estimated_seconds": seconds}

    def status(self) -> Mapping[str, Any]:
        snapshot = self.store.snapshot()
        if snapshot is None:
            return {
                "state": IDLE,
                "run_id": "",
                "active": False,
                "recovery_required": False,
                "local_evidence_committed": False,
                "upload_queued": False,
            }
        return {
            "state": snapshot.state,
            "run_id": snapshot.run_id,
            "active": snapshot.active,
            "recovery_required": snapshot.recovery_required,
            "local_evidence_committed": snapshot.local_evidence_committed,
            "upload_queued": snapshot.upload_queued,
            "current_plate_id": snapshot.current_plate_id,
            "transition_sequence": snapshot.transition_sequence,
            "error_code": snapshot.error_code,
        }

    def cancel(self) -> None:
        self._cancel_event.set()

    def start(
        self,
        *,
        stop_requested: Callable[[], bool],
        progress_callback: Callable[[str], None],
        start_request_id: Optional[str] = None,
    ) -> str:
        request_id = start_request_id or uuid.uuid4().hex
        snapshot, created = self.store.create_or_replay(
            start_request_id=request_id,
            recipe=self.recipe,
            expected_plate_ids=self.expected_plate_ids,
        )
        if not created:
            return snapshot.run_id
        if not self._run_lock.acquire(blocking=False):
            raise PetriProcessBusy("Petri process controller is already running")
        self._cancel_event.clear()

        def combined_stop() -> bool:
            return bool(self._cancel_event.is_set() or stop_requested())

        context: dict[str, Any] = {
            "run_id": snapshot.run_id,
            "recipe_id": self.recipe.recipe_id,
            "recipe_version": self.recipe.recipe_version,
            "process_profile_sha256": self.recipe.process_profile_sha256,
            "expected_plate_ids": list(self.expected_plate_ids),
            "plate_evidence": [],
        }
        active_state = PREFLIGHT
        try:
            for state in (PREFLIGHT, AUTHORIZED, HOME, LOCATE_TRAY, IDENTIFY_PLATES):
                active_state = state
                self._execute(state, context, combined_stop, progress_callback)
            for index, plate_id in enumerate(self.expected_plate_ids):
                context["current_plate_index"] = index
                context["current_plate_id"] = plate_id
                for state in (
                    APPROACH,
                    PICK_LID,
                    VERIFY_GRASP,
                    CAPTURE,
                    RETURN_LID,
                    VERIFY_RELEASE,
                    NEXT_PLATE,
                ):
                    active_state = state
                    self._execute(state, context, combined_stop, progress_callback)
            for state in (RETURN_HOME, LOCAL_COMMIT, QUEUE_UPLOAD):
                active_state = state
                self._execute(state, context, combined_stop, progress_callback)
            self._require_completion_context(context)
            self.store.transition(
                snapshot.run_id,
                COMPLETE,
                reason="process_complete",
                evidence={
                    "local_commit_id": context["local_commit_id"],
                    "local_evidence_sha256": context["local_evidence_sha256"],
                    "upload_queue_id": context["upload_queue_id"],
                    "plate_ids": list(self.expected_plate_ids),
                },
                updates={"active": 0, "recovery_required": 0, "error_code": ""},
            )
            progress_callback("COMPLETE")
            return snapshot.run_id
        except PetriProcessCancelled:
            self._handle_cancel(snapshot.run_id, active_state, context)
            return snapshot.run_id
        except Exception as exc:
            self._handle_fault(snapshot.run_id, active_state, context, exc)
            raise
        finally:
            self._run_lock.release()

    def _execute(
        self,
        state: str,
        context: dict[str, Any],
        stop_requested: Callable[[], bool],
        progress_callback: Callable[[str], None],
    ) -> None:
        self._check_preconditions(state, context)
        if stop_requested():
            raise PetriProcessCancelled("process stop requested")
        plate_id = str(context.get("current_plate_id") or "")
        updates: dict[str, Any] = {
            "current_plate_index": int(context.get("current_plate_index") or 0),
            "current_plate_id": plate_id,
            "context_json": _canonical_evidence(_public_context(context)),
        }
        self.store.transition(
            context["run_id"],
            state,
            reason="state_entered",
            evidence={
                "plate_id": plate_id,
                "preconditions": list(_PRECONDITIONS[state]),
            },
            updates=updates,
        )
        progress_callback(state)
        timeout = self.recipe.state_timeouts_seconds[state]
        deadline = float(self._monotonic()) + timeout
        output = self.operations.execute_state(
            state,
            _immutable_context(context),
            stop_requested=stop_requested,
            deadline_monotonic=deadline,
        )
        if stop_requested():
            raise PetriProcessCancelled("process stop requested")
        if float(self._monotonic()) > deadline:
            raise PetriProcessFault("state_timeout", state)
        evidence = _validated_operation_evidence(output, state)
        self._check_postconditions(state, context, evidence)
        context.update(evidence)
        if state == CAPTURE:
            context["plate_evidence"].append(
                {
                    "plate_id": context["current_plate_id"],
                    "capture_id": evidence["capture_id"],
                    "capture_sha256": evidence["capture_sha256"],
                }
            )
        state_updates: dict[str, Any] = {
            "context_json": _canonical_evidence(_public_context(context))
        }
        if state == IDENTIFY_PLATES:
            state_updates["identified_plate_ids_json"] = _canonical_json(
                evidence["identified_plate_ids"]
            )
        elif state == LOCAL_COMMIT:
            state_updates.update(
                {
                    "local_commit_id": evidence["local_commit_id"],
                    "local_evidence_sha256": evidence["local_evidence_sha256"],
                }
            )
        elif state == QUEUE_UPLOAD:
            state_updates["upload_queue_id"] = evidence["upload_queue_id"]
            self.store.enqueue_upload(context["run_id"], evidence["upload_queue_id"])
        self.store.transition(
            context["run_id"],
            state,
            reason="state_postconditions_met",
            evidence=dict(evidence),
            updates=state_updates,
        )

    def _check_preconditions(self, state: str, context: Mapping[str, Any]) -> None:
        missing = [name for name in _PRECONDITIONS[state] if not context.get(name)]
        if missing:
            raise PetriProcessFault("precondition_failed", state)
        if state == LOCAL_COMMIT and len(context.get("plate_evidence") or ()) != len(
            self.expected_plate_ids
        ):
            raise PetriProcessFault("capture_evidence_incomplete", state)

    def _check_postconditions(
        self,
        state: str,
        context: Mapping[str, Any],
        evidence: Mapping[str, Any],
    ) -> None:
        if evidence.get("ok") is not True:
            raise PetriProcessFault("postcondition_failed", state)
        required_true = {
            PREFLIGHT: "preflight_verified",
            AUTHORIZED: "authorization_verified",
            HOME: "home_verified",
            LOCATE_TRAY: "tray_located",
            APPROACH: "approach_verified",
            PICK_LID: "lid_pick_requested",
            VERIFY_GRASP: "grasp_verified",
            RETURN_LID: "lid_returned",
            VERIFY_RELEASE: "release_verified",
            RETURN_HOME: "home_verified",
        }
        required = required_true.get(state)
        if required is not None and evidence.get(required) is not True:
            raise PetriProcessFault("postcondition_failed", state)
        if state == PREFLIGHT:
            _sha256(evidence.get("preflight_sha256"), "preflight_sha256")
        elif state == AUTHORIZED:
            _identifier(
                evidence.get("authorization_receipt_id"),
                "authorization_receipt_id",
            )
        elif state in {HOME, RETURN_HOME}:
            _sha256(evidence.get("home_pose_sha256"), "home_pose_sha256")
        elif state == LOCATE_TRAY:
            _sha256(evidence.get("tray_pose_sha256"), "tray_pose_sha256")
        elif state == IDENTIFY_PLATES:
            identified = _plate_ids(evidence.get("identified_plate_ids") or ())
            if identified != self.expected_plate_ids:
                raise PetriProcessFault("plate_identity_mismatch", state)
            _sha256(
                evidence.get("identity_evidence_sha256"),
                "identity_evidence_sha256",
            )
        elif state in {APPROACH, RETURN_LID}:
            _identifier(
                evidence.get("trajectory_receipt_id"),
                "trajectory_receipt_id",
            )
        elif state == PICK_LID:
            _identifier(evidence.get("tool_receipt_id"), "tool_receipt_id")
        elif state == VERIFY_GRASP:
            _identifier(evidence.get("grasp_receipt_id"), "grasp_receipt_id")
        elif state == CAPTURE:
            _identifier(evidence.get("capture_id"), "capture_id")
            _sha256(evidence.get("capture_sha256"), "capture_sha256")
        elif state == VERIFY_RELEASE:
            _identifier(evidence.get("release_receipt_id"), "release_receipt_id")
        elif state == LOCAL_COMMIT:
            _identifier(evidence.get("local_commit_id"), "local_commit_id")
            _sha256(evidence.get("local_evidence_sha256"), "local_evidence_sha256")
        elif state == QUEUE_UPLOAD:
            _identifier(evidence.get("upload_queue_id"), "upload_queue_id")

    @staticmethod
    def _require_completion_context(context: Mapping[str, Any]) -> None:
        if not (
            context.get("local_commit_id")
            and context.get("local_evidence_sha256")
            and context.get("upload_queue_id")
        ):
            raise PetriProcessFault("completion_evidence_missing", COMPLETE)

    def _handle_cancel(
        self,
        run_id: str,
        active_state: str,
        context: Mapping[str, Any],
    ) -> None:
        self.store.transition(
            run_id,
            STOPPING,
            reason="stop_or_cancel_requested",
            evidence={"interrupted_state": active_state},
        )
        stop_evidence = self._safe_stop(active_state, context, "cancelled")
        if active_state in _MOTION_OR_TOOL_STATES:
            self.store.transition(
                run_id,
                RECOVERY_REQUIRED,
                reason="interrupted_physical_state",
                evidence=stop_evidence,
                updates={
                    "active": 1,
                    "recovery_required": 1,
                    "error_code": "process_interrupted",
                },
            )
        else:
            self.store.transition(
                run_id,
                CANCELLED,
                reason="cancelled_before_physical_uncertainty",
                evidence=stop_evidence,
                updates={"active": 0, "recovery_required": 0},
            )

    def _handle_fault(
        self,
        run_id: str,
        active_state: str,
        context: Mapping[str, Any],
        exc: BaseException,
    ) -> None:
        code = (
            exc.code if isinstance(exc, PetriProcessFault) else "state_operation_failed"
        )
        stop_evidence = self._safe_stop(active_state, context, code)
        self.store.transition(
            run_id,
            FAULT_LOCKED,
            reason="state_fault",
            evidence={
                "failed_state": active_state,
                "error_code": _error_code(code),
                "exception_type": type(exc).__name__[:64],
                "safe_stop": stop_evidence,
            },
            updates={
                "active": 1,
                "recovery_required": 1,
                "error_code": _error_code(code),
            },
        )

    def _safe_stop(
        self,
        state: str,
        context: Mapping[str, Any],
        reason: str,
    ) -> Mapping[str, Any]:
        try:
            result = self.operations.safe_stop(
                state,
                _immutable_context(context),
                reason=_error_code(reason),
            )
            evidence = _validated_operation_evidence(result, STOPPING)
        except Exception as exc:
            return {
                "ok": False,
                "safe_state_verified": False,
                "error_code": "safe_stop_failed",
                "exception_type": type(exc).__name__[:64],
            }
        return dict(evidence)

    def supervised_recovery(
        self,
        *,
        supervisor_id: str,
        stop_requested: Callable[[], bool],
    ) -> str:
        supervisor = _identifier(supervisor_id, "supervisor_id")
        snapshot = self.store.snapshot()
        if (
            snapshot is None
            or not snapshot.active
            or snapshot.state
            not in {
                FAULT_LOCKED,
                RECOVERY_REQUIRED,
            }
        ):
            raise PetriProcessFault("recovery_not_required", RECOVERY_REQUIRED)
        if not self._run_lock.acquire(blocking=False):
            raise PetriProcessBusy("Petri recovery is already running")
        try:
            self.store.transition(
                snapshot.run_id,
                RECOVERY_REQUIRED,
                reason="supervised_recovery_started",
                evidence={"supervisor_id": supervisor},
            )
            deadline = (
                float(self._monotonic()) + self.recipe.state_timeouts_seconds[HOME]
            )
            result = self.operations.supervised_recovery(
                MappingProxyType({"run_id": snapshot.run_id}),
                supervisor_id=supervisor,
                stop_requested=stop_requested,
                deadline_monotonic=deadline,
            )
            evidence = _validated_operation_evidence(result, RECOVERY_REQUIRED)
            if stop_requested() or float(self._monotonic()) > deadline:
                raise PetriProcessFault("recovery_interrupted", RECOVERY_REQUIRED)
            if not (
                evidence.get("ok") is True
                and evidence.get("home_verified") is True
                and evidence.get("tool_safe_verified") is True
            ):
                raise PetriProcessFault("recovery_unverified", RECOVERY_REQUIRED)
            self.store.transition(
                snapshot.run_id,
                IDLE,
                reason="supervised_recovery_verified",
                evidence={**evidence, "supervisor_id": supervisor},
                updates={
                    "active": 0,
                    "recovery_required": 0,
                    "error_code": "",
                    "current_plate_id": "",
                },
            )
            return snapshot.run_id
        except Exception as exc:
            current = self.store.snapshot(snapshot.run_id)
            if current is not None:
                self.store.transition(
                    snapshot.run_id,
                    RECOVERY_REQUIRED,
                    reason="supervised_recovery_failed",
                    evidence={
                        "error_code": "recovery_failed",
                        "exception_type": type(exc).__name__[:64],
                    },
                    updates={
                        "active": 1,
                        "recovery_required": 1,
                        "error_code": "recovery_failed",
                    },
                )
            raise
        finally:
            self._run_lock.release()


_PRECONDITIONS = {
    PREFLIGHT: (),
    AUTHORIZED: ("preflight_verified",),
    HOME: ("authorization_verified",),
    LOCATE_TRAY: ("home_verified",),
    IDENTIFY_PLATES: ("tray_located",),
    APPROACH: ("identified_plate_ids", "current_plate_id"),
    PICK_LID: ("approach_verified",),
    VERIFY_GRASP: ("lid_pick_requested",),
    CAPTURE: ("grasp_verified",),
    RETURN_LID: ("capture_id", "capture_sha256"),
    VERIFY_RELEASE: ("lid_returned",),
    NEXT_PLATE: ("release_verified",),
    RETURN_HOME: ("release_verified",),
    LOCAL_COMMIT: ("home_verified", "plate_evidence"),
    QUEUE_UPLOAD: ("local_commit_id", "local_evidence_sha256"),
}

_EVIDENCE_FIELDS = {
    PREFLIGHT: frozenset({"ok", "preflight_verified", "preflight_sha256"}),
    AUTHORIZED: frozenset({"ok", "authorization_verified", "authorization_receipt_id"}),
    HOME: frozenset({"ok", "home_verified", "home_pose_sha256"}),
    LOCATE_TRAY: frozenset({"ok", "tray_located", "tray_pose_sha256"}),
    IDENTIFY_PLATES: frozenset(
        {"ok", "identified_plate_ids", "identity_evidence_sha256"}
    ),
    APPROACH: frozenset({"ok", "approach_verified", "trajectory_receipt_id"}),
    PICK_LID: frozenset({"ok", "lid_pick_requested", "tool_receipt_id"}),
    VERIFY_GRASP: frozenset({"ok", "grasp_verified", "grasp_receipt_id"}),
    CAPTURE: frozenset({"ok", "capture_id", "capture_sha256"}),
    RETURN_LID: frozenset({"ok", "lid_returned", "trajectory_receipt_id"}),
    VERIFY_RELEASE: frozenset({"ok", "release_verified", "release_receipt_id"}),
    NEXT_PLATE: frozenset({"ok"}),
    RETURN_HOME: frozenset({"ok", "home_verified", "home_pose_sha256"}),
    LOCAL_COMMIT: frozenset({"ok", "local_commit_id", "local_evidence_sha256"}),
    QUEUE_UPLOAD: frozenset({"ok", "upload_queue_id"}),
    STOPPING: frozenset({"ok", "safe_state_verified", "error_code", "exception_type"}),
    RECOVERY_REQUIRED: frozenset({"ok", "home_verified", "tool_safe_verified"}),
}


def _validated_operation_evidence(value: Any, state: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise PetriProcessFault("operation_evidence_invalid", state)
    allowed = _EVIDENCE_FIELDS.get(state)
    if allowed is None or not set(value) <= allowed:
        raise PetriProcessFault("operation_evidence_invalid", state)
    encoded = _canonical_evidence(value)
    decoded = json.loads(encoded)
    if not isinstance(decoded, dict):
        raise PetriProcessFault("operation_evidence_invalid", state)
    return MappingProxyType(decoded)


def _public_context(context: Mapping[str, Any]) -> Mapping[str, Any]:
    return {
        key: value
        for key, value in context.items()
        if key
        in {
            "run_id",
            "recipe_id",
            "recipe_version",
            "process_profile_sha256",
            "expected_plate_ids",
            "identified_plate_ids",
            "current_plate_index",
            "current_plate_id",
            "plate_evidence",
            "local_commit_id",
            "local_evidence_sha256",
            "upload_queue_id",
        }
    }


def _immutable_context(value: Mapping[str, Any]) -> Mapping[str, Any]:
    """Return a recursively immutable, canonical copy for physical callbacks."""

    decoded = json.loads(_canonical_evidence(value))

    def freeze(item: Any) -> Any:
        if isinstance(item, dict):
            return MappingProxyType({key: freeze(child) for key, child in item.items()})
        if isinstance(item, list):
            return tuple(freeze(child) for child in item)
        return item

    return freeze(decoded)


def _snapshot(row: sqlite3.Row) -> PetriProcessSnapshot:
    expected = tuple(json.loads(row["expected_plate_ids_json"]))
    identified = tuple(json.loads(row["identified_plate_ids_json"]))
    return PetriProcessSnapshot(
        run_id=row["run_id"],
        start_request_id=row["start_request_id"],
        state=row["state"],
        active=bool(row["active"]),
        recipe_id=row["recipe_id"],
        recipe_version=row["recipe_version"],
        process_profile_sha256=row["process_profile_sha256"],
        expected_plate_ids=expected,
        identified_plate_ids=identified,
        current_plate_index=int(row["current_plate_index"]),
        current_plate_id=row["current_plate_id"],
        transition_sequence=int(row["transition_sequence"]),
        local_evidence_committed=bool(
            row["local_commit_id"] and row["local_evidence_sha256"]
        ),
        local_commit_id=row["local_commit_id"],
        local_evidence_sha256=row["local_evidence_sha256"],
        upload_queued=bool(row["upload_queue_id"]),
        upload_queue_id=row["upload_queue_id"],
        recovery_required=bool(row["recovery_required"]),
        error_code=row["error_code"],
        started_at=row["started_at"],
        updated_at=row["updated_at"],
    )


def _plate_ids(values: Sequence[str]) -> Tuple[str, ...]:
    if isinstance(values, (str, bytes)):
        raise ValueError("plate identities must be a sequence")
    result = tuple(values)
    if not 1 <= len(result) <= 96 or len(set(result)) != len(result):
        raise ValueError("plate identities are invalid")
    if any(
        not isinstance(value, str) or not _PLATE_ID.fullmatch(value) for value in result
    ):
        raise ValueError("plate identity is invalid")
    return result


def _identifier(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value):
        raise ValueError("{} is invalid".format(name))
    return value


def _sha256(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _SHA256.fullmatch(value):
        raise ValueError("{} is invalid".format(name))
    return value


def _hex32(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _HEX32.fullmatch(value):
        raise ValueError("{} is invalid".format(name))
    return value


def _error_code(value: Any) -> str:
    text = str(value or "process_error").strip().lower()
    normalized = re.sub(r"[^a-z0-9_]+", "_", text).strip("_")[:64]
    return normalized or "process_error"


def _bounded_text(value: Any, maximum: int) -> str:
    text = str(value or "")
    if any(ord(character) < 32 for character in text):
        raise ValueError("process text contains control characters")
    return text[:maximum]


def _canonical_json(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False)


def _canonical_evidence(value: Any) -> str:
    try:
        encoded = _canonical_json(value)
    except (TypeError, ValueError) as exc:
        raise ValueError("process evidence is not canonical JSON") from exc
    if len(encoded.encode("utf-8")) > _MAX_EVIDENCE_BYTES:
        raise ValueError("process evidence is too large")
    return encoded


def _utc_now() -> str:
    return (
        datetime.now(timezone.utc)
        .isoformat(timespec="microseconds")
        .replace("+00:00", "Z")
    )


def evidence_sha256(value: Mapping[str, Any]) -> str:
    """Return the canonical digest used by local evidence fixtures/adapters."""

    return hashlib.sha256(_canonical_evidence(value).encode("utf-8")).hexdigest()
