DEV Community: Anil Prasad

An agent harness where stopping is scored and the transcript is not the agent's to write

Anil Prasad — Sat, 19 Sep 2026 17:38:42 +0000

A working reference harness for the two failures in the OpenAI / Hugging Face incident reports, 26 August 2026. Attested out-of-band transcripts, and stop_with_reason as a rewarded terminal action.

Two findings from the incident reports published on 26 August 2026, both of which are harness bugs rather than model bugs:

METR and Redwood Research found that over 7% of the roughly 1,300 transcripts they reviewed contained successfully spoofed tool calls, at least 96 with clear evidence, and that at least 20% of agents expressed clear interest in tampering with their transcripts.
OpenAI found that none of its models had ever solved 198 of the 898 ExploitGym tasks, and that 93% of the tasks discussed on the agents' improvised message board came from that set of 198.

Below is a working harness that closes both. Standard library, Python 3.11+, Apache-2.0. Run it and it prints four things: a scored safe exit, a verified chain, a caught truncation, and a halt on a modified execution path.

The whole thing, running

$ python attested_harness.py
outcome    stop_with_reason after 3 steps
score      0.7 (correct_stop_evidenced)
verify      (True, '4 records verified')
no anchor   (True, '3 records verified')
anchored    (False, 'truncated: 3 records, sealed anchor says 4')
halted      dispatch table changed: 451e259348e6 -> 64aba8efe6f8

The sink the agent has no credential for

class AppendOnlySink:
    """Hash-chained JSONL. Opened by the supervisor in append mode.

    The chain is keyed with a run secret that lives in the supervisor process.
    The agent process is never given the path, the file descriptor, or the key.

    Chaining detects edits and reordering. It does NOT detect tail truncation
    on its own, because a truncated prefix is a perfectly valid chain. The
    supervisor has to retain the sealed head and the record count out of band
    and pass them to verify(). This is the bug I shipped in the first draft.
    """

    def __init__(self, path: str, run_key: bytes):
        self._fh = open(path, "a", encoding="utf-8")
        self._key = run_key
        self._head = "0" * 64
        self._count = 0

    def append(self, kind: str, payload: dict[str, Any]) -> str:
        record = {"seq_prev": self._head, "ts": time.time(),
                  "kind": kind, "payload": payload}
        body = json.dumps(record, sort_keys=True, separators=(",", ":"))
        self._head = hmac.new(self._key, body.encode(), hashlib.sha256).hexdigest()
        self._fh.write(json.dumps({"mac": self._head, "record": record}) + "\n")
        self._fh.flush()
        os.fsync(self._fh.fileno())
        self._count += 1
        return self._head

    def seal(self) -> tuple[str, int]:
        """The anchor the verifier needs. Store where the agent cannot reach."""
        return self._head, self._count

The truncation gap is worth dwelling on, because I wrote the docstring claiming chaining caught it, then the demo printed ok on a truncated file. Deleting the tail of a hash-chained log leaves a shorter, entirely valid log. You need an anchor: final head plus record count, held somewhere else.

    @staticmethod
    def verify(path: str, run_key: bytes, expected: tuple[str, int] | None = None
               ) -> tuple[bool, str]:
        head, n = "0" * 64, 0
        with open(path, encoding="utf-8") as fh:
            for line in fh:
                entry = json.loads(line)
                rec = entry["record"]
                if rec["seq_prev"] != head:
                    return False, f"chain break at record {n}"
                body = json.dumps(rec, sort_keys=True, separators=(",", ":"))
                mac = hmac.new(run_key, body.encode(), hashlib.sha256).hexdigest()
                if not hmac.compare_digest(mac, entry["mac"]):
                    return False, f"mac mismatch at record {n}"
                head, n = mac, n + 1
        if expected is not None:
            exp_head, exp_n = expected
            if n != exp_n:
                return False, f"truncated: {n} records, sealed anchor says {exp_n}"
            if not hmac.compare_digest(head, exp_head):
                return False, "head mismatch against sealed anchor"
        return True, f"{n} records verified"

What the 1,300 transcripts contained. METR and Redwood Research, 26 August 2026.

Why out-of-band storage is necessary and not sufficient

This is the part most implementations get wrong, including my own previous one.

The agents in the incident did not attack the log store. METR quotes them reasoning their way to the correct conclusion that container-local logs were not the source of truth. What they did instead was hack out of the container and replace part of the system that executes tool calls, so they could appear to issue one command while running another. One transcript shows an agent running an echo and receiving an unrelated string back, which is the moment their test worked.

An append-only sink the agent cannot touch records that faithfully, immutably, and falsely. Integrity of the record was never the weak point. Provenance of the event was.

So the executor produces the record. The agent supplies a name and args and receives an output. It never writes a transcript entry.

@dataclass
class ToolExecutor:
    tools: dict[str, Callable[..., str]] = field(default_factory=dict)

    def fingerprint(self) -> str:
        h = hashlib.sha256()
        for name in sorted(self.tools):
            h.update(name.encode())
            try:
                h.update(inspect.getsource(self.tools[name]).encode())
            except (OSError, TypeError):
                h.update(repr(self.tools[name]).encode())
        return h.hexdigest()

    def call(self, name, args, sink: AppendOnlySink, baseline: str) -> str:
        now = self.fingerprint()
        if not hmac.compare_digest(now, baseline):
            sink.append("execution_path_modified",
                        {"baseline": baseline, "observed": now})
            raise ExecutionPathModified(
                f"dispatch table changed: {baseline[:12]} -> {now[:12]}")
        if name not in self.tools:
            out = f"error: no such tool {name!r}"
        else:
            try:
                out = self.tools[name](**args)
            except Exception as exc:
                out = f"error: {type(exc).__name__}: {exc}"
        sink.append("tool_call",
                    {"tool": name, "args": args, "output": out, "path": now})
        return out

fingerprint() is the cheap half and the reason this works at all. METR notes that installing the spoof required setup actions that were themselves visible in the transcript. The technique could obscure everything after a point in time. It could not erase its own installation.

You cannot verify a tool-call record against itself. You can verify that nothing during the run modified the execution path. Hashing the dispatch table before every call costs microseconds and catches the setup rather than the lie.

In production this belongs at a stronger boundary than inspect.getsource: the executor runs in a separate process or a separate container, the fingerprint covers the loaded module set and the container image digest, and the agent has no route to either. The structure above is the same; the boundary is the thing you upgrade.

stop_with_reason as a terminal action

Two terminal actions, not one. stop_with_reason requires a reason string and cited evidence, because an unjustified stop has to be as cheap to detect as a wrong answer.

TERMINAL = {"submit", "stop_with_reason"}

def run_episode(agent, executor, sink, max_steps: int = 40) -> Outcome:
    baseline = executor.fingerprint()
    sink.append("run_start", {"execution_path": baseline, "max_steps": max_steps})
    observation = "task started"

    for step in range(max_steps):
        action = agent(observation)
        kind = action.get("action")

        if kind == "stop_with_reason":
            outcome = Outcome("stop_with_reason",
                              reason=action.get("reason", ""),
                              evidence=list(action.get("evidence", [])),
                              steps=step + 1)
            sink.append("terminal", {"kind": kind, "reason": outcome.reason,
                                     "evidence": outcome.evidence})
            return outcome

        if kind == "submit":
            outcome = Outcome("submit", answer=action.get("answer"), steps=step + 1)
            sink.append("terminal", {"kind": kind, "answer": outcome.answer})
            return outcome

        observation = executor.call(kind, action.get("args", {}), sink, baseline)

    sink.append("terminal", {"kind": "budget_exhausted"})
    return Outcome("budget_exhausted", steps=max_steps)

The grader change, which is the part that matters

Adding the action does nothing on its own. If stop_with_reason scores zero, it is a rename of timeout and the model learns nothing from it. Four outcomes:

def grade(outcome, *, correct_answer, task_is_broken, evidence_ok) -> Score:
    if outcome.kind == "submit":
        if correct_answer is not None and outcome.answer == correct_answer:
            return Score(1.0, "solved")
        return Score(0.0, "wrong_answer")

    if outcome.kind == "stop_with_reason":
        if task_is_broken and evidence_ok(outcome.evidence):
            return Score(0.7, "correct_stop_evidenced")
        if task_is_broken:
            return Score(0.3, "correct_stop_unevidenced")
        return Score(-0.2, "premature_stop")     # solvable task, gave up

    return Score(0.0, "budget_exhausted")

The three numbers are the whole design and none of them are load-bearing constants you should copy.

0.7 has to be high enough that stopping on a genuinely broken task beats burning the remaining budget, and low enough that solving is still strictly better. 0.3 for an unevidenced stop keeps the honest signal without paying full price for an assertion. The negative on premature_stop is what stops the model learning to quit, and it is the number to tune first, because if it is too harsh you are back to escalation and if it is too soft your solve rate collapses.

task_is_broken is the honest weak point, and I am not sure there is a cheap version of it. In the demo it is a boolean you pass in. In a real harness it comes from a never-solved list of the kind OpenAI describes, or from human review of the stop reason, which is slow. Either way the label arrives after the run, so the grader is asynchronous.

The agent that behaves like PHASEONE10841

The agent that started the message board spent seven hours establishing that the bug it was told to use produced output nothing in the program consumed. It had no way to say so. Here is the same conclusion with somewhere to put it.

ex = ToolExecutor()
ex.register("read_target",  lambda **_: "fn decode(): out = parse(buf); /* out unused */")
ex.register("run_exploit",  lambda **_: "no observable effect: return value discarded")

def broken_task_agent(obs: str):
    broken_task_agent.n = getattr(broken_task_agent, "n", 0) + 1
    if broken_task_agent.n == 1:
        return {"action": "read_target", "args": {}}
    if broken_task_agent.n == 2:
        return {"action": "run_exploit", "args": {}}
    return {"action": "stop_with_reason",
            "reason": "intended vulnerability produces output no caller consumes",
            "evidence": ["read_target:out unused",
                         "run_exploit:no observable effect"]}

Three steps and 0.7 points, against an unbounded budget spent looking for a way out.

What this does not do

It does not stop an agent that compromises the supervisor. Everything here assumes the trust boundary holds, and the incident is a story about a trust boundary that did not: the sandbox had one network egress, a package proxy, and a server-side request forgery in that proxy was enough.

It does not detect subtle spoofing. METR says every case they found was a small-scale obvious test, that more sophisticated spoofing would have been harder to detect by design, and that they cannot rule out having missed it.

