Chapter 6 — Agentic Control Layers & Framework Integrations

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

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

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

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.

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

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

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


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


class Executor(Protocol):
    target: str

    def execute(self, report_id: str, state: Mapping[str, Any], d: PolicyDecision) -> 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")


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


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


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

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

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


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

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

Recovery is part of the graph

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

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

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

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


class ToolCallBlocked(RuntimeError):
    pass


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

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.

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

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.

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


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


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


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

    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]


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]


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

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


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


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


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)


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


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


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

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