Chapter 5 — Confidence-Gated Architecture & Dynamic Routing

Previously: Chapter 4 gave us a clean evidence state and six typed answers from one call.

You will build: an autonomous Chief-of-Staff task router with auditable JSON queues. A model suggests a refund route. Does that mean the system should open a refund request immediately? The answer depends on more than the label: the available evidence, transaction status, approval rules, and the cost of a wrong route all matter.

A general LLM’s written “confidence: 0.9” cannot settle this. Even a probability produced by a dedicated decision model needs validation on the cases you intend to automate.

This chapter builds policy: explicit application rules that accept, redirect, or escalate a proposal.

Capstone increment: capstone-project/src/triage/policy_engine.py (confidence bands, per-target gates, overrides) and the decision trace.

5.0 The problem

Jev just told you: “send this to the refund pipeline, confidence 0.52.” What now?

If you blindly follow it, a coin-flip answer can move money. If you ignore every answer below 0.99, you automate almost nothing. And if the answer turns out wrong next week, you will have no record of why the system did what it did.

So we need three things the model cannot give us:

1. a rule for how sure is sure enough, set by how costly a mistake is (§5.1 to §5.3),
2. code that applies the rule and the business’s hard limits (§5.4),
3. a record of every decision, so mistakes can be explained and fixed (§5.5 to §5.7).

The model proposes a branch. Code decides whether that branch may execute. This chapter is the code that decides.

5.1 What confidence is (and isn’t)

• Confidence is concentration. It’s computed from the shape of the distribution: high when one option dominates, low when probability is spread out.
• Confidence isn’t correctness. Jev is trained with Reinforcement Learning for Calibrated Decisions (RLCD), so across many predictions, answers given 90 % should be right about 90 % of the time. Any single answer can still be wrong.
• Noul has no confidence field. The probability is the signal: 0.97 is a confident yes, 0.50 is “can’t tell”.
• LLM-reported probabilities aren’t calibrated the same way. If you prototyped on the adapter (Chapter 2), re-tune every threshold after switching to Jev.

💡 Mental Model: a weather forecast “90% chance of rain” doesn’t mean it will rain tomorrow. It means that on days with that forecast, it rains about nine times out of ten. Jev’s probabilities work the same way: reliable over many decisions, never a promise about one.

Log the top-two margin next to confidence. In a 4-way Choice, a 0.48 vs 0.46 split and a 0.48 vs 0.20 split can have similar confidence and mean different things.

5.2 Three bands, thresholds by consequence

Starting points from the Codila playbook. Measure on your data, then move them:

For the triage engine, the thresholds are asymmetric by destination. Sending a benign ticket to SecOps wastes a few minutes. Sending a real account takeover to Tier-1 loses money. So SecOps gets a lower bar than the refund pipeline.

💡 Mental Model: withdrawal limits A bank lets you check a balance with no questions, withdraw ₹5,000 with a PIN, and wire ₹5 lakh only after a call. The higher the consequence, the higher the bar. Confidence gates are withdrawal limits for decisions.

5.3 Routing work to the right model

The same pattern chooses which expensive resource to spend:

ROUTER = {
    "needs_reasoning": {"type": "score", "instructions": "How much thought does a good reply to `message` need?",
                        "criteria": ["A lookup or one-line fact",
                                     "A short explanation using product knowledge",
                                     "A judgment call with trade-offs or an unhappy customer"]},
}
# score < 0.7  -> template / DB lookup (no LLM)
# score < 1.5  -> small fast LLM
# otherwise    -> frontier LLM or human

A Score of “how much thought does this need?” becomes a router: cheap lookups never touch an LLM, and only hard cases reach the expensive one.

Flavio Copes’s A Deep Dive into Jev - intent-routing example, order-status lookups never touch an LLM, two intents get an LLM with different context, and complaints use a Score to choose between LLM and human. Expensive resources run only when needed.

5.4 Capstone increment: policy_engine.py

📄 Full source: capstone-project/src/triage/policy_engine.py

All risk tolerance lives in this one file, as numbers. We build it in five steps: constants → thresholds → the output shape → the decision ladder → tests.

Step 1: versions and vocabulary

# capstone-project/src/triage/policy_engine.py
"""Turns probabilities into an allowed destination. All risk tolerance lives here, in numbers."""
from __future__ import annotations

from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Any, Final, Literal, Mapping

from triage.contract import DispatchTarget, to_business_urgency
from triage.jev_client import DecisionResponse

