Chapter 11 — Industrial Decision Modeling with Kev Primitives

Previously: Chapter 10 gave us trustworthy, versioned edge snapshots and a KevDecisionClient.

You will 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.

SmartPlant increment: smartplant.contract, context_enricher, models, decision_service and the packaged test_scenarios.json.

11.0 The problem

A threshold alarm on the SCADA screen says something is wrong. It doesn’t say what to do, how soon, or whether doing it is even possible tonight. If you hand a language model the raw numbers and ask “what now?”, you get a fluent paragraph, and a paragraph can’t be routed into a work-order system, checked against a rule, or tested.

So we split the job into three layers, each in the place it does best:

Code gathers facts and does every comparison. Is the backup big enough? How close is the deadline?
The model answers a few small, typed questions about those facts.
Code checks the answers against hard safety rules before anyone acts.

The rest of this chapter builds those layers in that order.

11.1 From an alarm to four clear questions

We turn each missing piece of the alarm into one small question and send all four to Kev in a single fan-out call (Chapter 4).

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.

💡 Mental Model: label vs. degree 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. Asking both gives you a consistency check for free.

⚠️ Watch Out 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.

11.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. We’ll read the module in small pieces.

Step 1: Assemble the relevant evidence

state = {
    "asset": "M-104",
    "telemetry": {"vibration_zone": "D_danger", "trend": "rising_fast"},
    "maintenance": {"bearing_age_band": "overdue"},
    "production": {"order_priority": "CRITICAL"},
    "redundancy": {"backup_coverage": "none"},
    "data_quality": "valid",
}

Telemetry describes the condition. Maintenance adds relevant history. Production informs planning. Redundancy determines whether load transfer is possible. Data quality tells the service whether to ask for model judgment at all.

Define a small action contract for this example:

questions = {"operational_action": {
    "type": "choice",
    "instructions": "Which response should operations take next for `asset`?",
    "criteria": {
        "MONITOR": "Healthy condition with no meaningful deterioration",
        "INSPECT": "Confirm an abnormal condition with an inspection",
        "SCHEDULE_MAINTENANCE": "Plan corrective work for a clear degradation",
        "REQUEST_SHUTDOWN": "Request a controlled stop because running risks damage",
    },
}}

TRANSFER_LOAD is absent because no backup is available. This is an eligibility decision made before the model call, rather than a suggestion hidden in the instructions.

Step 2: Ask for the action proposal

from triage.jev_client import client_from_env

response = client_from_env().decide(state, questions)
action = response.choice("operational_action")

This isolated fragment assumes valid evidence and no protective trip. The complete service handles those gates first.

Step 3: Keep the proposal behind policy

review_required = action.confidence < 0.80  # Example gate; tune per checkpoint.
if action.choice == "REQUEST_SHUTDOWN":
    next_step = "authorized_operator_review"
elif review_required:
    next_step = "reliability_specialist_review"
else:
    next_step = "apply_equipment_policy"

Even a confident answer still passes equipment policy. We have not established a total ordering in which every action can safely replace every other action.

Snippet 1: the vocabulary. Two literal types pin down the only urgencies and actions that exist, and a version string identifies the contract in every audit record.

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

Snippet 2: the urgency question. instructions is the question, and criteria describes each label in plant terms. Backticked names such as telemetry refer to fields of the state.

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

Snippet 3: the action question. Same shape. Note that TRANSFER_LOAD is defined here, but whether it is offered is decided later, in code.

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

Snippet 4: the yes/no question. A Noul takes true/false criteria.

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

Snippet 5: the severity scale. A Score takes an ordered list of levels, each with examples. Five levels here, one example is enough per level to anchor it.

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

Snippet 6: the speculative question. Optional, instructions only.

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?",
    },
}

Snippet 7: validate the contract at import time. A malformed contract fails loudly when the module loads, not in production.

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

Snippet 8: prune options in code. eligible_actions() is a plain rule. If there’s no backup, TRANSFER_LOAD is removed before the call.

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]

Snippet 9: assemble the questions for one state. Copy the template, drop ineligible actions, add the speculative question, validate.

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

Snippet 10: two small helpers. A short hash identifies the exact contract used. The severity mapper turns Kev’s 0..4 score into the plant’s 1..5 scale, and rejects anything out of range.

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)

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

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.

