Chapter 13 — Scenario Evaluation & Benchmark Suite

A dashboard says the service is 95% accurate. That sounds reassuring until the remaining 5% includes the cases that damage equipment. We need to distinguish harmless disagreement from dangerous under-escalation.

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

You will build: a replay suite of 10 SmartPlant scenarios, a three-way benchmark (rules vs an LLM vs Kev) scored with industrial metrics, and a static dashboard.

SmartPlant increment: scenario_runner, benchmark_metrics and eval_dashboard.

13.0 The problem, in plain English

We have a controller that looks sensible. But “looks sensible” is how plants end up with a motor that quietly burned out while the model said everything was fine. We need evidence, and the evidence has to be fair.

Fair means three things. Every candidate gets the same input. Every candidate is graded against the same answer key, written by people who know the plant. And the grade counts the mistakes that matter most separately, because missing a failing motor is not the same kind of mistake as inspecting a healthy one.

Everything below is built for that: a set of replayable scenarios (§13.2), metrics that separate dangerous errors from harmless ones (§13.3), and a runner and dashboard that make the comparison repeatable (§13.4).

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

💡 Mental Model: the test rig Think of a motor test bench. The motor changes; the bench, the load and the instruments don’t. Our harness is the bench, and the engines are the motors.

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

13.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:

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

Snippet 1: imports. The runner reuses Part I’s client and breaker, and Part II’s orchestrator and audit log.

# 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

Snippet 2: load the scenarios. They’re packaged with smartplant, so no path tricks.

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

Snippet 3: rebuild one scenario from scratch. Seeded simulator and fault list, then stream the readings into the edge builder.

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

Snippet 4: enrich and return. Same enricher as production, fed with the scenario’s context.

    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)

Snippet 5: the Engine protocol and the rules engine. Anything with a name and a recommend() can be benchmarked.

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

Snippet 6: the S10 outage. A client that always times out.

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)

Snippet 7: any DecisionClient behind the same policy. Kev, Jev, an adapter LLM and the fake all use this one class.

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

⚠️ Watch Out ServiceEngine builds a fresh SmartPlantDecisionService per call, so one scenario’s failures can’t open a circuit breaker that affects the next.

Snippet 8: the offline stand-in, first half. It derives a severity from the zone and band, then an urgency and an action.

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

Snippet 9: the offline stand-in, second half. It packages those into the same typed answers that Kev returns.

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

Snippet 10: one result per scenario and engine.

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

Snippet 11: the runner. Time the decision, execute it through the same orchestrator, and check the audit trail is complete.

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

Snippet 12: choose engines. The rules are always in. The LLM is optional. If Kev isn’t available, an offline stand-in takes its place.

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

Snippet 13: the command line.

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

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

Computing the metrics

benchmark_metrics counts each metric from section 13.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.

Snippet 1: the vocabulary.

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

Snippet 2: one metrics record per engine. The comments say what each field measures and which direction is good.

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

Snippet 3: two tiny rate helpers. _rate_or_none returns None when the metric doesn’t apply.

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

Snippet 4: set up the counters.

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] = []

Snippet 5: count, scenario by scenario. Each line is one metric from the table in §13.3.

        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)

⚠️ Watch Out engine != "rules" on the fallback line is deliberate. The rules engine never calls the service, so “fallback success” has no meaning for it.

Snippet 6: turn counts into rates. p95 comes from 20-quantiles.

        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

Snippet 7: self-hosted cost.

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

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

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.

Snippet 1: imports.

# 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

Snippet 2: the styles. Light and dark themes with a handful of CSS variables.

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

Snippet 3: which metrics become columns. Each tuple is a field, a heading, and whether higher is better.

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

Snippet 4: a tiny bar chart. A small inline SVG plus a colour class from the thresholds.

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

Snippet 5: build the engine table rows.

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)

Snippet 6: build the scenario matrix. Green matches the expert label, amber disagrees, and red means a critical case was under-escalated.

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

Snippet 7: assemble the page.

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

Snippet 8: write it to disk.

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

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

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

Snippet 1: usage and imports.

# 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

Snippet 2: the flags.

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

Snippet 3: replay, score, print.

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

📄 Full source: mini-projects/11-smartplant-eval-suite/run_eval.py

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

Snippet 1: imports.

# 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

Snippet 2: replay and assert the safety metrics. Every engine must score 100 % validity and audit completeness, with zero critical misses.

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

Snippet 3: the cost formula.

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

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

13.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 14 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.