Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
The primary model times out on access-R900. A cheap backup is healthy, but it can't return the cited, structured answer that this private access request needs. Returning something quickly would turn a provider outage into an application error.
The cost engineering lesson supplied a fictional release support-release-2026-05-cost-v1, schema cited-support-answer-v3, and a 0.005000 USD single-generation ceiling. A model gateway can apply these requirements when choosing a large language model (LLM) path. The app sends a stable internal request; the gateway filters approved paths, invokes an adapter, checks the result, and records the decision. Privacy and human review also require authorization and workflow controls around that model call.
Compatibility is the constraint that makes every later choice meaningful. The gateway must enforce the full contract so application code can't satisfy one field by dropping another.
What belongs at the gateway boundary
An adapter gives application code a stable internal schema while translating provider wire formats. You can put that adapter inside a service or behind a shared gateway. A separate gateway centralizes credentials, routing, accounting, and telemetry, but also adds a dependency, latency, and an operational failure boundary.
For example, native schema configuration differs across APIs. These are request fields checked September 21, 2026, not interchangeable SDK helper arguments:
| API | Native JSON-schema output configuration |
|---|---|
| OpenAI Responses | text.format with type: "json_schema" |
| OpenAI Chat Completions | response_format with a json_schema object |
| Claude Messages | output_config.format with type: "json_schema" |
Claude also offers strict tool inputs; tools aren't the only route to structured output. Both providers document supported schema subsets and exceptions such as refusals and incomplete output. Keep application validation even when constrained decoding is enabled.[1][2]
Cost engineering separated generation cost from cache decisions. Now the gateway consumes that budget contract alongside product and safety requirements. A stored semantic-cache hit ends before this decision, but still needs its own authorization and freshness checks. The gateway decides where one generated support request may run.
Keep three terms separate:
| Term | Job | Example here |
|---|---|---|
| Adapter | Translates a stable application request into one provider's API shape | Send messages and parse a structured reply |
| Lane | Names an approved model-plus-adapter path and its surrounding controls | local-private-cited-review |
| Gateway | Compiles requirements, filters lanes, selects or falls back, and logs the decision | Reject a cheap lane that drops citations |
An OpenAI-shaped adapter only proves that a request can be translated. It doesn't prove that two lanes share a contract. Anthropic describes its OpenAI SDK compatibility layer as a way to test and compare Claude, not as a long-term production path. In that layer, prompt caching isn't supported, response_format is ignored, and the strict tool-calling flag is ignored too. For guaranteed schema conformance, Anthropic points you to native structured outputs on the Claude API.[3]
Record capabilities on each lane. Don't infer them from a familiar URL shape. Here, citation support means preserving application evidence identifiers in JSON for subsequent checking. It doesn't guarantee claim support or enable a provider's native citation blocks: Claude currently rejects native citations combined with output_config.format.[2] Test capability combinations, not just each Boolean in isolation. Human review belongs to the surrounding workflow. A lane can prepare a reviewable answer; it can't approve its own access grant.
Before routing access-R900, inventory the cost artifact, trusted request facts, a compiled contract, and lane capability records. All lanes, capabilities, costs, and timings below are teaching fixtures, not measurements or provider prices. This CPU lab neither authenticates requests nor runs adapters, reviews answers, creates a human queue, or enforces live spending. A deployed registry needs retained evaluations tied to model, adapter, schema, policy, and registry revisions. Price each attempt using its own effective model/tier/region rate card; the preceding lesson's GPT-5.4 row doesn't price every fallback.
Run the numbered Python cells in order using Python 3.10 or later; they share state and need no network or third-party packages. The first examples select routes. Later cells simulate attempts, advance a clock, and reject invalid replies. No example grants real access or calls a model.
1from dataclasses import dataclass
2from decimal import Decimal
3from enum import Enum
4import json
5import math
6
7COST_RELEASE_ID = "support-release-2026-05-cost-v1"
8GATEWAY_POLICY_ID = "gateway-policy-v1"
9REQUIRED_ANSWER_SCHEMA = "cited-support-answer-v3"
10MAX_GENERATED_ANSWER_USD = Decimal("0.005000")
11MAX_GENERATION_ATTEMPTS = 2
12REQUEST_DEADLINE_MS = 2_500
13MAX_REQUEST_SPEND_USD = Decimal("0.010000")
14ATTEMPT_RESERVATION_USD = Decimal("0.005000")
15
16def owned_text(value: object) -> bool:
17 return type(value) is str and bool(value.strip())
18
19def integer_count(value: object) -> bool:
20 return type(value) is int and value >= 0
21
22def money(value: object) -> bool:
23 return type(value) is Decimal and value.is_finite() and value >= 0
24
25def seconds(value: object) -> bool:
26 if type(value) not in (int, float):
27 return False
28 try:
29 return math.isfinite(value) and value >= 0
30 except OverflowError:
31 return False
32
33class DataClass(str, Enum):
34 PUBLIC = "public"
35 TENANT_PRIVATE = "tenant_private"
36
37@dataclass(frozen=True)
38class GatewayRequest:
39 request_id: str
40 task: str
41 data_class: DataClass
42 context_tokens: int
43 risk_amount_cents: int = 0
44 requires_citations: bool = False
45 requires_schema: bool = True
46
47 def __post_init__(self) -> None:
48 if not owned_text(self.request_id) or not owned_text(self.task) or type(self.data_class) is not DataClass:
49 raise ValueError("need owned request/task identifiers and a known data class")
50 if not integer_count(self.context_tokens) or self.context_tokens == 0 or not integer_count(self.risk_amount_cents):
51 raise ValueError("need positive context and nonnegative integer risk")
52 if type(self.requires_citations) is not bool or type(self.requires_schema) is not bool:
53 raise ValueError("requirement flags must be known Booleans")
54
55@dataclass(frozen=True)
56class RouteContract:
57 request_id: str
58 data_class: DataClass
59 context_tokens: int
60 needs_citations: bool
61 needs_schema: bool
62 needs_human_review: bool
63 max_answer_cost_usd: Decimal
64
65 def __post_init__(self) -> None:
66 if not owned_text(self.request_id) or type(self.data_class) is not DataClass or not integer_count(self.context_tokens) or self.context_tokens == 0:
67 raise ValueError("need a named contract, known data class, and positive context")
68 if any(type(value) is not bool for value in (self.needs_citations, self.needs_schema, self.needs_human_review)) or self.needs_schema is not True:
69 raise ValueError("known requirements and the inherited schema are required")
70 if not money(self.max_answer_cost_usd) or self.max_answer_cost_usd > MAX_GENERATED_ANSWER_USD:
71 raise ValueError("cost ceiling must fit the inherited policy")
72
73@dataclass(frozen=True)
74class Lane:
75 name: str
76 provider: str
77 allowed_data_classes: frozenset[DataClass]
78 max_context_tokens: int
79 supports_schema: bool
80 supports_citations: bool
81 supports_human_review: bool
82 evaluated_answer_cost_usd: Decimal
83 expected_latency_ms: int
84
85 def __post_init__(self) -> None:
86 if not owned_text(self.name) or not owned_text(self.provider):
87 raise ValueError("lane and provider must be named")
88 if type(self.allowed_data_classes) is not frozenset or not self.allowed_data_classes or any(type(value) is not DataClass for value in self.allowed_data_classes):
89 raise ValueError("need a nonempty set of known data classes")
90 if not integer_count(self.max_context_tokens) or self.max_context_tokens == 0 or not integer_count(self.expected_latency_ms):
91 raise ValueError("invalid context capacity or latency")
92 if any(type(value) is not bool for value in (self.supports_schema, self.supports_citations, self.supports_human_review)):
93 raise ValueError("capability flags must be known Booleans")
94 if not money(self.evaluated_answer_cost_usd):
95 raise ValueError("evaluated cost must be a finite nonnegative Decimal")
96
97LANES = [
98 Lane(
99 "fast-public-json", "hosted-fast", frozenset({DataClass.PUBLIC}),
100 16_000, True, False, False, Decimal("0.001100"), 260,
101 ),
102 Lane(
103 "public-cited-review", "hosted-cited", frozenset({DataClass.PUBLIC}),
104 64_000, True, True, True, Decimal("0.003800"), 860,
105 ),
106 Lane(
107 "primary-private-cited-review", "hosted-private", frozenset({DataClass.TENANT_PRIVATE}),
108 64_000, True, True, True, Decimal("0.004200"), 940,
109 ),
110 Lane(
111 "local-private-cited-review", "local-private", frozenset({DataClass.TENANT_PRIVATE}),
112 32_000, True, True, True, Decimal("0.004500"), 1_100,
113 ),
114 Lane(
115 "regional-private-cited-review", "regional-private", frozenset({DataClass.TENANT_PRIVATE}),
116 64_000, True, True, True, Decimal("0.004560"), 1_300,
117 ),
118 Lane(
119 "cheap-text-fallback", "hosted-cheap", frozenset({DataClass.PUBLIC, DataClass.TENANT_PRIVATE}),
120 32_000, False, False, False, Decimal("0.001500"), 420,
121 ),
122]
123
124print(f"cost_release={COST_RELEASE_ID}")
125print(f"gateway_policy={GATEWAY_POLICY_ID}")
126print(f"required_answer_schema={REQUIRED_ANSWER_SCHEMA}")
127print(f"max_generated_answer_usd={MAX_GENERATED_ANSWER_USD}")
128print(f"max_generation_attempts={MAX_GENERATION_ATTEMPTS}")
129print(f"request_deadline_ms={REQUEST_DEADLINE_MS}")
130print(f"registered_lanes={len(LANES)}")1cost_release=support-release-2026-05-cost-v1
2gateway_policy=gateway-policy-v1
3required_answer_schema=cited-support-answer-v3
4max_generated_answer_usd=0.005000
5max_generation_attempts=2
6request_deadline_ms=2500
7registered_lanes=6Compile requirements before choosing a lane
The bug fits in two early returns:
1if request.data_class == "tenant_private":
2 return "local-private"
3if request.risk_amount_cents >= 50_000:
4 return "human-review"For a private break-glass access request with 900 USD of risk, the first return silently drops the second requirement. Give each stage one job:
- Convert every request fact into one contract.
- Drop every lane that violates any contract field.
- Rank only the remaining lanes by measured preferences such as cost or latency.
Predict access-R900 before running the compiler: its risk requires human review, its private notes require a private lane, and its answer still needs cited-support-answer-v3. The next cell compiles those requirements. Neither check erases the other, and needs_schema means that exact schema, not "any JSON."
These records come from a trusted application boundary, not unchecked JSON from a client. Authenticate the caller, authorize the task and tenant, and derive classification and risk from server-side policy before constructing one. A client-supplied data_class="public" can't declassify private notes. The compiler below also forces private handling and citations for the access task, even if those flags are weakened accidentally.
The single context_tokens count is a fixture simplification. Tokenizers and prompt wrappers differ by lane. At dispatch, tokenize each candidate's complete input, including tools and evidence, using that lane's tokenizer; reserve output and other tokens against its actual limits. Recheck region, retention policy, credential scope, and registry revision too. A private-data Boolean can't express all those conditions.
1HIGH_RISK_ACCESS_CENTS = 50_000
2
3def compile_contract(request: GatewayRequest) -> RouteContract:
4 if type(request) is not GatewayRequest:
5 raise ValueError("need a validated gateway request")
6 if request.task not in {"prod_access_decision", "deploy_policy"}:
7 raise ValueError("unknown task")
8 access_task = request.task == "prod_access_decision"
9 return RouteContract(
10 request_id=request.request_id,
11 data_class=DataClass.TENANT_PRIVATE if access_task else request.data_class,
12 context_tokens=request.context_tokens,
13 needs_citations=access_task or request.requires_citations,
14 # The inherited release contract requires its schema on every route.
15 # A caller can't opt out by setting requires_schema=False.
16 needs_schema=True,
17 needs_human_review=request.risk_amount_cents >= HIGH_RISK_ACCESS_CENTS,
18 max_answer_cost_usd=MAX_GENERATED_ANSWER_USD,
19 )
20
21private_access = GatewayRequest(
22 request_id="access-R900",
23 task="prod_access_decision",
24 data_class=DataClass.TENANT_PRIVATE,
25 context_tokens=24_000,
26 risk_amount_cents=90_000,
27 requires_citations=True,
28)
29private_contract = compile_contract(private_access)
30schema_downgrade = GatewayRequest(
31 "schema-bypass",
32 "prod_access_decision",
33 DataClass.TENANT_PRIVATE,
34 2_000,
35 requires_schema=False,
36)
37assert compile_contract(schema_downgrade).needs_schema is True
38
39print(f"request={private_contract.request_id}")
40print(f"data_class={private_contract.data_class.value}")
41print(f"needs_citations={private_contract.needs_citations}")
42print(f"needs_human_review={private_contract.needs_human_review}")
43print(f"max_answer_cost_usd={private_contract.max_answer_cost_usd}")1request=access-R900
2data_class=tenant_private
3needs_citations=True
4needs_human_review=True
5max_answer_cost_usd=0.005000The output preserves every requirement. Now make each lane explain its rejection instead of returning a bare Boolean. If no lane survives, an operator needs to know whether the missing capability is private-data handling, context length, citations, review, or budget.
Before reading the rows, predict the result for access-R900: three private cited-review lanes should survive; cheap text should fail schema, citations, and review.
1def contract_violations(lane: Lane, contract: RouteContract) -> list[str]:
2 if type(lane) is not Lane or type(contract) is not RouteContract:
3 raise ValueError("need a validated lane and contract")
4 failures: list[str] = []
5 if contract.data_class not in lane.allowed_data_classes:
6 failures.append("data_boundary")
7 if lane.max_context_tokens < contract.context_tokens:
8 failures.append("context_length")
9 if contract.needs_schema and not lane.supports_schema:
10 failures.append("schema")
11 if contract.needs_citations and not lane.supports_citations:
12 failures.append("citations")
13 if contract.needs_human_review and not lane.supports_human_review:
14 failures.append("human_review")
15 if lane.evaluated_answer_cost_usd > contract.max_answer_cost_usd:
16 failures.append("budget")
17 return failures
18
19def compatible_lanes(contract: RouteContract) -> list[Lane]:
20 return [lane for lane in LANES if not contract_violations(lane, contract)]
21
22for lane in LANES:
23 failures = contract_violations(lane, private_contract)
24 status = "compatible" if not failures else "reject=" + ",".join(failures)
25 print(f"{lane.name}: {status}")1fast-public-json: reject=data_boundary,context_length,citations,human_review
2public-cited-review: reject=data_boundary
3primary-private-cited-review: compatible
4local-private-cited-review: compatible
5regional-private-cited-review: compatible
6cheap-text-fallback: reject=schema,citations,human_reviewThree private cited-review lanes survive the supplied fields. Cheap text looks tempting: its declared data classes include tenant-private and its cost estimate is below 0.005000 USD. It still fails schema, citations, and review. An estimated cost isn't a charge bound; the later executor demonstrates reservations separately.
💡 Key insight: Budget and a private-data flag aren't a contract.
cheap-text-fallbackfails three hard fields at once.
The gateway can rank survivors by a soft preference without weakening any requirement. Here it chooses lower fixture answer cost, then latency as a deterministic tie-breaker.
Expect docs-Q102 to take the fast public lane and access-R900 to take the lowest-cost private cited-review lane. The next cell makes that choice explicit and records its reasons.
1@dataclass(frozen=True)
2class RouteDecision:
3 request_id: str
4 lane: str | None
5 action: str
6 reasons: tuple[str, ...]
7
8def choose_primary(contract: RouteContract) -> RouteDecision:
9 feasible = compatible_lanes(contract)
10 if not feasible:
11 return RouteDecision(contract.request_id, None, "escalate", ("no_compatible_lane",))
12 lane = min(feasible, key=lambda candidate: (candidate.evaluated_answer_cost_usd, candidate.expected_latency_ms, candidate.name))
13 reasons = (
14 f"data={contract.data_class.value}",
15 f"review={str(contract.needs_human_review).lower()}",
16 f"citations={str(contract.needs_citations).lower()}",
17 f"budget<={contract.max_answer_cost_usd}",
18 )
19 return RouteDecision(contract.request_id, lane.name, "generate", reasons)
20
21status_request = GatewayRequest(
22 "docs-Q102", "deploy_policy", DataClass.PUBLIC, 2_000,
23 requires_citations=False,
24)
25
26for request in (status_request, private_access):
27 decision = choose_primary(compile_contract(request))
28 print(f"{decision.request_id} -> {decision.lane} action={decision.action}")
29 print(" " + " ".join(decision.reasons))1docs-Q102 -> fast-public-json action=generate
2 data=public review=false citations=false budget<=0.005000
3access-R900 -> primary-private-cited-review action=generate
4 data=tenant_private review=true citations=true budget<=0.005000
Let learned routing rank only legal lanes
The filter above is deliberately rule-based. Data boundaries, schema requirements, review requirements, and approved cost ceilings are policy constraints. A probabilistic classifier that relaxes any of them would make a plausible prediction more important than authorization.
Learned routing still has a job inside the feasible set. Consider three preferences:
- Task quality: Evaluate candidate models on your actual lookup, reconciliation, and reasoning cases. Parameter count or a "frontier" label doesn't establish the best route. RouteLLM learns from preference data and reports more than a twofold cost reduction in certain benchmark cases without reducing its measured quality. That isn't a guarantee for this private access workflow or a new model pair.[4]
- Latency: Set time-to-first-token (TTFT) and complete-answer targets for the product, then measure realistic load. There's no universal sub-500 ms requirement or hosted-only solution. Batch work has a different completion window; minutes of acceptable delay alone don't make a workload eligible for a 24-hour Batch path.
- Spend and quotas: Metering helps choose a cheaper approved lane, defer work, or reject it under a declared policy. A 90% budget warning is a policy choice, not a hard cap. Reservations, settlement, and atomic account-level admission are what prevent concurrent calls from overspending an enforceable bound.
Treat learned routers and heuristics as ranking engines over already-legal lanes. They're never a replacement for privacy, context, or schema validation gates.
The control boundary has two paths. A soft router may rank already-compatible candidates. A failure re-enters the same contract filter instead of jumping to a convenient provider.