💡 Mental Model: rules in code, judgment in criteria Criteria text describes situations. Anything that is a hard rule (“never offer an action that isn’t possible”) belongs in code, where it can be tested.

ch11-contract

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

ch11-enrichment

Snippet 0: imports and versioning. The module docstring states the rule of the chapter: code computes, Kev reads buckets.

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

Snippet 1: the three data records. Small frozen dataclasses. Each field is a fact the model will later see as a bucket, not as a raw number.

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

Snippet 2: the read services as Protocols. Any object with these methods will do, so tests and production can plug in different implementations.

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

Snippet 3: a mock that implements all three. MockPlantData is seeded per scenario.

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

Snippet 4: let code do the comparing. These helpers turn numbers into labels. Kev never compares 78 % with 91 %.

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"

Snippet 5: enrich(), step one. Fetch the three records, then work out the load the line needs.

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)

Snippet 6: enrich(), step two. Build the state: telemetry bands from the snapshot, plus data quality and the protective-limit flag.

    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,

Snippet 7: enrich(), step three. Add the plant context as buckets.

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

Snippet 8: a one-line summary for operator screens and logs.

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

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

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.

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

ch11-decision-service

Snippet 1: one result type for every engine. Rules, LLM or Kev all return a Recommendation, 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

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

The service reuses two Part I modules unchanged, triage.jev_client and triage.resilience.

Snippet 1b: imports. Note the Part I imports, reused unchanged.

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

Snippet 2: starting thresholds for Kev. Gates are lower than Part I’s, because Kev’s confidence statistic runs lower on the same evidence.

@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

Snippet 3: the service object. A client, a circuit breaker (which stops calls to a service known to be down) and the thresholds.

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

Snippet 4: the guard sequence in evaluate(). 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 breaker falls back too. Only then does the model get asked.

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

Snippet 5: the call, and what failure does. A retryable error counts toward opening the breaker. Any failure returns a deterministic fallback, never an exception to the operator.

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

Snippet 6: validate the answer before using it. A backend can return a label that doesn’t exist. That becomes a fallback too.

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

⚠️ Watch Out Treat the model’s answer as untrusted input. Even a well-behaved model can return a label that was never offered. _apply_policy() checks for that explicitly.

Snippet 7: read the answers. Pull each typed result and convert severity to the plant scale. Reject labels outside the contract.

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

Snippet 8: guard the offered menu.

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

Snippet 9: the safety floor, applied. The floor is the minimum response implied by code facts. A model may escalate past it, never below it.

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

Snippet 10: routing to the right people. A case goes to specialist review 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.

        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"

Snippet 11: build the result. TRANSFER_LOAD and REQUEST_SHUTDOWN always require approval.

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

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

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.

# 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. Treat them as starting points and calibrate them, meaning tune them against known outcomes on your own setup:

Replay the labeled scenarios (Chapter 13) against your Kev checkpoint.
Sweep each gate with capstone-project/tools/sweep_threshold.py (Chapter 5), and pick the lowest threshold that meets the error budget for that action.
Call /v1/systemone/permute on operational_action for 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.

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

Snippet 1: command-line options.

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

Snippet 2: the service, and the S10 override. Everything shares one service, except the scenario that simulates an outage.

    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)

Snippet 3: print the result.

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

📄 Full source: mini-projects/09-smartplant-decision-service/run_decision_service.py

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 run returns:

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 (which took about 4045 ms above):

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_LOAD is 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 10’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 10. 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

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

Snippet 1: a scriptable fake model. model(...) returns a function that answers every question with the values you choose.

# 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

Snippet 2: options that are not eligible are never offered.

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

Snippet 3: the model can’t go below the floor. Even if it says ROUTINE/MONITOR, danger-zone vibration forces URGENT/INSPECT.

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)

Snippet 4: production priority never suppresses an inspection.

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"

Snippet 5: risky actions need approval.

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

Snippet 6: conflict and low confidence go to a specialist.

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

Snippet 7: bad data never reaches the model.

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"

Snippet 8: an outage opens the breaker and falls back.

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"

Snippet 9: an invalid label becomes a fallback.

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

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

11.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_LOAD without 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 12 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.

Enjoying the book?

If this guide was helpful, consider starring the project or sharing your thoughts:

⭐ Star on GitHub 📝 Send Anonymous Feedback 💬 Share on X / LinkedIn

Keyboard shortcuts

Mastering TypeSafe Jev

Enjoying the book?