POLICY_VERSION: Final[str] = "triage-policy/1.0.0"
Band = Literal["low", "medium", "high"]
Priority = Literal["P1", "P2", "P3", "P4"]

• POLICY_VERSION is stored on every decision, so you can tell which rules produced it.
• Band is how sure we are; Priority is how fast humans must react.

Step 2: thresholds as data

Every number lives in one frozen dataclass. First the single-value thresholds:

@dataclass(frozen=True, slots=True)
class Thresholds:
    low_floor: float = 0.50
    fraud_override: float = 0.85
    fraud_suspect: float = 0.50
    manipulation: float = 0.50

• low_floor: below this Choice confidence, no auto-routing at all.
• fraud_override / fraud_suspect: when the fraud Noul alone overrides the model’s own routing.
• manipulation: when “text addressed to an AI” blocks automation.

Then the per-destination gates. Note how SecOps has a lower bar than refunds:

    min_conf: Mapping[str, float] = field(default_factory=lambda: MappingProxyType({
        "tier1_support": 0.70,
        "secops_agent": 0.60,        # false positive is cheap
        "refund_pipeline": 0.90,     # money moves
        "manual_escalation": 0.0,
    }))

    high_band: Mapping[str, float] = field(default_factory=lambda: MappingProxyType({
        "tier1_support": 0.85, "secops_agent": 0.80, "refund_pipeline": 0.95, "manual_escalation": 0.0,
    }))


DEFAULT_THRESHOLDS: Final = Thresholds()

💡 Mental Model: two dials per destination min_conf is the door: below it, the case is escalated to a human. high_band is the express lane: above it, no confirmation step is needed. Between the two, the case goes through but is flagged for confirmation.

Step 3: the output shape

The engine returns a single record, so every caller sees the same fields:

@dataclass(frozen=True, slots=True)
class PolicyDecision:
    destination: DispatchTarget
    band: Band
    priority: Priority
    urgency: int                      # business scale 1..5
    requires_confirmation: bool
    requires_human_approval: bool
    reasons: tuple[str, ...]
    policy_version: str = POLICY_VERSION

def _priority(urgency: int, fraud_p: float) -> Priority:
    if urgency >= 5 or (urgency >= 4 and fraud_p >= 0.5):
        return "P1"
    return "P2" if urgency >= 4 else "P3" if urgency >= 3 else "P4"

Step 4: the decision ladder

decide() reads the model’s three answers, then walks four rules in order. The first rule that fires wins. Start with the setup:

def decide(resp: DecisionResponse, state: Mapping[str, Any],
           t: Thresholds = DEFAULT_THRESHOLDS) -> PolicyDecision:
    fraud_p = resp.noul("is_fraud_suspected").p_yes
    urgency = to_business_urgency(resp.score("urgency_rating").score)
    target = resp.choice("dispatch_target")
    prio = _priority(urgency, fraud_p)
    reasons: list[str] = []

Rule 1: safety first. Text that tries to steer automation never auto-routes.

    # 1. Safety first: text that tries to steer automation never auto-routes.
    manip = resp.answers.get("contains_instructions_to_system")
    if manip is not None and resp.noul("contains_instructions_to_system").p_yes >= t.manipulation:
        return PolicyDecision("manual_escalation", "low", prio, urgency, False, True,
                              ("possible_manipulation",))

Rule 2: a fraud signal overrides the model’s own routing choice. Even if the Choice said tier1_support, a fraud Noul at 0.93 goes to SecOps.

    # 2. Fraud signal overrides the model's own routing choice.
    if fraud_p >= t.fraud_override:
        reasons.append(f"fraud_override p={fraud_p:.2f}")
        return PolicyDecision("secops_agent", "high", prio, urgency, False, False, tuple(reasons))
    if fraud_p >= t.fraud_suspect:
        reasons.append(f"fraud_suspected p={fraud_p:.2f}")
        return PolicyDecision("secops_agent", "medium", prio, urgency, True, False, tuple(reasons))

Rule 3: the per-destination confidence gate. Too uncertain means a human.

    # 3. Per-destination confidence gate.
    dest: DispatchTarget = target.choice  # type: ignore[assignment]  # contract guarantees label set
    if target.confidence < t.low_floor or target.confidence < t.min_conf[dest]:
        reasons.append(f"low_confidence {dest}@{target.confidence:.2f} margin={target.margin():.2f}")
        return PolicyDecision("manual_escalation", "low", prio, urgency, False, True, tuple(reasons))

    band: Band = "high" if target.confidence >= t.high_band[dest] else "medium"

