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

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

In this chapter. Now we act on them. We 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 gains fallback_rules, cmms_mock_service, audit_log and action_orchestrator.

11.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 6) 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 13).

11.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, much like a permit-to-work.

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.

In the mock services below, notice that create_work_order returns the existing ID for a key it has seen. fail_next and drop_next_write simulate an outage and a write that was acknowledged but lost. MockControlRequests only records requests.

# 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


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

    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

    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


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

The orchestrator carries a recommendation to a verified work order. Follow execute from top to bottom. It logs the snapshot and decision, and MONITOR exits early. If approval is needed, it checks the approver’s role against ROLES_ALLOWED 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.

# 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"]


@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 = ""


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


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


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


@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


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

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

        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

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

    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 {})})

    # ---- 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

    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

    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

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

11.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.

append chains each new record to the last hash. verify_chain replays the chain from the all-zeros GENESIS value. audit_complete confirms a decision can be reconstructed from its snapshot, decision and execution records.

# 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()


@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"]

    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

    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]


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


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)

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.

11.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 12, 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.

# 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}


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


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'])}")
    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")
    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",
    )


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


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

11.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.

SwitchableClient simulates the outage. In the loop, watch the key tuple, the three-sample debounce and the stop once the PLC trip latches.

# 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


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)


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

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

    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))
        # 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
        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())

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 KV 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 10’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 13 uses this audit log.

Step 7: Keep the audit log for Chapter 13. Run step 1 once more so audit/smartplant.jsonl holds a clean run with two work orders. Chapter 13 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.

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


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


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


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"


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


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)


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


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"

11.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 §11.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 12 replays the ten SmartPlant scenarios through all three and measures agreement, critical-case miss rate, unnecessary interventions, validity, latency, cost and fallback success.