"""Reference solution — Platform Reference Model & Budget Calculator. Everything here is deterministic: no wall clock, no randomness, no I/O. Money is carried as integer micro-USD so accumulated cost is exact and tests can compare with ``==``. Run ``python solution.py`` for the worked example from the phase WARMUP. """ from __future__ import annotations import math from dataclasses import dataclass, field from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple MINUTES_PER_DAY = 24 * 60 # Budgets and burn rates are ratios of floats, so an exact threshold comparison is # unreliable: ``1 - 0.999`` is 0.0009999999999998899, not 0.001, and a burn rate that is # mathematically exactly 14.4 computes as 14.399999999999986. Real alerting systems hit # this and silently fail to fire at the boundary. Compare with a tolerance instead. EPSILON = 1e-9 # -------------------------------------------------------------------------------------- # 1. Availability composition # -------------------------------------------------------------------------------------- def series_availability(values: Sequence[float]) -> float: """Availability of components that must ALL succeed: the product. An empty chain is 1.0 — the identity of a product, and the right answer for "a platform with no dependencies never fails because of a dependency". """ result = 1.0 for value in values: _check_probability(value, "availability") result *= value return result def parallel_availability(values: Sequence[float]) -> float: """Availability of redundant components: fails only if ALL fail. ``1 - prod(1 - A_i)``. An empty group is 0.0 — no members means nothing can serve. """ if not values: return 0.0 unavailability = 1.0 for value in values: _check_probability(value, "availability") unavailability *= 1.0 - value return 1.0 - unavailability def correlated_parallel_availability(values: Sequence[float], common_mode: float) -> float: """Redundancy with a shared failure mode. ``common_mode`` is the probability that a failure hits every member at once (shared region, shared deployment pipeline, shared upstream). With ``c`` the common-mode fraction and ``u_i = 1 - A_i``:: unavailability = c * mean(u_i) + (1 - c) * prod(u_i) At ``c = 0`` this reduces to :func:`parallel_availability`; at ``c = 1`` redundancy buys nothing and you get the average single-component availability. """ _check_probability(common_mode, "common_mode") if not values: return 0.0 unavailabilities = [] for value in values: _check_probability(value, "availability") unavailabilities.append(1.0 - value) independent = 1.0 for u in unavailabilities: independent *= u shared = sum(unavailabilities) / len(unavailabilities) return 1.0 - (common_mode * shared + (1.0 - common_mode) * independent) def _check_probability(value: float, label: str) -> None: if not 0.0 <= value <= 1.0: raise ValueError(f"{label} must be in [0, 1], got {value!r}") # -------------------------------------------------------------------------------------- # 2. The platform model # -------------------------------------------------------------------------------------- @dataclass(frozen=True) class Component: """One dependency in the request path. ``degradable=True`` means the request still succeeds when this component fails, with reduced quality. That single flag is the highest-leverage availability decision in the phase: a degradable dependency does not multiply into request availability. """ name: str availability: float degradable: bool = False p95_ms: int = 0 def __post_init__(self) -> None: _check_probability(self.availability, f"availability of {self.name!r}") if self.p95_ms < 0: raise ValueError("p95_ms must be >= 0") @dataclass(frozen=True) class PlatformModel: """A layered platform whose components sit in series on the request path.""" name: str components: Tuple[Component, ...] @classmethod def of(cls, name: str, components: Iterable[Component]) -> "PlatformModel": return cls(name=name, components=tuple(components)) def request_availability(self) -> float: """Availability of *answering at all* — degradable components excluded.""" return series_availability( [c.availability for c in self.components if not c.degradable] ) def quality_availability(self) -> float: """Availability of answering *at full quality* — every component counted.""" return series_availability([c.availability for c in self.components]) def downtime_minutes(self, window_days: int = 30, *, quality: bool = False) -> float: availability = self.quality_availability() if quality else self.request_availability() return (1.0 - availability) * window_days * MINUTES_PER_DAY def weakest_links(self, k: int = 3, *, include_degradable: bool = False) -> List[Tuple[str, float]]: """The top-k components by unavailability — where to spend, in order. Unavailabilities approximately add, so the largest ``1 - A`` terms dominate the composed number. Ties break on name so the result is deterministic. """ pool = [ c for c in self.components if include_degradable or not c.degradable ] ranked = sorted(pool, key=lambda c: (-(1.0 - c.availability), c.name)) return [(c.name, 1.0 - c.availability) for c in ranked[:k]] def with_degradable(self, *names: str) -> "PlatformModel": """Return a copy with the named components marked degradable.""" wanted = set(names) unknown = wanted - {c.name for c in self.components} if unknown: raise KeyError(f"unknown components: {sorted(unknown)}") return PlatformModel( name=self.name, components=tuple( Component(c.name, c.availability, c.name in wanted or c.degradable, c.p95_ms) for c in self.components ), ) def with_replaced(self, name: str, availability: float) -> "PlatformModel": """Return a copy with one component's availability replaced (e.g. after adding a redundant provider).""" if name not in {c.name for c in self.components}: raise KeyError(f"unknown component: {name!r}") return PlatformModel( name=self.name, components=tuple( Component(c.name, availability if c.name == name else c.availability, c.degradable, c.p95_ms) for c in self.components ), ) # -------------------------------------------------------------------------------------- # 3. Error budgets # -------------------------------------------------------------------------------------- @dataclass(frozen=True) class ErrorBudget: slo: float window_days: int = 30 def __post_init__(self) -> None: _check_probability(self.slo, "slo") if self.window_days <= 0: raise ValueError("window_days must be > 0") def total_minutes(self) -> float: return (1.0 - self.slo) * self.window_days * MINUTES_PER_DAY def allocate(self, shares: Mapping[str, float], *, tolerance: float = 1e-9) -> Dict[str, float]: """Split the budget across layers by weight. Weights must sum to 1.""" if not shares: raise ValueError("shares must not be empty") total = sum(shares.values()) if any(v < 0 for v in shares.values()): raise ValueError("shares must be non-negative") if abs(total - 1.0) > tolerance: raise ValueError(f"shares must sum to 1.0, got {total!r}") budget = self.total_minutes() return {layer: budget * weight for layer, weight in shares.items()} class BudgetLedger: """Tracks consumption of an allocated error budget. Deliberately never goes negative: a layer that has overspent reports 0 remaining and the overspend is visible via :meth:`overspend_for`. Reporting a negative budget makes dashboards lie about the platform total. """ def __init__(self, budget: ErrorBudget, shares: Mapping[str, float]) -> None: self.budget = budget self.allocations = budget.allocate(shares) self._consumed: Dict[str, float] = {layer: 0.0 for layer in self.allocations} def consume(self, layer: str, minutes: float) -> None: if layer not in self._consumed: raise KeyError(f"unknown layer: {layer!r}") if minutes < 0: raise ValueError("minutes must be >= 0") self._consumed[layer] += minutes def consumed_for(self, layer: str) -> float: return self._consumed[layer] def remaining_for(self, layer: str) -> float: return max(0.0, self.allocations[layer] - self._consumed[layer]) def overspend_for(self, layer: str) -> float: return max(0.0, self._consumed[layer] - self.allocations[layer]) def total_consumed(self) -> float: return sum(self._consumed.values()) def remaining(self) -> float: return max(0.0, self.budget.total_minutes() - self.total_consumed()) def fraction_remaining(self) -> float: total = self.budget.total_minutes() if total == 0: return 0.0 return self.remaining() / total def policy_state(self) -> str: """The error-budget policy, as a pure function of budget remaining. Agreed in writing before the first breach, so that when it triggers neither two-in-a-box owner has to argue about it. """ fraction = self.fraction_remaining() if fraction <= EPSILON: return "freeze" if fraction < 0.25: return "reliability-focus" if fraction < 0.50: return "elevated" return "normal" def burn_rate(observed_bad_ratio: float, slo: float) -> float: """How fast the budget is being spent, relative to spending it exactly on time.""" _check_probability(observed_bad_ratio, "observed_bad_ratio") _check_probability(slo, "slo") allowance = 1.0 - slo if allowance == 0.0: return math.inf if observed_bad_ratio > 0 else 0.0 return observed_bad_ratio / allowance def burn_rate_threshold(budget_fraction: float, window_hours: float, period_days: int = 30) -> float: """Derive the alerting threshold instead of memorizing 14.4. A window of ``window_hours`` is ``window_hours / (period_days * 24)`` of the period. Burning ``budget_fraction`` of the budget within it requires:: B = budget_fraction * (period_hours / window_hours) ``burn_rate_threshold(0.02, 1)`` -> 14.4; ``burn_rate_threshold(0.05, 6)`` -> 6.0. """ if window_hours <= 0: raise ValueError("window_hours must be > 0") if period_days <= 0: raise ValueError("period_days must be > 0") _check_probability(budget_fraction, "budget_fraction") period_hours = period_days * 24 return budget_fraction * (period_hours / window_hours) @dataclass(frozen=True) class AlertRule: name: str severity: str # "page" | "ticket" long_window_hours: float short_window_hours: float threshold: float DEFAULT_ALERT_RULES: Tuple[AlertRule, ...] = ( AlertRule("fast-burn", "page", 1.0, 1.0 / 12, 14.4), AlertRule("medium-burn", "page", 6.0, 0.5, 6.0), AlertRule("slow-burn", "ticket", 72.0, 6.0, 1.0), ) @dataclass(frozen=True) class MultiWindowAlertPolicy: """Fire only when a long window AND its short window both exceed the threshold. Long window alone is slow to resolve; short window alone pages on blips. Requiring both gives urgency *and* a clean auto-resolve. """ slo: float rules: Tuple[AlertRule, ...] = DEFAULT_ALERT_RULES def evaluate(self, bad_ratio_over: Callable[[float], float]) -> List[AlertRule]: """``bad_ratio_over(window_hours)`` returns the observed bad-event ratio.""" fired: List[AlertRule] = [] for rule in self.rules: long_burn = burn_rate(bad_ratio_over(rule.long_window_hours), self.slo) short_burn = burn_rate(bad_ratio_over(rule.short_window_hours), self.slo) floor = rule.threshold * (1.0 - EPSILON) if long_burn >= floor and short_burn >= floor: fired.append(rule) return fired def highest_severity(self, fired: Sequence[AlertRule]) -> Optional[str]: if any(rule.severity == "page" for rule in fired): return "page" if fired: return "ticket" return None # -------------------------------------------------------------------------------------- # 4. Latency budgets # -------------------------------------------------------------------------------------- @dataclass(frozen=True) class LatencyStage: name: str p95_ms: int group: Optional[str] = None # stages sharing a group run concurrently sheddable: bool = False def __post_init__(self) -> None: if self.p95_ms < 0: raise ValueError("p95_ms must be >= 0") @dataclass(frozen=True) class LatencyBudget: """Top-down budget. A parallel group contributes its MAX, not its sum.""" target_p95_ms: int stages: Tuple[LatencyStage, ...] @classmethod def of(cls, target_p95_ms: int, stages: Iterable[LatencyStage]) -> "LatencyBudget": if target_p95_ms <= 0: raise ValueError("target_p95_ms must be > 0") return cls(target_p95_ms=target_p95_ms, stages=tuple(stages)) def committed_ms(self, *, shed: Sequence[str] = ()) -> int: dropped = set(shed) groups: Dict[str, int] = {} total = 0 for stage in self.stages: if stage.name in dropped: continue if stage.group is None: total += stage.p95_ms else: groups[stage.group] = max(groups.get(stage.group, 0), stage.p95_ms) return total + sum(groups.values()) def headroom_ms(self, *, shed: Sequence[str] = ()) -> int: return self.target_p95_ms - self.committed_ms(shed=shed) def is_feasible(self, *, shed: Sequence[str] = ()) -> bool: return self.headroom_ms(shed=shed) >= 0 def fits_fallback(self, fallback_timeout_ms: int, *, shed: Sequence[str] = ()) -> bool: """A fallback that does not fit the remaining budget is decoration.""" if fallback_timeout_ms < 0: raise ValueError("fallback_timeout_ms must be >= 0") return self.headroom_ms(shed=shed) >= fallback_timeout_ms def shed_order(self) -> List[str]: """The degradation ladder: sheddable stages, most expensive first. Decided in daylight so it can be executed at 3 a.m. """ sheddable = [s for s in self.stages if s.sheddable] return [s.name for s in sorted(sheddable, key=lambda s: (-s.p95_ms, s.name))] def shed_until_fits(self, fallback_timeout_ms: int) -> List[str]: """Shed in ladder order until a fallback of the given timeout fits. Returns the stages shed (possibly empty; possibly the whole ladder without succeeding — the caller must re-check :meth:`fits_fallback`). """ shed: List[str] = [] for name in self.shed_order(): if self.fits_fallback(fallback_timeout_ms, shed=shed): return shed shed.append(name) return shed # -------------------------------------------------------------------------------------- # 5. Loop reliability # -------------------------------------------------------------------------------------- def loop_success(p: float, n: int) -> float: """``p ** n`` — the most important formula in agent engineering.""" _check_probability(p, "p") if n < 0: raise ValueError("n must be >= 0") return p ** n def effective_step_probability(p: float, retries: int) -> float: """``1 - (1 - p) ** attempts`` where attempts = retries + 1. Only valid when the step is idempotent AND the failure was transient. """ _check_probability(p, "p") if retries < 0: raise ValueError("retries must be >= 0") return 1.0 - (1.0 - p) ** (retries + 1) def max_steps_for_target(p: float, target: float) -> int: """Largest ``n`` with ``p ** n >= target``. ``p == 1`` never degrades, so there is no bound — raise rather than return a misleading sentinel. ``p == 0`` supports only the empty task. """ _check_probability(p, "p") _check_probability(target, "target") if p >= 1.0: raise ValueError("p == 1.0 has no step limit; the model is not useful there") if target <= 0.0: raise ValueError("target must be > 0") if p <= 0.0: return 0 return int(math.floor(math.log(target) / math.log(p))) # -------------------------------------------------------------------------------------- # 6. Cost # -------------------------------------------------------------------------------------- @dataclass(frozen=True) class TokenPrices: """Micro-USD per 1 000 tokens. Integers keep accumulated cost exact.""" input_per_1k: int cached_input_per_1k: int output_per_1k: int def __post_init__(self) -> None: for label in ("input_per_1k", "cached_input_per_1k", "output_per_1k"): if getattr(self, label) < 0: raise ValueError(f"{label} must be >= 0") @dataclass(frozen=True) class CostModel: prices: TokenPrices def step_cost_micros(self, tokens_in: int, tokens_cached: int, tokens_out: int) -> int: if min(tokens_in, tokens_cached, tokens_out) < 0: raise ValueError("token counts must be >= 0") if tokens_cached > tokens_in: raise ValueError("tokens_cached cannot exceed tokens_in") fresh = tokens_in - tokens_cached total = ( fresh * self.prices.input_per_1k + tokens_cached * self.prices.cached_input_per_1k + tokens_out * self.prices.output_per_1k ) # Integer division truncates toward zero; costs are per-1k so divide last. return total // 1000 def run_cost_micros( self, *, base_tokens: int, per_step_tokens: int, output_tokens_per_step: int, steps: int, cached_prefix_tokens: int = 0, ) -> int: """Cost of an ``n``-step run with an accumulating scratchpad. Step ``i`` (1-indexed) sends ``base + per_step * (i - 1)`` input tokens, so total input over the run is ``n*base + per_step * n(n-1)/2`` — quadratic in ``n``. """ if steps < 0: raise ValueError("steps must be >= 0") total = 0 for i in range(1, steps + 1): tokens_in = base_tokens + per_step_tokens * (i - 1) cached = min(cached_prefix_tokens, tokens_in) total += self.step_cost_micros(tokens_in, cached, output_tokens_per_step) return total @staticmethod def total_input_tokens(base_tokens: int, per_step_tokens: int, steps: int) -> int: """``n*b + a*n(n-1)/2`` — the quadratic scratchpad term, in closed form.""" if steps < 0: raise ValueError("steps must be >= 0") return steps * base_tokens + per_step_tokens * steps * (steps - 1) // 2 @staticmethod def cost_per_successful_action_micros(attempt_cost_micros: int, success_probability: float) -> int: _check_probability(success_probability, "success_probability") if success_probability == 0.0: raise ValueError("success_probability must be > 0") return round(attempt_cost_micros / success_probability) def effective_cost_with_cache(miss_cost: float, hit_cost: float, hit_rate: float) -> float: _check_probability(hit_rate, "hit_rate") return (1.0 - hit_rate) * miss_cost + hit_rate * hit_cost def cache_savings_fraction(miss_cost: float, hit_cost: float, hit_rate: float) -> float: _check_probability(hit_rate, "hit_rate") if miss_cost <= 0: raise ValueError("miss_cost must be > 0") return hit_rate * (1.0 - hit_cost / miss_cost) # -------------------------------------------------------------------------------------- # 7. The admission pipeline — which layer denies, and why # -------------------------------------------------------------------------------------- LAYERS: Tuple[str, ...] = ( "users_and_channels", "control_plane", "agent_kernel", "knowledge_foundation", "action_gateway", ) @dataclass(frozen=True) class AgentRegistration: agent_id: str tenant: str permitted_tools: Tuple[str, ...] granted_scopes: Tuple[str, ...] max_action_amount_micros: int evaluation_fresh: bool = True @dataclass(frozen=True) class ProposedAction: tenant: str # derived from the verified token, never from the request body agent_id: str tool: str channel: str user_authenticated: bool = True amount_micros: int = 0 resource_tenant: str = "" approvals: Tuple[str, ...] = () step_index: int = 1 run_cost_micros: int = 0 derived_from_untrusted_content: bool = False retrieved_tenants: Tuple[str, ...] = () def __post_init__(self) -> None: if self.amount_micros < 0: raise ValueError("amount_micros must be >= 0") if self.step_index < 1: raise ValueError("step_index must be >= 1") @dataclass(frozen=True) class Denial: layer: str code: str reason: str @dataclass(frozen=True) class AdmissionResult: allowed: bool denials: Tuple[Denial, ...] @property def primary(self) -> Optional[Denial]: """The earliest denial — what the caller is told.""" return self.denials[0] if self.denials else None @property def defence_depth(self) -> int: """How many *distinct layers* independently denied. 1 means a single point of failure with good intentions.""" return len({d.layer for d in self.denials}) @dataclass(frozen=True) class AdmissionPipeline: """Runs every layer's check and reports ALL denials, not just the first. Reporting all of them is the whole point of the phase: defence in depth is a claim you should be able to *measure*, and :attr:`AdmissionResult.defence_depth` measures it. """ registry: Mapping[str, AgentRegistration] tool_required_scope: Mapping[str, str] side_effecting_tools: frozenset dual_control_threshold_micros: int approval_capable_channels: frozenset max_steps: int = 25 max_run_cost_micros: int = 5_000_000 def evaluate(self, action: ProposedAction) -> AdmissionResult: denials: List[Denial] = [] registration = self.registry.get(action.agent_id) denials.extend(self._channel_checks(action)) denials.extend(self._control_plane_checks(action, registration)) denials.extend(self._kernel_checks(action)) denials.extend(self._knowledge_checks(action)) denials.extend(self._gateway_checks(action, registration)) order = {layer: i for i, layer in enumerate(LAYERS)} denials.sort(key=lambda d: (order[d.layer], d.code)) return AdmissionResult(allowed=not denials, denials=tuple(denials)) # -- layer 1 ----------------------------------------------------------------- def _channel_checks(self, action: ProposedAction) -> List[Denial]: out: List[Denial] = [] if not action.user_authenticated: out.append(Denial("users_and_channels", "UNAUTHENTICATED", "no authenticated human principal on the session")) if (action.amount_micros >= self.dual_control_threshold_micros and action.channel not in self.approval_capable_channels): out.append(Denial( "users_and_channels", "CHANNEL_CANNOT_APPROVE", f"channel {action.channel!r} cannot render an approval for an action " f"requiring dual control")) return out # -- layer 2 ----------------------------------------------------------------- def _control_plane_checks( self, action: ProposedAction, registration: Optional[AgentRegistration] ) -> List[Denial]: out: List[Denial] = [] if registration is None: out.append(Denial("control_plane", "AGENT_NOT_REGISTERED", f"agent {action.agent_id!r} is not in the agent registry")) return out if action.tool not in registration.permitted_tools: out.append(Denial("control_plane", "TOOL_NOT_PERMITTED", f"{action.tool!r} is not in the agent's registered tool set")) if not registration.evaluation_fresh: out.append(Denial("control_plane", "EVALUATION_STALE", "agent's evaluation results are stale; KYA posture failed")) if registration.tenant != action.tenant: out.append(Denial("control_plane", "AGENT_TENANT_MISMATCH", f"agent is registered to {registration.tenant!r} but the token " f"asserts {action.tenant!r}")) return out # -- layer 3 ----------------------------------------------------------------- def _kernel_checks(self, action: ProposedAction) -> List[Denial]: out: List[Denial] = [] if action.step_index > self.max_steps: out.append(Denial("agent_kernel", "STEP_BUDGET_EXCEEDED", f"step {action.step_index} exceeds max_steps={self.max_steps}")) if action.run_cost_micros > self.max_run_cost_micros: out.append(Denial("agent_kernel", "COST_CEILING_EXCEEDED", f"run cost {action.run_cost_micros} exceeds ceiling " f"{self.max_run_cost_micros}")) return out # -- layer 4 ----------------------------------------------------------------- def _knowledge_checks(self, action: ProposedAction) -> List[Denial]: out: List[Denial] = [] foreign = sorted({t for t in action.retrieved_tenants if t != action.tenant}) if foreign: out.append(Denial("knowledge_foundation", "CROSS_TENANT_RETRIEVAL", f"context includes material from {foreign}")) if action.derived_from_untrusted_content and action.tool in self.side_effecting_tools: out.append(Denial("knowledge_foundation", "UNTRUSTED_INSTRUCTION_SOURCE", "a side-effecting action was derived from retrieved content; " "retrieved content is data, never instruction")) return out # -- layer 5 ----------------------------------------------------------------- def _gateway_checks( self, action: ProposedAction, registration: Optional[AgentRegistration] ) -> List[Denial]: out: List[Denial] = [] if action.resource_tenant and action.resource_tenant != action.tenant: out.append(Denial("action_gateway", "TENANT_MISMATCH", f"resource belongs to {action.resource_tenant!r}, caller is " f"{action.tenant!r}")) required = self.tool_required_scope.get(action.tool) if required is not None and ( registration is None or required not in registration.granted_scopes ): out.append(Denial("action_gateway", "SCOPE_MISSING", f"credential lacks required scope {required!r}")) if registration is not None and action.amount_micros > registration.max_action_amount_micros: out.append(Denial("action_gateway", "ACTION_LIMIT_EXCEEDED", f"amount {action.amount_micros} exceeds the agent's limit " f"{registration.max_action_amount_micros}")) if action.amount_micros >= self.dual_control_threshold_micros: approvers = {a for a in action.approvals if a != action.agent_id} if len(approvers) < 2: out.append(Denial("action_gateway", "DUAL_CONTROL_REQUIRED", f"{len(approvers)} distinct human approver(s); 2 required")) return out # -------------------------------------------------------------------------------------- # Worked example # -------------------------------------------------------------------------------------- def _pct(x: float) -> str: return f"{x * 100:.3f}%" def _mins(x: float) -> str: hours, minutes = divmod(x, 60) return f"{x:.1f} min ({int(hours)}h {minutes:.0f}m)" def main() -> None: # pragma: no cover - narrative output print("=" * 78) print("1. THE FIVE-LAYER PLATFORM, COMPOSED") print("=" * 78) naive = PlatformModel.of("bank-ai-platform", [ Component("channel_apim", 0.9995, p95_ms=60), Component("control_plane_cached", 0.99999, p95_ms=10), Component("agent_kernel", 0.999, p95_ms=120), Component("model_layer", 0.998, p95_ms=800), Component("knowledge_foundation", 0.995, p95_ms=350), Component("action_gateway", 0.9995, p95_ms=900), ]) print(f" request availability : {_pct(naive.request_availability())}") print(f" monthly downtime : {_mins(naive.downtime_minutes())}") print(" weakest links :") for name, u in naive.weakest_links(3): print(f" {name:<24} unavailability {u:.5f}") print() print(" -> make retrieval degradable (BM25 stays serial at 0.999):") improved = naive.with_replaced("knowledge_foundation", 0.999) print(f" request availability: {_pct(improved.request_availability())}" f" downtime {_mins(improved.downtime_minutes())}") pair = correlated_parallel_availability([0.998, 0.998], common_mode=0.2) print(f" -> add a second model provider (common-mode c=0.2): {pair:.6f}") improved = improved.with_replaced("model_layer", pair) print(f" request availability: {_pct(improved.request_availability())}" f" downtime {_mins(improved.downtime_minutes())}") action_path = improved.request_availability() * 0.997 print(f" -> action path (x core banking 0.997): {_pct(action_path)}" f" downtime {_mins((1 - action_path) * 43200)}") print() print("=" * 78) print("2. ERROR BUDGET & ALERTING") print("=" * 78) budget = ErrorBudget(slo=0.995, window_days=30) shares = {"model_layer": 0.40, "integrations": 0.30, "agent_kernel": 0.20, "other": 0.10} ledger = BudgetLedger(budget, shares) print(f" total budget @ SLO {budget.slo}: {_mins(budget.total_minutes())}") for layer, minutes in ledger.allocations.items(): print(f" {layer:<16} {minutes:7.1f} min") ledger.consume("model_layer", 70.0) ledger.consume("integrations", 55.0) print(f" after two incidents -> remaining {_mins(ledger.remaining())}" f" ({ledger.fraction_remaining() * 100:.0f}%) state={ledger.policy_state()}") print() print(f" burn-rate threshold, 2% of budget in 1h : {burn_rate_threshold(0.02, 1):.1f}") print(f" burn-rate threshold, 5% of budget in 6h : {burn_rate_threshold(0.05, 6):.1f}") policy = MultiWindowAlertPolicy(slo=0.995) # A hard outage: 12% of requests failing over the last hour and still failing now. outage = {1.0: 0.12, 1 / 12: 0.15, 6.0: 0.03, 0.5: 0.15, 72.0: 0.004} fired = policy.evaluate(lambda w: outage.get(w, 0.004)) print(f" fired rules: {[r.name for r in fired]} -> " f"severity {policy.highest_severity(fired)}") print() print("=" * 78) print("3. LATENCY BUDGET") print("=" * 78) lb = LatencyBudget.of(3000, [ LatencyStage("ingress+authz", 60), LatencyStage("input_guardrails", 120, group="pre"), LatencyStage("retrieval", 200, group="pre"), LatencyStage("rerank", 150, sheddable=True), LatencyStage("model_ttft", 800), LatencyStage("action_gateway", 900), LatencyStage("output_guardrails", 60), LatencyStage("network+jitter", 200), ]) print(f" committed {lb.committed_ms()} ms headroom {lb.headroom_ms()} ms" f" feasible={lb.is_feasible()}") print(f" fits a 500 ms fallback? {lb.fits_fallback(500)}") print(f" fits a 700 ms fallback? {lb.fits_fallback(700)}" f" -> shed {lb.shed_until_fits(700)} then " f"{lb.fits_fallback(700, shed=lb.shed_until_fits(700))}") print() print("=" * 78) print("4. LOOP RELIABILITY & COST") print("=" * 78) for p in (0.99, 0.95, 0.90): row = " ".join(f"n={n}: {loop_success(p, n):.3f}" for n in (3, 5, 10, 20)) print(f" p={p}: {row}") print(f" p=0.95 with one retry -> {effective_step_probability(0.95, 1):.4f}") print(f" max steps at p=0.95 for a 0.90 task target -> {max_steps_for_target(0.95, 0.90)}") prices = TokenPrices(input_per_1k=3000, cached_input_per_1k=300, output_per_1k=15000) cost = CostModel(prices) tokens = CostModel.total_input_tokens(1000, 2000, 10) print(f" input tokens over a 10-step run (b=1000, a=2000): {tokens:,}") run = cost.run_cost_micros(base_tokens=1000, per_step_tokens=2000, output_tokens_per_step=300, steps=10) print(f" run cost: {run:,} micro-USD (${run / 1e6:.4f})") cached = cost.run_cost_micros(base_tokens=1000, per_step_tokens=2000, output_tokens_per_step=300, steps=10, cached_prefix_tokens=1000) print(f" with a 1000-token cached prefix: {cached:,} micro-USD " f"({(1 - cached / run) * 100:.1f}% saved)") for success in (0.70, 0.90): cpsa = CostModel.cost_per_successful_action_micros(run, success) print(f" cost per SUCCESSFUL action @ {success:.0%}: {cpsa:,} micro-USD " f"(${cpsa / 1e6:.4f})") print() print("=" * 78) print("5. ADMISSION — WHICH LAYER DENIES?") print("=" * 78) pipeline = AdmissionPipeline( registry={ "collections-01": AgentRegistration( agent_id="collections-01", tenant="retail", permitted_tools=("crm.read", "collections.note"), granted_scopes=("crm.read", "collections.write"), max_action_amount_micros=0, ) }, tool_required_scope={"payments.release": "payments.release"}, side_effecting_tools=frozenset({"payments.release", "collections.note"}), dual_control_threshold_micros=1_000_000_00, # 100 USD in micro-USD units approval_capable_channels=frozenset({"web", "teams"}), ) bad = ProposedAction( tenant="retail", agent_id="collections-01", tool="payments.release", channel="ivr", amount_micros=2_000_000_000, resource_tenant="wholesale", derived_from_untrusted_content=True, retrieved_tenants=("retail", "wholesale"), step_index=3, ) result = pipeline.evaluate(bad) print(f" allowed={result.allowed} distinct layers denying={result.defence_depth}") for d in result.denials: print(f" [{d.layer:<22}] {d.code:<28} {d.reason}") print(f" primary (what the caller sees): {result.primary.code}") good = ProposedAction(tenant="retail", agent_id="collections-01", tool="crm.read", channel="web") print(f" benign read -> allowed={pipeline.evaluate(good).allowed}") if __name__ == "__main__": # pragma: no cover main()