Chapter 7 — Agentic Control Layers & Framework Integrations

The router chooses the correct queue. A downstream call replies “success.” Can we close the case? Only if the intended record exists with the expected state.

A general LLM can describe an action as complete without observing the external system. An executor can also receive an acknowledgment before a write is visible or durable. Our application therefore needs a separate verification step.

Previously: Chapter 6 made the engine safe and resilient, with limits, gates, a circuit breaker, and labeled evals.

You will build: a two-stage BM25 + Jev re-ranker.

Capstone increment: executors.py and verifier.py, which close the loop so a decision becomes one observable action followed by fresh-state proof.

7.0 The problem

Up to now our engine reaches a decision and stops, like a manager who says “send this to SecOps” and walks away. Real systems must then do the thing, and make sure it was done.

Two things go wrong when AI is allowed to act:

• It does something you didn’t intend (a made-up tool, a guessed button coordinate).
• It says “done” when it isn’t (the ticket was never created, the refund was rejected).

The cure is a loop around the decision. Jev controls the forks. Code and agents still own execution, memory and verification. A Jev deployment is a loop around a typed decision, not a single model call.

💡 Mental Model: a courier with a receipt The decision says where the parcel goes. The executor is the courier: it delivers exactly one parcel and hands back a receipt. The verifier is you phoning the recipient to check the parcel arrived, rather than trusting the courier’s sticker that says “delivered”.

7.1 The control loop

The execution boundary. Commit these five lines to memory. They’re what make errors recoverable:

1. Jev decides among allowed options. A hallucinated tool can’t run because it was never in the candidate set.
2. Code validates schema, threshold, permission and budget.
3. The executor performs exactly one observable action and returns a receipt.
4. The verifier reads the external system again. A false “done” can’t close the loop.
5. The new observation becomes the next state. Never reuse a cached summary.

7.2 Capstone increment: executors with receipts

Each dispatch target maps to one executor. Executors are idempotent (keyed by report_id), do one thing, and return a receipt: what was attempted and where to look for proof. The handoff contract from the Codila playbook becomes a dataclass:

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

We build it in four steps: the receipt → the ports → one executor → the other executors.

Step 1: the receipt

Five fields from the table above, plus a timestamp:

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

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Mapping, Protocol

from triage.policy_engine import PolicyDecision

@dataclass(frozen=True, slots=True)
class Receipt:
    owner: str
    artifact_id: str
    acceptance: str
    evidence_uri: str
    escalation: str
    attempted_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

Step 2: the ports to the outside world

A port is a tiny interface describing what we need from a real system. In production these are real helpdesk, SecOps and ledger clients; in tests they are in-memory fakes. Every port has a write method and a read method, because the verifier needs to read back what the executor wrote.

# ---- Ports to the outside world (real clients in prod, in-memory fakes in tests) ----
class HelpdeskAPI(Protocol):
    def upsert_ticket(self, key: str, queue: str, priority: str, body: Mapping[str, Any]) -> str: ...
    def get_ticket(self, ticket_id: str) -> Mapping[str, Any] | None: ...


class SecOpsAPI(Protocol):
    def open_case(self, key: str, priority: str, evidence: Mapping[str, Any]) -> str: ...
    def get_case(self, case_id: str) -> Mapping[str, Any] | None: ...


class LedgerAPI(Protocol):
    def create_refund_request(self, key: str, needs_approval: bool, evidence: Mapping[str, Any]) -> str: ...
    def get_refund_request(self, refund_id: str) -> Mapping[str, Any] | None: ...

Then the interface every executor implements:

class Executor(Protocol):
    target: str

    def execute(self, report_id: str, state: Mapping[str, Any], d: PolicyDecision) -> Receipt: ...

💡 Mental Model: a wall socket A Protocol is a socket shape. The executor only cares that something with an upsert_ticket plug is attached. Whether it’s the real helpdesk or a test double doesn’t matter.

Step 3: one executor, end to end

The Tier-1 executor is the simplest. It creates (or updates) a ticket, keyed by triage:{report_id} so running it twice never creates a duplicate, and returns a receipt:

@dataclass(slots=True)
class Tier1Executor:
    helpdesk: HelpdeskAPI
    target: str = "tier1_support"

    def execute(self, report_id: str, state: Mapping[str, Any], d: PolicyDecision) -> Receipt:
        tid = self.helpdesk.upsert_ticket(f"triage:{report_id}", "tier1", d.priority, state)
        return Receipt("tier1_support", tid, "ticket exists in queue tier1 with matching priority",
                       f"helpdesk://tickets/{tid}", "unassigned after 4h -> manual_escalation")

• f"triage:{report_id}" is the idempotency key.
• acceptance is a sentence the verifier will check.
• escalation is the condition that hands control back to a human.

Step 4: the other three executors

SecOps opens a case:

@dataclass(slots=True)
class SecOpsExecutor:
    secops: SecOpsAPI
    target: str = "secops_agent"

    def execute(self, report_id: str, state: Mapping[str, Any], d: PolicyDecision) -> Receipt:
        cid = self.secops.open_case(f"triage:{report_id}", d.priority,
                                    {"signals": state.get("computed_signals"), "reasons": d.reasons})
        return Receipt("secops_agent", cid, "case status in {open, investigating} and priority set",
                       f"secops://cases/{cid}", "P1 not acknowledged in 15 min -> on-call page")

Refund creates a refund request and never computes an amount from model output:

@dataclass(slots=True)
class RefundExecutor:
    ledger: LedgerAPI
    target: str = "refund_pipeline"

    def execute(self, report_id: str, state: Mapping[str, Any], d: PolicyDecision) -> Receipt:
        # The executor NEVER computes an amount from model output; the ledger derives it from txn ids.
        rid = self.ledger.create_refund_request(f"triage:{report_id}", d.requires_human_approval,
                                                {"merchants": state.get("disputed_merchants")})
        accept = "status == pending_approval" if d.requires_human_approval else "status in {queued, processed}"
        return Receipt("refund_pipeline", rid, accept, f"ledger://refunds/{rid}",
                       "ledger rejects or 2 identical failures -> manual_escalation")

⚠️ Watch Out The refund executor passes transaction ids and merchants, never an amount. The ledger derives the amount from its own records. A model should never be the source of a number that moves money.

And manual escalation puts a case in front of a human:

@dataclass(slots=True)
class ManualEscalationExecutor:
    helpdesk: HelpdeskAPI
    target: str = "manual_escalation"

    def execute(self, report_id: str, state: Mapping[str, Any], d: PolicyDecision) -> Receipt:
        tid = self.helpdesk.upsert_ticket(f"triage:{report_id}", "escalations", d.priority,
                                          {"state": state, "reasons": d.reasons})
        return Receipt("manual_escalation", tid, "ticket exists in queue escalations",
                       f"helpdesk://tickets/{tid}", "n/a (already human)")

7.3 Capstone increment: the fresh-state verifier

Never trust a “done” label, whether it comes from a model or an executor. The verifier opens the artifact in the external system and checks the acceptance condition.

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

The three possible outcomes.

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

from dataclasses import dataclass
from typing import Literal

from triage.executors import HelpdeskAPI, LedgerAPI, Receipt, SecOpsAPI
from triage.policy_engine import PolicyDecision

Outcome = Literal["verified", "not_found", "mismatch"]

@dataclass(frozen=True, slots=True)
class Verification:
    outcome: Outcome
    detail: str

    @property
    def ok(self) -> bool:
        return self.outcome == "verified"

The verifier and its three ports.

@dataclass(slots=True)
class FreshStateVerifier:
    helpdesk: HelpdeskAPI
    secops: SecOpsAPI
    ledger: LedgerAPI

Branch 1: tickets (Tier-1 and manual escalation). Re-read the ticket, check it exists, then check queue and priority.

    def verify(self, receipt: Receipt, d: PolicyDecision) -> Verification:
        if receipt.owner in ("tier1_support", "manual_escalation"):
            t = self.helpdesk.get_ticket(receipt.artifact_id)
            if t is None:
                return Verification("not_found", "ticket missing")
            want_q = "tier1" if receipt.owner == "tier1_support" else "escalations"
            if t.get("queue") != want_q or t.get("priority") != d.priority:
                return Verification("mismatch", f"queue={t.get('queue')} priority={t.get('priority')}")
            return Verification("verified", f"ticket {receipt.artifact_id} in {want_q}")