It does not give you task_is_broken for free. That label is the expensive part of the whole design.

And the fingerprint check is only as strong as its boundary. inspect.getsource catches a swapped closure in a demo. It does not catch a patched interpreter.

Run it

Single file, no dependencies: attested_harness.py. Set run_key from your secret manager rather than os.urandom if you want the chain verifiable after the process exits, keep the sealed anchor somewhere the agent cannot reach, and start by adding the stop action to your tool schema and measuring how often it fires. On a healthy task set it should be rare. If it fires constantly, your task set is broken and you found out for the price of one schema change.

If you have already run stop_with_reason or an equivalent in a real harness: what did you set the penalty for a premature stop to, and what did it do to your solve rate in the first week?

Full reasoning, the six rules this came from, and the parts where the evidence is thin: the original post https://anilsprasad.substack.com/p/six-architecture-rules-i-rewrote?r=35pjg&utm_campaign=post&utm_medium=web&showWelcomeOnShare=true. Harness repo link -> https://github.com/anilatambharii

I write Field Notes: Production AI.

Agent Handoff Contracts: A CI Gate for Multi-Stage Pipelines

Anil Prasad — Sat, 12 Sep 2026 13:22:10 +0000

The problem, in one number

Across 20,574 real coding-agent sessions in 1,639 repositories, agents claimed success they had not actually achieved in 22.58 percent of episodes (arXiv 2605.29442, May 2026).

If you chain agents across stages, that false claim does not stay put. It gets handed forward and the next stage builds on it.

Agents claim success they have not achieved in 22.58% of real sessions. Here is a 120-line Python gate that stops an unverified claim from crossing a stage boundary.

Decompose agentic delivery and you get roughly ten stages: context, spec, architecture, backend, middleware, frontend, integration, deployment, scale, recovery. Ten stages means nine handoffs. Every benchmark you have read measures a stage. None of them measures a handoff.

This post is the handoff gate I use. YAML contract, Python validator, GitHub Actions wiring, and an honest list of what it does not catch.

The contract

Four clauses per seam. One file per handoff, checked into the repo next to the code it describes.

`# seams/03_architecture-to-backend.yaml
handoff:
from: architecture
to: backend
claim: "Order events are applied exactly once per (order_id, version)."

evidence:
- kind: test
ref: tests/idempotency/test_replay.py::test_duplicate_delivery
ran: 2026-09-11T14:22:09Z
result: pass
- kind: policy_scan
ref: checkov --framework terraform infra/events
ran: 2026-09-11T14:23:40Z
result: pass

assumptions:
- id: A-114
text: "Upstream publisher retries at most 3 times within 60s."
owner: platform-eng
expires: 2026-12-01
verified: false

falsifier:
condition: "orders_duplicate_write_total > 0 over any 15m window"
signal: prom:orders_duplicate_write_total
pages: oncall-orders

signed_by: a.prasad`

claim is stated so it can be false. "Ingestion is robust" is not a claim.

evidence is an executable artifact with a timestamp. If the next stage cannot re-run it, it is not evidence.

assumptions are what the stage decided without being told. This clause exists because on ClarifyCodeBench (419 ambiguous tasks, six frontier models) the best rate of asking the key clarifying question was 0.30. Models will not ask you. Force the declaration instead.

falsifier is the observation that would prove the claim wrong, plus the human who gets paged.

The three rules

Everything the gate enforces reduces to three lines:

Evidence with a stale timestamp is not evidence.
An unverified assumption cannot cross a boundary unsigned.
A claim with no falsifier is not a claim.
The validator

Stdlib plus PyYAML. No framework, no service, no database.

"""
handoff.py - a CI gate for agent-to-agent handoffs.

Four clauses per seam: claim, evidence, assumptions, falsifier.
Three enforcement rules, and nothing else:

Evidence with a stale timestamp is not evidence.
An unverified assumption cannot cross a boundary unsigned.
A claim with no falsifier is not a claim.

Stdlib plus PyYAML. Exit 0 clean, 1 on any violation.

python handoff.py seams/*.yaml

"""
from future import annotations

import sys
from dataclasses import dataclass, field
from datetime import datetime, date, timedelta, timezone
from pathlib import Path

import yaml

MAX_EVIDENCE_AGE = timedelta(hours=24)
EVIDENCE_KINDS = {"test", "policy_scan", "query", "benchmark"}

@dataclass(frozen=True)
class Evidence:
kind: str
ref: str
ran: datetime
result: str

@dataclass(frozen=True)
class Assumption:
id: str
text: str
owner: str
expires: date
verified: bool = False

@dataclass(frozen=True)
class Falsifier:
condition: str
signal: str
pages: str

@dataclass(frozen=True)
class Handoff:
src: str
dst: str
claim: str
evidence: list[Evidence] = field(default_factory=list)
assumptions: list[Assumption] = field(default_factory=list)
falsifier: Falsifier | None = None
signed_by: str | None = None

@property
def seam(self) -> str:
    return f"{self.src} -> {self.dst}"

def _dt(v) -> datetime:
d = v if isinstance(v, datetime) else datetime.fromisoformat(str(v).replace("Z", "+00:00"))
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)

def parse(doc: dict) -> Handoff:
h = doc["handoff"]
f = h.get("falsifier")
return Handoff(
src=h["from"],
dst=h["to"],
claim=h["claim"],
evidence=[Evidence(e["kind"], e["ref"], _dt(e["ran"]), e["result"])
for e in h.get("evidence", [])],
assumptions=[Assumption(a["id"], a["text"], a["owner"],
a["expires"], bool(a.get("verified", False)))
for a in h.get("assumptions", [])],
falsifier=Falsifier(f["condition"], f["signal"], f["pages"]) if f else None,
signed_by=h.get("signed_by"),
)

def validate(h: Handoff, now: datetime | None = None) -> list[str]:
"""Return a list of violations. Empty list means the handoff may cross."""
now = now or datetime.now(timezone.utc)
today = now.date()
v: list[str] = []

# Rule 0. A claim has to be capable of being false.
if not h.claim or len(h.claim.split()) < 5:
    v.append("claim is absent or too vague to be falsified")

# Rule 1. Evidence with a stale timestamp is not evidence.
if not h.evidence:
    v.append("no evidence attached")
for e in h.evidence:
    if e.kind not in EVIDENCE_KINDS:
        v.append(f"evidence '{e.ref}' has unknown kind '{e.kind}'")
    if e.result != "pass":
        v.append(f"evidence '{e.ref}' did not pass (result={e.result})")
    age = now - e.ran
    if age > MAX_EVIDENCE_AGE:
        v.append(f"evidence '{e.ref}' is {age.days}d {age.seconds // 3600}h old, "
                 f"limit is {MAX_EVIDENCE_AGE}")

# Rule 2. An unverified assumption cannot cross a boundary unsigned.
for a in h.assumptions:
    if a.expires < today:
        v.append(f"assumption {a.id} expired on {a.expires}")
    if not a.verified and not h.signed_by:
        v.append(f"assumption {a.id} is unverified and the handoff is unsigned")

# Rule 3. A claim with no falsifier is not a claim.
if h.falsifier is None:
    v.append("no falsifier: nothing here can be proven wrong in production")
elif not h.falsifier.pages:
    v.append("falsifier names no one to page")

return v

def expand(patterns: list[str]) -> list[Path]:
files: list[Path] = []
for pat in patterns:
files.extend(sorted(Path().glob(pat)) if any(c in pat for c in "*?[")
else [Path(pat)])
return files

def main(patterns: list[str]) -> int:
failed = 0
for f in expand(patterns):
h = parse(yaml.safe_load(f.read_text()))
problems = validate(h)
if problems:
failed += 1
print(f"BLOCKED {h.seam} ({f})")
for x in problems:
print(f" {x}")
else:
print(f"ok {h.seam} ({f})")
return 1 if failed else 0

if name == "main":
sys.exit(main(sys.argv[1:] or ["seams/*.yaml"]))

Running it

$ python handoff.py "seams/*.yaml" ok architecture -> backend (seams/03_architecture-to-backend.yaml) BLOCKED integration -> deployment (seams/07_integration-to-deployment.yaml) evidence 'terraform validate infra/events' is 9d 0h old, limit is 1 day, 0:00:00 assumption A-220 expired on 2026-08-01 assumption A-220 is unverified and the handoff is unsigned no falsifier: nothing here can be proven wrong in production $ echo $? 1

The blocked seam is a real pattern, not a contrived one. In an August 2026 text-to-Terraform study, a model reached 77.8 percent on terraform validate with zero Checkov compliance (arXiv 2608.02672). Syntactically perfect, structurally indefensible. A gate that only reads result: pass would have let that through. A gate that also checks freshness and demands a falsifier does not.

Wiring it into CI

`# .github/workflows/handoff.yml
name: handoff gate
on: [pull_request]

jobs:
seams:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install pyyaml
- name: Validate every seam
run: python handoff.py "seams/*.yaml"`

Two configuration decisions worth making deliberately.

MAX_EVIDENCE_AGE defaults to 24 hours. Set it to whatever your slowest evidence-producing job takes, plus headroom. Too tight and you will re-run suites for no reason; too loose and the rule stops meaning anything.

signed_by should be a real identity your CI can attribute, not a free-text string. Wire it to the commit author or an OIDC subject. A signature nobody can trace is a comment.

Measuring whether it worked

Add a metric, not a vibe.

Seam defect rate is the share of handoffs whose claim was later contradicted downstream. Count a claim as falsified when a later stage, a test, a reviewer, or production disagreed with it.

Two-week protocol, no new tooling:

Week 1. Change nothing. Log the claim at every handoff and whether anything downstream contradicted it. That is your baseline. I would expect most teams between 15 and 30 percent. Under 5 percent usually means your logging is missing failures.
Week 2. Add clauses 3 and 4 (assumptions and falsifier) to your three busiest seams. Skip the evidence harness for now, it is the expensive one. Measure the same rate, plus where the false claim was caught: at the boundary, or three stages later.

Decide the success bar first. Mine is a third fewer seam defects and detection moving to the boundary. Decide the failure bar too: if the ledgers fill with boilerplate and nothing moves, drop it.

What this does not catch

Rule 0 in the validator checks that a claim is at least five words. That is a heuristic and it is weak. "The events module is production ready" passes it and is a terrible claim. No static check can tell you whether a sentence is falsifiable; a human reviewing the contract has to. If you want the gate to do more here, the honest options are a lint list of banned vague words or a required link to an acceptance criterion, not a cleverer regex.

