Chapter 10 — Industrial Decision Modeling with Kev Primitives
Chapter 9 gave us trustworthy, versioned edge snapshots and a
KevDecisionClient.
In this chapter. We build the SmartPlant Operational Response Engine: a decision service that weighs motor degradation, backup capacity and production priority, and returns a typed recommendation with an explanation. The new code is
smartplant.contract,context_enricher,models,decision_serviceand the packagedtest_scenarios.json.
10.1 From an alarm to four clear questions
A threshold alarm on the SCADA screen tells the night-shift operator that something is wrong. It doesn’t say what to do, how soon, through which workflow, or how bad it is. We turn each of those into one small question and send all four to Kev in a single fan-out call (Chapter 3).
Kev answers with the same three primitives the hosted Jev model offers through its System One API. A Choice picks one label from a list and reports a confidence. A Noul gives the probability that a yes/no condition holds. A Score places the situation on an ordered scale of levels you describe.
| Decision | Primitive | Options or scale | Why this primitive |
|---|---|---|---|
| Intervention urgency | Choice | ROUTINE · PRIORITY · URGENT · CRITICAL | Maps one-to-one to CMMS SLA classes, so we need a discrete label to route on. |
| Preferred operational action | Choice | MONITOR · INSPECT · SCHEDULE_MAINTENANCE · TRANSFER_LOAD · REQUEST_SHUTDOWN | Exactly one response wins. Code decides which options are eligible. |
| Workflow routing | Noul | Probability the standard CMMS workflow fits | One checkable condition. A low probability sends the case to specialist review. |
| Operational severity | Score | 5 levels, mapped to plant scale 1–5 | An ordered degree with describable situations, used for ranking and floors. |
| Signals conflict (speculative) | Noul | Probability the signals disagree | A cheap extra question that catches “temperature up, vibration normal”. |
The last row is a speculative question, an optional extra that costs almost nothing in the same call.
Two choices here tend to come up in design reviews. Urgency is a Choice even though its labels are ordered, because the CMMS needs one SLA class and each class has a crisp, situational definition. Severity, the degree of the problem, is the Score, and asking both gives you a consistency check for free. Also, severity is not an engineering risk calculation. It is the model’s ordinal judgment, useful for ranking and gating. Formal limits such as trip points and ISO zones stay in code and config.
10.2 The decision contract
The decision contract is the versioned set of questions we send to the model. Each question has instructions and criteria: plain descriptions of what each option or level means on the plant. In the module below, look at how eligible_actions() prunes options before the call, and how to_severity_1_5() maps the raw score onto the plant’s 1–5 scale.
# smartplant-capstone/src/smartplant/contract.py
"""SmartPlant decision contract: four bounded questions (+1 speculative) about one asset state."""
from __future__ import annotations
import copy
import hashlib
import json
from typing import Any, Final, Literal, Mapping
CONTRACT_VERSION: Final[str] = "smartplant-contract/1.0.0"
Urgency = Literal["ROUTINE", "PRIORITY", "URGENT", "CRITICAL"]
Action = Literal["MONITOR", "INSPECT", "SCHEDULE_MAINTENANCE", "TRANSFER_LOAD", "REQUEST_SHUTDOWN"]
URGENCIES: Final[tuple[Urgency, ...]] = ("ROUTINE", "PRIORITY", "URGENT", "CRITICAL")
ACTIONS: Final[tuple[Action, ...]] = (
"MONITOR", "INSPECT", "SCHEDULE_MAINTENANCE", "TRANSFER_LOAD", "REQUEST_SHUTDOWN",
)
SMARTPLANT_QUESTIONS: Final[dict[str, dict[str, Any]]] = {
"intervention_urgency": {
"type": "choice",
"instructions": "How soon does `asset` need a maintenance intervention, given `telemetry`, "
"`data_quality` and `maintenance`?",
"criteria": {
"ROUTINE": "Readings are in normal bands with stable trends; next planned service is fine",
"PRIORITY": "One signal is outside its normal band or trending up, but no protective "
"limit is near; act within days",
"URGENT": "Several signals are degraded or one is rising fast toward a protective limit; "
"act this shift",
"CRITICAL": "A protective limit is reached or imminent, or damage is likely within the hour",
},
},
"operational_action": {
"type": "choice",
"instructions": "Which response should operations take next for `asset`, considering "
"`telemetry`, `production` and `redundancy`?",
"criteria": {
"MONITOR": "Asset is healthy or the change is minor; keep normal monitoring",
"INSPECT": "An operator should physically inspect the asset to confirm the condition",
"SCHEDULE_MAINTENANCE": "Condition is confirmed or clear enough to plan a repair at "
"the next suitable opportunity",
"TRANSFER_LOAD": "Asset should be unloaded by moving production to the backup in `redundancy`",
"REQUEST_SHUTDOWN": "Asset should be stopped in a controlled way because running on "
"risks damage or safety",
},
},
"standard_workflow": {
"type": "noul",
"instructions": "Can this case be handled through the standard maintenance work-order workflow, "
"without escalation to a reliability specialist?",
"criteria": {
"true": "Signals agree, data is trustworthy, and the pattern is a familiar wear or "
"service issue",
"false": "Signals conflict, data quality is questionable, or the pattern is unusual",
},
},
"operational_severity": {
"type": "score",
"instructions": "How severe is the current condition of `asset`?",
"criteria": [
{"what": "No abnormal condition", "examples": ["all bands normal, trends stable"]},
{"what": "Minor deviation with no functional impact",
"examples": ["vibration in acceptable zone, slowly rising"]},
{"what": "Degraded condition that will worsen without action",
"examples": ["vibration in alert zone and rising, temperature elevated"]},
{"what": "Serious condition; continued running risks equipment damage",
"examples": ["vibration in danger zone or temperature high and rising fast"]},
{"what": "Protective limit reached or failure imminent",
"examples": ["temperature at trip band", "vibration above danger zone and climbing fast"]},
],
},
}
SPECULATIVE_QUESTIONS: Final[dict[str, dict[str, Any]]] = {
"signals_conflict": {
"type": "noul",
"instructions": "Do the signals in `telemetry` point in different directions, for example "
"temperature rising while vibration stays normal?",
},
}
class ContractError(ValueError):
pass
def validate(questions: Mapping[str, Mapping[str, Any]]) -> None:
for qid, q in questions.items():
if not str(q.get("instructions", "")).strip():
raise ContractError(f"{qid}: instructions required")
crit = q.get("criteria")
if q["type"] == "choice" and not 2 <= len(crit or {}) <= 255:
raise ContractError(f"{qid}: Choice needs 2..255 options")
if q["type"] == "score" and not 2 <= len(crit or []) <= 10: # portable across Jev (2-10) and Kev
raise ContractError(f"{qid}: Score needs 2..10 levels")
if set(questions["operational_action"]["criteria"]) - set(ACTIONS):
raise ContractError("operational_action has labels outside ACTIONS")
def eligible_actions(state: Mapping[str, Any]) -> list[Action]:
"""Only offer actions that are valid NOW. Rules live in code, never in criteria text."""
allowed = set(ACTIONS)
if state["redundancy"]["backup_coverage"] == "none":
allowed.discard("TRANSFER_LOAD")
return [a for a in ACTIONS if a in allowed]
def build_questions(state: Mapping[str, Any], *, speculative: bool = True) -> dict[str, dict[str, Any]]:
qs = copy.deepcopy(SMARTPLANT_QUESTIONS)
keep = set(eligible_actions(state))
qs["operational_action"]["criteria"] = {
k: v for k, v in qs["operational_action"]["criteria"].items() if k in keep
}
if speculative:
qs.update(copy.deepcopy(SPECULATIVE_QUESTIONS))
validate(qs)
return qs
def contract_hash(questions: Mapping[str, Any]) -> str:
return hashlib.sha256(json.dumps(questions, sort_keys=True).encode()).hexdigest()[:12]
def to_severity_1_5(raw: float) -> int:
"""Kev/Jev return 0..4 (probability-weighted mean). Plant scale is integer 1..5."""
if not 0.0 <= raw <= 4.0:
raise ContractError(f"severity out of range: {raw}")
return int(min(5, max(1, round(raw) + 1)))
validate(SMARTPLANT_QUESTIONS)
Notice what the criteria don’t say: “don’t pick TRANSFER_LOAD if there’s no backup”. That’s an eligibility rule, so eligible_actions() removes the option before the call. The model can’t recommend what it was never offered.
10.3 Context enrichment: MES, CMMS and redundancy
Telemetry alone can’t answer “should we transfer load?”. Context enrichment pulls those facts from the MES, the CMMS and the redundancy records and merges them with the edge snapshot. Code does every comparison: whether the backup covers the required load, how close the deadline is, how old the bearing is.
The three read services are small Protocol interfaces, and MockPlantData implements all three for the scenarios. Watch how the band helpers turn raw numbers into labels, so Kev never compares 78 % with 91 % itself.
# smartplant-capstone/src/smartplant/context_enricher.py
"""Operational context: MES (production), CMMS (maintenance), redundancy -> decision state.
The enricher computes every comparison in code (does the backup cover the load? how close is
the deadline?) and hands Kev named buckets. Kev never compares 78 % with 91 %.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Final, Literal, Mapping, Protocol
from smartplant.edge_state_builder import EdgeSnapshot
DECISION_STATE_VERSION: Final[str] = "smartplant-state/1.0.0"
Priority = Literal["LOW", "NORMAL", "HIGH", "CRITICAL"]
@dataclass(frozen=True, slots=True)
class ProductionOrder:
order_id: str
priority: Priority
hours_to_deadline: float
required_load_pct: float # load the line needs from this asset (0..1)
@dataclass(frozen=True, slots=True)
class AssetRecord:
asset_id: str
criticality: Literal["A", "B", "C"] # A = line stops without it
days_since_bearing_replacement: int
open_work_orders: int
spare_bearing_in_stock: bool
last_inspection_finding: str = "none"
@dataclass(frozen=True, slots=True)
class BackupAsset:
asset_id: str | None
available: bool
approved_for_process: bool
capacity_pct: float # fraction of the primary's load it can carry (0..1)
class MesService(Protocol):
def current_order(self, asset_id: str) -> ProductionOrder | None: ...
class CmmsReader(Protocol):
def asset_record(self, asset_id: str) -> AssetRecord: ...
class RedundancyService(Protocol):
def backup_for(self, asset_id: str) -> BackupAsset: ...
@dataclass(slots=True)
class MockPlantData:
"""One object implementing all three read services, seeded per scenario."""
orders: dict[str, ProductionOrder] = field(default_factory=dict)
assets: dict[str, AssetRecord] = field(default_factory=dict)
backups: dict[str, BackupAsset] = field(default_factory=dict)
def current_order(self, asset_id: str) -> ProductionOrder | None:
return self.orders.get(asset_id)
def asset_record(self, asset_id: str) -> AssetRecord:
try:
return self.assets[asset_id]
except KeyError as e:
raise LookupError(f"asset {asset_id} not registered in CMMS") from e
def backup_for(self, asset_id: str) -> BackupAsset:
return self.backups.get(asset_id, BackupAsset(None, False, False, 0.0))
def backup_coverage(backup: BackupAsset, required_load_pct: float) -> str:
if not (backup.available and backup.approved_for_process and backup.capacity_pct > 0):
return "none"
return "full" if backup.capacity_pct >= required_load_pct else "partial"
def _bearing_age_band(days: int) -> str:
return "new" if days < 90 else "mid_life" if days < 365 else "aged"
def _deadline_band(hours: float | None) -> str:
if hours is None:
return "no_active_order"
return "under_8h" if hours < 8 else "8_to_48h" if hours <= 48 else "over_48h"
def enrich(snapshot: EdgeSnapshot, mes: MesService, cmms: CmmsReader,
redundancy: RedundancyService) -> dict[str, Any]:
order = mes.current_order(snapshot.asset_id)
record = cmms.asset_record(snapshot.asset_id)
backup = redundancy.backup_for(snapshot.asset_id)
required = order.required_load_pct if order else (snapshot.features.get("load_pct") or 0.0)
return {
"state_version": DECISION_STATE_VERSION,
"snapshot_hash": snapshot.snapshot_hash,
"asset": {"id": snapshot.asset_id, "criticality": record.criticality},
"telemetry": {
**snapshot.bands,
"latest_vibration_mm_s": snapshot.latest.get("vibration_rms_mm_s"),
"latest_bearing_temp_c": snapshot.latest.get("bearing_temp_c"),
"window_minutes": 20,
},
"data_quality": {"status": snapshot.data_quality, "issues": list(snapshot.quality_issues)},
"protective_limit_reached": snapshot.protective_limit_reached,
"maintenance": {
"bearing_age": _bearing_age_band(record.days_since_bearing_replacement),
"open_work_orders": record.open_work_orders > 0,
"spare_bearing_in_stock": record.spare_bearing_in_stock,
"last_inspection_finding": record.last_inspection_finding,
},
"production": {
"order_priority": order.priority if order else "NONE",
"deadline": _deadline_band(order.hours_to_deadline if order else None),
},
"redundancy": {
"backup_asset": backup.asset_id,
"backup_coverage": backup_coverage(backup, required),
},
}
def summarize(state: Mapping[str, Any]) -> str:
"""One line for operator screens and logs."""
t = state["telemetry"]
return (f"{state['asset']['id']}: vib {t['vibration_zone']}/{t['vibration_trend']}, "
f"temp {t['temperature_band']}/{t['temperature_trend']}, load {t['load_band']}, "
f"data {state['data_quality']['status']}, order {state['production']['order_priority']}, "
f"backup {state['redundancy']['backup_coverage']}")
What Kev actually sees (scenario S06)
Here is the real decision state the code above produces for “severe degradation, backup available”:
{
"state_version": "smartplant-state/1.0.0",
"snapshot_hash": "7f5ff2e4b643c232",
"asset": {
"id": "M-104",
"criticality": "A"
},
"telemetry": {
"vibration_zone": "D_danger",
"vibration_trend": "rising_fast",
"temperature_band": "elevated",
"temperature_trend": "rising_fast",
"load_band": "heavy",
"latest_vibration_mm_s": 7.95,
"latest_bearing_temp_c": 75.27,
"window_minutes": 20
},
"data_quality": {
"status": "valid",
"issues": []
},
"protective_limit_reached": false,
"maintenance": {
"bearing_age": "mid_life",
"open_work_orders": false,
"spare_bearing_in_stock": true,
"last_inspection_finding": "none"
},
"production": {
"order_priority": "HIGH",
"deadline": "8_to_48h"
},
"redundancy": {
"backup_asset": "M-105",
"backup_coverage": "full"
}
}
That’s a few hundred tokens of bands and booleans, plus two rounded readings for operator context. It fits comfortably inside Kev’s validated 8,192-token context and is cheap to cache: the Kev README reports 136 ms for Kev-4B on an Apple M5 when the text is cached, against 721 ms for new text.
10.4 The decision service
Think of the service as interlocks around one model call. Any guard can stop the sequence with a safe answer, and a policy step checks the model’s answer before anyone acts on it.
Every engine (rules, LLM or Kev) returns the same result type, so the operator screen doesn’t care where an answer came from. The source, reasons and confidence fields carry the audit trail.
# smartplant-capstone/src/smartplant/models.py
"""Shared result type: what every engine (rules, LLM, Kev) must return."""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from typing import Any, Literal
from smartplant.contract import Action, Urgency
Route = Literal["standard_cmms", "specialist_review"]
Source = Literal["kev", "jev", "adapter", "fake", "fallback", "safety_rules", "rules"]
@dataclass(frozen=True, slots=True)
class Recommendation:
asset_id: str
urgency: Urgency
action: Action
severity: int # 1..5
route: Route
requires_approval: bool
source: str
reasons: tuple[str, ...]
explanation: str
snapshot_hash: str
contract_version: str
model: str
latency_ms: float = 0.0
confidence: dict[str, float] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["reasons"] = list(self.reasons)
return d
The service reuses two Part I modules unchanged, triage.jev_client and triage.resilience. Read evaluate() top to bottom as the guard sequence. Protective limits are the PLC’s job, so the service mirrors them with rules and never waits on a model. Invalid data falls back to rules. An open circuit breaker, which stops calls to a service known to be down, also falls back. Only then does the model get asked.
# smartplant-capstone/src/smartplant/decision_service.py
"""SmartPlant decision service: state -> Kev (via the shared triage client) -> policy -> Recommendation.
Reuses Part I unchanged: `triage.jev_client` (DecisionClient, KevDecisionClient, FakeDecisionClient)
and `triage.resilience.CircuitBreaker`. Only the contract and the policy are domain-specific.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Any, Mapping
from triage.jev_client import DecisionClient, DecisionError, DecisionResponse
from triage.resilience import CircuitBreaker
from smartplant.contract import CONTRACT_VERSION, build_questions, to_severity_1_5
from smartplant.fallback_rules import (ACTION_RANK, URGENCY_RANK, explain, fallback_recommendation,
rule_baseline, safety_floor)
from smartplant.models import Recommendation
log = logging.getLogger("smartplant.decision_service")
POLICY_VERSION = "smartplant-policy/1.0.0"
@dataclass(frozen=True, slots=True)
class KevThresholds:
"""Starting points for Kev-4B. Kev's confidence is not Jev's: recalibrate per model and per plant."""
urgency_min_conf: float = 0.25
action_min_conf: Mapping[str, float] = field(default_factory=lambda: MappingProxyType({
"MONITOR": 0.30, "INSPECT": 0.20, "SCHEDULE_MAINTENANCE": 0.30,
"TRANSFER_LOAD": 0.45, "REQUEST_SHUTDOWN": 0.0,
}))
standard_workflow_min: float = 0.60
conflict_escalate: float = 0.50
@dataclass(slots=True)
class SmartPlantDecisionService:
client: DecisionClient
breaker: CircuitBreaker = field(default_factory=lambda: CircuitBreaker(failure_threshold=3, cooldown_s=30))
thresholds: KevThresholds = field(default_factory=KevThresholds)
def evaluate(self, state: Mapping[str, Any]) -> Recommendation:
# 1. Safety partition: protective limits are the PLC's job. Mirror them; never wait on a model.
if state["protective_limit_reached"]:
return rule_baseline(state, source="safety_rules", reason="protective_limit: model not consulted")
# 2. Untrusted evidence never reaches the model.
if state["data_quality"]["status"] == "invalid":
return fallback_recommendation(state, "invalid_data")
# 3. Fail fast when the decision service is known to be down.
if not self.breaker.allow():
return fallback_recommendation(state, "breaker_open")
questions = build_questions(state)
try:
resp = self.client.decide(state, questions)
except DecisionError as e:
if e.retryable:
self.breaker.record_failure()
else:
log.error("non-retryable decision error (contract/auth bug?): %s", e)
return fallback_recommendation(state, "decision_service_unavailable")
self.breaker.record_success()
try:
return self._apply_policy(state, questions, resp)
except (KeyError, TypeError, ValueError) as e: # invalid/missing answer from any backend
log.warning("invalid decision response: %s", e)
return fallback_recommendation(state, "invalid_response")
def _apply_policy(self, state: Mapping[str, Any], questions: Mapping[str, Any],
resp: DecisionResponse) -> Recommendation:
th = self.thresholds
u, a = resp.choice("intervention_urgency"), resp.choice("operational_action")
std = resp.noul("standard_workflow").p_yes
severity = to_severity_1_5(resp.score("operational_severity").score)
conflict = resp.noul("signals_conflict").p_yes if "signals_conflict" in resp.answers else 0.0
if a.choice not in questions["operational_action"]["criteria"]:
raise ValueError(f"action {a.choice!r} not offered") # adapters/LLMs can do this
if u.choice not in URGENCY_RANK:
raise ValueError(f"urgency {u.choice!r} not in contract")
urgency, action = u.choice, a.choice
reasons = [f"{resp.backend}: urgency={u.choice}@{u.confidence:.2f}",
f"action={a.choice}@{a.confidence:.2f}", f"severity={severity}",
f"standard_workflow p={std:.2f}"]
# Code-computed safety floor: a model may escalate past it, never below it.
floor = safety_floor(state)
if floor and URGENCY_RANK[urgency] < URGENCY_RANK[floor[0]]:
urgency = floor[0]
reasons.append(f"floor_raised_urgency->{urgency}")
if floor and ACTION_RANK[action] < ACTION_RANK[floor[1]]:
action = floor[1]
reasons.append(f"floor_raised_action->{action}")
if severity >= 4 and action == "MONITOR":
action = "INSPECT"
reasons.append("severity>=4 never just monitors")
route = "standard_cmms" if std >= th.standard_workflow_min else "specialist_review"
if conflict >= th.conflict_escalate:
route = "specialist_review"
reasons.append(f"signals_conflict p={conflict:.2f}")
if state["data_quality"]["status"] == "degraded":
route = "specialist_review"
reasons.append("degraded_data")
if u.confidence < th.urgency_min_conf or a.confidence < th.action_min_conf.get(action, 0.5):
route = "specialist_review"
reasons.append("low_confidence")
if urgency == "CRITICAL":
route = "specialist_review"
return Recommendation(
asset_id=state["asset"]["id"], urgency=urgency, action=action, severity=severity, # type: ignore[arg-type]
route=route, # type: ignore[arg-type]
requires_approval=action in ("TRANSFER_LOAD", "REQUEST_SHUTDOWN"),
source=resp.backend, reasons=tuple(reasons),
explanation=explain(state, action, urgency, reasons),
snapshot_hash=state["snapshot_hash"], contract_version=CONTRACT_VERSION,
model=resp.model, latency_ms=round(resp.latency_ms, 1),
confidence={"urgency": round(u.confidence, 3), "action": round(a.confidence, 3),
"standard_workflow": round(std, 3), "signals_conflict": round(conflict, 3)},
)
In _apply_policy(), a case goes to specialist review instead of the standard CMMS workflow when the workflow probability is low, the signals conflict, the data is degraded, confidence (the model’s own measure of how sure it is) is too low, or urgency is CRITICAL. TRANSFER_LOAD and REQUEST_SHUTDOWN always require approval.
The safety floor: production priority can’t win
The safety floor makes the rule “a high-priority order must never override an unsafe condition” executable. It is the minimum urgency and action implied by code facts (zones and bands), not by the model. A model may escalate past it, never below it.
# smartplant-capstone/src/smartplant/fallback_rules.py (excerpt)
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
Thresholds for Kev, not for Jev
The gates in KevThresholds are deliberately lower than Part I’s Jev gates, because Kev’s confidence statistic runs lower on the same evidence. Treat them as starting points and calibrate them, meaning tune them against known outcomes on your own setup:
- Replay the labeled scenarios (Chapter 12) against your Kev checkpoint.
- Sweep each gate with
capstone-project/tools/sweep_threshold.py(Chapter 4), and pick the lowest threshold that meets the error budget for that action. - Call
/v1/systemone/permuteonoperational_actionfor a sample of states. If the winner changes when the options are reordered, tighten that action’s gate or rewrite its criteria. - Re-run all three steps whenever the checkpoint, contract or policy version changes.
10.5 Mini-project 9: SmartPlant Operational Response Engine
The ten scenarios ship inside the package (smartplant/data/test_scenarios.json), so this mini-project and the Chapter 12 suite load the same file with load_scenarios() and no relative paths. Here is one entry:
{
"id": "S06", "name": "Severe degradation, backup available", "load": 0.9,
"faults": [{"mode": "gradual_degradation", "start_step": 20, "magnitude": 1.0}],
"context": {
"order": {"priority": "HIGH", "hours_to_deadline": 30, "required_load_pct": 0.85},
"asset": {"criticality": "A", "days_since_bearing_replacement": 200, "open_work_orders": 0,
"spare_bearing_in_stock": true, "last_inspection_finding": "none"},
"backup": {"asset_id": "M-105", "available": true, "approved_for_process": true, "capacity_pct": 0.95}
},
"expected": {"actions": ["TRANSFER_LOAD", "INSPECT", "SCHEDULE_MAINTENANCE"],
"urgencies": ["URGENT", "CRITICAL"], "route": null, "critical": true, "intervention": true},
"steps": 120, "seed": 11, "decision_service_down": false
}
The expected field holds sets of acceptable answers, not one label. Expert reviewers often agree on “inspect or schedule, urgently” without agreeing on one exact action.
In the runner, --offline swaps in a scripted stand-in for the model, and S10 uses an always-down client to simulate a timeout.
# mini-projects/09-smartplant-decision-service/run_decision_service.py
"""Mini-project 9: evaluate SmartPlant scenarios with Kev (or the offline stand-in).
DECISION_BACKEND=kev
KEV_BASE_URL=http://127.0.0.1:8009
uv run python mini-projects/09-smartplant-decision-service/run_decision_service.py S06
uv run python mini-projects/09-smartplant-decision-service/run_decision_service.py --offline
"""
from __future__ import annotations
import argparse
import json
from triage.jev_client import FakeDecisionClient, client_from_env
from smartplant.context_enricher import summarize
from smartplant.decision_service import SmartPlantDecisionService
from smartplant.scenario_runner import DownClient, build_state, load_scenarios, scripted_model
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("ids", nargs="*", help="scenario ids (default: all)")
ap.add_argument("--offline", action="store_true", help="use the scripted stand-in instead of a server")
ap.add_argument("--json", action="store_true")
args = ap.parse_args()
client = FakeDecisionClient(scripted_model) if args.offline else client_from_env()
service = SmartPlantDecisionService(client)
for spec in load_scenarios():
if args.ids and spec["id"] not in args.ids:
continue
state = build_state(spec)
# S10 simulates the decision service timing out; everything else uses the shared service.
svc = SmartPlantDecisionService(DownClient()) if spec.get("decision_service_down") else service
rec = svc.evaluate(state)
if args.json:
print(json.dumps({"scenario": spec["id"], **rec.to_dict()}))
continue
print(f"\n{spec['id']} {spec['name']}\n state : {summarize(state)}")
print(f" result: {rec.urgency} / {rec.action} / severity {rec.severity} / {rec.route}"
f"{' / APPROVAL REQUIRED' if rec.requires_approval else ''} [{rec.source}, {rec.latency_ms} ms]")
print(f" why : {rec.explanation}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Run it against a Kev server or offline:
# against a running Kev server
DECISION_BACKEND=kev
uv run python mini-projects/09-smartplant-decision-service/run_decision_service.py S06 S07
# anywhere, with the offline stand-in (measures the pipeline, not Kev)
uv run python mini-projects/09-smartplant-decision-service/run_decision_service.py --offline S02 S04 S06 S07 S10
The Kev retuns
S06 Severe degradation, backup available
state : M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
result: URGENT / TRANSFER_LOAD / severity 4 / specialist_review / APPROVAL REQUIRED [kev, 4045.2 ms]
why : TRANSFER_LOAD at URGENT urgency because vibration D_danger (rising_fast); bearing temperature elevated (rising_fast); load heavy; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage full.
S07 Severe degradation, backup unavailable
state : M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup none
result: URGENT / REQUEST_SHUTDOWN / severity 4 / specialist_review / APPROVAL REQUIRED [kev, 4078.5 ms]
why : REQUEST_SHUTDOWN at URGENT urgency because vibration D_danger (rising_fast); bearing temperature elevated (rising_fast); load heavy; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage none.
The offline run shows 0.0 ms latencies because it measures the pipeline, not Kev (it shows 4045.2 ms):
S02 Gradual bearing degradation
state : M-104: vib C_alert/rising_fast, temp normal/rising_fast, load normal, data valid, order NORMAL, backup full
result: URGENT / INSPECT / severity 4 / standard_cmms [fake, 0.0 ms]
why : INSPECT at URGENT urgency because vibration C_alert (rising_fast); bearing temperature normal (rising_fast); load normal; backup coverage full.
S04 Faulty temperature sensor (dropout)
state : M-104: vib A_good/stable, temp normal/stable, load normal, data invalid, order NORMAL, backup full
result: PRIORITY / INSPECT / severity 3 / specialist_review [fallback, 0.0 ms]
why : INSPECT at PRIORITY urgency because vibration A_good (stable); bearing temperature normal (stable); load normal; backup coverage full; data quality invalid.
S06 Severe degradation, backup available
state : M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup full
result: CRITICAL / TRANSFER_LOAD / severity 5 / specialist_review / APPROVAL REQUIRED [fake, 0.0 ms]
why : TRANSFER_LOAD at CRITICAL urgency because vibration D_danger (rising_fast); bearing temperature elevated (rising_fast); load heavy; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage full.
S07 Severe degradation, backup unavailable
state : M-104: vib D_danger/rising_fast, temp elevated/rising_fast, load heavy, data valid, order HIGH, backup none
result: CRITICAL / INSPECT / severity 5 / specialist_review [fake, 0.0 ms]
why : INSPECT at CRITICAL urgency because vibration D_danger (rising_fast); bearing temperature elevated (rising_fast); load heavy; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage none.
S10 Decision service timeout during degradation
state : M-104: vib C_alert/rising_fast, temp normal/rising_fast, load normal, data valid, order NORMAL, backup full
result: PRIORITY / INSPECT / severity 3 / specialist_review [fallback, 0.0 ms]
why : INSPECT at PRIORITY urgency because vibration C_alert (rising_fast); bearing temperature normal (rising_fast); load normal; backup coverage full.
Read the output as a policy reviewer would:
- In S04, the invalid data never reached the model (
fallback), and the case went to a specialist. - S06 and S07 differ only in backup availability. With a backup,
TRANSFER_LOADis offered and requires approval. Without one, it was never on the menu. - In S10, the decision service timed out, so conservative rules took over and sent the case to specialist review.
- Every explanation is assembled from recorded facts. Nothing is invented.
Hands-on: from a 10-minute telemetry run to a SmartPlant decision
This walk-through picks up where Chapter 9’s simulator left off. You’ll take ten minutes of simulated bearing wear, enrich each snapshot with plant context, and watch the decision service answer in your console. Every run stays within 10 simulated minutes (61 steps of 10 s), so you can read the whole timeline at a glance.
You need two terminals. Terminal 1 runs Kev, as in Chapter 9. Terminal 2 is the book workspace at the repo root. Each step that needs Kev also has an --offline variant that uses the scripted stand-in, so you can follow along before Kev is installed. The outputs shown are from offline runs; with Kev, the [fake] tag becomes [kev] and some calls may differ. That difference is what you’re here to see.
Step 1: Start Kev (terminal 1) and point the workspace at it (terminal 2).
# terminal 1: Kev in its own folder and venv (Chapter 9)
cd ~/src/kev && uv run --extra serve python -m kev.serve --run jaredpalmer/kev-0.8b --port 8009
# terminal 2: the book workspace
uv sync
export KEV_BASE_URL=http://127.0.0.1:8009
curl -s $KEV_BASE_URL/v1/models | jq '.' # confirm which checkpoint is loaded
Step 2: Simulate 10 minutes of bearing wear. Magnitude 2 walks the motor through all four vibration zones in ten minutes. A snapshot is written every minute.
uv run python mini-projects/08-iot-sensor-simulator/run_simulator.py \
--fault gradual_degradation --magnitude 2 --fault-start 6 \
--steps 61 --snapshot-every 6 --out snapshots.jsonl
t= 6 vib= 2.0 temp= 61.32 A_good rising normal valid -
t= 12 vib= 2.69 temp= 62.4 B_acceptable rising_fast normal valid -
t= 30 vib= 4.8 temp= 67.15 C_alert rising_fast normal valid -
t= 54 vib= 7.77 temp= 72.84 D_danger rising_fast normal valid -
t= 60 vib= 8.46 temp= 74.47 D_danger rising_fast normal valid -
wrote snapshots.jsonl
Step 3: See exactly what Kev is asked. The helper kev_console.py (in mini-projects/08-iot-sensor-simulator/) enriches a snapshot with demo plant data: a critical production order, a bearing 410 days old, and no backup motor. --show-state 5 prints the enriched state and the contract for minute 5, the first snapshot in zone C.
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --show-state 5 | less
The output
{
"state": {
"state_version": "smartplant-state/1.0.0",
"snapshot_hash": "8397592d263959d1",
"asset": {
"id": "M-104",
"criticality": "A"
},
"telemetry": {
"vibration_zone": "C_alert",
"vibration_trend": "rising_fast",
"temperature_band": "normal",
"temperature_trend": "rising_fast",
"load_band": "normal",
"latest_vibration_mm_s": 4.8,
"latest_bearing_temp_c": 67.15,
"window_minutes": 20
},
"data_quality": {
"status": "valid",
"issues": []
},
"protective_limit_reached": false,
"maintenance": {
"bearing_age": "aged",
"open_work_orders": false,
"spare_bearing_in_stock": true,
"last_inspection_finding": "none"
},
"production": {
"order_priority": "CRITICAL",
"deadline": "under_8h"
},
"redundancy": {
"backup_asset": null,
"backup_coverage": "none"
}
},
"questions": {
"intervention_urgency": {
"type": "choice",
"instructions": "How soon does `asset` need a maintenance intervention, given `telemetry`, `data_quality` and `maintenance`?",
"criteria": {
"ROUTINE": "Readings are in normal bands with stable trends; next planned service is fine",
"PRIORITY": "One signal is outside its normal band or trending up, but no protective limit is near; act within days",
"URGENT": "Several signals are degraded or one is rising fast toward a protective limit; act this shift",
"CRITICAL": "A protective limit is reached or imminent, or damage is likely within the hour"
}
},
"operational_action": {
"type": "choice",
"instructions": "Which response should operations take next for `asset`, considering `telemetry`, `production` and `redundancy`?",
"criteria": {
"MONITOR": "Asset is healthy or the change is minor; keep normal monitoring",
"INSPECT": "An operator should physically inspect the asset to confirm the condition",
"SCHEDULE_MAINTENANCE": "Condition is confirmed or clear enough to plan a repair at the next suitable opportunity",
"REQUEST_SHUTDOWN": "Asset should be stopped in a controlled way because running on risks damage or safety"
}
},
"standard_workflow": {
"type": "noul",
"instructions": "Can this case be handled through the standard maintenance work-order workflow, without escalation to a reliability specialist?",
"criteria": {
"true": "Signals agree, data is trustworthy, and the pattern is a familiar wear or service issue",
"false": "Signals conflict, data quality is questionable, or the pattern is unusual"
}
},
"operational_severity": {
"type": "score",
"instructions": "How severe is the current condition of `asset`?",
"criteria": [
{
"what": "No abnormal condition",
"examples": [
"all bands normal, trends stable"
]
},
{
"what": "Minor deviation with no functional impact",
"examples": [
"vibration in acceptable zone, slowly rising"
]
},
{
"what": "Degraded condition that will worsen without action",
"examples": [
"vibration in alert zone and rising, temperature elevated"
]
},
{
"what": "Serious condition; continued running risks equipment damage",
"examples": [
"vibration in danger zone or temperature high and rising fast"
]
},
{
"what": "Protective limit reached or failure imminent",
"examples": [
"temperature at trip band",
"vibration above danger zone and climbing fast"
]
}
]
},
"signals_conflict": {
"type": "noul",
"instructions": "Do the signals in `telemetry` point in different directions, for example temperature rising while vibration stays normal?"
}
}
}
Look at two things. The telemetry, maintenance, production and redundancy blocks are the context from 10.3. And operational_action offers only MONITOR, INSPECT, SCHEDULE_MAINTENANCE and REQUEST_SHUTDOWN: TRANSFER_LOAD was pruned because there’s no backup to transfer to. Kev can’t recommend an action the plant can’t perform.
Step 4: Give the plant a backup motor and look again.
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --show-state 5 --backup --order HIGH | less
Output
{
"state": {
"state_version": "smartplant-state/1.0.0",
"snapshot_hash": "8397592d263959d1",
"asset": {
"id": "M-104",
"criticality": "A"
},
"telemetry": {
"vibration_zone": "C_alert",
"vibration_trend": "rising_fast",
"temperature_band": "normal",
"temperature_trend": "rising_fast",
"load_band": "normal",
"latest_vibration_mm_s": 4.8,
"latest_bearing_temp_c": 67.15,
"window_minutes": 20
},
"data_quality": {
"status": "valid",
"issues": []
},
"protective_limit_reached": false,
"maintenance": {
"bearing_age": "aged",
"open_work_orders": false,
"spare_bearing_in_stock": true,
"last_inspection_finding": "none"
},
"production": {
"order_priority": "HIGH",
"deadline": "under_8h"
},
"redundancy": {
"backup_asset": "M-105",
"backup_coverage": "full"
}
},
"questions": {
"intervention_urgency": {
"type": "choice",
"instructions": "How soon does `asset` need a maintenance intervention, given `telemetry`, `data_quality` and `maintenance`?",
"criteria": {
"ROUTINE": "Readings are in normal bands with stable trends; next planned service is fine",
"PRIORITY": "One signal is outside its normal band or trending up, but no protective limit is near; act within days",
"URGENT": "Several signals are degraded or one is rising fast toward a protective limit; act this shift",
"CRITICAL": "A protective limit is reached or imminent, or damage is likely within the hour"
}
},
"operational_action": {
"type": "choice",
"instructions": "Which response should operations take next for `asset`, considering `telemetry`, `production` and `redundancy`?",
"criteria": {
"MONITOR": "Asset is healthy or the change is minor; keep normal monitoring",
"INSPECT": "An operator should physically inspect the asset to confirm the condition",
"SCHEDULE_MAINTENANCE": "Condition is confirmed or clear enough to plan a repair at the next suitable opportunity",
"TRANSFER_LOAD": "Asset should be unloaded by moving production to the backup in `redundancy`",
"REQUEST_SHUTDOWN": "Asset should be stopped in a controlled way because running on risks damage or safety"
}
},
"standard_workflow": {
"type": "noul",
"instructions": "Can this case be handled through the standard maintenance work-order workflow, without escalation to a reliability specialist?",
"criteria": {
"true": "Signals agree, data is trustworthy, and the pattern is a familiar wear or service issue",
"false": "Signals conflict, data quality is questionable, or the pattern is unusual"
}
},
"operational_severity": {
"type": "score",
"instructions": "How severe is the current condition of `asset`?",
"criteria": [
{
"what": "No abnormal condition",
"examples": [
"all bands normal, trends stable"
]
},
{
"what": "Minor deviation with no functional impact",
"examples": [
"vibration in acceptable zone, slowly rising"
]
},
{
"what": "Degraded condition that will worsen without action",
"examples": [
"vibration in alert zone and rising, temperature elevated"
]
},
{
"what": "Serious condition; continued running risks equipment damage",
"examples": [
"vibration in danger zone or temperature high and rising fast"
]
},
{
"what": "Protective limit reached or failure imminent",
"examples": [
"temperature at trip band",
"vibration above danger zone and climbing fast"
]
}
]
},
"signals_conflict": {
"type": "noul",
"instructions": "Do the signals in `telemetry` point in different directions, for example temperature rising while vibration stays normal?"
}
}
}
Now redundancy.backup_coverage is full and TRANSFER_LOAD is back on the menu. The contract changes with the plant, not just with the telemetry.
Step 5: Replay all ten minutes through the decision service. Each line is one snapshot: the zone and trend on the left, the SmartPlant recommendation on the right.
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl # Kev
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --offline # stand-in
06:01 vib= 2.0 A_good rising | ROUTINE MONITOR sev 1 [fake]
06:02 vib= 2.69 B_acceptable rising_fast | PRIORITY INSPECT sev 3 [fake]
06:05 vib= 4.8 C_alert rising_fast | URGENT INSPECT sev 4 [fake]
06:09 vib= 7.77 D_danger rising_fast | CRITICAL INSPECT sev 5 [fake]
06:10 vib= 8.46 D_danger rising_fast | CRITICAL INSPECT sev 5 [fake]
last explanation: INSPECT at CRITICAL urgency because vibration D_danger (rising_fast); bearing temperature normal (rising_fast); load normal; a critical-priority order is running (considered for planning, never overriding equipment protection); backup coverage none.
Even in zone D, the answer stays INSPECT: with no backup, transferring load isn’t possible, and the critical order is “considered for planning, never overriding equipment protection”. That’s the safety floor from 10.4 at work.
Step 6: Change the plant and watch the answer change.
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --offline --backup --order HIGH | tail -4
06:09 vib= 7.77 D_danger rising_fast | CRITICAL TRANSFER_LOAD sev 5 [fake] APPROVAL REQUIRED
06:10 vib= 8.46 D_danger rising_fast | CRITICAL TRANSFER_LOAD sev 5 [fake] APPROVAL REQUIRED
last explanation: TRANSFER_LOAD at CRITICAL urgency because vibration D_danger (rising_fast); bearing temperature normal (rising_fast); load normal; a high-priority order is running (considered for planning, never overriding equipment protection); backup coverage full.
Same telemetry, different plant: with a backup available, zone D now leads to a load transfer, and it’s flagged for operator approval. Chapter 11 turns that flag into a real approval step. Try --order NONE too, and notice the explanation drops the production clause.
Step 7: Compare Kev with the stand-in, side by side.
diff <(uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --offline) \
<(uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl)
Every line that differs is a place where the real model disagrees with the scripted one. Those are the cases worth reading closely, and the reason Chapter 12 builds a proper evaluation.
Step 8: Run the scenario catalogue. The same service answers the ten labelled scenarios. S04 (sensor dropout) and S10 (service timeout) never reach the model; S06 does.
uv run python mini-projects/09-smartplant-decision-service/run_decision_service.py --offline S04 S06 S10
DECISION_BACKEND=kev
uv run python mini-projects/09-smartplant-decision-service/run_decision_service.py S04 S06 S10
S04 Faulty temperature sensor (dropout)
result: PRIORITY / INSPECT / severity 3 / specialist_review [fallback, 0.0 ms]
S06 Severe degradation, backup available
result: CRITICAL / TRANSFER_LOAD / severity 5 / specialist_review / APPROVAL REQUIRED [fake, 0.0 ms]
S10 Decision service timeout during degradation
result: PRIORITY / INSPECT / severity 3 / specialist_review [fallback, 0.0 ms]
(Output trimmed to the result lines.) Bad data and a dead service both go to fallback, and only trustworthy evidence reaches Kev.
Step 9: Stop Kev and replay once more. Press Ctrl+C in terminal 1, then rerun step 5 without --offline. Every line now comes from [fallback]: the circuit breaker opens after a few failures and the deterministic rules take over. Chapter 11 builds on exactly this behaviour.
Tests
Each test scripts a fake model’s answers, which lets us prove the guards hold even when the model gets it wrong.
# smartplant-capstone/tests/test_decision_service.py
from __future__ import annotations
from typing import Any
from triage.jev_client import ChoiceResult, FakeDecisionClient, NoulResult, ScoreResult
from triage.resilience import CircuitBreaker
from smartplant.contract import build_questions, eligible_actions
from smartplant.decision_service import SmartPlantDecisionService
def model(urgency: str, action: str, sev: float, std: float = 0.8, conf: float = 0.6, conflict: float = 0.1):
def rule(state: Any, qs: Any):
offered = list(qs["operational_action"]["criteria"])
return {
"intervention_urgency": ChoiceResult(urgency, {urgency: conf}, conf),
"operational_action": ChoiceResult(action, {a: (conf if a == action else 0.0) for a in offered}, conf),
"standard_workflow": NoulResult(std),
"operational_severity": ScoreResult(sev, {}, 0.7, 5),
"signals_conflict": NoulResult(conflict),
}
return rule
def test_transfer_load_not_offered_without_backup(state_for) -> None:
st = state_for("S07")
assert "TRANSFER_LOAD" not in eligible_actions(st)
assert "TRANSFER_LOAD" not in build_questions(st)["operational_action"]["criteria"]
def test_model_cannot_go_below_safety_floor(state_for) -> None:
svc = SmartPlantDecisionService(FakeDecisionClient(model("ROUTINE", "MONITOR", 0.5)))
r = svc.evaluate(state_for("S06")) # vibration in danger zone
assert r.urgency == "URGENT" and r.action == "INSPECT"
assert any("floor_raised" in x for x in r.reasons)
def test_production_priority_never_suppresses_inspection(state_for) -> None:
svc = SmartPlantDecisionService(FakeDecisionClient(model("ROUTINE", "MONITOR", 1.0)))
r = svc.evaluate(state_for("S05")) # CRITICAL order, alert-zone vibration
assert r.action != "MONITOR"
def test_transfer_requires_approval(state_for) -> None:
svc = SmartPlantDecisionService(FakeDecisionClient(model("URGENT", "TRANSFER_LOAD", 3.0)))
r = svc.evaluate(state_for("S06"))
assert r.action == "TRANSFER_LOAD" and r.requires_approval
def test_conflict_and_low_confidence_go_to_specialist(state_for) -> None:
svc = SmartPlantDecisionService(FakeDecisionClient(model("PRIORITY", "INSPECT", 2.0, conflict=0.9)))
assert svc.evaluate(state_for("S09")).route == "specialist_review"
svc = SmartPlantDecisionService(FakeDecisionClient(model("PRIORITY", "INSPECT", 2.0, conf=0.1)))
assert "low_confidence" in svc.evaluate(state_for("S02")).reasons
def test_invalid_data_never_reaches_model(state_for) -> None:
calls = []
svc = SmartPlantDecisionService(FakeDecisionClient(lambda s, q: calls.append(1) or {}))
r = svc.evaluate(state_for("S04"))
assert not calls and r.source == "fallback" and r.route == "specialist_review"
def test_outage_falls_back_and_opens_breaker(state_for) -> None:
br = CircuitBreaker(failure_threshold=2, cooldown_s=60)
svc = SmartPlantDecisionService(FakeDecisionClient(model("PRIORITY", "INSPECT", 2.0), fail_times=5), br)
st = state_for("S02")
for _ in range(3):
r = svc.evaluate(st)
assert r.source == "fallback" and r.action == "INSPECT"
assert br.state.value == "open"
def test_invalid_label_from_backend_falls_back(state_for) -> None:
svc = SmartPlantDecisionService(FakeDecisionClient(model("PRIORITY", "REBOOT_PLC", 2.0)))
r = svc.evaluate(state_for("S02"))
assert r.source == "fallback" and "fallback:invalid_response" in r.reasons
10.6 Checklist
- You ask four atomic questions plus one speculative question in one call, with urgency as a Choice and severity as a Score.
- Ineligible actions are pruned in code, such as
TRANSFER_LOADwithout backup coverage. - Enrichment computes every comparison and hands Kev buckets.
- The safety floor comes from code facts, and production priority can’t lower it.
- Invalid data and an open breaker never call the model.
- Kev thresholds are calibrated on your checkpoint, and you’ve run the permute check on
operational_action. - Explanations are built only from recorded facts.
What’s next
A recommendation is still only a recommendation. Chapter 11 turns it into action: authorisation rules, operator approval for load transfer and shutdown, idempotent CMMS work orders, an immutable decision trail, and fallbacks for stale telemetry, failed sensors and an unreachable Kev.