A low-cost learned router recommends a public lane for a request containing private incident notes. What should happen?
Answer
The deterministic contract filter rejects the lane before any ranking decision can use it. Learned ranking can choose only among lanes already approved for the data boundary and required output contract.
Make fallback another contract decision
When a primary lane fails before output, the gateway may get a second chance to generate. It doesn't get a second contract. A cited access answer must remain private, structured, reviewable, and admitted under the same cost policy during an outage.
Gateway products expose fallback mechanisms, but a config file doesn't prove semantic compatibility. LiteLLM documents ordered fallbacks and separate chains for ordinary errors, context-window errors, and content-policy errors. Those knobs decide when another model group may be tried; your contract still decides which replacement is acceptable. Its allowed_fails and cooldown_time settings can temporarily remove failing deployments. One documented exception matters: fallback to a specific model ID bypasses cooldown checks for that replacement. Test the exact configured path before relying on its breaker behavior.[5]
⚠️ Common mistake: Copying an ordered fallback list from a proxy config and treating every backup as legal. The proxy will happily fail over. The contract still has to admit that lane.
Classify the failure before the gateway looks for a candidate. A transient rate limit or timeout before output is shown may be retried against a compatible lane if attempts, time, and spend remain. A quota-exhausted account isn't the same as a temporary throttle. Authentication failures and policy refusals stop here; switching credentials or providers to bypass them isn't recovery. Once a model has streamed visible text, silently switching models can produce a contradictory continuation. Stop that answer and expose an explicit retry or handoff.
This lab retries answer generation separately from tool writes. Generation isn't generally a pure deterministic function: sampling can vary, and hosted state, storage, or enabled tools can have effects beyond token production. A retry requires a known generation-only path and compatible conversation/input state.
Suppose an access-grant tool commits a write, then the explanation times out. Replaying the whole turn can duplicate the grant. Preserve its verified receipt, recheck authorization, and retry only the explanation when allowed. Write services need scoped operation identities, parameter comparison, retained results, and atomic commit or reconciliation. A pre-output timeout alone says nothing about whether a write committed.[6]
A context rejection also stops here. The filter already checked capacity, so a provider rejection means stale registry data or another mismatch. Investigate it instead of hiding it behind a downgrade.
Some production gateways retry a different schema-capable lane after a validation miss, as long as the caller never saw the bad payload. This lab escalates schema_invalid so a broken supports_schema flag can't hide behind another attempt. Cheap text is illegal either way.
The next cell makes the transparent-retry boundary explicit:
1class FailureKind(str, Enum):
2 RATE_LIMIT_BEFORE_OUTPUT = "rate_limit_before_output"
3 TIMEOUT_BEFORE_OUTPUT = "timeout_before_output"
4 QUOTA_EXHAUSTED = "quota_exhausted"
5 CONTEXT_REJECTED = "context_rejected"
6 MID_STREAM_DROP = "mid_stream_drop"
7 SCHEMA_INVALID = "schema_invalid"
8 AUTH_DENIED = "auth_denied"
9 POLICY_REFUSAL = "policy_refusal"
10
11def may_retry_transparently(failure: FailureKind) -> bool:
12 return type(failure) is FailureKind and failure in {
13 FailureKind.RATE_LIMIT_BEFORE_OUTPUT,
14 FailureKind.TIMEOUT_BEFORE_OUTPUT,
15 }
16
17for failure in FailureKind:
18 action = "try_compatible_fallback" if may_retry_transparently(failure) else "stop_or_escalate"
19 print(f"{failure.value}: {action}")1rate_limit_before_output: try_compatible_fallback
2timeout_before_output: try_compatible_fallback
3quota_exhausted: stop_or_escalate
4context_rejected: stop_or_escalate
5mid_stream_drop: stop_or_escalate
6schema_invalid: stop_or_escalate
7auth_denied: stop_or_escalate
8policy_refusal: stop_or_escalateFor access-R900, suppose the lower-cost primary lane times out before emitting output. The fallback selector excludes the failed lane, applies the same contract filter, and chooses from what remains.
Predict the result before looking at the output: local private should serve, regional private should remain available, and cheap text should still be rejected.
1def choose_fallback(contract: RouteContract, failed_lane: str) -> RouteDecision:
2 if not owned_text(failed_lane) or failed_lane not in {lane.name for lane in LANES}:
3 raise ValueError("need a known failed lane")
4 candidates = [
5 lane for lane in compatible_lanes(contract)
6 if lane.name != failed_lane
7 ]
8 if not candidates:
9 return RouteDecision(contract.request_id, None, "escalate", ("fallback_contract_unmet",))
10 lane = min(candidates, key=lambda candidate: (candidate.evaluated_answer_cost_usd, candidate.expected_latency_ms, candidate.name))
11 return RouteDecision(
12 contract.request_id,
13 lane.name,
14 "fallback_generate",
15 (f"primary_failed={failed_lane}", "contract_preserved=true"),
16 )
17
18primary = choose_primary(private_contract)
19fallback = choose_fallback(private_contract, primary.lane or "")
20print(f"primary={primary.lane}")
21print(f"fallback={fallback.lane} action={fallback.action}")
22print("cheap_text_rejected=" + ",".join(contract_violations(LANES[-1], private_contract)))1primary=primary-private-cited-review
2fallback=local-private-cited-review action=fallback_generate
3cheap_text_rejected=schema,citations,human_reviewBoth backup estimates fit the single-generation ceiling. That doesn't bound a future call's bill, and a failed primary can still consume work before its timeout reaches the gateway. A maximum for one call also doesn't cap cumulative task spending.
Keep two numbers separate: the admission ceiling used to filter lanes, and the request-total spend that sums usage from every attempt, failed primary plus fallback. Production policy also needs a retry budget: a maximum attempt count, one wall-clock deadline shared across attempts, and post-call pricing for usage reported by every provider attempt.
A critical invariant across sequential attempts is cumulative deadline propagation:
With a 2,500 ms total deadline and 1,800 ms already spent, only 700 ms remain. Giving fallback a fresh 2,500 ms timeout would permit 4,300 ms of attempt time, exceeding the original deadline. Real elapsed time also includes queueing, policy checks, retries, and adapter overhead. Use one monotonic deadline and carry the remaining budget into cancellation and timeout handling. A socket read timeout alone may reset between chunks; it isn't necessarily a total-call deadline, and local cancellation doesn't prove remote work stopped.
This lab allows at most one fallback (MAX_GENERATION_ATTEMPTS = 2) within a shared 2.5-second deadline. The executor below advances a simulated clock, including retry delay. If the primary consumes 1,800 ms, only 700 ms remain, not another 2,500 ms.