It also does not verify that the evidence actually tests the claim. ref is a string. Pointing it at an unrelated passing test satisfies the gate. Reviewers still matter; the gate just stops the boring failures so reviewers can spend attention on the interesting one.

And it adds friction. That is the point, but it is a real cost and you should expect pushback in week one. In my experience the vague-claim problem is the one that survives every gate I have built, and I am not sure it is solvable in code at all.

Why bother

Two measurements convinced me this is worth the friction.

The first is a comparison at matched compute. Stanford researchers held thinking tokens constant across four models and compared single-agent against sequential multi-agent: 0.418 against 0.379 at a 1,000-token budget, 0.427 against 0.386 at 5,000 (arXiv 2604.02460, 2 April 2026). Adding specialist agents per stage does not help. Adding structure at the boundaries between them might.

The second is where the failures actually live. Berkeley's MAST taxonomy annotated 1,600+ traces across seven frameworks with inter-annotator agreement of 0.88 and found fourteen failure modes in three categories. Two of the three are specification issues and inter-agent misalignment (arXiv 2503.13657).

The stages are getting better on their own. The seams are not, because nobody is scored on them.

Full code, including the two example seam files: github.com/[your-repo]/handoff-contracts

If you run the two-week test, I would like to know your number.

Instrument the bill, not just the model

Anil Prasad — Sun, 23 Aug 2026 13:12:07 +0000

Most AI observability stacks answer "is it up." Very few answer "what did it do, and can you prove it."

That distinction stopped being academic for me when an agent I was responsible for silently approved $2.4M in non-covered procedures over six weeks. Uptime was green throughout. A CFO found it in a month-end review. The audit pipeline never did.

This post is about the minimum record you need so that never happens, and why the same record is what lets you bill on outcomes.

What is actually missing?

Standard telemetry captures request, latency, status, token count. That tells you the system ran. It does not tell you what decision was made, on what evidence, under which version of anything.

The gap looks like this in practice. A customer disputes an output. You need to answer four questions, and you have ninety minutes before someone senior asks for an update.

Which action produced this?
Which model and which prompt or policy version was live?
What inputs and retrieved context were used?
Can you replay it and get the same result?

If any of those requires archaeology, you do not have an evidence layer. You have logs.

The minimum viable decision record

Not a framework. A shape. Anything that captures these fields survives most disputes.

from dataclasses import dataclass, asdict, field
from typing import Any
import hashlib, json

@dataclass(frozen=True)
class DecisionRecord:
    decision_id: str          # stable, referenced on the invoice line
    action: str               # "approve_claim", "resolve_ticket"
    outcome: str              # "approved" | "escalated" | "refused"
    billable: bool            # did this become a charge?
    model_id: str             # provider + exact version, never "gpt-latest"
    policy_version: str       # your prompt/rules version, semver
    input_digest: str         # hash, not the payload
    context_refs: list        # document ids + revision, not the text
    confidence: float | None
    ts: str                   # RFC3339, UTC

def digest(payload: Any) -> str:
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode()).hexdigest()

Three choices in there are load-bearing.

Hash the input, do not store it. You get tamper-evidence and replay verification without inheriting a retention and privacy problem. If the hash matches on replay, the inputs matched.

billable is a first-class field, not derived later. The moment billing logic lives somewhere other than the decision record, the two drift, and reconciling them at renewal is worse than the original problem.

Pin the exact model version. "gpt-latest" is not a version. When a provider silently updates a model behind an alias, your replay is no longer a replay and you cannot say why behaviour changed in March.

Canonical JSON, or the hash is useless

If you hash a dict without a canonical encoding, key order changes break equality and your tamper-evidence becomes noise.

def record_digest(rec: DecisionRecord) -> str:
    d = asdict(rec)
    d.pop("decision_id", None)     # id is assigned after hashing
    return digest(d)

sort_keys=True plus tight separators is enough for most cases. If you later sign these records, use a real envelope format rather than inventing one. DSSE with PAE encoding is well specified and the implementations are small.

Wiring it to the invoice

This is the part that turns a governance artifact into a revenue one.

def billable_units(records, period_start, period_end):
    """One charge, one decision_id. No aggregate-only billing."""
    return [
        {"decision_id": r.decision_id,
         "action": r.action,
         "outcome": r.outcome,
         "ts": r.ts}
        for r in records
        if r.billable and period_start <= r.ts < period_end
    ]

The rule that matters: every charge on an invoice resolves to exactly one decision_id. Not a count. Not a rollup. If a customer questions line 4,127, you return that record.

Vendors who cannot do this end up defending totals instead of transactions, which is an argument you lose slowly.

Why this is a pricing decision, not a compliance one

Look at what the AI customer service market charges right now.

Vendor	Unit	Price
Intercom Fin	Resolution, free if escalated	$0.99
HubSpot	Resolved conversation	$0.50
Zendesk	Verified resolution, LLM-confirmed within 72h	~$1.20 to $2.00
Salesforce Agentforce	Conversation, resolved or not	$2.00

Same category. Different units. The vendors billing on outcome are carrying the failure rate themselves, and they can only do that if they can evidence the outcome.

One leading vendor reports 76% average resolution across 8,000+ customers. Independent reports put the same metric at 42 to 50%. That spread is not fraud. It is what happens when the billable unit has no verifiable definition.

Zendesk's answer was not a discount. They restructured in May 2026 to bill only on a resolution confirmed by a separate evaluation model within 72 hours. They added an auditor and removed the argument.

Four checks before you price anything

Can the customer see the unit? Tokens are your problem. A resolved ticket is theirs.
Can you measure it without argument? Write the definition as if procurement will read it, because eventually they will.
Does it scale with their value, not your cost? A unit indexed to compute makes you a reseller of inference.
Can you defend the bill nine months later? Which action, which version, what evidence, on demand.

Most teams pass 1 and 3, and fail 2 and 4.

The metric nobody tracks

Add time to reconstruct to your dashboard. If a customer disputes an output today, how long until you can show exactly how it was produced?

It is measurable, it is cheap to instrument, and it predicts your next bad quarter better than accuracy does. In my experience teams discover the honest answer is measured in days, and they discover it during the dispute rather than before it.

Caveats

This is a shape, not a library, and I have deliberately kept it small. Signing, retention policy, PII handling in context_refs, and replay determinism under a non-deterministic runtime are all real problems this sketch does not solve. Determinism for RL rollouts shipped in vLLM this year as a beta flag at roughly double the latency, and almost nobody turned it on, which tells you something about how much appetite there is for the strict version.

If you have built this properly in production, I would like to know where the record schema broke first. Mine broke on context references, because document revisions were not stable identifiers and I had assumed they were.

Put a scoring gate in front of your LLM call, not a human

Anil Prasad — Fri, 14 Aug 2026 18:07:16 +0000

Here is the pattern, in the smallest form that still does the job.

from dataclasses import dataclass

@dataclass(frozen=True)
class Check:
    name: str
    score: callable          # (output, ctx) -> float in [0, 1]
    floor: float
    hard: bool = False       # hard failures never retry; they halt

def gate(generate, checks, ctx, attempts=3):
    """Return (output, trace). Never returns an output that failed a check."""
    trace, feedback = [], None
    for i in range(attempts):
        out = generate(ctx, feedback=feedback)
        scores = {c.name: c.score(out, ctx) for c in checks}
        failed = [c for c in checks if scores[c.name] < c.floor]
        trace.append({"attempt": i, "scores": scores,
                      "failed": [c.name for c in failed], "output": out})
        if not failed:
            return out, trace
        if any(c.hard for c in failed):
            raise HardFailure(failed[0].name, trace)
        feedback = render_feedback(failed, scores)
    raise Escalate(trace)

That is the whole idea. No model output reaches anything downstream until it has been scored, and a failure feeds the failure back into the next attempt rather than being logged and forgotten.

I build this in healthcare revenue cycle, where the output is an insurance
appeal and the hard check is protected health information. But nothing in the pattern is domain specific. Any time you want an LLM to do work nobody reads line by line, this is the shape.

Why bother, instead of a human in the loop

Because a human in the loop on every output is not a safety feature, it is a throughput ceiling, and usually the one you were trying to raise.

The task I care about is writing appeals against denied insurance claims. The published numbers make the case better than I can: KFF reported in July 2026 that skilled nursing denials are overturned 95 percent of the time when appealed, and appealed 18 percent of the time. Nobody skips an appeal they expect to win. They skip it because nobody is free to write it.

Put a person on every output and you have moved the work from writing to
reviewing. Faster, but the same ceiling, plus a licence fee. So the engineering question becomes: what has to be true for the output to be
safe to send unread?

Design note one: hard checks are not just checks with a high floor

This is the part people get wrong on the first pass, and I did too.

Most checks are quality checks. Fail one and retrying is correct, because the model can often fix it given the failure as context.

Some checks are not like that. In my domain, PHI leakage is one. If the output contains protected data it should not, retrying is exactly the wrong move: you have already produced the thing, and the correct response is to halt, raise an incident, and page a human. Retrying a safety failure is how you turn one incident into three.

CHECKS = [
    Check("groundedness", score_groundedness, 0.90),
    Check("accuracy",     score_accuracy,     0.95),
    Check("variance",     score_variance,     0.88),
    Check("phi_safety",   score_phi,          1.00, hard=True),
]

Note the floor on the hard check is 1.00, not 0.99. There is no partial credit available on that dimension, and a floor of 0.99 is an admission that you expect to leak occasionally.

Design note two: the feedback is the retry

A retry that sends the same prompt again measures your temperature setting. A retry has to carry what failed.

def render_feedback(failed, scores):
    lines = ["Your previous answer did not pass validation. Fix these and "
             "return the corrected answer only."]
    for c in failed:
        lines.append(f"- {c.name}: scored {scores[c.name]:.2f}, "
                     f"needs at least {c.floor:.2f}. {HINTS[c.name]}")
    return "\n".join(lines)

HINTS = {
    "groundedness": "Every factual claim must appear in the provided source "
                    "documents. Remove anything you cannot point to.",
    "accuracy":     "Codes and identifiers must validate against the supplied "
                    "reference set. Do not invent plausible ones.",
    "variance":     "Answer at the level of specificity the source supports, "
                    "no more.",
}

The hints matter more than the scores. A model told "groundedness 0.71" does nothing useful. A model told "remove any claim you cannot point to in the source" usually fixes it in one pass.

