"""Reference solution — the SRE console for a probabilistic system. Ordinary SRE assumes "correct" is a predicate. For an AI platform it is a *distribution*: the same input can produce a good answer, a mediocre one and a wrong one, and none of them is an error in the HTTP sense. That single fact breaks three habits, and this file is the consequence of each: * **quality does not belong in the availability SLI** — it is not measurable in real time, not attributable to the platform, and it makes the metric un-actionable during an incident. So: a hard SLO for availability, a *gated objective* for quality; * **the golden signals are incomplete** — cost per successful action and safety-block rate can move catastrophically while latency, traffic, errors and saturation are all green; * **debugging is trace-first** — a run is a tree, and non-determinism means you cannot re-run to reproduce, so the trace is the only artifact. Deterministic: an injected clock, integer micro-USD money divided last, sorted outputs. ``python solution.py`` runs the worked example. """ from __future__ import annotations import math from collections import Counter, defaultdict from dataclasses import dataclass, field, replace from enum import Enum from typing import (Any, Callable, Dict, FrozenSet, Iterable, List, Mapping, Optional, Sequence, Set, Tuple) #: Floating-point slack. A budget consumed exactly to its limit leaves ~3.5e-14, and a #: mathematically-exact burn rate of 14.4 computes as 14.399999999999986 — so a naive #: `>=` never fires. Same reasoning as Phase 00. EPSILON = 1e-9 # ====================================================================================== # 1. Events and the validity predicate # ====================================================================================== class Outcome(str, Enum): """What happened to one request. The classification is the design. ``USER_ABORTED`` and ``CLIENT_ERROR`` exist so they can be *excluded* from the denominator — see ``ValidityPredicate``. """ SUCCESS = "success" PLATFORM_ERROR = "platform_error" # our fault UPSTREAM_ERROR = "upstream_error" # a dependency's fault; still counts CLIENT_ERROR = "client_error" # a 4xx; NOT our availability USER_ABORTED = "user_aborted" # they closed the tab SAFETY_BLOCKED = "safety_blocked" # a guardrail refused it — WORKING AS INTENDED TIMEOUT = "timeout" @dataclass(frozen=True) class RequestEvent: """One request. The unit of every SLI here.""" tick: int tenant: str deployment: str outcome: Outcome latency_ms: int cost_micros: int = 0 tokens_in: int = 0 tokens_out: int = 0 synthetic: bool = False trace_id: str = "" @dataclass(frozen=True) class ValidityPredicate: """**Which events count in the denominator.** A design decision, written down. The event-ratio model is ``good / valid``, and almost every argument about an SLO is really an argument about *valid*. Three exclusions that are defensible, and each is contentious enough to need stating: * **client errors** — a malformed request is not an availability failure. But an endpoint that returns 400 for a valid request *is*, so this exclusion is only safe if 4xx is genuinely the client's fault; * **user aborts** — they closed the tab. Counting these makes the SLI track user patience; * **synthetic probes** — they are for detection, not for the number. Including them lets you improve the SLO by probing more, which is Goodhart's law with a cron job. And the one people get wrong: **safety blocks are excluded from *good* but counted in *valid*.** A guardrail refusing a request is the platform working correctly, so it is not a success — but if the guardrail suddenly refuses everything, availability *should* degrade. Excluding them entirely hides a total outage behind a green dashboard. """ exclude_client_errors: bool = True exclude_user_aborted: bool = True exclude_synthetic: bool = True tenants: Tuple[str, ...] = () # empty = all deployments: Tuple[str, ...] = () def is_valid(self, event: RequestEvent) -> bool: if self.exclude_synthetic and event.synthetic: return False if self.exclude_client_errors and event.outcome is Outcome.CLIENT_ERROR: return False if self.exclude_user_aborted and event.outcome is Outcome.USER_ABORTED: return False if self.tenants and event.tenant not in self.tenants: return False if self.deployments and event.deployment not in self.deployments: return False return True def describe(self) -> str: parts = [] if self.exclude_client_errors: parts.append("4xx excluded") if self.exclude_user_aborted: parts.append("aborts excluded") if self.exclude_synthetic: parts.append("synthetic excluded") if self.tenants: parts.append(f"tenants={list(self.tenants)}") return "; ".join(parts) or "everything counts" GOOD_OUTCOMES: FrozenSet[Outcome] = frozenset({Outcome.SUCCESS}) # ====================================================================================== # 2. The SLI # ====================================================================================== @dataclass(frozen=True) class SliResult: good: int valid: int ratio: float predicate: str @property def bad(self) -> int: return self.valid - self.good def format(self) -> str: return f"{self.ratio * 100:.3f}% ({self.good}/{self.valid})" def availability_sli(events: Sequence[RequestEvent], predicate: ValidityPredicate = ValidityPredicate()) -> SliResult: """``good / valid``. An empty window is 1.0, not 0.0 — and that choice matters. Zero traffic is not an outage, and treating it as one means every quiet Sunday burns the entire budget. """ valid = [e for e in events if predicate.is_valid(e)] good = [e for e in valid if e.outcome in GOOD_OUTCOMES] ratio = len(good) / len(valid) if valid else 1.0 return SliResult(len(good), len(valid), ratio, predicate.describe()) def latency_sli(events: Sequence[RequestEvent], *, threshold_ms: int, predicate: ValidityPredicate = ValidityPredicate()) -> SliResult: """Latency as a RATIO, not a percentile. "p99 < 2s" cannot be aggregated (percentiles do not average) and cannot be turned into an error budget. "99% of requests under 2s" is the same statement as a countable ratio, which composes across windows and regions and has a budget you can spend. """ valid = [e for e in events if predicate.is_valid(e)] good = [e for e in valid if e.outcome in GOOD_OUTCOMES and e.latency_ms <= threshold_ms] ratio = len(good) / len(valid) if valid else 1.0 return SliResult(len(good), len(valid), ratio, f"{predicate.describe()}; under {threshold_ms}ms") def percentile(values: Sequence[int], q: float) -> float: """Nearest-rank. Reported alongside the ratio SLI for diagnosis, never alerted on.""" if not values: return 0.0 ordered = sorted(values) rank = max(1, math.ceil(q * len(ordered))) return float(ordered[min(rank, len(ordered)) - 1]) # ====================================================================================== # 3. Error budgets # ====================================================================================== @dataclass(frozen=True) class Slo: name: str target: float # e.g. 0.995 window_ticks: int # the rolling window threshold_ms: Optional[int] = None # set for a latency SLO @property def budget_fraction(self) -> float: return 1.0 - self.target @dataclass(frozen=True) class BudgetState: slo: str target: float achieved: float allowed_bad: float actual_bad: int remaining_fraction: float exhausted: bool overspent_by: int # events beyond the budget; separate from "exhausted" def format(self) -> str: return (f"{self.slo}: {self.achieved * 100:.3f}% vs {self.target * 100:.3f}% " f"target — {self.remaining_fraction * 100:.1f}% of budget left") class ErrorBudget: """A rolling budget over a window. Two properties the tests pin, and both are deliberate: * the remaining fraction **never goes negative** — it clamps at 0, because a dashboard showing -340% tells you nothing you did not already know; * **overspend is surfaced separately** as a count, because "how far past" is a real question and clamping it away loses it. """ def __init__(self, slo: Slo) -> None: self.slo = slo def state(self, events: Sequence[RequestEvent], predicate: ValidityPredicate = ValidityPredicate()) -> BudgetState: sli = (latency_sli(events, threshold_ms=self.slo.threshold_ms, predicate=predicate) if self.slo.threshold_ms is not None else availability_sli(events, predicate=predicate)) allowed = sli.valid * self.slo.budget_fraction remaining = 1.0 if allowed <= 0 else 1.0 - (sli.bad / allowed) clamped = max(0.0, remaining) overspent = max(0, sli.bad - int(allowed)) return BudgetState( self.slo.name, self.slo.target, sli.ratio, allowed, sli.bad, clamped, clamped <= EPSILON, overspent) def allocate_budget(total_fraction: float, weights: Mapping[str, float]) -> Dict[str, float]: """Split a budget across layers, proportionally. The point of allocating is that a shared budget is a budget nobody owns. When the gateway spends the whole thing, the retrieval team learns about it from an incident rather than from their own number. """ total_weight = sum(weights.values()) if total_weight <= 0: raise ValueError("weights must sum to something positive") return {name: total_fraction * (w / total_weight) for name, w in sorted(weights.items())} # ====================================================================================== # 4. Burn-rate alerting # ====================================================================================== def burn_rate_threshold(budget_fraction_consumed: float, window_hours: float, *, period_days: int = 30) -> float: """**Derive** the threshold rather than memorizing 14.4. burn_rate = (fraction of budget consumed) / (fraction of period elapsed) Consuming 2% of a 30-day budget in one hour: (0.02) / (1 / 720) = 14.4 That is where the famous number comes from, and being able to derive it is the difference between operating an SLO and copying a runbook. """ if window_hours <= 0: raise ValueError("window must be positive") period_hours = period_days * 24 return budget_fraction_consumed / (window_hours / period_hours) def observed_burn_rate(sli: SliResult, slo: Slo) -> float: """How fast the budget is being spent, as a multiple of the sustainable rate. 1.0 means exactly on budget for the period. 14.4 means the whole month's budget in an hour. """ if slo.budget_fraction <= 0: return 0.0 error_rate = (sli.bad / sli.valid) if sli.valid else 0.0 return error_rate / slo.budget_fraction @dataclass(frozen=True) class BurnRateRule: name: str long_window_hours: float short_window_hours: float threshold: float severity: str # "page" | "ticket" min_events: int = 10 # the low-traffic guard #: The standard ladder, with its thresholds *derived*. #: #: Two windows per rule is the entire trick. The long window is the signal; the **short #: window is the reset**. Without it, a five-minute blip that consumed 2% of the budget #: keeps the alert firing for the rest of the hour, long after the problem is gone — and #: an alert that stays lit after the fix is an alert people learn to close. STANDARD_LADDER: Tuple[BurnRateRule, ...] = ( BurnRateRule("fast-burn", 1.0, 1.0 / 12, burn_rate_threshold(0.02, 1.0), "page", min_events=10), BurnRateRule("medium-burn", 6.0, 0.5, burn_rate_threshold(0.05, 6.0), "page", min_events=60), BurnRateRule("slow-burn", 24.0, 2.0, burn_rate_threshold(0.10, 24.0), "ticket", min_events=200), ) @dataclass(frozen=True) class AlertDecision: rule: str firing: bool severity: str long_burn: float short_burn: float reason: str class BurnRateAlerting: """Multi-window, multi-burn-rate. A rule fires only when **both** windows exceed the threshold and there is enough traffic to mean anything. The minimum-volume guard is the one people leave out and the one that determines whether the alerting survives. On a service doing two requests an hour, one failure is a 50% error rate and a burn rate in the hundreds. Page on that a few times at 3 a.m. and somebody raises the threshold until nothing ever fires. """ def __init__(self, slo: Slo, rules: Sequence[BurnRateRule] = STANDARD_LADDER, *, ticks_per_hour: int = 60) -> None: self.slo = slo self.rules = tuple(rules) self.ticks_per_hour = ticks_per_hour def _window(self, events: Sequence[RequestEvent], now: int, hours: float) -> List[RequestEvent]: cutoff = now - hours * self.ticks_per_hour return [e for e in events if e.tick > cutoff] def evaluate(self, events: Sequence[RequestEvent], now: int, predicate: ValidityPredicate = ValidityPredicate() ) -> List[AlertDecision]: out: List[AlertDecision] = [] for rule in self.rules: long_events = self._window(events, now, rule.long_window_hours) short_events = self._window(events, now, rule.short_window_hours) long_sli = self._sli(long_events, predicate) short_sli = self._sli(short_events, predicate) long_burn = observed_burn_rate(long_sli, self.slo) short_burn = observed_burn_rate(short_sli, self.slo) if long_sli.valid < rule.min_events: out.append(AlertDecision( rule.name, False, rule.severity, long_burn, short_burn, f"only {long_sli.valid} events; below the {rule.min_events} minimum")) continue # `* (1 - EPSILON)` because a mathematically-exact threshold computes just # under itself in floating point, so a naive `>=` never fires. gate = rule.threshold * (1 - EPSILON) long_hot = long_burn >= gate short_hot = short_burn >= gate if long_hot and short_hot: reason = (f"both windows over {rule.threshold:.1f}x " f"(long {long_burn:.1f}, short {short_burn:.1f})") out.append(AlertDecision(rule.name, True, rule.severity, long_burn, short_burn, reason)) elif long_hot: out.append(AlertDecision( rule.name, False, rule.severity, long_burn, short_burn, f"long window hot ({long_burn:.1f}x) but short window has recovered " f"({short_burn:.1f}x)")) else: out.append(AlertDecision( rule.name, False, rule.severity, long_burn, short_burn, f"burn rate {long_burn:.1f}x is under {rule.threshold:.1f}x")) return out def _sli(self, events: Sequence[RequestEvent], predicate: ValidityPredicate) -> SliResult: if self.slo.threshold_ms is not None: return latency_sli(events, threshold_ms=self.slo.threshold_ms, predicate=predicate) return availability_sli(events, predicate=predicate) def firing(self, events: Sequence[RequestEvent], now: int, predicate: ValidityPredicate = ValidityPredicate() ) -> List[AlertDecision]: return [d for d in self.evaluate(events, now, predicate) if d.firing] # ====================================================================================== # 5. The span tree # ====================================================================================== class SpanKind(str, Enum): AGENT = "agent" MODEL = "model" TOOL = "tool" RETRIEVAL = "retrieval" POLICY = "policy" GUARDRAIL = "guardrail" @dataclass(frozen=True) class Span: """One unit of work. Attribute names follow OTel's **GenAI semantic conventions**. Using the conventions rather than inventing names is not pedantry: it is what lets any OTel-aware backend chart your token usage without custom queries, and it is a migration you will otherwise do later under pressure. """ span_id: str parent_id: Optional[str] trace_id: str name: str kind: SpanKind start_tick: int end_tick: int attributes: Mapping[str, Any] = field(default_factory=dict) status: str = "ok" @property def duration(self) -> int: return self.end_tick - self.start_tick @property def cost_micros(self) -> int: return int(self.attributes.get("gen_ai.usage.cost_micros", 0)) class Tracer: """Derived span ids, injected clock — so a trace is reproducible in a test.""" def __init__(self, *, now: Callable[[], int]) -> None: self.now = now self.spans: List[Span] = [] self._counter = 0 def record(self, *, trace_id: str, parent_id: Optional[str], name: str, kind: SpanKind, start_tick: int, status: str = "ok", **attributes: Any) -> Span: self._counter += 1 span = Span(f"span-{self._counter}", parent_id, trace_id, name, kind, start_tick, self.now(), dict(attributes), status) self.spans.append(span) return span def trace(self, trace_id: str) -> List[Span]: return [s for s in self.spans if s.trace_id == trace_id] class SpanTree: """Reconstruct a run from a flat span list — and answer "where did the time go?" This is the artifact that replaces logs. Non-determinism means you cannot re-run to reproduce, so whatever the trace did not capture is gone. """ def __init__(self, spans: Sequence[Span]) -> None: self.spans = {s.span_id: s for s in spans} self.children: Dict[Optional[str], List[str]] = defaultdict(list) for span in sorted(spans, key=lambda s: (s.start_tick, s.span_id)): self.children[span.parent_id].append(span.span_id) @property def roots(self) -> List[str]: known = set(self.spans) return [s.span_id for s in self.spans.values() if s.parent_id is None or s.parent_id not in known] def walk(self, span_id: Optional[str] = None, depth: int = 0 ) -> Iterable[Tuple[int, Span]]: if span_id is None: for root in self.roots: yield from self.walk(root, 0) return yield depth, self.spans[span_id] for child in self.children.get(span_id, []): yield from self.walk(child, depth + 1) def self_time(self, span_id: str) -> int: """Duration minus the time covered by children. **The number that answers "where did the latency go?"** Total duration attributes a slow tool call to the whole run; self time attributes it to the tool. Note this subtracts the *union* of child intervals, so concurrent children are not double-counted — a naive sum makes a parallel fan-out look like negative self time, which people then clamp to zero and stop trusting. """ span = self.spans[span_id] intervals = sorted((self.spans[c].start_tick, self.spans[c].end_tick) for c in self.children.get(span_id, [])) covered = 0 cursor = span.start_tick for start, end in intervals: start = max(start, cursor) if end > start: covered += end - start cursor = end return max(0, span.duration - covered) def time_by_kind(self) -> Dict[str, int]: totals: Dict[str, int] = defaultdict(int) for span_id, span in self.spans.items(): totals[span.kind.value] += self.self_time(span_id) return dict(sorted(totals.items())) def total_cost_micros(self) -> int: return sum(s.cost_micros for s in self.spans.values()) def critical_path(self) -> List[str]: """The longest chain root-to-leaf by duration — what to optimize first.""" best: List[str] = [] best_duration = -1 def descend(span_id: str, path: List[str]) -> None: nonlocal best, best_duration children = self.children.get(span_id, []) if not children: duration = sum(self.spans[s].duration for s in path) if duration > best_duration: best_duration, best = duration, list(path) return for child in children: descend(child, path + [child]) for root in self.roots: descend(root, [root]) return best def errors(self) -> List[Span]: return sorted((s for s in self.spans.values() if s.status != "ok"), key=lambda s: s.span_id) # ====================================================================================== # 6. Cardinality governance # ====================================================================================== @dataclass(frozen=True) class LabelSpec: name: str estimated_values: int description: str = "" @dataclass(frozen=True) class MetricSpec: name: str labels: Tuple[LabelSpec, ...] @property def series_count(self) -> int: """Series are the **product** of label cardinalities, not the sum. This is why the failure is a cliff rather than a slope: adding one label with 100 values multiplies everything, and a metric goes from affordable to unaffordable in one commit. """ total = 1 for label in self.labels: total *= max(1, label.estimated_values) return total @dataclass(frozen=True) class CardinalityVerdict: metric: str series: int budget: int ok: bool worst_label: Optional[str] reason: str #: Labels that must never appear on a metric. Each of these is unbounded, and each has #: taken down a metrics backend somewhere. FORBIDDEN_LABELS: FrozenSet[str] = frozenset({ "user_id", "trace_id", "span_id", "request_id", "session_id", "prompt", "conversation_id", "document_id", "email", "account_number", }) class CardinalityBudget: """Reject an unaffordable metric **at definition time**. Cardinality is the metric backend's cliff, and the failure mode is specific: the backend does not degrade, it falls over — and it falls over during an incident, because that is when a new label seemed useful. The rule that follows: **high-cardinality identifiers live on traces and in the accounting store; metrics carry tenant, deployment and outcome.** A trace backend is built for unbounded ids; a time-series backend is not. """ def __init__(self, *, max_series_per_metric: int = 10_000, max_total_series: int = 200_000) -> None: self.max_series_per_metric = max_series_per_metric self.max_total_series = max_total_series self.registered: Dict[str, MetricSpec] = {} def check(self, spec: MetricSpec) -> CardinalityVerdict: forbidden = sorted({l.name for l in spec.labels} & FORBIDDEN_LABELS) if forbidden: return CardinalityVerdict( spec.name, spec.series_count, self.max_series_per_metric, False, forbidden[0], f"label {forbidden[0]!r} is unbounded; it belongs on a trace, not a " f"metric") series = spec.series_count if series > self.max_series_per_metric: worst = max(spec.labels, key=lambda l: l.estimated_values) return CardinalityVerdict( spec.name, series, self.max_series_per_metric, False, worst.name, f"{series:,} series exceeds the per-metric budget of " f"{self.max_series_per_metric:,}; {worst.name!r} contributes " f"{worst.estimated_values}x") projected = self.total_series() + series if projected > self.max_total_series: return CardinalityVerdict( spec.name, series, self.max_total_series, False, None, f"would take the total to {projected:,}, over the " f"{self.max_total_series:,} budget") return CardinalityVerdict(spec.name, series, self.max_series_per_metric, True, None, "within budget") def register(self, spec: MetricSpec) -> MetricSpec: verdict = self.check(spec) if not verdict.ok: raise ValueError(f"{spec.name}: {verdict.reason}") self.registered[spec.name] = spec return spec def total_series(self) -> int: return sum(s.series_count for s in self.registered.values()) # ====================================================================================== # 7. The degradation ladder # ====================================================================================== @dataclass(frozen=True) class DegradationStep: """One rung. ``user_visible`` is the field that makes the ladder honest. A degradation the user is not told about is a silent quality change, and it is how a platform loses trust: the answers got worse and nobody said anything. """ name: str description: str saves: str user_visible: bool reversible: bool = True #: **Written in daylight, executed at 3 a.m.** The order is the design decision, and it #: goes cheapest-loss-first: shed accuracy before capability, capability before #: availability. STANDARD_LADDER_STEPS: Tuple[DegradationStep, ...] = ( DegradationStep("disable-rerank", "skip the cross-encoder reranker", "~30% of retrieval latency", user_visible=False), DegradationStep("smaller-model", "route to the small model", "~70% of token cost, ~50% latency", user_visible=True), DegradationStep("cache-only", "serve only from the semantic cache", "all model cost", user_visible=True), DegradationStep("read-only", "refuse side-effecting tools", "all downstream write load", user_visible=True), DegradationStep("queue", "accept and defer", "everything except the queue", user_visible=True), DegradationStep("reject", "refuse new work", "everything", user_visible=True), ) @dataclass(frozen=True) class LadderState: level: int active: Tuple[str, ...] reason: str @property def degraded(self) -> bool: return self.level > 0 class DegradationLadder: """Shed load in a declared order, driven by burn rate. Two asymmetries that are the whole operational content: * **descend fast, ascend slowly.** Going down one rung at a time during a real incident is too slow; coming back up quickly re-creates the load that caused it. So: jump straight to the level the burn rate implies, and recover one rung at a time with a hold period. * **recovery needs hysteresis.** Without it the ladder oscillates — degrade, load drops, restore, load returns, degrade — which is worse than staying degraded, because the user sees the answer quality flapping. """ def __init__(self, steps: Sequence[DegradationStep] = STANDARD_LADDER_STEPS, *, thresholds: Sequence[float] = (2.0, 6.0, 14.4, 30.0, 60.0, 100.0), recovery_hold_ticks: int = 10, now: Optional[Callable[[], int]] = None) -> None: if len(thresholds) < len(steps): raise ValueError("every step needs a threshold") self.steps = tuple(steps) self.thresholds = tuple(thresholds) self.recovery_hold_ticks = recovery_hold_ticks self.now = now or (lambda: 0) self.level = 0 self._last_change = self.now() self.history: List[Tuple[int, int, str]] = [] def target_level(self, burn_rate: float) -> int: level = 0 for i, threshold in enumerate(self.thresholds[:len(self.steps)], start=1): if burn_rate >= threshold * (1 - EPSILON): level = i return level def evaluate(self, burn_rate: float) -> LadderState: target = self.target_level(burn_rate) now = self.now() if target > self.level: reason = f"burn rate {burn_rate:.1f}x; descending to level {target}" self.level = target # jump, do not step self._last_change = now self.history.append((now, self.level, reason)) elif target < self.level: if now - self._last_change < self.recovery_hold_ticks: return LadderState( self.level, self._active(), f"burn rate {burn_rate:.1f}x has recovered; holding level " f"{self.level} for " f"{self.recovery_hold_ticks - (now - self._last_change)} more ticks") self.level -= 1 # one rung at a time self._last_change = now reason = f"burn rate {burn_rate:.1f}x; ascending to level {self.level}" self.history.append((now, self.level, reason)) else: reason = f"burn rate {burn_rate:.1f}x; holding level {self.level}" return LadderState(self.level, self._active(), reason) def _active(self) -> Tuple[str, ...]: return tuple(s.name for s in self.steps[:self.level]) @property def user_visible_degradation(self) -> bool: return any(s.user_visible for s in self.steps[:self.level]) # ====================================================================================== # 8. Cost governance # ====================================================================================== @dataclass(frozen=True) class CostReport: """``cost per successful action`` is the unit economic. Note the denominator: **successful** actions. Cost per request improves when you fail faster, which is the wrong incentive — a platform that errors instantly has an excellent cost per request. """ tenant: str total_micros: int successful_actions: int cost_per_action_micros: int wasted_micros: int # spent on requests that did not succeed @property def waste_fraction(self) -> float: return self.wasted_micros / self.total_micros if self.total_micros else 0.0 def format(self) -> str: return (f"{self.tenant}: ${self.total_micros / 1_000_000:.2f} total, " f"{self.successful_actions} actions, " f"${self.cost_per_action_micros / 1_000_000:.4f}/action, " f"{self.waste_fraction * 100:.1f}% wasted") def cost_report(events: Sequence[RequestEvent], tenant: str) -> CostReport: """Money is integer micro-USD throughout; divide only for display.""" scoped = [e for e in events if e.tenant == tenant] total = sum(e.cost_micros for e in scoped) good = [e for e in scoped if e.outcome in GOOD_OUTCOMES] wasted = total - sum(e.cost_micros for e in good) per_action = total // len(good) if good else 0 return CostReport(tenant, total, len(good), per_action, wasted) @dataclass(frozen=True) class BreakerVerdict: tenant: str tripped: bool spent_micros: int budget_micros: int reason: str @property def fraction_used(self) -> float: return self.spent_micros / self.budget_micros if self.budget_micros else 0.0 class CostCircuitBreaker: """**A cost control is an availability control.** A runaway agent loop can spend a month's budget in an hour, and the failure looks like nothing — every request succeeds. So the breaker trips per tenant, which is the property that matters: one tenant's loop must not exhaust another tenant's budget or the platform's. Two thresholds, because "warn then stop" is what makes it usable: an alert at 80% gives somebody a chance to look before the tenant is cut off. """ def __init__(self, *, budgets_micros: Mapping[str, int], warn_fraction: float = 0.8) -> None: self.budgets = dict(budgets_micros) self.warn_fraction = warn_fraction self._spend: Dict[str, int] = defaultdict(int) self.tripped: Set[str] = set() def record(self, tenant: str, micros: int) -> None: self._spend[tenant] += micros def spent(self, tenant: str) -> int: return self._spend[tenant] def check(self, tenant: str) -> BreakerVerdict: budget = self.budgets.get(tenant, 0) spent = self._spend[tenant] if budget <= 0: return BreakerVerdict(tenant, True, spent, 0, "no budget is configured for this tenant") if spent >= budget: self.tripped.add(tenant) return BreakerVerdict(tenant, True, spent, budget, f"spent {spent} of {budget} micro-USD") if spent >= budget * self.warn_fraction: return BreakerVerdict( tenant, False, spent, budget, f"at {spent / budget * 100:.0f}% of budget — warn") return BreakerVerdict(tenant, False, spent, budget, "within budget") def allow(self, tenant: str) -> bool: return not self.check(tenant).tripped def reset(self, tenant: str) -> None: self._spend[tenant] = 0 self.tripped.discard(tenant) # ====================================================================================== # 9. Capacity forecasting # ====================================================================================== @dataclass(frozen=True) class CapacityForecast: resource: str current: float limit: float growth_per_period: float periods_to_limit: Optional[float] lead_time_periods: int alert: bool reason: str @property def headroom_fraction(self) -> float: return max(0.0, 1.0 - self.current / self.limit) if self.limit else 0.0 def forecast_capacity(samples: Sequence[float], *, limit: float, lead_time_periods: int, safety_factor: float = 1.5 ) -> CapacityForecast: """Project usage against a **provider limit**, and alert with procurement lead time. Two things that make this different from ordinary capacity planning: * **the operative limit is the provider's quota, not CPU.** You will hit an Azure OpenAI TPM ceiling long before a machine is busy, and CPU-based headroom alerts are silent all the way to a hard 429; * **the alert must fire a lead time early.** GPU quota increases take weeks and GPU hardware takes months. An alert at 90% utilization is an alert that arrives after the decision point, which makes it a notification rather than a control. The safety factor is deliberate over-caution: starting a procurement conversation early costs a meeting, and starting it late costs a quarter. """ if limit <= 0: raise ValueError("limit must be positive") if len(samples) < 2: return CapacityForecast("", samples[-1] if samples else 0.0, limit, 0.0, None, lead_time_periods, False, "not enough samples to project") # Least-squares slope. Robust enough for a trend, and deterministic. n = len(samples) mean_x = (n - 1) / 2 mean_y = sum(samples) / n denominator = sum((i - mean_x) ** 2 for i in range(n)) slope = (sum((i - mean_x) * (y - mean_y) for i, y in enumerate(samples)) / denominator) if denominator else 0.0 current = samples[-1] if slope <= 0: return CapacityForecast("", current, limit, slope, None, lead_time_periods, current >= limit, "usage is flat or falling" if current < limit else "already at the limit") periods = (limit - current) / slope horizon = lead_time_periods * safety_factor alert = periods <= horizon reason = (f"{periods:.1f} periods to the limit, inside the " f"{horizon:.1f}-period alert horizon " f"({lead_time_periods} lead time x {safety_factor})" if alert else f"{periods:.1f} periods to the limit, outside the " f"{horizon:.1f}-period horizon") return CapacityForecast("", current, limit, slope, periods, lead_time_periods, alert, reason) # ====================================================================================== # 10. Incident classification # ====================================================================================== @dataclass(frozen=True) class Baseline: """What was pinned when the eval last passed. **The whole diagnostic technique.** In a deterministic system you re-run to reproduce. Here you cannot, so the only way to answer "what changed?" is to have recorded what everything *was* — and if the model version is not pinned, "the provider changed the model" is a hypothesis you can never confirm or exclude. """ model_version: str prompt_version: str corpus_version: str policy_version: str eval_score: float recorded_tick: int @dataclass(frozen=True) class Hypothesis: cause: str confidence: str # "confirmed" | "likely" | "possible" | "excluded" evidence: str def classify_regression(baseline: Baseline, current: Baseline, *, eval_drop_threshold: float = 0.05) -> List[Hypothesis]: """"The model changed" vs "our prompt changed" vs "the corpus changed". Deliberately mechanical: it compares recorded versions. That is the point — the technique is not clever inference, it is **having pinned the versions in the first place**, and every "confirmed" below is only available because somebody did. """ out: List[Hypothesis] = [] drop = baseline.eval_score - current.eval_score if drop < eval_drop_threshold: out.append(Hypothesis("no-regression", "excluded", f"eval moved {drop:+.3f}, inside the " f"{eval_drop_threshold} threshold")) return out changed = False for field_name, cause in (("model_version", "model-changed"), ("prompt_version", "prompt-changed"), ("corpus_version", "corpus-changed"), ("policy_version", "policy-changed")): before, after = getattr(baseline, field_name), getattr(current, field_name) if before != after: changed = True out.append(Hypothesis(cause, "confirmed", f"{field_name}: {before} -> {after}")) else: out.append(Hypothesis(cause, "excluded", f"{field_name} is unchanged at {before}")) if not changed: # Nothing we control moved. Either the provider changed the model behind a stable # version string, or the input distribution changed. Both are real, and being # able to say "it is one of these two" is a much better position than "something # changed". out.append(Hypothesis( "provider-silent-change-or-input-drift", "likely", "every pinned version is unchanged and the eval dropped; the provider " "changed behaviour behind a stable version, or the input distribution moved")) return sorted(out, key=lambda h: (h.confidence != "confirmed", h.cause)) # ====================================================================================== # Worked example # ====================================================================================== def _clock(start: int = 0, step: int = 1) -> Callable[[], int]: state = {"t": start - step} def now() -> int: state["t"] += step return state["t"] return now def _events(spec: Sequence[Tuple[int, Outcome, int]], *, tenant: str = "wholesale", cost: int = 4_200) -> List[RequestEvent]: return [RequestEvent(tick, tenant, "gpt-frontier", outcome, latency, cost_micros=cost if outcome is Outcome.SUCCESS else cost // 2) for tick, outcome, latency in spec] def main() -> None: # pragma: no cover - narrative output print("=" * 78) print("1. THE VALIDITY PREDICATE IS THE ARGUMENT") print("=" * 78) mixed = [ RequestEvent(1, "wholesale", "d", Outcome.SUCCESS, 800), RequestEvent(2, "wholesale", "d", Outcome.PLATFORM_ERROR, 100), RequestEvent(3, "wholesale", "d", Outcome.CLIENT_ERROR, 20), RequestEvent(4, "wholesale", "d", Outcome.USER_ABORTED, 5_000), RequestEvent(5, "wholesale", "d", Outcome.SAFETY_BLOCKED, 60), RequestEvent(6, "wholesale", "d", Outcome.SUCCESS, 900, synthetic=True), ] for label, predicate in ( ("everything counts", ValidityPredicate(False, False, False)), ("standard", ValidityPredicate()), ("4xx counted", ValidityPredicate(exclude_client_errors=False)), ): sli = availability_sli(mixed, predicate) print(f" {label:<18} {sli.format():<20} [{sli.predicate}]") print(" -> the same six events, three different numbers. Almost every argument") print(" about an SLO is really an argument about the denominator.") print(" -> and note the safety block: excluded from GOOD, counted in VALID. If a") print(" guardrail starts refusing everything, availability SHOULD degrade.") print() print("=" * 78) print("2. LATENCY AS A RATIO, NOT A PERCENTILE") print("=" * 78) latencies = [200, 250, 300, 400, 500, 800, 1_200, 1_900, 2_400, 9_000] events = _events([(i, Outcome.SUCCESS, ms) for i, ms in enumerate(latencies)]) print(f" p50={percentile(latencies, 0.5):.0f}ms " f"p95={percentile(latencies, 0.95):.0f}ms " f"p99={percentile(latencies, 0.99):.0f}ms") sli = latency_sli(events, threshold_ms=2_000) print(f" 'under 2000ms': {sli.format()}") print(" -> percentiles cannot be averaged across windows or regions, and there is") print(" no error budget for a p99. A ratio composes and has a budget you spend.") print() print("=" * 78) print("3. DERIVING THE BURN-RATE LADDER") print("=" * 78) print(f" {'rule':<14} {'budget':<9} {'window':<10} {'threshold':<12} severity") for consumed, hours, name, severity in ( (0.02, 1.0, "fast-burn", "page"), (0.05, 6.0, "medium-burn", "page"), (0.10, 24.0, "slow-burn", "ticket"), (0.02, 24.0, "(too slow)", "-"), ): threshold = burn_rate_threshold(consumed, hours) print(f" {name:<14} {consumed * 100:>5.0f}% {hours:>5.1f}h " f"{threshold:>8.2f}x {severity}") print(" -> 14.4 is not magic: 2% of a 30-day budget in 1 hour is 0.02/(1/720).") print(" Deriving it is the difference between operating an SLO and copying a") print(" runbook.") print() print("=" * 78) print("4. TWO WINDOWS, AND THE LOW-TRAFFIC GUARD") print("=" * 78) slo = Slo("availability", 0.995, window_ticks=43_200) alerting = BurnRateAlerting(slo) burst = ([RequestEvent(t, "w", "d", Outcome.SUCCESS, 300) for t in range(0, 50)] + [RequestEvent(t, "w", "d", Outcome.PLATFORM_ERROR, 100) for t in range(50, 56)] + [RequestEvent(t, "w", "d", Outcome.SUCCESS, 300) for t in range(56, 60)]) print(" during the burst (now=56):") for decision in alerting.evaluate(burst, now=56): mark = "FIRING" if decision.firing else " " print(f" {mark} {decision.rule:<13} {decision.reason}") recovered = burst + [RequestEvent(t, "w", "d", Outcome.SUCCESS, 300) for t in range(60, 115)] print(" 55 ticks later, fully recovered (now=115):") for decision in alerting.evaluate(recovered, now=115): mark = "FIRING" if decision.firing else " " print(f" {mark} {decision.rule:<13} {decision.reason}") print(" -> the SHORT window is the reset. Without it the alert stays lit for the") print(" rest of the hour, long after the fix — and an alert that stays lit after") print(" the fix is an alert people learn to close.") print() quiet = [RequestEvent(1, "w", "d", Outcome.SUCCESS, 300), RequestEvent(2, "w", "d", Outcome.PLATFORM_ERROR, 100)] print(" a low-traffic service: 1 failure out of 2 requests") sli = availability_sli(quiet) print(f" error rate {(1 - sli.ratio) * 100:.0f}%, burn rate " f"{observed_burn_rate(sli, slo):.0f}x") for decision in alerting.evaluate(quiet, now=2): print(f" {'FIRING' if decision.firing else ' '} {decision.rule:<13} " f"{decision.reason}") print(" -> a burn rate of 100x and it does not page. Without that guard, somebody") print(" raises the threshold at 3am until nothing ever fires.") print() print("=" * 78) print("5. ERROR BUDGETS, AND ALLOCATING THEM") print("=" * 78) month = ([RequestEvent(t, "w", "d", Outcome.SUCCESS, 300) for t in range(9_940)] + [RequestEvent(t, "w", "d", Outcome.PLATFORM_ERROR, 100) for t in range(9_940, 10_000)]) budget = ErrorBudget(slo) state = budget.state(month) print(f" {state.format()}") print(f" allowed {state.allowed_bad:.0f} bad events, actual {state.actual_bad}, " f"overspent by {state.overspent_by}") worse = month + [RequestEvent(t, "w", "d", Outcome.PLATFORM_ERROR, 100) for t in range(10_000, 10_200)] state = budget.state(worse) print(f" after 200 more failures: remaining " f"{state.remaining_fraction * 100:.1f}%, exhausted={state.exhausted}, " f"overspent by {state.overspent_by}") print(" -> remaining clamps at 0 (a dashboard showing -340% helps nobody) and the") print(" overspend is reported separately, because 'how far past' is a real") print(" question.") print() allocation = allocate_budget(slo.budget_fraction, {"gateway": 1, "kernel": 2, "retrieval": 1, "action-gateway": 1}) print(" allocating the 0.5% budget across layers:") for layer, fraction in allocation.items(): print(f" {layer:<16} {fraction * 100:.4f}% " f"({fraction / slo.budget_fraction * 100:.0f}% of the total)") print(" -> a shared budget is a budget nobody owns. When the gateway spends it all,") print(" the retrieval team should learn that from their number, not an incident.") print() print("=" * 78) print("6. THE SPAN TREE — WHERE DID THE LATENCY GO?") print("=" * 78) now = _clock(start=0, step=0) tracer = Tracer(now=lambda: 0) # ends set explicitly below spans = [ Span("s1", None, "t-1", "investigate-payment", SpanKind.AGENT, 0, 4_100), Span("s2", "s1", "t-1", "policy.evaluate", SpanKind.POLICY, 5, 12, {"policy.version": "v7", "policy.decision": "allow"}), Span("s3", "s1", "t-1", "guardrail.input", SpanKind.GUARDRAIL, 12, 18, {"guardrail.verdict": "allow"}), Span("s4", "s1", "t-1", "retrieval.hybrid", SpanKind.RETRIEVAL, 20, 900, {"retrieval.k": 20, "retrieval.chunks": 12}), Span("s5", "s4", "t-1", "vector.search", SpanKind.RETRIEVAL, 25, 180, {}), Span("s6", "s4", "t-1", "bm25.search", SpanKind.RETRIEVAL, 25, 120, {}), Span("s7", "s4", "t-1", "rerank", SpanKind.MODEL, 190, 880, {"gen_ai.request.model": "reranker-v2", "gen_ai.usage.cost_micros": 300}), Span("s8", "s1", "t-1", "model.complete", SpanKind.MODEL, 905, 3_400, {"gen_ai.system": "azure.openai", "gen_ai.request.model": "gpt-frontier-2026-02-11", "gen_ai.usage.input_tokens": 4_812, "gen_ai.usage.output_tokens": 380, "gen_ai.usage.cost_micros": 3_900}), Span("s9", "s1", "t-1", "payments.lookup", SpanKind.TOOL, 3_410, 4_050, {"tool.side_effect": "read", "http.status_code": 200}), ] tree = SpanTree(spans) print(f" {'span':<34} {'kind':<11} {'total':>7} {'self':>7}") for depth, span in tree.walk(): indent = " " * depth print(f" {indent}{span.name:<{32 - len(indent)}} {span.kind.value:<11} " f"{span.duration:>6}ms {tree.self_time(span.span_id):>6}ms") print() print(" self time by kind:") total = sum(tree.time_by_kind().values()) for kind, ms in sorted(tree.time_by_kind().items(), key=lambda kv: -kv[1]): print(f" {kind:<12} {ms:>6}ms ({ms / total * 100:>4.1f}%)") print(f" critical path: {' -> '.join(tree.spans[s].name for s in tree.critical_path())}") print(f" total cost: ${tree.total_cost_micros() / 1_000_000:.4f}") print(" -> SELF time, not total. Total duration blames the root span for everything;") print(" self time says the model call is 2.5s of the 4.1s and the rerank is 0.7s.") print(" And it subtracts the UNION of child intervals, so the two concurrent") print(" searches under retrieval are not double-counted.") print() print("=" * 78) print("7. CARDINALITY IS A CLIFF, NOT A SLOPE") print("=" * 78) budgeter = CardinalityBudget(max_series_per_metric=10_000) candidates = [ MetricSpec("requests_total", ( LabelSpec("tenant", 12), LabelSpec("deployment", 6), LabelSpec("outcome", 7))), MetricSpec("requests_by_model", ( LabelSpec("tenant", 12), LabelSpec("deployment", 6), LabelSpec("outcome", 7), LabelSpec("model", 8), LabelSpec("region", 3))), MetricSpec("latency_by_user", ( LabelSpec("tenant", 12), LabelSpec("user_id", 40_000))), MetricSpec("tokens_by_prompt", ( LabelSpec("tenant", 12), LabelSpec("prompt", 500))), ] for spec in candidates: verdict = budgeter.check(spec) mark = "OK " if verdict.ok else "FAIL" print(f" {mark} {spec.name:<20} {verdict.series:>10,} series") if not verdict.ok: print(f" {verdict.reason}") else: budgeter.register(spec) print(f" registered total: {budgeter.total_series():,} series") print(" -> series are the PRODUCT of label cardinalities. Adding one label with 100") print(" values multiplies everything, which is why a metric goes from affordable") print(" to unaffordable in a single commit — and why the backend falls over") print(" during an incident, when a new label seemed useful.") print() print("=" * 78) print("8. THE DEGRADATION LADDER") print("=" * 78) ladder_clock = _clock(start=0, step=1) ladder = DegradationLadder(recovery_hold_ticks=5, now=ladder_clock) print(f" {'burn':<8} {'level':<7} {'active'}") for burn in (1.0, 3.0, 20.0, 20.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 1.0, 0.5): state = ladder.evaluate(burn) print(f" {burn:<8.1f} {state.level:<7} {list(state.active)}") print(" -> descend FAST (jump straight to the level the burn rate implies), ascend") print(" SLOWLY (one rung, with a hold). Without hysteresis the ladder") print(" oscillates, and the user sees answer quality flapping.") print() print(" the ladder itself, written in daylight:") for i, step in enumerate(STANDARD_LADDER_STEPS, 1): visible = "USER-VISIBLE" if step.user_visible else "invisible" print(f" {i}. {step.name:<16} {step.description:<38} " f"saves {step.saves:<28} {visible}") print() print("=" * 78) print("9. COST PER SUCCESSFUL ACTION, AND THE BREAKER") print("=" * 78) spend = ( _events([(t, Outcome.SUCCESS, 800) for t in range(200)], tenant="wholesale") + _events([(t, Outcome.PLATFORM_ERROR, 200) for t in range(30)], tenant="wholesale") + _events([(t, Outcome.SUCCESS, 800) for t in range(50)], tenant="retail") ) for tenant in ("wholesale", "retail"): print(f" {cost_report(spend, tenant).format()}") print(" -> the denominator is SUCCESSFUL actions. Cost per REQUEST improves when") print(" you fail faster, which is exactly the wrong incentive.") print() breaker = CostCircuitBreaker(budgets_micros={"wholesale": 1_000_000, "retail": 500_000}) for tenant, amount, label in (("wholesale", 700_000, "normal week"), ("wholesale", 150_000, "a busy day"), ("wholesale", 400_000, "a runaway loop"), ("retail", 50_000, "normal")): breaker.record(tenant, amount) verdict = breaker.check(tenant) mark = "TRIPPED" if verdict.tripped else ("warn" if verdict.fraction_used >= 0.8 else "ok") print(f" {tenant:<10} +{amount:>8} ({label:<16}) -> {mark:<8}" f"{verdict.fraction_used * 100:>4.0f}% {verdict.reason}") print(f" retail still allowed: {breaker.allow('retail')}") print(" -> per tenant, so one runaway loop cannot exhaust anybody else's budget.") print(" A cost control IS an availability control: the runaway looks like") print(" success on every other signal.") print() print("=" * 78) print("10. CAPACITY IS A FORECAST") print("=" * 78) print(" limit = the PROVIDER's quota (TPM, GPU count), not CPU.") print(f" {'':<5} {'resource':<26} {'now':>6} {'growth':>8} {'periods':>9} lead") for label, samples, limit, lead in ( ("PAYG tokens/min quota", [40, 44, 48, 53, 58, 64, 70], 100.0, 2), ("PTU quota, 4wk lead", [40, 44, 48, 53, 58, 64, 70], 100.0, 5), ("GPU capacity, 12wk lead", [31, 34, 38, 42, 47, 51, 55], 100.0, 12), ("flat, no growth", [70, 69, 71, 70, 70, 71, 70], 100.0, 12), ): forecast = forecast_capacity(samples, limit=limit, lead_time_periods=lead) mark = "ALERT" if forecast.alert else " " periods = (f"{forecast.periods_to_limit:.1f}" if forecast.periods_to_limit is not None else "-") print(f" {mark} {label:<26} {forecast.current:>5.0f}% " f"{forecast.growth_per_period:>+7.1f} {periods:>9} {lead}") print(f" {forecast.reason}") print() print(" -> the same growth curve, three verdicts, because the LEAD TIME differs.") print(" GPU capacity alerts at 55% utilization: 11 periods of headroom is not") print(" enough when the procurement takes 12. A 90%-utilization alert arrives") print(" after the decision point, which makes it a notification, not a control.") print() print("=" * 78) print("11. WHAT CHANGED? — ONLY ANSWERABLE IF YOU PINNED IT") print("=" * 78) baseline = Baseline("gpt-frontier-2026-02-11", "prompt-v7", "corpus-2026-02-01", "policy-v12", 0.94, recorded_tick=1_000) for label, current in ( ("prompt changed", replace(baseline, prompt_version="prompt-v8", eval_score=0.81)), ("model changed", replace(baseline, model_version="gpt-frontier-2026-03-02", eval_score=0.79)), ("nothing changed", replace(baseline, eval_score=0.80)), ("no regression", replace(baseline, eval_score=0.93)), ): print(f" {label} (eval {baseline.eval_score:.2f} -> {current.eval_score:.2f}):") for hypothesis in classify_regression(baseline, current): if hypothesis.confidence in ("confirmed", "likely", "excluded") and \ hypothesis.confidence != "excluded": print(f" {hypothesis.confidence.upper():<10} {hypothesis.cause:<38} " f"{hypothesis.evidence}") print(" -> the third case is the interesting one: nothing WE control moved, so it") print(" is a provider change behind a stable version string or an input drift.") print(" 'It is one of these two' is a much better position than 'something") print(" changed', and it is only available because the versions were pinned.") if __name__ == "__main__": # pragma: no cover main()