flowchart LR
CLIENT["Client with virtual key"] --> GATE["Request plane<br/>identity - budget - route - deadline"]
GATE --> AGENT(["Agent runtime"])
AGENT --> EFFECT["Effect gateway"]
CONTROL["Control plane<br/>catalog - policy - release bundle"] -->|"atomic version pointer"| GATE
CONTROL -->|"pinned model and tools"| AGENT
DATA[("Data plane<br/>corpus - cache - checkpoint<br/>effect ledger - receipt")]
AGENT <--> DATA
EFFECT <--> DATA
GATE -. "usage + decision" .-> EVIDENCE[("Evidence plane<br/>traces - evals - lineage - audit")]
AGENT -. "trajectory" .-> EVIDENCE
EFFECT -. "authoritative receipt" .-> EVIDENCE
26 The Production Platform: Gateways, Cost, Durable Execution, and Delivery
At 16:40 on a Friday, three product teams discover that their agents share one provider account. One team’s fallback loop retries a rate-limited request against two models; another prepends a large corpus on every turn; a third runs an evaluation sweep. The account burns twenty-three thousand cost units before anyone can attribute the traffic to a tenant, a release, or a task, and the only way to stop the runaway loop is to disable the single provider key — which takes down every healthy application with it. Nothing here is a missing dashboard. The organization has no enforcement point between what an application intends and the scarce model capacity it consumes.
This chapter builds that enforcement point and measures it, one piece at a time, entirely in memory. The parts are (i) a gateway that authenticates a caller, admits a journey against a budget, and routes or cascades over stub backends whose cost, quality, and latency we compare with real numbers; (ii) a cost model that prices a whole journey, computes when a cache pays back, applies a batch discount, and reproduces a loop-cost blow-up before capping it; (iii) reliability wrappers — a retry budget that exhausts, a propagated deadline, an idempotency key that stops a double charge, and a hedge whose tail-latency win we measure; (iv) a durable-execution engine with deterministic replay, a crash resumed mid-task, and a saga that compensates; (v) a release manifest we diff by content hash; and (vi) a release argument — one property-based test and an eval-gated canary whose gate we watch flip. A journey is the whole user outcome, including model calls, retrieval, tools, retries, and review; one API request is only a hop within it. The raw agent loop this platform wraps is the one from Chapter 16; the offline evaluation it gates is the one from Chapter 22; the final authorization at the effect boundary remains Chapter 24’s. Everything runs offline and deterministically, so every number on these pages is produced when the book is built. The stub backends are visible scripts, not models — the point is to exercise the control machinery, which is identical whether the thing behind it is a stub or a frontier model.
26.1 The platform in four planes
The Friday incident is a boundary failure, so the first design move is to name the boundaries. A platform is easier to reason about — and to keep from becoming one tangled cluster — when we separate four planes by the kind of authority each holds. The request plane acts synchronously on one journey: it authenticates, admits, routes, propagates a deadline, and invokes effects. The data plane stores state that outlives a request: prompts, corpora, checkpoints, effect records, and cache entries. The control plane publishes immutable configuration ahead of traffic: catalogs, policies, budgets, and release bundles. The evidence plane records what happened: traces, usage, eval results, and audit exports. The distinction is about responsibility, not necessarily four deployments.
The one rule that keeps the planes honest is the direction of authority. The control plane publishes downward into the request path; the evidence plane records what the request path did but must never become an authority the hot path waits on. A trace collector should not decide whether a refund is authorized, and its outage should not block safe service. Figure 26.1 locates authority and evidence so that a later trace can be interpreted against the exact rules a request used.
Where the planes physically run is a second decision, and it follows constraints rather than fashion. A managed model API minimizes serving work and makes model substitution fast, at the cost of provider availability, data-handling, and quota dependencies. Self-hosting buys control over weights, batching, and isolation while adding serving expertise and capacity risk. On-premises or sovereign deployments satisfy locality and procurement constraints but lose elasticity. Edge or hybrid systems cut network latency and keep an offline path, but memory, power, and version convergence become first-class limits. The decision is per workload and data class, and the useful way to see it is as a position on two axes — how much control and residency a workload demands, against how much elasticity and fast substitution it needs. Figure 26.2 places the five topologies and annotates a single organization running three different workloads.
Show the code that draws this figure
import matplotlib.pyplot as plt
topologies = {
"Managed API": (0.15, 0.95),
"Regional managed": (0.42, 0.78),
"Self-hosted": (0.66, 0.50),
"On-prem / sovereign": (0.90, 0.24),
"Edge / hybrid": (0.78, 0.14),
}
workloads = {
"public drafting": ("Managed API", (8, 10)),
"tenant-private support": ("Regional managed", (10, -6)),
"high-volume classification": ("Self-hosted", (8, -16)),
}
fig, ax = plt.subplots(figsize=(6.6, 4.0))
for name, (x, y) in topologies.items():
ax.scatter(x, y, s=90, color="#315b8a", zorder=3)
ax.annotate(name, (x, y), textcoords="offset points", xytext=(8, 6), fontsize=9)
for label, (topo, offset) in workloads.items():
x, y = topologies[topo]
ax.annotate(label, (x, y), textcoords="offset points", xytext=offset,
fontsize=8, color="#b13f3f",
arrowprops=dict(arrowstyle="->", color="#b13f3f", lw=0.8))
ax.set_xlabel("control and residency need ->")
ax.set_ylabel("elasticity and substitution speed ->")
ax.set_xlim(0, 1.05)
ax.set_ylim(0, 1.1)
ax.spines[["top", "right"]].set_visible(False)
ax.grid(alpha=0.2)
fig.tight_layout()
plt.show()
Record residency, latency, availability, staffing, customization, isolation, update cadence, and exit cost for each choice; Chapter 27 turns residency and retention into operated controls, and Chapter 28 uses the staffing and substitution evidence for build-versus-buy.
NoteLandscape 2026 (dated)
Verify live: 2026-07-20. Current AI-gateway options span dedicated projects and AI extensions to general API gateways; current durable-execution choices include Temporal, Restate, and DBOS. Managed accelerator endpoints, serverless GPU products, and sovereign regions change frequently. Select by the verified mechanism — identity propagation, policy enforcement, replay semantics, storage control, quotas, and failure behavior — not by the category label. Appendix C owns the current registry.
26.2 One gateway: identity, admission, routing, and cascades
We are ready to build the request plane and measure it. A gateway does six things on the way in — authenticate, admit, route, propagate a deadline, meter, and emit evidence — and the first correctness rule is about trust. A client header saying tenant=bank-a is data, not identity. The gateway issues a virtual key and maps it server-side to a tenant, its scopes, and its budget; it never forwards its own broad provider credential to an application. We start with the identity a request carries and the price schedule that settles it.
# @save
from __future__ import annotations
import hashlib
import json
import math
import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable
@dataclass(frozen=True)
class RequestContext:
"""The trusted identity a journey carries, derived only from a virtual key.
A gateway builds this from a server-side key lookup, never from a
model-controlled or client-declared field. Everything downstream —
budgets, routing scopes, effect authorization — reads identity from here,
which is why the tenant and scopes are fixed at admission and immutable.
"""
request_id: str
tenant: str
scopes: frozenset[str]
deadline_s: float
@dataclass(frozen=True)
class Usage:
"""Metered tokens and per-call charges for one model call in a journey.
The cost of a journey is more than input and output tokens: it includes
per-call tool charges and, for reviewed journeys, a share of human review.
Carrying all four terms here is what lets ``breakdown`` show where a
journey's money actually goes instead of hiding it behind a token count.
"""
input_tokens: int
output_tokens: int
input_per_million: float
output_per_million: float
tool_units: float = 0.0
review_units: float = 0.0
@property
def cost(self) -> float:
"""Return the total cost of this call in currency units."""
tokens = (
self.input_tokens * self.input_per_million
+ self.output_tokens * self.output_per_million
) / 1_000_000
return tokens + self.tool_units + self.review_units
def breakdown(self) -> dict[str, float]:
"""Return per-component cost so a journey's spend can be attributed."""
return {
"input": self.input_tokens * self.input_per_million / 1_000_000,
"output": self.output_tokens * self.output_per_million / 1_000_000,
"tool": self.tool_units,
"review": self.review_units,
}Admission happens before any expensive work, and it must reserve an estimate for the whole journey, not just the first call — a request that can take eight agent steps and fan out to four workers has a very different maximum exposure than a single extraction. The gateway holds a per-tenant budget and rejects a journey whose estimate would exceed it.
# @save
class BudgetError(RuntimeError):
"""Raised when a whole journey has no remaining spend authority."""
@dataclass
class Deployment:
"""A model deployment the gateway may route to, with its cost per call."""
name: str
capabilities: frozenset[str]
cost_per_call: float
healthy: bool = True
class Gateway:
"""Authenticate, admit, route, and settle one agent journey.
The gateway is the single enforcement point the Friday incident lacked:
identity comes from a virtual key, a journey above its tenant budget is
rejected before any call, routing considers only deployments that satisfy
the requested capability and are healthy, and settlement charges realized
usage back to the tenant. It deliberately holds provider credentials so an
application never does.
"""
def __init__(
self,
keys: dict[str, tuple[str, frozenset[str]]],
limits: dict[str, float],
deployments: list[Deployment],
) -> None:
self.keys = keys
self.limits = limits
self.deployments = deployments
self.spend = {tenant: 0.0 for tenant in limits}
def authenticate(
self, virtual_key: str, request_id: str, deadline_s: float
) -> RequestContext:
"""Map a virtual key to a trusted identity, or refuse an unknown key.
Args:
virtual_key: The opaque key presented by the caller.
request_id: An id used to correlate evidence for this request.
deadline_s: The absolute time budget the journey starts with.
Returns:
The ``RequestContext`` whose tenant and scopes come from the
server-side key table, never from a client-supplied field.
Raises:
PermissionError: If the key is not in the table.
"""
if virtual_key not in self.keys:
raise PermissionError("unknown virtual key")
tenant, scopes = self.keys[virtual_key]
return RequestContext(request_id, tenant, scopes, deadline_s)
def admit(self, context: RequestContext, estimated_cost: float) -> None:
"""Reject a journey whose estimate would exceed the tenant budget.
Args:
context: The authenticated identity for this journey.
estimated_cost: The whole-journey estimate to reserve.
Raises:
BudgetError: If reserving the estimate would exceed the limit.
"""
limit = self.limits[context.tenant]
if self.spend[context.tenant] + estimated_cost > limit:
raise BudgetError("tenant journey budget exhausted")
def route(self, context: RequestContext, capability: str) -> Deployment:
"""Select the least-cost healthy deployment satisfying a capability.
Args:
context: The authenticated identity (its scopes gate capabilities).
capability: The logical capability the journey needs, e.g. "chat".
Returns:
The cheapest eligible ``Deployment``.
Raises:
LookupError: If no healthy deployment offers the capability.
"""
candidates = [
d for d in self.deployments if capability in d.capabilities and d.healthy
]
if not candidates:
raise LookupError("no healthy deployment satisfies the capability")
return min(candidates, key=lambda d: d.cost_per_call)
def settle(self, context: RequestContext, usage: Usage) -> float:
"""Charge realized usage to the tenant and return the new total spend."""
self.spend[context.tenant] += usage.cost
return self.spend[context.tenant]Watch one journey pass through: a virtual key becomes an identity, an estimate is admitted, a capability resolves to the cheaper of two deployments, and metered usage settles against the tenant.
gateway = Gateway(
keys={"vk-acme": ("acme", frozenset({"chat"}))},
limits={"acme": 0.05},
deployments=[
Deployment("small", frozenset({"chat"}), cost_per_call=1.0),
Deployment("large", frozenset({"chat", "vision"}), cost_per_call=5.0),
],
)
context = gateway.authenticate("vk-acme", "req-7", deadline_s=30.0)
usage = Usage(20_000, 2_000, input_per_million=0.5, output_per_million=1.5)
gateway.admit(context, usage.cost)
route = gateway.route(context, "chat")
print(f"tenant={context.tenant} routed={route.name} "
f"journey_cost={usage.cost:.4f} tenant_spend={gateway.settle(context, usage):.4f}")tenant=acme routed=small journey_cost=0.0130 tenant_spend=0.0130The identity came from the key, not the payload; the over-budget guard sits before routing; and route picked small because it was the cheapest deployment that could satisfy chat. That last choice — pick the cheapest eligible deployment — is only the simplest of four different operations, and confusing them is a common production bug. Routing chooses a deployment before a call. A cascade tries a cheap deployment first and escalates after a grader judges the result insufficient. A fallback switches provider after an operational failure, to preserve availability. A hedge issues a redundant attempt before the first completes, to cut tail latency. They differ in when they fire and what they optimize; here we measure the first two, because their cost-quality tradeoff is the one a cost review always asks about.
To measure anything we need something to serve. A real model is nondeterministic and needs a GPU, so we use a stub backend whose accuracy is a visible rule: a backend answers a task correctly when its strength meets the task’s difficulty. That is not a model — it is a scriptable stand-in that lets the control logic (route versus cascade) be exercised and priced offline.
# @save
@dataclass(frozen=True)
class Backend:
"""A stub model deployment with a fixed cost, quality, and latency profile.
The backend answers a task correctly when its ``strength`` meets the
task's difficulty. This deterministic rule stands in for a real model's
accuracy curve so routing and cascading can be compared and priced without
a network or an accelerator; only the numbers change with a real model.
"""
name: str
strength: float
cost_per_call: float
latency_ms: float
def answer(self, difficulty: float) -> bool:
"""Return whether this backend answers a task of the given difficulty."""
return self.strength >= difficultyThree strategies run over the same workload. run_single sends every task to one backend. run_router asks a cheap classifier to predict difficulty — imperfectly, with noise — and routes hard-looking tasks to the strong backend. run_cascade always tries the cheap backend, then escalates to the strong one whenever a grader judges the cheap answer insufficient; the grader is an oracle here, save for a small false-accept rate that keeps it honest.
# @save
@dataclass(frozen=True)
class WorkloadResult:
"""Aggregate cost, quality, and latency for one routing strategy."""
strategy: str
accuracy: float
total_cost: float
mean_latency_ms: float
escalations: int
def run_single(backend: Backend, difficulties: list[float]) -> WorkloadResult:
"""Serve every task with one backend and aggregate the outcome.
Args:
backend: The single deployment used for the whole workload.
difficulties: One difficulty per task in the workload.
Returns:
The strategy's accuracy, total cost, mean latency, and (zero)
escalations.
"""
correct = sum(backend.answer(d) for d in difficulties)
n = len(difficulties)
return WorkloadResult(
f"single:{backend.name}", correct / n,
backend.cost_per_call * n, backend.latency_ms, 0,
)
def run_router(
small: Backend, large: Backend, difficulties: list[float],
noise: float, seed: int,
) -> WorkloadResult:
"""Route each task by a noisy difficulty estimate, then serve it once.
A cheap classifier estimates difficulty as the true value plus Gaussian
noise; tasks estimated above the small backend's strength go to the large
backend. Misestimates are the router's characteristic failure: an easy
task sent to the large model wastes money, a hard task kept on the small
model loses quality.
Args:
small: The cheap, weaker backend.
large: The expensive, stronger backend.
difficulties: One difficulty per task.
noise: Standard deviation of the classifier's estimation error.
seed: Seed for the deterministic noise stream.
Returns:
The routed workload's aggregate result.
"""
rng = random.Random(seed)
correct = cost = latency = escalations = 0
for d in difficulties:
estimate = d + rng.gauss(0.0, noise)
chosen = large if estimate > small.strength else small
escalations += chosen is large
correct += chosen.answer(d)
cost += chosen.cost_per_call
latency += chosen.latency_ms
n = len(difficulties)
return WorkloadResult("router", correct / n, cost, latency / n, escalations)
def run_cascade(
small: Backend, large: Backend, difficulties: list[float],
false_accept: float, seed: int,
) -> WorkloadResult:
"""Try the cheap backend first, escalate when a grader rejects its answer.
Every task pays the small backend. A grader (an oracle, except that it
wrongly accepts a wrong answer with probability ``false_accept``) decides
whether to escalate; escalated tasks also pay the large backend and its
latency. The cascade's promise is large-backend quality at a fraction of
its cost, bought with higher latency on the escalated tail.
Args:
small: The cheap backend tried first.
large: The expensive backend used on escalation.
difficulties: One difficulty per task.
false_accept: Probability the grader accepts a wrong cheap answer.
seed: Seed for the deterministic grader stream.
Returns:
The cascade's aggregate result.
"""
rng = random.Random(seed)
correct = cost = latency = escalations = 0
for d in difficulties:
cost += small.cost_per_call
small_ok = small.answer(d)
accept = small_ok or rng.random() < false_accept
if accept:
correct += small_ok
latency += small.latency_ms
else:
escalations += 1
correct += large.answer(d)
cost += large.cost_per_call
latency += small.latency_ms + large.latency_ms
n = len(difficulties)
return WorkloadResult("cascade", correct / n, cost, latency / n, escalations)Now the measurement. Four hundred tasks with difficulty spread across the range, a weak cheap backend and a strong expensive one, and we print all five strategies side by side.
small = Backend("small", strength=0.55, cost_per_call=1.0, latency_ms=200.0)
large = Backend("large", strength=0.97, cost_per_call=6.0, latency_ms=900.0)
tasks = [i / 400 for i in range(400)]
results = [
run_single(small, tasks),
run_single(large, tasks),
run_router(small, large, tasks, noise=0.15, seed=1),
run_cascade(small, large, tasks, false_accept=0.05, seed=2),
]
print(f"{'strategy':16} {'accuracy':>9} {'cost':>8} {'mean_ms':>9} {'escalations':>12}")
for r in results:
print(f"{r.strategy:16} {r.accuracy:9.3f} {r.total_cost:8.0f} "
f"{r.mean_latency_ms:9.0f} {r.escalations:12d}")strategy accuracy cost mean_ms escalations
single:small 0.552 400 200 0
single:large 0.973 2400 900 0
router 0.932 1360 536 192
cascade 0.945 1396 574 166Read the table as a Pareto frontier. The cheap single backend is fast and cheap and wrong on the hard half; the strong single backend is accurate but pays full price and full latency on every task, including the easy ones it never needed. The router lands in between on all three axes. The cascade is the interesting one: it reaches nearly the strong backend’s accuracy while paying the strong price only on the escalated fraction — cheaper than serving everything to the large model, at the cost of higher latency on that escalated tail, because those tasks pay both backends. Figure 26.3 plots the three axes so the tradeoff is visible at a glance.
Show the code that draws this figure
labels = [r.strategy.replace("single:", "") for r in results]
colors = ["#b7791f", "#b13f3f", "#557a3c", "#315b8a"]
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(8.4, 3.4))
for r, label, color in zip(results, labels, colors):
ax1.scatter(r.total_cost, r.accuracy, s=90, color=color, zorder=3)
ax1.annotate(label, (r.total_cost, r.accuracy),
textcoords="offset points", xytext=(6, 5), fontsize=8)
ax1.set_xlabel("total cost (units)")
ax1.set_ylabel("accuracy")
ax1.set_title("cost vs quality")
ax2.bar(labels, [r.mean_latency_ms for r in results], color=colors, width=0.6)
ax2.set_ylabel("mean latency (ms)")
ax2.set_title("latency")
ax2.tick_params(axis="x", rotation=20)
for ax in (ax1, ax2):
ax.spines[["top", "right"]].set_visible(False)
ax.grid(alpha=0.2)
fig.tight_layout()
plt.show()
Two operational rules fall out of the distinction. First, resolve a product alias such as support-answer through capability filters — schema support, context length, region, data class, release, policy — before optimizing cost or latency; a cheap model that cannot satisfy the contract is not a candidate, however cheap. Second, do not provider-shop around a policy refusal: if provider A refuses a harmful request, sending it to B until one complies converts resilience into a security bypass. Fallback is for overload and transient faults while time remains; authentication, invalid-request, and policy errors are terminal. Record fallback rate and escalation rate separately — rising fallback points at provider health, rising escalation at task or routing quality.
26.3 Price the journey
Per-token list price is an input to cost, not the answer. The unit that matters is cost per successful business outcome, and to compute it we attribute every charge to one journey identity. For a journey with calls j = 1, \ldots, m, let x_j and y_j be metered input and output tokens at unit prices p^{(x)}_j and p^{(y)}_j, let u_j be other per-call tool charges, and let h be the journey’s share of human review. The journey cost is
C_{\text{journey}} = \sum_{j=1}^{m}\left(x_j p^{(x)}_j + y_j p^{(y)}_j + u_j\right) + h. \tag{26.1}
The Usage type from the last section already carries all four terms, so summing a journey is direct. The reason to keep the terms apart is that at realistic scales the token cost is often not where the money goes — a single human review or a paid tool call can dwarf a few thousand tokens. Figure 26.4 breaks one journey down.
Show the code that draws this figure
journey = Usage(20_000, 2_000, input_per_million=0.5, output_per_million=1.5,
tool_units=0.004, review_units=0.010)
parts = journey.breakdown()
fig, ax = plt.subplots(figsize=(6.4, 3.2))
left = 0.0
palette = {"input": "#315b8a", "output": "#557a3c", "tool": "#b7791f", "review": "#b13f3f"}
for name, value in parts.items():
ax.barh("journey", value, left=left, color=palette[name], label=f"{name} ({value:.3f})")
left += value
ax.set_xlabel(f"cost units (total {journey.cost:.3f})")
ax.legend(ncol=4, fontsize=8, loc="upper center", bbox_to_anchor=(0.5, -0.25))
ax.spines[["top", "right"]].set_visible(False)
fig.tight_layout()
plt.show()
Caching is the first lever, and it has a clean break-even. Suppose an uncached prefix costs c, writing it to the cache costs wc, each cached read costs dc, and the prefix is reused r times after the write. Uncached, the total is (r+1)c; cached, it is (w + rd)c. Caching pays back when
r \ge \frac{w-1}{1-d}. \tag{26.2}
# @save
def cache_breakeven(write_multiplier: float, read_multiplier: float) -> float:
"""Return the minimum reuses for a prefix cache to pay for its write.
Solves @eq-ch26-cache: a cache whose write costs ``write_multiplier`` times
an uncached call and whose reads cost ``read_multiplier`` times an uncached
call breaks even after ``(w - 1) / (1 - d)`` reuses.
Args:
write_multiplier: Cost of writing the cache, relative to one call.
read_multiplier: Cost of a cached read, relative to one call.
Returns:
The break-even reuse count (a real number; round up for whole reuses).
"""
return (write_multiplier - 1) / (1 - read_multiplier)
def cached_cost(reuses: int, write_multiplier: float, read_multiplier: float) -> float:
"""Return the cost of one cache write plus ``reuses`` discounted reads."""
return write_multiplier + reuses * read_multiplier
def batch_saving(synchronous_cost: float, discount: float) -> float:
"""Return the cost of a deferrable workload run through a batch API.
Providers commonly price asynchronous, latency-tolerant work — nightly
evaluations, bulk classification, backfills — at a discount to the
synchronous rate. Moving that work to a batch queue is often the single
largest cost lever for an agent platform's offline jobs.
Args:
synchronous_cost: What the workload would cost at the live rate.
discount: Fractional discount the batch tier offers, in [0, 1].
Returns:
The batch-tier cost of the same workload.
"""
return synchronous_cost * (1 - discount)With a write premium of 1.25 and a read at a tenth of an uncached call, the break-even is under one reuse — a prefix reused even once already pays for itself.
w, d = 1.25, 0.10
print(f"break-even reuses = {cache_breakeven(w, d):.3f} -> one reuse suffices")
print(f"uncached 2 calls = 2.000 cached (write + 1 read) = {cached_cost(1, w, d):.3f}")
sweep_cost = 4.0
print(f"nightly eval sweep: sync {sweep_cost:.2f} -> batch (50% off) "
f"{batch_saving(sweep_cost, 0.5):.2f}")break-even reuses = 0.278 -> one reuse suffices
uncached 2 calls = 2.000 cached (write + 1 read) = 1.350
nightly eval sweep: sync 4.00 -> batch (50% off) 2.00The batch line is the other easy lever: work that tolerates latency — evaluation sweeps, bulk classification, backfills — belongs on a discounted asynchronous tier, halving its cost in this schedule. A semantic cache has no such price-only optimum, because lowering its similarity threshold raises the hit rate and the false-reuse rate; tune it against correctness and privacy slices, not savings alone.
Now the pathology from the Friday incident. Agent loops multiply cost in two compounding ways. First, retries: if an attempt independently needs another attempt with probability q < 1, the expected attempt count is (1-q)^{-1}, so at q = 0.2 offered load already grows by a quarter — and independent retry layers multiply, so a client retrying three times, a gateway twice, and an SDK three times can turn one intent into eighteen attempts.
# @save
def retry_multiplier(q: float) -> float:
"""Return the expected attempts when each independently retries with prob q.
Args:
q: Per-attempt probability that another attempt is needed, in [0, 1).
Returns:
The geometric-series expectation ``1 / (1 - q)``.
"""
return 1.0 / (1.0 - q)
def goodput(throughput: float, qualifying_fraction: float) -> float:
"""Return goodput: the rate of completions that satisfy the task contract."""
return throughput * qualifying_fractionThe second multiplier is the transcript itself: because every turn resends the whole conversation, a loop that keeps retrying pays a growing per-turn cost. We reproduce the blow-up by simulating many journeys that retry a soft failure until they succeed, each turn costing more than the last, and then we cap it.
# @save
def simulate_agent_loop(
rng: random.Random, p_success: float, per_turn_tokens: int, max_turns: int
) -> tuple[int, int]:
"""Simulate one retry-until-success journey with a growing transcript.
Each turn resends the whole transcript, so turn ``t`` costs
``per_turn_tokens * t``; the journey ends on the first Bernoulli success or
when ``max_turns`` caps it. The uncapped tail of this distribution is the
denial-of-wallet pathology a platform must bound.
Args:
rng: Seeded random source, so the simulation is reproducible.
p_success: Per-turn probability the journey completes.
per_turn_tokens: Base token cost of the first turn.
max_turns: Hard cap on turns; the lever that bounds the tail.
Returns:
A tuple of (total tokens spent, turns taken).
"""
total = 0
for turn in range(1, max_turns + 1):
total += per_turn_tokens * turn
if rng.random() < p_success:
return total, turn
return total, max_turnsdef spend_distribution(max_turns: int) -> list[int]:
rng = random.Random(7)
return sorted(simulate_agent_loop(rng, 0.5, 500, max_turns)[0]
for _ in range(5000))
for cap, label in [(50, "effectively uncapped"), (4, "capped at 4 turns")]:
spend = spend_distribution(cap)
p99 = spend[int(0.99 * len(spend))]
print(f"{label:20} mean={sum(spend)/len(spend):7.0f} "
f"p99={p99:6d} max={spend[-1]:6d} tokens")effectively uncapped mean= 1945 p99= 14000 max= 39000 tokens
capped at 4 turns mean= 1616 p99= 5000 max= 5000 tokensThe uncapped run has a fat tail: most journeys finish in one or two turns, but a few run long enough to spend an order of magnitude more, and it is those few that drain a shared account on a Friday afternoon. Capping turns leaves the common case untouched and simply amputates the tail. The lesson generalizes: set step, token, time, and concurrency budgets at the harness, and enforce the currency reservation at the gateway. Figure 26.5 puts the two cost curves — cache break-even and retry amplification — side by side.
Show the code that draws this figure
reuses = list(range(0, 11))
savings = [(r + 1) - cached_cost(r, 1.25, 0.10) for r in reuses]
qs = [i / 100 for i in range(0, 91)]
mult = [retry_multiplier(q) for q in qs]
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(8.4, 3.4))
ax1.axhline(0, color="#666666", lw=1)
ax1.plot(reuses, savings, color="#315b8a", lw=2, marker="o", ms=4)
ax1.set(xlabel="cache reuses", ylabel="savings (call-cost units)",
title="prefix-cache break-even")
ax2.plot(qs, mult, color="#b13f3f", lw=2)
ax2.scatter([0.2], [retry_multiplier(0.2)], color="#b13f3f", zorder=3)
ax2.annotate("q=0.2 -> 1.25x", (0.2, 1.25), textcoords="offset points",
xytext=(6, 8), fontsize=8)
ax2.set(xlabel="per-attempt retry probability q", ylabel="offered-load multiplier",
title="retry amplification")
for ax in (ax1, ax2):
ax.spines[["top", "right"]].set_visible(False)
ax.grid(alpha=0.2)
fig.tight_layout()
plt.show()
Capacity is governed by the tightest resource, and the honest capacity metric is goodput — the rate of completions that satisfy the task and service contract, G = Qs for raw throughput Q and qualifying fraction s — because a faster deployment that emits more invalid outputs can serve fewer good ones. To bound offered load on the tightest dimension we use a token bucket keyed on that dimension, so one enormous prompt cannot hide behind a low request count.
# @save
class TokenBucket:
"""Rate-limit a resource with an injected clock, so tests need no real time.
Tokens refill continuously at ``refill_per_s`` up to ``capacity``; a
request is admitted only if enough tokens are present, and consuming them
applies backpressure on the constrained dimension (input tokens, say,
rather than request count). The clock is injected so drain-and-refill
behavior is exercised deterministically.
"""
def __init__(self, capacity: float, refill_per_s: float,
now: Callable[[], float]) -> None:
self.capacity = capacity
self.refill_per_s = refill_per_s
self.now = now
self.tokens = capacity
self.updated_at = now()
def allow(self, requested: float) -> bool:
"""Admit a request of ``requested`` tokens, refilling for elapsed time.
Args:
requested: Tokens the request would consume.
Returns:
True and consumes the tokens when enough are available; False and
consumes nothing otherwise.
"""
current = self.now()
elapsed = max(0.0, current - self.updated_at)
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_per_s)
self.updated_at = current
if requested > self.tokens:
return False
self.tokens -= requested
return Trueclock = [0.0]
bucket = TokenBucket(capacity=100, refill_per_s=10, now=lambda: clock[0])
print("spend 80:", bucket.allow(80), " then 30:", bucket.allow(30), "(drained)")
clock[0] = 1.0
print("after 1s refill, 30:", bucket.allow(30), " goodput(Q=120, s=0.9):",
goodput(120, 0.9))spend 80: True then 30: False (drained)
after 1s refill, 30: True goodput(Q=120, s=0.9): 108.026.4 Reliability: deadlines, retry budgets, and hedging
A timeout says the caller stopped waiting. It does not say the server stopped working, the model call went unbilled, or the external effect did not occur. So the first reliability rule is to classify a failure before reacting to it: a retryable fault (a connection reset before any response, a declared overload) may be retried within the same journey; a terminal one (invalid schema, failed authentication, policy refusal) must stop; and an ambiguous one — a connection lost after a write was sent — must be reconciled by a stable identity, never retried with a fresh one.
# @save
TERMINAL = frozenset({"policy_refusal", "invalid_request", "auth"})
RETRYABLE = frozenset({"overloaded", "connection_reset", "rate_limited"})
def classify_failure(kind: str) -> str:
"""Classify a failure as retryable, terminal, or ambiguous.
Args:
kind: A short failure label from the provider or transport.
Returns:
"terminal", "retryable", or "ambiguous"; the last covers writes whose
outcome is unknown and must be reconciled, not blindly retried.
"""
if kind in TERMINAL:
return "terminal"
if kind in RETRYABLE:
return "retryable"
return "ambiguous"
def propagate_deadline(remaining_s: float, reserved_s: float) -> float:
"""Return the timeout a downstream call may use, reserving tail time.
A service that receives 800 ms remaining must not start a fixed two-second
timeout: it must pass down what is left minus time reserved for validation,
effect recording, and the response path.
Args:
remaining_s: Time left on the journey's absolute deadline.
reserved_s: Time to hold back for the caller's own tail work.
Returns:
The non-negative budget a downstream call may spend.
"""
return max(0.0, remaining_s - reserved_s)A retry budget bounds total extra work while a deadline bounds elapsed usefulness; the two together stop a retry loop from either running forever or running past the point where its answer still matters. We give the budget an injected clock and watch it consume attempts until one of the two limits bites.
# @save
@dataclass
class RetryBudget:
"""Bound retries by an attempt count and an absolute deadline together.
A retryable failure consumes one attempt and is allowed only while both
attempts and deadline remain; a terminal failure stops immediately.
Separating the two limits is what makes the budget honest: a fast failure
loop runs out of attempts, a slow one runs out of time.
"""
attempts_left: int
deadline_s: float
now: Callable[[], float]
def consume(self, kind: str) -> float:
"""Consume one retry for a retryable failure, or refuse to.
Args:
kind: The failure label being handled.
Returns:
The time remaining after consuming the attempt.
Raises:
RuntimeError: On a terminal failure (do not retry).
TimeoutError: When attempts or the deadline are exhausted.
"""
remaining = self.deadline_s - self.now()
if classify_failure(kind) == "terminal":
raise RuntimeError("terminal failure; do not retry")
if self.attempts_left <= 0 or remaining <= 0:
raise TimeoutError("retry budget exhausted")
self.attempts_left -= 1
return remainingelapsed = [0.0]
budget = RetryBudget(attempts_left=5, deadline_s=1.0, now=lambda: elapsed[0])
print("downstream budget from 0.8s, reserving 0.2s:",
propagate_deadline(0.8, 0.2))
for step in range(6):
elapsed[0] = step * 0.25
try:
remaining = budget.consume("overloaded")
print(f"t={elapsed[0]:.2f}s retry allowed attempts_left="
f"{budget.attempts_left} remaining={remaining:.2f}s")
except TimeoutError as exc:
print(f"t={elapsed[0]:.2f}s stop: {exc}")
breakdownstream budget from 0.8s, reserving 0.2s: 0.6000000000000001
t=0.00s retry allowed attempts_left=4 remaining=1.00s
t=0.25s retry allowed attempts_left=3 remaining=0.75s
t=0.50s retry allowed attempts_left=2 remaining=0.50s
t=0.75s retry allowed attempts_left=1 remaining=0.25s
t=1.00s stop: retry budget exhaustedThe budget stops the loop when the clock crosses the deadline, well before the attempt count would have — exactly the case a naive “retry three times” would get wrong. The last pattern, hedging, spends more to buy tail latency: after a threshold, issue a second attempt and take whichever returns first (Dean and Barroso 2013). It helps when a small fraction of components are slow for uncorrelated reasons, but it costs extra capacity during the busiest period, so the question is always whether the tail win is worth the spend. We measure it on a latency distribution with a heavy tail.
# @save
def hedge_latency(first_ms: float, second_ms: float, threshold_ms: float
) -> tuple[float, bool]:
"""Return the effective latency and whether a hedge fired.
If the first attempt finishes by ``threshold_ms``, its latency stands and
no hedge fires. Otherwise a second attempt starts at the threshold and the
request finishes when either returns, so the effective latency is the
smaller of the first attempt and threshold-plus-second.
Args:
first_ms: Latency of the primary attempt.
second_ms: Latency the hedged attempt would take on its own.
threshold_ms: Delay before the hedge is issued.
Returns:
A tuple of (effective latency in ms, whether the hedge fired).
"""
if first_ms <= threshold_ms:
return first_ms, False
return min(first_ms, threshold_ms + second_ms), Truedef draw_latency(rng: random.Random) -> float:
# 95% fast, 5% a slow tail — the shape hedging targets.
if rng.random() < 0.05:
return rng.gauss(1500, 300)
return rng.gauss(200, 30)
rng = random.Random(0)
base, hedged, fired = [], [], 0
for _ in range(20_000):
first, second = draw_latency(rng), draw_latency(rng)
base.append(first)
eff, did = hedge_latency(first, second, threshold_ms=400)
hedged.append(eff)
fired += did
def pct(xs, p):
return sorted(xs)[int(p * len(xs))]
print(f"baseline p50={pct(base,0.5):5.0f}ms p99={pct(base,0.99):5.0f}ms")
print(f"hedged p50={pct(hedged,0.5):5.0f}ms p99={pct(hedged,0.99):5.0f}ms")
print(f"hedge fired on {fired/len(base):.1%} of requests (the extra cost)")baseline p50= 203ms p99= 1740ms
hedged p50= 203ms p99= 629ms
hedge fired on 4.9% of requests (the extra cost)The hedge barely touches the median — most requests finish before the threshold — but it collapses the ninety-ninth percentile, because a slow first attempt is very likely paired with a fast second. The bill for that is the few percent of requests that ran twice. Figure 26.6 shows the tail before and after.
Show the code that draws this figure
fig, ax = plt.subplots(figsize=(6.4, 3.3))
groups = ["p50", "p99"]
base_v = [pct(base, 0.5), pct(base, 0.99)]
hedge_v = [pct(hedged, 0.5), pct(hedged, 0.99)]
x = range(len(groups))
ax.bar([i - 0.2 for i in x], base_v, width=0.4, label="baseline", color="#b13f3f")
ax.bar([i + 0.2 for i in x], hedge_v, width=0.4, label="hedged", color="#315b8a")
ax.set_xticks(list(x))
ax.set_xticklabels(groups)
ax.set_ylabel("latency (ms)")
ax.legend()
ax.spines[["top", "right"]].set_visible(False)
fig.tight_layout()
plt.show()
Hedging works for side-effect-free reads. For a write, running twice is exactly the danger the next section is about.
26.5 Close the crash window with an effect ledger
Now the second incident. A worker calls a refund provider; the provider commits the refund and returns a receipt; before the worker records that receipt, its process dies. Replay restarts from the last checkpoint and calls the provider again. If it uses a fresh idempotency key, the customer is refunded twice. The unsafe interval — between the external commit and the durable record that tells replay not to repeat it — is the crash window, and no ordinary checkpoint can atomically commit a local row together with an unrelated provider across a network. Figure 26.7 shows exactly where the duplicate is born and which stable fact resolves it.
sequenceDiagram
participant W as Worker
participant L as Effect ledger
participant P as Provider
W->>L: reserve(stable key, payload digest)
L-->>W: RESERVED
W->>P: apply(key, exact payload)
P-->>W: committed(receipt-1)
Note over W: process dies before recording the receipt
W->>L: recover(same stable key)
W->>P: apply(same key, same payload)
P-->>W: existing receipt-1 (no second effect)
W->>L: record(receipt-1) -> RECORDED
An effect ledger records the application’s intent separately from workflow progress, and it moves an effect through a small set of states, each of which answers one recovery question. RESERVED says the key and payload digest are durable; EXECUTED says a receipt is known in this process; RECORDED says that receipt is durable so replay may return it; FAILED says a terminal result needs review. Figure 26.8 draws the machine.
stateDiagram-v2
[*] --> INTENDED: create exact intent
INTENDED --> RESERVED: persist key + payload digest
RESERVED --> EXECUTED: provider returns receipt
EXECUTED --> RECORDED: persist authoritative receipt
RESERVED --> FAILED: terminal provider decision
RECORDED --> [*]
The ledger’s core is small. A reserve binds a stable key to a payload digest and refuses to let the same key be reused for a different payload — so a changed amount under an old key is rejected, not silently applied.
# @save
class EffectState(str, Enum):
"""The durable states of one external effect, in recovery order."""
INTENDED = "INTENDED"
RESERVED = "RESERVED"
EXECUTED = "EXECUTED"
RECORDED = "RECORDED"
FAILED = "FAILED"
@dataclass
class EffectRecord:
"""One effect's durable identity, digest, state, and authoritative receipt."""
key: str
payload_digest: str
state: EffectState
receipt: str | None = None
class EffectLedger:
"""Keep a stable, payload-bound effect identity across crash and replay.
The ledger is the application-owned record that a checkpoint cannot be: it
binds a business idempotency key to a payload digest before the provider is
called, so recovery re-asks by that key and a substituted payload is
refused. It stores intent, not workflow progress.
"""
def __init__(self) -> None:
self.records: dict[str, EffectRecord] = {}
@staticmethod
def digest(payload: dict[str, Any]) -> str:
"""Return a canonical SHA-256 digest of an effect payload."""
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(encoded).hexdigest()
def reserve(self, key: str, payload: dict[str, Any]) -> EffectRecord:
"""Reserve a stable key for an exact payload, or reject a substitution.
Args:
key: The business idempotency key (stable across retries).
payload: The exact effect payload; its digest is bound to the key.
Returns:
The record for this key, moved to RESERVED on first reservation.
Raises:
ValueError: If the key already exists for a different payload.
"""
digest = self.digest(payload)
record = self.records.get(key)
if record and record.payload_digest != digest:
raise ValueError("idempotency key reused for a different payload")
if record is None:
record = EffectRecord(key, digest, EffectState.INTENDED)
self.records[key] = record
if record.state == EffectState.INTENDED:
record.state = EffectState.RESERVED
return recordThe provider is idempotent on the stable key — it returns the same receipt for a repeated key and commits at most one effect — and execute_once drives the whole sequence, including the FAILED branch a terminal provider error takes. The injected crash models process death precisely in the dangerous window: after the provider commits, before the receipt is recorded.
# @save
class InjectedCrash(RuntimeError):
"""Simulate worker death after the provider committed, before recording."""
class IdempotentProvider:
"""A provider that commits at most one effect per stable key.
It returns the same receipt for a repeated key, which is the property that
makes recovery safe: a worker that crashes and retries with the same key
sees the original receipt instead of a second effect. ``effect_count`` is
the ground truth the tests assert on.
"""
def __init__(self) -> None:
self.receipts: dict[str, str] = {}
self.effect_count = 0
def apply(self, key: str, payload: dict[str, Any]) -> str:
"""Commit the effect once for a new key, else return its receipt."""
if key not in self.receipts:
self.effect_count += 1
self.receipts[key] = f"receipt-{self.effect_count}"
return self.receipts[key]
def execute_once(
ledger: EffectLedger,
provider: IdempotentProvider,
key: str,
payload: dict[str, Any],
crash_after_provider: bool = False,
) -> str:
"""Execute an effect once, or recover it safely after a crash.
Recovery is a composition, not magic: a durable reservation (the ledger)
plus a provider that honors a stable key. If the record is already
RECORDED, the receipt is returned without touching the provider. Otherwise
the provider is called with the stable key; a crash in the window between
the call and recording leaves the record RESERVED, so the next attempt
re-asks the provider by the same key and still sees one effect.
Args:
ledger: The durable effect ledger.
provider: The idempotent external provider.
key: The stable business idempotency key.
payload: The exact effect payload.
crash_after_provider: If True, raise in the dangerous window to model
a worker dying before it recorded the receipt.
Returns:
The authoritative receipt for this effect.
Raises:
InjectedCrash: When ``crash_after_provider`` is set, before recording.
"""
record = ledger.reserve(key, payload)
if record.state == EffectState.RECORDED:
return record.receipt # type: ignore[return-value]
try:
receipt = provider.apply(key, payload)
if crash_after_provider:
raise InjectedCrash("died before recording provider receipt")
record.state = EffectState.EXECUTED
record.receipt = receipt
record.state = EffectState.RECORDED
return receipt
except InjectedCrash:
raise
except Exception:
record.state = EffectState.FAILED
raiseNow the demonstration the old chapter only asserted in a test: crash three times in the window, then recover, and print the whole trace. The provider must commit exactly one effect no matter how many deaths occurred.
ledger, provider = EffectLedger(), IdempotentProvider()
payload = {"order_id": "A-17", "amount_cents": 4999}
key = "refund:A-17:v1"
for attempt in range(1, 4):
try:
execute_once(ledger, provider, key, payload, crash_after_provider=True)
except InjectedCrash:
state = ledger.records[key].state.value
print(f"attempt {attempt}: crashed in window ledger={state} "
f"provider_effects={provider.effect_count}")
receipt = execute_once(ledger, provider, key, payload)
print(f"recovery: receipt={receipt} ledger={ledger.records[key].state.value} "
f"provider_effects={provider.effect_count}")attempt 1: crashed in window ledger=RESERVED provider_effects=1
attempt 2: crashed in window ledger=RESERVED provider_effects=1
attempt 3: crashed in window ledger=RESERVED provider_effects=1
recovery: receipt=receipt-1 ledger=RECORDED provider_effects=1Three deaths, one effect. Each crash left the record RESERVED — durable proof that an attempt was in flight — and recovery re-asked the provider with refund:A-17:v1, which returned the original receipt-1 rather than moving money again. Try to reuse that key with a changed amount and the ledger refuses:
try:
execute_once(ledger, provider, key, {"order_id": "A-17", "amount_cents": 9999})
except ValueError as exc:
print("substituted payload rejected:", exc)
print("provider effects still:", provider.effect_count)substituted payload rejected: idempotency key reused for a different payload
provider effects still: 1That is the whole content of “exactly once” for an external effect: a stable, payload-bound identity plus a provider that honors it. A transactional outbox solves the adjacent local problem — write business state and an outgoing intent in one database transaction, then relay the intent at least once — but it prevents a lost message, not a duplicate; the consumer still deduplicates by message identity (Richardson 2019).
NoteIn the interview
Q. Your workflow crashes after a payment provider charges a card but before your code records the receipt. On resume it charges again. Why, and what is the fix? The charge and the local record live in different systems, so they cannot commit atomically — that gap is the crash window, and “the engine guarantees exactly once” does not cover an external provider. The fix is a stable, payload-bound idempotency key reserved before the call (refund:{order}:{intent_version}), plus a provider that returns the original receipt for a repeated key; recovery re-asks by that key instead of minting a new one. The trap is claiming a workflow engine makes external effects exactly-once on its own, or keying the operation on the network attempt (uuid4() per retry) rather than the business operation.
26.6 Durable execution: replay and compensation
A framework checkpointer stores agent state so a loop can resume; a durable-execution engine records enough workflow history to reconstruct decisions, schedule retries and timers, and move work to another worker after a failure. The engine’s central trick is deterministic replay: on recovery it re-runs the workflow code but returns each past step’s recorded result instead of re-executing it. This matters most for a model call, because re-executing one changes its output, its cost, and possibly the next branch the workflow takes. We build a tiny engine that makes the point concrete.
# @save
@dataclass
class WorkflowEngine:
"""A minimal durable-execution engine: journal activities, replay them.
A workflow calls :meth:`activity` for every non-deterministic or effectful
step. The first time an activity id is seen its function runs and the
result is journaled; on any later pass the journaled result is returned
without running the function again. That is exactly what lets a crashed
workflow resume — replay reconstructs past decisions from the journal
rather than re-executing paid, non-deterministic calls.
"""
journal: dict[str, Any] = field(default_factory=dict)
executed: list[str] = field(default_factory=list)
def activity(self, activity_id: str, fn: Callable[[], Any]) -> Any:
"""Run ``fn`` once and journal it, or replay the journaled result.
Args:
activity_id: Stable id naming this step in the workflow history.
fn: The step body; executed only the first time this id is seen.
Returns:
The recorded result for ``activity_id`` — fresh on first
execution, replayed verbatim thereafter.
"""
if activity_id in self.journal:
return self.journal[activity_id]
result = fn()
self.journal[activity_id] = result
self.executed.append(activity_id)
return resultFirst, why the model call must not re-run. A workflow summarizes a document with a call whose token count differs every time — as a real model’s would. We record it once, then show what a naive re-execution would return versus what replay returns.
calls = {"n": 0}
def flaky_summarize() -> int:
calls["n"] += 1
return 100 + 7 * calls["n"] # a real model call: different tokens each time
engine = WorkflowEngine()
first_run = engine.activity("summarize", flaky_summarize)
naive_rerun = flaky_summarize() # what re-executing would give
replayed = engine.activity("summarize", flaky_summarize) # what replay gives
print(f"first run recorded {first_run} tokens")
print(f"naive re-execution would return {naive_rerun} tokens (diverged!)")
print(f"replay returns {replayed} tokens; real executions so far: {engine.executed}")first run recorded 107 tokens
naive re-execution would return 114 tokens (diverged!)
replay returns 107 tokens; real executions so far: ['summarize']The naive re-execution diverges — different tokens, different cost, a possibly different downstream branch — while replay returns the recorded value and the function is not called again. Figure 26.9 draws the three paths.
flowchart LR
subgraph first["First run"]
A1["call model"] --> R1["journal tokens = T1"]
end
subgraph replay["Replay after crash"]
J["read journal"] --> RT["return recorded T1"]
end
subgraph naive["Naive re-execution (wrong)"]
A2["call model again"] --> R2["new tokens T2 != T1"]
R2 --> D["different cost, latency, next branch"]
end
Now a crash inside a workflow. A booking charges a card — an effect — then sends a receipt. We inject a death during the receipt send, after the charge is journaled, and then resume with the same engine. The charge must replay from the journal without a second call.
sends = {"n": 0}
def send_receipt() -> str:
sends["n"] += 1
if sends["n"] == 1:
raise InjectedCrash("died sending the receipt")
return "receipt-sent"
def booking_workflow(engine: WorkflowEngine, provider: IdempotentProvider) -> str:
engine.activity("charge", lambda: provider.apply("charge:order-9", {"cents": 5000}))
engine.activity("send_receipt", send_receipt)
return "done"
provider = IdempotentProvider()
engine = WorkflowEngine()
try:
booking_workflow(engine, provider)
except InjectedCrash as exc:
print(f"crashed: {exc} journal so far: {engine.executed}")
booking_workflow(engine, provider) # resume with the same durable journal
print(f"resumed: journal {engine.executed} provider_effects={provider.effect_count}")crashed: died sending the receipt journal so far: ['charge']
resumed: journal ['charge', 'send_receipt'] provider_effects=1On resume the charge replayed from the journal — it is not in executed a second time — so the card was charged exactly once even though the workflow ran twice. An activity can still run more than once if a worker dies after the external effect but before the engine records completion, which is why the effect ledger of the previous section stays necessary unless the engine and the data store share a documented atomic boundary. Whether to reach for a checkpointer or a full engine is a lifetime decision:
| Decision | Framework checkpointer | Durable-execution engine |
|---|---|---|
| unit of resume | graph or loop state | workflow history plus activities |
| long timers, external signals | application implements them | native durable primitive |
| deployment migration | deserialize into compatible code | replay history under version rules |
| external effect contract | application-owned | still application-owned unless a shared atomic boundary is documented |
| best fit | conversational or bounded agent state | multi-service, long-running business process |
Against a real engine, the model call is wrapped as a recorded activity — shown here, not executed, because it needs the engine’s runtime:
from durable_engine import workflow, activity # a Temporal-class engine
@activity
def summarize(doc: str) -> dict:
return call_model(doc) # non-deterministic + paid: recorded once
@workflow
def review_workflow(doc: str) -> dict:
summary = summarize(doc) # replay returns the recorded result
approval = wait_for_signal("human_approval", timeout="14 days") # durable timer
return {"summary": summary, "approved": approval}The long human wait becomes a durable timer and signal: the workflow suspends without holding a process and resumes when a validated signal arrives or the timer expires. Chapter 17 defined approval mechanics and Chapter 24 the authorization binding; the platform guarantee here is only that a restart, a deploy, or a two-week wait neither loses nor duplicates the suspended intent.
The last durable pattern is compensation, and it is not time travel. When a multi-step booking fails partway, a saga runs a compensating action for each committed step — but a compensation is a new forward effect with its own stable identity, not an undo. Our saga books a flight and a hotel, fails on the event, and compensates the two bookings in reverse.
# @save
class EffectRejected(RuntimeError):
"""A provider declined an effect (e.g. an event is sold out)."""
def book_trip(engine: WorkflowEngine, provider: IdempotentProvider,
event_available: bool) -> list[str]:
"""Book flight, hotel, and event as a saga; compensate on failure.
Each booking is a durable activity. If the event booking is rejected, the
already-committed steps are compensated in reverse order, each compensation
being a new forward effect with its own stable key — because you cannot
un-charge a card, only issue a matching reversal.
Args:
engine: The durable engine journaling each step.
provider: The idempotent effect provider.
event_available: Whether the event booking succeeds.
Returns:
The provider receipts, in the order effects were committed (bookings
first, then any compensations).
Raises:
EffectRejected: After compensations run, to signal the saga failed.
"""
compensations: list[tuple[str, str]] = []
engine.activity("flight", lambda: provider.apply("flight:T1", {"seat": "12A"}))
compensations.append(("cancel_flight", "flight:T1"))
engine.activity("hotel", lambda: provider.apply("hotel:T1", {"nights": 2}))
compensations.append(("cancel_hotel", "hotel:T1"))
try:
if not event_available:
raise EffectRejected("event sold out")
engine.activity("event", lambda: provider.apply("event:T1", {"seats": 2}))
except EffectRejected:
for name, target in reversed(compensations):
provider.apply(f"{name}:{target}", {"compensates": target})
raise
return list(provider.receipts.values())provider = IdempotentProvider()
engine = WorkflowEngine()
try:
book_trip(engine, provider, event_available=False)
except EffectRejected as exc:
print("saga failed:", exc)
for key, receipt in provider.receipts.items():
print(f" {receipt:11} {key}")
print("total forward effects (2 bookings + 2 compensations):", provider.effect_count)saga failed: event sold out
receipt-1 flight:T1
receipt-2 hotel:T1
receipt-3 cancel_hotel:hotel:T1
receipt-4 cancel_flight:flight:T1
total forward effects (2 bookings + 2 compensations): 4The flight and hotel were booked, the event was rejected, and the two bookings were compensated with cancel_hotel:hotel:T1 then cancel_flight:flight:T1 — new effects with stable ids, applied in reverse. Had the event been an irreversible effect that had already committed, no compensation could undo it; that is why effects are classified as reversible, compensatable, or irreversible before an autonomous workflow is allowed to drive them (Garcia-Molina and Salem 1987).
26.7 Version the full artifact surface
“Model version” does not identify an agent release. The output can change when the tokenizer, the system prompt, a tool schema, the retrieval corpus snapshot, the embedder, the reranker, a decoding setting, or the judge rubric changes — a dozen surfaces, any of which silently alters behavior. A release bundle maps every material surface to an immutable identifier and is addressed by the content hash of that map, so two releases are the same release exactly when their bundles hash equal. Because the map is order-independent, the hash must be too.
# @save
def bundle_digest(surface: dict[str, str]) -> str:
"""Return a canonical content hash of a complete release bundle.
Every correctness-bearing surface — model, tokenizer, prompt, tools,
corpus, embedder, judge — maps to an immutable id here. Canonical JSON
makes the digest independent of key order, so the same surfaces always
hash equal and any single change flips the digest.
Args:
surface: A map from surface name to its immutable identifier.
Returns:
The hex SHA-256 digest addressing this exact bundle.
"""
encoded = json.dumps(surface, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(encoded).hexdigest()
def manifest_diff(old: dict[str, str], new: dict[str, str]) -> set[str]:
"""Return the surfaces whose identifier changed between two bundles."""
return {k for k in old.keys() | new.keys() if old.get(k) != new.get(k)}We hash a manifest, confirm that reordering its keys leaves the digest unchanged, then change exactly one surface and watch the digest flip while manifest_diff names precisely what moved.
v42 = {"model": "m-1", "tokenizer": "tok-3", "prompt": "sys-7",
"corpus": "docs-2026-07-01", "judge": "rubric-4"}
reordered = dict(reversed(list(v42.items())))
v43 = {**v42, "corpus": "docs-2026-07-15"}
print("digest v42 :", bundle_digest(v42)[:16])
print("reordered :", bundle_digest(reordered)[:16], " (same bundle, same hash)")
print("digest v43 :", bundle_digest(v43)[:16], " (one surface changed)")
print("changed surfaces:", manifest_diff(v42, v43))digest v42 : f11f996090271306
reordered : f11f996090271306 (same bundle, same hash)
digest v43 : 72c57c3182ca9dea (one surface changed)
changed surfaces: {'corpus'}Order does not matter; a single corpus bump does. A request captures the resolved hash at admission and keeps it for the whole journey, so caches key on it, checkpoints store it so resume uses compatible tools, and effect records store it so an incident can reconstruct the exact code and authorization that produced a receipt. Rollback publishes a new control-plane pointer to a proven prior bundle; it never edits v42 in place.
Recovery of the data behind those surfaces needs per-store objectives. Recovery point objective (RPO) is the maximum tolerable data loss in time; recovery time objective (RTO) is the target time to restore service. A rebuildable vector index can tolerate an older backup as long as its source documents, chunker, embedder, and deletion log are retained; an effect ledger may need near-zero RPO because a lost receipt can duplicate money movement; a prompt cache can often be discarded. State that cannot be rebuilt needs a tested restore, not merely a backup job with a green timestamp — and end-to-end lineage from source document through chunk, embedding, retrieved context, model call, and effect receipt is what makes deletion propagation (Chapter 18), forensics (Chapter 24), and incident reconstruction (Chapter 27) possible (Kleppmann 2017).
26.8 Test the deterministic shell
An evaluation asks whether a probabilistic system performs a task; a test asks whether deterministic code obeys an invariant. A release needs both, and a ninety-two-percent task pass rate says nothing about whether concurrent budget checks overspend, a retry changes an effect key, or yesterday’s checkpoint still loads. The most valuable test family for the machinery in this chapter is the property-based test: rather than pinning one example, it generates many inputs and asserts an invariant across all of them. The invariant we most want is the one Section 26.5 claimed — that no number of crashes produces more than one effect — so we assert it over hundreds of random crash schedules.
def effect_count_is_one(crashes: int, cents: int, seed: int) -> bool:
ledger, provider = EffectLedger(), IdempotentProvider()
key = f"refund:{seed}:v1"
payload = {"cents": cents}
for _ in range(crashes):
try:
execute_once(ledger, provider, key, payload, crash_after_provider=True)
except InjectedCrash:
pass
execute_once(ledger, provider, key, payload)
return provider.effect_count == 1
rng = random.Random(2026)
cases = [(rng.randint(0, 12), rng.randint(1, 100_000), i) for i in range(500)]
held = all(effect_count_is_one(c, cents, s) for c, cents, s in cases)
print(f"property held over {len(cases)} random crash schedules: {held}")property held over 500 random crash schedules: TrueFive hundred random schedules, the invariant holds in every one. A dedicated tool such as Hypothesis automates this generation and shrinks a failure to its minimal case; the hand-rolled loop keeps the chapter dependency-free while teaching the same idea. Two neighboring families are worth naming. A metamorphic test changes execution while preserving a relation — for example, “adding a retry never increases the external effect count,” which we can check by comparing a clean run to a crash-and-recover run and asserting the effect counts are equal. A mutation test deliberately breaks the code — say, making reserve accept any payload for an existing key — and confirms the suite notices; a test that stays green under that mutation was never really testing the substitution guard. Mocks help for rare errors but often grant the exact idempotency you forgot to verify, so crash and transaction guarantees need an integration environment with the real database and engine; quarantine a flaky test only with an owner, a diagnosis, and an expiry, never as a silent retry.
26.9 Deliver through gates, shadow, and canary
The last piece is getting a new bundle into production without exposing every user to its mistakes. Progressive delivery is a funnel that widens exposure only as evidence accumulates: an offline gate runs tests and evaluations against the exact bundle; a shadow stage sends the candidate a copy of traffic but lets it create no effects; a canary routes a small real-effect slice with strong abort rules; a ramp widens it in stages; and converge retires the old version only after its journeys drain. The reason the canary slice can be small is statistical: if a candidate independently fails an extra fraction p of cases, the chance of catching at least one failure in n independent cases is P = 1 - (1-p)^n, so a target detection probability \gamma needs n \ge \log(1-\gamma)/\log(1-p) cases.
# @save
def wilson_lower(successes: int, trials: int, z: float = 1.96) -> float:
"""Return a Wilson lower confidence bound for a success rate.
The Wilson interval is well-behaved near 0 and 1 and at small samples,
where the naive normal interval misbehaves. A canary uses the lower bound —
not the point estimate — so a good-looking rate from few trials cannot
promote a release on luck.
Args:
successes: Number of successful canary cases.
trials: Total canary cases (must be positive).
z: Normal quantile for the confidence level (1.96 ~ 95%).
Returns:
The lower confidence bound on the true success rate.
Raises:
ValueError: If ``trials`` is not positive.
"""
if trials <= 0:
raise ValueError("trials must be positive")
rate = successes / trials
denom = 1 + z * z / trials
center = rate + z * z / (2 * trials)
radius = z * math.sqrt(rate * (1 - rate) / trials + z * z / (4 * trials ** 2))
return max(0.0, (center - radius) / denom)
def detection_probability(sample_size: int, regression_rate: float) -> float:
"""Return the chance of seeing at least one independent regression."""
return 1 - (1 - regression_rate) ** sample_size
def canary_sample_size(regression_rate: float, target_prob: float) -> int:
"""Return the independent cases needed to detect a regression with target_prob.
Args:
regression_rate: The extra failure fraction to be able to catch.
target_prob: The detection probability to reach, in (0, 1).
Returns:
The smallest case count meeting the target under the one-hit model.
"""
return math.ceil(math.log(1 - target_prob) / math.log(1 - regression_rate))The gate itself combines two rules of different kinds. Quality uses the Wilson lower bound against a release floor — uncertainty-aware, so few trials cannot promote on luck — while safety is zero-tolerance: any critical failure holds the release outright.
# @save
@dataclass(frozen=True)
class CanaryDecision:
"""The outcome of a canary gate: promote or hold, with the reasons why."""
effect: str
observed_rate: float
lower_bound: float
reasons: tuple[str, ...]
def canary_gate(successes: int, trials: int, minimum_rate: float,
critical_failures: int = 0) -> CanaryDecision:
"""Decide promotion from an uncertainty bound plus a zero-tolerance rule.
The candidate promotes only if no critical invariant failed and the Wilson
lower bound clears the release floor; otherwise it holds. Combining a
statistical quality test with an absolute safety test is what lets one
unauthorized transfer block a release that otherwise looks excellent.
Args:
successes: Successful canary cases.
trials: Total canary cases.
minimum_rate: The quality floor the lower bound must clear.
critical_failures: Count of zero-tolerance failures observed.
Returns:
A ``CanaryDecision`` naming the effect and every holding reason.
"""
lower = wilson_lower(successes, trials)
reasons: list[str] = []
if critical_failures:
reasons.append("critical invariant failed")
if lower < minimum_rate:
reasons.append("quality lower bound misses release floor")
return CanaryDecision("PROMOTE" if not reasons else "HOLD",
successes / trials, lower, tuple(reasons))Now watch the gate flip across three scenarios that a naive point-estimate gate would get wrong. A strong result on a hundred cases promotes; the same observed rate on twenty-five cases holds, because the lower bound has not cleared the floor; and a perfect run with one critical failure holds regardless of the rate.
floor = 0.90
scenarios = [
("98 of 100", canary_gate(98, 100, floor)),
("24 of 25", canary_gate(24, 25, floor)),
("100 of 100, 1 critical", canary_gate(100, 100, floor, critical_failures=1)),
]
for name, decision in scenarios:
print(f"{name:26} rate={decision.observed_rate:.2f} "
f"lower={decision.lower_bound:.3f} -> {decision.effect} "
f"{list(decision.reasons)}")
print(f"\ncases to catch a 5% regression at 80% confidence: "
f"{canary_sample_size(0.05, 0.80)}")98 of 100 rate=0.98 lower=0.930 -> PROMOTE []
24 of 25 rate=0.96 lower=0.805 -> HOLD ['quality lower bound misses release floor']
100 of 100, 1 critical rate=1.00 lower=0.963 -> HOLD ['critical invariant failed']
cases to catch a 5% regression at 80% confidence: 32The observed rate is nearly identical in the first two rows, yet only the larger sample promotes — the gate is buying its confidence with cases, not with a lucky ratio. Figure 26.10 puts the exposure funnel beside the detection curve so the two ideas — shrink exposure, grow evidence — sit together.
Show the code that draws this figure
stages = ["offline", "shadow", "canary", "ramp", "converge"]
exposure = [0.0, 0.0, 0.05, 0.35, 1.0]
n = list(range(1, 81))
detect = [detection_probability(k, 0.05) for k in n]
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(8.6, 3.4))
ax1.barh(stages[::-1], exposure[::-1], color="#315b8a")
ax1.set_xlabel("fraction of users exposed to real effects")
ax1.set_title("delivery funnel")
ax2.plot(n, detect, color="#2f855a", lw=2)
ax2.axhline(0.8, ls="--", color="#666666", lw=1)
ax2.scatter([32], [detection_probability(32, 0.05)], color="#2f855a", zorder=3)
ax2.annotate("32 cases", (32, 0.8), textcoords="offset points", xytext=(6, -14),
fontsize=8)
ax2.set(xlabel="independent canary cases", ylabel="detection probability",
title="detect one 5% regression", ylim=(0, 1))
for ax in (ax1, ax2):
ax.spines[["top", "right"]].set_visible(False)
ax.grid(alpha=0.2)
fig.tight_layout()
plt.show()
Two closing rules. Shadowing is not free — it can duplicate model and tool cost, disclose inputs to another provider, and mutate caches — so a shadow candidate must have its write capabilities stripped and its traffic excluded where consent or residency forbids duplication. And rollback has two meanings that a delivery pipeline must both support: technical rollback moves traffic to a prior bundle, while business rollback repairs effects already caused — reconcile transactions, contact users, invoke compensation. Chapter 27 owns incident execution; the pipeline’s job is to retain the identities, receipts, and release hashes that make both rollbacks possible. Chapter 22’s paired release contract remains the canonical treatment for full comparative evaluation; the compact Wilson gate here illustrates the uncertainty discipline that gate applies.
26.10 Summary
A production platform is the enforcement point the Friday incident lacked: the boundary between application intent and spendable capacity. We built one and measured it — a gateway that makes identity, budget, and routing explicit; a cost model that turns caching, batching, and loop pathologies into numbers; reliability and durable-execution machinery that make an external effect happen exactly once across crashes and compensate when a saga fails; a content-hashed release manifest; and a release argument of property tests and a powered canary. The recurring lesson: a prompt enforces nothing, but a stable key, a recorded activity, and an uncertainty-aware gate do. Chapter 27 operates what we shipped.
26.11 Exercises
- Cascade knobs. Using
run_cascade, sweep the grader’sfalse_acceptrate from 0.0 to 0.3 and plot accuracy and total cost against it. At what false-accept rate does the cascade’s accuracy fall below the router’s? Explain, in terms of Equation 26.1, why a worse grader costs both quality and money. - Atomic admission.
Gateway.admitchecks a balance but does not reserve it, so two concurrent journeys that individually fit can together overspend. Add a reserve-and-release protocol (reserve at admit, settle or release at the end) and write a test in which two estimates individually fit the limit but the second is rejected once the first has reserved. - Break-even under expiry. Extend Equation 26.2 so a cached prefix expires before reuse with probability e and a miss pays both the lookup and the full uncached cost. Derive the new break-even reuse count, implement it beside
cache_breakeven, and plot the boundary over e for the chapter’s write and read multipliers. - Reconciliation worker. Implement a worker that resolves ledger records left
RESERVEDafter a crash by querying the provider. Model three provider answers —committed,not_found,unknown— and give each a safe transition. Defend, in code and a sentence, whyunknownmust never invent a new effect key. - Saga with an irreversible step. Modify
book_tripso the event is irreversible and succeeds, but a later “insurance” step fails.- Show that the two compensatable bookings can be reversed while the irreversible event cannot.
- Route the irreversible effect to a manual-review outcome instead of a compensation, and print the resulting effect log.
- Argue which effects an autonomous workflow should be allowed to drive without a human, using your classification.
- Replay divergence, measured. Wrap
flaky_summarizein a workflow whose next branch depends on whether the token count exceeds a threshold. Run it once, then run a naive re-execution and a journaled replay, and show that only the naive path can take a different branch. Relate this to why paid, non-deterministic calls belong inside recorded activities. - Power a rare-failure canary. A critical failure is expected once in ten thousand journeys. Use
canary_sample_sizeto show how many cases a live canary would need to expect one such event, explain why waiting to observe one may be unacceptable, then design a release argument that combines targeted offline adversarial cases, invariant gates, and shadow evidence to bound exposure without live proof. - Manifest completeness. Write a release-bundle manifest for a permission-aware RAG agent covering every surface named in Section 26.7. Mutate the tokenizer, the corpus snapshot, the policy, and the judge each independently, and show with
bundle_digestandmanifest_diffthat each mutation flips the digest and names exactly the surface that changed.