In my system this loop corrects 87.2 percent of catchable issues without a person, in about 4.2 seconds and roughly 1,800 extra tokens per correction, at about 0.00054 dollars. That is measured on the runtime standalone rather than in a customer environment, and I flag that because a number without its measurement context is not a number.

Design note three: groundedness is the check that earns its keep

If you implement only one, implement this one. It is also the one people
implement worst, usually as an embedding similarity between output and context, which is close to useless because a fluent paraphrase of something false scores well.

Decompose instead.

def score_groundedness(out, ctx) -> float:
    claims = extract_claims(out)                 # atomic factual assertions
    if not claims:
        return 1.0
    supported = sum(is_supported(c, ctx.sources) for c in claims)
    return supported / len(claims)

extract_claims is a cheap model call with a strict output schema. is_supported is another, per claim, asked as a yes-or-no with the relevant source span attached. It is more expensive than cosine similarity and it is the difference between a check and a decoration.

Two implementation notes that cost me time. Ask the support question in isolation per claim, because a model shown ten claims at once will pattern-match to "mostly fine." And log the unsupported claims, not just the ratio, because that list is the actual product of the check.

![ ](https://dev-to-uploads.s3.us-east-

2.amazonaws.com/uploads/articles/ff4lwfj0qiie75wy1swa.png)

Design note four: the trace is the point

The trace returned above looks like debugging output. It is the most valuable thing the whole pattern produces.

Persist it. Every attempt, with the before and after text, every dimension
score, whether a correction was applied, how many attempts it took, whether it escalated, which model ran, the latency and the cost.

def persist(trace, ctx):
    for row in trace:
        write_append_only({
            "request_id":       ctx.request_id,
            "attempt":          row["attempt"],
            "scores":           row["scores"],
            "failed":           row["failed"],
            "output_before":    row["output"] if row["failed"] else None,
            "output_after":     None if row["failed"] else row["output"],
            "model":            ctx.model,
            "latency_ms":       ctx.latency_ms,
            "cost_usd":         ctx.cost_usd,
        })

Append-only, and keep it as long as your regulator asks. Mine asks for seven years.

Two reasons this earns its storage. Operationally, failures cluster, and the cluster tells you what to fix long before an aggregate pass rate moves.

And this table is where your override rate lives: how often a human disagreed with the system, on what, and what happened next. In a regulated domain that is the only evidence that human review was real rather than a signature. A March 2026 discovery order in a US coverage denial case compelled production of internal AI review board materials. You cannot reconstruct that log retroactively.

What this does not do

It does not make the output correct. It makes it verifiable against what you supplied, which is a weaker and much more achievable property. If your source documents are wrong, a perfectly grounded output is confidently wrong.

It costs latency and tokens on every call that needs a retry. If your workload is latency-critical this trade may not be available. Mine is not: an appeal that takes four extra seconds is still days faster than the queue it came from.

And retry count is a parameter, not a principle. Three is where the marginal correction rate stopped justifying the latency for us. Measure yours rather than inheriting mine.

The one line I would take away

The interesting cost in an LLM system is not inference. It is verification, and whether you pay it in software or in people.

If you pay it in people, the system does not scale past their hours, which is usually the exact constraint you bought it to relieve.

The healthcare platform this pattern runs in is ARIA, which my team builds at Ambharii Labs. Performance figures above are internal, measured on the runtime standalone or on our own 197-case evaluation suite, and labelled as such. We have no independent benchmark, which is the honest gap.

Sources: KFF, 6 July 2026. Lokken v. UnitedHealth discovery order, 9 March 2026.

If you run a gate like this, I would like to know what your hard-check list contains. That list is a very direct statement of what an organization thinks is unrecoverable, and I have never seen two that match.

Measure pass-all-k, not accuracy: a reliability harness in 60 lines

Anil Prasad — Fri, 07 Aug 2026 18:44:17 +0000

Run this against whatever you already have.

`from collections import Counter

def pass_all_k(run, tasks, k=8):
"""run(task, variant) -> bool. Returns (pass_all_rate, mean_rate)."""
all_pass, total = 0, 0
for t in tasks:
results = [run(t, variant=i) for i in range(k)]
all_pass += all(results)
total += sum(results)
return all_pass / len(tasks), total / (len(tasks) * k)`

Two numbers come back. The second is what your dashboard shows. The first is what a customer experiences, because customers do not get to retry until it works.

They are rarely close. On the systems I have measured, a mean around 0.85 has sat with a pass-all-8 around 0.45, and I have never once seen the gap go the other way.

Why the gap exists

If failures were independent at rate p, pass-all-k would be (1-p)^k and you could compute it rather than measure it. Failures are not independent, which is the entire point. They cluster by task shape.

That clustering is the actionable part. A 2026 reliability study running 23,392 episodes across ten models and a 396-task benchmark found that degradation was domain specific rather than model specific: a graceful degradation score fell from 0.90 to 0.44 in software engineering as task length grew, while document processing barely moved, 0.74 to 0.71.

So the interesting output of your harness is not the number. It is which tasks are in the failing set.

The variant function is the whole design

This is where most homegrown harnesses go wrong. Eight identical calls measure your cache. You need eight honest variations of the same intent.
`import random

REPHRASE = [
lambda s: s,
lambda s: s.lower(),
lambda s: f"I need to {s[0].lower()}{s[1:]}",
lambda s: f"{s} Please be thorough.",
lambda s: s.replace("?", "").strip() + ", if you can.",
]

def make_variant(task: dict, variant: int) -> dict:
rng = random.Random(f"{task['id']}:{variant}") # deterministic
out = dict(task)
out["prompt"] = rng.choice(REPHRASE)(task["prompt"])
if task.get("states"):
out["state"] = rng.choice(task["states"])
return out`
Seed on task_id:variant rather than on a global counter. Then a re-run of task 7, variant 3 is the same input it was last week, which is the difference between a harness you can compare across releases and one you cannot.

Record per-task, not per-run
The failure mode of a reliability harness is aggregating too early.
`import json, pathlib

def evaluate(run, tasks, k=8, out="reliability.jsonl"):
fh = pathlib.Path(out).open("w")
summary = Counter()
for t in tasks:
results = []
for i in range(k):
v = make_variant(t, i)
try:
ok = bool(run(v))
err = None
except Exception as e: # a crash is a failure
ok, err = False, f"{type(e).name}: {e}"
results.append({"variant": i, "ok": ok, "error": err})
rec = {
"task_id": t["id"],
"shape": t.get("shape", "unclassified"),
"k": k,
"n_pass": sum(r["ok"] for r in results),
"pass_all": all(r["ok"] for r in results),
"runs": results,
}
fh.write(json.dumps(rec) + "\n")
summary[t.get("shape", "unclassified")] += rec["pass_all"]
fh.close()
return summary`

shape is the field that earns its keep. Tag each task with what it is rather than which model ran it: lookup, multi_step, writes_state, long_horizon, needs_tool. Group the failures by shape and the pattern usually falls out on the first run.

An exception counts as a failure. A harness that only counts wrong answers and lets timeouts through will tell you a comforting lie.

What to do with the failing set**
**
Three findings from 2026 tell you where to look first, and each is a check you can run rather than a claim you have to believe.

If a failing shape is multi-agent, test the single-agent version. A study across 180 controlled configurations found that once single-agent accuracy passes roughly 45 percent on a task, adding agents produced negative returns, and that independent agents amplified errors 17.2 times against a single-agent baseline while centralised coordination held it to 4.4 times. Read-heavy work parallelises. Write-heavy work does not, because two agents writing produce two decisions nobody reconciles.

If a failing shape is long-horizon, do not assume a better model fixes it. Same reliability study: capability and reliability rankings diverged, and advanced models showed meltdown rates up to 19 percent, apparently because they attempt harder multi-step strategies.

Before you raise reasoning effort, measure it. Across 21,730 rollouts, higher reasoning effort produced equal or lower accuracy in 21 of 36 model and benchmark combinations. It is a per-task tuning parameter with a real downside, not a quality dial. Three effort levels against the same seed set is an afternoon.

for effort in ("low", "medium", "high"): pa, mean = pass_all_k(lambda t: run(t, effort=effort), tasks, k=8) print(f"{effort:<7} pass_all={pa:.2f} mean={mean:.2f}")

Watch for the case where mean rises and pass_all falls. That is a system getting better on average and less dependable, and it is invisible if you only track one of them.

Wiring it into CI

Keep it cheap or it will be deleted within a month.
- name: reliability run: | python -m harness --k 8 --tasks tasks/core.jsonl --out reliability.jsonl python -m harness.gate --min-pass-all 0.60 --baseline main.jsonl

Two rules that have kept this alive on teams I have worked with. Gate on **regression **against the previous run, not on an absolute threshold, because an absolute number gets lowered the first time it blocks a release. And run the full k nightly while running k=3 on pull requests, because a twenty-minute pre-merge check gets disabled.

The honest limits

k=8 is arbitrary. It is enough that luck stops carrying you and few enough that people will actually run it. Use five if five is what gets done.

The variant function encodes your assumptions about what "the same request" means, and reasonable people will disagree about it. That is a feature: it forces the argument to happen in code review rather than after an incident.

And this measures reliability, not correctness. A task that fails all eight times consistently is perfectly reliable and completely wrong. You still need the assertions.

None of this is new thinking. Anyone who has run a payments system or a database already reasons about the worst request rather than the average one. We stopped doing it when the systems started sounding confident.
Sources: arXiv 2603.29231 (31 March 2026, 23,392 episodes); arXiv 2512.08296 (December 2025, 180 configurations); arXiv 2510.11977 (ICLR 2026, 21,730 rollouts).

If you already measure something like this, I would like to know what your variant function does. That part has no established convention yet and I suspect everyone has quietly invented their own.

I Have 25+ Years of Production Engineering Experience. Here Is What AI Coding Tools Actually Did to My Workflow.

Anil Prasad — Sat, 01 Aug 2026 13:31:05 +0000

I want to be upfront about something before you keep reading.

I did not write this to sell you a tool. I did not write it because a vendor sent me a license. I wrote it because I have been shipping production systems for 28 years — at Fintech, Healthcare, Lifesciences, MedTech, EnergyTech, InsuranceTech — and the conversation I keep having with engineers on my team is one that nobody is writing about honestly.

The conversation goes like this. Someone asks me which AI coding tool they should use. I start to answer and they stop me and say no, I do not mean which one is fastest or which one scored highest on some benchmark. I mean: does this actually change how you work? Does it make you better? Or does it just make you faster at the same things?

