flowchart TD
    LOAD["Завантажити pallet/cycle"] --> PICK["Mock PICKED"]
    PICK --> CAM["Mock AT_CAMERA"]
    CAM --> QR{"QR v1 валідний та узгоджений?"}
    QR -->|так| ID["IDENTIFIED + event у транзакції"]
    ID --> ROUTE["ROUTE_ASSIGNED"]
    ROUTE --> CONV["Mock ON_CONVEYOR"]
    CONV --> ACK{"Сенсор підтвердив?"}
    ACK -->|так| STORED["STORED + cell empty + cycle complete"]
    QR -->|ні| ERR["ERROR / STOP"]
    ACK -->|ні| REC["RECOVERY_REQUIRED / STOP"]
    PICK -->|restart| REC

# safe-smoke
import json
import re

ID_RE = re.compile(r"^[A-Z][A-Z0-9_-]{2,31}$")

def validate(raw: str) -> dict:
    if len(raw.encode("utf-8")) > 512:
        raise ValueError("payload-too-large")
    data = json.loads(raw)
    if not isinstance(data, dict) or not {"schema", "id", "type"} <= data.keys():
        raise ValueError("missing-fields")
    if not set(data) <= {"schema", "id", "type", "batch"}:
        raise ValueError("extra-fields")
    if data["schema"] != "merc-i5.workpiece.v1" or not ID_RE.fullmatch(data["id"]):
        raise ValueError("invalid-schema-or-id")
    return data

valid = '{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"LAB-2026-01"}'
assert validate(valid)["id"] == "BARREL-0001"
for bad in ('{"schema":"x","id":"bad","type":"barrel"}',
            '{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","secret":"x"}'):
    try:
        validate(bad)
    except ValueError:
        pass
    else:
        raise AssertionError("invalid QR accepted")
print("safe-smoke: QR v1 validation PASS")

