Chapter 12 — Execution, Human-in-the-Loop & Safety Boundaries

Previously: Chapter 11 left us with typed, explained Recommendations from Kev, sitting behind a safety floor that our code owns.

You will build: a closed-loop work-order and fallback controller. It turns validated recommendations into CMMS work orders, asks an authorised operator before a load transfer or shutdown, and keeps working when telemetry or Kev fails.

SmartPlant increment: fallback_rules, cmms_mock_service, audit_log and action_orchestrator.

12.0 The problem, in plain English

A recommendation is just text on a screen until somebody does something about it. The moment a system acts, four new questions appear:

1. Who said yes? Moving production to a backup motor is a decision with consequences. A person with the right authority must approve it.
2. What if we do it twice? Industrial networks drop and retry messages. Two retries must not become two repair jobs.
3. Did it actually happen? “The server said OK” is not proof the work order exists.
4. What if something is down? Sensors freeze, gateways lag, Kev reboots. The plant can’t wait for us to fix it.

This chapter answers each in turn: approvals (§12.2), idempotent execution and verification (§12.2), an audit trail (§12.3), and predefined fallbacks (§12.4).

12.1 Where the model stops and the plant starts

If you’ve worked with PLCs, you know the pattern. Control logic can ask for anything, but protective trips and permits decide what the machine actually does. We treat Kev the same way.

Part I’s five-line boundary (Chapter 7) still holds. Industry adds two more lines:

1. Kev chooses only among the options we allow.
2. Our code validates schema, thresholds, permissions and budgets.
3. A recommendation is not an authorisation. TRANSFER_LOAD and REQUEST_SHUTDOWN need a named operator with an authorised role. This is human-in-the-loop: a person must say yes first.
4. The executor (the code that carries out the decision) performs one idempotent action. That’s either a CMMS work order or a request to the control room. It never sends a machine command.
5. The verifier reads the work order back from the CMMS, the way you’d confirm a write rather than trust the acknowledgement.
6. Every step goes into a hash-chained audit log: the snapshot, the decision, the approval, the execution, the verification and the outcome.
7. The maintenance outcome comes back as labelled data for evaluation and fine-tuning (Chapter 14).

💡 Mental Model: permit-to-work On a real plant, a hot-work permit lets a technician do something dangerous after a named supervisor signs. Kev proposes. A supervisor signs. Only then does the system file the paperwork.

12.2 Guardrails and approvals

Nobody wants a model shifting load at 3 a.m. without a supervisor knowing. Nobody wants a routine inspection waiting on a signature either. So low-risk actions go straight through, and the two that touch production need sign-off.

Three rules keep this safe:

• We check authorisation in code, against the role the approver actually has, not the role the request asked for.
• Silence doesn’t count as consent. A timeout is recorded as approval_timeout and handled like a denial.
• A denial never drops the case. The asset still gets an inspection work order, and the reason goes into the audit trail.

Idempotency

An action is idempotent when doing it twice has the same effect as doing it once. OT networks are flaky and we retry, so this matters. Without it, a retry would hand the planner two jobs for one bearing.

The work order’s idempotency key is asset:action:hour. Retries, duplicate evaluations and replays all reuse the same work order. If the condition worsens within the episode, we escalate the open work order (PRIORITY to URGENT) instead of opening another. Control-room requests are idempotent per work order.

💡 Mental Model: the lift button Pressing the call button for a lift five times doesn’t summon five lifts. The key asset:action:hour plays the same role: repeats are absorbed.

The mock CMMS

We start with the world the orchestrator talks to. These are mocks, so nothing touches a real machine.

Snippet 1: the work order record. idempotency_key is what makes retries safe.

# smartplant-capstone/src/smartplant/cmms_mock_service.py
"""Mock CMMS and control-request services. Nothing here touches a real machine."""
from __future__ import annotations

import itertools
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Iterator, Literal

WOStatus = Literal["open", "in_progress", "completed", "cancelled"]


class CmmsUnavailable(RuntimeError):
    pass

@dataclass(slots=True)
class WorkOrder:
    wo_id: str
    idempotency_key: str
    asset_id: str
    kind: str                 # INSPECT | SCHEDULE_MAINTENANCE | TRANSFER_LOAD | REQUEST_SHUTDOWN
    priority: str             # ROUTINE | PRIORITY | URGENT | CRITICAL
    description: str
    evidence: dict[str, Any]
    status: WOStatus = "open"
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    finding: str | None = None

Snippet 2: the in-memory CMMS. Two switches simulate trouble: fail_next (outage) and drop_next_write (acknowledged but lost).

@dataclass(slots=True)
class InMemoryCmms:
    work_orders: dict[str, WorkOrder] = field(default_factory=dict)
    by_key: dict[str, str] = field(default_factory=dict)
    fail_next: int = 0                       # simulate outages
    drop_next_write: bool = False            # simulate an acknowledged-but-lost write
    _ids: Iterator[int] = field(default_factory=lambda: itertools.count(1001))

Snippet 3: create a work order, idempotently. Notice the early return for a key we’ve seen, and the “acknowledged, never stored” branch that tests the verifier later.

    def create_work_order(self, idempotency_key: str, asset_id: str, kind: str, priority: str,
                          description: str, evidence: dict[str, Any]) -> str:
        if self.fail_next > 0:
            self.fail_next -= 1
            raise CmmsUnavailable("CMMS returned 503")
        if idempotency_key in self.by_key:                 # retries never create duplicates
            return self.by_key[idempotency_key]
        wo_id = f"WO-{next(self._ids)}"
        self.by_key[idempotency_key] = wo_id
        if self.drop_next_write:
            self.drop_next_write = False
            return wo_id                                   # acknowledged, never stored
        self.work_orders[wo_id] = WorkOrder(wo_id, idempotency_key, asset_id, kind, priority,
                                            description, evidence)
        return wo_id

Snippet 4: read, escalate, complete.

    def get_work_order(self, wo_id: str) -> WorkOrder | None:
        return self.work_orders.get(wo_id)

    def escalate_priority(self, wo_id: str, priority: str) -> None:
        self.work_orders[wo_id].priority = priority

    def complete(self, wo_id: str, finding: str) -> None:
        wo = self.work_orders[wo_id]
        wo.status, wo.finding = "completed", finding