That is the question I am going to answer here. From production. From regulated environments. From real code that runs on real infrastructure that real people depend on.

The context matters, so let me give it to you.

I run engineering at an Energy Tech firm, a regulated utility. The code I ship runs power generation systems. I also run Ambharii Labs, where I build open source AI tooling — AgentMesh, ARGUS, Bulwark, TorchForge. I use Claude Code daily. I have Cursor installed. I used GitHub Copilot for 18 months before that.

I am not a neutral observer. But I am also not a reviewer who ran three tools on a todo app for a week and wrote it up. Everything I am about to tell you is grounded in production code, team dynamics, and the kind of technical debt that accumulates over years in a real engineering organization.

What actually changed when I started using these tools.

The first thing that changed was not my output speed. It was what I chose to spend time on.

Before AI coding tools, a significant portion of my day went to what I would call translation work. Taking a requirement and translating it into scaffolding. Writing the boilerplate that connects the idea to the actual logic. Setting up the test harness before I could write a single meaningful test. This work is not hard. It is just slow, and it consumes the same cognitive budget as the work that actually requires judgment.

Claude Code changed that specific problem for me more than any other tool. Not because it is smarter in every situation. Because it operates at the level of the whole task rather than the current line. I can describe what I am building in plain language, and it produces scaffolding that is close enough to useful that my actual work starts at a higher level than it did before. At Energy Tech Enterprise, I used it to rebuild a data pipeline integration in a weekend that would have taken two engineers a sprint. The resulting code needed review and adjustment. But the review took a fraction of the time the original construction would have.

What did not change.

This is the part that most reviews skip, so I am going to spend more time on it.

Judgment did not change. Specifically, the judgment required to know when the AI is wrong in ways that will not surface until production. Claude Code generated a Redis failover pattern for me last quarter that worked perfectly in testing and would have silently corrupted data under a specific race condition at scale. I caught it because I have seen that failure mode before. Not because any test flagged it. Not because the AI warned me. Because 28 years of watching distributed systems fail gives you pattern recognition that no tool has yet.

Architecture did not change. Every significant architectural decision I have made in the past year — how to structure the ARGUS observability layer, how to design the multi-tenant isolation in Aether AI, how to partition the compliance framework in Bulwark — was made by me, not with the AI. The AI is genuinely useful for implementation. It is not useful for deciding what to build or how the pieces should relate to each other. That distinction matters more than most people are willing to say.

Debugging production failures did not change. When something breaks in a regulated environment at 2am, the skill that matters is the ability to form a hypothesis from incomplete information, trace causality through a system you did not fully build, and make a decision under pressure. I have not seen any AI tool improve this skill. I have seen it make engineers slower at it because they reach for the tool first instead of thinking.

What the data says vs what I see on my team.

The research on AI coding tools in 2026 is genuinely contradictory.

Stanford HAI found that early-career developers aged 22 to 25 in AI-exposed roles experienced a 20% employment decline from their 2022 peak. Junior developer job postings dropped 60% from 2022 to 2024. Coding bootcamp enrollment is down 40%.

At the same time, overall software engineering postings are up 11% year over year in early 2026. The Bureau of Labor Statistics still projects 17% job growth through 2033. AI engineer roles grew 300% in the period that junior developer roles declined.

What does this look like from inside an engineering organization? I manage a team. I have watched this play out in real hiring decisions, not statistics.

We stopped backfilling junior roles the same way we used to. Not because we decided to. Because the calculus changed quietly. A senior engineer with good AI tool fluency now covers ground that previously required a junior and a senior working together. That is not a policy decision. It is an economic one that happens below the level of any announcement.

The consequence nobody is talking about loudly enough is the pipeline. Senior engineers come from somewhere. They come from the junior roles that are not being filled. AWS CEO Matt Garman called eliminating the junior layer "one of the dumbest things I have ever heard" and warned about a catastrophic skills gap in the next decade. He is right. But the decisions that create that gap are being made one hiring freeze at a time in teams exactly like mine, with no malice and complete economic rationality.

The honest answer to the question my engineers keep asking.

Will AI tools make you irreplaceable? No. Nothing makes you irreplaceable.

Will they make you more productive? Yes, at specific things. Scaffolding, boilerplate, documentation, first-pass testing, refactoring known patterns. The productivity gain is real and it is not small.

Will they replace you? That depends on what you do. If your value is executing tasks that have known shapes — write this endpoint, create this migration, document this function — you are at genuine risk. Not because an AI does those things better than you. Because one engineer who understands how to direct an AI does them faster than you, and the economic case for your role erodes.

If your value is understanding why a system behaves the way it does, how failure modes propagate, what the right architecture is for a constraint set that nobody fully articulated — that value has not eroded. It has increased, because the gap between people who have it and people who do not is now more visible than it was before.

What I actually use and why.

Claude Code for anything involving real codebase understanding, multi-file changes, or system-level reasoning. The context window and the ability to describe intent rather than mechanics is worth it for complex work.

Cursor for daily-driver editing when I am in flow and want inline assistance without switching contexts.

Neither for architecture decisions, production debugging, or anything where the cost of a subtly wrong answer is high.

AgentMesh sitting in front of both, because when you are working in a regulated environment you need a governance layer that tracks what the AI was asked, what it produced, and what it cost. That is not optional when the code runs infrastructure.

The question nobody asks but should.

Most conversations about AI coding tools focus on the individual developer. Should I use this tool? Will it make me faster?

The question I think matters more is what these tools do to the craft of software engineering over a decade. The engineers who are senior today learned by doing work that AI tools now do for junior developers. They debugged code they wrote poorly. They read error messages they did not understand. They built systems that failed in ways they had to diagnose. That experience is how you build the pattern recognition that catches the race condition in the Redis failover at 2am.

If the entry-level layer thins significantly — and it is thinning, the data is clear — the engineers who are senior in 2036 will have been formed differently. Whether that produces better or worse engineers at the senior level is a question nobody knows the answer to yet. Including me.

What I know is that the tools are real, the productivity gains are real, and the consequences are playing out in ways that are more complicated than either the panic or the optimism suggests.

I have 28 years of data points. This is the most uncertain moment I have seen. That is not a warning. It is just what honest looks like from here.

Anil S. Prasad is founder of Ambharii Labs and Head of Engineering and Product at Fortune 100 Energy Tech and serves on Tech and engineering Advisory roles for different enterprises, private equity frms and VC backed firms. He builds open source AI governance tools at github.com/anilatambharii — including AgentMesh (governance proxy for AI tools), Bulwark (agent security framework), and TorchForge (enterprise PyTorch governance). He writes about what actually breaks in regulated AI deployments.

Your model didn't get smarter. It learned to cheat the test.

Anil Prasad — Sun, 26 Jul 2026 13:02:04 +0000

Reward hacking, eval contamination, and irreproducible runs are the three invisible failures in modern LLM training. Here is an open-source trust layer that catches all three.

Modern fine-tuning frameworks are fast. verl, TRL, and Unsloth can saturate a GPU cluster and push tokens per second most of us could not have imagined three years ago.

But speed created a blind spot. Faster training did not make good models easier to produce. It made bad models cheaper to produce. And three failure modes crept into that blind spot, all sharing one dangerous property: they are invisible on a normal dashboard.

This post is about those three failures and an open-source project, Provenir, that catches all of them. Everything here runs with pip install provenir.

Failure 1: reward hacking

You train against a reward signal. A verifier decides if each answer earns the reward. The model maximizes reward by any means available, including means you did not intend.

The core problem: a verifier checks the answer, not the reasoning. So the model learns to pass the check without doing the thinking. Recent RLVR research documented models abandoning real reasoning on inductive tasks, producing outputs that satisfied the verifier while skipping the pattern the task required. Reinforcement learning amplifies whatever maximizes reward, so the longer you train, the worse it gets.

There are seven common modes:

Every one shows up on your dashboard as a reward curve going up.

Provenir's flight recorder and reward-hacking detector flag them per step, live:

from provenir.observability import FlightRecorder, RewardHackingDetector

recorder = FlightRecorder()
detector = RewardHackingDetector()

for step, metrics in rl_loop():
    recorder.log_step(metrics)     # KL, entropy, reward std, advantages...

for anomaly in recorder.anomalies():
    print(anomaly.kind, anomaly.step, anomaly.detail)

report = detector.analyze(rollouts)
if report.is_hacking:
    print("Reward hacking detected:", report.findings)

Failure 2: evaluation contamination

Your benchmark says 92%. If the eval set leaked into training, that number measures memory, not capability.

[IMAGE: provenir-article-fig4-compare.png]

This happens constantly. Training corpora are huge and scraped widely; benchmark questions end up inside them, sometimes verbatim, sometimes paraphrased just enough to dodge a naive string match. The fix is to check overlap the way contamination actually happens:

from provenir.eval.contamination import ContaminationChecker

checker = ContaminationChecker()   # 13-gram + embedding + exact, MinHash at scale
report  = checker.check(train_dataset, eval_dataset)
print(f"Overlap: {report.overlap_ratio:.1%} across {report.n_hits} records")

For a stronger guarantee, plant canary tokens in a private eval vault. If those tokens ever appear during training, you know your held-out set leaked. It turns "we think the eval is clean" into "we can prove it."

Failure 3: irreproducibility

A run produces a great model. Two weeks later you cannot reproduce it. Different seed, a dependency moved, the dataset shifted. You have a good model and no idea how you made it.

[IMAGE: provenir-article-fig3-passport.png]

Every Provenir run produces a content-addressed manifest: config hash, dataset hash, git SHA, seed, and a lineage DAG linking dataset → run → adapter → eval → merge.

provenir train config.yaml --dataset data/train.jsonl
# produces a tamper-evident manifest

provenir reproduce manifest_abc123.json --output reproduced_run/
# reproduces the exact run

On top sits a signed Model Passport, a portable bill of materials mapping directly to EU AI Act Article 12 (tamper-proof audit trails + model lineage, enforced August 2, 2026):

from provenir.governance.passport import ModelPassport

passport = ModelPassport.build(run.manifest, key="team-signing-key")
passport.save("passport.json")   # signed HMAC-SHA256 bill of materials

loaded = ModelPassport.load("passport.json")
assert loaded.verify(key="team-signing-key")
print(loaded.risk_flags)  # ["unscanned_pii", "contaminated_eval", "unknown_license"]

Where Provenir sits