Rule 4: hard business rules. These are deterministic and never delegated to the model. A refund above ₹50,000 always needs a human approval, however confident Jev is.

    # 4. Hard business rules: deterministic, never delegated to the model.
    needs_approval = False
    if dest == "refund_pipeline":
        amount_band = state.get("computed_signals", {}).get("disputed_amount_band")
        if amount_band == "above_50k":
            needs_approval = True
            reasons.append("refund_above_50k_requires_approval")
    reasons.append(f"{dest}@{target.confidence:.2f}")
    return PolicyDecision(dest, band, prio, urgency, band == "medium", needs_approval, tuple(reasons))

⚠️ Watch Out Rule order is the policy. Rule 1 comes before Rule 2 on purpose: if a message is both a manipulation attempt and a high-fraud-score message, we want a human, not an automatic SecOps route an attacker could trigger. When you add a rule, decide its rung deliberately and write a test.

Step 5: test the policy with no network

It’s pure code, so tests need no network and have no flakiness. First a helper that fabricates a response with the fake types from Chapter 2:

# capstone-project/tests/test_policy_engine.py
from triage.jev_client import ChoiceResult, DecisionResponse, NoulResult, ScoreResult
from triage.policy_engine import decide

def resp(fraud: float, urgency_raw: float, target: str, conf: float, manip: float = 0.0) -> DecisionResponse:
    probs = {k: 0.0 for k in ("tier1_support", "secops_agent", "refund_pipeline", "manual_escalation")}
    probs[target] = conf
    return DecisionResponse(
        answers={
            "is_fraud_suspected": NoulResult(fraud),
            "urgency_rating": ScoreResult(urgency_raw, {}, 0.9, 5),
            "dispatch_target": ChoiceResult(target, probs, conf),
            "contains_instructions_to_system": NoulResult(manip),
        },
        model="fake-0", input_tokens=None, latency_ms=0.0, backend="fake",
    )

Then one test per rule:

def test_fraud_overrides_model_route() -> None:
    d = decide(resp(0.93, 3.8, "tier1_support", 0.9), {})
    assert d.destination == "secops_agent" and d.priority == "P1"

def test_low_confidence_refund_escalates() -> None:
    d = decide(resp(0.05, 2.0, "refund_pipeline", 0.72), {})
    assert d.destination == "manual_escalation" and d.requires_human_approval

def test_big_refund_needs_approval_even_when_confident() -> None:
    d = decide(resp(0.02, 2.9, "refund_pipeline", 0.97),
               {"computed_signals": {"disputed_amount_band": "above_50k"}})
    assert d.destination == "refund_pipeline" and d.requires_human_approval

def test_manipulation_never_auto_routes() -> None:
    d = decide(resp(0.0, 1.0, "refund_pipeline", 0.99, manip=0.8), {})
    assert d.destination == "manual_escalation"

📄 Full source: capstone-project/tests/test_policy_engine.py

5.5 The decision trace: what you log

Every decision writes one record. When a route goes wrong, this is how you tell wrong input, wrong model answer, wrong threshold, executor failure and false DONE apart.

💡 Mental Model: a flight recorder Nobody records a flight to watch it. They record it so that when something goes wrong, they can replay exactly what the pilot saw, what the instruments said and what was done. The decision trace is that recorder.

📄 Full source: capstone-project/src/triage/trace.py

Imports.

# capstone-project/src/triage/trace.py
from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

from triage.contract import CONTRACT_VERSION, contract_hash
from triage.jev_client import ChoiceResult, DecisionResponse, NoulResult, ScoreResult
from triage.policy_engine import PolicyDecision

Two helpers. A short hash of the state (so you can compare evidence without storing it), and a converter that adds the margin to Choice answers.

def state_hash(state: Mapping[str, Any]) -> str:
    return hashlib.sha256(json.dumps(state, sort_keys=True, default=str).encode()).hexdigest()[:16]

def _answer_dict(a: NoulResult | ChoiceResult | ScoreResult) -> dict[str, Any]:
    d = asdict(a)
    if isinstance(a, ChoiceResult):
        d["margin"] = round(a.margin(), 4)
    return d

The trace record. Every field is something from the table above:

@dataclass(slots=True)
class DecisionTrace:
    report_id: str
    state_hash: str
    contract_version: str
    contract_hash: str
    model: str
    backend: str
    latency_ms: float
    input_tokens: int | None
    answers: dict[str, Any]
    policy: dict[str, Any]
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    executor_receipt: dict[str, Any] | None = None
    verification: dict[str, Any] | None = None