[
  {"id":"normal","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"LAB-2026-01"},"camera":true,"sensor":true,"duplicate_delivery":false,"conflict":false,"restart":false},
  {"id":"duplicate-delivery","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"LAB-2026-01"},"camera":true,"sensor":true,"duplicate_delivery":true,"conflict":false,"restart":false},
  {"id":"conflicting-payload","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"OTHER"},"camera":true,"sensor":true,"duplicate_delivery":false,"conflict":true,"restart":false},
  {"id":"invalid-extra-field","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","secret":"forbidden"},"camera":true,"sensor":true,"duplicate_delivery":false,"conflict":false,"restart":false},
  {"id":"camera-lost","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"LAB-2026-01"},"camera":false,"sensor":true,"duplicate_delivery":false,"conflict":false,"restart":false},
  {"id":"sensor-timeout","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"LAB-2026-01"},"camera":true,"sensor":false,"duplicate_delivery":false,"conflict":false,"restart":false},
  {"id":"restart-mid-cycle","payload":{"schema":"merc-i5.workpiece.v1","id":"BARREL-0001","type":"barrel","batch":"LAB-2026-01"},"camera":true,"sensor":true,"duplicate_delivery":false,"conflict":false,"restart":true}
]

from __future__ import annotations

import argparse
import json
import re
import sqlite3
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from pathlib import Path

ID_RE = re.compile(r"^[A-Z][A-Z0-9_-]{2,31}$")
ALLOWED = {"schema", "id", "type", "batch"}


@dataclass
class Outcome:
    case_id: str
    cycle_status: str
    workpiece_state: str
    events: int
    errors: int
    cell_empty: bool
    destination: str | None
    reason: str


@contextmanager
def transaction(con: sqlite3.Connection):
    con.execute("BEGIN IMMEDIATE")
    try:
        yield
    except Exception:
        con.rollback()
        raise
    else:
        con.commit()


def canonical(data: dict) -> str:
    return json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def validate_payload(data: object) -> dict:
    if not isinstance(data, dict):
        raise ValueError("payload-not-object")
    if not {"schema", "id", "type"} <= set(data):
        raise ValueError("missing-fields")
    if not set(data) <= ALLOWED:
        raise ValueError("extra-fields")
    if data["schema"] != "merc-i5.workpiece.v1":
        raise ValueError("unknown-schema")
    if not isinstance(data["id"], str) or not ID_RE.fullmatch(data["id"]):
        raise ValueError("invalid-id")
    if not isinstance(data["type"], str) or not data["type"]:
        raise ValueError("invalid-type")
    if "batch" in data and not isinstance(data["batch"], str):
        raise ValueError("invalid-batch")
    if len(canonical(data).encode("utf-8")) > 512:
        raise ValueError("payload-too-large")
    return data


def open_memory(schema_path: Path) -> sqlite3.Connection:
    con = sqlite3.connect(":memory:", isolation_level=None)
    con.row_factory = sqlite3.Row
    con.executescript(schema_path.read_text(encoding="utf-8"))
    assert con.execute("PRAGMA foreign_keys").fetchone()[0] == 1
    return con


def seed(con: sqlite3.Connection, cycle_id: str) -> str:
    payload = {"schema":"merc-i5.workpiece.v1", "id":"BARREL-0001", "type":"barrel", "batch":"LAB-2026-01"}
    with transaction(con):
        con.execute("INSERT INTO workpieces(id,type,batch,payload_json,state) VALUES(?,?,?,?,?)",
                    (payload["id"], payload["type"], payload["batch"], canonical(payload), "ON_PALLET"))
        con.execute("INSERT INTO pallets(id,rows,columns) VALUES(?,?,?)", ("PALLET-A", 1, 1))
        con.execute("INSERT INTO pallet_cells(pallet_id,row_index,column_index,workpiece_id) VALUES(?,?,?,?)",
                    ("PALLET-A", 0, 0, payload["id"]))
        con.execute("INSERT INTO storage_locations(id,description) VALUES(?,?)", ("BIN-A", "Навчальне місце A"))
        con.execute("INSERT INTO routes(id,workpiece_type,destination_id,enabled) VALUES(?,?,?,1)",
                    (1, "barrel", "BIN-A"))
        con.execute("INSERT INTO cycles(id,pallet_id,status) VALUES(?,?,?)", (cycle_id, "PALLET-A", "RUNNING"))
    return payload["id"]


def event(con: sqlite3.Connection, key: str, cycle: str, item: str, kind: str, details: dict) -> bool:
    cur = con.execute(
        "INSERT OR IGNORE INTO events(idempotency_key,cycle_id,workpiece_id,event_type,details_json) VALUES(?,?,?,?,?)",
        (key, cycle, item, kind, canonical(details)),
    )
    return cur.rowcount == 1


def transition(con: sqlite3.Connection, cycle: str, item: str, expected: str, target: str, key: str) -> None:
    row = con.execute("SELECT state FROM workpieces WHERE id=?", (item,)).fetchone()
    if row is None or row["state"] != expected:
        raise RuntimeError(f"state-mismatch:{expected}->{target}")
    con.execute("UPDATE workpieces SET state=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (target, item))
    if not event(con, key, cycle, item, target, {"from":expected,"to":target}):
        raise RuntimeError("duplicate-transition-event")


def latch(con: sqlite3.Connection, cycle: str, item: str, status: str, code: str) -> None:
    with transaction(con):
        con.execute("UPDATE cycles SET status=? WHERE id=?", (status, cycle))
        if status == "ERROR":
            con.execute("UPDATE workpieces SET state='ERROR' WHERE id=?", (item,))
            con.execute("INSERT INTO errors(cycle_id,workpiece_id,code,message) VALUES(?,?,?,?)",
                        (cycle, item, code, code))
        event(con, f"{cycle}:fault:{code}", cycle, item, status, {"code":code})


def scan(con: sqlite3.Connection, cycle: str, data: dict, delivery: str) -> str:
    data = validate_payload(data)
    encoded = canonical(data)
    with transaction(con):
        row = con.execute("SELECT payload_json,state FROM workpieces WHERE id=?", (data["id"],)).fetchone()
        if row is None:
            raise RuntimeError("unknown-workpiece")
        if row["payload_json"] != encoded:
            raise RuntimeError("conflicting-payload")
        inserted = event(con, f"{cycle}:scan:{delivery}", cycle, data["id"], "QR_SCANNED", data)
        if row["state"] == "AT_CAMERA":
            con.execute("UPDATE workpieces SET state='IDENTIFIED' WHERE id=?", (data["id"],))
        elif row["state"] not in ("IDENTIFIED", "ROUTE_ASSIGNED"):
            raise RuntimeError("scan-state-invalid")
    return "NEW" if inserted else "IDEMPOTENT"


def run_case(schema: Path, case: dict) -> Outcome:
    con = open_memory(schema)
    cycle = f"CYCLE-{case['id']}"
    item = seed(con, cycle)
    reason = "ok"
    try:
        with transaction(con):
            transition(con, cycle, item, "ON_PALLET", "PICKED", f"{cycle}:picked")
        if bool(case["restart"]):
            latch(con, cycle, item, "RECOVERY_REQUIRED", "restart-mid-cycle")
        elif not bool(case["camera"]):
            latch(con, cycle, item, "RECOVERY_REQUIRED", "camera-lost")
        else:
            with transaction(con):
                transition(con, cycle, item, "PICKED", "AT_CAMERA", f"{cycle}:at-camera")
            first = scan(con, cycle, case["payload"], "message-1")
            if bool(case["duplicate_delivery"]):
                assert scan(con, cycle, case["payload"], "message-1") == "IDEMPOTENT"
            with transaction(con):
                transition(con, cycle, item, "IDENTIFIED", "ROUTE_ASSIGNED", f"{cycle}:route")
                destination = con.execute(
                    "SELECT destination_id FROM routes WHERE workpiece_type=? AND enabled=1", (case["payload"]["type"],)
                ).fetchone()
                if destination is None:
                    raise RuntimeError("route-not-found")
                transition(con, cycle, item, "ROUTE_ASSIGNED", "ON_CONVEYOR", f"{cycle}:conveyor")
            if not bool(case["sensor"]):
                latch(con, cycle, item, "RECOVERY_REQUIRED", "sensor-timeout")
            else:
                with transaction(con):
                    transition(con, cycle, item, "ON_CONVEYOR", "STORED", f"{cycle}:stored")
                    con.execute("UPDATE pallet_cells SET workpiece_id=NULL WHERE workpiece_id=?", (item,))
                    con.execute("UPDATE storage_locations SET occupied_by=? WHERE id=?", (item, destination["destination_id"]))
                    con.execute("UPDATE cycles SET status='COMPLETED',finished_at=CURRENT_TIMESTAMP WHERE id=?", (cycle,))
            reason = first.lower()
    except (ValueError, RuntimeError, sqlite3.DatabaseError) as exc:
        reason = str(exc)
        latch(con, cycle, item, "ERROR", reason)

    cycle_status = con.execute("SELECT status FROM cycles WHERE id=?", (cycle,)).fetchone()[0]
    state = con.execute("SELECT state FROM workpieces WHERE id=?", (item,)).fetchone()[0]
    events = con.execute("SELECT COUNT(*) FROM events").fetchone()[0]
    errors = con.execute("SELECT COUNT(*) FROM errors").fetchone()[0]
    cell_empty = con.execute("SELECT workpiece_id IS NULL FROM pallet_cells").fetchone()[0] == 1
    destination_row = con.execute("SELECT id FROM storage_locations WHERE occupied_by=?", (item,)).fetchone()
    outcome = Outcome(str(case["id"]), cycle_status, state, events, errors, cell_empty,
                      None if destination_row is None else destination_row["id"], reason)
    con.close()
    return outcome


def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--schema", type=Path, required=True); p.add_argument("--cases", type=Path, required=True)
    p.add_argument("--out", type=Path, required=True); args = p.parse_args()
    cases = json.loads(args.cases.read_text(encoding="utf-8"))
    results = [run_case(args.schema, case) for case in cases]
    statuses = {r.case_id:r.cycle_status for r in results}
    assert statuses["normal"] == statuses["duplicate-delivery"] == "COMPLETED"
    assert statuses["conflicting-payload"] == statuses["invalid-extra-field"] == "ERROR"
    for key in ("camera-lost", "sensor-timeout", "restart-mid-cycle"):
        assert statuses[key] == "RECOVERY_REQUIRED"
    duplicate = next(r for r in results if r.case_id == "duplicate-delivery")
    normal = next(r for r in results if r.case_id == "normal")
    assert duplicate.events == normal.events
    args.out.write_text(json.dumps([asdict(r) for r in results], ensure_ascii=False, indent=2), encoding="utf-8")
    print("PASS: COMPLETED=2 ERROR=2 RECOVERY_REQUIRED=3")
    return 0


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

python lab22_warehouse.py --schema codes/storage/schema.sql --cases lab22-cases.json --out lab22-results.json

python -m json.tool lab22-results.json