"""Executable contract for an inspectable AI system boundary.""" from __future__ import annotations from dataclasses import dataclass from typing import Literal ComponentKind = Literal[ "deterministic", "probabilistic", "data", "human", "external", ] @dataclass(frozen=True) class Component: name: str kind: ComponentKind owner: str contract: str failure_mode: str evidence: str @dataclass(frozen=True) class Flow: source: str target: str payload: str classification: str owner: str contract: str failure_behavior: str auth_context: str evidence: str @dataclass(frozen=True) class SystemBoundary: name: str purpose: str decision_owner: str components: tuple[Component, ...] flows: tuple[Flow, ...] def validate_boundary(boundary: SystemBoundary) -> None: """Fail closed when the map cannot support ownership or failure analysis.""" for field_name in ("name", "purpose", "decision_owner"): if not getattr(boundary, field_name).strip(): raise ValueError(f"{field_name} must be explicit") if not boundary.components: raise ValueError("at least one component is required") names = [component.name for component in boundary.components] if len(names) != len(set(names)): raise ValueError("component names must be unique") allowed_kinds = { "deterministic", "probabilistic", "data", "human", "external", } for component in boundary.components: if component.kind not in allowed_kinds: raise ValueError(f"unknown component kind: {component.kind}") for field_name in ("name", "owner", "contract", "failure_mode", "evidence"): if not getattr(component, field_name).strip(): raise ValueError( f"component {component.name or ''} needs {field_name}" ) if not any(component.kind == "probabilistic" for component in boundary.components): raise ValueError("an AI boundary needs a probabilistic component") if not boundary.flows: raise ValueError("an AI boundary needs at least one flow") known = set(names) adjacency = {name: set() for name in names} seen_flows: set[tuple[str, str, str]] = set() for flow in boundary.flows: if flow.source not in known or flow.target not in known: raise ValueError( f"flow endpoint is outside the map: {flow.source}->{flow.target}" ) if flow.source == flow.target: raise ValueError(f"flow must cross components: {flow.source}") for field_name in ( "payload", "classification", "owner", "contract", "failure_behavior", "auth_context", "evidence", ): value = getattr(flow, field_name) if not isinstance(value, str) or not value.strip(): raise ValueError( f"flow {flow.source}->{flow.target} needs {field_name}" ) identity = (flow.source, flow.target, flow.payload) if identity in seen_flows: raise ValueError(f"duplicate flow: {flow.source}->{flow.target} {flow.payload}") seen_flows.add(identity) adjacency[flow.source].add(flow.target) adjacency[flow.target].add(flow.source) reachable: set[str] = set() pending = [names[0]] while pending: current = pending.pop() if current in reachable: continue reachable.add(current) pending.extend(adjacency[current] - reachable) disconnected = sorted(known - reachable) if disconnected: raise ValueError(f"disconnected components: {', '.join(disconnected)}") def boundary_report(boundary: SystemBoundary) -> str: validate_boundary(boundary) kind_by_name = { component.name: component.kind for component in boundary.components } external_flows = sum( kind_by_name[flow.source] == "external" or kind_by_name[flow.target] == "external" for flow in boundary.flows ) return "\n".join( ( f"boundary={boundary.name}", f"components={len(boundary.components)} flows={len(boundary.flows)}", "probabilistic_components=" + str(sum(component.kind == "probabilistic" for component in boundary.components)), "human_decision_points=" + str(sum(component.kind == "human" for component in boundary.components)), f"external_flows={external_flows}", ) ) EXAMPLE_BOUNDARY = SystemBoundary( name="refund-triage", purpose="Prioritize refund requests; a person authorizes every payment.", decision_owner="Support operations", components=( Component( "customer-request", "external", "Customer support", "Authenticated text request", "Spoofed identity or adversarial content", "Authentication and ingestion audit event", ), Component( "request-parser", "deterministic", "Support platform", "Validated request schema", "Rejects a valid request or accepts malformed input", "Schema validation counters and request trace", ), Component( "policy-snapshot", "data", "Finance policy", "Versioned refund rules", "Stale or inapplicable policy", "Policy version and freshness record", ), Component( "triage-model", "probabilistic", "ML platform", "Score plus model and policy versions", "Misranking or score shift", "Slice metrics and versioned inference trace", ), Component( "operations-reviewer", "human", "Support operations", "Approve, reject, or escalate with reason", "Automation bias or inconsistent judgment", "Decision reason, override, and escalation audit", ), Component( "refund-service", "deterministic", "Payments platform", "Idempotent authorized refund command", "Duplicate or unauthorized payment", "Authorization, idempotency, and payment audit record", ), ), flows=( Flow( "customer-request", "request-parser", "request", "confidential", "Support platform", "RequestEnvelope v1", "Reject malformed input and return a safe error", "Authenticated customer session", "Request ID and validation outcome", ), Flow( "request-parser", "triage-model", "validated features", "confidential", "ML platform", "TriageFeatures v1", "Stop scoring and route to the manual queue", "Tenant-bound service identity", "Trace ID, schema version, and rejection reason", ), Flow( "policy-snapshot", "triage-model", "policy version", "internal", "Finance policy", "RefundPolicySnapshot v1", "Stop scoring when policy is missing or stale", "Read-only workload identity", "Policy version and freshness result", ), Flow( "triage-model", "operations-reviewer", "score and evidence", "confidential", "Support operations", "TriageEvidence v1", "Present an unranked manual queue when scoring fails", "Case-scoped operator role", "Model, policy, and evidence versions", ), Flow( "operations-reviewer", "refund-service", "authorized decision", "restricted", "Payments platform", "AuthorizedRefund v1", "Reject unauthorized work and escalate without side effects", "Operator authorization plus service identity", "Principal, reason, policy version, and idempotency key", ), ), ) if __name__ == "__main__": print(boundary_report(EXAMPLE_BOUNDARY))