Branch 2: SecOps cases. The case must exist and be open or investigating.

        if receipt.owner == "secops_agent":
            c = self.secops.get_case(receipt.artifact_id)
            if c is None:
                return Verification("not_found", "case missing")
            if c.get("status") not in {"open", "investigating"}:
                return Verification("mismatch", f"status={c.get('status')}")
            return Verification("verified", f"case {receipt.artifact_id} {c.get('status')}")

Branch 3: refunds. The allowed status depends on whether human approval was required. A refund that needed approval must be pending_approval; it must not be processed.

        if receipt.owner == "refund_pipeline":
            r = self.ledger.get_refund_request(receipt.artifact_id)
            if r is None:
                return Verification("not_found", "refund request missing")
            allowed = {"pending_approval"} if d.requires_human_approval else {"queued", "processed"}
            if r.get("status") not in allowed:
                return Verification("mismatch", f"status={r.get('status')} expected {sorted(allowed)}")
            return Verification("verified", f"refund {receipt.artifact_id} {r.get('status')}")
        return Verification("mismatch", f"unknown owner {receipt.owner}")

⚠️ Watch Out Each branch has the same three-step shape: does it exist? is it in the expected state? then verified. If you add a fifth executor, add a fourth branch here, or unknown owner will make every action look failed.

Recovery is part of the graph

Recovery depends on counts, budgets and permissions, so it lives in code:

7.4 Integrating with agent frameworks

LangChain

LangChain’s public integration ships two experimental middlewares: ModelRouterMiddleware (Jev picks which model tier handles a turn) and AutoModeMiddleware (Jev reviews a selected tool call before it runs). They’re experimental and their signatures are still moving, so check the current LangChain docs before wiring them. The pattern underneath is framework-agnostic, and you can use it today with any tool-calling agent:

💡 Mental Model: a bouncer at the door Before any tool runs, a bouncer asks two questions: “Does this serve what the user asked for?” and “How bad is the worst case?” If either answer is wrong, the tool doesn’t run.

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

Snippet 1: imports and type variables. ParamSpec lets the guard wrap any function and keep its signature.

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

import functools
import json
from typing import Any, Callable, Final, ParamSpec, TypeVar

from triage.jev_client import DecisionClient, DecisionError

P = ParamSpec("P")
R = TypeVar("R")

Snippet 2: the two guard questions. One Noul (does the call match the goal?) and one Score (how bad is the worst outcome?).

GUARD_QUESTIONS: Final = {
    "matches_user_goal": {
    "type": "noul",
    "instructions": "Does calling `tool` with `args` directly serve `user_goal`?"
    },
    "harm": {
        "type": "score",
        "instructions": "What is the worst outcome if `tool` runs with `args`?",
        "criteria": [
            "Nothing changes outside this conversation",
            "A reversible change to the user's own data",
            "Money moves, messages are sent, or data is deleted"
        ]
    },
}

Snippet 3: the failure signal.

class ToolCallBlocked(RuntimeError):
    pass

Snippet 4: the decorator and its state. It builds the evidence from the tool name, its arguments and the user’s goal.

