Chapter 12 — Scenario Evaluation & Benchmark Suite

Chapter 11 closed the loop: approvals, idempotent work orders, an audit chain and fallbacks are all in place.

In this chapter. Now we need evidence that the whole thing makes good calls. We’ll build a replay suite of 10 SmartPlant scenarios, run a three-way benchmark (rules vs an LLM vs Kev) scored with industrial metrics, and publish the results to a static dashboard. The SmartPlant modules added here are scenario_runner, benchmark_metrics and eval_dashboard.

12.1 Three engines, one contract

Before a new controller goes live, you run acceptance tests on a rig where you control every input. We do the same for decision engines, and the test is only fair if every engine gets byte-identical state, the same contract (the fixed set of questions and allowed answer labels), and the same policy and executor downstream. The only thing that changes between runs is who answers the questions. We compare three engines:

The rules baseline is the yardstick: the hard-coded thresholds a plant would use with no model at all. Engines B and C reuse Part I’s DecisionClient implementations unchanged, so any difference comes from the model, not the plumbing. An offline stand-in, scripted_model, lets the suite run on any laptop. Its numbers measure the pipeline and never tell you anything about Kev.

12.2 The ten scenarios

Each scenario is labelled: a reproducible plant situation paired with the answer expert reviewers accept. That answer is the ground truth we score against, like the expected-result column on a commissioning test sheet.

The expected answers are sets, because expert reviewers agree on a range more often than on a single answer. Some cases (S04, S09 and S10) are meant to go to a human rather than get a forced automatic answer.

12.3 Industrial metrics

One accuracy number hides what matters on a plant. A false trip costs production; a missed trip can cost the motor. So we count them separately. ↓ means lower is better.

The critical-case miss rate is the false-negative rate on cases that could break equipment. p50 is the median decision time; p95 is the time 95 % of decisions come in under.

The ship gate

Metrics only help if they decide something. The ship gate is a pass/fail check, like a site acceptance test. A new engine, checkpoint, contract or policy version must pass it before it leaves shadow mode:

12.4 The suite

Replaying scenarios

scenario_runner is like replaying historian data through a test rig. build_state rebuilds each scenario from scratch (seeded simulator, edge snapshot, enrichment), so every engine sees byte-identical state. The Engine protocol lets rules, LLM and Kev plug in the same way, and DownClient fakes the S10 timeout. scripted_model is the offline stand-in.

# smartplant-capstone/src/smartplant/scenario_runner.py
"""Replay the 10 SmartPlant scenarios through any engine: rules, LLM (adapter), Kev, or an offline stand-in.

Each scenario is rebuilt from scratch (seeded simulator -> edge snapshot -> enrichment), so every
engine sees byte-identical state. Results feed benchmark_metrics and eval_dashboard.
"""
from __future__ import annotations

import argparse
import json
import tempfile
import time
from dataclasses import dataclass
from datetime import timedelta
from importlib import resources
from pathlib import Path
from typing import Any, Mapping, Protocol, Sequence

from triage.jev_client import (Answer, ChoiceResult, DecisionClient, DecisionError, DecisionResponse,
                               FakeDecisionClient, NoulResult, Questions, ScoreResult)
from triage.resilience import CircuitBreaker

from smartplant.action_orchestrator import ActionOrchestrator, ScriptedApprover
from smartplant.audit_log import AuditLog, audit_complete
from smartplant.cmms_mock_service import InMemoryCmms, MockControlRequests
from smartplant.context_enricher import AssetRecord, BackupAsset, MockPlantData, ProductionOrder, enrich
from smartplant.decision_service import SmartPlantDecisionService
from smartplant.edge_state_builder import EdgeStateBuilder
from smartplant.fallback_rules import rule_baseline
from smartplant.fault_injector import Fault, FaultMode
from smartplant.models import Recommendation
from smartplant.sensor_simulator import AssetProfile, SensorSimulator


def load_scenarios(path: str | Path | None = None) -> list[dict[str, Any]]:
    """Packaged with smartplant so every mini-project loads it without path tricks."""
    raw = (Path(path).read_text(encoding="utf-8") if path else
           resources.files("smartplant.data").joinpath("test_scenarios.json").read_text(encoding="utf-8"))
    return json.loads(raw)["scenarios"]