A hard request-spend cap needs more than estimated lane cost: reserve a conservative charge bound before each attempt, reconcile actual usage, and retain reservations for timed-out calls whose usage is unknown. Concurrent requests need atomic reservations at the tenant/account boundary. The fixture uses a separate 0.010000 USD request cap and a 0.005000 USD reservation per attempt. Real bounds must include all billable token categories and enabled tools; an estimate or typical cost isn't such a bound.
There's another outage cost. If every request still tries the sick primary first, you pay for that failed attempt over and over. The next section stops that herd.
The primary private lane times out. A cheap lane is healthy and under budget, but it returns plain text without citations or human-review support. Is it an acceptable fallback?
Answer
No. Budget is only one contract field. The gateway must reject the cheap lane because it drops required schema, citations, and review handling.
Keep an outage from multiplying
The fallback path protects one request. Under load, every request can still poke an unhealthy primary before moving to the backup. When an upstream provider suffers degraded capacity, multiple concurrent requests time out simultaneously. If clients retry naively without backoff or coordination, traffic spikes multiplicatively against an already struggling upstream, turning a temporary slowdown into a total outage.
To break that feedback loop, resilient gateways combine two core patterns:
- Jittered exponential backoff: Fixed retry intervals cause clients to synchronize their retry attempts into destructive pulses. Full jittered backoff distributes delays randomly between zero and an exponential ceiling: Jitter reduces synchronized retry clusters; it doesn't guarantee smooth traffic or solve overload by itself. The exponential index starts at zero for the first retry in this convention.[7]
- Circuit breakers: A circuit breaker suppresses normal calls after chosen health failures cross a threshold. A compatible healthy replacement may exist; otherwise escalate. After cooldown, this policy allows one probe. Other implementations allow a bounded probe set. Probe success can close the circuit; failure reopens it. Define health failures separately from authentication errors, refusals, or unsupported contracts.
This compact implementation models the three states: closed, open, and half_open. Its key is provider because each fixture provider names one upstream target.
A production registry may need provider, endpoint, deployment, or lane keys depending on failure scope. This local breaker rejects malformed clocks and expires an unanswered probe. It has no synchronization or result lease: concurrent workers can race, and an old completion can overwrite newer state. Shared breakers need atomic probe leases, generations that reject stale results, bounded backoff, and actual probe cancellation. A cooldown timer alone isn't proof that an upstream recovered.
1class CircuitStatus(str, Enum):
2 CLOSED = "closed"
3 OPEN = "open"
4 HALF_OPEN = "half_open"
5
6@dataclass
7class CircuitState:
8 status: CircuitStatus = CircuitStatus.CLOSED
9 failures: int = 0
10 opened_until: float = 0.0
11 probe_until: float = 0.0
12
13class CircuitBreaker:
14 def __init__(self, threshold: int = 2, cooldown_seconds: float = 10.0,
15 probe_timeout_seconds: float = 2.5) -> None:
16 if not integer_count(threshold) or threshold == 0 or not seconds(cooldown_seconds) or cooldown_seconds == 0 or not seconds(probe_timeout_seconds) or probe_timeout_seconds == 0:
17 raise ValueError("positive finite threshold, cooldown, and probe lifetime required")
18 self.threshold = threshold
19 self.cooldown_seconds = cooldown_seconds
20 self.probe_timeout_seconds = probe_timeout_seconds
21 self.states: dict[str, CircuitState] = {}
22
23 def permit(self, provider: str, now: float) -> bool:
24 if not owned_text(provider) or not seconds(now) or not seconds(now + self.cooldown_seconds) or not seconds(now + self.probe_timeout_seconds):
25 return False
26 state = self.states.setdefault(provider, CircuitState())
27 if state.status == CircuitStatus.HALF_OPEN:
28 if now >= state.probe_until:
29 state.status = CircuitStatus.OPEN
30 state.opened_until = now + self.cooldown_seconds
31 return False
32 if state.status == CircuitStatus.OPEN:
33 if now < state.opened_until:
34 return False
35 state.status = CircuitStatus.HALF_OPEN
36 state.probe_until = now + self.probe_timeout_seconds
37 return True
38 return state.status == CircuitStatus.CLOSED
39
40 def failure(self, provider: str, now: float) -> None:
41 if not owned_text(provider) or not seconds(now) or not seconds(now + self.cooldown_seconds):
42 raise ValueError("need a named target and finite nonnegative clock")
43 state = self.states.setdefault(provider, CircuitState())
44 state.failures += 1
45 if state.status == CircuitStatus.HALF_OPEN or state.failures >= self.threshold:
46 state.status = CircuitStatus.OPEN
47 state.opened_until = now + self.cooldown_seconds
48
49 def success(self, provider: str) -> None:
50 if not owned_text(provider):
51 raise ValueError("need a named target")
52 self.states[provider] = CircuitState()
53
54breaker = CircuitBreaker()
55breaker.failure("hosted-private", 100.0)
56breaker.failure("hosted-private", 101.0)
57print(f"during_cooldown={breaker.permit('hosted-private', 105.0)}")
58print(f"probe_after_cooldown={breaker.permit('hosted-private', 112.0)}")
59breaker.success("hosted-private")
60print(f"after_success={breaker.states['hosted-private'].status.value}")
61
62unanswered_probe = CircuitBreaker(threshold=1)
63unanswered_probe.failure("probe-target", 100.0)
64assert unanswered_probe.permit("probe-target", 110.0)
65assert not unanswered_probe.permit("probe-target", 110.1)
66assert not unanswered_probe.permit("probe-target", 112.5)
67assert unanswered_probe.states["probe-target"].status is CircuitStatus.OPEN
68print(f"expired_probe={unanswered_probe.states['probe-target'].status.value}")1during_cooldown=False
2probe_after_cooldown=True
3after_success=closed
4expired_probe=openWhat should an open circuit breaker do after a provider crosses its failure threshold?
Answer
Stop sending normal traffic to that lane for a bounded period, then allow controlled probes before closing the circuit. Continuing ordinary retries would amplify latency and provider load during the outage.
Execute one outage path and record why
To investigate an incident, retain enough evidence to reconstruct lane selection: policy and registry revisions, approved cost artifact, effective lane/model/adapter, contract fields, attempt/task identities, status, and pricing evidence. A string ID or an audit row alone doesn't prove that the decision or its provenance is truthful.
That record lets a later review distinguish a model-quality problem from a bad policy decision. The next cell follows one private request through timeout, partial output, and circuit cooldown.
The executor consumes scripted outcomes instead of making HTTP calls. Its defaults assume successful completion with fixed durations and charges; failure overrides let you break that assumption. Each attempted lane advances the simulated clock and contributes its supplied charge. An open circuit consumes no provider attempt or charge in this fixture. served means transport completion only; no payload has yet passed reply validation or human review.
In a real adapter, propagate the remaining deadline to connection, read, and total-call timeouts; disable hidden SDK retries or charge them to the same attempt budget. Cancellation doesn't prove the remote provider stopped generating or billing. Honor applicable Retry-After delays and use bounded backoff with jitter when retrying under load. The fixture accepts an explicit delay so that boundary is testable.[8]
1@dataclass(frozen=True)
2class RouteEvent:
3 request_id: str
4 policy_id: str
5 cost_release_id: str
6 action: str
7 lane: str | None
8 contract_summary: str
9 reason: str
10 evaluated_cost_usd: Decimal
11 attempted_lanes: tuple[str, ...] = ()
12 elapsed_ms: int = 0
13 fixture_spend_usd: Decimal = Decimal("0")
14
15def lane_by_name(name: str) -> Lane:
16 return next(lane for lane in LANES if lane.name == name)
17
18def summarize_contract(contract: RouteContract) -> str:
19 return (
20 f"data={contract.data_class.value};"
21 f"context_tokens={contract.context_tokens};"
22 f"schema={str(contract.needs_schema).lower()};"
23 f"schema_id={REQUIRED_ANSWER_SCHEMA};"
24 f"citations={str(contract.needs_citations).lower()};"
25 f"review={str(contract.needs_human_review).lower()};"
26 f"budget<={contract.max_answer_cost_usd}"
27 )
28
29@dataclass(frozen=True)
30class Attempt:
31 duration_ms: int
32 charge_usd: Decimal
33 failure: FailureKind | None = None
34
35 def __post_init__(self) -> None:
36 if not integer_count(self.duration_ms) or not money(self.charge_usd):
37 raise ValueError("need a nonnegative integer duration and finite Decimal charge")
38 if self.failure is not None and type(self.failure) is not FailureKind:
39 raise ValueError("need a known failure or explicit completion")
40
41def execute_with_failure(
42 request: GatewayRequest,
43 failure: FailureKind | None,
44 now: float = 200.0,
45 scripted: dict[str, Attempt] | None = None,
46 retry_delay_ms: int = 0,
47 request_cap: Decimal = MAX_REQUEST_SPEND_USD,
48) -> RouteEvent:
49 if failure is not None and type(failure) is not FailureKind:
50 raise ValueError("need a classified failure or explicit completion")
51 if not seconds(now):
52 raise ValueError("need a finite nonnegative fixture clock")
53 if not money(request_cap):
54 raise ValueError("request cap must be a finite nonnegative Decimal")
55 if scripted is None:
56 scripted = {}
57 if type(scripted) is not dict or any(
58 not owned_text(name) or name not in {lane.name for lane in LANES}
59 or type(attempt) is not Attempt or attempt.charge_usd > ATTEMPT_RESERVATION_USD
60 for name, attempt in scripted.items()
61 ):
62 raise ValueError("need known scripted lanes and charges within the reserved bound")
63 contract = compile_contract(request)
64 contract_summary = summarize_contract(contract)
65 candidates = sorted(
66 compatible_lanes(contract),
67 key=lambda lane: (lane.evaluated_answer_cost_usd, lane.expected_latency_ms, lane.name),
68 )
69 if type(retry_delay_ms) is not int or retry_delay_ms < 0:
70 raise ValueError("retry delay must be nonnegative integer milliseconds")
71 # Synthetic known charges, distinct from registry cost estimates.
72 charges = dict(zip((lane.name for lane in LANES), map(Decimal, (
73 "0.001100", "0.003800", "0.004200", "0.004500", "0.004560", "0.001500",
74 ))))
75 attempted: list[str] = []
76 elapsed = 0
77 spend = Decimal("0")
78 reason = "no_primary_lane" if not candidates else "no_permitted_compatible_lane"
79
80 def event(action: str, lane: Lane | None = None) -> RouteEvent:
81 return RouteEvent(
82 request.request_id, GATEWAY_POLICY_ID, COST_RELEASE_ID, action,
83 lane.name if lane else None, contract_summary, reason,
84 lane.evaluated_answer_cost_usd if lane else Decimal("0"),
85 tuple(attempted), elapsed, spend,
86 )
87
88 for index, lane in enumerate(candidates):
89 if elapsed >= REQUEST_DEADLINE_MS:
90 reason = "deadline_exhausted"
91 break
92 if len(attempted) >= MAX_GENERATION_ATTEMPTS:
93 reason = "attempts_exhausted"
94 break
95 if spend + ATTEMPT_RESERVATION_USD > request_cap:
96 reason = "request_spend_reservation_denied"
97 break
98 if not breaker.permit(lane.provider, now + elapsed / 1000):
99 if index == 0:
100 reason = "primary_circuit_open"
101 continue
102 attempt = scripted.get(lane.name, Attempt(
103 lane.expected_latency_ms, charges[lane.name], failure if index == 0 else None,
104 ))
105 attempted.append(lane.name)
106 remaining = REQUEST_DEADLINE_MS - elapsed
107 elapsed += min(attempt.duration_ms, remaining)
108 spend += attempt.charge_usd
109 outcome = attempt.failure
110 if attempt.duration_ms >= remaining:
111 outcome = FailureKind.TIMEOUT_BEFORE_OUTPUT
112 if outcome is None:
113 breaker.success(lane.provider)
114 reason = "primary_contract_match" if index == 0 else reason + ";contract_preserved"
115 return event("served" if index == 0 else "served_fallback", lane)
116 # Auth, refusal, and schema failures aren't provider-health evidence here.
117 if outcome in {FailureKind.TIMEOUT_BEFORE_OUTPUT, FailureKind.MID_STREAM_DROP}:
118 breaker.failure(lane.provider, now + elapsed / 1000)
119 elif breaker.states[lane.provider].status == CircuitStatus.HALF_OPEN:
120 breaker.success(lane.provider) # reachable, even if response was unusable
121 reason = ("primary_" if index == 0 else "fallback_") + outcome.value
122 if not may_retry_transparently(outcome):
123 return event("escalate")
124 elapsed += min(retry_delay_ms, REQUEST_DEADLINE_MS - elapsed)
125 return event("escalate")
126
127breaker = CircuitBreaker()
128before_output = execute_with_failure(private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT, now=200.0)
129mid_stream = execute_with_failure(private_access, FailureKind.MID_STREAM_DROP, now=203.0)
130breaker.failure("local-private", 204.0)
131breaker.failure("local-private", 205.0)
132skip_open_local = execute_with_failure(private_access, None, now=206.0)
133
134# These are scripted charges, not provider-reported usage.
135primary_timeout_usage = Decimal("0.004200")
136fallback_usage = Decimal("0.004500")
137request_attempt_costs = [primary_timeout_usage, fallback_usage]
138request_total_spend = sum(request_attempt_costs, Decimal("0"))
139assert before_output.evaluated_cost_usd == Decimal("0.004500")
140assert request_total_spend == Decimal("0.008700")
141assert request_total_spend > MAX_GENERATED_ANSWER_USD
142assert before_output.fixture_spend_usd == request_total_spend
143assert before_output.elapsed_ms == 2_040
144
145print(f"before_output={before_output.action} lane={before_output.lane} reason={before_output.reason}")
146print(f"audit_policy={before_output.policy_id} cost_release={before_output.cost_release_id}")
147print(f"audit_contract={before_output.contract_summary}")
148print(f"mid_stream={mid_stream.action} lane={mid_stream.lane} reason={mid_stream.reason}")
149print(f"circuit_after_failures={breaker.states['hosted-private'].status.value}")
150print(f"open_local_skipped={skip_open_local.action} lane={skip_open_local.lane} reason={skip_open_local.reason}")
151print(f"attempt_costs={','.join(str(c) for c in request_attempt_costs)}")
152print(f"request_total_spend={request_total_spend} admission_ceiling={MAX_GENERATED_ANSWER_USD}")1before_output=served_fallback lane=local-private-cited-review reason=primary_timeout_before_output;contract_preserved
2audit_policy=gateway-policy-v1 cost_release=support-release-2026-05-cost-v1
3audit_contract=data=tenant_private;context_tokens=24000;schema=true;schema_id=cited-support-answer-v3;citations=true;review=true;budget<=0.005000
4mid_stream=escalate lane=None reason=primary_mid_stream_drop
5circuit_after_failures=open
6open_local_skipped=served_fallback lane=regional-private-cited-review reason=primary_circuit_open;contract_preserved
7attempt_costs=0.004200,0.004500
8request_total_spend=0.008700 admission_ceiling=0.005000Replay policy before promoting it
Three hand-picked examples can't establish a gateway policy. Start with a mechanical replay of low-risk traffic, high-risk cases, private data, outage paths, and unsupported contracts. For every simulated success, check that the selected lane preserves the contract and stays inside the admission ceiling. That doesn't establish answer quality.
The replay below adds a request with a private context too large for any approved private lane. It also opens the primary circuit on a simulated timeout, then checks that the next request uses a contract-preserving fallback during cooldown. Unsupported work escalates instead of silently truncating evidence or sending private notes somewhere unapproved.
1too_large_private_request = GatewayRequest(
2 "access-long-context", "prod_access_decision", DataClass.TENANT_PRIVATE, 70_000,
3 risk_amount_cents=90_000, requires_citations=True,
4)
5
6breaker = CircuitBreaker(threshold=1, cooldown_seconds=10.0)
7replay_cases = [
8 (status_request, None),
9 (private_access, None),
10 (private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT),
11 (private_access, None),
12 (too_large_private_request, None),
13]
14events = [
15 execute_with_failure(request, failure, now=300.0 + 3 * index)
16 for index, (request, failure) in enumerate(replay_cases)
17]
18
19served = []
20expected_actions = ["served", "served", "served_fallback", "served_fallback", "escalate"]
21for (request, _), event, expected in zip(replay_cases, events, expected_actions):
22 assert event.action == expected
23 assert event.elapsed_ms <= REQUEST_DEADLINE_MS
24 assert len(event.attempted_lanes) <= MAX_GENERATION_ATTEMPTS
25 assert event.fixture_spend_usd <= MAX_REQUEST_SPEND_USD
26 if event.action.startswith("served"):
27 lane = lane_by_name(event.lane or "")
28 assert not contract_violations(lane, compile_contract(request))
29 assert event.evaluated_cost_usd <= MAX_GENERATED_ANSWER_USD
30 served.append(event)
31
32for event in events:
33 print(f"{event.request_id}: {event.action} lane={event.lane}")
34route_field_violations = sum(
35 len(contract_violations(lane_by_name(event.lane), compile_contract(request)))
36 for (request, _), event in zip(replay_cases, events)
37 if event.action.startswith("served")
38)
39assert route_field_violations == 0
40print(f"compatible_transport_completions={len(served)}/{len(events)}")
41print(f"route_field_violations={route_field_violations}")1docs-Q102: served lane=fast-public-json
2access-R900: served lane=primary-private-cited-review
3access-R900: served_fallback lane=local-private-cited-review
4access-R900: served_fallback lane=local-private-cited-review
5access-long-context: escalate lane=None
6compatible_transport_completions=4/5
7route_field_violations=0The replay reads no answers. Its cost assertion compares supplied lane estimates; it doesn't verify a future charge bound or a provider bill. The zero violation count covers these five mechanical cases, not population safety.
After each real attempt, normalize its usage and price it with that lane's effective model/tier/region rate card using the preceding lesson's ledger mechanics. Pin the card's contents and retain unknown usage as unresolved. Cross-provider fallback neither transfers a prefix cache nor guarantees the replacement has no independently warm prefix. Before promotion, join route events with answer correctness, citation support, and measured spending.
A routing policy can satisfy the mechanical contract and still send too many difficult public questions to a weak but formally compatible lane.
Only after hard requirements pass should you test learned or heuristic ranking for quality and cost. Useful comparison metrics are generated-answer correctness, citation correctness, human-escalation rate, fallback success rate, p95 latency, and actual spend by lane.
Now break the runtime, not just the route table. Each case starts with a fresh breaker. Two failed attempts must stop, a slow primary must leave less time for its fallback, and a request cap must reject a reservation even when each lane passes admission.
1primary_name = "primary-private-cited-review"
2local_name = "local-private-cited-review"
3cases = {
4 "fallback_also_fails": ({
5 local_name: Attempt(500, Decimal("0.002000"), FailureKind.TIMEOUT_BEFORE_OUTPUT),
6 }, 0, MAX_REQUEST_SPEND_USD, "attempts_exhausted", 2),
7 "shared_deadline": ({
8 primary_name: Attempt(1_800, Decimal("0.004200"), FailureKind.TIMEOUT_BEFORE_OUTPUT),
9 }, 0, MAX_REQUEST_SPEND_USD, "deadline_exhausted", 2),
10 "retry_delay_uses_deadline": ({}, 2_000, MAX_REQUEST_SPEND_USD, "deadline_exhausted", 1),
11 "request_cap": ({}, 0, Decimal("0.008000"), "request_spend_reservation_denied", 1),
12 "auth_failure": ({
13 primary_name: Attempt(20, Decimal("0"), FailureKind.AUTH_DENIED),
14 }, 0, MAX_REQUEST_SPEND_USD, "primary_auth_denied", 1),
15}
16for name, (scripted, delay, cap, reason, attempts) in cases.items():
17 breaker = CircuitBreaker()
18 result = execute_with_failure(
19 private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT,
20 scripted=scripted, retry_delay_ms=delay, request_cap=cap,
21 )
22 assert result.action == "escalate" and result.reason == reason
23 assert len(result.attempted_lanes) == attempts
24 assert result.elapsed_ms <= REQUEST_DEADLINE_MS
25 assert result.fixture_spend_usd <= cap
26 print(f"{name}: {reason}, attempts={attempts}, elapsed_ms={result.elapsed_ms}")1fallback_also_fails: attempts_exhausted, attempts=2, elapsed_ms=1440
2shared_deadline: deadline_exhausted, attempts=2, elapsed_ms=2500
3retry_delay_uses_deadline: deadline_exhausted, attempts=1, elapsed_ms=2500
4request_cap: request_spend_reservation_denied, attempts=1, elapsed_ms=940
5auth_failure: primary_auth_denied, attempts=1, elapsed_ms=20The 0.008000 USD case stops after spending 0.004200 USD: another 0.005000 USD reservation would exceed the cap. Knowing a fallback's typical charge is 0.004500 USD doesn't make it safe to reserve less than the bound. All charges are known in this fixture; unknown real usage must stay reserved until reconciled.
Validate the reply, not just the lane
A green lane row isn't a green answer. Adapters must normalize provider-specific text blocks, tool-call identifiers, stop reasons, refusals, and usage into an explicit application result. A truncated JSON object isn't success. Neither is a valid object containing a made-up citation or a model-authored claim that a human approved it.
The next cell checks a small local reply shape after adapter normalization. It isn't the full inherited cited-support-answer-v3 schema. We assume the application supplies currently authorized evidence IDs for this request; this function doesn't resolve permissions or fetch passages. Membership catches unknown IDs, not unsupported wording. Likewise, review="pending" is an expected marker, not proof of an actual review queue or human approval.
1def validate_reply(payload: object, contract: RouteContract, evidence_ids: frozenset[str]) -> dict:
2 if type(contract) is not RouteContract or type(evidence_ids) is not frozenset or any(not owned_text(item) for item in evidence_ids):
3 raise ValueError("need a validated contract and owned evidence identifiers")
4 fields = {"schema", "status", "answer", "citations", "review"}
5 if type(payload) is not dict or set(payload) != fields:
6 raise ValueError("unexpected reply shape")
7 if payload["schema"] != REQUIRED_ANSWER_SCHEMA or payload["status"] != "complete":
8 raise ValueError("wrong schema, refusal, or incomplete reply")
9 if type(payload["answer"]) is not str or not payload["answer"].strip():
10 raise ValueError("missing answer")
11 citations = payload["citations"]
12 if type(citations) is not list or any(type(item) is not str for item in citations):
13 raise ValueError("invalid citations")
14 if len(set(citations)) != len(citations) or not set(citations) <= evidence_ids:
15 raise ValueError("duplicate or unknown citation")
16 if contract.needs_citations and not citations:
17 raise ValueError("citations required")
18 # Require the marker; a real workflow must separately create/verify handoff state.
19 expected_review = "pending" if contract.needs_human_review else "not_required"
20 if payload["review"] != expected_review:
21 raise ValueError("invalid review state")
22 return payload | {"citations": citations.copy()}
23
24good_reply = {
25 "schema": REQUIRED_ANSWER_SCHEMA, "status": "complete",
26 "answer": "Send this access request for human review.",
27 "citations": ["E1"], "review": "pending",
28}
29assert validate_reply(good_reply, private_contract, frozenset({"E1"})) == good_reply
30for name, patch in {
31 "refused": {"status": "refused"},
32 "truncated": {"status": "length"},
33 "wrong_schema": {"schema": "plain-json-v1"},
34 "invented_citation": {"citations": ["E99"]},
35 "self_approved": {"review": "approved"},
36}.items():
37 try:
38 validate_reply(good_reply | patch, private_contract, frozenset({"E1"}))
39 except ValueError:
40 print(f"{name}: blocked before publication")
41 else:
42 raise AssertionError(name)1refused: blocked before publication
2truncated: blocked before publication
3wrong_schema: blocked before publication
4invented_citation: blocked before publication
5self_approved: blocked before publicationBefore publishing, apply the complete versioned schema, factual/citation checks, and current workflow state. Returning a copied local result prevents later mutation of the caller's citation list from changing this validated result; it doesn't certify its claims. Keep provider credentials server-side and scoped to approved deployments. A valid provider key doesn't authorize a user to read another tenant's evidence or invoke an access tool. Audit records need principal/tenant and policy references, restricted access and retention, and appropriate redaction; never include credentials.
A generation retry must leave a completed write alone
Suppose a separately authorized human approved the access grant and the write succeeded, but the explanation timed out. Retrying generation should consume the recorded receipt. It shouldn't execute the grant again.
This in-memory service uses fake identities and receipts. It demonstrates repeated-key handling and a generation simulation that never invokes the write function. It neither authenticates a reviewer nor attaches a receipt to a real model prompt. A real approval must bind the operation's tenant, principal, target, duration, policy, and lifetime; a Boolean alone isn't that artifact.
The local ledger binds the key to tenant, principal, and exact parameters, rejecting changed intent. Production also needs retained evidence and a durable atomic write-plus-idempotency transaction, or reconciliation when an external commit can't share that transaction. Define retention and late-retry behavior; forgetting a key can allow a later duplicate write.[6]
1@dataclass(frozen=True)
2class Caller:
3 principal: str
4 tenant: str
5 permissions: frozenset[str]
6
7 def __post_init__(self) -> None:
8 if not owned_text(self.principal) or not owned_text(self.tenant) or type(self.permissions) is not frozenset or any(not owned_text(value) for value in self.permissions):
9 raise ValueError("need owned caller identity and explicit permission names")
10
11ledger: dict[tuple[str, str, str], tuple[tuple[str, int], str]] = {}
12write_count = 0
13
14def grant_fixture(caller: Caller, tenant: str, operation_key: str,
15 target: str, minutes: int, human_approved: bool) -> str:
16 global write_count
17 if type(caller) is not Caller or not owned_text(tenant):
18 raise PermissionError("need a validated caller and named tenant")
19 if caller.tenant != tenant or "access:grant" not in caller.permissions:
20 raise PermissionError("grant not authorized")
21 if human_approved is not True:
22 raise PermissionError("trusted human approval required")
23 if not owned_text(operation_key) or not owned_text(target) or not integer_count(minutes) or minutes == 0:
24 raise ValueError("invalid operation")
25 key = (tenant, caller.principal, operation_key)
26 parameters = (target, minutes)
27 if key in ledger:
28 previous, receipt = ledger[key]
29 if previous != parameters:
30 raise ValueError("idempotency key reused with different parameters")
31 return receipt
32 write_count += 1
33 receipt = f"grant-receipt-{write_count}"
34 ledger[key] = (parameters, receipt)
35 return receipt
36
37caller = Caller("reviewer-7", "tenant-a", frozenset({"access:grant"}))
38receipt = grant_fixture(caller, "tenant-a", "access-R900-grant", "prod-console", 15, True)
39# Retrying after a lost write response returns the same receipt.
40assert grant_fixture(caller, "tenant-a", "access-R900-grant", "prod-console", 15, True) == receipt
41breaker = CircuitBreaker()
42generation = execute_with_failure(private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT)
43assert generation.action == "served_fallback" and write_count == 1
44for tenant, minutes, approved in [("tenant-b", 15, True), ("tenant-a", 30, True), ("tenant-a", 15, False)]:
45 try:
46 grant_fixture(caller, tenant, "access-R900-grant", "prod-console", minutes, approved)
47 except (PermissionError, ValueError):
48 pass
49 else:
50 raise AssertionError("unsafe write accepted")
51assert write_count == 1
52print(f"receipt={receipt} writes={write_count} generation={generation.action}")
53print("cross-tenant, changed-parameters, and unapproved writes blocked")1receipt=grant-receipt-1 writes=1 generation=served_fallback
2cross-tenant, changed-parameters, and unapproved writes blockedHand the support agent a policy artifact
The next lesson builds a complete support agent, but it shouldn't recreate routing rules inside orchestration code. Export a versioned artifact that states the intended requirements and escalation rules. Lane selection and retry behavior stay in the gateway; the agent adds retrieval, tool authorization, and human handoff on top. This fixture artifact isn't approval to deploy: the registry, real adapters, durable accounting, output validation, and representative quality evaluation still need integration evidence.
The final cell emits that handoff: inherited cost contract and schema, demonstrated primary/fallback choices, and explicit attempt, deadline, and spend limits.
1policy_artifact = {
2 "policy_id": GATEWAY_POLICY_ID,
3 "cost_release_id": COST_RELEASE_ID,
4 "required_answer_schema": REQUIRED_ANSWER_SCHEMA,
5 "max_generated_answer_usd": str(MAX_GENERATED_ANSWER_USD),
6 "retry_limits": {
7 "max_generation_attempts": MAX_GENERATION_ATTEMPTS,
8 "request_deadline_ms": REQUEST_DEADLINE_MS,
9 "request_spend_cap_usd": str(MAX_REQUEST_SPEND_USD),
10 "attempt_reservation_usd": str(ATTEMPT_RESERVATION_USD),
11 },
12 "fixture_examples": {
13 "public_deploy_policy": "fast-public-json",
14 "private_high_risk_access": "primary-private-cited-review",
15 "private_high_risk_access_fallback": "local-private-cited-review",
16 },
17 "escalate_when": [
18 "no lane preserves all contract fields",
19 "failure occurs after visible output begins",
20 "approved private context capacity is exceeded",
21 "retry attempts or request deadline are exhausted",
22 "request spend reservation is denied",
23 "failure is not eligible for transparent generation retry",
24 "normalized reply fails the versioned application contract",
25 ],
26}
27
28print(json.dumps(policy_artifact, indent=2))1{
2 "policy_id": "gateway-policy-v1",
3 "cost_release_id": "support-release-2026-05-cost-v1",
4 "required_answer_schema": "cited-support-answer-v3",
5 "max_generated_answer_usd": "0.005000",
6 "retry_limits": {
7 "max_generation_attempts": 2,
8 "request_deadline_ms": 2500,
9 "request_spend_cap_usd": "0.010000",
10 "attempt_reservation_usd": "0.005000"
11 },
12 "fixture_examples": {
13 "public_deploy_policy": "fast-public-json",
14 "private_high_risk_access": "primary-private-cited-review",
15 "private_high_risk_access_fallback": "local-private-cited-review"
16 },
17 "escalate_when": [
18 "no lane preserves all contract fields",
19 "failure occurs after visible output begins",
20 "approved private context capacity is exceeded",
21 "retry attempts or request deadline are exhausted",
22 "request spend reservation is denied",
23 "failure is not eligible for transparent generation retry",
24 "normalized reply fails the versioned application contract"
25 ]
26}Before moving on, change one boundary at a time:
- Set the primary duration to 1,400 ms. A 1,100 ms fallback reaches exactly 2,500 ms and is rejected under this fixture's strict-before-deadline rule. At 1,399 ms, it completes at 2,499 ms.
- Open all three compatible circuits. No lane should be attempted and spend should stay zero. Closing only regional should permit it without relaxing the contract.
- Replace
E1withE99in a well-formed reply. The response must be blocked even though lane selection passed. - Remove
access:grantfrom the caller and replay the same operation key. Authorization is checked again; a cached receipt isn't permission to perform or disclose a write.