def guarded(client: DecisionClient, tool_name: str, get_goal: Callable[[], str],
            max_harm: float = 1.2, min_match: float = 0.8) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Wrap any tool function (plain, LangChain @tool body, MCP handler) with a Jev pre-check."""
    def deco(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            state = {"tool": tool_name, "args": json.loads(json.dumps(kwargs, default=str)),
                     "user_goal": get_goal()}

Snippet 5: the single decision call, then the guardrail. Two ways to say no: the guard is down, or the answer is unsafe.

            try:
                r = client.decide(state, GUARD_QUESTIONS)
            except DecisionError as e:
                raise ToolCallBlocked(f"guard unavailable: {e}") from e
            harm, match = r.score("harm").score, r.noul("matches_user_goal").p_yes
            if harm > max_harm or match < min_match:
                raise ToolCallBlocked(f"{tool_name}: harm={harm:.2f} goal_match={match:.2f} -> needs approval")
            return fn(*args, **kwargs)
        return wrapper
    return deco

⚠️ Watch Out If the guard cannot be reached, the tool is blocked, not allowed. A guard that fails open is not a guard.

Coding agents: install the TypeSafe skill first

Agents trained on LLM APIs ask one question per call and invent request fields. Install the official skill before you ask an agent to integrate Jev:

# Claude Code
claude plugin marketplace add typesafe-ai/skills
claude plugin install typesafe@typesafe-ai
# Cursor, Codex, others (project-local; add -g only on purpose)
npx skills add typesafe-ai/skills --skill typesafe-ai

A good first prompt: “Using the TypeSafe skill, explore the project and find opportunities for using intelligent judgement to stand in for complex parsing or other fragile code.” Review the proposal before any code is written, and make the agent keep every question and threshold in one file.

Browser, computer-use and mobile agents: bound the action space

Browser Use, Stagehand, Cua and Mobile Jev all converge on the same shape. The driver turns the screen into a table of valid action IDs, and Jev picks one. The model never invents coordinates or selectors.

Published results show the whole loop matters more than the model:

The Retriever AI row is the one to remember. A cheap decision model doesn’t guarantee a cheap workflow when candidate preparation grows. Profile observation, decision, execution and verification separately.

7.5 Mini-project 6: two-stage re-ranker (BM25 → Jev Noul)

Background: what is BM25?

BM25 (“Best Matching 25”) is the standard keyword-ranking formula for search. It scores how well a document matches a query from the words they share. It’s the default ranking in Elasticsearch, OpenSearch, Lucene/Lucene.NET and Azure AI Search, so if you’ve used any of those from C#, you’ve already used BM25.

The idea in three rules. For each query word found in a document, BM25 adds points, and three things control how many:

1. Rare words count more (IDF). A match on “chargeback” says far more than a match on “the” or “payment”, which appear everywhere.
2. Repeats help, with diminishing returns (k1). Saying “refund” three times beats once, but the tenth mention adds almost nothing. That stops keyword-stuffed documents from winning.
3. Long documents are penalised a little (b). A 5,000-word policy will contain more words by chance than a 50-word clause. BM25 scales scores by document length relative to the average.

For each query term:

score += IDF(term) × tf × (k1 + 1) / (tf + k1 × (1 − b + b × docLength / avgDocLength))

The usual defaults are k1 = 1.5 and b = 0.75, the values in the book’s bm25_search.py. The whole implementation there is about 25 lines of plain Python.

What it’s good and bad at.

Why the book pairs it with Jev or Kev. That trade-off is the reason for this chapter’s two-stage design:

1. BM25 (stage 1): cheaply narrows the whole corpus to a shortlist (for example, 30 passages). It’s tuned for recall, so the right answer is somewhere in the list.
2. Jev or Kev (stage 2): asks one yes/no question per shortlisted passage, “does this state the rule that decides this case?”, and re-sorts by that probability. It’s tuned for precision, so the right answer moves to the top.

💡 Mental Model: librarian, then expert A librarian who knows only keywords pulls 30 books off the shelf in seconds (stage 1). A subject expert then skims those 30 and says which one actually answers your question (stage 2). Neither works well alone, but together they are fast and accurate.

Your new eval_cases.jsonl shows the gap. BM25 missed “someone used my card without my permission” because the right clause says “transactions the customer did not authorise”, and it shares almost no words with the query. Matching meaning like that is the job the decision model is there for. On TypeSafe’s legal-retrieval test, adding the re-ranking stage raised top-1 accuracy from 5 % to 18 %.

The alternative for stage 1 is embedding (vector) search, which matches by meaning but needs an embedding model and a vector index. Many production systems run both and merge the results (hybrid search). BM25 stays popular because it costs almost nothing and is very hard to break.

The two-stage idea

Stage 1 maximises recall cheaply and deterministically. Stage 2 asks one stable Noul per (query, passage) pair and sorts by probability. On 40 CLERC legal queries with 30-passage BM25 shortlists, TypeSafe reported:

In the capstone we use it to attach the right refund/chargeback policy clause to a case before the refund executor runs. That’s a legitimate second request, because it needs retrieved text the first call didn’t have.

Stage 1: bm25_search.py

📄 Full source: mini-projects/06-retrieval-reranker/bm25_search.py

Snippet 1: usage.

# mini-projects/06-retrieval-reranker/bm25_search.py
"""Stage 1: pure-Python BM25 over a small corpus. Fast, deterministic, recall-oriented.

  uv run python mini-projects/06-retrieval-reranker/bm25_search.py "money debited but UPI payment failed"
