flowchart LR
    SPEC["Versioned spec + cycle ID"] --> PRE["Preflight"]
    PRE --> FEED["Mock feeder"]
    FEED --> CONV["Mock conveyor + sensor"]
    CONV --> OBS["Mock camera/classifier"]
    OBS --> GATE{"class + confidence + transform valid?"}
    GATE -->|ні| REJ["REJECT / no goal"]
    GATE -->|так| RG["Mock robot + GripperAdapter"]
    RG --> DB["SQLite :memory: transaction + audit"]
    DB --> DONE["COMPLETE"]
    PRE -->|fault| REC["RECOVERY_REQUIRED"]
    FEED -->|timeout| REC
    CONV -->|timeout| REC
    OBS -->|lost| REC
    RG -->|no ack| REC
    DB -->|rollback| REC

# safe-smoke
SAFE = {"feeder":"STOP", "conveyor":"STOP", "robot":"NO_NEW_GOAL", "gripper":"HOLD_OR_STOP"}

def terminal_safe(outputs: dict[str, str]) -> bool:
    return outputs == SAFE

def recover(operator_ack: bool, cause_removed: bool, reconciled: bool, area_clear: bool) -> bool:
    return all((operator_ack, cause_removed, reconciled, area_clear))

assert terminal_safe(dict(SAFE))
assert not recover(True, True, False, True)
assert recover(True, True, True, True)
print("safe-smoke: final-cell invariants PASS")

{
  "mode":"simulation-only",
  "confidence_min":0.80,
  "timeouts":{"feed":3,"transport":4,"robot":3,"gripper":2},
  "routes":{"red-cylinder":"BOX_A","blue-cube":"BOX_B"},
  "cases":[
    {"acceptance_id":"AC-02","id":"normal","cycle":"C-001","feed":2,"transport":3,"camera":true,"label":"red-cylinder","confidence":0.92,"robot":2,"gripper":1,"grip_ok":true,"db_ok":true,"restart":false,"expected":"COMPLETE"},
    {"acceptance_id":"AC-02","id":"boundary-timeouts","cycle":"C-009","feed":3,"transport":4,"camera":true,"label":"blue-cube","confidence":0.80,"robot":3,"gripper":2,"grip_ok":true,"db_ok":true,"restart":false,"expected":"COMPLETE"},
    {"acceptance_id":"AC-03","id":"unknown","cycle":"C-002","feed":1,"transport":1,"camera":true,"label":"UNKNOWN","confidence":0.30,"robot":0,"gripper":0,"grip_ok":false,"db_ok":true,"restart":false,"expected":"REJECT"},
    {"acceptance_id":"AC-04","id":"duplicate","cycle":"C-001","feed":0,"transport":0,"camera":true,"label":"red-cylinder","confidence":1.0,"robot":0,"gripper":0,"grip_ok":true,"db_ok":true,"restart":false,"expected":"IDEMPOTENT"},
    {"acceptance_id":"AC-05","id":"feed-timeout","cycle":"C-003","feed":4,"transport":1,"camera":true,"label":"red-cylinder","confidence":0.90,"robot":1,"gripper":1,"grip_ok":true,"db_ok":true,"restart":false,"expected":"RECOVERY_REQUIRED"},
    {"acceptance_id":"AC-06","id":"camera-lost","cycle":"C-004","feed":1,"transport":1,"camera":false,"label":"red-cylinder","confidence":0.90,"robot":1,"gripper":1,"grip_ok":true,"db_ok":true,"restart":false,"expected":"RECOVERY_REQUIRED"},
    {"acceptance_id":"AC-07","id":"robot-timeout","cycle":"C-005","feed":1,"transport":1,"camera":true,"label":"blue-cube","confidence":0.90,"robot":4,"gripper":1,"grip_ok":true,"db_ok":true,"restart":false,"expected":"RECOVERY_REQUIRED"},
    {"acceptance_id":"AC-08","id":"sensor-timeout","cycle":"C-006","feed":1,"transport":5,"camera":true,"label":"red-cylinder","confidence":0.90,"robot":1,"gripper":1,"grip_ok":true,"db_ok":true,"restart":false,"expected":"RECOVERY_REQUIRED"},
    {"acceptance_id":"AC-09","id":"db-failure","cycle":"C-007","feed":1,"transport":1,"camera":true,"label":"red-cylinder","confidence":0.90,"robot":1,"gripper":1,"grip_ok":true,"db_ok":false,"restart":false,"expected":"RECOVERY_REQUIRED"},
    {"acceptance_id":"AC-10","id":"restart","cycle":"C-008","feed":1,"transport":1,"camera":true,"label":"red-cylinder","confidence":0.90,"robot":1,"gripper":1,"grip_ok":true,"db_ok":true,"restart":true,"expected":"RECOVERY_REQUIRED"}
  ]
}