def build_state(spec: Mapping[str, Any], asset_id: str = "M-104") -> dict[str, Any]:
    profile = AssetProfile(asset_id=asset_id)
    faults = [Fault(FaultMode(f["mode"]), f["start_step"], f.get("duration_steps"),
                    f.get("magnitude", 1.0), f.get("channel", "bearing_temp_c")) for f in spec["faults"]]
    sim = SensorSimulator(profile, seed=spec["seed"], load=lambda _s: spec["load"], faults=faults)
    builder = EdgeStateBuilder(profile)
    last = None
    for r in sim.stream(spec["steps"]):
        builder.add(r)
        last = r
    assert last is not None
    snap = builder.snapshot(last.ts + timedelta(seconds=5 + spec.get("stale_seconds", 0)))
    c = spec["context"]
    plant = MockPlantData(
        orders={asset_id: ProductionOrder(f"PO-{spec['id']}", c["order"]["priority"],
                                          c["order"]["hours_to_deadline"], c["order"]["required_load_pct"])}
        if c.get("order") else {},
        assets={asset_id: AssetRecord(asset_id, **c["asset"])},
        backups={asset_id: BackupAsset(**c["backup"])},
    )
    return enrich(snap, plant, plant, plant)


# ---------- Engines ----------
class Engine(Protocol):
    name: str

    def recommend(self, state: Mapping[str, Any], spec: Mapping[str, Any]) -> Recommendation: ...


class RuleEngine:
    name = "rules"

    def recommend(self, state: Mapping[str, Any], spec: Mapping[str, Any]) -> Recommendation:
        return rule_baseline(state)


class DownClient:
    """Simulates the decision service timing out (scenario S10)."""
    backend = "down"

    def decide(self, state: Any, questions: Questions) -> DecisionResponse:
        raise DecisionError("timeout after 5.0s", retryable=True)


@dataclass(slots=True)
class ServiceEngine:
    """Any DecisionClient (Kev, Jev, adapter-LLM, fake) behind the same SmartPlant policy."""
    name: str
    client: DecisionClient

    def recommend(self, state: Mapping[str, Any], spec: Mapping[str, Any]) -> Recommendation:
        client: DecisionClient = DownClient() if spec.get("decision_service_down") else self.client
        svc = SmartPlantDecisionService(client, CircuitBreaker(failure_threshold=3, cooldown_s=30))
        return svc.evaluate(state)


def scripted_model(state: Any, questions: Questions) -> Mapping[str, Answer]:
    """OFFLINE STAND-IN for Kev so the harness runs anywhere. Its numbers measure the pipeline, not Kev."""
    t = state["telemetry"]
    sev = {"A_good": 0.2, "B_acceptable": 1.0, "C_alert": 2.0, "D_danger": 3.0, "unknown": 1.5}[t["vibration_zone"]]
    sev = max(sev, {"normal": 0.2, "elevated": 1.5, "high": 3.0, "trip": 4.0, "unknown": 1.5}[t["temperature_band"]])
    if "rising_fast" in (t["vibration_trend"], t["temperature_trend"]):
        sev = min(4.0, sev + 0.6)
    urgency = "ROUTINE" if sev < 0.9 else "PRIORITY" if sev < 2.3 else "URGENT" if sev < 3.6 else "CRITICAL"
    offered = list(questions["operational_action"]["criteria"])
    if sev >= 3.0 and "TRANSFER_LOAD" in offered:
        action = "TRANSFER_LOAD"
    elif sev >= 0.9:
        action = "INSPECT"
    else:
        action = "MONITOR"
    conflict = 0.85 if t["temperature_trend"] in ("rising", "rising_fast") and t["vibration_trend"] == "stable" else 0.1

    def choice(label: str, labels: Sequence[str], conf: float) -> ChoiceResult:
        rest = (1 - 0.6) / max(1, len(labels) - 1)
        return ChoiceResult(label, {l: (0.6 if l == label else rest) for l in labels}, conf)

    return {
        "intervention_urgency": choice(urgency, ["ROUTINE", "PRIORITY", "URGENT", "CRITICAL"], 0.55),
        "operational_action": choice(action, offered, 0.5),
        "standard_workflow": NoulResult(0.3 if conflict > 0.5 else 0.8),
        "operational_severity": ScoreResult(sev, {}, 0.6, 5),
        "signals_conflict": NoulResult(conflict),
    }