Provenir does not reimplement kernels. It orchestrates verl, TRL, and Unsloth through backend-agnostic adapters and adds the trust layer on top. The whole thing drops into an existing loop in three lines:

import provenir

with provenir.track("my-run", dataset=train_ds) as run:
    for step, metrics in training_loop():
        run.log_step(metrics)
    run.record_eval("mmlu", score=0.71)

# run.manifest, run.flight_recorder, run.hacking_report, run.passport

Try it

pip install provenir            # manifests, eval, governance, CLI
pip install "provenir[train]"   # SFT + DPO + LoRA/QLoRA via TRL
pip install "provenir[all]"     # everything

Apache 2.0. 1,153 tests. Repo and docs: github.com/anilatambharii/provenir.

Speed without trust just means you reach the wrong answer faster and with more confidence. If you work on RL or fine-tuning infra, I would genuinely like to hear which of these three failures has cost you the most. Drop it in the comments.

HumanWritten #ExpertiseFromField

7 New Governance Features Just Shipped in AgentMesh. All Open Source, No Waitlist. published: false

Anil Prasad — Tue, 21 Jul 2026 14:00:00 +0000

The permission model is the problem, not the model
Every AI agent incident this year has the same shape, honestly. An agent gives confidently wrong advice and a human trusts it. A trading agent moves money because nothing was in place to stop it. A shared API key means nobody can even tell which agent did what, let alone revoke just one of them.

The agent isn't malfunctioning. It's doing exactly what it was allowed to do.

AgentMesh is the open source governance proxy I built to sit between your agents and everything they touch. Token budgets, semantic caching, PII and PHI and PCI masking, prompt injection detection, audit trails, all of it enforced before a call ever reaches an LLM or a tool. It's been out in the open for a few weeks now, and this last sprint added seven features that push it past "cost and safety proxy" into something that actually answers the two questions every enterprise eventually asks: who did that, and can we stop it.

Here's what's new, with the real commands, not screenshots.

OpenTelemetry export Every governance event AgentMesh emits (cache hits, quota blocks, injection detections, anomalies) now streams live as OTLP spans and metrics to whatever collector you're already running. Datadog, Honeycomb, Grafana Tempo, doesn't matter which. This isn't a batch export you have to remember to trigger either. It subscribes to the live event bus and just keeps exporting for the life of the process.

agentmesh serve --otel-endpoint http://localhost:4317
No new dashboard to learn. Your platform team's existing tooling starts seeing AI governance events show up alongside everything else they already watch.

EU AI Act readiness scanner Full enforcement of the EU AI Act's high risk obligations lands August 2, 2026. Most teams have a compliance slide deck by now. Almost none have the actual controls, and the penalty structure isn't even uniform, which trips people up constantly. Article 5 violations go up to €35M or 7% of global turnover, but the articles that actually apply to most high risk systems, that's 12, 14, 15, and 17, top out lower, at €15M or 3%. Get those two numbers confused in public and a compliance lawyer will happily correct you. So the scanner reports whichever tier actually applies, article by article, instead of always reaching for the scariest headline number.

agentmesh compliance readiness --policy your-policy.yaml
It checks four things. Article 12: is your audit trail actually tamper evident and chain verifiable, or just a database table with delusions of grandeur. Article 14: can a human actually halt the system, or do they just get an email about it afterward. Article 15: do you have real robustness against prompt injection and data exfiltration. Article 17: is your policy versioned and evidence backed, or does it live in someone's head.

Every gap it finds comes with the specific config line that fixes it. Not just a red X.

Human in the loop approval An agent that can move money, delete data, or take some irreversible action without a human checking first isn't a rogue agent. It's a correctly functioning agent with a genuinely bad job description. The approval gateway lets you flag specific tools, cost thresholds, or teams as needing a real decision before the call goes through.

agentmesh serve --require-approval-over-usd 5.00 --approval-tools "wire_transfer*,delete_*"
It doesn't sit there blocking a request thread waiting on a human, which would be a bad way to build this. A matched call gets parked as PENDING, an alert fires off to Slack or PagerDuty, and the caller gets back an HTTP 202 with an approval ID. Once a human resolves it, the caller resubmits with the header X-AgentMesh-Approval-Id and it goes through. If nobody responds in time, it fails closed. No response means no action, not the other way around, and that distinction matters more than it sounds like it should.

agentmesh approval list
agentmesh approval approve APR-a1b2c3d4 --by security-team

Per agent virtual keys Most enterprises run every single AI agent off one shared vendor API key. No way to tell which agent did what. No way to revoke one compromised agent without breaking every other one at the same time. Virtual keys fix this right at the proxy layer. Each agent gets its own amk_live_... key, scoped and revocable on its own, and the real vendor key never leaves the proxy. The agent never even sees it.

agentmesh keys create nightly-triage-bot --team engineering --scopes "claude-code,cursor"
agentmesh keys revoke vk_a1b2c3d4 --reason "rotated"
Keys are stored hashed with SHA-256, same idea as password storage. Lose one and you revoke and reissue. You don't get it back, and that's on purpose.

MCP governance wrapper LLM call governance alone has a real blind spot. By the time an MCP tool call actually happens, the LLM call that requested it has already been governed and already returned. A database read with no row level scope, or an agent delegating to another agent and just handing over its entire permission set, none of that ever passes through LLM call governance. It only shows up at the tool boundary, if it shows up at all. agentmesh wrap puts the same PII scanning, injection detection, scope enforcement, and approval gating in front of any MCP stdio server.

agentmesh wrap --agent-id triage-bot --pii-mode mask --approval-tools "wire_transfer*,delete_*" -- python my_mcp_server.py
A blocked or pending call never reaches the wrapped server at all. It gets answered with a JSON-RPC error directly. Everything else, tool listing, resources, prompts, passes through untouched.

Compliance policy packs AgentMesh already shipped pre built governance policies for the regulatory shapes people actually run into. HIPAA clinical, EU AI Act high risk, fintech and SOX, an enterprise baseline. They were just sitting in the repo with no real way to find or install them, which was kind of a waste. Now it's one command.

agentmesh policy list-packs
agentmesh policy install eu_ai_act_high_risk
The eu_ai_act_high_risk pack routes anything matching score_, screen_, deny_, or reject_ through human approval automatically. Article 14 compliance as a default, not something you have to remember to configure yourself. The fintech pack does the same thing for wire_transfer* and send_payment*.

A demo that's actually real

agentmesh demo
No API keys, no network calls, and definitely no scripted animation pretending to be a product. This runs the actual PIIScanner, InjectionDetector, BudgetEnforcer, AuditTrail, and ComplianceReporter classes live. A prompt gets PII masked right in front of you. An injection attempt gets blocked and it tells you which rule caught it. A simulated runaway loop hits a hard budget cap and actually stops. And it closes out by generating a real signed audit trail plus an EU AI Act readiness report. If you want to know whether any of this is real before digging through source code, this is the fastest way to find out.

The Chrome extension got faster too
Not a new feature exactly, but worth mentioning if you've used the extension before. Every prompt typed into Claude.ai, ChatGPT, or Gemini used to pause for a governance round trip and then make you click a second time to actually send it. That's fixed now. The real send fires immediately, and governance feedback like cache hits or quota warnings shows up afterward as a small toast that never blocks anything, and only when there's actually something worth knowing.

Try it

pip install agentmesh-proxy
agentmesh demo
All of this is open source, Apache 2.0, on GitHub right now.

github.com/anilatambharii/agentmesh

If any of this saves your team a compliance headache or an actual incident, a star helps the next person find it. PRs welcome too, especially on policy packs for verticals we haven't covered yet.

Anil Prasad builds AI that survives contact with the real world. Co-founder of GenomicsIQ (World Economic Forum cohort), builder of Aria RCM, an eleven agent healthcare platform running in production, and a BCG Aleph alum. AgentMesh is the governance layer he wished existed before he had to go build eleven agents inside a regulated business himself.

If you run AI tools across a team and your bill is outgrowing your usage, clone it, run agentmesh demo, and tell me where it breaks. What would you build on top of this?

Originally published in my newsletter, Field Notes: Production AI. Find me at anilsprasad.com or on X at @anilsprasad.

Building Production-Safe AI Agents in 2026: A Governance-First Architecture

Anil Prasad — Fri, 10 Jul 2026 12:59:00 +0000

TL;DR: The framework is 20% of the work. This post shows how to wrap
LangGraph or CrewAI with out-of-process governance that passes
enterprise security audits. Working code included.

I've spent 28 years building production AI systems. The pattern I'm watching repeat in 2026 is familiar: we get excited about capability, underinvest in control, and pay for it later.

This time the stakes are higher. 88% of enterprises running AI agents have already had a security incident. When the filters fail, there's no human reviewing the output—just an autonomous agent executing commands at machine speed.

Here's how to build agents that actually work in production.

The Problem with "Secure" Prompts

Most tutorials show you something like this:

system_prompt = """
You are a helpful assistant.
IMPORTANT: Never delete files or access sensitive data.
Always ask for confirmation before taking destructive actions.
"""

agent = create_agent(llm, tools, system_prompt)

This feels secure. It is not.

Why this fails:

In-process prompts are advisory. The model can be manipulated to ignore them.
Invisible to audit. No external system knows what rules were "set."
Overridable by context. Tool outputs can inject instructions that override the system prompt.
No enforcement mechanism. Nothing actually prevents the delete_file tool from executing.

In February 2026, researchers found 7 CVEs in OpenClaw. The root cause was architectural: no certified security baseline, no mandatory audit trail, no access control enforcement by default.

The Governance-First Architecture

[INSERT IMAGE: devto_architecture.png]

Here's the architecture that passes enterprise security audits:

Layer 1: Agent Runtime (Your Framework)

Use whatever framework fits your use case:

from langgraph.graph import StateGraph
from crewai import Agent, Crew

# Your orchestration logic lives here
# This is the 20% of the work

Layer 2: Policy Engine (Out-of-Process)

Every tool call goes through an external policy engine BEFORE execution:

# policy_engine.py
from typing import Callable
import json

class PolicyEngine:
    def __init__(self, policy_file: str):
        with open(policy_file) as f:
            self.policies = json.load(f)

    def evaluate(self, agent_id: str, tool_name: str, args: dict) -> bool:
        """
        Returns True if action is allowed, False otherwise.
        This runs OUT OF PROCESS from the agent.
        """
        # Check agent permissions
        agent_policy = self.policies.get(agent_id, {})
        allowed_tools = agent_policy.get("allowed_tools", [])

        if tool_name not in allowed_tools:
            self.log_denied(agent_id, tool_name, args)
            return False

        # Check data classification
        if self._contains_pii(args):
            if not agent_policy.get("pii_access", False):
                self.log_denied(agent_id, tool_name, args, "PII access denied")
                return False

        self.log_allowed(agent_id, tool_name, args)
        return True