Snippet 5: control-room requests. MockControlRequests only records requests. It sends no commands, by design.

@dataclass(slots=True)
class MockControlRequests:
    """Load-transfer and shutdown REQUESTS for the control-room workflow. It sends no commands."""
    requests: dict[str, dict[str, Any]] = field(default_factory=dict)
    _ids: Iterator[int] = field(default_factory=lambda: itertools.count(1))

    def submit(self, kind: str, asset_id: str, approved_by: str, work_order: str) -> str:
        for rid, req in self.requests.items():          # idempotent per work order
            if req["work_order"] == work_order and req["kind"] == kind:
                return rid
        rid = f"CR-{next(self._ids):04d}"
        self.requests[rid] = {"kind": kind, "asset_id": asset_id, "approved_by": approved_by,
                              "work_order": work_order, "status": "pending_control_room"}
        return rid

    def get(self, request_id: str) -> dict[str, Any] | None:
        return self.requests.get(request_id)

📄 Full source: smartplant-capstone/src/smartplant/cmms_mock_service.py

The orchestrator

The orchestrator carries a recommendation to a verified work order. It logs the snapshot and decision, and MONITOR exits early. If approval is needed, it checks the approver’s role and falls back to INSPECT on failure. Then it writes with retry, reads back, escalates if needed, and only submits a control-room request for an approved, verified action.

Snippet 1: imports and the role table. ROLES_ALLOWED is the whole authorisation policy, in one place.

# smartplant-capstone/src/smartplant/action_orchestrator.py
"""Closed-loop execution: Recommendation -> approval -> authorized, idempotent CMMS action -> verification.

A recommendation is not an authorization. High-consequence actions need a named operator with the
right role; everything is written to the hash-chained audit log; every write is read back.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Literal, Mapping, Protocol

from smartplant.audit_log import AuditLog
from smartplant.cmms_mock_service import CmmsUnavailable, InMemoryCmms, MockControlRequests
from smartplant.models import Recommendation

log = logging.getLogger("smartplant.orchestrator")

ROLES_ALLOWED: Mapping[str, frozenset[str]] = {
    "TRANSFER_LOAD": frozenset({"shift_supervisor", "plant_manager"}),
    "REQUEST_SHUTDOWN": frozenset({"shift_supervisor", "plant_manager"}),
}
_RANK = {"ROUTINE": 0, "PRIORITY": 1, "URGENT": 2, "CRITICAL": 3}
Status = Literal["no_action", "executed", "approval_denied", "approval_timeout", "failed_escalated"]

Snippet 2: the approval messages. An ApprovalRequest goes out, an ApprovalDecision comes back. approved=None means nobody answered.

@dataclass(frozen=True, slots=True)
class ApprovalRequest:
    asset_id: str
    action: str
    urgency: str
    explanation: str
    reasons: tuple[str, ...]

@dataclass(frozen=True, slots=True)
class ApprovalDecision:
    approved: bool | None          # None = no answer before the timeout
    operator_id: str
    role: str
    note: str = ""

Snippet 3: the approver is a swappable Protocol.

class Approver(Protocol):
    def approve(self, req: ApprovalRequest) -> ApprovalDecision: ...

Snippet 4: a scripted approver for tests and replays. It records what it was asked, like a real operator would.

@dataclass(slots=True)
class ScriptedApprover:
    """Tests and replays: a fixed answer, recorded like a real operator's."""
    answer: bool | None = True
    operator_id: str = "op-17"
    role: str = "shift_supervisor"
    seen: list[ApprovalRequest] = field(default_factory=list)

    def approve(self, req: ApprovalRequest) -> ApprovalDecision:
        self.seen.append(req)
        return ApprovalDecision(self.answer, self.operator_id, self.role, "scripted")

Snippet 5: a console approver for the hands-on.

class ConsoleApprover:
    """Mini-project 10: prompt the operator in the terminal."""

    def __init__(self, operator_id: str, role: str) -> None:
        self.operator_id, self.role = operator_id, role

    def approve(self, req: ApprovalRequest) -> ApprovalDecision:
        print(f"\n[APPROVAL] {req.action} on {req.asset_id} ({req.urgency})\n  {req.explanation}")
        ans = input("  approve? [y/N/skip] ").strip().lower()
        verdict = True if ans == "y" else None if ans == "skip" else False
        return ApprovalDecision(verdict, self.operator_id, self.role, "console")

Snippet 6: the result type.

@dataclass(frozen=True, slots=True)
class ExecutionResult:
    status: Status
    action_taken: str
    work_order_id: str | None
    control_request_id: str | None
    verified: bool
    detail: str

Snippet 7: the orchestrator and its dependencies.

@dataclass(slots=True)
class ActionOrchestrator:
    cmms: InMemoryCmms
    control: MockControlRequests
    approver: Approver
    audit: AuditLog
    max_write_attempts: int = 2

Snippet 8: execute(), part 1. Log, then short-circuit MONITOR. The correlation ID ties every later event for this snapshot together.

    def execute(self, rec: Recommendation, state: Mapping[str, Any],
                now: datetime | None = None) -> ExecutionResult:
        now = now or datetime.now(timezone.utc)
        corr = f"{rec.asset_id}:{rec.snapshot_hash}"
        self.audit.append("snapshot", corr, dict(state))
        self.audit.append("decision", corr, rec.to_dict())

        if rec.action == "MONITOR":
            return self._done(corr, ExecutionResult("no_action", "MONITOR", None, None, True, "monitoring only"))

Snippet 9: execute(), part 2. Approval. authorized needs both a yes and an allowed role. If either is missing, we inspect instead and say why.

        action, priority, status, approver_id = rec.action, rec.urgency, "executed", None
        if rec.requires_approval:
            d = self.approver.approve(ApprovalRequest(rec.asset_id, rec.action, rec.urgency,
                                                      rec.explanation, rec.reasons))
            authorized = d.approved is True and d.role in ROLES_ALLOWED.get(rec.action, frozenset())
            self.audit.append("approval", corr, {"requested": rec.action, "approved": d.approved,
                                                 "operator": d.operator_id, "role": d.role,
                                                 "authorized": authorized, "note": d.note})
            if not authorized:
                # Safe default: never silently drop the case; inspect instead and say why.
                status = "approval_timeout" if d.approved is None else "approval_denied"
                action = "INSPECT"
                priority = "URGENT" if rec.urgency in ("URGENT", "CRITICAL") else rec.urgency
            else:
                approver_id = d.operator_id