# ---------- Runner ----------
@dataclass(frozen=True, slots=True)
class ScenarioResult:
    scenario_id: str
    scenario_name: str
    engine: str
    recommendation: Recommendation
    wall_ms: float
    audit_complete: bool
    execution_status: str


def run(engines: Sequence[Engine], scenarios: Sequence[Mapping[str, Any]],
        audit_dir: Path | None = None) -> list[ScenarioResult]:
    audit_dir = audit_dir or Path(tempfile.mkdtemp(prefix="smartplant-audit-"))
    out: list[ScenarioResult] = []
    for eng in engines:
        audit = AuditLog(audit_dir / f"{eng.name}.jsonl")
        orch = ActionOrchestrator(InMemoryCmms(), MockControlRequests(), ScriptedApprover(True), audit)
        for spec in scenarios:
            state = build_state(spec)
            t0 = time.perf_counter()
            rec = eng.recommend(state, spec)
            wall = (time.perf_counter() - t0) * 1000
            result = orch.execute(rec, state)
            out.append(ScenarioResult(spec["id"], spec["name"], eng.name, rec, round(wall, 1),
                                      audit_complete(audit, f"{rec.asset_id}:{rec.snapshot_hash}"),
                                      result.status))
    return out


def default_engines(use_kev: bool, use_llm: bool) -> list[Engine]:
    engines: list[Engine] = [RuleEngine()]
    if use_llm:
        from triage.jev_client import AdapterDecisionClient
        engines.append(ServiceEngine("llm-gpt-4o-mini", AdapterDecisionClient("openai", "gpt-4o-mini")))
    if use_kev:
        from triage.jev_client import KevDecisionClient
        engines.append(ServiceEngine("kev", KevDecisionClient()))
    if not use_kev:
        engines.append(ServiceEngine("kev-offline-standin", FakeDecisionClient(scripted_model)))
    return engines


