"""Reference solution — the control plane: KYA, policy-as-code, continuous authorization. The layer that knows every agent in the bank. Being that layer is the entire justification for having a platform rather than twelve teams with API keys. Three ideas carry it: * KNOW YOUR AGENT is a RUNTIME property, not an onboarding checklist — owner, tools, model version, evaluation freshness, current posture, all answerable per request; * POLICY IS CODE in a versioned, signed bundle, because a rule in a wiki is a suggestion and a rule in a bundle is a control with an audit trail; * FAIL STATIC is the third posture nobody remembers: when the control plane is unreachable the data plane keeps enforcing the last known-good bundle, alarms on staleness, and hard-stops eventually. Deterministic: an injected clock, sorted outputs, hash-based bundle signatures. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import hmac import json from dataclasses import dataclass, field, replace from enum import Enum from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple # ====================================================================================== # 1. The decision inputs — subject, action, resource, environment # ====================================================================================== _CLASSIFICATION_ORDER = ("public", "internal", "confidential", "restricted") def classification_rank(name: str) -> int: try: return _CLASSIFICATION_ORDER.index(name) except ValueError: raise ValueError(f"unknown data classification: {name!r}") from None @dataclass(frozen=True) class Subject: """The BLENDED principal: an agent acting for a user. Both halves constrain what is allowed, and both appear in the decision record. A subject with an agent and no user is a workload acting on its own behalf — which is a different, narrower thing, and the policy must be able to tell. """ agent_id: str user_id: Optional[str] tenant: str scopes: Tuple[str, ...] = () delegation_chain: Tuple[str, ...] = () clearance: str = "internal" @property def is_user_scoped(self) -> bool: return self.user_id is not None @dataclass(frozen=True) class Resource: resource_type: str # "tool" | "data" | "model" resource_id: str tenant: str classification: str = "internal" barrier: Optional[str] = None @dataclass(frozen=True) class Environment: """Everything about *now* that the policy may read. ``anomaly_score`` and ``evaluation_age_ticks`` are the posture signals: they are what make authorization *continuous* rather than a fact established at session start. """ tick: int = 0 channel: str = "api" anomaly_score: float = 0.0 evaluation_age_ticks: int = 0 approvals: Tuple[str, ...] = () step_index: int = 1 amount_micros: int = 0 @dataclass(frozen=True) class Request: subject: Subject action: str resource: Resource environment: Environment = Environment() # ====================================================================================== # 2. Policy — default-deny, deny-overrides # ====================================================================================== class Effect(str, Enum): ALLOW = "allow" DENY = "deny" #: A condition reads the whole request and returns a boolean. Keeping conditions as #: callables avoids writing an expression parser — the EVALUATION SEMANTICS are the #: lesson, not the grammar. Condition = Callable[[Request], bool] @dataclass(frozen=True) class Rule: """One policy rule. ``actions``/``resource_types``/``tenants`` are cheap structural matches; ``condition`` is the expressive part. Empty tuples mean "any", which is the convention every policy language uses and the one that produces an accidental match-everything rule when a field is forgotten — hence the validation in ``PolicyBundle``. """ name: str effect: Effect actions: Tuple[str, ...] = () resource_types: Tuple[str, ...] = () tenants: Tuple[str, ...] = () condition: Optional[Condition] = None reason: str = "" def matches(self, request: Request) -> bool: if self.actions and not any(_action_matches(a, request.action) for a in self.actions): return False if self.resource_types and request.resource.resource_type not in self.resource_types: return False if self.tenants and request.subject.tenant not in self.tenants: return False if self.condition is not None and not self.condition(request): return False return True def _action_matches(pattern: str, action: str) -> bool: """Exact, or a single trailing wildcard: ``payments.*`` matches ``payments.release``.""" if pattern == action: return True if pattern.endswith(".*"): return action.startswith(pattern[:-1]) return False @dataclass(frozen=True) class Decision: """The artifact. Not a boolean — a boolean cannot be shown to an examiner. ``policy_version`` is the field that makes this evidence: six months later, "which policy allowed this?" has an answer that can be looked up. """ effect: Effect reason: str rule_name: str policy_version: str obligations: Tuple[str, ...] = () matched_rules: Tuple[str, ...] = () evaluated_at: int = 0 @property def allowed(self) -> bool: return self.effect is Effect.ALLOW # ====================================================================================== # 3. The policy bundle — versioned, signed, atomically activated # ====================================================================================== class BundleError(Exception): pass @dataclass(frozen=True) class PolicyBundle: """A versioned, signed set of rules. Signing matters more than it looks: **atomic activation of a verified bundle** is the mechanism that makes fail-static real. A malformed or unsigned bundle is rejected and the previous one stays live, so a bad push degrades to "stale" rather than "broken". """ version: str rules: Tuple[Rule, ...] created_tick: int signature: str = "" def digest(self) -> str: """Over the version and the rules' STRUCTURE. Conditions are callables and cannot be hashed — which is exactly why a production bundle carries Rego or Cedar source rather than code, and is one of the miniature's honest limits.""" material = json.dumps({ "version": self.version, "rules": [ {"name": r.name, "effect": r.effect.value, "actions": list(r.actions), "resource_types": list(r.resource_types), "tenants": list(r.tenants), "has_condition": r.condition is not None, "reason": r.reason} for r in self.rules ], }, sort_keys=True, separators=(",", ":")) return hashlib.sha256(material.encode()).hexdigest() def sign(self, secret: bytes) -> "PolicyBundle": signature = hmac.new(secret, self.digest().encode(), hashlib.sha256).hexdigest() return replace(self, signature=signature) def verify(self, secret: bytes) -> None: if not self.signature: raise BundleError(f"bundle {self.version} is unsigned") expected = hmac.new(secret, self.digest().encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, self.signature): raise BundleError(f"bundle {self.version} has a bad signature") def validate(self) -> None: """Structural checks that catch the mistakes a policy author actually makes.""" names = [r.name for r in self.rules] duplicates = sorted({n for n in names if names.count(n) > 1}) if duplicates: raise BundleError(f"duplicate rule names: {duplicates}") for rule in self.rules: # An ALLOW with no actions, no resource types, no tenants and no condition # matches every request. It is never intended, and it is silent. if (rule.effect is Effect.ALLOW and not rule.actions and not rule.resource_types and not rule.tenants and rule.condition is None): raise BundleError( f"rule {rule.name!r} is an unconditional allow-everything") # ====================================================================================== # 4. The policy decision point # ====================================================================================== class PolicyEngine: """Default-deny, deny-overrides. Two combining rules, both non-negotiable in a bank: * **default-deny** — no matching rule means DENY. A policy set that fails open the moment somebody forgets a case is not a control. * **deny-overrides** — any matching DENY beats every ALLOW, and evaluation does not stop at the first allow. Order-independence is what makes a policy set reviewable: you can read the rules in any order and reason about the result. """ def __init__(self, bundle: PolicyBundle) -> None: bundle.validate() self.bundle = bundle def evaluate(self, request: Request) -> Decision: matched: List[Rule] = [r for r in self.bundle.rules if r.matches(request)] denies = [r for r in matched if r.effect is Effect.DENY] allows = [r for r in matched if r.effect is Effect.ALLOW] names = tuple(sorted(r.name for r in matched)) tick = request.environment.tick if denies: # Deterministic choice among several denies: the first in bundle order, so # the same request always reports the same rule. rule = denies[0] return Decision(Effect.DENY, rule.reason or f"denied by {rule.name}", rule.name, self.bundle.version, matched_rules=names, evaluated_at=tick) if allows: rule = allows[0] return Decision(Effect.ALLOW, rule.reason or f"allowed by {rule.name}", rule.name, self.bundle.version, matched_rules=names, evaluated_at=tick) return Decision(Effect.DENY, "no matching rule (default deny)", "-", self.bundle.version, matched_rules=names, evaluated_at=tick) # ====================================================================================== # 5. Bundle distribution — and the third posture # ====================================================================================== class Posture(str, Enum): """What the data plane does when the control plane is unreachable. FAIL_OPEN is a security hole. FAIL_SHUT makes the control plane's availability multiply into the data plane's, which is the Phase 00 composition trap. FAIL_STATIC is the option people forget exists, and it is the right one. """ FAIL_OPEN = "fail_open" FAIL_SHUT = "fail_shut" FAIL_STATIC = "fail_static" @dataclass(frozen=True) class DistributorStatus: version: str last_successful_activation: int staleness_ticks: int stale: bool hard_stopped: bool class BundleDistributor: """Polls for a bundle, verifies it, activates it atomically. The three thresholds are the design: * refresh interval — how often we try; * staleness alarm — how long before somebody is told; * hard stop — how long before we refuse to serve, because a policy bundle that is hours old in a bank is worse than an outage. """ def __init__(self, *, secret: bytes, now: Callable[[], int], initial: PolicyBundle, staleness_alarm_ticks: int = 300, hard_stop_ticks: int = 1800, posture: Posture = Posture.FAIL_STATIC) -> None: initial.verify(secret) initial.validate() self.secret = secret self.now = now self.staleness_alarm_ticks = staleness_alarm_ticks self.hard_stop_ticks = hard_stop_ticks self.posture = posture self._active = initial self._activated_at = now() self.failed_activations = 0 self.alarms: List[str] = [] @property def active(self) -> PolicyBundle: return self._active def offer(self, bundle: PolicyBundle) -> bool: """Try to activate. A rejected bundle leaves the previous one live. That is the whole of fail-static: a bad push degrades to STALE, never to BROKEN. """ try: bundle.verify(self.secret) bundle.validate() except BundleError as exc: self.failed_activations += 1 self.alarms.append(f"rejected bundle {bundle.version}: {exc}") return False if bundle.created_tick < self._active.created_tick: self.failed_activations += 1 self.alarms.append( f"rejected bundle {bundle.version}: older than the active one") return False self._active = bundle # atomic swap self._activated_at = self.now() return True def status(self) -> DistributorStatus: age = self.now() - self._activated_at return DistributorStatus( version=self._active.version, last_successful_activation=self._activated_at, staleness_ticks=age, stale=age >= self.staleness_alarm_ticks, hard_stopped=age >= self.hard_stop_ticks) def engine(self) -> PolicyEngine: """Refuse to serve past the hard stop — the one case where fail-static becomes fail-shut, deliberately and with a stated threshold.""" if self.status().hard_stopped: raise BundleError( f"policy bundle is {self.status().staleness_ticks} ticks old; " f"past the hard stop of {self.hard_stop_ticks}") return PolicyEngine(self._active) # ====================================================================================== # 6. The agent registry — the KYA database # ====================================================================================== class AgentState(str, Enum): DRAFT = "draft" APPROVED = "approved" ACTIVE = "active" SUSPENDED = "suspended" RETIRED = "retired" AGENT_TRANSITIONS: Mapping[AgentState, frozenset] = { AgentState.DRAFT: frozenset({AgentState.APPROVED, AgentState.RETIRED}), AgentState.APPROVED: frozenset({AgentState.ACTIVE, AgentState.RETIRED}), AgentState.ACTIVE: frozenset({AgentState.SUSPENDED, AgentState.RETIRED}), AgentState.SUSPENDED: frozenset({AgentState.ACTIVE, AgentState.RETIRED}), } @dataclass(frozen=True) class AgentRecord: """Everything KYA must answer, at runtime, for every agent in production.""" agent_id: str owner: str # a HUMAN tenant: str permitted_tools: Tuple[str, ...] max_classification: str = "internal" model_deployment: str = "" model_version: str = "" # pinned; an unpinned model is a finding autonomy_band: str = "read_only" # read_only | assisted | autonomous state: AgentState = AgentState.DRAFT last_evaluation_tick: int = 0 evaluation_score: float = 0.0 max_action_micros: int = 0 class AgentRegistry: """The authoritative inventory. The reason the platform exists.""" def __init__(self) -> None: self._agents: Dict[str, AgentRecord] = {} def register(self, record: AgentRecord) -> AgentRecord: if record.agent_id in self._agents: raise ValueError(f"{record.agent_id} is already registered") if not record.owner: raise ValueError("every agent needs a human owner") if not record.model_version: raise ValueError( "an agent must pin a model version; an unpinned model changes under you") classification_rank(record.max_classification) self._agents[record.agent_id] = record return record def get(self, agent_id: str) -> AgentRecord: try: return self._agents[agent_id] except KeyError: raise KeyError(f"unknown agent: {agent_id}") from None def transition(self, agent_id: str, target: AgentState) -> AgentRecord: record = self.get(agent_id) if target not in AGENT_TRANSITIONS.get(record.state, frozenset()): raise ValueError( f"{record.state.value} -> {target.value} is not a legal transition") updated = replace(record, state=target) self._agents[agent_id] = updated return updated def record_evaluation(self, agent_id: str, *, tick: int, score: float) -> AgentRecord: updated = replace(self.get(agent_id), last_evaluation_tick=tick, evaluation_score=score) self._agents[agent_id] = updated return updated def by_owner(self, owner: str) -> List[AgentRecord]: return sorted((a for a in self._agents.values() if a.owner == owner), key=lambda a: a.agent_id) def all(self) -> List[AgentRecord]: return sorted(self._agents.values(), key=lambda a: a.agent_id) # ====================================================================================== # 7. Tool registry and authorization-aware capability discovery # ====================================================================================== class SideEffect(str, Enum): READ = "read" WRITE_IDEMPOTENT = "write_idempotent" WRITE_NON_IDEMPOTENT = "write_non_idempotent" IRREVERSIBLE = "irreversible" @dataclass(frozen=True) class ToolRecord: tool_id: str description: str side_effect: SideEffect required_scopes: Tuple[str, ...] = () classification: str = "internal" tenants: Tuple[str, ...] = () owner: str = "unknown" class ToolRegistry: def __init__(self) -> None: self._tools: Dict[str, ToolRecord] = {} def publish(self, tool: ToolRecord) -> None: if tool.tool_id in self._tools: raise ValueError(f"{tool.tool_id} is already published") classification_rank(tool.classification) self._tools[tool.tool_id] = tool def get(self, tool_id: str) -> Optional[ToolRecord]: return self._tools.get(tool_id) def all(self) -> List[ToolRecord]: return sorted(self._tools.values(), key=lambda t: t.tool_id) @dataclass(frozen=True) class Capability: tool_id: str description: str side_effect: SideEffect @dataclass(frozen=True) class PostureFinding: """A KYA check that failed, and how hard it bites. The distinction is the whole subtlety of posture. Some failures are **categorical** — a suspended agent should do nothing at all, and an ownerless one has no accountable human. Others are **graduated**: an agent whose evaluation went stale this morning is not dangerous to *read* with, but it has no business releasing a payment. Collapsing the two is the common mistake, and it fails in both directions. Treat everything as categorical and a routine eval lapse takes the fleet down, which teaches operators to raise the thresholds until they never fire. Treat everything as graduated and a suspended agent keeps reading customer data. """ check: str detail: str blocks_reads: bool def __str__(self) -> str: return self.detail class ControlPlane: """The composition: registries + policy + posture, answering two questions. * ``discover`` — what may this agent see RIGHT NOW? * ``authorize`` — may it do this specific thing RIGHT NOW? Discovery is an authorization decision, not a lookup. A capability the principal cannot use must not be visible: its name and description are information, and a model that can see a tool will eventually try to call it. """ def __init__(self, *, agents: AgentRegistry, tools: ToolRegistry, distributor: BundleDistributor, now: Callable[[], int], max_evaluation_age_ticks: int = 1000, anomaly_block_writes: float = 0.5, anomaly_block_all: float = 0.8) -> None: self.agents = agents self.tools = tools self.distributor = distributor self.now = now self.max_evaluation_age_ticks = max_evaluation_age_ticks self.anomaly_block_writes = anomaly_block_writes self.anomaly_block_all = anomaly_block_all self.decision_log: List[Decision] = [] # -- KYA posture --------------------------------------------------------------- def posture_checks(self, record: AgentRecord, environment: Environment) -> List[PostureFinding]: """The KYA questions, evaluated NOW. Returns every failure, in a stable order. Every failure, not the first: an operator fixing one at a time is an operator who discovers the next one on the next attempt. """ findings: List[PostureFinding] = [] if record.state is not AgentState.ACTIVE: findings.append(PostureFinding( "state", f"agent is {record.state.value}", blocks_reads=True)) if not record.owner: findings.append(PostureFinding( "owner", "agent has no owner", blocks_reads=True)) if not record.model_version: findings.append(PostureFinding( "model", "agent has no pinned model version", blocks_reads=True)) age = environment.tick - record.last_evaluation_tick if age > self.max_evaluation_age_ticks: # GRADUATED. A lapsed evaluation is a reason not to let it act, not a reason # to take it offline — and a control that takes the fleet offline every time # an eval job is late is a control operators will disable. findings.append(PostureFinding( "evaluation", f"evaluation is {age} ticks old", blocks_reads=False)) if environment.anomaly_score >= self.anomaly_block_all: findings.append(PostureFinding( "anomaly", f"anomaly score {environment.anomaly_score:.2f}", blocks_reads=True)) elif environment.anomaly_score >= self.anomaly_block_writes: findings.append(PostureFinding( "anomaly", f"anomaly score {environment.anomaly_score:.2f}", blocks_reads=False)) return findings def blocking_findings(self, record: AgentRecord, environment: Environment, *, high_impact: bool) -> List[PostureFinding]: """The subset that actually stops THIS request.""" return [f for f in self.posture_checks(record, environment) if f.blocks_reads or high_impact] # -- discovery ------------------------------------------------------------------ def discover(self, subject: Subject, environment: Environment = Environment()) -> List[Capability]: """The tools this subject may use right now — filtered by registry, posture and policy, in that order. Posture degrades the answer the same way it degrades authorization. A categorical failure returns NOTHING: an agent that is suspended should be visibly unable to act, not quietly operating on a reduced tool set. A graduated failure returns the READ capabilities only — which is the honest thing to show an agent that may look but not touch, and it means the model is never offered a tool the next call would refuse. """ record = self.agents.get(subject.agent_id) findings = self.posture_checks(record, environment) if any(f.blocks_reads for f in findings): return [] reads_only = bool(findings) allowed_rank = classification_rank(record.max_classification) held = set(subject.scopes) out: List[Capability] = [] for tool in self.tools.all(): if tool.tool_id not in record.permitted_tools: continue if reads_only and tool.side_effect is not SideEffect.READ: continue if tool.tenants and subject.tenant not in tool.tenants: continue if classification_rank(tool.classification) > allowed_rank: continue if not set(tool.required_scopes) <= held: continue probe = Request( subject=subject, action=tool.tool_id, resource=Resource("tool", tool.tool_id, subject.tenant, tool.classification), environment=environment) if not self.distributor.engine().evaluate(probe).allowed: continue out.append(Capability(tool.tool_id, tool.description, tool.side_effect)) return out # -- authorization --------------------------------------------------------------- def authorize(self, request: Request, *, high_impact: bool = False) -> Decision: """The full decision: posture first (cheap, and it short-circuits), then policy. Every outcome is logged with the policy version, because a decision without one cannot answer an examiner's question — and that includes the posture denials, which is easy to forget precisely because policy was never consulted. """ try: record = self.agents.get(request.subject.agent_id) except KeyError: decision = Decision(Effect.DENY, "agent is not registered", "kya:unregistered", self.distributor.active.version, evaluated_at=request.environment.tick) self.decision_log.append(decision) return decision blocking = self.blocking_findings(record, request.environment, high_impact=high_impact) if blocking: decision = Decision(Effect.DENY, "; ".join(f.detail for f in blocking), "kya:posture", self.distributor.active.version, evaluated_at=request.environment.tick) self.decision_log.append(decision) return decision decision = self.distributor.engine().evaluate(request) self.decision_log.append(decision) return decision # ====================================================================================== # 8. Continuous authorization # ====================================================================================== @dataclass(frozen=True) class Lease: """A cached decision with a TTL. Caching is not an optimization here; without it the PDP is on the synchronous path of every step and its availability multiplies into the platform's. The TTL is the revocation-latency budget, stated as a number. """ decision: Decision issued_at: int ttl_ticks: int request_fingerprint: str agent_id: str def is_live(self, tick: int) -> bool: return tick - self.issued_at < self.ttl_ticks def fingerprint(request: Request) -> str: """What makes two requests 'the same decision'. Deliberately EXCLUDES the environment's volatile fields — anomaly score, evaluation age, tick — because including them would make every lease a miss. Which is precisely why a lease has a TTL *and* why some events must invalidate it explicitly (§ revoke). """ material = "|".join([ request.subject.agent_id, request.subject.user_id or "-", request.subject.tenant, ",".join(sorted(request.subject.scopes)), request.action, request.resource.resource_type, request.resource.resource_id, request.resource.tenant, request.resource.classification, ]) return hashlib.blake2b(material.encode(), digest_size=12).hexdigest() class ContinuousAuthorizer: """Re-evaluates during a long-running task, not only at session start. An agent task can run for an hour while entitlements, agent posture and risk signals all change. A decision made at minute zero and cached for the session is a decision about conditions that no longer hold. Three triggers force re-evaluation regardless of the TTL: * the policy bundle version changed; * the action is high-impact (a side-effect class that is not a read); * the lease was explicitly revoked (a kill switch that beats the refresh interval). """ def __init__(self, *, control_plane: ControlPlane, now: Callable[[], int], lease_ttl_ticks: int = 60) -> None: self.control_plane = control_plane self.now = now self.lease_ttl_ticks = lease_ttl_ticks self._leases: Dict[str, Lease] = {} self._revoked: Set[str] = set() self.evaluations = 0 self.lease_hits = 0 def authorize(self, request: Request, *, high_impact: bool = False) -> Decision: key = fingerprint(request) tick = request.environment.tick lease = self._leases.get(key) reusable = ( lease is not None and lease.is_live(tick) and not high_impact and request.subject.agent_id not in self._revoked and lease.decision.policy_version == self.control_plane.distributor.active.version ) if reusable: self.lease_hits += 1 return lease.decision self.evaluations += 1 decision = self.control_plane.authorize(request, high_impact=high_impact) if decision.allowed and not high_impact: self._leases[key] = Lease(decision, tick, self.lease_ttl_ticks, key, request.subject.agent_id) return decision def revoke_agent(self, agent_id: str) -> int: """The kill switch. Beats the lease TTL and the bundle refresh interval — which is the point: 'we suspended it but it kept acting for 60 seconds' is a sentence you do not want to say to an examiner. Note the two halves. Dropping the leases is what makes revocation *fast*; the ``_revoked`` set is what keeps it fast, because a request arriving one tick later would otherwise mint a fresh lease from a control plane that has not yet caught up. Returns the number of leases dropped, because "how many in-flight decisions did we just invalidate?" is an incident question. """ self._revoked.add(agent_id) dropped = [k for k, lease in self._leases.items() if lease.agent_id == agent_id] for key in dropped: del self._leases[key] return len(dropped) def reinstate_agent(self, agent_id: str) -> None: self._revoked.discard(agent_id) # ====================================================================================== # 9. Evaluation pipeline — quality as an authorization input # ====================================================================================== @dataclass(frozen=True) class EvalCase: case_id: str kind: str # "golden" | "safety" | "regression" expected: str @dataclass(frozen=True) class EvalResult: agent_id: str tick: int total: int passed: int safety_failures: int @property def score(self) -> float: return self.passed / self.total if self.total else 0.0 @dataclass(frozen=True) class PromotionVerdict: promoted: bool reasons: Tuple[str, ...] class EvaluationPipeline: """Golden sets, safety suites, regression gates — and the link that matters. Evaluation status is an *authorization input* (via the posture check), not a dashboard. That single link is what makes quality a control rather than a report: an agent whose evaluation is stale or failing cannot act. """ def __init__(self, *, registry: AgentRegistry, now: Callable[[], int], min_score: float = 0.85, allow_safety_failures: int = 0) -> None: self.registry = registry self.now = now self.min_score = min_score self.allow_safety_failures = allow_safety_failures self.history: List[EvalResult] = [] def run(self, agent_id: str, cases: Sequence[EvalCase], respond: Callable[[EvalCase], str]) -> EvalResult: self.registry.get(agent_id) passed = 0 safety_failures = 0 for case in cases: ok = respond(case) == case.expected passed += 1 if ok else 0 if case.kind == "safety" and not ok: safety_failures += 1 result = EvalResult(agent_id, self.now(), len(cases), passed, safety_failures) self.history.append(result) self.registry.record_evaluation(agent_id, tick=result.tick, score=result.score) return result def gate(self, result: EvalResult, *, baseline: Optional[EvalResult] = None) -> PromotionVerdict: """The promotion gate. A safety failure is disqualifying regardless of score — an aggregate that averages away a safety failure is a gate that does not gate.""" reasons: List[str] = [] if result.safety_failures > self.allow_safety_failures: reasons.append(f"{result.safety_failures} safety failure(s)") if result.score < self.min_score: reasons.append(f"score {result.score:.2f} below {self.min_score:.2f}") if baseline is not None and result.score < baseline.score: reasons.append( f"regression: {result.score:.2f} < baseline {baseline.score:.2f}") return PromotionVerdict(not reasons, tuple(reasons)) # ====================================================================================== # 10. Tracing and lineage at agent and tool granularity # ====================================================================================== @dataclass(frozen=True) class Span: """One unit of work, with the identity and decision attached. 'Tracing at agent and tool granularity' means exactly this: each agent step and each tool call is its own span, carrying who, what, which policy, and which model. """ span_id: str parent_id: Optional[str] trace_id: str name: str kind: str # "agent" | "tool" | "model" | "policy" agent_id: str user_id: Optional[str] tenant: str started_at: int ended_at: int policy_version: str = "" decision: Optional[str] = None model_version: str = "" attributes: Mapping[str, str] = field(default_factory=dict) @property def duration(self) -> int: return self.ended_at - self.started_at class Tracer: def __init__(self, *, now: Callable[[], int]) -> None: self.now = now self.spans: List[Span] = [] self._counter = 0 def _next_id(self) -> str: self._counter += 1 return f"span-{self._counter}" def record(self, *, trace_id: str, parent_id: Optional[str], name: str, kind: str, subject: Subject, started_at: int, policy_version: str = "", decision: Optional[str] = None, model_version: str = "", **attributes: str) -> Span: span = Span( span_id=self._next_id(), parent_id=parent_id, trace_id=trace_id, name=name, kind=kind, agent_id=subject.agent_id, user_id=subject.user_id, tenant=subject.tenant, started_at=started_at, ended_at=self.now(), policy_version=policy_version, decision=decision, model_version=model_version, attributes=dict(attributes)) self.spans.append(span) return span def trace(self, trace_id: str) -> List[Span]: return [s for s in self.spans if s.trace_id == trace_id] def lineage(self, trace_id: str) -> Dict[str, object]: """What produced this outcome: the agents, tools, models and policy versions. This is the query an examiner's question reduces to, and it is only answerable because every span carries identity and version. """ spans = self.trace(trace_id) return { "trace_id": trace_id, "agents": sorted({s.agent_id for s in spans}), "users": sorted({s.user_id for s in spans if s.user_id}), "tools": sorted({s.name for s in spans if s.kind == "tool"}), "models": sorted({s.model_version for s in spans if s.model_version}), "policy_versions": sorted({s.policy_version for s in spans if s.policy_version}), "decisions": [s.decision for s in spans if s.decision], "span_count": len(spans), } # ====================================================================================== # Worked example # ====================================================================================== def _clock(start: int = 1000, step: int = 1) -> Callable[[], int]: state = {"t": start - step} def now() -> int: state["t"] += step return state["t"] return now SECRET = b"control-plane-signing-key" def _build_rules() -> Tuple[Rule, ...]: return ( Rule("deny-cross-tenant", Effect.DENY, condition=lambda r: r.resource.tenant != r.subject.tenant, reason="resource belongs to another tenant"), Rule("deny-above-clearance", Effect.DENY, condition=lambda r: classification_rank(r.resource.classification) > classification_rank(r.subject.clearance), reason="resource classification exceeds the subject's clearance"), Rule("deny-barrier", Effect.DENY, condition=lambda r: r.resource.barrier is not None, reason="resource is behind an information barrier"), Rule("deny-irreversible-without-user", Effect.DENY, actions=("payments.release",), condition=lambda r: not r.subject.is_user_scoped, reason="an irreversible action requires a user-scoped credential"), Rule("deny-irreversible-without-dual-control", Effect.DENY, actions=("payments.release",), condition=lambda r: len({a for a in r.environment.approvals if a != r.subject.agent_id}) < 2, reason="two distinct human approvers required"), Rule("deny-high-anomaly-writes", Effect.DENY, actions=("payments.*", "crm.write"), condition=lambda r: r.environment.anomaly_score >= 0.5, reason="anomalous behaviour; write actions suspended"), Rule("allow-reads", Effect.ALLOW, actions=("payments.lookup", "crm.read", "policy.read"), reason="read actions are permitted within tenant and clearance"), Rule("allow-writes-with-scope", Effect.ALLOW, actions=("crm.write",), condition=lambda r: "crm.write" in r.subject.scopes, reason="write permitted with the crm.write scope"), Rule("allow-release-with-approvals", Effect.ALLOW, actions=("payments.release",), condition=lambda r: "payments.release" in r.subject.scopes, reason="release permitted with scope, subject to the deny rules"), ) def main() -> None: # pragma: no cover - narrative output now = _clock() bundle = PolicyBundle("v1", _build_rules(), created_tick=now()).sign(SECRET) distributor = BundleDistributor(secret=SECRET, now=now, initial=bundle, staleness_alarm_ticks=50, hard_stop_ticks=200) agents = AgentRegistry() agents.register(AgentRecord( agent_id="payments-investigator", owner="layla.almansouri", tenant="wholesale", permitted_tools=("payments.lookup", "crm.read", "crm.write"), max_classification="confidential", model_deployment="azure-gpt-uae-ptu", model_version="gpt-frontier-2026-02-11", autonomy_band="assisted", last_evaluation_tick=1000, evaluation_score=0.94)) agents.transition("payments-investigator", AgentState.APPROVED) agents.transition("payments-investigator", AgentState.ACTIVE) tools = ToolRegistry() for tool in ( ToolRecord("payments.lookup", "Look up a payment", SideEffect.READ, ("payments.read",), "confidential", ("wholesale",)), ToolRecord("payments.release", "Release a held payment", SideEffect.IRREVERSIBLE, ("payments.release",), "restricted", ("wholesale",)), ToolRecord("crm.read", "Read a customer record", SideEffect.READ, ("crm.read",)), ToolRecord("crm.write", "Append a CRM note", SideEffect.WRITE_IDEMPOTENT, ("crm.write",)), ToolRecord("hr.lookup", "Look up an employee", SideEffect.READ, ("hr.read",), "confidential", ("group",)), ): tools.publish(tool) plane = ControlPlane(agents=agents, tools=tools, distributor=distributor, now=now, max_evaluation_age_ticks=100) subject = Subject(agent_id="payments-investigator", user_id="u-42", tenant="wholesale", scopes=("payments.read", "crm.read", "crm.write"), delegation_chain=("orchestrator",), clearance="confidential") print("=" * 78) print("1. CAPABILITY DISCOVERY IS AN AUTHORIZATION DECISION") print("=" * 78) caps = plane.discover(subject, Environment(tick=1010)) for cap in caps: print(f" {cap.tool_id:<20} {cap.side_effect.value:<20} {cap.description}") print(" not shown, and each for a different reason:") print(" payments.release — not in the agent's registered tool set") print(" hr.lookup — restricted to tenant 'group'") print(" a tool this agent cannot call must not appear: its name is information,") print(" and a model that can see a tool will eventually try to call it.") print() print("=" * 78) print("2. DEFAULT-DENY, DENY-OVERRIDES") print("=" * 78) cases = [ ("read in tenant", Request(subject, "payments.lookup", Resource("tool", "payments.lookup", "wholesale", "confidential"), Environment(tick=1010))), ("cross-tenant read", Request(subject, "payments.lookup", Resource("tool", "payments.lookup", "retail", "confidential"), Environment(tick=1010))), ("above clearance", Request(subject, "payments.lookup", Resource("tool", "payments.lookup", "wholesale", "restricted"), Environment(tick=1010))), ("behind a barrier", Request(subject, "crm.read", Resource("tool", "crm.read", "wholesale", "internal", barrier="advisory"), Environment(tick=1010))), ("unmatched action", Request(subject, "treasury.trade", Resource("tool", "treasury.trade", "wholesale"), Environment(tick=1010))), ] for label, request in cases: decision = plane.authorize(request) print(f" {label:<20} {decision.effect.value.upper():<6} {decision.rule_name:<24} " f"{decision.reason}") print(" -> an unmatched action is DENIED. A policy set that fails open the moment") print(" somebody forgets a case is not a control.") print() print("=" * 78) print("3. DENY BEATS ALLOW, AND ORDER DOES NOT MATTER") print("=" * 78) release_subject = replace(subject, scopes=subject.scopes + ("payments.release",)) request = Request(release_subject, "payments.release", Resource("tool", "payments.release", "wholesale", "confidential"), Environment(tick=1010, approvals=("alice",))) decision = plane.authorize(request) print(f" one approver -> {decision.effect.value.upper()}: {decision.reason}") print(f" matched rules: {list(decision.matched_rules)}") print(" both an ALLOW and a DENY matched; the DENY wins, regardless of order.") request = replace(request, environment=Environment(tick=1010, approvals=("alice", "bob"))) decision = plane.authorize(request) print(f" two approvers -> {decision.effect.value.upper()}: {decision.reason}") print() print("=" * 78) print("4. KYA POSTURE — EVALUATED NOW, NOT AT ONBOARDING") print("=" * 78) read = Request(subject, "crm.read", Resource("tool", "crm.read", "wholesale")) write = Request(subject, "crm.write", Resource("tool", "crm.write", "wholesale")) stale = Environment(tick=1200) # evaluation was at 1000, max age 100 d_read = plane.authorize(replace(read, environment=stale)) d_write = plane.authorize(replace(write, environment=stale), high_impact=True) print(f" stale eval, read -> {d_read.effect.value.upper()}") print(f" stale eval, write -> {d_write.effect.value.upper()}: {d_write.reason}") print(" GRADUATED. A control that takes the fleet offline every time an eval job") print(" is late is a control operators will quietly disable.") anomalous = Environment(tick=1010, anomaly_score=0.9) decision = plane.authorize(replace(read, environment=anomalous)) print(f" anomaly 0.90 -> {decision.effect.value.upper()}: {decision.reason}") agents.transition("payments-investigator", AgentState.SUSPENDED) decision = plane.authorize(replace(read, environment=Environment(tick=1010))) print(f" suspended agent -> {decision.effect.value.upper()}: {decision.reason}") print(" CATEGORICAL. Nothing at all, for any action.") agents.transition("payments-investigator", AgentState.ACTIVE) caps = plane.discover(subject, stale) print(f" discovery, stale eval : {[c.tool_id for c in caps]} — reads only, so the") print(" model is never offered a tool the next call would refuse.") agents.transition("payments-investigator", AgentState.SUSPENDED) print(f" discovery, suspended : " f"{[c.tool_id for c in plane.discover(subject, Environment(tick=1010))]}") agents.transition("payments-investigator", AgentState.ACTIVE) print() print("=" * 78) print("5. FAIL STATIC — THE THIRD POSTURE") print("=" * 78) bad = PolicyBundle("v2-broken", _build_rules(), created_tick=now()) # unsigned accepted = distributor.offer(bad) print(f" unsigned bundle offered -> accepted={accepted}, " f"active is still {distributor.active.version}") tampered = PolicyBundle("v2-tampered", _build_rules(), created_tick=now()) tampered = replace(tampered, signature="deadbeef") distributor.offer(tampered) print(f" tampered bundle offered -> active is still {distributor.active.version}") print(f" alarms: {distributor.alarms}") print(" -> a bad push degrades to STALE, never to BROKEN. That is fail-static:") print(" not fail-open (a hole) and not fail-shut (a self-inflicted outage).") good = PolicyBundle("v2", _build_rules() + ( Rule("deny-friday-releases", Effect.DENY, actions=("payments.release",), condition=lambda r: r.environment.tick % 7 == 0, reason="release freeze window"),), created_tick=now()).sign(SECRET) print(f" valid bundle offered -> accepted={distributor.offer(good)}, " f"active is now {distributor.active.version}") status = distributor.status() print(f" status: version={status.version} age={status.staleness_ticks} " f"stale={status.stale} hard_stopped={status.hard_stopped}") print() print("=" * 78) print("6. CONTINUOUS AUTHORIZATION") print("=" * 78) continuous = ContinuousAuthorizer(control_plane=plane, now=now, lease_ttl_ticks=30) base = Request(subject, "crm.read", Resource("tool", "crm.read", "wholesale"), Environment(tick=1010)) for tick in (1010, 1015, 1020): continuous.authorize(replace(base, environment=Environment(tick=tick))) print(f" 3 reads within the lease TTL -> evaluations={continuous.evaluations} " f"lease_hits={continuous.lease_hits}") continuous.authorize(replace(base, environment=Environment(tick=1100))) print(f" after the TTL expires -> evaluations={continuous.evaluations}") agents.record_evaluation("payments-investigator", tick=1100, score=0.94) continuous.authorize(replace(base, environment=Environment(tick=1105)), high_impact=True) print(f" a high-impact action -> evaluations={continuous.evaluations} " f"(never leased)") # The kill switch, against a LIVE lease — the case that actually matters. agents.transition("payments-investigator", AgentState.SUSPENDED) decision = continuous.authorize(replace(base, environment=Environment(tick=1112))) print(f" suspended, lease still live -> {decision.effect.value.upper()} " f"(the control plane says no; the lease says yes, and the lease wins)") dropped = continuous.revoke_agent("payments-investigator") decision = continuous.authorize(replace(base, environment=Environment(tick=1113))) print(f" kill switch ({dropped} lease dropped) -> {decision.effect.value.upper()}: " f"{decision.reason}") print(" -> a task running for an hour outlives the conditions that admitted it,") print(" and a suspension that waits out a TTL is a suspension with a latency") print(" you will have to state to an examiner.") continuous.reinstate_agent("payments-investigator") agents.transition("payments-investigator", AgentState.ACTIVE) print() print("=" * 78) print("7. EVALUATION AS AN AUTHORIZATION INPUT") print("=" * 78) pipeline = EvaluationPipeline(registry=agents, now=now, min_score=0.85) cases = [EvalCase(f"g{i}", "golden", "ok") for i in range(8)] + [ EvalCase("s1", "safety", "refuse"), EvalCase("s2", "safety", "refuse")] def good_agent(case: EvalCase) -> str: return case.expected def leaky_agent(case: EvalCase) -> str: return "ok" if case.kind != "safety" else "leaked" passing = pipeline.run("payments-investigator", cases, good_agent) print(f" passing run: score={passing.score:.2f} " f"safety_failures={passing.safety_failures} " f"-> {pipeline.gate(passing)}") failing = pipeline.run("payments-investigator", cases, leaky_agent) verdict = pipeline.gate(failing, baseline=passing) print(f" leaky run : score={failing.score:.2f} " f"safety_failures={failing.safety_failures}") print(f" verdict : promoted={verdict.promoted} reasons={list(verdict.reasons)}") print(" -> a safety failure is disqualifying REGARDLESS of score. An aggregate") print(" that averages away a safety failure is a gate that does not gate.") print() print("=" * 78) print("8. TRACING AND LINEAGE AT AGENT AND TOOL GRANULARITY") print("=" * 78) tracer = Tracer(now=now) trace_id = "trace-88" # Every span's start comes from the SAME injected clock the tracer ends it with. # Two clocks is how you get negative durations in production, and it is always # embarrassing. started = now() agents.record_evaluation("payments-investigator", tick=started, score=0.94) root = tracer.record(trace_id=trace_id, parent_id=None, name="investigate-PMT-771", kind="agent", subject=subject, started_at=started, model_version="gpt-frontier-2026-02-11") started = now() step = Request(subject, "payments.lookup", Resource("tool", "payments.lookup", "wholesale", "confidential"), Environment(tick=started)) decision = plane.authorize(step) tracer.record(trace_id=trace_id, parent_id=root.span_id, name="policy.evaluate", kind="policy", subject=subject, started_at=started, policy_version=decision.policy_version, decision=f"{decision.effect.value}:{decision.rule_name}") started = now() tracer.record(trace_id=trace_id, parent_id=root.span_id, name="payments.lookup", kind="tool", subject=subject, started_at=started, policy_version=decision.policy_version, result="HELD") started = now() tracer.record(trace_id=trace_id, parent_id=root.span_id, name="model.complete", kind="model", subject=subject, started_at=started, model_version="gpt-frontier-2026-02-11", tokens="412") for span in tracer.trace(trace_id): parent = span.parent_id or "-" print(f" {span.span_id:<8} parent={parent:<8} {span.kind:<7} {span.name:<22} " f"dur={span.duration}") print() for key, value in tracer.lineage(trace_id).items(): print(f" {key:<18} {value}") print() print(" That dict is what an examiner's question reduces to — and it is answerable") print(" only because every span carries identity, policy version and model version.") if __name__ == "__main__": # pragma: no cover main()