⚠️ Watch Out d.role in ROLES_ALLOWED[...] checks the role the approver actually holds. Never trust a role named in the request.

Snippet 10: execute(), part 3. Write with retry, then verify. The key makes the write idempotent. A None means the CMMS stayed down.

        key = f"{rec.asset_id}:{action}:{now:%Y%m%d%H}"          # one work order per asset/action/hour
        wo_id = self._create_with_retry(key, rec, action, priority, corr)
        if wo_id is None:
            return self._done(corr, ExecutionResult("failed_escalated", action, None, None, False,
                                                    "CMMS unavailable; paging maintenance supervisor"))
        wo_id, verified = self._verify(wo_id, key, rec, action, priority, corr)

Snippet 11: execute(), part 4. Escalate, don’t duplicate.

        existing = self.cmms.get_work_order(wo_id)
        note = ""
        if verified and existing is not None and _RANK[priority] > _RANK[existing.priority]:
            # Same episode, worse condition: escalate the open work order instead of opening another.
            old = existing.priority
            self.cmms.escalate_priority(wo_id, priority)
            self.audit.append("execution", corr, {"escalated": wo_id, "from": old, "to": priority})
            note = f" (escalated {old}->{priority})"

Snippet 12: execute(), part 5. Control-room request only if verified and approved.

        cr_id = None
        if verified and approver_id and action in ROLES_ALLOWED:
            cr_id = self.control.submit(action, rec.asset_id, approver_id, wo_id)
        return self._done(corr, ExecutionResult(status, action, wo_id, cr_id, verified,
                                                f"{action} via {wo_id}{note}" + (f" + {cr_id}" if cr_id else "")))

Snippet 13: close the loop with the outcome. What the technician found becomes labelled data for Chapter 14.

    def record_outcome(self, rec: Recommendation, wo_id: str, finding: str,
                       expert_labels: Mapping[str, Any] | None = None) -> None:
        """Close the loop with what the technician actually found. Expert labels feed fine-tuning (Ch13)."""
        self.cmms.complete(wo_id, finding)
        self.audit.append("outcome", f"{rec.asset_id}:{rec.snapshot_hash}",
                          {"work_order": wo_id, "finding": finding, "expert_labels": dict(expert_labels or {})})

Snippet 14: retry the write. A failed attempt is logged. Two attempts, then give up and escalate.

    # ---- internals ----
    def _create_with_retry(self, key: str, rec: Recommendation, action: str, priority: str,
                           corr: str) -> str | None:
        for attempt in range(1, self.max_write_attempts + 1):
            try:
                return self.cmms.create_work_order(key, rec.asset_id, action, priority, rec.explanation,
                                                   {"snapshot_hash": rec.snapshot_hash, "reasons": list(rec.reasons),
                                                    "model": rec.model, "source": rec.source})
            except CmmsUnavailable as e:
                log.warning("CMMS write failed (attempt %d): %s", attempt, e)
                self.audit.append("execution", corr, {"error": str(e), "attempt": attempt})
        return None

Snippet 15: verify the write by reading it back. If the CMMS acknowledged a write that it didn’t store, we re-issue under a new key and check again.

    def _verify(self, wo_id: str, key: str, rec: Recommendation, action: str, priority: str,
                corr: str) -> tuple[str, bool]:
        wo = self.cmms.get_work_order(wo_id)
        ok = wo is not None and wo.kind == action and wo.status == "open"
        if not ok:   # acknowledged but not stored: write again under a new key, then re-read
            retry_id = self._create_with_retry(f"{key}:reissue", rec, action, priority, corr)
            if retry_id:
                wo_id = retry_id
                again = self.cmms.get_work_order(retry_id)
                ok = again is not None and again.kind == action
        self.audit.append("verification", corr, {"work_order": wo_id, "verified": ok})
        return wo_id, ok

Snippet 16: log the result.

    def _done(self, corr: str, result: ExecutionResult) -> ExecutionResult:
        self.audit.append("execution", corr, {"status": result.status, "action": result.action_taken,
                                              "work_order": result.work_order_id,
                                              "control_request": result.control_request_id,
                                              "verified": result.verified, "detail": result.detail})
        return result

📄 Full source: smartplant-capstone/src/smartplant/action_orchestrator.py

In practice, a flaky CMMS costs you a retry, not a duplicate job. A wrong-role approval costs you an inspection, not a shutdown.

12.3 Auditability and outcome tracking

After an incident, someone will ask what the system saw, what Kev said, which rules fired, who approved, what happened and what the technician found. That might be an auditor, or you six months later. The log must answer all of it.

We hash-chain the log. Each record carries the hash of the previous one, so editing or deleting history is detectable.

💡 Mental Model: numbered pages in a ledger Each page quotes the fingerprint of the page before it. Tear one out or rewrite one and every later fingerprint stops matching.

Snippet 1: imports and the fingerprint function.

# smartplant-capstone/src/smartplant/audit_log.py
"""Append-only, hash-chained audit trail: snapshot -> decision -> rules -> approval -> action -> outcome.

Each record carries the SHA-256 of the previous one, so editing or deleting any line breaks
`verify_chain`. Ship the file (or each record) to WORM storage for real immutability.
"""
from __future__ import annotations

import hashlib
import json
import threading
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Literal

EventType = Literal["snapshot", "decision", "approval", "execution", "verification", "outcome", "fallback"]
GENESIS = "0" * 64

def _digest(prev: str, body: dict[str, Any]) -> str:
    return hashlib.sha256((prev + json.dumps(body, sort_keys=True, default=str)).encode()).hexdigest()

Snippet 2: the log object. It resumes from the last hash if the file already exists.