from __future__ import annotations

import argparse
import json
import sqlite3
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path


class State(str, Enum):
    IDLE="IDLE"; PREFLIGHT="PREFLIGHT"; FEEDING="FEEDING"; TRANSPORTING="TRANSPORTING"
    INSPECTING="INSPECTING"; PICKING="PICKING"; PERSISTING="PERSISTING"
    COMPLETE="COMPLETE"; REJECT="REJECT"; IDEMPOTENT="IDEMPOTENT"
    FAULT="FAULT"; RECOVERY_REQUIRED="RECOVERY_REQUIRED"

SAFE = {"feeder":"STOP", "conveyor":"STOP", "robot":"NO_NEW_GOAL", "gripper":"HOLD_OR_STOP"}


@dataclass
class Outcome:
    acceptance_id: str
    case_id: str
    cycle_id: str
    state: str
    reason: str
    route: str | None
    outputs: dict[str, str]
    ledger_rows: int
    transitions: list[str] = field(default_factory=list)


def no_secret_keys(value: object) -> bool:
    if isinstance(value, dict):
        for key, item in value.items():
            if any(word in str(key).lower() for word in ("password", "token", "private_key")):
                return False
            if not no_secret_keys(item):
                return False
    elif isinstance(value, list):
        return all(no_secret_keys(item) for item in value)
    return True


class ReferenceCell:
    def __init__(self, spec: dict) -> None:
        if spec.get("mode") != "simulation-only" or not no_secret_keys(spec):
            raise ValueError("AC-01 failed")
        self.spec = spec
        self.db = sqlite3.connect(":memory:")
        self.db.execute("CREATE TABLE ledger(cycle_id TEXT PRIMARY KEY, route TEXT NOT NULL, status TEXT NOT NULL)")
        self.db.commit()
        self.state = State.IDLE
        self.outputs = dict(SAFE)
        self.log: list[str] = []

    def reset_runtime(self) -> None:
        self.state = State.IDLE; self.outputs = dict(SAFE); self.log = []

    def move(self, target: State, reason: str) -> None:
        self.log.append(f"{self.state.value}->{target.value}:{reason}")
        self.state = target

    def safe_stop(self) -> None:
        self.outputs = dict(SAFE)

    def fail(self, reason: str) -> None:
        self.safe_stop(); self.move(State.FAULT, reason); self.move(State.RECOVERY_REQUIRED, "manual-reconciliation")

    def ledger_count(self) -> int:
        return int(self.db.execute("SELECT COUNT(*) FROM ledger").fetchone()[0])

    def run(self, case: dict) -> Outcome:
        self.reset_runtime(); cycle = str(case["cycle"]); route = None; reason = "ok"
        self.move(State.PREFLIGHT, "start")
        if self.db.execute("SELECT 1 FROM ledger WHERE cycle_id=?", (cycle,)).fetchone():
            self.move(State.IDEMPOTENT, "completed-cycle-already-recorded")
        else:
            self.move(State.FEEDING, "preflight-pass"); self.outputs["feeder"] = "RUN"
        if self.state is State.FEEDING:
            if int(case["feed"]) > int(self.spec["timeouts"]["feed"]): self.fail("feed-timeout")
            else:
                self.outputs.update(feeder="STOP", conveyor="RUN"); self.move(State.TRANSPORTING, "feed-ack")
        if self.state is State.TRANSPORTING:
            if int(case["transport"]) > int(self.spec["timeouts"]["transport"]): self.fail("sensor-timeout")
            else:
                self.outputs["conveyor"] = "STOP"; self.move(State.INSPECTING, "sensor-and-stop-ack")
        if self.state is State.INSPECTING:
            if bool(case["restart"]): self.fail("restart-mid-cycle")
            elif not bool(case["camera"]): self.fail("camera-lost")
            elif str(case["label"]) not in self.spec["routes"] or float(case["confidence"]) < float(self.spec["confidence_min"]):
                reason = "classification-rejected"; self.safe_stop(); self.move(State.REJECT, reason)
            else:
                route = str(self.spec["routes"][str(case["label"])]); self.outputs["robot"] = "MOCK_GOAL"
                self.move(State.PICKING, "class-transform-route-valid")
        if self.state is State.PICKING:
            if int(case["robot"]) > int(self.spec["timeouts"]["robot"]): self.fail("robot-timeout")
            elif int(case["gripper"]) > int(self.spec["timeouts"]["gripper"]) or not bool(case["grip_ok"]): self.fail("grip-failed")
            else:
                self.outputs.update(robot="NO_NEW_GOAL", gripper="HOLD_OR_STOP"); self.move(State.PERSISTING, "mock-place-ack")
        if self.state is State.PERSISTING:
            before = self.ledger_count()
            try:
                self.db.execute("BEGIN")
                if not bool(case["db_ok"]): raise sqlite3.OperationalError("injected-db-failure")
                self.db.execute("INSERT INTO ledger(cycle_id,route,status) VALUES(?,?,?)", (cycle, route, "COMPLETE"))
                self.db.commit(); self.safe_stop(); self.move(State.COMPLETE, "transaction-committed")
            except sqlite3.DatabaseError as exc:
                self.db.rollback(); assert self.ledger_count() == before; self.fail(str(exc))
        if self.state in (State.COMPLETE, State.REJECT, State.IDEMPOTENT, State.RECOVERY_REQUIRED):
            self.safe_stop()
        assert self.outputs == SAFE
        reason = self.log[-1].split(":", 1)[1] if self.log else reason
        return Outcome(str(case["acceptance_id"]), str(case["id"]), cycle, self.state.value,
                       reason, route, dict(self.outputs), self.ledger_count(), list(self.log))

    def recover(self, operator_ack: bool, cause_removed: bool, reconciled: bool, area_clear: bool) -> bool:
        if self.state is not State.RECOVERY_REQUIRED or not all((operator_ack,cause_removed,reconciled,area_clear)):
            return False
        self.safe_stop(); self.move(State.IDLE, "manual-recovery"); return True