Building one from a decision. One call turns a response and a policy decision into a trace:

    @classmethod
    def build(cls, report_id: str, state: Mapping[str, Any], questions: Mapping[str, Any],
              resp: DecisionResponse, decision: PolicyDecision) -> "DecisionTrace":
        return cls(
            report_id=report_id, state_hash=state_hash(state),
            contract_version=CONTRACT_VERSION, contract_hash=contract_hash(questions),
            model=resp.model, backend=resp.backend, latency_ms=round(resp.latency_ms, 1),
            input_tokens=resp.input_tokens,
            answers={k: _answer_dict(v) for k, v in resp.answers.items()},
            policy=asdict(decision),
        )

Writing it. Append-only JSON lines, easy to ship to any log store:

    def append_jsonl(self, path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(self), default=str) + "\n")

🔒 Redact PII before traces leave the service boundary. Store the state hash in shared logs and keep the full state in an access-controlled store.

5.6 Calibrating a threshold (don’t guess it)

Split labeled examples into development and holdout. On dev, sweep thresholds and pick the lowest one that meets the required error rate for that action. Freeze it, evaluate once on holdout, and store the result next to the policy version. If you keep re-tuning on holdout, it has become training data.

⚠️ Watch Out Looking at the holdout set twice quietly turns it into a training set. Tune on dev, test once on holdout, write the number down.

📄 Full source: capstone-project/tools/sweep_threshold.py

Two small records. One labeled example, and one point on the sweep curve:

# capstone-project/tools/sweep_threshold.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Sequence

@dataclass(frozen=True, slots=True)
class Row:
    predicted: str
    confidence: float
    expected: str

@dataclass(frozen=True, slots=True)
class SweepPoint:
    threshold: float
    automation_share: float   # fraction auto-routed
    precision: float          # accuracy among auto-routed
    auto_errors: int

The sweep. For each candidate threshold, how much would we automate, and how accurate would that automation be?

def sweep(rows: Sequence[Row], target: str, grid: Sequence[float]) -> list[SweepPoint]:
    rel = [r for r in rows if r.predicted == target]
    out: list[SweepPoint] = []
    for th in grid:
        auto = [r for r in rel if r.confidence >= th]
        correct = sum(r.predicted == r.expected for r in auto)
        out.append(SweepPoint(th, len(auto) / max(1, len(rel)),
                              correct / len(auto) if auto else 1.0, len(auto) - correct))
    return out

The pick. The lowest threshold that still meets your precision target maximises automation:

def pick(points: Sequence[SweepPoint], min_precision: float) -> float | None:
    ok = [p for p in points if p.precision >= min_precision and p.automation_share > 0]
    return min(ok, key=lambda p: p.threshold).threshold if ok else None

Run it on a JSONL file of labeled rows:

if __name__ == "__main__":
    import json, sys
    data = [Row(**json.loads(l)) for l in open(sys.argv[1], encoding="utf-8")]
    pts = sweep(data, "refund_pipeline", [x / 100 for x in range(50, 100, 5)])
    for p in pts:
        print(f"th={p.threshold:.2f} auto={p.automation_share:5.1%} precision={p.precision:6.1%} errors={p.auto_errors}")
    print("chosen:", pick(pts, min_precision=0.99))

Watch four numbers per destination: precision of auto-routed, automation share, escalation share, and total workflow cost. Raise thresholds for irreversible actions. Lower them only when unnecessary escalations cost more than the occasional wrong route.

5.7 Rollout: earn autonomy one branch at a time

5.8 Mini-project 4: the Chief-of-Staff task router

A CLI that accepts a goal, asks Jev for worker (Choice), complexity (Score) and needs approval (Noul), gates it, and writes an auditable job into queue/<destination>/.

File 1: the queue (queue_manager.py)

📄 Full source: mini-projects/04-chief-of-staff-router/queue_manager.py

The four allowed queues, and a function that refuses unknown ones:

# mini-projects/04-chief-of-staff-router/queue_manager.py
from __future__ import annotations

import json
import os
import tempfile
import uuid
from pathlib import Path
from typing import Any, Mapping

QUEUES = ("research", "write", "tool", "review")

def enqueue(root: Path, destination: str, job: Mapping[str, Any]) -> Path:
    if destination not in QUEUES:
        raise ValueError(f"unknown queue {destination!r}")
    folder = root / destination
    folder.mkdir(parents=True, exist_ok=True)
    final = folder / f"{uuid.uuid4().hex}.json"