@dataclass(slots=True)
class AuditLog:
    path: Path
    _last: str = GENESIS
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def __post_init__(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        if self.path.exists():
            lines = self.path.read_text(encoding="utf-8").splitlines()
            if lines:
                self._last = json.loads(lines[-1])["hash"]

Snippet 3: append chains each new record to the last hash. The lock keeps concurrent writers from forking the chain.

    def append(self, event: EventType, correlation_id: str, payload: dict[str, Any]) -> str:
        with self._lock:
            body = {"ts": datetime.now(timezone.utc).isoformat(), "event": event,
                    "correlation_id": correlation_id, "payload": payload, "prev": self._last}
            h = _digest(self._last, body)
            with self.path.open("a", encoding="utf-8") as f:
                f.write(json.dumps({**body, "hash": h}, default=str) + "\n")
            self._last = h
            return h

Snippet 4: read events back.

    def events(self, correlation_id: str | None = None) -> list[dict[str, Any]]:
        if not self.path.exists():
            return []
        rows = [json.loads(l) for l in self.path.read_text(encoding="utf-8").splitlines() if l]
        return [r for r in rows if correlation_id is None or r["correlation_id"] == correlation_id]

Snippet 5: verify_chain replays from the all-zeros GENESIS value.

def verify_chain(path: Path) -> bool:
    prev = GENESIS
    for line in path.read_text(encoding="utf-8").splitlines():
        rec = json.loads(line)
        h = rec.pop("hash")
        if rec["prev"] != prev or _digest(prev, rec) != h:
            return False
        prev = h
    return True

Snippet 6: can we reconstruct this decision? audit_complete checks that the snapshot, decision and execution records are all present.

REQUIRED_EVENTS = ("snapshot", "decision", "execution")


def audit_complete(log: AuditLog, correlation_id: str) -> bool:
    """Can this decision be reconstructed? Input snapshot, decision and execution must all be present."""
    seen = {e["event"] for e in log.events(correlation_id)}
    return all(ev in seen for ev in REQUIRED_EVENTS)

📄 Full source: smartplant-capstone/src/smartplant/audit_log.py

⚠️ Watch Out A hash chain makes tampering evident, not impossible. For real immutability, ship each record to WORM (write-once) storage, such as object lock or an append-only DB, as it’s written. Keep the chain for fast verification.

12.4 Fallback and resilience

Things fail on a factory floor all the time. A transmitter freezes, a gateway buffers for ten minutes, the Kev box reboots during a driver update. So every failure has a predefined, deterministic path: the same failure always gets the same response.

A latch works as it does in a PLC: once set, it stays set until someone resets it. A circuit breaker stops calling a failing service after repeated errors, then tries again after a cooldown.

The fallback rules double as the rule-based baseline in Chapter 13, so you benchmark exactly what runs when Kev is down. safety_floor sets the minimum response; a model may go above it but never below. The rules never recommend TRANSFER_LOAD, and explain uses recorded facts only.

Snippet 1: ranks and imports. Ranks let code compare “more urgent than” with plain integers.

# smartplant-capstone/src/smartplant/fallback_rules.py
"""Deterministic local rules: the safety floor, the Kev-down fallback, and the rule-based baseline.

These run on the edge with no model and no network. They are conservative on purpose:
when in doubt they inspect and escalate, and they never recommend TRANSFER_LOAD.
"""
from __future__ import annotations

from typing import Any, Mapping

from smartplant.contract import CONTRACT_VERSION, Action, Urgency
from smartplant.models import Recommendation

ACTION_RANK: dict[str, int] = {
    "MONITOR": 0, "INSPECT": 1, "SCHEDULE_MAINTENANCE": 2, "TRANSFER_LOAD": 3, "REQUEST_SHUTDOWN": 4,
}
URGENCY_RANK: dict[str, int] = {"ROUTINE": 0, "PRIORITY": 1, "URGENT": 2, "CRITICAL": 3}

Snippet 2: the safety floor. The same function Chapter 11 used.

def safety_floor(state: Mapping[str, Any]) -> tuple[Urgency, Action] | None:
    """Minimum response implied by code-computed facts. A model may escalate past it, never below it."""
    t = state["telemetry"]
    if state["protective_limit_reached"] or t["temperature_band"] == "trip":
        return "CRITICAL", "REQUEST_SHUTDOWN"
    if t["vibration_zone"] == "D_danger" or (
            t["temperature_band"] == "high" and t["temperature_trend"] == "rising_fast"):
        return "URGENT", "INSPECT"
    if t["temperature_band"] == "high" or t["vibration_zone"] == "C_alert":
        return "PRIORITY", "INSPECT"
    return None

Snippet 3: the rule baseline, first branch. A protective limit means CRITICAL / REQUEST_SHUTDOWN. Invalid data means inspect and escalate.

def rule_baseline(state: Mapping[str, Any], *, source: str = "rules", reason: str = "rule_baseline") -> Recommendation:
    """Static thresholds and if/else: Engine A in the Chapter 12 benchmark, and the Kev-down fallback."""
    t, dq = state["telemetry"], state["data_quality"]["status"]
    reasons = [reason]
    floor = safety_floor(state)
    if floor and floor[1] == "REQUEST_SHUTDOWN":
        urgency, action, sev, route = "CRITICAL", "REQUEST_SHUTDOWN", 5, "specialist_review"
        reasons.append("protective_limit_reached: PLC trip is authoritative")

    elif dq == "invalid":
        urgency, action, sev, route = "PRIORITY", "INSPECT", 3, "specialist_review"
        reasons.append(f"untrusted_data: {','.join(state['data_quality']['issues'])}")

Snippet 4: the rule baseline, the middle. Floor-based responses, then early-trend detection, then the quiet default.

    elif floor:
        urgency, action = floor
        sev = 4 if urgency == "URGENT" else 3
        route = "specialist_review" if urgency == "URGENT" else "standard_cmms"
        reasons.append(f"threshold: vib={t['vibration_zone']} temp={t['temperature_band']}/{t['temperature_trend']}")

    elif t["vibration_trend"] in ("rising", "rising_fast") or (
            t["temperature_band"] == "elevated" and t["temperature_trend"] in ("rising", "rising_fast")):
        urgency, action, sev, route = "PRIORITY", "INSPECT", 2, "standard_cmms"
        reasons.append("early_degradation_trend")
    else:
        urgency, action, sev, route = "ROUTINE", "MONITOR", 1, "standard_cmms"
    if dq == "degraded" and route == "standard_cmms" and action != "MONITOR":
        route = "specialist_review"
        reasons.append("degraded_data")

Snippet 5: the rule baseline, the result.

    return Recommendation(
        asset_id=state["asset"]["id"], urgency=urgency, action=action, severity=sev, route=route,  # type: ignore[arg-type]
        requires_approval=action in ("TRANSFER_LOAD", "REQUEST_SHUTDOWN"), source=source,
        reasons=tuple(reasons), explanation=explain(state, action, urgency, reasons),  # type: ignore[arg-type]
        snapshot_hash=state["snapshot_hash"], contract_version=CONTRACT_VERSION, model="rules-1.0",
    )

Snippet 6: the Kev-down fallback. Without a model opinion, anything beyond monitoring gets a human second look.

def fallback_recommendation(state: Mapping[str, Any], why: str) -> Recommendation:
    """Used when Kev is unreachable, the breaker is open, or the data cannot be trusted."""
    rec = rule_baseline(state, source="fallback", reason=f"fallback:{why}")
    if rec.action != "MONITOR" and rec.route != "specialist_review":
        # Without a model opinion, anything beyond monitoring gets a human second look.
        return Recommendation(**{**rec.to_dict(), "reasons": rec.reasons + ("fallback_needs_review",),
                                 "route": "specialist_review"})
    return rec

Snippet 7: explanations from facts only. Every phrase comes from a recorded field. Nothing is invented.

def explain(state: Mapping[str, Any], action: str, urgency: str, reasons: list[str] | tuple[str, ...]) -> str:
    """Assemble an explanation from recorded facts only. Never invent a rationale."""
    t, p, r = state["telemetry"], state["production"], state["redundancy"]
    facts = [f"vibration {t['vibration_zone']} ({t['vibration_trend']})",
             f"bearing temperature {t['temperature_band']} ({t['temperature_trend']})",
             f"load {t['load_band']}"]
    if p["order_priority"] in ("HIGH", "CRITICAL"):
        facts.append(f"a {p['order_priority'].lower()}-priority order is running "
                     "(considered for planning, never overriding equipment protection)")
    facts.append(f"backup coverage {r['backup_coverage']}")
    if state["data_quality"]["status"] != "valid":
        facts.append(f"data quality {state['data_quality']['status']}")
    return f"{action} at {urgency} urgency because " + "; ".join(facts) + "."

📄 Full source: smartplant-capstone/src/smartplant/fallback_rules.py

12.5 Mini-project 10: a closed-loop work-order and fallback controller

The controller runs a degrading motor through an hour of simulated time. Deciding on every 10-second sample would flood the CMMS the way a badly tuned alarm floods a SCADA screen. So it decides on meaningful change (zone, band, data quality, trip) or on a heartbeat, a periodic check even when nothing changed. It also debounces: a new zone must hold for three samples before it triggers a decision, because trends jitter and zones don’t. Partway through, the demo pulls the plug on the decision service.

Snippet 1: usage and imports. The docstring doubles as the run recipe.

# mini-projects/10-closed-loop-controller/run_controller.py
"""Mini-project 10: closed loop with operator approval and a Kev outage mid-run.

uv run python mini-projects/10-closed-loop-controller/run_controller.py --offline --auto-approve
DECISION_BACKEND=kev 
uv run python mini-projects/10-closed-loop-controller/run_controller.py --kev-down-at 120
uv run python mini-projects/10-closed-loop-controller/run_controller.py --offline --steps 61 --every 6 --fault-start 6 --magnitude 2
"""

from __future__ import annotations

import argparse
from datetime import timedelta
from pathlib import Path

from triage.jev_client import DecisionClient, FakeDecisionClient, client_from_env
from triage.resilience import CircuitBreaker

from smartplant.action_orchestrator import ActionOrchestrator, ConsoleApprover, ScriptedApprover
from smartplant.audit_log import AuditLog, verify_chain
from smartplant.cmms_mock_service import InMemoryCmms, MockControlRequests
from smartplant.context_enricher import AssetRecord, BackupAsset, MockPlantData, ProductionOrder, enrich, summarize
from smartplant.decision_service import SmartPlantDecisionService
from smartplant.edge_state_builder import EdgeStateBuilder
from smartplant.fault_injector import Fault, FaultMode
from smartplant.scenario_runner import DownClient, scripted_model
from smartplant.sensor_simulator import AssetProfile, SensorSimulator

Snippet 2: SwitchableClient simulates the outage.

class SwitchableClient:
    """Lets the demo pull the plug on the decision service at a chosen step."""

    def __init__(self, inner: DecisionClient) -> None:
        self.inner, self.down, self.backend = inner, False, inner.backend

    def decide(self, state, questions):
        return (DownClient() if self.down else self.inner).decide(state, questions)

Snippet 3: command-line options.

def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--steps", type=int, default=360)
    ap.add_argument("--every", type=int, default=30, help="decide every N steps (or on band change)")
    ap.add_argument("--offline", action="store_true")
    ap.add_argument("--auto-approve", action="store_true")
    ap.add_argument("--kev-down-at", type=int, default=None)
    ap.add_argument("--fault-start", type=int, default=40, help="step where bearing wear begins")
    ap.add_argument("--magnitude", type=float, default=0.8, help="wear rate (2 reaches zone D in ~9 minutes)")
    ap.add_argument("--audit", type=Path, default=Path("audit/smartplant.jsonl"))
    args = ap.parse_args()

Snippet 4: wire up the world. A simulator with a fault, an edge builder, mock plant data, a service and an orchestrator.

    profile = AssetProfile("M-104")
    sim = SensorSimulator(profile, load=lambda _s: 0.9,
                          faults=[Fault(FaultMode.GRADUAL_DEGRADATION, args.fault_start, magnitude=args.magnitude)])
    builder = EdgeStateBuilder(profile)

    plant = MockPlantData(
        orders={"M-104": ProductionOrder("PO-7781", "HIGH", 30, 0.85)},
        assets={"M-104": AssetRecord("M-104", "A", 410, 0, True)},
        backups={"M-104": BackupAsset("M-105", True, True, 0.95)},
    )

    client = SwitchableClient(FakeDecisionClient(scripted_model) if args.offline else client_from_env())
    service = SmartPlantDecisionService(client, CircuitBreaker(failure_threshold=2, cooldown_s=60))
    approver = ScriptedApprover(True) if args.auto_approve else ConsoleApprover("op-17", "shift_supervisor")
    orch = ActionOrchestrator(InMemoryCmms(), MockControlRequests(), approver, AuditLog(args.audit))

Snippet 5: the loop starts. Track the last key and a debounce counter, feed each sample to the builder, and pull the plug at the chosen step.

    last_key: tuple[object, ...] | None = None
    pending: tuple[object, ...] | None = None
    pending_count = 0
    for r in sim.stream(args.steps):
        builder.add(r)
        if args.kev_down_at is not None and r.seq == args.kev_down_at:
            client.down = True
            print(f"--- t={r.seq}: decision service unreachable ---")
        if r.seq < builder.limits.min_samples:          # warm-up: not enough evidence yet
            continue
        snap = builder.snapshot(r.ts + timedelta(seconds=1))

Snippet 6: decide only on meaningful change. The key tuple captures what matters. A new key must hold for three samples (30 s) before we act. A heartbeat covers the quiet periods.

        # Decide on a meaningful change (zones/bands/quality/trip) or on a heartbeat,
        # not on every 10 s sample: trends jitter, zones don't.
        key = (snap.bands["vibration_zone"], snap.bands["temperature_band"], snap.data_quality,
               snap.protective_limit_reached)
        if key != last_key and not snap.protective_limit_reached:
            # Debounce: a new zone must hold for 3 samples (30 s) before we act on it.
            pending_count = pending_count + 1 if key == pending else 1
            pending = key
            if pending_count < 3:
                continue
        elif r.seq % args.every != 0 and key == last_key:
            continue
        last_key = key

💡 Mental Model: trends jitter, zones don’t A slope computed over a few samples wobbles. A zone label either changed or it didn’t. We watch the labels.

Snippet 7: decide, execute, print, stop on a trip.

        state = enrich(snap, plant, plant, plant)
        rec = service.evaluate(state)
        res = orch.execute(rec, state, now=r.ts)
        print(f"t={r.seq:4d} {summarize(state)}\n       -> {rec.urgency}/{rec.action} [{rec.source}] "
              f"=> {res.status} {res.detail}")

        if snap.protective_limit_reached:
            print("PLC protective trip latched: motor stopped by the PLC; loop ends until operator reset.")
            break
    print(f"audit chain intact: {verify_chain(args.audit)}  ({args.audit})")
    return 0

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

📄 Full source: mini-projects/10-closed-loop-controller/run_controller.py

Run it offline, or against a real Kev server:

uv run python mini-projects/10-closed-loop-controller/run_controller.py --offline --auto-approve --kev-down-at 120
# with a real Kev server and a human at the keyboard:
DECISION_BACKEND=kev 
uv run python mini-projects/10-closed-loop-controller/run_controller.py --kev-down-at 120

A typical offline run looks like this:

t=   8 M-104: vib A_good/rising, temp normal/stable, load heavy, data valid, order HIGH, backup full
       -> ROUTINE/MONITOR [fake] => no_action monitoring only
t=  30 M-104: vib A_good/stable, temp normal/stable, load heavy, data valid, order HIGH, backup full
       -> ROUTINE/MONITOR [fake] => no_action monitoring only
t=  46 M-104: vib B_acceptable/rising, temp normal/stable, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/INSPECT [fake] => executed INSPECT via WO-1001
t=  60 M-104: vib B_acceptable/rising, temp normal/rising, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/INSPECT [fake] => executed INSPECT via WO-1001
t=  90 M-104: vib B_acceptable/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/INSPECT [fake] => executed INSPECT via WO-1001
t=  91 M-104: vib C_alert/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fake] => executed INSPECT via WO-1001 (escalated PRIORITY->URGENT)
--- t=120: decision service unreachable ---
t= 120 M-104: vib C_alert/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/INSPECT [fallback] => executed INSPECT via WO-1001
circuit OPEN after 2 failures
t= 147 M-104: vib D_danger/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
t= 150 M-104: vib D_danger/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
t= 161 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
t= 180 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
t= 210 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
t= 226 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> CRITICAL/REQUEST_SHUTDOWN [safety_rules] => executed REQUEST_SHUTDOWN via WO-1002 + CR-0001
PLC protective trip latched: motor stopped by the PLC; loop ends until operator reset.
audit chain intact: True  (audit/smartplant.jsonl)