"""

Snippet 2: imports, a tokenizer pattern and the corpus path.

from __future__ import annotations

import argparse
import json
import math
import re
import sys
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence

_TOKEN = re.compile(r"[a-z0-9]+")
DEFAULT_CORPUS = Path(__file__).parent / "policy_clauses.json"
# DEFAULT_CORPUS = Path(__file__).with_name("policy_clauses.json")

Snippet 3: lower-case words only.

def tokenize(text: str) -> list[str]:
    return _TOKEN.findall(text.lower())

@dataclass(frozen=True, slots=True)
class Hit:
    doc_id: int
    score: float

Snippet 4: build the index. For each document store word counts, lengths and each word’s rarity (IDF):

class BM25:
    def __init__(self, docs: Sequence[str], k1: float = 1.5, b: float = 0.75) -> None:
        if not docs:
            raise ValueError("empty corpus")
        self.k1, self.b = k1, b
        self.tf = [Counter(tokenize(d)) for d in docs]
        self.len = [sum(c.values()) for c in self.tf]
        self.avg = sum(self.len) / len(self.len)
        df: Counter[str] = Counter()
        for c in self.tf:
            df.update(c.keys())
        n = len(docs)
        self.idf = {t: math.log(1 + (n - f + 0.5) / (f + 0.5)) for t, f in df.items()}

Snippet 5: score a query. For each document, add up the BM25 points for every query word it contains:

    def search(self, query: str, k: int = 30) -> list[Hit]:
        q = tokenize(query)
        scores: list[Hit] = []
        for i, tf in enumerate(self.tf):
            s = 0.0
            for t in q:
                f = tf.get(t, 0)
                if f:
                    s += self.idf[t] * f * (self.k1 + 1) / (f + self.k1 * (1 - self.b + self.b * self.len[i] / self.avg))
            if s > 0:
                scores.append(Hit(i, s))
        return sorted(scores, key=lambda h: h.score, reverse=True)[:k]

Snippet 6: load and prepare the corpus.

def load_corpus(path: Path = DEFAULT_CORPUS) -> list[dict[str, Any]]:
    """JSON list of {"id", "title", "text"}; the searchable document is title + text."""
    rows = json.loads(path.read_text(encoding="utf-8"))
    if not rows or not all({"id", "text"} <= r.keys() for r in rows):
        raise ValueError(f"{path}: expected a non-empty list of objects with 'id' and 'text'")
    return rows

def as_documents(rows: Sequence[dict[str, Any]]) -> list[str]:
    return [f"{r.get('title', '')}. {r['text']}" for r in rows]

Snippet 7: the command line.

def main(argv: Sequence[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="BM25 search over a JSON corpus")
    ap.add_argument("query")
    ap.add_argument("--corpus", type=Path, default=DEFAULT_CORPUS)
    ap.add_argument("-k", type=int, default=5)
    args = ap.parse_args(argv)
    corpus_path = Path(__file__).parent / args.corpus
    rows = load_corpus(corpus_path)
    hits = BM25(as_documents(rows)).search(args.query, k=args.k)

    if not hits:
        print("no matching documents")
        return 1
    for rank, h in enumerate(hits, 1):
        print(f"{rank:2}. {rows[h.doc_id]['id']}  bm25={h.score:5.2f}  {rows[h.doc_id].get('title', '')}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

Stage 1 on its own, against the sample corpus of 12 refund and chargeback clauses (policy_clauses.json):

uv run python mini-projects/06-retrieval-reranker/bm25_search.py "money debited but UPI payment failed" -k 3

 1. R02  bm25= 9.90  Failed but debited UPI payments
 2. R01  bm25= 1.57  Duplicate charges
 3. R07  bm25= 1.54  Refund timelines

Stage 2: jev_reranker.py

📄 Full source: mini-projects/06-retrieval-reranker/jev_reranker.py

Snippet 1: usage.

# mini-projects/06-retrieval-reranker/jev_reranker.py
"""Stage 2: re-rank a BM25 shortlist with one Noul per (case, passage) pair, in one call.

  uv run python mini-projects/06-retrieval-reranker/jev_reranker.py "UPI payment failed but money debited"
  uv run python mini-projects/06-retrieval-reranker/jev_reranker.py --eval             # top-k vs BM25 alone
  uv run python mini-projects/06-retrieval-reranker/jev_reranker.py --eval --offline   # no API key needed
