"""Lab 01 — The evidence engine. Every other phase in this track emits an artifact. This is where those artifacts become **evidence**. 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, which model version decided, which policy version allowed it, and who reviewed it."* Work top to bottom; each section's tests pass once that section is done: 1. risk tiering 2. the model configuration 3. the inventory 4. lineage 5. residency 6. reproducibility 7. third-party governance 8. the evidence pack 9. control coverage pytest # against your work LAB_MODULE=solution pytest # against the reference Determinism rules: the clock is injected, records are hash-chained, traversal is sorted, and money is integer micro-units. """ 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 what 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'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 #: TODO: one policy per tier. The tests pin these relationships: #: * TIER_1 requires independent validation AND board approval; TIER_3 requires neither; #: * a higher tier caps autonomy LOWER, demands MORE eval cases, and is monitored MORE #: often (a smaller ``monitoring_days``); #: * revalidation is more frequent at a higher tier. #: Suggested: T1 (365d reval, 1d monitoring, "assisted", 500 evals), #: T2 (730, 7, "bounded", 200), T3 (1095, 30, "autonomous", 50). TIER_POLICIES: Mapping[RiskTier, TierPolicy] = {} @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]]: """TODO: assign a tier, and say **why** — return ``(tier, sorted_reasons)``. TIER_1 for irreversible actions, financial impact at or above 100,000 (micro-units: 100_000_000_000), or regulatory reporting. TIER_2 for customer impact, restricted data, or any financial impact. TIER_3 otherwise — and it still gets a reason. A tiering function that returns a bare tier is one nobody can challenge, and being challengeable is the point of writing it down. """ raise NotImplementedError # ====================================================================================== # 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 all of them together are the model, which makes a prompt edit a **model change**: revalidation, approval, a version bump. People resist this because it makes prompt changes expensive. 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: """TODO: a stable digest over everything that determines the output distribution — and NOT over ``config_id`` or ``version``, which are labels.""" raise NotImplementedError def differs_from(self, other: "ModelConfiguration") -> List[str]: """TODO: the sorted names of the fields that differ.""" raise NotImplementedError @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: raise NotImplementedError def register(self, entry: InventoryEntry) -> InventoryEntry: """TODO: refuse a duplicate id, a missing owner, a missing business sponsor, and a missing purpose — a model with no stated purpose cannot be validated against anything.""" raise NotImplementedError def get(self, entry_id: str) -> InventoryEntry: raise NotImplementedError def validate(self, entry_id: str, *, validator: str, state: ValidationState, conditions: Sequence[str] = ()) -> InventoryEntry: """TODO: record the validation — and **refuse when the validator is 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. """ raise NotImplementedError def promote(self, entry_id: str, *, autonomy_band: str) -> InventoryEntry: """TODO: the gate. Collect **every** reason to refuse, then raise with all of them: * the tier requires independent validation and the state is not APPROVED or APPROVED_WITH_CONDITIONS; * validation has EXPIRED; * fewer eval cases than the tier requires; * the requested autonomy exceeds the tier's maximum (see ``_autonomy_rank``). 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 plan for it. """ raise NotImplementedError def record_change(self, entry_id: str, configuration: ModelConfiguration ) -> Tuple[InventoryEntry, List[str]]: """TODO: a configuration change **invalidates the validation** and drops the model out of production. An unchanged configuration changes nothing. This is the operational consequence of "the agent configuration is the model". """ raise NotImplementedError def due_for_revalidation(self) -> List[InventoryEntry]: """TODO: validated entries past their tier's ``revalidation_days``. A never-validated entry is not *due* — it simply is not validated.""" raise NotImplementedError def in_production(self) -> List[InventoryEntry]: raise NotImplementedError def all(self) -> List[InventoryEntry]: raise NotImplementedError def history(self) -> List[Tuple[int, str, str]]: """TODO: (tick, entry_id, event) for registration, validation, promotion and configuration changes.""" raise NotImplementedError _AUTONOMY_ORDER = ("read_only", "assisted", "bounded", "autonomous") def _autonomy_rank(band: str) -> int: """TODO: the index in ``_AUTONOMY_ORDER``; raise ``InventoryError`` on an unknown band rather than defaulting.""" raise NotImplementedError # ====================================================================================== # 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 here. Every layer emits 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: """TODO: a canonical sha256 over every field.""" raise NotImplementedError 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 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: """TODO: refuse a duplicate id, and refuse an artifact deriving from one that does not exist yet — **evidence must be emitted in causal order**, and enforcing that at write time is what stops a graph that cannot be walked.""" raise NotImplementedError def get(self, artifact_id: str) -> Artifact: raise NotImplementedError def for_trace(self, trace_id: str) -> List[Artifact]: """TODO: sorted by ``(tick, artifact_id)`` — causal order.""" raise NotImplementedError def ancestors(self, artifact_id: str) -> List[Artifact]: """TODO: **walk backwards** — everything that contributed to this output, sorted. This is the examiner's actual question. "What did the agent use to decide?" is a reachability query, answerable only because every emitter recorded its inputs. """ raise NotImplementedError def descendants(self, artifact_id: str) -> List[Artifact]: """TODO: 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.""" raise NotImplementedError def of_kind(self, trace_id: str, kind: ArtifactKind) -> List[Artifact]: raise NotImplementedError def check_acyclic(self) -> None: """TODO: raise ``LineageError`` naming the cycle if one exists.""" raise NotImplementedError def orphans(self, trace_id: str) -> List[Artifact]: """TODO: artifacts that derive from nothing AND that nothing derives from — excluding SESSION, which is legitimately a root. 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. """ raise NotImplementedError # ====================================================================================== # 4. Residency # ====================================================================================== @dataclass(frozen=True) class ResidencyRule: 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: raise NotImplementedError 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; a record says what did. This is the evidence half of Phase 13's reachability proof: that phase proves no path *can* leave the region; this one proves none *did*. """ def __init__(self, rules: Sequence[ResidencyRule]) -> None: raise NotImplementedError def rule_for(self, classification: str) -> ResidencyRule: """TODO: raise ``LineageError`` for an unknown classification — an unclassified data flow is a finding, not a default.""" raise NotImplementedError def check_trace(self, graph: LineageGraph, trace_id: str) -> List[ResidencyViolation]: """TODO: check INFERENCE, RETRIEVAL and TOOL_CALL artifacts. * a record missing its classification or region is a violation — **unprovable is a violation**, because "we think it stayed in region" is not evidence; * a region outside the rule's permitted set is a violation; * transit through a region outside the set is a violation unless the rule permits transit. """ raise NotImplementedError # ====================================================================================== # 5. Reproducibility # ====================================================================================== #: TODO: the pins required to re-derive a past decision. The tests expect six: #: base_model_version, prompt_version, retrieval_snapshot, policy_version, #: tool_set_version, guardrail_version #: #: 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, ...] = () @dataclass(frozen=True) class ReproducibilityReport: trace_id: str reproducible: bool present: Tuple[str, ...] missing: Tuple[str, ...] caveats: Tuple[str, ...] def format(self) -> str: raise NotImplementedError def check_reproducibility(graph: LineageGraph, trace_id: str) -> ReproducibilityReport: """TODO: gather the pins from every artifact's attributes; report what is missing. An empty-string pin does not count as present. Two caveats to record without failing: * a ``temperature`` above 0 — **even with every pin, the output is not bit-reproducible.** What is reproducible is the *decision context*, which is what an examiner actually needs. Claiming bit-reproducibility for a sampled model is a claim that will be tested; * ``provider_managed_version`` — a silent provider update cannot be excluded. """ raise NotImplementedError # ====================================================================================== # 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: **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: raise NotImplementedError def register(self, entry: ThirdPartyModel) -> ThirdPartyModel: """TODO: refuse an entry with no data-use terms — "do they train on our data?" is the first question asked, every time.""" raise NotImplementedError def all(self) -> List[ThirdPartyModel]: raise NotImplementedError def assess_concentration(self) -> List[ConcentrationFinding]: """TODO: findings, sorted most-severe first. * traffic share **aggregated per provider** at or above the threshold -> high; * exit readiness NONE or IDENTIFIED -> high above 20% traffic, else medium; * exit readiness TESTED with a stale (or absent) test -> medium; * LIVE needs no test — the traffic *is* the test; * deprecation notice under 90 days -> medium (less than a validation cycle); * no declared sub-processors -> low. """ raise NotImplementedError # ====================================================================================== # 7. The evidence pack # ====================================================================================== #: TODO: what the examiner's question decomposes into — the artifact kind and the question #: it answers. The tests expect six: SESSION, POLICY_DECISION, EXECUTION_STEP, RETRIEVAL, #: INFERENCE, ACTION. #: #: The discipline: **if you cannot name whose question an artifact answers, it does not #: belong in the pack.** REQUIRED_ARTIFACTS: Mapping[ArtifactKind, str] = {} #: Required only when the action was at or 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: raise NotImplementedError class EvidencePackError(Exception): pass class EvidenceGenerator: """Generate the pack. **Generated, not assembled.** A generator reads records 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 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: raise NotImplementedError def generate(self, trace_id: str, *, strict: bool = True) -> EvidencePack: """TODO: * no artifacts for the trace -> raise; * every kind in ``REQUIRED_ARTIFACTS`` must be present, or it is missing — report ``"kind (the question it answers)"``; * an ACTION at or above the dual-control threshold additionally requires an APPROVAL; * check the graph is acyclic, run the reproducibility check and the residency check; * hash-chain the artifacts in causal order, and sign the pack; * ``strict`` raises on anything missing; otherwise return an incomplete pack. """ raise NotImplementedError def verify(self, pack: EvidencePack) -> Tuple[bool, Optional[str]]: """TODO: recompute the chain and the signature. Return ``(ok, problem)``. Both checks are needed: the chain catches an edited or dropped artifact, and the signature catches an edited chain head. """ raise NotImplementedError # ====================================================================================== # 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.* It 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") #: TODO: the controls this track actually built, each naming the artifact it emits (or #: None). The tests expect unique ids, at least five CBUAE controls, at least one SR 11-7 #: control, and at least one **silent** control (``emits=None``). #: #: Include ``C-05 Dual control above threshold`` emitting APPROVAL — the tests use it. CONTROL_CATALOGUE: Tuple[Control, ...] = () @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]: """TODO: generate from the catalogue, one row per framework, sorted. It surfaces the **silent controls**, which are not necessarily wrong — egress allow-listing is enforced by the network and is genuinely hard to attach to a trace. But they must be *named*, because at audit time they need some other form of evidence, and knowing which controls those are is the difference between a prepared answer and a scramble. """ raise NotImplementedError def verify_control_evidence(graph: LineageGraph, trace_id: str, controls: Sequence[Control] = CONTROL_CATALOGUE ) -> List[str]: """TODO: which evidence-emitting controls left no artifact on this trace? The check that turns the catalogue from a document into a test. """ raise NotImplementedError # ====================================================================================== # Worked example # ====================================================================================== def build_trace(graph: LineageGraph, trace_id: str, *, with_approval: bool = True, region: str = "uaenorth", pin_retrieval: bool = True) -> None: """TODO: emit the artifacts a real run would, in causal order. **The tests use this**, so build it early. The shape, with ids the tests expect (prefixed ``{trace_id}:``): sess SESSION (root) pol POLICY_DECISION <- sess doc1 DOCUMENT (root) doc2 DOCUMENT (root) ret RETRIEVAL <- pol, doc1, doc2 guard GUARDRAIL <- ret inf INFERENCE <- guard step EXECUTION_STEP <- inf appr APPROVAL <- step (only when with_approval) act ACTION <- step[, appr] ``ret``, ``inf`` and ``act`` carry ``data_classification="confidential"`` and ``region``. ``inf`` carries every pin except ``retrieval_snapshot``, which is on ``ret`` and omitted when ``pin_retrieval`` is False. ``act`` carries ``value_micros=250_000_000_000``. """ raise NotImplementedError def main() -> None: # pragma: no cover """TODO (optional): once the tests pass, build the eight-section demo. Compare against ``python solution.py`` — but only after your own runs. """ raise NotImplementedError if __name__ == "__main__": # pragma: no cover main()