The actual Kev output:

uv run python mini-projects/10-closed-loop-controller/run_controller.py --kev-down-at 120

[APPROVAL] TRANSFER_LOAD on M-104 (PRIORITY)
  TRANSFER_LOAD at PRIORITY urgency because vibration A_good (rising); bearing temperature normal (stable); load heavy; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage full.
  approve? [y/N/skip] y

t=   8 M-104: vib A_good/rising, temp normal/stable, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/TRANSFER_LOAD [kev] => executed TRANSFER_LOAD via WO-1001 + CR-0001
t=  30 M-104: vib A_good/stable, temp normal/stable, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/MONITOR [kev] => no_action monitoring only

What to notice:

• t=46–91: one work order (WO-1001) covers the whole episode. It’s escalated from PRIORITY to URGENT when the zone worsens, with no duplicates.
• t=120: Kev goes away, and the very next decision comes from fallback. After two failures the breaker opens and the service stops calling a dead endpoint.
• t=147–210: vibration reaches zone D with no model available. The fallback never recommends TRANSFER_LOAD, which needs a model opinion and a human. The code safety floor keeps urgency at URGENT.
• t=226: the protective limit is reached. The service records safety_rules, documents a controlled-shutdown request and stops, because the PLC already tripped the motor.
• End: the audit chain verifies.