"""

Snippet 2: imports. Stage 1 is imported, not copied.

from __future__ import annotations

import argparse
import json
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence

from bm25_search import BM25, as_documents, load_corpus, tokenize
from triage.jev_client import Answer, DecisionClient, DecisionError, FakeDecisionClient, NoulResult, client_from_env

DEFAULT_EVAL = Path(__file__).parent / "eval_cases.jsonl"
# DEFAULT_EVAL = Path(__file__).with_name("eval_cases.jsonl")

Snippet 3: the input and output shape. A ranked result holds the passage, Jev’s probability, and the BM25 score for tie-breaking.

@dataclass(frozen=True, slots=True)
class Ranked:
    doc_id: int
    p_relevant: float
    bm25: float

Snippet 4: stage 1 feeds stage 2. Take the shortlist, and truncate each passage so one long clause can’t blow the token budget.

def rerank(client: DecisionClient, query: str, docs: Sequence[str], index: BM25,
           shortlist: int = 30, min_p: float = 0.5) -> tuple[list[Ranked], float]:
    hits = index.search(query, k=shortlist)
    if not hits:
        return [], 0.0
    passages = [docs[h.doc_id][:1_500] for h in hits]

Snippet 5: the single decision call, one Noul per passage. Notice the failure path: if stage 2 is down, return the stage-1 order instead of nothing.

    qs = {f"p{i}": {"type": "noul",
                    "instructions": f"Does `passages[{i}]` state the rule that decides `case`?"}
          for i in range(len(passages))}
    t0 = time.perf_counter()
    try:
        r = client.decide({"case": query, "passages": passages}, qs)
    except DecisionError:
        # Degrade to stage-1 order; never return nothing because stage 2 failed.
        return [Ranked(h.doc_id, 0.0, h.score) for h in hits], 0.0
    ms = (time.perf_counter() - t0) * 1000

Snippet 6: branch on the typed output. Sort by probability, break ties by BM25, and drop passages below min_p, but always keep at least one.

    ranked = sorted((Ranked(h.doc_id, r.noul(f"p{i}").p_yes, h.score) for i, h in enumerate(hits)),
                    key=lambda x: (x.p_relevant, x.bm25), reverse=True)
    return [x for x in ranked if x.p_relevant >= min_p] or ranked[:1], ms

⚠️ Watch Out or ranked[:1] means a case with no confident passage still gets its best guess rather than nothing, and min_p filtering is only a view. The decision about whether the clause is trustworthy belongs to the policy layer, not here.

Snippet 7: measuring accuracy.

def top_k_accuracy(results: Sequence[Sequence[int]], gold: Sequence[int], k: int) -> float:
    return sum(g in r[:k] for r, g in zip(results, gold, strict=True)) / len(gold)

Snippet 8: an offline stand-in that rewards passages whose title words appear in the case. It is not a model.

# ---- Offline stand-in: rewards passages whose TITLE terms appear in the case. Not a model. ----
_STOP = {"the", "a", "an", "and", "or", "of", "to", "my", "i", "is", "it", "for", "on", "in", "was", "but"}

def offline_model(state: Any, questions: Mapping[str, Any]) -> Mapping[str, Answer]:
    case = {w for w in tokenize(state["case"]) if w not in _STOP}
    out: dict[str, Answer] = {}
    for qid in questions:
        passage = state["passages"][int(qid[1:])]
        title = {w for w in tokenize(passage.split(".", 1)[0]) if w not in _STOP}
        body = {w for w in tokenize(passage) if w not in _STOP}
        score = 0.6 * len(case & title) / max(1, len(title)) + 0.4 * len(case & body) / max(1, len(case))
        out[qid] = NoulResult(round(min(0.97, 0.05 + score), 3))
    return out

Snippet 9: the evaluation loop. For each labeled case, run BM25 alone and BM25 plus re-rank, and print the first-place result for each.

def _evaluate(client: DecisionClient, rows: list[dict[str, Any]], cases: list[dict[str, str]],
              shortlist: int) -> int:
    docs = as_documents(rows)
    index = BM25(docs)
    ids = [r["id"] for r in rows]
    bm25_runs, jev_runs, gold, total_ms = [], [], [], 0.0
    for c in cases:
        bm25_runs.append([h.doc_id for h in index.search(c["case"], k=shortlist)])
        ranked, ms = rerank(client, c["case"], docs, index, shortlist=shortlist, min_p=0.0)
        jev_runs.append([x.doc_id for x in ranked])
        gold.append(ids.index(c["gold"]))
        total_ms += ms

        top_b = ids[bm25_runs[-1][0]] if bm25_runs[-1] else "-"
        top_j = ids[jev_runs[-1][0]] if jev_runs[-1] else "-"
        mark = "ok " if top_j == c["gold"] else "MISS"
        print(f"{mark} gold={c['gold']}  bm25@1={top_b}  rerank@1={top_j}  {c['case'][:60]}")
    print(f"\n{'':10}{'Top-1':>8}{'Top-3':>8}")
    for name, runs in (("BM25", bm25_runs), ("re-ranked", jev_runs)):
        print(f"{name:10}{top_k_accuracy(runs, gold, 1):8.0%}{top_k_accuracy(runs, gold, 3):8.0%}")
    print(f"\nstage-2 decision time: {total_ms:.0f} ms over {len(cases)} cases")
    return 0

Snippet 10: the command line and main flow.

def main(argv: Sequence[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Two-stage retrieval: BM25 shortlist -> System One re-rank")
    ap.add_argument("query", nargs="?", help="a case to find the deciding clause for")
    default_corpus_path = Path(__file__).parent / "policy_clauses.json"
    ap.add_argument("--corpus", type=Path, default=default_corpus_path)
    ap.add_argument("--eval", nargs="?", const=DEFAULT_EVAL, type=Path,
                    help="JSONL of {case, gold} to compare BM25 vs re-ranked accuracy")
    ap.add_argument("--shortlist", type=int, default=8)
    ap.add_argument("--min-p", type=float, default=0.5)
    ap.add_argument("--offline", action="store_true", help="use the keyword stand-in instead of Jev/Kev")
    args = ap.parse_args(argv)
    if not args.query and not args.eval:
        ap.error("give a query or --eval")

    corpus_path = Path(__file__).parent / args.corpus
    rows = load_corpus(corpus_path)
    client: DecisionClient = FakeDecisionClient(offline_model) if args.offline else client_from_env()
    if args.eval:
        cases = [json.loads(l) for l in args.eval.read_text(encoding="utf-8").splitlines() if l.strip()]
        return _evaluate(client, rows, cases, args.shortlist)

    docs = as_documents(rows)
    ranked, ms = rerank(client, args.query, docs, BM25(docs), shortlist=args.shortlist, min_p=args.min_p)
    if not ranked:
        print("no candidate passages")
        return 1
    for rank, x in enumerate(ranked, 1):
        print(f"{rank:2}. {rows[x.doc_id]['id']}  p={x.p_relevant:.2f}  bm25={x.bm25:5.2f}  {rows[x.doc_id]['title']}")
    print(f"\nstage-2 decision time: {ms:.0f} ms ({client.backend})")
    return 0

if __name__ == "__main__":
    sys.exit(main())

Run it. Pass a case to get the ranked clauses, or --eval to compare BM25 alone against BM25 plus re-ranking on eval_cases.jsonl (12 customer cases, each labelled with its deciding clause). Most cases are deliberately paraphrased, so they don’t share words with their clause:

uv run python mini-projects/06-retrieval-reranker/jev_reranker.py "Someone used my card without my permission"
uv run python mini-projects/06-retrieval-reranker/jev_reranker.py --eval            # Jev / Kev via DECISION_BACKEND
uv run python mini-projects/06-retrieval-reranker/jev_reranker.py --eval --offline  # no key: plumbing check only

For: uv run python mini-projects/06-retrieval-reranker/jev_reranker.py --eval 

ok  gold=R01  bm25@1=R01  rerank@1=R01  I was billed two times for one order, please return the extr
ok  gold=R02  bm25@1=R02  rerank@1=R02  My UPI payment failed but money got debited from my bank acc
MISS gold=R03  bm25@1=R01  rerank@1=R01  Someone used my card without my permission last night.
ok  gold=R04  bm25@1=R04  rerank@1=R04  The courier says the parcel is lost and nothing ever arrived
MISS gold=R05  bm25@1=R02  rerank@1=R02  I stopped my plan an hour after it auto-renewed, can I get m
MISS gold=R06  bm25@1=R11  rerank@1=R04  Two of the five things I ordered were not in the box.
ok  gold=R07  bm25@1=R07  rerank@1=R07  How long until the money shows up back on my credit card?
ok  gold=R08  bm25@1=R08  rerank@1=R08  I need 75,000 rupees back for a returned laptop.
ok  gold=R09  bm25@1=R09  rerank@1=R09  Can I still dispute a card transaction from four months ago 
ok  gold=R10  bm25@1=R10  rerank@1=R10  I paid the delivery person in cash and sent the item back, h
MISS gold=R11  bm25@1=R02  rerank@1=R02  I was charged in dollars on an overseas site and want it rev
ok  gold=R12  bm25@1=R12  rerank@1=R12  Can I swap my unused gift card for cash?

             Top-1   Top-3
BM25           67%     83%
re-ranked      67%     75%

stage-2 decision time: 1 ms over 12 cases (offline stand-in)

For uv run python mini-projects/06-retrieval-reranker/jev_reranker.py "Someone used my card without my permission"
 1. R03  p=0.85  bm25= 1.23  Unauthorised transactions

stage-2 decision time: 560 ms (jev)

Read the misses, not the totals. “Someone used my card without my permission” never says “did not authorise”, “stopped my plan” never says “subscription”, and “charged in dollars on an overseas site” never says “foreign currency”. BM25 can’t bridge those gaps, and neither can the offline stand-in, which is word matching too. Those are exactly the cases a semantic judge exists for, so run --eval against Jev or Kev and compare. If re-ranking doesn’t win on your labelled cases, keep BM25 alone.

One call or thirty? Put all 30 pairs in one request when passages are short: the case text is sent once. With long passages, split into 3 × 10 so each request stays well under the 32k-token state + longest question limit. Measure both with the Chapter 2 harness.

The same “rank wide, read narrow” pattern powers TypeSafe’s Hermes skill router (182 skills). Wrong skill loads fell 16.8 % → 7.3 % and needless loads 9.8 % → 4.0 % across 488 requests, with 7 previously-correct picks newly broken. Keep a no_skill / none option and treat the output as a shortlist, not a command.

7.6 Checklist

• Every executor performs one action, is idempotent, and returns a receipt with acceptance + evidence
• The verifier reads the external system; model “done” is a hypothesis
• Recovery transitions are code: counts, budgets, timeouts
• Agent tool calls pass through a guard; action menus are rebuilt after every step
• Retrieval is two-stage: deterministic recall, then Jev precision, then verification
• Each loop stage is profiled separately (observe / decide / execute / verify)

What’s next

We’ve built the whole loop on Jev. Is Jev the right engine for your workload, region and privacy constraints? Chapter 8 benchmarks it against the alternatives (a self-hosted open model, a commercial competitor, general LLMs with structured output, the adapter, and a classical classifier) on the same contract and the same labeled set.