Chapter 4 — Confidence-Gated Architecture & Dynamic Routing

Previously: Chapter 3 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.

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

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

4.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 1), re-tune every threshold after switching to Jev.

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.

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

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

In the Deep Dive’s 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.

4.4 Capstone increment: policy_engine.py

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


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


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


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

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

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

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

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

Test the policy with the fake client from Chapter 1. It’s pure code, so no network and no flakiness:

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


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"

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

# 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


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


@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

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

    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.

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

# 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


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


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


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.

4.7 Rollout: earn autonomy one branch at a time

4.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>/.

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

# 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

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}


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

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

4.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 5 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.