Then an atomic write, so a worker never sees a half-written job:

    # Atomic write: a worker never sees a half-written job.
    fd, tmp = tempfile.mkstemp(dir=folder, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(job, f, indent=2, default=str)
        os.replace(tmp, final)
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise
    return final

💡 Mental Model: write to a scratch file, then rename Renaming a file is instant and all-or-nothing, while writing a file takes time and can be interrupted. So write to a temp file, then os.replace it into place. Readers see either no job or a whole job.

File 2: the router (chief.py)

📄 Full source: mini-projects/04-chief-of-staff-router/chief.py

Snippet 1: imports.

# mini-projects/04-chief-of-staff-router/chief.py
from __future__ import annotations

import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Final

from queue_manager import enqueue
from triage.jev_client import DecisionError, client_from_env

Snippet 2: the three questions and the per-queue gates. review has a gate of 0.0 because it is the human exit and always accepts.

QUESTIONS: Final = {
    "worker": {"type": "choice", "instructions": "Who should handle `goal` next?",
               "criteria": {"research": "Needs finding or validating external evidence before anything is written",
                            "write": "Needs drafting, rewriting or summarising text from evidence already in hand",
                            "tool": "One deterministic operation: a lookup, a file move, an API call",
                            "review": "Sensitive, irreversible, or unclear; a person must look first"}},
    "complexity": {"type": "score", "instructions": "How complex is the next step for `goal`?",
                   "criteria": ["Mechanical", "Multi-step", "Ambiguous or high-risk"]},
    "needs_approval": {"type": "noul",
                       "instructions": "Would the next step for `goal` send, publish, pay, or delete something?"},
}
MIN_CONF: Final = {"research": 0.70, "write": 0.70, "tool": 0.90, "review": 0.0}

Snippet 3: the input shape. The state is just the goal and the list of valid workers.

def main() -> int:
    ap = argparse.ArgumentParser(description="Route a goal to a worker queue via Jev")
    ap.add_argument("goal")
    ap.add_argument("--queue-root", type=Path, default=Path("queue"))
    args = ap.parse_args()
    state = {"goal": args.goal.strip(), "available_workers": list(MIN_CONF)}

Snippet 4: the single decision call, with a safe failure path. If Jev is down, the goal goes to review; it is never lost.

    try:
        r = client_from_env().decide(state, QUESTIONS)
    except DecisionError as e:
        path = enqueue(args.queue_root, "review", {"state": state, "error": str(e)})
        print(f"decision unavailable -> review: {path}", file=sys.stderr)
        return 2

Snippet 5: branch on the typed output. Three independent guardrails can each send the job to a human:

    worker, approval = r.choice("worker"), r.noul("needs_approval")
    dest = worker.choice
    if approval.p_yes >= 0.70 or worker.confidence < MIN_CONF[dest] or r.score("complexity").score > 1.5:
        dest = "review"

• approval-needed probability ≥ 0.70 (it would send, publish, pay or delete),
• confidence below that queue’s gate,
• complexity above 1.5 on the 0 to 2 scale.

Snippet 6: write the auditable job.

    job = {"created_at": datetime.now(timezone.utc).isoformat(), "model": r.model, "state": state,
           "decision": {"worker": worker.choice, "probabilities": dict(worker.probabilities),
                        "confidence": worker.confidence, "complexity": r.score("complexity").score,
                        "needs_approval": approval.p_yes},
           "destination": dest, "latency_ms": round(r.latency_ms, 1)}
    print(f"queued for {dest}: {enqueue(args.queue_root, dest, job)}")
    return 0

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

Run it. A research task is queued for research, and a destructive request goes to a human:

sheikudumanali@sheikmac $ uv run mini-projects/04-chief-of-staff-router/chief.py "Find the last three RBI circulars on UPI fraud liability and summarise them"
queued for research: queue/research/d219....json
sheikudumanali@sheikmac $ uv run mini-projects/04-chief-of-staff-router/chief.py "Delete all rows in the staging customers table"
queued for review: queue/review/9406...json

5.9 Checklist

• Every threshold is a named constant in policy_engine.py with a version
• Thresholds differ by consequence, not one global number
• Deterministic business rules (amount limits, permissions) run after the model and can only make things safer
• Decision trace stores full distribution, margin, thresholds, model id, contract hash
• Thresholds chosen on dev, checked once on holdout, recorded
• Rollout starts in shadow mode

What’s next

Our gates assume the model’s answer and the input are honest, and that the service is up. Chapter 6 drops all three assumptions: what Jev can’t do (math, dates, counting), prompt injection in customer text, and a circuit breaker with deterministic fallback so a Jev outage never drops a fraud report.