Hands-on: run the closed loop for 10 simulated minutes

The defaults above simulate an hour. For learning, it’s easier to squeeze the same story into ten minutes: wear starts at step 6, magnitude 2 reaches zone D by minute 9, and the controller re-checks at least once a minute (--every 6). Each step has an offline variant; outputs shown are from offline runs. With Kev in terminal 1 (as in Chapter 11’s hands-on), drop --offline and set DECISION_BACKEND=kev.

export KEV_BASE_URL=http://127.0.0.1:8009        # terminal 2, only for the Kev runs
alias loop='uv run python mini-projects/10-closed-loop-controller/run_controller.py --steps 61 --every 6 --fault-start 6 --magnitude 2'

(The alias just saves typing. On Windows PowerShell, paste the full command instead.)

Step 1: A clean run with automatic approval. Start with a fresh audit log so the chain begins at the genesis record.

rm -rf audit
loop --offline --auto-approve                    # or: DECISION_BACKEND=kev loop --auto-approve

t=   8 M-104: vib A_good/rising_fast, temp normal/rising, load heavy, data valid, order HIGH, backup full
       -> ROUTINE/MONITOR [fake] => no_action monitoring only
t=  11 M-104: vib B_acceptable/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/INSPECT [fake] => executed INSPECT via WO-1001
t=  29 M-104: vib C_alert/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fake] => executed INSPECT via WO-1001 (escalated PRIORITY->URGENT)
t=  50 M-104: vib D_danger/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> CRITICAL/TRANSFER_LOAD [fake] => executed TRANSFER_LOAD via WO-1002 + CR-0001
t=  60 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> CRITICAL/TRANSFER_LOAD [fake] => executed TRANSFER_LOAD via WO-1002 + CR-0001
audit chain intact: True  (audit/smartplant.jsonl)

