"""Reference solution — the evidence engine. Every other phase in this track emits an artifact. This is where those artifacts become **evidence**, and where the difference is made concrete. An examiner does not ask "do you log?" They ask: "On 12 March, an agent initiated a payment of AED 250,000 for customer X. Show me: who authorized it, what the agent was permitted to do at that moment, what data it used to decide, which model version produced the decision, which policy version allowed it, and who reviewed it." A log line answers none of that. **Evidence is a linked set of records** with shared join keys and a tamper-evident chain, generated as a by-product of serving. Anything assembled afterwards is a reconstruction, and an examiner can tell — because a reconstruction has gaps it cannot explain. Three ideas carry the file: * **join keys are the design** — `trace_id` on every artifact, from day one; * **the agent configuration is the model** — so a prompt change is a model change; * **a control that emits no artifact does not exist** as far as audit is concerned. Deterministic: an injected clock, hash-chained records, sorted traversal. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import json from collections import defaultdict, deque 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. Risk tiering # ====================================================================================== class RiskTier(str, Enum): """Impact-based, and the tier drives everything else. Tiering by *impact* rather than by technique is the move that makes model risk tractable for an agentic platform: "is it an LLM?" is not a risk question, and "can it move money?" is. """ TIER_1 = "tier_1" # material financial, customer or regulatory impact TIER_2 = "tier_2" # moderate — informs a human decision TIER_3 = "tier_3" # low — internal productivity @dataclass(frozen=True) class TierPolicy: """What a tier *changes*. A tier that changes nothing is a label. Note ``max_autonomy``: this is the link back to [Phase 10](../../phase-10-action-gateway/index.md)'s autonomy ladder. Risk tier is not a documentation exercise — it is the input that decides whether an agent may act without a human. """ tier: RiskTier requires_independent_validation: bool requires_board_approval: bool revalidation_days: int monitoring_days: int max_autonomy: str # read_only | assisted | bounded | autonomous min_eval_cases: int TIER_POLICIES: Mapping[RiskTier, TierPolicy] = { RiskTier.TIER_1: TierPolicy(RiskTier.TIER_1, True, True, 365, 1, "assisted", 500), RiskTier.TIER_2: TierPolicy(RiskTier.TIER_2, True, False, 730, 7, "bounded", 200), RiskTier.TIER_3: TierPolicy(RiskTier.TIER_3, False, False, 1095, 30, "autonomous", 50), } @dataclass(frozen=True) class ImpactAssessment: """The inputs that decide a tier. Answered by the business, not by engineering.""" max_financial_impact_micros: int affects_customers: bool regulatory_reporting: bool irreversible_actions: bool processes_restricted_data: bool def assign_tier(impact: ImpactAssessment) -> Tuple[RiskTier, List[str]]: """Assign a tier, and say **why** — the reasons are what a validator reviews. A tiering function that returns a bare tier is a tiering function nobody can challenge, and being challengeable is the point of writing it down. """ reasons: List[str] = [] tier = RiskTier.TIER_3 if impact.irreversible_actions: reasons.append("takes irreversible actions") tier = RiskTier.TIER_1 if impact.max_financial_impact_micros >= 100_000_000_000: # 100,000 reasons.append("financial impact at or above 100,000") tier = RiskTier.TIER_1 if impact.regulatory_reporting: reasons.append("feeds regulatory reporting") tier = RiskTier.TIER_1 if tier is not RiskTier.TIER_1: if impact.affects_customers: reasons.append("affects customer outcomes") tier = RiskTier.TIER_2 if impact.processes_restricted_data: reasons.append("processes restricted data") tier = RiskTier.TIER_2 if impact.max_financial_impact_micros > 0: reasons.append("has some financial impact") tier = RiskTier.TIER_2 if not reasons: reasons.append("no material financial, customer or regulatory impact") return tier, sorted(reasons) # ====================================================================================== # 2. The model inventory # ====================================================================================== class ValidationState(str, Enum): NOT_SUBMITTED = "not_submitted" IN_REVIEW = "in_review" APPROVED = "approved" APPROVED_WITH_CONDITIONS = "approved_with_conditions" REJECTED = "rejected" EXPIRED = "expired" @dataclass(frozen=True) class ModelConfiguration: """**The agent configuration IS the model.** This is the phase's central argument and it is contested every time. SR 11-7 was written for a scorecard: weights in, score out, and "the model" is unambiguous. For an agent, the output distribution is determined by the weights *and* the system prompt *and* the retrieval configuration *and* the tool set *and* the guardrails. Change any one and the behaviour changes — often more than a weights change would. So the defensible position is that all of them together are the model, which makes a prompt edit a **model change** with everything that implies: revalidation, approval, a version bump. People resist this because it makes prompt changes expensive. The counter is that a prompt change *is* expensive; the only question is whether the cost is paid before or after an incident. """ config_id: str version: str base_model: str base_model_version: str prompt_version: str retrieval_config_version: str tool_set_version: str guardrail_version: str temperature: float = 0.0 def fingerprint(self) -> str: """Everything that determines the output distribution.""" material = json.dumps({ "base_model": self.base_model, "base_model_version": self.base_model_version, "prompt_version": self.prompt_version, "retrieval_config_version": self.retrieval_config_version, "tool_set_version": self.tool_set_version, "guardrail_version": self.guardrail_version, "temperature": self.temperature, }, sort_keys=True, separators=(",", ":")) return hashlib.sha256(material.encode()).hexdigest()[:16] def differs_from(self, other: "ModelConfiguration") -> List[str]: fields = ("base_model", "base_model_version", "prompt_version", "retrieval_config_version", "tool_set_version", "guardrail_version", "temperature") return sorted(f for f in fields if getattr(self, f) != getattr(other, f)) @dataclass(frozen=True) class InventoryEntry: entry_id: str name: str owner: str # a named human business_sponsor: str # a named human on the business side purpose: str tier: RiskTier tier_reasons: Tuple[str, ...] configuration: ModelConfiguration validation_state: ValidationState validated_at: Optional[int] = None validated_by: Optional[str] = None conditions: Tuple[str, ...] = () autonomy_band: str = "read_only" in_production: bool = False eval_case_count: int = 0 last_monitored_at: Optional[int] = None class InventoryError(Exception): pass class ModelInventory: """The register. "Which models does the bank run?" as a query rather than a survey. Its real value is not the list — it is that **it is the gate**. Promotion goes through it, so a model that is not in the inventory cannot reach production, which is the only mechanism that keeps an inventory current. """ def __init__(self, *, now: Callable[[], int]) -> None: self.now = now self._entries: Dict[str, InventoryEntry] = {} self._history: List[Tuple[int, str, str]] = [] def register(self, entry: InventoryEntry) -> InventoryEntry: if entry.entry_id in self._entries: raise InventoryError(f"{entry.entry_id} is already registered") if not entry.owner or not entry.business_sponsor: raise InventoryError( "every model needs a named technical owner AND a business sponsor") if not entry.purpose: raise InventoryError("a model with no stated purpose cannot be validated") self._entries[entry.entry_id] = entry self._history.append((self.now(), entry.entry_id, "registered")) return entry def get(self, entry_id: str) -> InventoryEntry: try: return self._entries[entry_id] except KeyError: raise InventoryError(f"unknown model: {entry_id}") from None def validate(self, entry_id: str, *, validator: str, state: ValidationState, conditions: Sequence[str] = ()) -> InventoryEntry: """Independent validation. The validator **may not be the owner**. That single check is the whole of "independent". A validation function that reports to the team that built the model is a review, not a validation — and the distinction is what SR 11-7 is mostly about. """ entry = self.get(entry_id) if validator == entry.owner: raise InventoryError( f"{validator} owns this model and cannot validate it; validation must be " f"independent of development") updated = replace(entry, validation_state=state, validated_at=self.now(), validated_by=validator, conditions=tuple(conditions)) self._entries[entry_id] = updated self._history.append((self.now(), entry_id, f"validation:{state.value}")) return updated def promote(self, entry_id: str, *, autonomy_band: str) -> InventoryEntry: """The gate. Every reason to refuse is checked, and all of them are reported. Reporting all of them matters: a team that fixes one blocker at a time and discovers the next on the next attempt learns to resent the gate rather than to plan for it. """ entry = self.get(entry_id) policy = TIER_POLICIES[entry.tier] problems: List[str] = [] if policy.requires_independent_validation and entry.validation_state not in ( ValidationState.APPROVED, ValidationState.APPROVED_WITH_CONDITIONS): problems.append( f"{entry.tier.value} requires independent validation; state is " f"{entry.validation_state.value}") if entry.validation_state is ValidationState.EXPIRED: problems.append("validation has expired") if entry.eval_case_count < policy.min_eval_cases: problems.append( f"{entry.eval_case_count} eval cases; {entry.tier.value} requires " f"{policy.min_eval_cases}") if _autonomy_rank(autonomy_band) > _autonomy_rank(policy.max_autonomy): problems.append( f"autonomy {autonomy_band!r} exceeds the {entry.tier.value} maximum of " f"{policy.max_autonomy!r}") if problems: raise InventoryError(f"{entry_id} cannot be promoted: " + "; ".join(problems)) updated = replace(entry, in_production=True, autonomy_band=autonomy_band) self._entries[entry_id] = updated self._history.append((self.now(), entry_id, f"promoted:{autonomy_band}")) return updated def record_change(self, entry_id: str, configuration: ModelConfiguration) -> Tuple[InventoryEntry, List[str]]: """A configuration change **invalidates the validation**. Which is the operational consequence of "the agent configuration is the model". Edit the prompt, and the model is no longer the one that was validated — so it drops out of production until it is validated again. """ entry = self.get(entry_id) changed = configuration.differs_from(entry.configuration) if not changed: return entry, [] updated = replace(entry, configuration=configuration, validation_state=ValidationState.NOT_SUBMITTED, validated_at=None, validated_by=None, in_production=False) self._entries[entry_id] = updated self._history.append( (self.now(), entry_id, f"config-change:{','.join(changed)}")) return updated, changed def due_for_revalidation(self) -> List[InventoryEntry]: out: List[InventoryEntry] = [] for entry in self._entries.values(): if entry.validated_at is None: continue policy = TIER_POLICIES[entry.tier] if self.now() - entry.validated_at >= policy.revalidation_days: out.append(entry) return sorted(out, key=lambda e: e.entry_id) def in_production(self) -> List[InventoryEntry]: return sorted((e for e in self._entries.values() if e.in_production), key=lambda e: e.entry_id) def all(self) -> List[InventoryEntry]: return sorted(self._entries.values(), key=lambda e: e.entry_id) def history(self) -> List[Tuple[int, str, str]]: return list(self._history) _AUTONOMY_ORDER = ("read_only", "assisted", "bounded", "autonomous") def _autonomy_rank(band: str) -> int: try: return _AUTONOMY_ORDER.index(band) except ValueError: raise InventoryError(f"unknown autonomy band: {band!r}") from None # ====================================================================================== # 3. The lineage graph # ====================================================================================== class ArtifactKind(str, Enum): """One per emitting layer. The set IS the evidence pack's contents.""" SESSION = "session" # channel: who asked, from where POLICY_DECISION = "policy_decision" # control plane: allowed, under which version EXECUTION_STEP = "execution_step" # kernel: what the agent did RETRIEVAL = "retrieval" # knowledge: which documents, which versions INFERENCE = "inference" # model layer: which model, tokens, region TOOL_CALL = "tool_call" # action gateway: what happened downstream GUARDRAIL = "guardrail" # guardrails: what was blocked or masked APPROVAL = "approval" # HITL: who approved ACTION = "action" # the effect on the bank DOCUMENT = "document" # a source document, at a version @dataclass(frozen=True) class Artifact: """One evidence record. ``trace_id`` is the **join key**, and it is the single most important field in this file. Every layer emits its own artifact into its own store — the policy decision to the audit log, the inference record to the gateway's ledger, the retrieval to the search service — and the only thing that makes them one story is a shared id present on all of them from the first line of code. Retrofitting a join key means the first N months of evidence cannot be assembled, ever. """ artifact_id: str kind: ArtifactKind trace_id: str tick: int emitted_by: str # the component derived_from: Tuple[str, ...] = () # artifact ids attributes: Mapping[str, Any] = field(default_factory=dict) def digest(self) -> str: material = json.dumps({ "artifact_id": self.artifact_id, "kind": self.kind.value, "trace_id": self.trace_id, "tick": self.tick, "emitted_by": self.emitted_by, "derived_from": sorted(self.derived_from), "attributes": self.attributes, }, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(material.encode()).hexdigest() class LineageError(Exception): pass class LineageGraph: """The graph from an output back to every input. A DAG, and the acyclicity matters: a cycle would mean an output that is its own ancestor, which is either a bug in the emitters or a genuine feedback loop nobody intended. Either way it is a finding. """ def __init__(self) -> None: self._artifacts: Dict[str, Artifact] = {} self._by_trace: Dict[str, List[str]] = defaultdict(list) def add(self, artifact: Artifact) -> Artifact: if artifact.artifact_id in self._artifacts: raise LineageError(f"duplicate artifact: {artifact.artifact_id}") for parent in artifact.derived_from: if parent not in self._artifacts: raise LineageError( f"{artifact.artifact_id} derives from {parent}, which does not " f"exist — evidence must be emitted in causal order") self._artifacts[artifact.artifact_id] = artifact self._by_trace[artifact.trace_id].append(artifact.artifact_id) return artifact def get(self, artifact_id: str) -> Artifact: try: return self._artifacts[artifact_id] except KeyError: raise LineageError(f"unknown artifact: {artifact_id}") from None def for_trace(self, trace_id: str) -> List[Artifact]: return sorted((self._artifacts[a] for a in self._by_trace.get(trace_id, [])), key=lambda a: (a.tick, a.artifact_id)) def ancestors(self, artifact_id: str) -> List[Artifact]: """**Walk backwards** — everything that contributed to this output. This is the examiner's actual question. "What did the agent use to decide?" is a reachability query over this graph, and it is answerable only because every emitter recorded what it derived from. """ self.get(artifact_id) seen: Set[str] = set() queue = deque([artifact_id]) while queue: current = queue.popleft() for parent in self._artifacts[current].derived_from: if parent not in seen: seen.add(parent) queue.append(parent) return sorted((self._artifacts[a] for a in seen), key=lambda a: (a.tick, a.artifact_id)) def descendants(self, artifact_id: str) -> List[Artifact]: """Forward — **the impact query**. "This document was wrong; what did it affect?" The direction people forget to build, and the one asked during a remediation. """ self.get(artifact_id) children: Dict[str, List[str]] = defaultdict(list) for artifact in self._artifacts.values(): for parent in artifact.derived_from: children[parent].append(artifact.artifact_id) seen: Set[str] = set() queue = deque(children.get(artifact_id, [])) while queue: current = queue.popleft() if current in seen: continue seen.add(current) queue.extend(children.get(current, [])) return sorted((self._artifacts[a] for a in seen), key=lambda a: (a.tick, a.artifact_id)) def of_kind(self, trace_id: str, kind: ArtifactKind) -> List[Artifact]: return [a for a in self.for_trace(trace_id) if a.kind is kind] def check_acyclic(self) -> None: colour: Dict[str, int] = defaultdict(int) # 0 white, 1 grey, 2 black def visit(node: str, path: List[str]) -> None: if colour[node] == 1: cycle = path[path.index(node):] + [node] raise LineageError(f"cycle in lineage: {' -> '.join(cycle)}") if colour[node] == 2: return colour[node] = 1 for parent in sorted(self._artifacts[node].derived_from): visit(parent, path + [node]) colour[node] = 2 for artifact_id in sorted(self._artifacts): visit(artifact_id, []) def orphans(self, trace_id: str) -> List[Artifact]: """Artifacts nothing derives from and which derive from nothing. Usually an emitter that forgot to record its inputs, which means part of the story is disconnected — and a disconnected artifact is worse than a missing one, because it looks like evidence. """ artifacts = self.for_trace(trace_id) referenced: Set[str] = set() for artifact in artifacts: referenced.update(artifact.derived_from) return [a for a in artifacts if not a.derived_from and a.artifact_id not in referenced and a.kind is not ArtifactKind.SESSION] # ====================================================================================== # 4. Residency # ====================================================================================== @dataclass(frozen=True) class ResidencyRule: """Per data classification, which regions may process it.""" classification: str permitted_regions: FrozenSet[str] permit_transit: bool = False # may it *traverse* another region? @dataclass(frozen=True) class ResidencyViolation: trace_id: str artifact_id: str classification: str region: str permitted: Tuple[str, ...] detail: str def __str__(self) -> str: return (f"{self.artifact_id}: {self.classification} processed in {self.region}; " f"permitted {list(self.permitted)} — {self.detail}") class ResidencyChecker: """Verify, **per inference record**, that processing stayed in-jurisdiction. Note what this is not: a configuration check. It reads the *actual* records of what happened, which is the only thing that survives an examiner asking "show me". A configuration says what should have happened; an inference record says what did. This is the evidence half of [Phase 13](../../phase-13-cloud-infrastructure-backbone/index.md)'s reachability proof: that phase proves no path *can* leave the region; this one proves none *did*. """ def __init__(self, rules: Sequence[ResidencyRule]) -> None: self.rules = {r.classification: r for r in rules} def rule_for(self, classification: str) -> ResidencyRule: try: return self.rules[classification] except KeyError: raise LineageError( f"no residency rule for classification {classification!r}; an " f"unclassified data flow is a finding, not a default") from None def check_trace(self, graph: LineageGraph, trace_id: str) -> List[ResidencyViolation]: violations: List[ResidencyViolation] = [] for artifact in graph.for_trace(trace_id): if artifact.kind not in (ArtifactKind.INFERENCE, ArtifactKind.RETRIEVAL, ArtifactKind.TOOL_CALL): continue classification = artifact.attributes.get("data_classification") region = artifact.attributes.get("region") if classification is None or region is None: violations.append(ResidencyViolation( trace_id, artifact.artifact_id, str(classification), str(region), (), "the record does not state its classification or region — " "unprovable is a violation")) continue rule = self.rule_for(classification) if region not in rule.permitted_regions: violations.append(ResidencyViolation( trace_id, artifact.artifact_id, classification, region, tuple(sorted(rule.permitted_regions)), "processed outside the permitted regions")) continue transit = artifact.attributes.get("transit_regions", ()) if transit and not rule.permit_transit: outside = sorted(set(transit) - rule.permitted_regions) if outside: violations.append(ResidencyViolation( trace_id, artifact.artifact_id, classification, region, tuple(sorted(rule.permitted_regions)), f"transited {outside}, which the rule does not permit")) return violations # ====================================================================================== # 5. Reproducibility # ====================================================================================== #: The pins required to re-derive a past decision. The one people forget is #: ``retrieval_snapshot`` — the corpus changes continuously, so without a snapshot id the #: same query returns different documents tomorrow and the decision cannot be reproduced #: even with every other pin in place. REQUIRED_PINS: Tuple[str, ...] = ( "base_model_version", "prompt_version", "retrieval_snapshot", "policy_version", "tool_set_version", "guardrail_version", ) @dataclass(frozen=True) class ReproducibilityReport: trace_id: str reproducible: bool present: Tuple[str, ...] missing: Tuple[str, ...] caveats: Tuple[str, ...] def format(self) -> str: if self.reproducible: base = f"{self.trace_id}: reproducible" return base + (f" (caveats: {'; '.join(self.caveats)})" if self.caveats else "") return f"{self.trace_id}: NOT reproducible — missing {list(self.missing)}" def check_reproducibility(graph: LineageGraph, trace_id: str) -> ReproducibilityReport: """Can this decision be re-derived? And the honest caveat that must be stated rather than hidden: **even with every pin, a temperature above zero means the output is not bit-reproducible.** What is reproducible is the *decision context* — the same inputs, the same permitted behaviour — which is what an examiner actually needs. Claiming bit-reproducibility for a sampled model is a claim that will be tested. """ pins: Dict[str, Any] = {} caveats: List[str] = [] for artifact in graph.for_trace(trace_id): for pin in REQUIRED_PINS: if pin in artifact.attributes and artifact.attributes[pin]: pins[pin] = artifact.attributes[pin] temperature = artifact.attributes.get("temperature") if temperature is not None and temperature > 0: caveats.append( f"temperature {temperature} — the decision context is reproducible, the " f"exact output is not") if artifact.attributes.get("provider_managed_version"): caveats.append( "the provider manages this version; a silent update cannot be excluded") missing = tuple(p for p in REQUIRED_PINS if p not in pins) return ReproducibilityReport(trace_id, not missing, tuple(sorted(pins)), missing, tuple(sorted(set(caveats)))) # ====================================================================================== # 6. Third-party governance # ====================================================================================== class ExitReadiness(str, Enum): NONE = "none" # no alternative identified IDENTIFIED = "identified" # an alternative exists on paper TESTED = "tested" # the alternative has been exercised LIVE = "live" # the alternative carries production traffic @dataclass(frozen=True) class ThirdPartyModel: """A provider register entry. ``exit_readiness`` is the field that matters, and the reason is the one line in this section worth remembering: **an exit plan that has never been executed is a document.** A regulator asking about concentration risk is asking whether you *can* move, and the only convincing answer is that some traffic already runs on the alternative. """ provider: str model: str version: str data_use_terms: str # e.g. "no training on customer data" sub_processors: Tuple[str, ...] regions: Tuple[str, ...] deprecation_notice_days: int contractual_sla: Optional[float] exit_readiness: ExitReadiness alternative: Optional[str] = None last_exit_test_tick: Optional[int] = None traffic_share: float = 0.0 @dataclass(frozen=True) class ConcentrationFinding: severity: str # "high" | "medium" | "low" detail: str class ThirdPartyRegister: def __init__(self, *, now: Callable[[], int], exit_test_interval_days: int = 180, concentration_threshold: float = 0.8) -> None: self.now = now self.exit_test_interval_days = exit_test_interval_days self.concentration_threshold = concentration_threshold self._entries: Dict[str, ThirdPartyModel] = {} def register(self, entry: ThirdPartyModel) -> ThirdPartyModel: key = f"{entry.provider}/{entry.model}" if not entry.data_use_terms: raise InventoryError( f"{key}: data-use terms are mandatory — 'do they train on our data?' is " f"the first question asked") self._entries[key] = entry return entry def all(self) -> List[ThirdPartyModel]: return sorted(self._entries.values(), key=lambda e: (e.provider, e.model)) def assess_concentration(self) -> List[ConcentrationFinding]: """Concentration risk is answered with an **architecture**, not a paragraph.""" findings: List[ConcentrationFinding] = [] by_provider: Dict[str, float] = defaultdict(float) for entry in self._entries.values(): by_provider[entry.provider] += entry.traffic_share for provider, share in sorted(by_provider.items()): if share >= self.concentration_threshold: findings.append(ConcentrationFinding( "high", f"{provider} carries {share * 100:.0f}% of traffic, at or above the " f"{self.concentration_threshold * 100:.0f}% threshold")) for entry in self.all(): key = f"{entry.provider}/{entry.model}" if entry.exit_readiness in (ExitReadiness.NONE, ExitReadiness.IDENTIFIED): findings.append(ConcentrationFinding( "high" if entry.traffic_share > 0.2 else "medium", f"{key}: exit readiness is {entry.exit_readiness.value} — an exit " f"plan that has never been executed is a document")) elif entry.exit_readiness is ExitReadiness.TESTED: last = entry.last_exit_test_tick if last is None or self.now() - last >= self.exit_test_interval_days: findings.append(ConcentrationFinding( "medium", f"{key}: the exit path was last tested " f"{'never' if last is None else self.now() - last}" f"{'' if last is None else ' days ago'}; the interval is " f"{self.exit_test_interval_days}")) if entry.deprecation_notice_days < 90: findings.append(ConcentrationFinding( "medium", f"{key}: {entry.deprecation_notice_days} days of deprecation notice " f"is less than a validation cycle")) if not entry.sub_processors: findings.append(ConcentrationFinding( "low", f"{key}: no sub-processors declared — verify rather than assume")) return sorted(findings, key=lambda f: ( {"high": 0, "medium": 1, "low": 2}[f.severity], f.detail)) # ====================================================================================== # 7. The evidence pack # ====================================================================================== #: What the examiner's question decomposes into. Each entry names the artifact kind and #: the question it answers — which is the discipline: **if you cannot name whose question #: an artifact answers, it does not belong in the pack.** REQUIRED_ARTIFACTS: Mapping[ArtifactKind, str] = { ArtifactKind.SESSION: "who asked, from where, authenticated how", ArtifactKind.POLICY_DECISION: "what the agent was permitted to do, under which policy", ArtifactKind.EXECUTION_STEP: "what the agent did, step by step", ArtifactKind.RETRIEVAL: "what data it used to decide", ArtifactKind.INFERENCE: "which model version produced the decision", ArtifactKind.ACTION: "what happened to the bank", } #: Required only when the action was above the dual-control threshold. CONDITIONAL_ARTIFACTS: Mapping[ArtifactKind, str] = { ArtifactKind.APPROVAL: "who reviewed and approved it", } @dataclass(frozen=True) class EvidencePack: trace_id: str generated_at: int complete: bool artifacts: Tuple[Artifact, ...] missing: Tuple[str, ...] reproducibility: ReproducibilityReport residency_violations: Tuple[ResidencyViolation, ...] chain_head: str signature: str def summary(self) -> str: status = "COMPLETE" if self.complete else "INCOMPLETE" return (f"{self.trace_id}: {status}, {len(self.artifacts)} artifacts, " f"reproducible={self.reproducibility.reproducible}, " f"residency violations={len(self.residency_violations)}") class EvidencePackError(Exception): pass class EvidenceGenerator: """Generate the pack. **Generated, not assembled.** The distinction is the phase. A generator reads records that were emitted as a by-product of serving and links them by a join key. An *assembler* goes looking for what it can find, and its output has gaps it cannot explain — which is exactly what an examiner probes. So a missing artifact is a **failure with the artifact named**, not a pack with a hole in it. Failing loudly is what makes the gap a fixable engineering problem rather than an audit finding. """ def __init__(self, *, graph: LineageGraph, inventory: ModelInventory, residency: ResidencyChecker, now: Callable[[], int], secret: bytes = b"evidence-signing-key", dual_control_threshold_micros: int = 100_000_000_000) -> None: self.graph = graph self.inventory = inventory self.residency = residency self.now = now self.secret = secret self.dual_control_threshold_micros = dual_control_threshold_micros def generate(self, trace_id: str, *, strict: bool = True) -> EvidencePack: artifacts = self.graph.for_trace(trace_id) if not artifacts: raise EvidencePackError(f"no artifacts for trace {trace_id}") present = {a.kind for a in artifacts} missing = [f"{kind.value} ({question})" for kind, question in REQUIRED_ARTIFACTS.items() if kind not in present] # Conditional: an action at or above the threshold needs an approval record. actions = [a for a in artifacts if a.kind is ArtifactKind.ACTION] needs_approval = any( int(a.attributes.get("value_micros", 0)) >= self.dual_control_threshold_micros for a in actions) if needs_approval and ArtifactKind.APPROVAL not in present: missing.append( f"approval ({CONDITIONAL_ARTIFACTS[ArtifactKind.APPROVAL]}) — required " f"because the action was at or above the dual-control threshold") self.graph.check_acyclic() reproducibility = check_reproducibility(self.graph, trace_id) violations = tuple(self.residency.check_trace(self.graph, trace_id)) chain_head = self._chain(artifacts) pack = EvidencePack( trace_id=trace_id, generated_at=self.now(), complete=not missing, artifacts=tuple(artifacts), missing=tuple(sorted(missing)), reproducibility=reproducibility, residency_violations=violations, chain_head=chain_head, signature="") pack = replace(pack, signature=self._sign(pack)) if strict and missing: raise EvidencePackError( f"cannot produce a complete pack for {trace_id}; missing: " + "; ".join(sorted(missing))) return pack def _chain(self, artifacts: Sequence[Artifact]) -> str: """Hash-chain the artifacts in causal order — Phase 10's mechanism, applied to the whole pack rather than one log.""" head = "0" * 64 for artifact in artifacts: head = hashlib.sha256((head + artifact.digest()).encode()).hexdigest() return head def _sign(self, pack: EvidencePack) -> str: material = json.dumps({ "trace_id": pack.trace_id, "generated_at": pack.generated_at, "complete": pack.complete, "chain_head": pack.chain_head, "missing": list(pack.missing), }, sort_keys=True, separators=(",", ":")) return hashlib.blake2b(self.secret + material.encode(), digest_size=16).hexdigest() def verify(self, pack: EvidencePack) -> Tuple[bool, Optional[str]]: if self._chain(pack.artifacts) != pack.chain_head: return False, "the artifact chain does not match the recorded head" unsigned = replace(pack, signature="") if self._sign(unsigned) != pack.signature: return False, "the pack signature does not match its contents" return True, None # ====================================================================================== # 8. Control-to-evidence mapping # ====================================================================================== @dataclass(frozen=True) class Control: """A control, the component that implements it, and **the artifact it emits**. That third column is the phase's operating principle: *a control that emits no artifact does not exist as far as audit is concerned.* A control with no evidence cannot be shown to have run, which for an examiner is indistinguishable from not having run. """ control_id: str name: str component: str phase: str emits: Optional[ArtifactKind] frameworks: Tuple[str, ...] # e.g. ("SR 11-7", "CBUAE", "LLM06") CONTROL_CATALOGUE: Tuple[Control, ...] = ( Control("C-01", "Agent identity and delegation chain", "identity fabric", "Phase 08", ArtifactKind.SESSION, ("CBUAE", "SR 11-7")), Control("C-02", "Policy-gated execution", "control plane", "Phase 09", ArtifactKind.POLICY_DECISION, ("CBUAE", "LLM06")), Control("C-03", "Know Your Agent posture", "control plane", "Phase 09", ArtifactKind.POLICY_DECISION, ("SR 11-7",)), Control("C-04", "Contract enforcement and idempotency", "action gateway", "Phase 10", ArtifactKind.ACTION, ("CBUAE",)), Control("C-05", "Dual control above threshold", "action gateway", "Phase 10", ArtifactKind.APPROVAL, ("CBUAE", "LLM06")), Control("C-06", "Hash-chained audit log", "action gateway", "Phase 10", ArtifactKind.ACTION, ("CBUAE",)), Control("C-07", "Injection containment (taint rule)", "guardrails", "Phase 11", ArtifactKind.GUARDRAIL, ("LLM01", "LLM06")), Control("C-08", "PII/MNPI detection and masking", "guardrails", "Phase 11", ArtifactKind.GUARDRAIL, ("LLM02", "CBUAE")), Control("C-09", "Information-barrier retrieval filter", "retrieval", "Phase 11", ArtifactKind.RETRIEVAL, ("CBUAE", "LLM08")), Control("C-10", "Egress allow-listing", "network", "Phase 11/13", None, ("LLM02",)), Control("C-11", "Residency-aware routing", "LLM gateway", "Phase 04", ArtifactKind.INFERENCE, ("CBUAE",)), Control("C-12", "Grounding and citation", "retrieval", "Phase 06", ArtifactKind.RETRIEVAL, ("LLM09",)), Control("C-13", "Model inventory and risk tiering", "governance", "Phase 15", None, ("SR 11-7",)), Control("C-14", "Independent validation", "governance", "Phase 15", None, ("SR 11-7",)), Control("C-15", "Evaluation pipeline as a gate", "control plane", "Phase 09", None, ("SR 11-7", "LLM04")), Control("C-16", "Trace and lineage", "observability", "Phase 14", ArtifactKind.EXECUTION_STEP, ("CBUAE", "SR 11-7")), Control("C-17", "Cost and quota enforcement", "LLM gateway", "Phase 04", None, ("LLM10",)), Control("C-18", "Third-party model governance", "governance", "Phase 15", None, ("CBUAE", "LLM03")), ) @dataclass(frozen=True) class CoverageRow: framework: str controls: Tuple[str, ...] evidence_emitting: Tuple[str, ...] silent: Tuple[str, ...] # controls that emit nothing def control_coverage(controls: Sequence[Control] = CONTROL_CATALOGUE ) -> List[CoverageRow]: """Generated from the catalogue, and it surfaces the **silent controls**. A silent control is not necessarily wrong — egress allow-listing is enforced by the network and is genuinely hard to attach to a trace. But it must be *named*, because at audit time "we allow-list egress" needs some other form of evidence, and knowing which controls those are is the difference between a prepared answer and a scramble. """ by_framework: Dict[str, List[Control]] = defaultdict(list) for control in controls: for framework in control.frameworks: by_framework[framework].append(control) rows: List[CoverageRow] = [] for framework in sorted(by_framework): members = sorted(by_framework[framework], key=lambda c: c.control_id) rows.append(CoverageRow( framework, tuple(c.control_id for c in members), tuple(c.control_id for c in members if c.emits is not None), tuple(c.control_id for c in members if c.emits is None))) return rows def verify_control_evidence(graph: LineageGraph, trace_id: str, controls: Sequence[Control] = CONTROL_CATALOGUE ) -> List[str]: """Which evidence-emitting controls left no artifact on this trace? The check that turns the catalogue from a document into a test. """ present = {a.kind for a in graph.for_trace(trace_id)} return sorted(f"{c.control_id} ({c.name}) emits {c.emits.value}, which is absent" for c in controls if c.emits is not None and c.emits not in present) # ====================================================================================== # Worked example # ====================================================================================== def _clock(start: int = 0, step: int = 1) -> Callable[[], int]: state = {"t": start - step} def now() -> int: state["t"] += step return state["t"] return now def build_trace(graph: LineageGraph, trace_id: str, *, with_approval: bool = True, region: str = "uaenorth", pin_retrieval: bool = True) -> None: """Emit the artifacts a real run would, in causal order.""" def art(aid: str, kind: ArtifactKind, tick: int, emitter: str, derived: Sequence[str] = (), **attrs: Any) -> None: graph.add(Artifact(f"{trace_id}:{aid}", kind, trace_id, tick, emitter, tuple(f"{trace_id}:{d}" for d in derived), attrs)) art("sess", ArtifactKind.SESSION, 1, "channel", user="layla.almansouri", channel="teams", auth="entra-oidc", agent="payments-investigator", delegation_chain="layla.almansouri -> orchestrator -> payments-investigator") art("pol", ArtifactKind.POLICY_DECISION, 2, "control-plane", ["sess"], effect="allow", rule="allow-release-with-approvals", policy_version="2026-03-11.4", kya_posture="ok") art("doc1", ArtifactKind.DOCUMENT, 3, "knowledge", doc_id="case-note-991", version="v3", classification="confidential") art("doc2", ArtifactKind.DOCUMENT, 3, "knowledge", doc_id="beneficiary-registry", version="2026-03-10", classification="internal") retrieval_attrs = dict(query="why is PMT-771 held", k=8, returned=2, data_classification="confidential", region=region) if pin_retrieval: retrieval_attrs["retrieval_snapshot"] = "idx-2026-03-11T06:00Z" art("ret", ArtifactKind.RETRIEVAL, 4, "retrieval", ["pol", "doc1", "doc2"], **retrieval_attrs) art("guard", ArtifactKind.GUARDRAIL, 5, "guardrails", ["ret"], verdict="mask", masked=1, guardrail_version="gr-2026-02") art("inf", ArtifactKind.INFERENCE, 6, "llm-gateway", ["guard"], base_model_version="gpt-frontier-2026-02-11", prompt_version="pi-v7", tool_set_version="ts-v3", guardrail_version="gr-2026-02", policy_version="2026-03-11.4", temperature=0.0, input_tokens=4812, output_tokens=380, cost_micros=3900, data_classification="confidential", region=region, deployment="azure-openai-uaenorth") art("step", ArtifactKind.EXECUTION_STEP, 7, "agent-kernel", ["inf"], step=1, action="propose payments.release", rationale="beneficiary verified") if with_approval: art("appr", ArtifactKind.APPROVAL, 8, "hitl", ["step"], approvers="ahmed.k,sara.m", rationale="verified against the registry", evidence="case-note-991 v3") art("act", ArtifactKind.ACTION, 9, "action-gateway", ["step"] + (["appr"] if with_approval else []), tool="payments.release", payment_id="PMT-771", value_micros=250_000_000_000, idempotency_key="idem-771", outcome="success", reference="REF-0091", data_classification="confidential", region=region) def main() -> None: # pragma: no cover - narrative output print("=" * 78) print("1. RISK TIERING — IMPACT, NOT TECHNIQUE") print("=" * 78) assessments = [ ("payments-investigator", ImpactAssessment(250_000_000_000, True, False, True, True)), ("credit-memo-drafter", ImpactAssessment(0, True, False, False, True)), ("meeting-summariser", ImpactAssessment(0, False, False, False, False)), ("regulatory-report-assembler", ImpactAssessment(0, False, True, False, True)), ] for name, impact in assessments: tier, reasons = assign_tier(impact) policy = TIER_POLICIES[tier] print(f" {name:<30} {tier.value}") print(f" because: {'; '.join(reasons)}") print(f" -> validation={'independent' if policy.requires_independent_validation else 'self'}" f" board={policy.requires_board_approval}" f" max autonomy={policy.max_autonomy}" f" evals>={policy.min_eval_cases}") print(" -> tiering by IMPACT, not by technique. 'Is it an LLM?' is not a risk") print(" question; 'can it move money?' is. And note the tier sets the AUTONOMY") print(" BAND — the register is wired to Phase 10, not filed.") print() print("=" * 78) print("2. THE AGENT CONFIGURATION IS THE MODEL") print("=" * 78) now = _clock(start=1000) inventory = ModelInventory(now=now) config = ModelConfiguration( "cfg-payments", "v1", "gpt-frontier", "gpt-frontier-2026-02-11", "pi-v7", "rc-v2", "ts-v3", "gr-2026-02", temperature=0.0) tier, reasons = assign_tier(assessments[0][1]) entry = inventory.register(InventoryEntry( "M-001", "Payments Investigator", owner="layla.almansouri", business_sponsor="head-of-payments-ops", purpose="Investigate held payments and propose release", tier=tier, tier_reasons=tuple(reasons), configuration=config, validation_state=ValidationState.NOT_SUBMITTED, eval_case_count=620)) print(f" registered M-001, fingerprint {config.fingerprint()}") try: inventory.validate("M-001", validator="layla.almansouri", state=ValidationState.APPROVED) except InventoryError as exc: print(f" owner validating: REFUSED — {exc}") inventory.validate("M-001", validator="model-risk.omar", state=ValidationState.APPROVED_WITH_CONDITIONS, conditions=("monthly monitoring", "no autonomy above assisted")) print(f" independent validation by model-risk.omar: " f"{inventory.get('M-001').validation_state.value}") inventory.promote("M-001", autonomy_band="assisted") print(f" promoted: in_production={inventory.get('M-001').in_production}") try: inventory.promote("M-001", autonomy_band="autonomous") except InventoryError as exc: print(f" promoting to autonomous: REFUSED — {exc}") print() print(" now edit the PROMPT — nothing about the weights changes:") updated, changed = inventory.record_change( "M-001", replace(config, prompt_version="pi-v8")) print(f" changed: {changed}") print(f" validation_state -> {updated.validation_state.value}, " f"in_production -> {updated.in_production}") print(" -> the output distribution is determined by the weights AND the prompt AND") print(" retrieval AND tools AND guardrails. So a prompt edit is a MODEL CHANGE,") print(" and it drops out of production until revalidated. People resist this") print(" because it makes prompt changes expensive. A prompt change IS expensive;") print(" the only question is whether the cost is paid before or after.") print() print("=" * 78) print("3. THE LINEAGE GRAPH — JOIN KEYS ARE THE DESIGN") print("=" * 78) graph = LineageGraph() build_trace(graph, "trace-payment-771") print(f" {'artifact':<34} {'kind':<16} {'emitted by':<16} derived from") for artifact in graph.for_trace("trace-payment-771"): short = artifact.artifact_id.split(":")[-1] parents = [p.split(":")[-1] for p in artifact.derived_from] print(f" {short:<34} {artifact.kind.value:<16} {artifact.emitted_by:<16} " f"{parents or '-'}") graph.check_acyclic() print(f" acyclic: yes") print() ancestors = graph.ancestors("trace-payment-771:act") print(f" what contributed to the ACTION ({len(ancestors)} artifacts):") for artifact in ancestors: print(f" {artifact.artifact_id.split(':')[-1]:<8} {artifact.kind.value}") print(" -> that walk IS the examiner's question. It is answerable only because every") print(" emitter recorded what it derived from, and because one join key —") print(" trace_id — is on all of them.") print() impacted = graph.descendants("trace-payment-771:doc1") print(f" and FORWARD — 'case-note-991 was wrong; what did it affect?':") print(f" {[a.artifact_id.split(':')[-1] for a in impacted]}") print(" -> the direction people forget to build, and the one asked during a") print(" remediation.") print() print("=" * 78) print("4. RESIDENCY — PROVED PER RECORD, NOT ASSERTED") print("=" * 78) checker = ResidencyChecker([ ResidencyRule("restricted", frozenset({"uaenorth"})), ResidencyRule("confidential", frozenset({"uaenorth", "uaecentral"})), ResidencyRule("internal", frozenset({"uaenorth", "uaecentral", "westeurope"})), ]) print(f" compliant trace: {checker.check_trace(graph, 'trace-payment-771') or 'no violations'}") offshore = LineageGraph() build_trace(offshore, "trace-offshore", region="westeurope") print(" the same run, routed to westeurope:") for violation in checker.check_trace(offshore, "trace-offshore"): print(f" {violation}") print(" -> this reads the RECORDS of what happened, not the configuration. Phase 13") print(" proves no path CAN leave the region; this proves none DID. An examiner") print(" asks for the second.") print() print("=" * 78) print("5. REPRODUCIBILITY — AND THE PIN EVERYONE FORGETS") print("=" * 78) report = check_reproducibility(graph, "trace-payment-771") print(f" {report.format()}") print(f" pins present: {list(report.present)}") unpinned = LineageGraph() build_trace(unpinned, "trace-unpinned", pin_retrieval=False) print(f" {check_reproducibility(unpinned, 'trace-unpinned').format()}") print(" -> the retrieval snapshot. The corpus changes continuously, so without it") print(" the same query returns different documents tomorrow and the decision") print(" cannot be reproduced even with every other pin in place.") sampled = LineageGraph() build_trace(sampled, "trace-sampled") sampled.add(Artifact("trace-sampled:inf2", ArtifactKind.INFERENCE, "trace-sampled", 10, "llm-gateway", ("trace-sampled:inf",), {"temperature": 0.7, "provider_managed_version": True})) caveated = check_reproducibility(sampled, "trace-sampled") print(f" a sampled run: reproducible={caveated.reproducible}") for caveat in caveated.caveats: print(f" caveat: {caveat}") print(" -> state the caveat rather than hiding it. Claiming bit-reproducibility for") print(" a sampled model is a claim that will be tested.") print() print("=" * 78) print("6. THIRD-PARTY GOVERNANCE AND CONCENTRATION RISK") print("=" * 78) register = ThirdPartyRegister(now=_clock(start=400), exit_test_interval_days=180) register.register(ThirdPartyModel( "azure-openai", "gpt-frontier", "2026-02-11", data_use_terms="no training on customer data; 30-day abuse retention, opt-out", sub_processors=("microsoft",), regions=("uaenorth",), deprecation_notice_days=180, contractual_sla=0.999, exit_readiness=ExitReadiness.LIVE, alternative="self-hosted-llama", last_exit_test_tick=350, traffic_share=0.85)) register.register(ThirdPartyModel( "anthropic-via-bedrock", "claude", "2026-01", data_use_terms="no training on customer data", sub_processors=("aws",), regions=("uaenorth",), deprecation_notice_days=60, contractual_sla=None, exit_readiness=ExitReadiness.IDENTIFIED, traffic_share=0.10)) register.register(ThirdPartyModel( "self-hosted", "llama-70b", "3.3", data_use_terms="n/a — runs in our tenancy", sub_processors=(), regions=("uaenorth",), deprecation_notice_days=3650, contractual_sla=None, exit_readiness=ExitReadiness.LIVE, traffic_share=0.05)) for entry in register.all(): name = f"{entry.provider}/{entry.model}" print(f" {name:<32} {entry.traffic_share * 100:>5.0f}% " f"exit={entry.exit_readiness.value:<11} " f"notice={entry.deprecation_notice_days}d") print() for finding in register.assess_concentration(): print(f" [{finding.severity.upper():<6}] {finding.detail}") print(" -> an exit plan that has never been executed is a document. The only") print(" convincing answer to concentration risk is that some traffic ALREADY") print(" runs on the alternative.") print() print("=" * 78) print("7. THE EVIDENCE PACK — GENERATED, NOT ASSEMBLED") print("=" * 78) generator = EvidenceGenerator(graph=graph, inventory=inventory, residency=checker, now=_clock(start=9000)) pack = generator.generate("trace-payment-771") print(f" {pack.summary()}") print(f" chain head: {pack.chain_head[:32]}...") print(f" signature : {pack.signature}") ok, problem = generator.verify(pack) print(f" verifies : {ok}") print() print(" the examiner's question, answered from the pack:") answers = { "who authorized it": ("sess", ("user", "delegation_chain")), "what was it permitted to do": ("pol", ("rule", "policy_version")), "what data did it use": ("ret", ("query", "returned", "retrieval_snapshot")), "which model version": ("inf", ("base_model_version", "prompt_version")), "which policy version": ("pol", ("policy_version",)), "who reviewed it": ("appr", ("approvers", "rationale")), "what happened": ("act", ("payment_id", "value_micros", "reference")), } by_id = {a.artifact_id.split(":")[-1]: a for a in pack.artifacts} for question, (aid, fields) in answers.items(): artifact = by_id[aid] values = ", ".join(f"{f}={artifact.attributes.get(f)}" for f in fields) print(f" {question:<30} {values}") print() incomplete = LineageGraph() build_trace(incomplete, "trace-no-approval", with_approval=False) bad_generator = EvidenceGenerator(graph=incomplete, inventory=inventory, residency=checker, now=_clock(start=9000)) try: bad_generator.generate("trace-no-approval") except EvidencePackError as exc: print(f" a 250,000 action with no approval record:") print(f" REFUSED — {exc}") print(" -> a missing artifact FAILS with the artifact named. Failing loudly makes") print(" the gap a fixable engineering problem; producing a pack with a hole makes") print(" it an audit finding.") print() tampered_graph = LineageGraph() build_trace(tampered_graph, "trace-tamper") tamper_gen = EvidenceGenerator(graph=tampered_graph, inventory=inventory, residency=checker, now=_clock(start=9000)) tampered_pack = tamper_gen.generate("trace-tamper") edited = list(tampered_pack.artifacts) edited[-1] = replace(edited[-1], attributes={**edited[-1].attributes, "value_micros": 1}) tampered_pack = replace(tampered_pack, artifacts=tuple(edited)) ok, problem = tamper_gen.verify(tampered_pack) print(f" after editing the action's value: verifies={ok} — {problem}") print() print("=" * 78) print("8. A CONTROL THAT EMITS NO ARTIFACT DOES NOT EXIST") print("=" * 78) print(f" {'framework':<12} {'controls':<7} {'emit evidence':<14} silent") for row in control_coverage(): print(f" {row.framework:<12} {len(row.controls):<7} " f"{len(row.evidence_emitting):<14} {list(row.silent) or '-'}") print() print(" silent controls, named:") for control in CONTROL_CATALOGUE: if control.emits is None: print(f" {control.control_id} {control.name} ({control.component})") print(" -> not necessarily wrong — egress allow-listing is enforced by the network") print(" and is genuinely hard to attach to a trace. But it must be NAMED, because") print(" at audit time it needs some other form of evidence, and knowing which") print(" controls those are is the difference between a prepared answer and a") print(" scramble.") print() missing = verify_control_evidence(graph, "trace-payment-771") print(f" controls with no artifact on this trace: {missing or 'none'}") if __name__ == "__main__": # pragma: no cover main()