"""Reference solution — the operating model, as code. Most candidates treat the leadership half of this JD as boilerplate. It is not: **two-in-a-box is a named operating model with specific mechanics**, and the interview will probe whether you have actually run one. The distinguishing property is that accountability is **undivided**, not partitioned. A normal EM/PM split says "you own tech, I own product", and it fails exactly at the boundary where AI platforms fail — where a product decision (autonomy band, onboarding pace) *is* an engineering risk decision. And the thing that makes shared accountability survive a real disagreement is not goodwill. It is **instruments**: an error-budget policy signed before the first breach, a disagreement protocol agreed in advance, ADRs that outlive both owners, an ORR gate that is a checklist rather than a feeling. This file builds them. The organizing idea throughout: **A standard in code is a control. A standard in a wiki is a suggestion.** Deterministic: an injected clock, sorted outputs, derived identifiers. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import json from collections import 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) # ====================================================================================== # 1. The operational readiness review # ====================================================================================== class Criticality(str, Enum): """Mandatory criteria are **not** weighted — they are gates. A weighted scoring system where everything is weighted lets a high total paper over a missing runbook. Splitting the two is what makes the ORR a gate rather than a grade. """ MANDATORY = "mandatory" ADVISORY = "advisory" @dataclass(frozen=True) class OrrCriterion: criterion_id: str question: str criticality: Criticality weight: int # advisory only; mandatory criteria are pass/fail category: str evidence_required: str # WHAT proves it — not "confirm that you have" def __post_init__(self) -> None: if self.criticality is Criticality.MANDATORY and self.weight: raise ValueError( f"{self.criterion_id}: a mandatory criterion must not carry a weight; " f"weighting it invites trading it away") #: The gate. Note ``evidence_required`` on every row: an ORR question answered "yes" is a #: belief; one answered with an artifact is a check. #: #: The six agent-specific rows at the end are the ones a generic ORR does not have, and #: they are the ones that matter for this platform. ORR_CRITERIA: Tuple[OrrCriterion, ...] = ( # -- the classic gate ------------------------------------------------------------ OrrCriterion("ORR-01", "Are SLOs defined and instrumented?", Criticality.MANDATORY, 0, "reliability", "a dashboard showing the SLI over 7 days"), OrrCriterion("ORR-02", "Have alerts been tested by INJECTING failure?", Criticality.MANDATORY, 0, "reliability", "a fault-injection run and the resulting page"), OrrCriterion("ORR-03", "Has the runbook been rehearsed by someone not on the team?", Criticality.MANDATORY, 0, "operability", "a rehearsal record naming the person and the date"), OrrCriterion("ORR-04", "Has rollback been tested in production-like conditions?", Criticality.MANDATORY, 0, "operability", "a rollback test record with the elapsed time"), OrrCriterion("ORR-05", "Are dependencies mapped with blast radius?", Criticality.MANDATORY, 0, "architecture", "a dependency diagram with composed availability"), OrrCriterion("ORR-06", "Is capacity headroom verified against the PROVIDER limit?", Criticality.MANDATORY, 0, "capacity", "a forecast against quota with the lead time stated"), OrrCriterion("ORR-07", "Is on-call trained and rostered?", Criticality.MANDATORY, 0, "operability", "a rota with names, and a completed shadow shift"), OrrCriterion("ORR-08", "Is there a documented degradation ladder?", Criticality.MANDATORY, 0, "reliability", "the ladder, with each rung's user-visible impact"), # -- agent-specific: the rows a generic ORR does not have ------------------------- OrrCriterion("ORR-09", "Is the evaluation suite passing at the tier's threshold?", Criticality.MANDATORY, 0, "agent", "an eval run with the score, the case count and the failures"), OrrCriterion("ORR-10", "Is the red-team suite passing on CONTAINMENT?", Criticality.MANDATORY, 0, "agent", "a red-team run with the containment rate"), OrrCriterion("ORR-11", "Have tool scopes been reviewed against least privilege?", Criticality.MANDATORY, 0, "agent", "a scope review naming the reviewer"), OrrCriterion("ORR-12", "Is a per-tenant cost ceiling configured?", Criticality.MANDATORY, 0, "cost", "the configured budget and the breaker's threshold"), OrrCriterion("ORR-13", "Is the autonomy band assigned and enforced?", Criticality.MANDATORY, 0, "agent", "the inventory entry showing tier and band"), OrrCriterion("ORR-14", "Can an evidence pack be generated for a sample action?", Criticality.MANDATORY, 0, "governance", "a generated pack for a test action"), # -- advisory: weighted, and genuinely optional ----------------------------------- OrrCriterion("ORR-15", "Is there a load test at 2x expected peak?", Criticality.ADVISORY, 15, "capacity", "load-test results"), OrrCriterion("ORR-16", "Is there a documented cost model per action?", Criticality.ADVISORY, 10, "cost", "cost per successful action"), OrrCriterion("ORR-17", "Has a chaos experiment been run?", Criticality.ADVISORY, 10, "reliability", "a chaos run and its findings"), OrrCriterion("ORR-18", "Is there a second engineer familiar with the system?", Criticality.ADVISORY, 20, "operability", "a named second"), OrrCriterion("ORR-19", "Are ADRs written for the major decisions?", Criticality.ADVISORY, 10, "architecture", "links to the ADRs"), OrrCriterion("ORR-20", "Is documentation current within 30 days?", Criticality.ADVISORY, 10, "operability", "a last-reviewed date"), ) @dataclass(frozen=True) class OrrAnswer: criterion_id: str satisfied: bool evidence: str = "" note: str = "" @dataclass(frozen=True) class OrrResult: service: str passed: bool advisory_score: int advisory_max: int mandatory_failures: Tuple[str, ...] unanswered: Tuple[str, ...] evidence_gaps: Tuple[str, ...] reasons: Tuple[str, ...] @property def advisory_percent(self) -> float: return (self.advisory_score / self.advisory_max * 100) if self.advisory_max else 0.0 def format(self) -> str: status = "PASS" if self.passed else "FAIL" return (f"{self.service}: {status} — advisory {self.advisory_score}/" f"{self.advisory_max} ({self.advisory_percent:.0f}%), " f"{len(self.mandatory_failures)} mandatory failure(s)") class OrrScorer: """Score an ORR. **Any mandatory failure is a fail, at any advisory score.** That is the whole design. The moment a mandatory criterion can be outweighed, the review becomes a negotiation, and the thing that gets negotiated away is always the runbook rehearsal — because it is the most inconvenient and the least visible. """ def __init__(self, criteria: Sequence[OrrCriterion] = ORR_CRITERIA, *, advisory_threshold: float = 70.0, require_evidence: bool = True) -> None: self.criteria = tuple(criteria) self.advisory_threshold = advisory_threshold self.require_evidence = require_evidence def score(self, service: str, answers: Sequence[OrrAnswer]) -> OrrResult: by_id = {a.criterion_id: a for a in answers} mandatory_failures: List[str] = [] unanswered: List[str] = [] evidence_gaps: List[str] = [] reasons: List[str] = [] advisory_score = 0 advisory_max = 0 for criterion in self.criteria: answer = by_id.get(criterion.criterion_id) if criterion.criticality is Criticality.ADVISORY: advisory_max += criterion.weight if answer is None: # An unanswered mandatory criterion is a FAILURE, not a gap. Silence is # not a pass; that distinction is what stops an ORR being completed by # omission. unanswered.append(criterion.criterion_id) if criterion.criticality is Criticality.MANDATORY: mandatory_failures.append(criterion.criterion_id) reasons.append(f"{criterion.criterion_id} unanswered: " f"{criterion.question}") continue if not answer.satisfied: if criterion.criticality is Criticality.MANDATORY: mandatory_failures.append(criterion.criterion_id) reasons.append( f"{criterion.criterion_id} not satisfied: {criterion.question}" + (f" ({answer.note})" if answer.note else "")) continue if self.require_evidence and not answer.evidence: # "Yes" with no artifact is a belief. For a mandatory criterion that is a # failure — an ORR of unevidenced yeses is a form somebody filled in. evidence_gaps.append(criterion.criterion_id) if criterion.criticality is Criticality.MANDATORY: mandatory_failures.append(criterion.criterion_id) reasons.append( f"{criterion.criterion_id} claimed without evidence; required: " f"{criterion.evidence_required}") continue if criterion.criticality is Criticality.ADVISORY: advisory_score += criterion.weight advisory_ok = (advisory_score / advisory_max * 100 >= self.advisory_threshold if advisory_max else True) if not mandatory_failures and not advisory_ok: reasons.append( f"advisory score {advisory_score}/{advisory_max} is below the " f"{self.advisory_threshold:.0f}% threshold") passed = not mandatory_failures and advisory_ok return OrrResult(service, passed, advisory_score, advisory_max, tuple(sorted(set(mandatory_failures))), tuple(sorted(unanswered)), tuple(sorted(set(evidence_gaps))), tuple(sorted(set(reasons)))) # ====================================================================================== # 2. The error-budget policy machine # ====================================================================================== class BudgetState(str, Enum): NORMAL = "normal" # > 50% remaining ELEVATED = "elevated" # 20-50% RELIABILITY_FOCUS = "reliability_focus" # < 20% FREEZE = "freeze" # exhausted class ChangeClass(str, Enum): """Change classes, ordered by risk. The policy permits a *prefix* of this list.""" EMERGENCY_FIX = "emergency_fix" # always permitted; the platform is broken RELIABILITY = "reliability" # makes the platform more reliable BUG_FIX = "bug_fix" CONFIG = "config" FEATURE = "feature" EXPERIMENT = "experiment" # the riskiest, and the first to go #: What each state permits. Note that ``EMERGENCY_FIX`` and ``RELIABILITY`` are permitted #: in **every** state including FREEZE — a freeze that blocks reliability work is a freeze #: that extends itself. STATE_POLICY: Mapping[BudgetState, FrozenSet[ChangeClass]] = { BudgetState.NORMAL: frozenset(ChangeClass), BudgetState.ELEVATED: frozenset({ ChangeClass.EMERGENCY_FIX, ChangeClass.RELIABILITY, ChangeClass.BUG_FIX, ChangeClass.CONFIG, ChangeClass.FEATURE}), BudgetState.RELIABILITY_FOCUS: frozenset({ ChangeClass.EMERGENCY_FIX, ChangeClass.RELIABILITY, ChangeClass.BUG_FIX}), BudgetState.FREEZE: frozenset({ ChangeClass.EMERGENCY_FIX, ChangeClass.RELIABILITY}), } @dataclass(frozen=True) class Exception_: """An exception to the policy. **Deliberately expensive.** Four properties, each closing a way a policy dies: * **it expires** — an exception without an expiry is a policy change nobody agreed to; * **both owners must approve** — one owner cannot suspend the shared instrument; * **it names a reason** — "business need" is not a reason; * **it is counted** — the exception rate is the health metric for the policy itself. Two a quarter is a working policy; ten is a policy that has been replaced by a habit. """ exception_id: str change_class: ChangeClass reason: str approved_by: Tuple[str, ...] granted_at: int expires_at: int used: bool = False def is_live(self, tick: int) -> bool: return not self.used and tick < self.expires_at class PolicyError(Exception): pass @dataclass(frozen=True) class ChangeVerdict: permitted: bool state: BudgetState change_class: ChangeClass reason: str exception_used: Optional[str] = None class ErrorBudgetPolicy: """State from budget remaining; permission from state; exceptions, expensively. **Signed before the first breach.** That is the entire point: a policy agreed while the budget is healthy is a rule, and one negotiated during a breach is an argument somebody wins on seniority. """ def __init__(self, *, owners: Tuple[str, str], now: Callable[[], int], thresholds: Mapping[BudgetState, float] = None, max_exceptions_per_window: int = 2, exception_window_ticks: int = 90) -> None: if len(set(owners)) != 2: raise PolicyError("two-in-a-box needs two distinct owners") self.owners = tuple(sorted(owners)) self.now = now self.thresholds = dict(thresholds or { BudgetState.NORMAL: 0.5, BudgetState.ELEVATED: 0.2, BudgetState.RELIABILITY_FOCUS: 0.0}) self.max_exceptions_per_window = max_exceptions_per_window self.exception_window_ticks = exception_window_ticks self._exceptions: Dict[str, Exception_] = {} self._counter = 0 self.decisions: List[ChangeVerdict] = [] def state_for(self, budget_remaining: float) -> BudgetState: if budget_remaining > self.thresholds[BudgetState.NORMAL]: return BudgetState.NORMAL if budget_remaining > self.thresholds[BudgetState.ELEVATED]: return BudgetState.ELEVATED if budget_remaining > self.thresholds[BudgetState.RELIABILITY_FOCUS]: return BudgetState.RELIABILITY_FOCUS return BudgetState.FREEZE def grant_exception(self, change_class: ChangeClass, *, reason: str, approved_by: Sequence[str], ttl_ticks: int = 7) -> Exception_: """**Both owners**, a real reason, and an expiry.""" approvers = tuple(sorted(set(approved_by))) if set(approvers) != set(self.owners): raise PolicyError( f"an exception needs both owners {list(self.owners)}; got " f"{list(approvers)} — one owner cannot suspend the shared instrument") if len(reason.split()) < 5: raise PolicyError( "an exception needs a real reason; 'business need' is not a reason") recent = self.recent_exceptions() if len(recent) >= self.max_exceptions_per_window: raise PolicyError( f"{len(recent)} exceptions in the last {self.exception_window_ticks} " f"ticks is at the limit of {self.max_exceptions_per_window}; the policy " f"is being replaced by a habit — renegotiate it instead") self._counter += 1 exception = Exception_(f"exc-{self._counter}", change_class, reason, approvers, self.now(), self.now() + ttl_ticks) self._exceptions[exception.exception_id] = exception return exception def recent_exceptions(self) -> List[Exception_]: cutoff = self.now() - self.exception_window_ticks return sorted((e for e in self._exceptions.values() if e.granted_at > cutoff), key=lambda e: e.exception_id) def live_exceptions(self) -> List[Exception_]: return sorted((e for e in self._exceptions.values() if e.is_live(self.now())), key=lambda e: e.exception_id) def may_ship(self, change_class: ChangeClass, budget_remaining: float, *, consume_exception: bool = True) -> ChangeVerdict: state = self.state_for(budget_remaining) if change_class in STATE_POLICY[state]: verdict = ChangeVerdict(True, state, change_class, f"{state.value} permits {change_class.value}") self.decisions.append(verdict) return verdict for exception in self.live_exceptions(): if exception.change_class is change_class: if consume_exception: self._exceptions[exception.exception_id] = replace(exception, used=True) verdict = ChangeVerdict( True, state, change_class, f"{state.value} would forbid {change_class.value}; permitted by " f"exception {exception.exception_id} ({exception.reason})", exception.exception_id) self.decisions.append(verdict) return verdict verdict = ChangeVerdict( False, state, change_class, f"{state.value} permits only " f"{sorted(c.value for c in STATE_POLICY[state])}") self.decisions.append(verdict) return verdict def exception_rate(self) -> float: """The health metric **for the policy itself**.""" return len(self.recent_exceptions()) / max(1, self.max_exceptions_per_window) # ====================================================================================== # 3. The decision router and the disagreement protocol # ====================================================================================== class DecisionClass(str, Enum): """Reversibility and visibility decide who must sign. The classification is the mechanism: a two-in-a-box pair that requires both signatures on everything is a pair that cannot move, and one that requires neither is two people who will disagree in public later. """ REVERSIBLE_INTERNAL = "reversible_internal" # one owner REVERSIBLE_EXTERNAL = "reversible_external" # one owner, the other informed IRREVERSIBLE_INTERNAL = "irreversible_internal" # both IRREVERSIBLE_EXTERNAL = "irreversible_external" # both, plus a forum def classify_decision(*, reversible: bool, externally_visible: bool) -> DecisionClass: if reversible: return (DecisionClass.REVERSIBLE_EXTERNAL if externally_visible else DecisionClass.REVERSIBLE_INTERNAL) return (DecisionClass.IRREVERSIBLE_EXTERNAL if externally_visible else DecisionClass.IRREVERSIBLE_INTERNAL) REQUIRED_SIGNERS: Mapping[DecisionClass, int] = { DecisionClass.REVERSIBLE_INTERNAL: 1, DecisionClass.REVERSIBLE_EXTERNAL: 1, DecisionClass.IRREVERSIBLE_INTERNAL: 2, DecisionClass.IRREVERSIBLE_EXTERNAL: 2, } class DisagreementKind(str, Enum): """Separating these is the protocol's central move. A **factual** disagreement has an answer: measure it. Most disagreements that feel like values turn out to be factual once somebody asks "what would change your mind?" A **values** disagreement does not, and the failure mode there is averaging — a design that is the midpoint of two coherent positions and is worse than either. Escalate **both written positions** instead, and let whoever owns the trade-off choose one. """ FACTUAL = "factual" VALUES = "values" UNCLEAR = "unclear" @dataclass(frozen=True) class Position: owner: str summary: str rationale: str what_would_change_my_mind: str # ← the field that classifies the disagreement @dataclass(frozen=True) class Disagreement: disagreement_id: str topic: str positions: Tuple[Position, ...] kind: DisagreementKind resolution: Optional[str] = None resolved_by: Optional[str] = None escalated_to: Optional[str] = None committed_by: Tuple[str, ...] = () def classify_disagreement(positions: Sequence[Position]) -> DisagreementKind: """Factual if **every** position names something that would change its holder's mind. That is the operational test, and it is a good one: a position that nothing would change is not an engineering position, and asking the question surfaces that in a minute rather than a meeting. """ if len(positions) < 2: raise ValueError("a disagreement needs at least two positions") stated = [bool(p.what_would_change_my_mind.strip()) for p in positions] if all(stated): return DisagreementKind.FACTUAL if not any(stated): return DisagreementKind.VALUES return DisagreementKind.UNCLEAR # ====================================================================================== # 4. The ADR store # ====================================================================================== class AdrStatus(str, Enum): PROPOSED = "proposed" ACCEPTED = "accepted" SUPERSEDED = "superseded" REJECTED = "rejected" @dataclass(frozen=True) class Adr: """One decision. Immutable once accepted; **superseded, never edited.** Editing an accepted ADR destroys the thing it exists for: a record of what was decided *and why, at the time, with the information then available*. A superseded ADR plus its successor tells you the decision changed and why; an edited one tells you nothing and quietly rewrites history. """ adr_id: str title: str status: AdrStatus context: str options: Tuple[str, ...] decision: str positive_consequences: Tuple[str, ...] negative_consequences: Tuple[str, ...] # ← required; see accept() decision_class: DecisionClass signers: Tuple[str, ...] created_at: int accepted_at: Optional[int] = None superseded_by: Optional[str] = None disagreement_id: Optional[str] = None def digest(self) -> str: return hashlib.sha256(json.dumps({ "adr_id": self.adr_id, "title": self.title, "context": self.context, "options": list(self.options), "decision": self.decision, "positive": list(self.positive_consequences), "negative": list(self.negative_consequences), "signers": sorted(self.signers), }, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16] class AdrError(Exception): pass class AdrStore: """Architectural memory that outlives both owners. Which is the actual justification. Two-in-a-box means two people hold the platform's architecture in their heads, and heads leave. An ADR is the mechanism by which the *reasoning* survives — and the reasoning is what a successor needs, not the decision. """ def __init__(self, *, owners: Tuple[str, str], now: Callable[[], int]) -> None: self.owners = tuple(sorted(owners)) self.now = now self._adrs: Dict[str, Adr] = {} self._counter = 0 def propose(self, *, title: str, context: str, options: Sequence[str], decision: str, positive: Sequence[str], negative: Sequence[str], decision_class: DecisionClass, disagreement_id: Optional[str] = None) -> Adr: if len(options) < 2: raise AdrError( "an ADR needs at least two options; one option is not a decision, it is " "a description") self._counter += 1 adr = Adr(f"ADR-{self._counter:04d}", title, AdrStatus.PROPOSED, context, tuple(options), decision, tuple(positive), tuple(negative), decision_class, (), self.now(), disagreement_id=disagreement_id) self._adrs[adr.adr_id] = adr return adr def accept(self, adr_id: str, *, signers: Sequence[str]) -> Adr: """Enough signers for the decision class, and **negative consequences stated**. The negative-consequences requirement is the one that changes ADR quality. An ADR with none has not been thought about — every real decision costs something, and naming the cost is what lets a successor tell whether the trade-off still holds. """ adr = self.get(adr_id) if adr.status is not AdrStatus.PROPOSED: raise AdrError(f"{adr_id} is {adr.status.value} and cannot be accepted") if not adr.negative_consequences: raise AdrError( f"{adr_id} states no negative consequences; every real decision costs " f"something, and an ADR without the cost has not been thought about") unique = tuple(sorted(set(signers))) required = REQUIRED_SIGNERS[adr.decision_class] if len(unique) < required: raise AdrError( f"{adr_id} is {adr.decision_class.value} and needs {required} signer(s); " f"got {len(unique)}") unknown = sorted(set(unique) - set(self.owners)) if unknown: raise AdrError(f"{unknown} are not owners of this platform") accepted = replace(adr, status=AdrStatus.ACCEPTED, signers=unique, accepted_at=self.now()) self._adrs[adr_id] = accepted return accepted def reject(self, adr_id: str, *, signers: Sequence[str]) -> Adr: adr = self.get(adr_id) if adr.status is not AdrStatus.PROPOSED: raise AdrError(f"{adr_id} is {adr.status.value}") rejected = replace(adr, status=AdrStatus.REJECTED, signers=tuple(sorted(set(signers)))) self._adrs[adr_id] = rejected return rejected def supersede(self, adr_id: str, *, by: str) -> Adr: """The only legal way to change an accepted decision.""" adr = self.get(adr_id) successor = self.get(by) if adr.status is not AdrStatus.ACCEPTED: raise AdrError(f"only an accepted ADR can be superseded; {adr_id} is " f"{adr.status.value}") if successor.status is not AdrStatus.ACCEPTED: raise AdrError(f"{by} must itself be accepted before it can supersede") superseded = replace(adr, status=AdrStatus.SUPERSEDED, superseded_by=by) self._adrs[adr_id] = superseded return superseded def amend(self, adr_id: str, **_: Any) -> Adr: """Always refuses. The method exists so the refusal is discoverable.""" adr = self.get(adr_id) if adr.status is AdrStatus.ACCEPTED: raise AdrError( f"{adr_id} is accepted and immutable; supersede it with a new ADR — " f"editing destroys the record of what was decided and why, at the time") raise AdrError("ADRs are not edited; propose a new one") def get(self, adr_id: str) -> Adr: try: return self._adrs[adr_id] except KeyError: raise AdrError(f"unknown ADR: {adr_id}") from None def accepted(self) -> List[Adr]: return sorted((a for a in self._adrs.values() if a.status is AdrStatus.ACCEPTED), key=lambda a: a.adr_id) def all(self) -> List[Adr]: return sorted(self._adrs.values(), key=lambda a: a.adr_id) def chain(self, adr_id: str) -> List[Adr]: """The supersession chain — how this decision evolved.""" out = [self.get(adr_id)] seen = {adr_id} while out[-1].superseded_by: nxt = out[-1].superseded_by if nxt in seen: raise AdrError(f"supersession cycle at {nxt}") seen.add(nxt) out.append(self.get(nxt)) return out # ====================================================================================== # 5. The design-review checklist # ====================================================================================== @dataclass(frozen=True) class DesignDocument: """A structured design. The **fields are the review**. Requiring the document to be structured is itself the intervention: a prose document can omit the blast radius without anyone noticing, and a typed one cannot. """ title: str author: str what_it_denies: str = "" # the five questions from Phase 00 blast_radius: str = "" degradation_behaviour: str = "" artifacts_emitted: Tuple[str, ...] = () operator_at_3am: str = "" dependencies: Tuple[str, ...] = () side_effecting_tools: Tuple[str, ...] = () data_classifications: Tuple[str, ...] = () autonomy_band: str = "" slo: str = "" retry_policy: str = "" idempotency: str = "" residency: str = "" reversible: bool = True @dataclass(frozen=True) class Finding: severity: str # "blocker" | "major" | "minor" rule: str detail: str def __str__(self) -> str: return f"[{self.severity.upper()}] {self.rule}: {self.detail}" #: A rule reads a design and returns a finding, or None. ReviewRule = Callable[[DesignDocument], Optional[Finding]] def _blocker(rule: str, detail: str) -> Finding: return Finding("blocker", rule, detail) #: The standing red flags, assembled from every phase in this track. The five questions #: from Phase 00 are blockers because a design that cannot answer them has not been #: designed. def rule_denies(d: DesignDocument) -> Optional[Finding]: if not d.what_it_denies: return _blocker("what-does-it-deny", "a component that denies nothing is not a control (Phase 00)") return None def rule_blast_radius(d: DesignDocument) -> Optional[Finding]: if not d.blast_radius: return _blocker("blast-radius", "no stated blast radius; you cannot reason about the failure " "without it (Phase 00)") return None def rule_degradation(d: DesignDocument) -> Optional[Finding]: if not d.degradation_behaviour: return _blocker("degradation", "no degradation behaviour; 'it fails' is not a design " "(Phase 14)") return None def rule_artifacts(d: DesignDocument) -> Optional[Finding]: if not d.artifacts_emitted: return _blocker("emits-nothing", "a control that emits no artifact does not exist as far as " "audit is concerned (Phase 15)") return None def rule_operator(d: DesignDocument) -> Optional[Finding]: if not d.operator_at_3am: return _blocker("who-operates-it", "nobody named as the 3 a.m. operator (Phase 14)") return None def rule_idempotency(d: DesignDocument) -> Optional[Finding]: if d.side_effecting_tools and not d.idempotency: return _blocker("idempotency", f"side-effecting tools {list(d.side_effecting_tools)} with no " f"idempotency story (Phase 10)") return None def rule_retry(d: DesignDocument) -> Optional[Finding]: if d.side_effecting_tools and not d.retry_policy: return Finding("major", "retry-policy", "side-effecting tools with no stated retry policy; the default " "will be wrong (Phase 10)") return None def rule_residency(d: DesignDocument) -> Optional[Finding]: if any(c in ("restricted", "confidential") for c in d.data_classifications) \ and not d.residency: return _blocker("residency", "restricted or confidential data with no residency statement " "(Phase 15)") return None def rule_autonomy(d: DesignDocument) -> Optional[Finding]: if d.side_effecting_tools and not d.autonomy_band: return _blocker("autonomy-band", "side-effecting tools with no assigned autonomy band " "(Phase 15)") return None def rule_slo(d: DesignDocument) -> Optional[Finding]: if not d.slo: return Finding("major", "slo", "no SLO stated (Phase 14)") return None def rule_dependencies(d: DesignDocument) -> Optional[Finding]: if not d.dependencies: return Finding("minor", "dependencies", "no dependencies listed; verify this is genuinely standalone") return None def rule_irreversible_review(d: DesignDocument) -> Optional[Finding]: if not d.reversible and not d.autonomy_band: return _blocker("irreversible-unbanded", "an irreversible design with no autonomy band") return None STANDING_RULES: Tuple[ReviewRule, ...] = ( rule_denies, rule_blast_radius, rule_degradation, rule_artifacts, rule_operator, rule_idempotency, rule_retry, rule_residency, rule_autonomy, rule_slo, rule_dependencies, rule_irreversible_review, ) @dataclass(frozen=True) class ReviewResult: document: str approved: bool findings: Tuple[Finding, ...] @property def blockers(self) -> Tuple[Finding, ...]: return tuple(f for f in self.findings if f.severity == "blocker") def format(self) -> str: return (f"{self.document}: {'APPROVED' if self.approved else 'BLOCKED'} — " f"{len(self.blockers)} blocker(s), {len(self.findings)} finding(s)") class DesignReview: """Run the standing rules. **Any blocker blocks.** The value is not that the rules are clever — they are deliberately obvious. It is that they are *standing*: the same list every time, so a reviewer having a bad day still catches the missing blast radius, and an author knows in advance what will be asked. """ def __init__(self, rules: Sequence[ReviewRule] = STANDING_RULES) -> None: self.rules = tuple(rules) def review(self, document: DesignDocument) -> ReviewResult: findings: List[Finding] = [] for rule in self.rules: try: finding = rule(document) except Exception as exc: # noqa: BLE001 - fail closed findings.append(_blocker("rule-error", f"a review rule failed: {exc}")) continue if finding: findings.append(finding) ordered = tuple(sorted(findings, key=lambda f: ( {"blocker": 0, "major": 1, "minor": 2}[f.severity], f.rule))) return ReviewResult(document.title, not any(f.severity == "blocker" for f in ordered), ordered) # ====================================================================================== # 6. Incident review tracking # ====================================================================================== class ActionStatus(str, Enum): OPEN = "open" IN_PROGRESS = "in_progress" DONE = "done" DROPPED = "dropped" # explicitly abandoned, with a reason @dataclass(frozen=True) class ActionItem: action_id: str incident_id: str description: str owner: str # a named human due_tick: int status: ActionStatus = ActionStatus.OPEN closed_at: Optional[int] = None drop_reason: str = "" def is_overdue(self, tick: int) -> bool: return self.status in (ActionStatus.OPEN, ActionStatus.IN_PROGRESS) \ and tick > self.due_tick @dataclass(frozen=True) class Incident: incident_id: str title: str severity: str started_at: int mitigated_at: Optional[int] = None resolved_at: Optional[int] = None budget_consumed: float = 0.0 blameless_review_at: Optional[int] = None @property def time_to_mitigate(self) -> Optional[int]: return (self.mitigated_at - self.started_at) if self.mitigated_at else None @property def time_to_resolve(self) -> Optional[int]: return (self.resolved_at - self.started_at) if self.resolved_at else None @dataclass(frozen=True) class ReviewHealth: """The honest measure of a post-mortem culture. Not "did we write post-mortems" — everybody writes post-mortems. **Did the actions get done?** A process whose actions are never completed is a writing exercise, and the completion rate is the only number that distinguishes the two. """ total_actions: int completed: int overdue: int dropped: int completion_rate: float overdue_rate: float reviews_held: int reviews_due: int @property def counted(self) -> int: """The denominator: total minus dropped, because dropped items were decided.""" return self.total_actions - self.dropped def format(self) -> str: return (f"{self.completed}/{self.counted} actions complete " f"({self.completion_rate * 100:.0f}%; {self.dropped} dropped, " f"excluded), {self.overdue} overdue, " f"{self.reviews_held}/{self.reviews_due} reviews held") class IncidentTracker: def __init__(self, *, now: Callable[[], int], review_due_ticks: int = 5) -> None: self.now = now self.review_due_ticks = review_due_ticks self._incidents: Dict[str, Incident] = {} self._actions: Dict[str, ActionItem] = {} self._counter = 0 def record(self, incident: Incident) -> Incident: self._incidents[incident.incident_id] = incident return incident def mitigate(self, incident_id: str) -> Incident: """**Mitigation is not a fix.** The distinction is the incident-command one: mitigation stops the bleeding, the fix removes the cause, and conflating them is how a mitigated incident is closed and recurs.""" incident = self._incidents[incident_id] updated = replace(incident, mitigated_at=self.now()) self._incidents[incident_id] = updated return updated def resolve(self, incident_id: str) -> Incident: incident = self._incidents[incident_id] if incident.mitigated_at is None: raise ValueError( f"{incident_id} was never mitigated; an incident cannot be resolved " f"before it stopped") updated = replace(incident, resolved_at=self.now()) self._incidents[incident_id] = updated return updated def hold_review(self, incident_id: str) -> Incident: incident = self._incidents[incident_id] updated = replace(incident, blameless_review_at=self.now()) self._incidents[incident_id] = updated return updated def add_action(self, incident_id: str, *, description: str, owner: str, due_in: int) -> ActionItem: if incident_id not in self._incidents: raise ValueError(f"unknown incident: {incident_id}") if not owner: raise ValueError( "an action item needs a named owner; 'the team' completes nothing") self._counter += 1 action = ActionItem(f"AI-{self._counter:03d}", incident_id, description, owner, self.now() + due_in) self._actions[action.action_id] = action return action def complete(self, action_id: str) -> ActionItem: action = self._actions[action_id] updated = replace(action, status=ActionStatus.DONE, closed_at=self.now()) self._actions[action_id] = updated return updated def drop(self, action_id: str, *, reason: str) -> ActionItem: """Dropping is legitimate and must be **explicit and reasoned**. An action quietly left open forever is worse than one dropped with a reason: the first corrupts the completion metric and the second is a decision. """ if len(reason.split()) < 3: raise ValueError("dropping an action item needs a stated reason") action = self._actions[action_id] updated = replace(action, status=ActionStatus.DROPPED, closed_at=self.now(), drop_reason=reason) self._actions[action_id] = updated return updated def overdue(self) -> List[ActionItem]: return sorted((a for a in self._actions.values() if a.is_overdue(self.now())), key=lambda a: (a.due_tick, a.action_id)) def health(self) -> ReviewHealth: actions = list(self._actions.values()) completed = sum(1 for a in actions if a.status is ActionStatus.DONE) dropped = sum(1 for a in actions if a.status is ActionStatus.DROPPED) overdue = len(self.overdue()) # Dropped items are excluded from the denominator: they were decided, not missed. counted = len(actions) - dropped reviews_due = sum(1 for i in self._incidents.values() if i.resolved_at is not None and self.now() - i.resolved_at >= self.review_due_ticks) reviews_held = sum(1 for i in self._incidents.values() if i.blameless_review_at is not None) return ReviewHealth( len(actions), completed, overdue, dropped, completed / counted if counted else 1.0, overdue / counted if counted else 0.0, reviews_held, reviews_due) # ====================================================================================== # 7. The forum playbook # ====================================================================================== @dataclass(frozen=True) class Forum: """Five audiences who ask different questions and reward different answers. Bringing the same deck to all five is the standard mistake, and each rejects it for a different reason. """ name: str wants: str artifact: str fails_when: str from_phase: str FORUMS: Tuple[Forum, ...] = ( Forum("Enterprise Architecture", "how it fits the target state, and what it standardizes rather than forks", "a reference architecture diagram with the five layers, plus the ADRs", "you present a bespoke design with no story about convergence", "Phase 00"), Forum("Cyber", "the threat model, the trust boundaries, and what a compromise reaches", "the identity model, the taint/containment argument, and the red-team results", "you claim you prevent prompt injection", "Phases 08, 11"), Forum("Model Risk", "what the model is, who validated it, and what it is NOT for", "the inventory entry, the validation pack, and the tiering rationale", "'the model' means only the weights", "Phase 15"), Forum("Internal Audit", "evidence that the control operated — not that it exists", "a generated evidence pack, and the control-to-evidence coverage map", "you describe controls instead of showing their artifacts", "Phase 15"), Forum("Group CTTO", "cost, capability, concentration risk, and the two-year direction", "cost per successful action, the capacity forecast, the exit readiness", "you present engineering detail instead of unit economics", "Phases 05, 14, 15"), ) def forum_brief(forum_name: str) -> Forum: for forum in FORUMS: if forum.name.lower() == forum_name.lower(): return forum raise KeyError(f"unknown forum: {forum_name}") # ====================================================================================== # 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 OWNERS = ("layla.almansouri", "omar.haddad") # engineering lead, product owner def _answers(*, fail: Sequence[str] = (), no_evidence: Sequence[str] = (), omit: Sequence[str] = ()) -> List[OrrAnswer]: out: List[OrrAnswer] = [] for criterion in ORR_CRITERIA: if criterion.criterion_id in omit: continue satisfied = criterion.criterion_id not in fail evidence = "" if criterion.criterion_id in no_evidence else "link://evidence" out.append(OrrAnswer(criterion.criterion_id, satisfied, evidence)) return out def main() -> None: # pragma: no cover - narrative output print("=" * 78) print("1. THE ORR IS A GATE, NOT A GRADE") print("=" * 78) scorer = OrrScorer() print(f" {len(ORR_CRITERIA)} criteria: " f"{sum(1 for c in ORR_CRITERIA if c.criticality is Criticality.MANDATORY)} " f"mandatory, " f"{sum(1 for c in ORR_CRITERIA if c.criticality is Criticality.ADVISORY)} " f"advisory") print(f" {scorer.score('payments-investigator', _answers()).format()}") print() result = scorer.score("payments-investigator", _answers(fail=["ORR-03"])) print(f" one mandatory failure (runbook not rehearsed):") print(f" {result.format()}") print(f" reason: {result.reasons[0]}") print(" -> 100% of the advisory score and it still fails. The moment a mandatory") print(" criterion can be outweighed, the review becomes a negotiation — and the") print(" thing negotiated away is always the runbook rehearsal, because it is the") print(" most inconvenient and the least visible.") print() result = scorer.score("payments-investigator", _answers(no_evidence=["ORR-02"])) print(f" claimed without evidence (alerts 'tested'):") print(f" {result.reasons[0]}") print(" -> 'yes' with no artifact is a belief. An ORR of unevidenced yeses is a") print(" form somebody filled in.") print() result = scorer.score("payments-investigator", _answers(omit=["ORR-10"])) print(f" a criterion simply left blank (red-team suite):") print(f" {result.reasons[0]}") print(" -> silence is not a pass. Otherwise an ORR is completed by omission.") print() print(" the six rows a generic ORR does not have:") for criterion in ORR_CRITERIA: if criterion.category == "agent" or criterion.criterion_id in ("ORR-12", "ORR-14"): print(f" {criterion.criterion_id} {criterion.question}") print(f" evidence: {criterion.evidence_required}") print() print("=" * 78) print("2. THE ERROR-BUDGET POLICY, SIGNED BEFORE THE FIRST BREACH") print("=" * 78) now = _clock(start=100) policy = ErrorBudgetPolicy(owners=OWNERS, now=now, max_exceptions_per_window=2) print(f" {'budget':<9} {'state':<20} permits") for remaining in (0.80, 0.35, 0.10, 0.0): state = policy.state_for(remaining) permitted = sorted(c.value for c in STATE_POLICY[state]) print(f" {remaining * 100:>6.0f}% {state.value:<20} {permitted}") print(" -> EMERGENCY_FIX and RELIABILITY are permitted in EVERY state including") print(" freeze. A freeze that blocks reliability work is a freeze that extends") print(" itself.") print() print(" shipping decisions at 8% budget remaining:") for change in (ChangeClass.FEATURE, ChangeClass.BUG_FIX, ChangeClass.RELIABILITY, ChangeClass.EXPERIMENT): verdict = policy.may_ship(change, 0.08) mark = "SHIP " if verdict.permitted else "BLOCK" print(f" {mark} {change.value:<16} {verdict.reason}") print() try: policy.grant_exception(ChangeClass.FEATURE, reason="business need", approved_by=[OWNERS[0]]) except PolicyError as exc: print(f" one owner granting an exception: REFUSED — {exc}") try: policy.grant_exception(ChangeClass.FEATURE, reason="urgent", approved_by=list(OWNERS)) except PolicyError as exc: print(f" a one-word reason: REFUSED — {exc}") exception = policy.grant_exception( ChangeClass.FEATURE, reason="regulatory deadline on 31 March requires the reporting agent to ship", approved_by=list(OWNERS), ttl_ticks=5) print(f" both owners, a real reason, expiry at tick {exception.expires_at}: " f"{exception.exception_id}") verdict = policy.may_ship(ChangeClass.FEATURE, 0.08) print(f" ship a feature -> {verdict.permitted} via {verdict.exception_used}") verdict = policy.may_ship(ChangeClass.FEATURE, 0.08) print(f" ship another -> {verdict.permitted} " f"(the exception was consumed)") print() policy.grant_exception( ChangeClass.FEATURE, reason="second genuine exception with a stated business case", approved_by=list(OWNERS)) try: policy.grant_exception( ChangeClass.FEATURE, reason="a third exception this quarter for convenience", approved_by=list(OWNERS)) except PolicyError as exc: print(f" a third exception in the window: REFUSED") print(f" {exc}") print(" -> the exception RATE is the health metric for the policy itself. Two a") print(" quarter is a working policy; ten is a policy replaced by a habit, and") print(" the right response is to renegotiate it rather than keep granting.") print() print("=" * 78) print("3. THE DECISION ROUTER") print("=" * 78) print(f" {'reversible':<12} {'external':<10} {'class':<26} signers") for reversible in (True, False): for external in (False, True): cls = classify_decision(reversible=reversible, externally_visible=external) print(f" {str(reversible):<12} {str(external):<10} {cls.value:<26} " f"{REQUIRED_SIGNERS[cls]}") print(" -> a pair that needs both signatures on everything cannot move; one that") print(" needs neither is two people who will disagree in public later.") print() print("=" * 78) print("4. THE DISAGREEMENT PROTOCOL") print("=" * 78) factual = [ Position(OWNERS[0], "self-host the 70B model", "PTU cost at our volume exceeds the GPU cost", "a cost model showing PTU cheaper at our actual volume"), Position(OWNERS[1], "stay on PTUs", "self-hosting adds an operational burden we cannot staff", "a staffing plan showing we can carry the on-call"), ] values = [ Position(OWNERS[0], "no autonomous payment release, ever", "the residual risk is not one I am willing to carry", ""), Position(OWNERS[1], "autonomous release under 10,000 AED", "the customer experience gain is worth the residual risk", ""), ] unclear = [factual[0], values[1]] for label, positions in (("factual", factual), ("values", values), ("unclear", unclear)): kind = classify_disagreement(positions) print(f" {label:<9} -> {kind.value}") if kind is DisagreementKind.FACTUAL: print(" resolve by MEASURING: build the cost model, run it") elif kind is DisagreementKind.VALUES: print(" escalate BOTH WRITTEN POSITIONS; do not average") else: print(" one position has no falsifier — ask for one first") print(" -> 'what would change your mind?' is the operational test. Most") print(" disagreements that feel like values turn out to be factual once") print(" somebody asks, and it takes a minute rather than a meeting.") print(" -> and for a genuine values disagreement, AVERAGING is the failure mode: a") print(" design at the midpoint of two coherent positions is worse than either.") print() print("=" * 78) print("5. ADRs — IMMUTABLE, SUPERSEDED, WITH THE COSTS NAMED") print("=" * 78) adr_clock = _clock(start=500) store = AdrStore(owners=OWNERS, now=adr_clock) try: store.propose(title="Use Kafka", context="c", options=["Kafka"], decision="Kafka", positive=["p"], negative=["n"], decision_class=DecisionClass.IRREVERSIBLE_INTERNAL) except AdrError as exc: print(f" one option: REFUSED — {exc}") adr = store.propose( title="Self-host the 70B model for restricted-data workloads", context="Residency requires in-region inference; PTU capacity in uaenorth is " "constrained and the queue is 6 weeks.", options=["Azure OpenAI PTU only", "Self-host on AKS GPU pool", "Hybrid: PTU for internal, self-host for restricted"], decision="Hybrid, routing by data classification at the gateway.", positive=["residency is provable per record", "PTU capacity is freed for internal workloads", "the self-hosted path doubles as the exit plan for concentration risk"], negative=["a GPU node pool to operate, and a 9-minute cold start", "two model behaviours to evaluate and two prompt variants to maintain", "on-call now needs GPU expertise the team does not currently have"], decision_class=DecisionClass.IRREVERSIBLE_INTERNAL) print(f" proposed {adr.adr_id}: {adr.title}") try: store.accept(adr.adr_id, signers=[OWNERS[0]]) except AdrError as exc: print(f" one signer on an irreversible decision: REFUSED — {exc}") accepted = store.accept(adr.adr_id, signers=list(OWNERS)) print(f" accepted by both: {accepted.status.value}, digest {accepted.digest()}") print(" negative consequences, which is the section that matters:") for consequence in accepted.negative_consequences: print(f" - {consequence}") try: store.amend(adr.adr_id, decision="actually, PTU only") except AdrError as exc: print(f" editing it: REFUSED — {exc}") successor = store.propose( title="Move restricted-data inference back to PTU", context="PTU capacity in uaenorth became available in Q3; the GPU on-call " "burden proved higher than estimated.", options=["Keep the hybrid", "Return to PTU only", "Self-host everything"], decision="Return to PTU for restricted data; retain the self-hosted path at 5% " "traffic as the tested exit route.", positive=["removes the GPU on-call burden", "keeps a live exit path for concentration risk"], negative=["re-introduces provider concentration for restricted workloads", "the 5% self-hosted path still costs a GPU pool"], decision_class=DecisionClass.IRREVERSIBLE_INTERNAL) store.accept(successor.adr_id, signers=list(OWNERS)) store.supersede(adr.adr_id, by=successor.adr_id) print() print(f" supersession chain:") for link in store.chain(adr.adr_id): print(f" {link.adr_id} [{link.status.value}] {link.title}") print(" -> superseded, never edited. The chain says the decision changed AND why;") print(" an edit would say neither, and would quietly rewrite history.") print() print("=" * 78) print("6. THE STANDING DESIGN-REVIEW CHECKLIST") print("=" * 78) reviewer = DesignReview() thin = DesignDocument( title="Treasury reconciliation agent", author="a-team", side_effecting_tools=("treasury.post_adjustment",), data_classifications=("restricted",), reversible=False) result = reviewer.review(thin) print(f" {result.format()}") for finding in result.findings: print(f" {finding}") print() good = DesignDocument( title="Treasury reconciliation agent (v2)", author="a-team", what_it_denies="posts outside the tolerance band; any adjustment without dual " "control", blast_radius="treasury ledger only; no customer-facing surface", degradation_behaviour="read-only mode: proposes adjustments for human posting", artifacts_emitted=("policy decision", "action record", "approval", "inference record"), operator_at_3am="treasury-platform on-call, runbook RB-114", dependencies=("core banking", "LLM gateway", "action gateway"), side_effecting_tools=("treasury.post_adjustment",), data_classifications=("restricted",), autonomy_band="assisted", slo="99.5% availability, 99% under 5s", retry_policy="irreversible: attempted once, never retried", idempotency="keyed on (period, account, adjustment_ref)", residency="uaenorth only; verified per inference record", reversible=False) result = reviewer.review(good) print(f" {result.format()}") for finding in result.findings: print(f" {finding}") print(" -> the rules are deliberately obvious. The value is that they are STANDING:") print(" the same list every time, so a reviewer having a bad day still catches") print(" the missing blast radius, and an author knows in advance what is asked.") print() print("=" * 78) print("7. POST-MORTEM ACTIONS — THE ONLY HONEST MEASURE") print("=" * 78) tracker = IncidentTracker(now=_clock(start=1000), review_due_ticks=5) inc = tracker.record(Incident("INC-001", "Model gateway saturation", "sev2", 1000, budget_consumed=0.31)) try: tracker.resolve("INC-001") except ValueError as exc: print(f" resolving before mitigating: REFUSED — {exc}") tracker.mitigate("INC-001") tracker.resolve("INC-001") tracker.hold_review("INC-001") print(f" INC-001 mitigated in {tracker._incidents['INC-001'].time_to_mitigate} " f"ticks, resolved in {tracker._incidents['INC-001'].time_to_resolve}") print(" -> mitigation stops the bleeding; the fix removes the cause. Conflating") print(" them is how a mitigated incident is closed and then recurs.") try: tracker.add_action("INC-001", description="improve monitoring", owner="", due_in=10) except ValueError as exc: print(f" an action owned by nobody: REFUSED — {exc}") a1 = tracker.add_action("INC-001", description="add PTU headroom alert", owner="layla.almansouri", due_in=10) a2 = tracker.add_action("INC-001", description="document the degradation ladder", owner="omar.haddad", due_in=10) a3 = tracker.add_action("INC-001", description="rewrite the gateway in Rust", owner="layla.almansouri", due_in=10) tracker.complete(a1.action_id) tracker.drop(a3.action_id, reason="scope creep; the alert addresses the actual cause") for _ in range(20): tracker.now() print() print(f" {tracker.health().format()}") print(f" overdue: {[a.action_id + ' (' + a.owner + ')' for a in tracker.overdue()]}") print(" -> dropped items are excluded from the denominator: they were DECIDED, not") print(" missed. An action quietly left open forever is worse than one dropped") print(" with a reason — the first corrupts the metric, the second is a decision.") print() print("=" * 78) print("8. FIVE FORUMS, FIVE DIFFERENT ANSWERS") print("=" * 78) for forum in FORUMS: print(f" {forum.name}") print(f" wants : {forum.wants}") print(f" bring : {forum.artifact}") print(f" fails when: {forum.fails_when}") print() print(" -> bringing the same deck to all five is the standard mistake, and each") print(" rejects it for a different reason. Cyber does not want the cost model;") print(" the CTTO does not want the threat model; Internal Audit does not want a") print(" description of a control, they want the artifact it emitted.") if __name__ == "__main__": # pragma: no cover main()