(Trimmed.) Three things to notice. One work order, WO-1001, follows the inspection from zone B to zone C and is escalated rather than duplicated. Zone D creates a second work order plus a control-room request, CR-0001. And the repeated decisions at t=54 and t=60 reuse both, because the actions are idempotent.

Step 2: Be the shift supervisor. Without --auto-approve, the controller stops and asks you before any load transfer. Answer y the first time.

rm -rf audit
loop --offline

[APPROVAL] TRANSFER_LOAD on M-104 (CRITICAL)
  TRANSFER_LOAD at CRITICAL urgency because vibration D_danger (rising_fast); bearing temperature normal (rising_fast); load heavy; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage full.
  approve? [y/N/skip] y
t=  50 M-104: vib D_danger/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> CRITICAL/TRANSFER_LOAD [fake] => executed TRANSFER_LOAD via WO-1002 + CR-0001

Step 3: Say no. Run it again and answer n (or just press Enter, since silence is not consent).

rm -rf audit
loop --offline

  approve? [y/N/skip] n
t=  60 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> CRITICAL/TRANSFER_LOAD [fake] => approval_denied INSPECT via WO-1001

The denial doesn’t drop the case. The motor still gets an inspection work order at the same urgency, and the refusal is in the audit trail.

Step 4: Read the audit trail. Each decision leaves a linked chain of events under one correlation ID: snapshot, decision, approval, verification, execution.

jq -r '[.event, .correlation_id, (.payload.action // .payload.status // "")] | @tsv' audit/smartplant.jsonl | tail -8
jq -c 'select(.event=="approval") | .payload' audit/smartplant.jsonl

{"requested":"TRANSFER_LOAD","approved":true,"operator":"op-17","role":"shift_supervisor","authorized":true,"note":"console"}

The approval records who approved, the role they actually hold, and whether that role was authorised, which is what an auditor asks for after an incident.

Step 5: Pull the plug on Kev at step 40 (minute 6.7).

rm -rf audit
loop --offline --auto-approve --kev-down-at 40   # or: DECISION_BACKEND=kev loop --auto-approve --kev-down-at 40

--- t=40: decision service unreachable ---
t=  42 M-104: vib C_alert/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> PRIORITY/INSPECT [fallback] => executed INSPECT via WO-1001
circuit OPEN after 2 failures
t=  50 M-104: vib D_danger/rising_fast, temp normal/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
t=  60 M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
       -> URGENT/INSPECT [fallback] => executed INSPECT via WO-1001
audit chain intact: True  (audit/smartplant.jsonl)

Compare this with step 1. With the model gone, zone D no longer produces TRANSFER_LOAD: the fallback never proposes an action that needs a model opinion and a human. It keeps the inspection open at URGENT instead, and the breaker stops calling the dead endpoint after two failures.

Step 6: Try to rewrite history. Change one recorded decision, then verify the chain.

cp audit/smartplant.jsonl audit/backup.jsonl
sed -i.bak '2s/"urgency": "ROUTINE"/"urgency": "CRITICAL"/' audit/smartplant.jsonl
uv run python -c "from pathlib import Path; from smartplant.audit_log import verify_chain; print(verify_chain(Path('audit/smartplant.jsonl')))"

False

One edited field breaks every hash after it. Restore the copy (cp audit/backup.jsonl audit/smartplant.jsonl) before moving on, because Chapter 14 uses this audit log.

Step 7: Keep the audit log for Chapter 14. Run step 1 once more so audit/smartplant.jsonl holds a clean run with two work orders. Chapter 14 closes those work orders with technician findings and turns them into training data.

Tests

These tests cover approvals and roles, idempotency, outages and lost writes, tamper detection, training-data export and escalation.

Snippet 1: helpers. orch() builds an orchestrator with a scripted approver. as_action() forces a recommendation to a given action.

# smartplant-capstone/tests/test_orchestrator.py
from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path

from smartplant.action_orchestrator import ActionOrchestrator, ScriptedApprover
from smartplant.audit_log import AuditLog, audit_complete, verify_chain
from smartplant.cmms_mock_service import InMemoryCmms, MockControlRequests
from smartplant.fallback_rules import rule_baseline
from smartplant.models import Recommendation

NOW = datetime(2026, 10, 1, 8, 0, tzinfo=timezone.utc)

