AI Agent Reliability Testing: Catching Silent Failures
Tom PiaggioCo-Founder at Autonoma
AI agent reliability testing is the pre-production discipline of verifying that an LLM agent takes the correct sequence of actions, not just that its final response reads well, across the natural non-determinism of sampling temperature, model updates, and tool-call ordering. It moves the test target from the last message in a transcript to the full trajectory: which tools got called, in what order, with what arguments, and what state resulted. Done well, it uses a recorded behavioral baseline to catch drift between runs and a staged rollout to catch whatever the baseline missed.
Your support agent tells a customer their refund is on the way. The message is warm, specific, and confident: correct order number, correct amount, a realistic timeline. Read in isolation, it looks like the agent nailed the job. It also never called the refund API. The eval that graded this transcript scored it 9 out of 10 for helpfulness. Nobody checked whether the tool call happened, because nobody was looking at anything but the text.
That's a silent failure: an agent that returns a plausible-looking response that is actually wrong, and a test suite with no way to notice because it only reads the final answer. Final-answer checks (does the response contain the right keywords, does an LLM judge rate it "helpful") catch the outputs that read badly. They are structurally blind to the ones that read fine and did the wrong thing underneath. As agents get better at sounding right, this gap gets more dangerous, not less, because the failures that used to look broken now look shipped.
A response that reads correctly and an action that executed correctly are not the same claim. Only one of them is a bug waiting to reach production.
Where the Non-Determinism Actually Comes From
Reliability testing exists because the same input to the same agent does not reliably produce the same behavior, and that's worth taking apart instead of treating as a black box. Five sources account for most of it.
Sampling temperature is the obvious one: any temperature above zero means the model is drawing from a probability distribution over next tokens, not returning a fixed answer, so two identical calls can diverge at the first token that has a close runner-up. Less obvious, and more disruptive in practice, is model version drift. Providers update models behind the same API endpoint and version alias, quietly, without a version bump you control for. An agent that behaved one way on Monday can behave differently on Tuesday with zero code changes on your side, because the model underneath the alias changed. Tool-call ordering is a third source: when an agent has several valid tools available for a step (search the docs, call the CRM, ask a clarifying question), which one it reaches for first is itself a probabilistic choice, and a different first choice cascades into a different trajectory even when the final answer converges. Retrieval variance adds a fourth axis for any agent backed by a vector store: near-duplicate chunks, index updates, or a slightly different embedding call can change which context gets retrieved, which changes what the model has to reason over. And context-window truncation is the quiet one: as a conversation or a tool-call history grows, older turns get dropped or summarized to fit the window, and an agent can lose the exact instruction or constraint that mattered three steps back without any error being raised.
None of these are bugs in the traditional sense. They are the normal operating behavior of a system built on a probabilistic model, and a recent academic effort to formalize this, "Towards a Science of AI Agent Reliability", argues for treating agent reliability as its own measurable discipline rather than folding it into general software reliability engineering, precisely because these sources of variance don't have direct analogues in deterministic code. The practical consequence is that a test suite built for deterministic software (run it once, assert equality, done) will pass agents that are actually unreliable, and fail agents that are actually fine, because it's measuring the wrong thing.
Every one of those five sources produces divergent trajectories, not just divergent wording, which is why a suite built to catch this has to look at the trajectory and not just the final message, and why a static, one-shot test tells you almost nothing about whether an agent will behave the same way tomorrow. That question, whether the agent's chosen action was actually correct against a live product rather than just plausible in isolation, is exactly where Autonoma comes in, and it's worth pausing on before getting into the suite itself.
A Runnable Reliability Suite: Testing the Trajectory, Not the Reply
If final-answer checks miss silent failures, the fix is to assert on the trajectory: the ordered sequence of steps, tool calls, and intermediate state the agent produced on the way to its answer, not just the answer itself. Concretely, that means recording every step as a structured event (which tool got called, with what arguments, what it returned, and what the agent decided to do next) and writing assertions against that sequence, the same way you'd assert against a function's return value, except the "return value" here is a list of steps.
A step-level assertion can check things a final-answer check structurally cannot: that a refund_api.create call happened at all, that it happened after the order was looked up and not before, that its arguments matched the amount the agent's text described, and that no tool call was skipped even though the final message implied it had run. This is the same underlying idea as AgentAssay, a benchmark built specifically to grade agent trajectories rather than final outputs, and the reasoning behind it applies directly here: an agent can be wrong in the middle and still produce a final answer that sounds right, so the test has to reach into the middle.
"""trajectory_recorder.py=======================Reference implementation of a TRAJECTORY RECORDER for testing AI agents.Core idea (see the blog post "AI Agent Reliability Testing: Catching SilentFailures"): testing an agent by grading only its final answer text is notenough. An agent can produce a perfectly confident, well-worded finalmessage ("Your refund has been processed!") while silently skipping thetool call that was actually supposed to perform the mutation. This is aSILENT FAILURE: the trajectory (the sequence of tool calls the agentactually made) diverges from what the final text claims happened.This module provides two things:1. TrajectoryRecorder - wraps a stub agent's tool calls and appends a structured event (tool name, arguments, result, timestamp) to an in-memory trajectory log for every step the agent takes. This gives you a ground-truth, step-level record of what actually happened, independent of what the agent's final natural-language answer says.2. assert_step_sequence(trajectory, expected_pattern) - a step-level assertion helper. This is TRAJECTORY testing, not FINAL-ANSWER testing: instead of asserting on the final answer text, you assert on the shape of the trajectory itself: - a lookup step must precede any mutating step (you can't mutate an entity you never looked up), and - a mutating step's arguments must reference an entity a prior lookup step actually returned (you can't refund order #A9128 if your only lookup returned order #A0001).Run this file directly (`python trajectory_recorder.py`) to see both aPASSING trajectory (the agent does everything right) and a FAILING one(the agent's final text claims success but the refund tool call wassilently skipped) evaluated by assert_step_sequence."""from __future__ import annotationsimport jsonfrom dataclasses import dataclass, fieldfrom typing import Any, Callable, Optional# ---------------------------------------------------------------------------# 1. The recorder# ---------------------------------------------------------------------------@dataclassclass TrajectoryRecorder: """Records every tool-call step an agent takes as a structured event. Each event is a plain dict, so the whole trajectory is trivially JSON-serializable and can be diffed by baseline_differ.py or replayed by canary_gate.py. """ trajectory: list = field(default_factory=list) _clock: int = field(default=0, repr=False) def _tick(self) -> int: # A monotonically increasing logical clock stands in for a # wall-clock timestamp. Using a logical counter instead of # time.time() keeps recorded trajectories byte-for-byte # reproducible across runs, which matters once you start diffing # trajectories against a stored baseline. self._clock += 1 return self._clock def record_step(self, tool_name: str, arguments: dict, result: Any) -> dict: """Append one step to the trajectory and return the recorded event.""" event = { "step": len(self.trajectory) + 1, "tool": tool_name, "arguments": arguments, "result": result, "timestamp": self._tick(), } self.trajectory.append(event) return event def track(self, tool_name: str) -> Callable: """Decorator form: wraps a stub tool function so every call to it is automatically recorded as a trajectory step. @recorder.track("lookup_order") def lookup_order(order_id): ... This is the shape you'd use to instrument a real agent's tool registry without touching the agent's own control flow. """ def decorator(fn: Callable) -> Callable: def wrapped(*args, **kwargs): result = fn(*args, **kwargs) # Record positional args under a stable synthetic key so # the log stays JSON-friendly regardless of the wrapped # function's signature. arguments = dict(kwargs) if args: arguments["_args"] = list(args) self.record_step(tool_name, arguments, result) return result return wrapped return decorator def as_json(self) -> str: return json.dumps(self.trajectory, indent=2)# ---------------------------------------------------------------------------# 2. Step-level assertions# ---------------------------------------------------------------------------class StepSequenceError(AssertionError): """Raised when a trajectory does not match its expected pattern."""def assert_step_sequence(trajectory: list, expected_pattern: list) -> None: """Validate a recorded trajectory against an expected step pattern. This is the step-level assertion at the heart of trajectory testing: we never look at what the agent *said* in its final answer, only at what it *did*. That is exactly what catches a silent failure, where the final text is confident but a required tool call never happened. `expected_pattern` is a list of step specs, each a dict with: - "tool": exact tool name expected to appear in the trajectory. - "kind": "lookup" | "mutate" | "notify" - lookup steps establish entities that later mutate steps are allowed to act on; mutate steps require at least one prior lookup. - "entity_arg" (mutate steps only): the argument key in this step whose value must equal an entity ID actually returned by some prior lookup step. - "entity_result_key" (lookup steps only): the key in this step's `result` dict that holds the entity ID it looked up. Defaults to "id". Raises StepSequenceError on any violation, including a mutating step that is entirely MISSING from the trajectory - the classic silent-failure signature. """ recorded_tools = [step["tool"] for step in trajectory] known_entities: set = set() # entity IDs surfaced by lookup steps so far seen_lookup = False cursor = 0 # walk the trajectory forward as we consume expected steps for expected in expected_pattern: tool = expected["tool"] kind = expected.get("kind", "other") # Find the next occurrence of this tool at or after the cursor. found_index: Optional[int] = None for i in range(cursor, len(trajectory)): if trajectory[i]["tool"] == tool: found_index = i break if found_index is None: raise StepSequenceError( f"expected step '{tool}' ({kind}) was never recorded in the " f"trajectory. Recorded tools were: {recorded_tools}. This is " f"the silent-failure signature: the agent's final answer may " f"claim success, but the required tool call never happened." ) step = trajectory[found_index] if kind == "lookup": entity_key = expected.get("entity_result_key", "id") result = step.get("result") or {} entity_id = result.get(entity_key) if isinstance(result, dict) else None if entity_id is not None: known_entities.add(entity_id) seen_lookup = True elif kind == "mutate": if not seen_lookup: raise StepSequenceError( f"mutating step '{tool}' occurred with no prior lookup " f"step. A mutation must always be preceded by a lookup " f"that establishes the entity being mutated." ) entity_arg = expected.get("entity_arg") if entity_arg: referenced_entity = step["arguments"].get(entity_arg) if referenced_entity not in known_entities: raise StepSequenceError( f"mutating step '{tool}' referenced entity " f"{referenced_entity!r} via argument '{entity_arg}', " f"but no prior lookup step returned that entity. " f"Known entities at this point: {sorted(known_entities)}." ) cursor = found_index + 1# ---------------------------------------------------------------------------# 3. Demo: a stub support agent handling a refund request# ---------------------------------------------------------------------------def _run_refund_scenario(skip_refund_call: bool) -> list: """Simulates a stub support agent working through a refund request. Steps: lookup_order -> check_policy -> refund_api.create -> send_email When `skip_refund_call` is True we simulate the silent-failure case: the agent still calls send_email with confident "refund processed" language, but the refund_api.create mutating call never happens. """ recorder = TrajectoryRecorder() # Step 1 (lookup): agent looks up the order the customer is asking about. order = {"id": "order_9F21", "status": "delivered", "total": 84.00} recorder.record_step("lookup_order", {"order_id": "order_9F21"}, order) # Step 2 (lookup): agent checks whether this order is eligible for refund. policy_result = {"eligible": True, "reason": "delivered_within_30_days"} recorder.record_step("check_policy", {"order_id": order["id"]}, policy_result) # Step 3 (mutate): the actual refund. This is the step that must exist # for the refund to have really happened. if not skip_refund_call: refund_result = {"refund_id": "rf_001", "status": "succeeded", "amount": 84.00} recorder.record_step( "refund_api.create", {"order_id": order["id"], "amount": 84.00}, refund_result, ) # Step 4 (notify): agent emails the customer regardless. This is the # dangerous part: the email text ("your refund has been processed") # reads identically whether or not step 3 actually ran. recorder.record_step( "send_email", {"to": "customer@example.com", "body": "Your refund has been processed."}, {"sent": True}, ) return recorder.trajectoryEXPECTED_REFUND_PATTERN = [ {"tool": "lookup_order", "kind": "lookup", "entity_result_key": "id"}, {"tool": "check_policy", "kind": "lookup"}, {"tool": "refund_api.create", "kind": "mutate", "entity_arg": "order_id"}, {"tool": "send_email", "kind": "notify"},]def _print_check(label: str, trajectory: list) -> None: print(f"\n--- {label} ---") print(json.dumps(trajectory, indent=2)) try: assert_step_sequence(trajectory, EXPECTED_REFUND_PATTERN) print(f"[PASS] {label}: trajectory matches expected step sequence.") except StepSequenceError as exc: print(f"[FAIL] {label}: {exc}")if __name__ == "__main__": good_trajectory = _run_refund_scenario(skip_refund_call=False) _print_check("Healthy run (refund_api.create present)", good_trajectory) bad_trajectory = _run_refund_scenario(skip_refund_call=True) _print_check( "Silent-failure run (refund_api.create SKIPPED, email still sent)", bad_trajectory, )
The recorder above wraps each tool call the agent makes and appends a structured event (tool name, arguments, result, timestamp) to a trajectory log, then exposes an assert_step_sequence helper that checks the recorded steps against an expected pattern, for example "a lookup step must precede any mutating step, and a mutating step's arguments must reference an entity that a prior lookup actually returned." That single assertion catches the refund example from the opening of this piece, because the trajectory would simply be missing the refund_api.create event that the final text implied.
Building a Behavioral Baseline You Diff Against
A single trajectory assertion catches a single known failure mode. What it doesn't catch is drift: the agent slowly behaving differently over weeks as a model alias updates upstream, as a prompt gets a small edit, or as a tool's underlying data shifts. Catching that requires a reference point to diff against, which is what a behavioral baseline is for.
The idea is straightforward: pick a representative set of scenarios, run the agent once under controlled conditions, and record the full trajectory of each run as your baseline (a "golden" trajectory), the same way a visual regression tool keeps a reference screenshot. Every subsequent run of that scenario gets compared against the baseline, not just for a matching final answer, but step by step: same tools called in the same order, same categories of arguments, same terminal action.
How you log AI agent behavior for this to work matters more than the diffing logic itself. At minimum, a usable trajectory log needs: the exact input (prompt, conversation history, any retrieved context) so a run is reproducible; every tool call with its full arguments and raw result, not a summary of it; every intermediate decision point where the agent chose between multiple valid next steps; and the final action taken, separate from the final text generated, because those are two different claims and only the log lets you check both. Store this as structured, append-only JSON per run, keyed by scenario and timestamp, so you can pull any two runs and diff them months apart without having re-instrumented anything.
Four of five steps match exactly. The fifth step called the same tool with different arguments, the kind of drift a final-answer check has no way to see and a baseline diff catches immediately.
"""baseline_differ.py====================Reference implementation of a BASELINE DRIFT detector for AI agenttrajectories.Where trajectory_recorder.py catches an agent that skips a required stepentirely, baseline_differ.py catches a subtler failure mode: BEHAVIORALDRIFT. The agent still calls the right tools in the right order, but aprompt change, model upgrade, or a tweaked system message quietly changes*what it does with them*: different arguments, a different terminalaction, or a dropped/added tool call, compared to a known-good referencerun.diff_trajectories(reference, candidate) compares two trajectory logs (thesame JSON shape produced by trajectory_recorder.py) and reports threeindependent categories of divergence: 1. Tool-call presence drift - tools called in one run but not the other. 2. Argument-level drift - same tool, same position, different argument value. Treated as a WARNING: small drift (a different order ID, a slightly reworded note) is often benign, but always worth surfacing. 3. Terminal-action drift - did the *category* of the final tool call change? This is a HARD FAILURE: if your reference run ends in `refund_api.create` and the candidate run ends in `send_email` only, something structurally different happened, wording aside.Run this file directly (`python baseline_differ.py`) to see ahuman-readable drift report for the refund scenario, with the candidate'srefund amount drifted from the reference (84.00 -> 42.00)."""from __future__ import annotationsimport jsonclass BaselineDriftError(Exception): """Raised by diff_trajectories(..., strict=True) on a hard failure."""def _tool_category(tool_name): """Coarse category for a tool name, used for terminal-action comparison. A tool name changing from "refund_api.create" to "refund_api.retry" should still count as the same *category* of terminal action (a mutating refund call); this groups by the part before the dot, or the whole name if there is no dot. """ return tool_name.split(".")[0] if tool_name else tool_namedef diff_trajectories(reference: list, candidate: list, strict: bool = False) -> dict: """Compare a reference trajectory against a candidate trajectory. Returns a report dict: { "missing_in_candidate": [tool names in reference not in candidate], "unexpected_in_candidate": [tool names in candidate not in reference], "argument_drift": [ {step, tool, arg, reference_value, candidate_value} ], "terminal_action_changed": bool, "reference_terminal": tool name or None, "candidate_terminal": tool name or None, "status": "ok" | "warning" | "fail", "failures": [human-readable failure strings], "warnings": [human-readable warning strings], } "status" is the non-zero-equivalent gate for CI: "ok" means no drift at all, "warning" means only benign drift (argument-level or extra/unexpected calls), and "fail" means a hard failure (a missing tool call, or a changed terminal action). If strict=True and status == "fail", raises BaselineDriftError instead of just returning the report. """ ref_tools = [s["tool"] for s in reference] cand_tools = [s["tool"] for s in candidate] # --- 1. Presence drift --------------------------------------------- missing_in_candidate = [t for t in ref_tools if t not in cand_tools] unexpected_in_candidate = [t for t in cand_tools if t not in ref_tools] # --- 2. Argument drift (for tool calls shared by both runs) --------- argument_drift = [] shared_len = min(len(reference), len(candidate)) for i in range(shared_len): ref_step, cand_step = reference[i], candidate[i] if ref_step["tool"] != cand_step["tool"]: # Position drifted entirely (a different tool ran here); the # presence-drift lists above already capture that story, so # there is nothing meaningful to argument-diff at this index. continue ref_args = ref_step.get("arguments", {}) cand_args = cand_step.get("arguments", {}) for key in sorted(set(ref_args) | set(cand_args)): ref_val = ref_args.get(key) cand_val = cand_args.get(key) if ref_val != cand_val: argument_drift.append( { "step": i + 1, "tool": ref_step["tool"], "arg": key, "reference_value": ref_val, "candidate_value": cand_val, } ) # --- 3. Terminal-action drift --------------------------------------- reference_terminal = ref_tools[-1] if ref_tools else None candidate_terminal = cand_tools[-1] if cand_tools else None terminal_changed = _tool_category(reference_terminal) != _tool_category( candidate_terminal ) failures = [] warnings = [] if missing_in_candidate: failures.append( "tool call(s) present in reference but missing from candidate: " f"{missing_in_candidate}" ) if unexpected_in_candidate: warnings.append( "tool call(s) present in candidate but not in reference: " f"{unexpected_in_candidate}" ) for d in argument_drift: warnings.append( f"step {d['step']} ({d['tool']}): argument '{d['arg']}' drifted " f"from {d['reference_value']!r} to {d['candidate_value']!r}" ) if terminal_changed: failures.append( "terminal action category changed: reference ended with " f"'{reference_terminal}', candidate ended with '{candidate_terminal}'" ) status = "fail" if failures else ("warning" if warnings else "ok") report = { "missing_in_candidate": missing_in_candidate, "unexpected_in_candidate": unexpected_in_candidate, "argument_drift": argument_drift, "terminal_action_changed": terminal_changed, "reference_terminal": reference_terminal, "candidate_terminal": candidate_terminal, "status": status, "failures": failures, "warnings": warnings, } if strict and status == "fail": raise BaselineDriftError(f"baseline drift check failed: {'; '.join(failures)}") return reportdef print_drift_report(report: dict) -> None: """Human-readable rendering of a diff_trajectories report.""" print(f"Status: {report['status'].upper()}") print( f"Terminal action: reference='{report['reference_terminal']}' " f"candidate='{report['candidate_terminal']}' " f"(changed={report['terminal_action_changed']})" ) if report["missing_in_candidate"]: print(f"\nMissing in candidate: {report['missing_in_candidate']}") if report["unexpected_in_candidate"]: print(f"Unexpected in candidate: {report['unexpected_in_candidate']}") if report["argument_drift"]: print(f"\nArgument-level drift ({len(report['argument_drift'])} field(s)):") for d in report["argument_drift"]: print( f" [step {d['step']}] {d['tool']}.{d['arg']}: " f"{d['reference_value']!r} -> {d['candidate_value']!r}" ) else: print("\nArgument-level drift: none") if report["failures"]: print("\nHARD FAILURES:") for f in report["failures"]: print(f" - {f}")# ---------------------------------------------------------------------------# Demo: reuse the refund scenario from trajectory_recorder.py, with the# candidate's refund amount drifted from the reference by one step# (84.00 -> 42.00), everything else identical.# ---------------------------------------------------------------------------def _build_reference_trajectory() -> list: return [ { "step": 1, "tool": "lookup_order", "arguments": {"order_id": "order_9F21"}, "result": {"id": "order_9F21", "status": "delivered", "total": 84.00}, "timestamp": 1, }, { "step": 2, "tool": "check_policy", "arguments": {"order_id": "order_9F21"}, "result": {"eligible": True, "reason": "delivered_within_30_days"}, "timestamp": 2, }, { "step": 3, "tool": "refund_api.create", "arguments": {"order_id": "order_9F21", "amount": 84.00}, "result": {"refund_id": "rf_001", "status": "succeeded", "amount": 84.00}, "timestamp": 3, }, { "step": 4, "tool": "send_email", "arguments": { "to": "customer@example.com", "body": "Your refund has been processed.", }, "result": {"sent": True}, "timestamp": 4, }, ]def _build_candidate_trajectory() -> list: candidate = json.loads(json.dumps(_build_reference_trajectory())) # deep copy # Simulate behavioral drift: the candidate model computed a different # refund amount (e.g. it silently applied a 50% partial-refund policy # it wasn't instructed to use). Everything else - the tool sequence, # the lookup, the policy check, the final email - is untouched. candidate[2]["arguments"]["amount"] = 42.00 candidate[2]["result"]["amount"] = 42.00 return candidateif __name__ == "__main__": reference = _build_reference_trajectory() candidate = _build_candidate_trajectory() report = diff_trajectories(reference, candidate) print("=== Baseline Drift Report: reference vs. candidate ===\n") print_drift_report(report) drifted_steps = {d["step"] for d in report["argument_drift"]} matched_steps = len(reference) - len(drifted_steps) print( f"\nSummary: {len(drifted_steps)} of {len(reference)} step(s) show " f"argument-level drift, {matched_steps} step(s) matched exactly " "(tool + arguments identical to the reference)." )
The differ walks two trajectory logs, a reference run and a candidate run, and reports three categories of divergence separately: which tools were called that weren't in the baseline (or vice versa), which arguments changed for a tool call both runs share, and whether the final action category changed even if the wording didn't. That last category is the one final-answer testing can never surface, and it's usually the one worth paging someone over. Treat small argument drift (a rephrased search query, a reordered list of retrieved chunks) as a warning threshold, and treat a changed or missing terminal action as a hard failure. Catching model version drift and other silent behavioral changes over time is really an extension of ordinary regression testing, applied to a system whose behavior is a probability distribution instead of a fixed function, and a baseline diff is the mechanism that makes agent regression testing possible instead of hand-wavy.
How Autonoma Verifies the Agent's Actual Action
Everything above happens against a simulated or recorded version of your agent: a trajectory log, a baseline you compare to, a shadow run you diff. That's the right layer for the agent's own decision-making, and it's where sampling temperature, model drift, and tool-call ordering actually live. It doesn't answer a related but different question: once the agent's chosen action reaches your real application, did the application actually do the thing the agent intended? A trajectory log can confirm the agent called refund_api.create with the right arguments. It can't confirm the refund actually posted, that the customer's account balance updated, or that the confirmation email went out, because none of that is visible from inside the agent's own trajectory.
That's the layer Autonoma runs. Instead of grading a transcript or diffing a log, Autonoma's Planner reads your application's actual code (routes, components, the flows an agent's actions would trigger) and plans test cases against them, an Execution Agent drives the real UI in an isolated PreviewKit environment, Autonoma's managed preview-environments layer, to carry out those cases, a GenerationReviewer classifies what happened (a real bug, an agent error, a test-plan mismatch), and a HealingAgent keeps the suite working as your app's UI changes, while a DiffsAgent maintains the test cases against every code diff so the suite doesn't silently rot as the product evolves. Applied to an AI agent's output, that means the behavioral baseline extends past the trajectory log and onto the live app: not "did the agent's log say it called the refund tool" but "did the refund actually appear in the account, correctly, in the UI a support rep or customer would actually see." The response looked plausible becomes the agent's action was actually correct, verified against the same running application your users touch, with the same per-PR maintenance discipline (via code diffs) that keeps the suite from rotting the moment your product changes.
That distinction matters most for exactly the failure mode this piece opened with. A silent failure is silent precisely because the text is convincing and the log can be well-formed while the underlying system state is wrong or absent. Autonoma is not a replacement for trajectory testing, retrieval checks, or a baseline differ; it's the layer that confirms the thing all of that testing is ultimately trying to protect, the actual state of the actual application, actually changed the way the agent claimed it would.
Canary and Staged Rollout for Agents
A baseline diff catches drift you can measure before deploying a new prompt, model version, or tool. It can't catch everything, because your baseline scenarios are necessarily a sample, not the full space of real user inputs. The remaining gap gets closed the way it's always been closed for any probabilistic system in production: a staged rollout, sized so a regression that slips past your offline suite reaches a small fraction of real traffic before it reaches all of it.
The pattern has three pieces. Percentage rollout routes a small slice of real traffic (5%, then 25%, then 100%) to the new agent version, while the rest keeps running the known-good version. Shadow runs go further: the new version processes a copy of real requests without its output ever reaching the user, so you get real-world trajectory data to diff against the baseline with zero user-facing risk, at the cost of double the inference spend for whatever traffic you shadow. And automatic rollback ties a reliability metric (baseline-diff pass rate, hard-failure rate on terminal actions, whatever your suite emits) to a threshold that, if breached during a canary stage, reroutes traffic back to the previous version without a human having to notice a dashboard first.
The point of a canary rollout isn't to prevent every regression. It's to guarantee that when one slips through, it reaches five percent of your users instead of all of them.
"""canary_gate.py================Reference implementation of a CANARY / STAGED ROLLOUT gate for AI agents.Trajectory recording (trajectory_recorder.py) catches an agent skipping arequired step. Baseline diffing (baseline_differ.py) catches an agent'sbehavior quietly drifting from a known-good reference. canary_gate.py isthe third leg: even with both of those checks passing in pre-production,you still want a staged, reversible rollout in production, becausereliability regressions can be probabilistic (an agent that silentlyfails 1 request in 10) and won't always show up in a small pre-prod testsuite.CanaryGate simulates routing an increasing percentage of productiontraffic through a new agent version across stages (e.g. 5% -> 25% ->100%), scoring each stage's reliability against a threshold, andautomatically halting the rollout and rolling back to the previousknown-good version the moment a stage's score drops below the configuredtolerance, without waiting for a human to notice.Run this file directly (`python canary_gate.py`) to see a simulatedrollout where the 25% stage's reliability regresses below threshold, andthe gate halts/rolls back automatically instead of continuing to 100%."""from __future__ import annotationsfrom dataclasses import dataclass, fieldfrom typing import Callable, Optional@dataclassclass StageResult: stage_pct: int sample_size: int passes: int score: float threshold: float passed: bool@dataclassclass CanaryGate: """Drives a staged canary rollout of a new agent version. stages: percentage of traffic routed through the candidate version at each stage, in increasing order (e.g. [5, 25, 100]). threshold: minimum acceptable reliability score (0.0-1.0) for a stage to pass. A stage scoring below this halts the rollout. sample_size: number of simulated trajectory comparisons to run per stage. Larger samples make a real regression more reliably detectable versus ordinary noise. """ stages: list threshold: float = 0.90 sample_size: int = 20 history: list = field(default_factory=list, repr=False) halted: bool = field(default=False, repr=False) halted_at_stage: Optional[int] = field(default=None, repr=False) def _score_stage(self, stage_pct: int, sample_fn: Callable[[], bool]) -> StageResult: """Runs `sample_size` simulated trajectory checks for this stage. `sample_fn` returns True for a simulated request whose trajectory matches the reliability baseline (in a real system: the output of diff_trajectories(reference, candidate) reporting status == "ok"), and False for one that doesn't (a regression, i.e. status == "fail"). """ passes = sum(1 for _ in range(self.sample_size) if sample_fn()) score = passes / self.sample_size return StageResult( stage_pct=stage_pct, sample_size=self.sample_size, passes=passes, score=score, threshold=self.threshold, passed=score >= self.threshold, ) def run(self, run_trajectory: Callable[[int], bool]) -> list: """Executes the staged rollout. `run_trajectory(stage_pct)` is called once per simulated request and returns True/False for that request's trajectory check at the given stage percentage. The rollout stops at the first stage whose score falls below `threshold` and rolls back to the previous known-good version instead of advancing to the next stage. """ self.history = [] self.halted = False self.halted_at_stage = None for stage_pct in self.stages: result = self._score_stage(stage_pct, lambda: run_trajectory(stage_pct)) self.history.append(result) if not result.passed: # AUTOMATIC ROLLBACK: reliability regressed at this stage, # so we halt immediately rather than ramping up exposure # to a version that is already failing at a small # percentage of traffic. Production traffic falls back to # the previous known-good version. self.halted = True self.halted_at_stage = stage_pct break # Stage cleared: proceed to the next, larger traffic slice. return self.history# ---------------------------------------------------------------------------# Demo: a rollout where the 25% stage regresses below threshold# ---------------------------------------------------------------------------def _simulate_trajectory_check(stage_pct: int, call_index: int) -> bool: """Deterministic stand-in for "does this simulated request's trajectory match the reliability baseline?" In a production version of this gate, this would run a real request through the candidate agent, record its trajectory with TrajectoryRecorder, and call diff_trajectories(reference, candidate) per request, treating status == "ok" (or "warning") as a pass and status == "fail" as a regression. Here we use a fixed, reproducible pattern per stage so the demo output is stable across runs. """ if stage_pct == 5: # The 5% canary stage behaves exactly like the reference version. return True if stage_pct == 25: # A regression only shows up once a wider slice of real traffic # variety hits the candidate: 3 out of every 10 requests fail. return (call_index % 10) < 7 # The 100% stage would be healthy again, but the gate halts before # this is ever reached because the 25% stage fails below threshold. return Truedef main() -> None: gate = CanaryGate(stages=[5, 25, 100], threshold=0.90, sample_size=20) counters = {5: 0, 25: 0, 100: 0} def run_trajectory(stage_pct: int) -> bool: counters[stage_pct] += 1 return _simulate_trajectory_check(stage_pct, counters[stage_pct] - 1) print("=== Canary Rollout Simulation ===") print( f"Stages: {gate.stages} | threshold: {gate.threshold} | " f"sample size per stage: {gate.sample_size}\n" ) gate.run(run_trajectory) for result in gate.history: status = "PASS" if result.passed else "FAIL" print( f"[stage {result.stage_pct:>3}% traffic] score={result.score:.2f} " f"({result.passes}/{result.sample_size} passed) " f"threshold={result.threshold:.2f} -> {status}" ) if gate.halted: print( f"\nROLLOUT HALTED at stage {gate.halted_at_stage}% traffic: " "reliability score fell below threshold." ) print( "Automatic rollback: traffic routed back to the previous " "known-good version." ) remaining = [s for s in gate.stages if s > gate.halted_at_stage] if remaining: print(f"Stages never reached: {remaining}") else: print("\nRollout completed successfully: candidate promoted to 100% traffic.")if __name__ == "__main__": main()
The gate above simulates staged traffic across rollout stages, scores each stage against the reliability metrics your baseline differ produces, and halts the rollout the moment a stage's score drops below a configured tolerance, rolling remaining traffic back to the previous version automatically. Wire the same threshold logic into whatever deploy pipeline promotes your agent, and a bad model update or a broken prompt edit gets contained to a small percentage of sessions instead of discovered from a support queue.
Each layer catches what the one below it structurally cannot: a final-answer check can't see a skipped tool call, and a trajectory check alone can't see drift across weeks or a regression that only shows up in shadow traffic.
Catching Confident-Wrong Before It Reaches a User
None of this is about distrusting agents in the abstract. It's about the specific, narrow failure mode where an agent is wrong in a way that looks exactly like being right, and building the layers that catch it before a user does. Final-answer checks catch obviously bad output. Step-level and trajectory assertions catch a skipped or misordered tool call hiding behind a well-worded response. A behavioral baseline catches the same scenario quietly behaving differently three weeks from now. A canary rollout catches the regression your baseline scenarios didn't happen to cover. And a live-application testing layer like Autonoma catches the case where every log looked fine and the refund still never posted.
Build the layers in that order, cheapest and fastest first. A trajectory recorder and a handful of step-level assertions take an afternoon and immediately catch the class of bug that final-answer grading is structurally unable to see. A behavioral baseline is worth adding the moment you have more than one agent version to compare, which is to say almost immediately. Canary rollout and shadow runs earn their operational overhead once real user traffic is on the line. None of the layers replace each other; each one is there because the layer below it has a specific, well-understood blind spot, and the cost of skipping a layer is a confident-wrong output that nobody notices until a customer does.
Everything here is scoped to a single trajectory: one run, one baseline, one canary decision. The layer above it is conversational: does the agent still reach the right outcome across several turns of a user pushing back or a tool call failing mid-conversation. Agent simulation testing covers that three-actor pattern (user simulator, agent, judge) in raw Python, and it's worth building once the single-trajectory suite above is solid. For the full picture of where reliability testing fits alongside the rest of an agent's test surface, how to test an AI agent end to end walks through the layers this piece assumes.
Frequently Asked Questions
AI agent reliability testing is the pre-production practice of verifying that an LLM agent takes the correct sequence of actions, not just that its final response reads well, across the natural non-determinism introduced by sampling temperature, model version updates, tool-call ordering, retrieval variance, and context-window truncation. It typically involves trajectory-level assertions, a recorded behavioral baseline to diff new runs against, and a staged rollout to contain regressions the offline suite missed.
A final-answer check only reads the text an agent produced. It can't tell whether the tool calls that were supposed to happen actually happened, in the right order, with the right arguments. An agent can generate a confident, well-worded response while skipping or misordering the underlying action, and a final-answer grader has no visibility into that gap because it never looks past the last message.
Five sources account for most of it: sampling temperature (any temperature above zero draws from a probability distribution over tokens), silent model version updates behind the same API alias, tool-call ordering when multiple valid tools are available for a step, retrieval variance from vector store updates or embedding differences, and context-window truncation dropping earlier instructions as a conversation grows.
Record each run as a structured, append-only trajectory log keyed by scenario and timestamp: the exact input (prompt, history, retrieved context), every tool call with full arguments and raw results, every point where the agent chose between valid next steps, and the final action taken as a field separate from the final text generated. That log is what a baseline differ or a step-level assertion runs against.
A canary rollout routes a small percentage of real traffic to a new agent version while most traffic keeps using the known-good version, optionally combined with shadow runs that process a copy of real requests without surfacing the output to users. A reliability metric (baseline-diff pass rate or hard-failure rate on terminal actions) is monitored per stage, and an automatic rollback reroutes traffic back if that metric regresses past a threshold.
