"""Lab 01 — The SRE console for a probabilistic system. Ordinary SRE assumes "correct" is a predicate. For an AI platform it is a *distribution*, and that single fact breaks three habits: quality cannot go in the availability SLI, the four golden signals are incomplete, and debugging is trace-first because you cannot re-run to reproduce. Work top to bottom; each section's tests pass once that section is done: 1. the validity predicate 2. SLIs 3. error budgets 4. burn rate 5. multi-window alerting 6. the span tree 7. cardinality governance 8. the degradation ladder 9. cost governance 10. capacity forecasting 11. incident classification pytest # against your work LAB_MODULE=solution pytest # against the reference Determinism rules: the clock is injected, money is integer micro-USD divided last, span ids are derived, and every collection you return is sorted. """ 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. """ 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: 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 defensible exclusions: * **client errors** — a malformed request is not an availability failure. Only safe if 4xx is genuinely the client's fault; * **user aborts** — counting these makes the SLI track user patience; * **synthetic probes** — 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 is the platform working — so not a success — but if it starts refusing 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: raise NotImplementedError def describe(self) -> str: """TODO: a short human-readable summary, for the SLI to carry.""" raise NotImplementedError #: TODO: only SUCCESS. Note what is NOT here: SAFETY_BLOCKED and UPSTREAM_ERROR. GOOD_OUTCOMES: FrozenSet[Outcome] = frozenset() # ====================================================================================== # 2. The SLI # ====================================================================================== @dataclass(frozen=True) class SliResult: good: int valid: int ratio: float predicate: str @property def bad(self) -> int: raise NotImplementedError def format(self) -> str: raise NotImplementedError def availability_sli(events: Sequence[RequestEvent], predicate: ValidityPredicate = ValidityPredicate()) -> SliResult: """TODO: ``good / valid``. An empty window is **1.0, not 0.0** — zero traffic is not an outage, and treating it as one means every quiet Sunday burns the entire budget. """ raise NotImplementedError def latency_sli(events: Sequence[RequestEvent], *, threshold_ms: int, predicate: ValidityPredicate = ValidityPredicate()) -> SliResult: """TODO: latency as a RATIO — the fraction of valid requests that both succeeded and landed at or under the threshold (inclusive). 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. """ raise NotImplementedError def percentile(values: Sequence[int], q: float) -> float: """TODO: nearest-rank. Reported alongside the ratio for diagnosis, never alerted on. Empty is 0.0.""" raise NotImplementedError # ====================================================================================== # 3. Error budgets # ====================================================================================== @dataclass(frozen=True) class Slo: name: str target: float # e.g. 0.995 window_ticks: int threshold_ms: Optional[int] = None # set for a latency SLO @property def budget_fraction(self) -> float: raise NotImplementedError @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: raise NotImplementedError class ErrorBudget: """A rolling budget over a window.""" def __init__(self, slo: Slo) -> None: self.slo = slo def state(self, events: Sequence[RequestEvent], predicate: ValidityPredicate = ValidityPredicate()) -> BudgetState: """TODO: pick the right SLI (latency if ``threshold_ms`` is set), then: * ``allowed = valid * budget_fraction``; * ``remaining = 1 - bad/allowed``, **clamped at 0** — a dashboard showing -340% tells you nothing you did not already know; * ``overspent_by`` = events beyond the allowance, reported **separately**, because "how far past" is a real question and clamping loses it; * ``exhausted`` when remaining is at (or within EPSILON of) zero. """ raise NotImplementedError def allocate_budget(total_fraction: float, weights: Mapping[str, float]) -> Dict[str, float]: """TODO: split proportionally; raise on non-positive total weight. A shared budget is a budget nobody owns. When the gateway spends the whole thing, the retrieval team should learn that from their own number rather than from an incident. """ raise NotImplementedError # ====================================================================================== # 4. Burn-rate alerting # ====================================================================================== def burn_rate_threshold(budget_fraction_consumed: float, window_hours: float, *, period_days: int = 30) -> float: """TODO: **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. Raise on a non-positive window. """ raise NotImplementedError def observed_burn_rate(sli: SliResult, slo: Slo) -> float: """TODO: the error rate as a multiple of the sustainable rate. 1.0 means exactly on budget for the period; 14.4 means the whole month in an hour. An empty window burns nothing. """ raise NotImplementedError @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 #: TODO: the standard ladder, with thresholds **derived** via burn_rate_threshold: #: fast-burn 2% in 1h -> 14.4x page (short window 5m, min 10) #: medium-burn 5% in 6h -> 6.0x page (short window 30m, min 60) #: slow-burn 10% in 24h -> 3.0x ticket (short window 2h, min 200) #: #: 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 keeps the alert firing for the #: rest of the hour — and an alert that stays lit after the fix is one people learn to #: close. STANDARD_LADDER: Tuple[BurnRateRule, ...] = () @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.""" def __init__(self, slo: Slo, rules: Sequence[BurnRateRule] = STANDARD_LADDER, *, ticks_per_hour: int = 60) -> None: raise NotImplementedError def evaluate(self, events: Sequence[RequestEvent], now: int, predicate: ValidityPredicate = ValidityPredicate() ) -> List[AlertDecision]: """TODO: one decision per rule, firing only when **both** windows exceed the threshold AND the long window has at least ``min_events``. Always give a reason, firing or not — including which condition failed, because "why didn't this page?" is a real question at 3 a.m. Compare against ``threshold * (1 - EPSILON)``: a mathematically-exact threshold computes just under itself in floating point, so a naive `>=` never fires. 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 and somebody raises the threshold until nothing ever fires. """ raise NotImplementedError def firing(self, events: Sequence[RequestEvent], now: int, predicate: ValidityPredicate = ValidityPredicate() ) -> List[AlertDecision]: raise NotImplementedError # ====================================================================================== # 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** (``gen_ai.system``, ``gen_ai.request.model``, ``gen_ai.usage.input_tokens``). Using the conventions 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: raise NotImplementedError @property def cost_micros(self) -> int: """TODO: from ``gen_ai.usage.cost_micros``; 0 when absent.""" raise NotImplementedError class Tracer: def __init__(self, *, now: Callable[[], int]) -> None: # TODO: span ids are DERIVED (``span-1``), never random. raise NotImplementedError def record(self, *, trace_id: str, parent_id: Optional[str], name: str, kind: SpanKind, start_tick: int, status: str = "ok", **attributes: Any) -> Span: raise NotImplementedError def trace(self, trace_id: str) -> List[Span]: raise NotImplementedError 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: # TODO: index by id, and build a children map keyed by parent id. Sort children # by (start_tick, span_id) so a walk is deterministic. raise NotImplementedError @property def roots(self) -> List[str]: """TODO: spans with no parent — **or a parent that is not in this set**, so an orphaned span is still reachable rather than silently dropped.""" raise NotImplementedError def walk(self, span_id: Optional[str] = None, depth: int = 0) -> Iterable[Tuple[int, Span]]: raise NotImplementedError def self_time(self, span_id: str) -> int: """TODO: duration minus the time covered by children. **The number that answers "where did the latency go?"** Subtract the **union** of child intervals, not their sum — concurrent children would otherwise be double-counted, making a parallel fan-out look like negative self time, which people then clamp to zero and stop trusting. Never return a negative number. """ raise NotImplementedError def time_by_kind(self) -> Dict[str, int]: """TODO: total SELF time per kind, sorted. The sum equals the root's duration.""" raise NotImplementedError def total_cost_micros(self) -> int: raise NotImplementedError def critical_path(self) -> List[str]: """TODO: the longest root-to-leaf chain by summed duration — what to optimize.""" raise NotImplementedError def errors(self) -> List[Span]: raise NotImplementedError # ====================================================================================== # 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: """TODO: the **product** of label cardinalities, not the sum. This is why the failure is a cliff rather than a slope: one label with 100 values multiplies everything, and a metric goes from affordable to unaffordable in a single commit. """ raise NotImplementedError @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 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: it does not degrade, it falls over — during an incident, because that is when a new label seemed useful. The rule: **high-cardinality identifiers live on traces and in the accounting store; metrics carry tenant, deployment and outcome.** """ def __init__(self, *, max_series_per_metric: int = 10_000, max_total_series: int = 200_000) -> None: raise NotImplementedError def check(self, spec: MetricSpec) -> CardinalityVerdict: """TODO: three rejections, in this order: 1. a **forbidden label** — reject regardless of estimated size, and say it belongs on a trace; 2. the per-metric budget — name the **worst-contributing label**; 3. the total budget, given what is already registered. """ raise NotImplementedError def register(self, spec: MetricSpec) -> MetricSpec: """TODO: check, raise ``ValueError`` on rejection, and do NOT register it.""" raise NotImplementedError def total_series(self) -> int: raise NotImplementedError # ====================================================================================== # 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 #: TODO: **written in daylight, executed at 3 a.m.** Cheapest loss first — shed accuracy #: before capability, capability before availability. The tests expect six steps starting #: with ``disable-rerank`` (the only one that is NOT user-visible), then #: ``smaller-model``, ``cache-only``, ``read-only``, ``queue``, ``reject``. STANDARD_LADDER_STEPS: Tuple[DegradationStep, ...] = () @dataclass(frozen=True) class LadderState: level: int active: Tuple[str, ...] reason: str @property def degraded(self) -> bool: raise NotImplementedError 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. * **recovery needs hysteresis.** Without it the ladder oscillates — degrade, load drops, restore, load returns — which is worse than staying degraded, because the user sees 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: # TODO: raise ValueError when there are fewer thresholds than steps. raise NotImplementedError def target_level(self, burn_rate: float) -> int: """TODO: the highest level whose threshold the burn rate meets (use EPSILON).""" raise NotImplementedError def evaluate(self, burn_rate: float) -> LadderState: """TODO: * target **above** the current level -> **jump straight there**, immediately; * target **below** -> only after ``recovery_hold_ticks`` since the last change, and then **one rung at a time**; * otherwise hold. Always give a reason, and record every transition in ``history``. """ raise NotImplementedError @property def user_visible_degradation(self) -> bool: raise NotImplementedError # ====================================================================================== # 8. Cost governance # ====================================================================================== @dataclass(frozen=True) class CostReport: 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: raise NotImplementedError def format(self) -> str: raise NotImplementedError def cost_report(events: Sequence[RequestEvent], tenant: str) -> CostReport: """TODO: ``cost per SUCCESSFUL action`` is the unit economic. Note the denominator. Cost per *request* improves when you fail faster, which is the wrong incentive — a platform that errors instantly has an excellent cost per request. Money is integer micro-USD throughout; divide only for display. Zero successes means zero per-action rather than a ZeroDivisionError. """ raise NotImplementedError @dataclass(frozen=True) class BreakerVerdict: tenant: str tripped: bool spent_micros: int budget_micros: int reason: str @property def fraction_used(self) -> float: raise NotImplementedError 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's budget. """ def __init__(self, *, budgets_micros: Mapping[str, int], warn_fraction: float = 0.8) -> None: raise NotImplementedError def record(self, tenant: str, micros: int) -> None: raise NotImplementedError def spent(self, tenant: str) -> int: raise NotImplementedError def check(self, tenant: str) -> BreakerVerdict: """TODO: a tenant with **no configured budget is denied** (fail closed); at or over budget trips; at or over ``warn_fraction`` warns without tripping. 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. """ raise NotImplementedError def allow(self, tenant: str) -> bool: raise NotImplementedError def reset(self, tenant: str) -> None: raise NotImplementedError # ====================================================================================== # 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: raise NotImplementedError def forecast_capacity(samples: Sequence[float], *, limit: float, lead_time_periods: int, safety_factor: float = 1.5) -> CapacityForecast: """TODO: project usage against a **provider limit** and alert with procurement lead time. * fewer than two samples -> no projection, no alert; * fit a least-squares slope over the samples; * slope <= 0 -> no projection; alert only if already at the limit; * otherwise ``periods = (limit - current) / slope``, and alert when that is inside ``lead_time_periods * safety_factor``. Two things 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. And **the alert must fire a lead time early**: GPU quota takes weeks and hardware takes months, so a 90%-utilization alert arrives after the decision point, which makes it a notification rather than a control. Always give a reason. Raise on a non-positive limit. """ raise NotImplementedError # ====================================================================================== # 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]: """TODO: "the model changed" vs "our prompt changed" vs "the corpus changed". * a drop under the threshold -> a single ``no-regression`` hypothesis, and stop; * each of the four pinned versions -> ``confirmed`` when it changed, ``excluded`` when it did not. Naming the exclusions matters as much as the confirmations; * **nothing changed and the eval dropped** -> a ``likely`` hypothesis that the provider changed behaviour behind a stable version string, or the input distribution moved. Sort ``confirmed`` first. This is deliberately mechanical — the technique is not clever inference, it is **having pinned the versions in the first place**. """ raise NotImplementedError # ====================================================================================== # Worked example # ====================================================================================== def main() -> None: # pragma: no cover """TODO (optional): once the tests pass, build the eleven-section demo. Compare against ``python solution.py`` — but only after your own runs. """ raise NotImplementedError if __name__ == "__main__": # pragma: no cover main()