def main() -> int:
    p = argparse.ArgumentParser(); p.add_argument("--spec", type=Path, required=True); p.add_argument("--out", type=Path, required=True)
    args = p.parse_args(); spec = json.loads(args.spec.read_text(encoding="utf-8")); cell = ReferenceCell(spec)
    results: list[Outcome] = []
    checks: dict[str, bool] = {"AC-01": spec.get("mode") == "simulation-only" and no_secret_keys(spec)}
    for case in spec["cases"]:
        before = cell.ledger_count(); result = cell.run(case); results.append(result)
        acceptance_id = str(case["acceptance_id"])
        case_pass = result.state == str(case["expected"]) and result.outputs == SAFE
        if acceptance_id == "AC-02":
            case_pass = case_pass and result.ledger_rows == before + 1
        if acceptance_id in ("AC-04", "AC-09"):
            case_pass = case_pass and result.ledger_rows == before
        if acceptance_id in ("AC-03", "AC-06"):
            case_pass = case_pass and not any("->PICKING:" in entry for entry in result.transitions)
        checks[acceptance_id] = checks.get(acceptance_id, True) and case_pass
        if result.state == "RECOVERY_REQUIRED":
            checks["AC-11"] = checks.get("AC-11", True) and not cell.recover(False, True, True, True) and cell.recover(True, True, True, True)
    checks["AC-12"] = all(r.cycle_id and r.transitions and r.reason for r in results) and no_secret_keys([asdict(r) for r in results])
    for required in (f"AC-{number:02d}" for number in range(1, 13)):
        checks.setdefault(required, False)
    evidence = {"checks": checks, "outcomes": [asdict(r) for r in results]}
    args.out.write_text(json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8")
    for key in sorted(checks): print(f"{key}: {'PASS' if checks[key] else 'FAIL'}")
    if not all(checks.values()): return 1
    print("ALL ACCEPTANCE TESTS PASS (simulation only)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python lab24_acceptance.py --spec lab24-spec.json --out lab24-evidence.json
python -m json.tool lab24-evidence.json

{
  "project":"MERC-I5 final cell mock",
  "revision":"<commit-sha-or-archive-checksum>",
  "mode":"simulation-only",
  "acceptance_evidence":"lab24-evidence.json",
  "hardware_tested":false
}