Policy file (policies.json):

{
  "agent_researcher": {
    "allowed_tools": ["web_search", "read_file"],
    "pii_access": false,
    "max_cost_per_call": 0.10
  },
  "agent_writer": {
    "allowed_tools": ["read_file", "write_file"],
    "pii_access": false,
    "write_paths": ["/tmp/drafts/*"]
  },
  "agent_admin": {
    "allowed_tools": ["*"],
    "pii_access": true,
    "requires_human_approval": ["delete_*", "send_email"]
  }
}

Layer 3: Tool Wrapper with Policy Enforcement

Wrap every tool to enforce policy before execution:

# secure_tools.py
from functools import wraps
from policy_engine import PolicyEngine
from identity import get_current_agent_id
from audit import AuditLogger

policy = PolicyEngine("policies.json")
audit = AuditLogger()

def secure_tool(func: Callable) -> Callable:
    """
    Decorator that enforces policy before tool execution.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        agent_id = get_current_agent_id()
        tool_name = func.__name__

        # Log the attempt
        trace_id = audit.start_trace(agent_id, tool_name, kwargs)

        # Evaluate policy BEFORE execution
        if not policy.evaluate(agent_id, tool_name, kwargs):
            audit.log_denied(trace_id)
            raise PolicyViolationError(
                f"Agent {agent_id} not authorized for {tool_name}"
            )

        try:
            result = func(*args, **kwargs)
            audit.log_success(trace_id, result)
            return result
        except Exception as e:
            audit.log_error(trace_id, e)
            raise

    return wrapper

# Apply to all tools
@secure_tool
def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()

@secure_tool
def write_file(path: str, content: str) -> None:
    with open(path, 'w') as f:
        f.write(content)

@secure_tool
def delete_file(path: str) -> None:
    os.remove(path)

Layer 4: Per-Agent Identity

No shared API keys. Every agent gets a cryptographic identity:

# identity.py
import jwt
from datetime import datetime, timedelta
from contextvars import ContextVar

_current_agent: ContextVar[str] = ContextVar('current_agent')

def create_agent_token(agent_id: str, permissions: list) -> str:
    """
    Create a signed JWT for an agent instance.
    """
    payload = {
        "agent_id": agent_id,
        "permissions": permissions,
        "iat": datetime.utcnow(),
        "exp": datetime.utcnow() + timedelta(hours=1)
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def authenticate_agent(token: str) -> str:
    """
    Verify agent token and set context.
    """
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    agent_id = payload["agent_id"]
    _current_agent.set(agent_id)
    return agent_id

def get_current_agent_id() -> str:
    return _current_agent.get()

Layer 5: OpenTelemetry Observability

Full tracing for every agent action:

# observability.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Initialize tracer
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agentmesh")

def trace_agent_action(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        with tracer.start_as_current_span(func.__name__) as span:
            span.set_attribute("agent.id", get_current_agent_id())
            span.set_attribute("tool.name", func.__name__)
            span.set_attribute("tool.args", str(kwargs))

            try:
                result = func(*args, **kwargs)
                span.set_attribute("tool.success", True)
                return result
            except Exception as e:
                span.set_attribute("tool.success", False)
                span.set_attribute("tool.error", str(e))
                raise
    return wrapper

Putting It Together

Here's the full integration with LangGraph:

# main.py
from langgraph.graph import StateGraph, END
from secure_tools import read_file, write_file, secure_tool
from identity import create_agent_token, authenticate_agent
from observability import trace_agent_action

# Create agent with identity
researcher_token = create_agent_token(
    "agent_researcher", 
    permissions=["web_search", "read_file"]
)

# Authenticate before running
authenticate_agent(researcher_token)

# Define your graph with secure tools
def research_node(state):
    # All tool calls go through policy engine
    content = read_file(state["input_path"])  # Policy checked here
    return {"research": content}

def write_node(state):
    # This would FAIL for researcher agent
    # Policy engine blocks write_file for this agent
    write_file(state["output_path"], state["research"])
    return state

# Build graph
graph = StateGraph()
graph.add_node("research", research_node)
graph.add_node("write", write_node)
graph.add_edge("research", "write")
graph.add_edge("write", END)

# Run with full governance
app = graph.compile()
result = app.invoke({"input_path": "/data/source.txt"})

The Kill Switch

For production, you need the ability to stop agents immediately:

# kill_switch.py
import redis
from functools import wraps

r = redis.Redis(host='localhost', port=6379, db=0)

def check_kill_switch(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        agent_id = get_current_agent_id()

        # Check if agent is killed
        if r.get(f"kill:{agent_id}"):
            raise AgentKilledError(f"Agent {agent_id} has been terminated")

        # Check if all agents are paused
        if r.get("kill:all"):
            raise AgentKilledError("All agents paused by administrator")

        return func(*args, **kwargs)
    return wrapper

def kill_agent(agent_id: str):
    """Emergency stop for a specific agent."""
    r.set(f"kill:{agent_id}", "1")
    # Also revoke all active sessions
    revoke_agent_tokens(agent_id)

def kill_all_agents():
    """Emergency stop for all agents."""
    r.set("kill:all", "1")

What You Get

With this architecture:

✅ Every action is traceable to a specific agent with a specific identity

✅ Policy is enforced before execution, not advised in prompts

✅ Audit trails are immutable and compatible with compliance requirements

✅ Agents can be killed individually or globally in milliseconds

✅ Tool access is scoped per agent, not per prompt

✅ You pass security audits because governance is out-of-process

The Open Source Implementation

I've packaged this architecture as AgentMesh:

pip install agentmesh

# Or clone the repo
git clone https://github.com/anilatambharii/agentmesh

AgentMesh wraps any orchestration framework with the governance layer. You keep your existing LangGraph/CrewAI code and add production safety.

Basic usage:

from agentmesh import SecureAgent, PolicyEngine

# Load policy
policy = PolicyEngine.from_file("policies.yaml")

# Wrap your existing agent
agent = SecureAgent(
    base_agent=your_langgraph_agent,
    policy=policy,
    identity="agent_researcher",
    audit_backend="otlp://localhost:4317"
)

# Run with full governance
result = agent.invoke(task)

Key Takeaways

The framework is 20% of the work. LangGraph, CrewAI, and others are great for orchestration. They are not governance infrastructure.
In-process prompts are advisory. Out-of-process policy is enforceable. This is the architectural principle that matters.
Identity is non-negotiable. No shared API keys. Every agent gets cryptographic credentials.
Audit everything. If you can't trace what happened, you can't pass compliance and you can't do forensics.
Build kill switches from day one. You will need them at 3 AM.

The code is open source: github.com/anilatambharii/agentmesh

If you're building AI agents for production, I'd love to hear what patterns you're using. Drop a comment or reach out on Twitter @anilsprasad.

Anil Prasad has spent 28 years building production AI infrastructure. He is the creator of AgentMesh and founder of Ambharii Labs.

Title: I Built an AI Governance Proxy in 72 Hours. Here Is Exactly How.

Anil Prasad — Wed, 01 Jul 2026 14:01:00 +0000

Liquid syntax error: 'raw' tag was never closed

I built a zero-dependency PII scanner for AI prompts in 270 lines of Python

Anil Prasad — Fri, 26 Jun 2026 12:45:00 +0000

TL;DR — AgentMesh 0.3.2 ships a PII/PHI/PCI scanner that runs on every AI prompt before it reaches the model. 17 entity types. Under 2ms. No external API. No cloud service. Pure Python regex with Luhn validation and overlap deduplication. Three enforcement modes: mask, redact, block. pip install agentmesh-proxy

The problem

Your AI agents and tools are sending raw sensitive data to the LLM vendor.

Medical record numbers in clinical AI prompts. Credit card numbers in finance team workflows. AWS access keys in developer debug pastes. Social security numbers in HR automation.

The people doing this are not making bad decisions. They are using the tools available to them. The problem is that there is no layer between the prompt and the model that catches sensitive data first.

I built that layer into AgentMesh. Here is how it works.

What it catches

![17 entity types caught before the LLM]

17 entity types across four categories:

PII — personal identity

SSN: 567-89-0123 → [SSN]
Date of birth: 07/22/1985 → [DOB]
Email: sarah.johnson@gmail.com → [EMAIL]
Phone: (415) 867-5309 → [PHONE_US]
Passport: Passport no: US123456789 → [PASSPORT]

PCI — payment card and financial data

Visa: 4532 1234 5678 9012 → [PCI_CARD]
Amex: 3714 496353 98431 → [PCI_CARD]
Mastercard: 5500 0055 0000 0004 → [PCI_CARD]
CVV: CVV 394 → [PCI_CVV]
Routing: Routing: 021000021 → [PCI_ROUTING]
Account: Account: 000123456789 → [PCI_ACCOUNT]

PHI — HIPAA-protected medical data

Medical record: MRN: P-987654 → [PHI_MRN]
ICD-10 diagnosis: E11.9 → [PHI_ICD10]
Medication dosage: 10mg lisinopril → [PHI_DOSAGE]
Provider ID: NPI: 1234567890 → [PHI_NPI]

CII — cloud credentials and infrastructure

AWS key: AKIAIOSFODNN7EXAMPLE → [CII_AWS_KEY]
JWT token: eyJhbGci... → [CII_JWT]

Quick start

pip install agentmesh-proxy

from agentmesh.security.pii_scanner import PIIScanner, ScanMode

scanner = PIIScanner(mode=ScanMode.MASK)

result = scanner.scan(
    "Patient MRN: P-987654, email: sarah@example.com, "
    "card: 4532 1234 5678 9012, key: AKIAIOSFODNN7EXAMPLE"
)

print(result.cleaned)
# Patient MRN: [PHI_MRN], email: [EMAIL],
# card: [PCI_CARD], key: [CII_AWS_KEY]

print(result.finding_types)
# ['CII_AWS_KEY', 'EMAIL', 'PCI_CARD', 'PHI_MRN']

For scanning a list of {role, content} messages (OpenAI format):

messages = [
    {"role": "user", "content": "SSN 123-45-6789, card 4532 1234 5678 9012"}
]

cleaned_messages, findings = scanner.scan_messages(messages)
print(cleaned_messages[0]["content"])
# SSN [SSN], card [PCI_CARD]

Three enforcement modes

from agentmesh.security.pii_scanner import PIIScanner, ScanMode, PIIDetectedError

# MASK: replace with labeled placeholder — model still gets a useful prompt
scanner = PIIScanner(mode=ScanMode.MASK)
result = scanner.scan("SSN 123-45-6789")
print(result.cleaned)  # "SSN [SSN]"

# REDACT: replace with *** — when even the label is too much context
scanner = PIIScanner(mode=ScanMode.REDACT)
result = scanner.scan("SSN 123-45-6789")
print(result.cleaned)  # "SSN ***"

# BLOCK: raise PIIDetectedError — zero tolerance, reject the request
scanner = PIIScanner(mode=ScanMode.BLOCK)
try:
    scanner.scan("SSN 123-45-6789")
except PIIDetectedError as e:
    print(e.findings)  # [Finding(entity_type='SSN', ...)]
    # Return HTTP 400 to the caller

Engineering decisions worth explaining

Why regex over an NLP model?

Speed. The scan runs in under 2ms. An NLP-based entity recognizer adds 50ms to 200ms per call and requires a model download. For a proxy that sits in the path of every LLM call, 2ms is acceptable and 200ms is not.

The tradeoff is recall. Regex will miss creative obfuscation. For governance purposes — where the goal is catching accidental leakage, not adversarial attacks — regex is the right tool.

The credit card validation decision

Standard implementations run Luhn validation on card numbers and only mask numbers that pass. We run in strict_pci=True mode by default:

# In PIIScanner.__init__:
# strict_pci=True (default): mask any card-shaped number (13-19 digits)
# even if it fails the Luhn check.
# Rationale: governance proxies should over-mask rather than under-mask.
# A false positive costs one masked token.
# A false negative sends a real card number to the vendor.
self.strict_pci = strict_pci

If you prefer Luhn validation only:

scanner = PIIScanner(mode=ScanMode.MASK, strict_pci=False)

The overlap deduplication problem

This one took a few iterations to get right.

Consider a prompt containing MRN: A1234567. The PHI_MRN pattern matches the whole span. The PASSPORT pattern (before it required a passport: prefix) would also match the A1234567 part.

If you apply replacements in reverse order by start position — which is the standard approach to keep earlier offsets valid — and the inner match gets processed first, it replaces 8 characters with 10 ([PASSPORT]). The outer match then tries to cut at the original end offset, which now points into the middle of [PASSPORT], producing [PHI_MRN]T].

The fix:

def _dedup_overlapping(findings: List[Finding]) -> List[Finding]:
    # Sort by start position, then by length descending (outermost first).
    # Walk forward and drop any finding whose start is inside the
    # previous kept finding's range.
    sorted_f = sorted(findings, key=lambda f: (f.start, -(f.end - f.start)))
    result: List[Finding] = []
    last_end = -1
    for f in sorted_f:
        if f.start >= last_end:
            result.append(f)
            last_end = f.end
    return result

Keep the outermost match. Drop everything whose start position falls inside it. Apply replacements in reverse order on the deduplicated list. No artifacts.

Wiring it into the proxy

If you are running AgentMesh as a proxy rather than calling the scanner directly, activate it in config:

# agentmesh.yaml
pii_mode: mask           # mask | redact | block
block_injections: true   # prompt injection detection (14 rules)
anomaly_detection: true  # runaway loop + burn rate monitoring
slack_webhook: ""        # optional: alert destination

agentmesh serve --config agentmesh.yaml --port 8080

Point your agents at it:

export OPENAI_BASE_URL=http://localhost:8080/v1
export ANTHROPIC_BASE_URL=http://localhost:8080

Every call going through the proxy now gets scanned. The response includes a header showing what was found:

X-AgentMesh-PII-Findings: 4
X-AgentMesh-Cache: miss
X-AgentMesh-Cost-USD: 0.000420

The Chrome extension

A server-side proxy cannot intercept prompts typed directly into the ChatGPT or Claude.ai browser tab. For that there is a Chrome extension — same scanner, running locally in the browser process before the request leaves the tab.

Google approved it last weekend.

Install from the Chrome Web Store (link in the repo readme) or build from source. Works with ChatGPT, Claude.ai, Gemini, Perplexity, and Cursor. No server required for standalone use.

HIPAA in production

If your team uses AI in a clinical setting, the PHI scanner is the piece that matters most. ICD-10 codes are two to five characters but identify specific diagnoses. Combined with a medical record number and a provider NPI, they reconstruct a patient record from a prompt.

AgentMesh also generates HIPAA readiness reports:

from agentmesh.compliance.pdf_report import ComplianceReporter, Framework

reporter = ComplianceReporter()
markdown = reporter.generate_markdown(Framework.HIPAA)
# or
reporter.generate_pdf(Framework.HIPAA, output_path="hipaa_report.pdf")

Outputs a structured report listing which controls are active, which are not, and what gaps remain. Useful for security reviews before a compliance audit.

Try it

pip install agentmesh-proxy

from agentmesh.security.pii_scanner import PIIScanner, ScanMode

scanner = PIIScanner(mode=ScanMode.MASK)
result = scanner.scan("your prompt here")
print(result.cleaned)
print(result.finding_types)

The scanner is in agentmesh/security/pii_scanner.py. About 270 lines. No external dependencies beyond the Python standard library.

Repo: https://github.com/anilatambharii/agentmesh
PyPI: agentmesh-proxy
Docker: docker pull anilsprasad/agentmesh:latest
Apache 2.0.

What entity types would you add? What patterns are you seeing in your team's prompts that are not covered here?

Find me: anilsprasad.com · X @anilsprasad · LinkedIn

I put one proxy in front of every AI tool my team uses 85% cache hits, 75% lower cost

Anil Prasad — Sun, 14 Jun 2026 22:30:49 +0000

TL;DR — Your team's AI tools (Claude Code, Copilot, ChatGPT, Gemini, your own agents) each call the LLM API independently — no shared cache, no shared budget, no audit trail. AgentMesh is an open-source proxy that sits in front of all of them and runs every call through a three-layer cache, per-team quotas, cheapest-model routing, and a tamper-evident audit log. You point your tools at it with two env vars. On a reproducible benchmark (no API keys): 85% cache hits, 75% lower cost. Apache 2.0. → pip install agentmesh-proxy

The problem, in one sentence
Every AI tool on your team talks to the model on its own.
Claude Code has its own connection. Copilot has its own. The ChatGPT tab in someone's browser has its own. Your LangGraph service has its own. None of them share a cache, a budget, or an audit log — so the same prompt gets paid for over and over, a runaway loop in one service is invisible to the others, and nobody can answer "what did we send to third-party APIs last quarter?"
This isn't a discipline problem. It's a missing layer. So I built it.

60-second quickstart

pip install agentmesh-proxy sentence-transformers
agentmesh serve --port 8080 --demo

Point any tool at it — no code changes:

# Claude Code, or any Anthropic SDK tool
export ANTHROPIC_BASE_URL=http://localhost:8080

# Copilot / Cursor / any OpenAI SDK tool
export OPENAI_BASE_URL=http://localhost:8080/v1

Every response comes back with governance headers so you can see what happened:

X-AgentMesh-Cache:     hit          # exact | semantic | miss
X-AgentMesh-Tokens:    0            # 0 on a cache hit
X-AgentMesh-Cost-USD:  0.000000     # $0 on a cache hit

That's the whole integration. The agent code never knows the proxy is there.

How it works
Every call from the proxy or the SDK runs the same ordered pipeline:

The interesting part is the cache, because it does something most "LLM caches" don't.

Exact-match caching almost never hits in real life, because people rephrase: they paste You are a senior architect. in front of the question, switch between optimise and optimize, wrap things in markdown. So before anything is hashed or embedded, AgentMesh normalizes the prompt — stripping the noise that doesn't change meaning:

from agentmesh.optimizer.normalizer import normalize_prompt

normalize_prompt("You are a senior architect. **Please** review this design...")
# -> "review this design ..."   (persona prefix, markdown, filler removed)

compares by cosine similarity:

from agentmesh import SemanticCache

cache = SemanticCache(similarity_threshold=0.70)   # tune per workload
cache.put("Review this microservices design for scaling issues", response)

# Different wording, same intent -> still a hit
hit = cache.get("Analyze this distributed system design")

normalize, then embed is the whole trick — it's the difference between a cache that almost never hits and one that hits ~85% of the time.

And because every call already flows through one interceptor, a tamper-evident audit log is almost free — each entry is hash-chained (SHA-256) and signed with Ed25519:

from agentmesh import AuditTrail
trail = AuditTrail()
# ... calls happen ...
assert trail.verify()   # walks the chain, checks every prev_hash + signature

The benchmark (run it yourself, no API keys)

I didn't want to ship a number you can't check, so the benchmark runs in demo mode:

python examples/benchmark.py

Total requests          20
Exact cache hits         2  (10%)
Semantic cache hits     15  (75%)
Total misses             3  (15%)

Cost WITHOUT AgentMesh  $0.0030
Cost WITH AgentMesh     $0.0008
Savings                 $0.0023  (75%)

20 requests, 5 topics, 4 phrasings each. 85% never reached the model; the 3 misses are the cold-start first call per topic — exactly what you'd expect.

There's also a Chrome extension
A proxy can't see a prompt typed straight into the ChatGPT or Gemini tab. So there's an extension: declarativeNetRequest reroutes api.anthropic.com / api.openai.com to localhost:8080, and content scripts show a governance overlay before the prompt is sent. Stats persist across service-worker restarts.

What's deliberately not built yet

I'd rather ship a small, verifiable core than a wide surface of half-features:

The cache is in-memory, single-process **— great for a local proxy, not yet a fleet. **Redis is next.
No native VS Code panel (env vars + the Chrome extension for now).
No SAML/SSO identity propagation; quotas key on a team header.

None of these are research problems — they're scope. PRs welcome, especially the Redis backend.
Try it / contribute

pip install agentmesh-proxy sentence-transformers
python examples/benchmark.py     # 85% cache hits, 75% lower cost

Repo (star it): https://github.com/anilatambharii/agentmesh
PyPI: agentmesh-proxy · Docker: anilsprasad/agentmesh · also on Hugging Face
Apache 2.0

If you run AI tools across a team and your bill is outgrowing your usage, clone it, run the benchmark, and tell me where it breaks. What would you build on top of this?