def main(argv: Sequence[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Replay SmartPlant scenarios across engines")
    ap.add_argument("--kev", action="store_true", help="use a running Kev server (KEV_BASE_URL)")
    ap.add_argument("--llm", action="store_true", help="add the LLM baseline via system-one-adapter")
    ap.add_argument("--scenarios", default=None, help="path to a scenarios JSON (default: packaged)")
    ap.add_argument("--out", default="eval_results.json")
    args = ap.parse_args(argv)
    results = run(default_engines(args.kev, args.llm), load_scenarios(args.scenarios))
    rows = [{"scenario": r.scenario_id, "name": r.scenario_name, "engine": r.engine,
             "wall_ms": r.wall_ms, "audit_complete": r.audit_complete, "execution": r.execution_status,
             **r.recommendation.to_dict()} for r in results]
    Path(args.out).write_text(json.dumps(rows, indent=2), encoding="utf-8")
    for r in results:
        rec = r.recommendation
        print(f"{r.scenario_id} {r.engine:22} {rec.urgency:9} {rec.action:21} sev={rec.severity} "
              f"{rec.route:18} src={rec.source:12} {r.wall_ms:7.1f}ms")
    print(f"wrote {args.out}")
    return 0


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

The runner times each decision, executes it through the same ActionOrchestrator, and checks the audit trail is complete.

Computing the metrics

benchmark_metrics counts each metric from section 12.3 per engine. HIGH_URGENCY defines “escalated” and DISRUPTIVE defines an intervention. The rules engine gets None for fallback success, since it never calls the service.

# smartplant-capstone/src/smartplant/benchmark_metrics.py
"""Industrial evaluation metrics for the three-way benchmark (rules vs LLM vs Kev)."""
from __future__ import annotations

import statistics
from collections import defaultdict
from dataclasses import asdict, dataclass
from typing import Any, Mapping, Sequence

from smartplant.contract import ACTIONS, URGENCIES
from smartplant.scenario_runner import ScenarioResult

HIGH_URGENCY = {"URGENT", "CRITICAL"}
DISRUPTIVE = {"SCHEDULE_MAINTENANCE", "TRANSFER_LOAD", "REQUEST_SHUTDOWN"}


@dataclass(frozen=True, slots=True)
class EngineMetrics:
    engine: str
    n: int
    agreement: float                    # action AND urgency within expert-accepted sets
    critical_miss_rate: float           # critical cases not escalated to URGENT/CRITICAL (lower is better)
    unnecessary_intervention_rate: float  # healthy cases that got a disruptive action (lower is better)
    response_validity: float            # labels inside the contract + action was eligible
    escalation_quality: float           # expected-specialist cases that reached a specialist
    fallback_success: float | None      # service-down cases handled safely (None = not applicable)
    audit_completeness: float
    p50_ms: float
    p95_ms: float
    cost_per_state_usd: float

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


def _rate(num: int, den: int, empty: float = 1.0) -> float:
    return round(num / den, 3) if den else empty


def _rate_or_none(num: int, den: int) -> float | None:
    return round(num / den, 3) if den else None


def compute(results: Sequence[ScenarioResult], scenarios: Sequence[Mapping[str, Any]],
            usd_per_state: Mapping[str, float] | None = None) -> list[EngineMetrics]:
    """usd_per_state: engine -> cost per decision (e.g. Kev GPU-hour / throughput, LLM tokens)."""
    spec = {s["id"]: s for s in scenarios}
    by_engine: dict[str, list[ScenarioResult]] = defaultdict(list)
    for r in results:
        by_engine[r.engine].append(r)
    out: list[EngineMetrics] = []
    for engine, rows in by_engine.items():
        agree = crit_n = crit_miss = healthy_n = unnecessary = valid = 0
        spec_n = spec_hit = down_n = down_ok = audit_ok = 0
        lat: list[float] = []
        for r in rows:
            exp, rec = spec[r.scenario_id]["expected"], r.recommendation
            agree += rec.action in exp["actions"] and rec.urgency in exp["urgencies"]
            if exp["critical"]:
                crit_n += 1
                crit_miss += rec.urgency not in HIGH_URGENCY
            if not exp["intervention"]:
                healthy_n += 1
                unnecessary += rec.action in DISRUPTIVE
            valid += rec.action in ACTIONS and rec.urgency in URGENCIES and 1 <= rec.severity <= 5
            if exp.get("route") == "specialist_review":
                spec_n += 1
                spec_hit += rec.route == "specialist_review"
            if spec[r.scenario_id].get("decision_service_down") and engine != "rules":
                down_n += 1
                down_ok += rec.source == "fallback" and rec.action != "MONITOR"
            audit_ok += r.audit_complete
            lat.append(r.wall_ms)
        q = statistics.quantiles(lat, n=20) if len(lat) >= 2 else lat * 19
        out.append(EngineMetrics(
            engine=engine, n=len(rows),
            agreement=_rate(agree, len(rows)),
            critical_miss_rate=_rate(crit_miss, crit_n, 0.0),
            unnecessary_intervention_rate=_rate(unnecessary, healthy_n, 0.0),
            response_validity=_rate(valid, len(rows)),
            escalation_quality=_rate(spec_hit, spec_n),
            fallback_success=_rate_or_none(down_ok, down_n),
            audit_completeness=_rate(audit_ok, len(rows)),
            p50_ms=round(statistics.median(lat), 1), p95_ms=round(q[18], 1),
            cost_per_state_usd=(usd_per_state or {}).get(engine, 0.0),
        ))
    return out


def kev_cost_per_state(gpu_usd_per_hour: float, decisions_per_second: float) -> float:
    """Self-hosted cost: you pay for the box, not the token. e.g. $2.50/h at 100 req/s."""
    if decisions_per_second <= 0:
        raise ValueError("throughput must be positive")
    return gpu_usd_per_hour / 3600 / decisions_per_second

What a Kev decision costs

With self-hosted Kev you pay for the box, not the token. Using the Kev README’s ~101 requests/s for Kev-4B on an H100 container and an illustrative $2.50/h, that’s about $0.0000069 per decision. A laptop running Kev-0.8B costs nothing extra, but measure its throughput before you promise a decision cadence.

Rendering the dashboard

eval_dashboard writes one dependency-free HTML file you can attach to a review. It has an engine metrics table and a scenario matrix; in the matrix, red means a critical case was under-escalated.

# smartplant-capstone/src/smartplant/eval_dashboard.py
"""Static, dependency-free HTML dashboard for the SmartPlant benchmark."""
from __future__ import annotations

import html
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping, Sequence

from smartplant.benchmark_metrics import EngineMetrics
from smartplant.scenario_runner import ScenarioResult

_CSS = """
:root{--bg:#fff;--fg:#1f2328;--muted:#656d76;--line:#d0d7de;--ok:#1a7f37;--bad:#cf222e;--warn:#9a6700;--bar:#0969da}
@media (prefers-color-scheme:dark){:root{--bg:#0d1117;--fg:#e6edf3;--muted:#8d96a0;--line:#30363d;--ok:#3fb950;--bad:#f85149;--warn:#d29922;--bar:#4493f8}}
body{background:var(--bg);color:var(--fg);font:14px/1.5 system-ui,sans-serif;margin:0;padding:24px 16px;max-width:1100px;margin-inline:auto}
h1{font-size:22px;margin:0 0 4px}p.sub{color:var(--muted);margin:0 0 24px}
table{border-collapse:collapse;width:100%;margin:12px 0 28px;font-variant-numeric:tabular-nums}
th,td{border-bottom:1px solid var(--line);padding:6px 8px;text-align:left;vertical-align:top}
th{color:var(--muted);font-weight:600}.ok{color:var(--ok)}.bad{color:var(--bad);font-weight:600}.warn{color:var(--warn)}
.wrap{overflow-x:auto}small{color:var(--muted)}
"""

_COLUMNS = [("agreement", "Agreement ↑", True), ("critical_miss_rate", "Critical miss ↓", False),
            ("unnecessary_intervention_rate", "Unnecessary intervention ↓", False),
            ("response_validity", "Valid responses ↑", True), ("escalation_quality", "Escalation quality ↑", True),
            ("fallback_success", "Fallback success ↑", True), ("audit_completeness", "Audit complete ↑", True)]


def _bar(value: float | None, good_high: bool) -> str:
    if value is None:
        return '<small>n/a</small>'
    w = int(max(0.0, min(1.0, value)) * 80)
    cls = "ok" if (value >= 0.9 if good_high else value <= 0.05) else "bad" if (value < 0.6 if good_high else value > 0.2) else "warn"
    return (f'<svg width="84" height="10" role="img" aria-label="{value:.0%}"><rect width="80" height="8" y="1" rx="2" '
            f'fill="currentColor" opacity=".12"/><rect width="{w}" height="8" y="1" rx="2" fill="var(--bar)"/></svg> '
            f'<span class="{cls}">{value:.0%}</span>')


def render(metrics: Sequence[EngineMetrics], results: Sequence[ScenarioResult],
           scenarios: Sequence[Mapping[str, Any]], title: str = "SmartPlant decision benchmark") -> str:
    engines = [m.engine for m in metrics]
    head = "".join(f"<th>{h}</th>" for _, h, _ in _COLUMNS)
    rows = "".join(
        f"<tr><td><b>{html.escape(m.engine)}</b><br><small>n={m.n}</small></td>"
        + "".join(f"<td>{_bar(getattr(m, k), up)}</td>" for k, _, up in _COLUMNS)
        + f"<td>{m.p50_ms:.0f} / {m.p95_ms:.0f} ms</td><td>${m.cost_per_state_usd:.6f}</td></tr>"
        for m in metrics)
    cell = {(r.scenario_id, r.engine): r for r in results}
    grid = ""
    for s in scenarios:
        exp = s["expected"]
        tds = ""
        for e in engines:
            r = cell.get((s["id"], e))
            if r is None:
                tds += "<td>–</td>"
                continue
            rec = r.recommendation
            hit = rec.action in exp["actions"] and rec.urgency in exp["urgencies"]
            miss = exp["critical"] and rec.urgency not in ("URGENT", "CRITICAL")
            cls = "bad" if miss else "ok" if hit else "warn"
            tds += (f'<td class="{cls}">{rec.action}<br><small>{rec.urgency} · sev {rec.severity} · '
                    f'{rec.route.replace("_", " ")} · {rec.source}</small></td>')
        grid += (f"<tr><td><b>{s['id']}</b> {html.escape(s['name'])}<br><small>expect "
                 f"{'/'.join(exp['actions'])} @ {'/'.join(exp['urgencies'])}</small></td>{tds}</tr>")
    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    return f"""<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"><title>{html.escape(title)}</title>
<style>{_CSS}</style></head><body>
<h1>{html.escape(title)}</h1><p class="sub">Generated {stamp}. Same states, same contract, same policy for every engine.
Measured results only; an offline stand-in measures the pipeline, not a model.</p>
<h2>Engine metrics</h2><div class="wrap"><table><tr><th>Engine</th>{head}<th>p50 / p95</th><th>Cost / state</th></tr>{rows}</table></div>
<h2>Scenario matrix</h2><p class="sub"><span class="ok">green</span> = matches expert label,
<span class="warn">amber</span> = disagrees with the expert label, <span class="bad">red</span> = critical case under-escalated.</p>
<div class="wrap"><table><tr><th>Scenario</th>{''.join(f'<th>{html.escape(e)}</th>' for e in engines)}</tr>{grid}</table></div>
</body></html>"""


def write(path: str | Path, metrics: Sequence[EngineMetrics], results: Sequence[ScenarioResult],
          scenarios: Sequence[Mapping[str, Any]]) -> Path:
    p = Path(path)
    p.write_text(render(metrics, results, scenarios), encoding="utf-8")
    return p

12.5 Mini-project 11: run the replay suite

This script wires the three modules together and prints one JSON line of metrics per engine. The cost flags take your own numbers: amortised box cost, measured Kev throughput, and LLM cost from your token usage.

# mini-projects/11-smartplant-eval-suite/run_eval.py
"""Mini-project 11: replay the 10 scenarios across engines and build the dashboard.

uv run python mini-projects/11-smartplant-eval-suite/run_eval.py                  # rules + offline stand-in
uv run python mini-projects/11-smartplant-eval-suite/run_eval.py --kev --llm      # real Kev + LLM baseline
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path

from smartplant.benchmark_metrics import compute, kev_cost_per_state
from smartplant.eval_dashboard import write
from smartplant.scenario_runner import default_engines, load_scenarios, run


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--kev", action="store_true")
    ap.add_argument("--llm", action="store_true")
    ap.add_argument("--gpu-usd-per-hour", type=float, default=2.50, help="Kev box cost (amortised)")
    ap.add_argument("--kev-rps", type=float, default=20.0, help="measured Kev throughput, decisions/s")
    ap.add_argument("--llm-usd-per-state", type=float, default=0.00025, help="measured from your token usage")
    ap.add_argument("--out", type=Path, default=Path("smartplant_eval.html"))
    args = ap.parse_args()

    scenarios = load_scenarios()
    results = run(default_engines(args.kev, args.llm), scenarios)
    kev_cost = kev_cost_per_state(args.gpu_usd_per_hour, args.kev_rps)
    costs = {"kev": kev_cost, "kev-offline-standin": 0.0, "llm-gpt-4o-mini": args.llm_usd_per_state}
    metrics = compute(results, scenarios, costs)
    for m in metrics:
        print(json.dumps(m.to_dict()))
    print(f"dashboard: {write(args.out, metrics, results, scenarios).resolve()}")
    return 0


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

The first command runs anywhere. The real three-way benchmark needs a Kev server, OPENAI_API_KEY and the adapter group.

# anywhere (rules + offline stand-in)
uv run python mini-projects/11-smartplant-eval-suite/run_eval.py
# the real three-way benchmark: Kev server up, OPENAI_API_KEY set, adapter group installed
uv sync --group adapter
uv run python mini-projects/11-smartplant-eval-suite/run_eval.py --kev --llm --kev-rps 20 --gpu-usd-per-hour 0

Offline run (rules vs the scripted stand-in):

{"engine": "rules", "n": 10, "agreement": 1.0, "critical_miss_rate": 0.0, "unnecessary_intervention_rate": 0.0, "response_validity": 1.0, "escalation_quality": 0.6, "fallback_success": null, "audit_completeness": 1.0, "p50_ms": 0.0, "p95_ms": 0.0, "cost_per_state_usd": 0.0}
{"engine": "kev-offline-standin", "n": 10, "agreement": 1.0, "critical_miss_rate": 0.0, "unnecessary_intervention_rate": 0.0, "response_validity": 1.0, "escalation_quality": 1.0, "fallback_success": 1.0, "audit_completeness": 1.0, "p50_ms": 0.2, "p95_ms": 0.2, "cost_per_state_usd": 0.0}
dashboard: /path/to/smartplant_eval.html

Reading the results without fooling yourself

• The rules score perfect agreement. That’s expected, and you should treat it as a warning. The scenarios were generated with the same zone and band thresholds the rules use, so the rules are being graded on their own homework. Their weak spot shows up in escalation quality (60 %). Rules can’t tell “conflicting signals” or “degraded data” apart from a clean case, so S09 (conflicting signals) goes to the standard workflow instead of a specialist. S10 also counts against them. With the decision service down, policy wants every non-MONITOR call reviewed, and static rules have no idea that “the model is unavailable” is even a state.
• The stand-in’s numbers prove the plumbing works, nothing more. Run with --kev and --llm and the same dashboard shows what your checkpoint actually does.
• Critical-miss rate is the gate. If a model raises agreement but misses even one critical case, it doesn’t ship. The safety floor in code exists so that this number is bounded by your thresholds, and not by whatever mood the model is in.
• Validity comes free with Kev, but not with LLMs. The adapter path can return labels that aren’t on the menu, and the service turns those into fallback:invalid_response. Count them rather than letting them disappear.
• Ten scenarios are a smoke test, not an evaluation. Before you trust a threshold, build a held-out set from real plant history (alarms, work orders, technician findings), with labels reviewed by reliability engineers. Chapter 13 shows how the audit trail produces exactly that dataset.

What the published Kev numbers suggest you’ll see

Kev’s own evaluation suites (new sources, test split) give a rough idea of what to expect. Accuracy runs 0.697 for 0.8B, 0.838 for 4B, 0.852 for 9B and 0.896 for 27B. Hosted Jev is reported at 0.857 on development data. If you allow a 5 % error budget, Kev automates 0.45–0.57 of decisions, against 0.70 for Jev; the rest get escalated. On a plant workload, expect the 4B and 9B models to beat the rules on the ambiguous cases (S05 and S09), and to escalate more often than Jev would. Fine-tuning on your own labels is how you win that automation back.

Hands-on: run the acceptance test yourself

This is the commissioning step: the same ten scenarios, run against each engine, with a pass/fail answer at the end. It needs no simulated time at all, because scenarios are fixed snapshots. The final step reuses the 10-minute snapshots.jsonl from Chapter 10’s hands-on to compare engines on a live-looking timeline. Outputs shown are from offline runs.

Step 1: Run the suite offline and open the dashboard.

uv run python mini-projects/11-smartplant-eval-suite/run_eval.py
open smartplant_eval.html                 # macOS; use xdg-open on Linux or start on Windows

You get one JSON line per engine (the same as in 12.4) and a dashboard file. In the dashboard, look at the Scenario matrix first: it shows which scenario each engine got right and which way it was wrong.

Step 2: Apply the ship gate in one command. The gate’s hard rules from 12.3 are zero critical misses, 100 % response validity, 100 % audit completeness and 100 % fallback success. jq can check them straight from the output.

uv run python mini-projects/11-smartplant-eval-suite/run_eval.py | grep '^{' | jq -r '
  [.engine,
   (if .critical_miss_rate == 0 and .response_validity == 1 and .audit_completeness == 1
       and (.fallback_success == null or .fallback_success == 1) then "PASS" else "REJECT" end),
   "agreement=\(.agreement)", "critical_miss=\(.critical_miss_rate)", "escalation=\(.escalation_quality)"] | @tsv'

rules	PASS	agreement=1.0	critical_miss=0.0	escalation=0.6
kev-offline-standin	PASS	agreement=1.0	critical_miss=0.0	escalation=1.0

The rules pass the hard gate but escalate only 60 % of the cases that need a specialist. That’s the gap a model has to close to earn its place.

Step 3: Put real Kev through the same test. With Kev running in terminal 1:

export KEV_BASE_URL=http://127.0.0.1:8009
uv run python mini-projects/11-smartplant-eval-suite/run_eval.py --kev --kev-rps 20 --gpu-usd-per-hour 2.50

A third line appears with "engine": "kev". Run the step 2 filter on it. If critical_miss_rate is above zero, the checkpoint doesn’t ship, however good its agreement looks. Kev’s latency and cost columns are now real too: at $2.50 an hour and 20 decisions a second, a decision costs about $2.50 ÷ 3600 ÷ 20 ≈ $0.000035.

Step 4 (optional): Add the LLM baseline. This needs an OPENAI_API_KEY and the adapter group.

uv sync --group adapter
uv run python mini-projects/11-smartplant-eval-suite/run_eval.py --kev --llm --kev-rps 20 --gpu-usd-per-hour 2.50

Watch response_validity for the LLM row. Kev can’t answer outside the contract, but the LLM can, and every invalid answer becomes a fallback.

Step 5: Compare engines on your 10-minute run. Scenarios are snapshots. Your simulator run is a timeline, which shows when each engine escalates. Use the snapshots.jsonl from Chapter 10’s hands-on:

paste <(uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --offline | cut -d'[' -f1) \
      <(uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl | cut -d'|' -f2)

Each row shows the minute, the zone, the stand-in’s call and Kev’s call side by side. The minute at which an engine first says URGENT is your time to escalation. A model that escalates a minute later than the stand-in on a fast-wearing bearing is a finding worth writing down, even if its scenario score is perfect.

Step 6: Record what you ran. Before you show anyone the numbers, note the Kev checkpoint (curl -s $KEV_BASE_URL/v1/models), the contract and policy versions, and the dashboard file. A result without versions can’t be compared with the next run.

Tests

The tests replay all ten scenarios, assert the safety-critical metrics, and check the cost formula.

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

from smartplant.benchmark_metrics import compute, kev_cost_per_state
from smartplant.eval_dashboard import render
from smartplant.scenario_runner import RuleEngine, ServiceEngine, load_scenarios, run, scripted_model
from triage.jev_client import FakeDecisionClient


def test_ten_scenarios_replay_and_score(tmp_path) -> None:
    scenarios = load_scenarios()
    assert len(scenarios) == 10
    results = run([RuleEngine(), ServiceEngine("standin", FakeDecisionClient(scripted_model))], scenarios, tmp_path)
    metrics = {m.engine: m for m in compute(results, scenarios, {"standin": kev_cost_per_state(2.5, 100)})}
    for m in metrics.values():
        assert m.response_validity == 1.0 and m.audit_completeness == 1.0
        assert m.critical_miss_rate == 0.0 and m.unnecessary_intervention_rate == 0.0
    assert metrics["standin"].fallback_success == 1.0
    html = render(list(metrics.values()), results, scenarios)
    assert "S10" in html and "Critical miss" in html


def test_kev_cost_model() -> None:
    assert round(kev_cost_per_state(3.6, 100), 8) == 0.00001

12.6 Checklist

• Every engine gets identical state, contract, policy and executor.
• Expert labels are sets, and each scenario records its critical flag and expected route.
• The ship gate is a critical-case miss rate of 0.
• Invalid LLM labels are counted, not silently coerced into valid ones.
• Kev cost is computed from measured throughput and the real hardware cost.
• A held-out set from real plant history exists before anyone tunes thresholds.

What’s next

Every piece of SmartPlant now exists and has been measured. Chapter 13 assembles it into the deployed capstone. It covers edge topology, OT network security, monitoring, governance and scaling to many assets, plus the feedback loop that fine-tunes Kev on your own audit trail.