def orch(tmp_path: Path, answer=True, role="shift_supervisor", cmms: InMemoryCmms | None = None):
    approver = ScriptedApprover(answer, role=role)
    return ActionOrchestrator(cmms or InMemoryCmms(), MockControlRequests(), approver,
                              AuditLog(tmp_path / "audit.jsonl")), approver


def as_action(rec: Recommendation, action: str) -> Recommendation:
    return Recommendation(**{**rec.to_dict(), "reasons": rec.reasons, "action": action,
                             "requires_approval": action in ("TRANSFER_LOAD", "REQUEST_SHUTDOWN")})

Snippet 2: MONITOR creates nothing.

def test_monitor_creates_nothing(tmp_path, state_for) -> None:
    o, _ = orch(tmp_path)
    st = state_for("S01")
    res = o.execute(rule_baseline(st), st, NOW)
    assert res.status == "no_action" and not o.cmms.work_orders

Snippet 3: idempotency. Run twice, get one work order.

def test_inspect_creates_verified_work_order_idempotently(tmp_path, state_for) -> None:
    o, _ = orch(tmp_path)
    st = state_for("S02")
    rec = rule_baseline(st)
    a, b = o.execute(rec, st, NOW), o.execute(rec, st, NOW)
    assert a.verified and a.work_order_id == b.work_order_id and len(o.cmms.work_orders) == 1

Snippet 4: approval and roles. A wrong role is denied, and silence times out. Both fall back to an inspection.

def test_transfer_needs_authorized_approval(tmp_path, state_for) -> None:
    st = state_for("S06")
    rec = as_action(rule_baseline(st), "TRANSFER_LOAD")
    o, approver = orch(tmp_path)
    ok = o.execute(rec, st, NOW)
    assert ok.status == "executed" and ok.control_request_id and approver.seen
    o2, _ = orch(tmp_path / "b", role="maintenance_tech")          # wrong role
    denied = o2.execute(rec, st, NOW)
    assert denied.status == "approval_denied" and denied.action_taken == "INSPECT"
    o3, _ = orch(tmp_path / "c", answer=None)                       # nobody answered
    assert o3.execute(rec, st, NOW).status == "approval_timeout"

Snippet 5: outage and lost write. Five CMMS failures end in failed_escalated. A lost write is detected and re-issued.

def test_cmms_outage_and_lost_write(tmp_path, state_for) -> None:
    st = state_for("S02")
    rec = rule_baseline(st)
    o, _ = orch(tmp_path, cmms=InMemoryCmms(fail_next=5))
    assert o.execute(rec, st, NOW).status == "failed_escalated"
    o2, _ = orch(tmp_path / "lost", cmms=InMemoryCmms(drop_next_write=True))
    res = o2.execute(rec, st, NOW)
    assert res.verified and res.work_order_id in o2.cmms.work_orders

Snippet 6: tamper evidence. Edit one record, and verify_chain flips to false.

def test_audit_chain_is_complete_and_tamper_evident(tmp_path, state_for) -> None:
    o, _ = orch(tmp_path)
    st = state_for("S02")
    rec = rule_baseline(st)
    res = o.execute(rec, st, NOW)
    o.record_outcome(rec, res.work_order_id, "outer race spalling confirmed; bearing replaced")
    path = tmp_path / "audit.jsonl"
    assert verify_chain(path) and audit_complete(o.audit, f"{rec.asset_id}:{rec.snapshot_hash}")
    lines = path.read_text().splitlines()
    rec1 = json.loads(lines[1])
    rec1["payload"]["urgency"] = "ROUTINE"                           # someone "fixes" history
    lines[1] = json.dumps(rec1)
    path.write_text("\n".join(lines) + "\n")
    assert not verify_chain(path)

Snippet 7: outcomes become training data.

def test_outcomes_with_expert_labels_become_kev_training_data(tmp_path, state_for) -> None:
    from smartplant.training_export import examples_from_audit, write_split
    o, _ = orch(tmp_path)
    st = state_for("S02")
    rec = rule_baseline(st)
    res = o.execute(rec, st, NOW)
    o.record_outcome(rec, res.work_order_id, "outer race spalling",
                     {"intervention_urgency": "URGENT", "operational_action": "SCHEDULE_MAINTENANCE",
                      "standard_workflow": True, "operational_severity": 4})
    ex = examples_from_audit((tmp_path / "audit.jsonl").read_text().splitlines())
    assert len(ex) == 1
    q = ex[0]["questions"]
    assert q["operational_severity"]["label"] == 3 and q["standard_workflow"]["label"] is True
    train, held = write_split(ex, tmp_path / "ft")
    assert train.read_text().count("\n") == 1

Snippet 8: worse condition escalates the open work order.

def test_worse_condition_escalates_open_work_order(tmp_path, state_for) -> None:
    o, _ = orch(tmp_path)
    st = state_for("S02")
    first = rule_baseline(st)                                         # PRIORITY / INSPECT
    worse = Recommendation(**{**first.to_dict(), "reasons": first.reasons, "urgency": "URGENT"})
    a = o.execute(first, st, NOW)
    b = o.execute(worse, st, NOW)
    assert a.work_order_id == b.work_order_id and "escalated PRIORITY->URGENT" in b.detail
    assert o.cmms.work_orders[a.work_order_id].priority == "URGENT"

📄 Full source: smartplant-capstone/tests/test_orchestrator.py

12.6 Checklist

• TRANSFER_LOAD and REQUEST_SHUTDOWN need approval, the role is checked in code, and silence counts as denial.
• Denials and timeouts still produce an inspection work order.
• Work orders are idempotent per asset, action and episode, and are escalated, never duplicated.
• Every write is read back from the CMMS, and lost writes are re-issued.
• The hash-chained audit log covers snapshot, decision, approval, execution, verification and outcome.
• Every failure mode in §12.4 has a test or a scenario.
• Decisions happen on meaningful change, with debouncing, not on every sample.

What’s next

We now have three ways to make the same decision: hand-written rules, a general LLM through the adapter, and Kev behind guardrails. Chapter 13 replays the ten SmartPlant scenarios through all three and measures agreement, critical-case miss rate, unnecessary interventions, validity, latency, cost and fallback success.