"""Deterministic safety-case completeness audit for invented feature evidence.""" from dataclasses import asdict, dataclass, fields import hashlib import json import re IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,127}") SHA256 = re.compile(r"[a-f0-9]{64}") EVIDENCE_KINDS = ("test", "monitor", "process") EVIDENCE_STATUSES = ("pass", "fail", "expired") def identity(value): if type(value) is not str or not IDENTITY.fullmatch(value): raise ValueError("exact bounded identity required") return value def bounded_text(value, lower=1, upper=500): if type(value) is not str or not lower <= len(value) <= upper: raise ValueError("exact bounded text required") if value != value.strip() or any(ord(char) < 32 for char in value): raise ValueError("normalized printable text required") return value def integer(value, lower, upper): if type(value) is not int or not lower <= value <= upper: raise ValueError("bounded non-boolean integer required") return value def exact_bool(value): if type(value) is not bool: raise ValueError("exact boolean required") return value def sequence(value, lower, upper): if type(value) not in (tuple, list): raise ValueError("bounded tuple or list required") integer(len(value), lower, upper) return tuple(value) def identities(value, lower, upper): result = sequence(value, lower, upper) for item in result: identity(item) if len(set(result)) != len(result): raise ValueError("duplicate identity") return result def scope_tuple(value): return identities(value, 8, 8) def digest(value): encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode() return hashlib.sha256(encoded).hexdigest() def seal(record): if type(record.content_id) is not str: raise ValueError("content digest requires exact string") data = asdict(record) data.pop("content_id") return digest(data) def validate_record(record, cls): if type(record) is not cls: raise ValueError("concrete frozen record required") try: rebuilt = cls(**{field.name: getattr(record, field.name) for field in fields(cls)}) except (AttributeError, TypeError) as error: raise ValueError("malformed record") from error if record != rebuilt or record.content_id != rebuilt.content_id: raise ValueError("noncanonical or modified record") return rebuilt @dataclass(frozen=True) class SafetyCaseContract: scope: tuple = ( "feature-policy-v1", "hazard-taxonomy-v1", "control-catalog-v1", "evidence-schema-v1", "risk-matrix-v1", "review-policy-v1", "launch-policy-v1", "fixture-v1", ) max_residual_risk: int = 6 maximum_hazards: int = 32 require_independent_review: bool = True content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) integer(self.max_residual_risk, 1, 25) integer(self.maximum_hazards, 1, 256) if exact_bool(self.require_independent_review) is not True: raise ValueError("safety case requires independent review") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("safety-case contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ControlEvidence: scope: tuple evidence_id: str control_id: str kind: str status: str owner: str revision: str content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.evidence_id) identity(self.control_id) if type(self.kind) is not str or self.kind not in EVIDENCE_KINDS: raise ValueError("unknown exact evidence kind") if type(self.status) is not str or self.status not in EVIDENCE_STATUSES: raise ValueError("unknown exact evidence status") identity(self.owner) identity(self.revision) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("control evidence digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class HazardClaim: scope: tuple hazard_id: str harm: str owner: str inherent_likelihood: int inherent_severity: int residual_likelihood: int residual_severity: int control_ids: tuple evidence_ids: tuple launch_condition: str content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.hazard_id) bounded_text(self.harm) identity(self.owner) for value in ( self.inherent_likelihood, self.inherent_severity, self.residual_likelihood, self.residual_severity, ): integer(value, 1, 5) if self.residual_likelihood > self.inherent_likelihood: raise ValueError("residual likelihood cannot exceed inherent likelihood") if self.residual_severity > self.inherent_severity: raise ValueError("residual severity cannot exceed inherent severity") object.__setattr__(self, "control_ids", identities(self.control_ids, 1, 16)) object.__setattr__(self, "evidence_ids", identities(self.evidence_ids, 1, 32)) bounded_text(self.launch_condition) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("hazard claim digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SafetyCase: scope: tuple contract_content_id: str feature_id: str hazards: tuple evidence: tuple independent_reviewer: str review_complete: bool content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) if type(self.contract_content_id) is not str or not SHA256.fullmatch( self.contract_content_id ): raise ValueError("exact lowercase SHA-256 required") identity(self.feature_id) hazards = tuple( validate_record(item, HazardClaim) for item in sequence(self.hazards, 1, 256) ) evidence = tuple( validate_record(item, ControlEvidence) for item in sequence(self.evidence, 1, 512) ) if any(item.scope != self.scope for item in hazards + evidence): raise ValueError("safety-case record outside declared scope") if len({item.hazard_id for item in hazards}) != len(hazards): raise ValueError("duplicate hazard identity") if len({item.evidence_id for item in evidence}) != len(evidence): raise ValueError("duplicate evidence identity") identity(self.independent_reviewer) exact_bool(self.review_complete) object.__setattr__(self, "hazards", hazards) object.__setattr__(self, "evidence", evidence) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("safety case digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SafetyCaseReport: decision: str violations: tuple maximum_residual_risk: int launch_conditions: tuple evidence_id: str claim: str = "LOCAL_SAFETY_CASE_AUDIT_NOT_SAFETY_CERTIFICATION" def audit_safety_case(contract: SafetyCaseContract, case: SafetyCase) -> SafetyCaseReport: """Check declared hazard-control-evidence links without testing the feature.""" contract = validate_record(contract, SafetyCaseContract) case = validate_record(case, SafetyCase) if case.scope != contract.scope or case.contract_content_id != contract.content_id: raise ValueError("case belongs to another safety contract") if len(case.hazards) > contract.maximum_hazards: raise ValueError("case exceeds hazard budget") evidence_by_id = {item.evidence_id: item for item in case.evidence} violations = [] maximum_residual_risk = 0 conditions = [] referenced_evidence = set() for hazard in case.hazards: maximum_residual_risk = max( maximum_residual_risk, hazard.residual_likelihood * hazard.residual_severity, ) conditions.append(f"{hazard.hazard_id}:{hazard.launch_condition}") if hazard.residual_likelihood * hazard.residual_severity > contract.max_residual_risk: violations.append(f"{hazard.hazard_id}:residual-risk") for evidence_id in hazard.evidence_ids: referenced_evidence.add(evidence_id) evidence = evidence_by_id.get(evidence_id) if evidence is None: violations.append(f"{hazard.hazard_id}:missing-evidence:{evidence_id}") continue if evidence.control_id not in hazard.control_ids: violations.append(f"{hazard.hazard_id}:control-mismatch:{evidence_id}") if evidence.status != "pass": violations.append(f"{hazard.hazard_id}:evidence-{evidence.status}:{evidence_id}") covered_controls = { evidence_by_id[evidence_id].control_id for evidence_id in hazard.evidence_ids if evidence_id in evidence_by_id } for control_id in hazard.control_ids: if control_id not in covered_controls: violations.append(f"{hazard.hazard_id}:unsubstantiated-control:{control_id}") for evidence in case.evidence: if evidence.evidence_id not in referenced_evidence: violations.append(f"unlinked-evidence:{evidence.evidence_id}") if contract.require_independent_review and not case.review_complete: violations.append("independent-review-incomplete") decision = "BLOCK" if violations else "READY_WITH_CONDITIONS" evidence_id = digest( { "contract": contract.content_id, "case": case.content_id, "decision": decision, "violations": violations, "maximum_residual_risk": maximum_residual_risk, } ) return SafetyCaseReport( decision, tuple(violations), maximum_residual_risk, tuple(conditions), evidence_id, ) def illustrative_fixture(): contract = SafetyCaseContract() evidence = ( ControlEvidence( contract.scope, "evidence-grounding-suite-v12", "control-grounding-gate", "test", "pass", "eval-team", "dataset-2026-09", ), ControlEvidence( contract.scope, "evidence-escalation-drill-v4", "control-human-escalation", "process", "pass", "support-ops", "drill-2026-q3", ), ) hazards = ( HazardClaim( contract.scope, "hazard-unsupported-guidance", "A generated support answer could present unsupported account guidance as fact.", "support-product", 4, 4, 2, 3, ("control-grounding-gate", "control-human-escalation"), ("evidence-grounding-suite-v12", "evidence-escalation-drill-v4"), "Keep retrieval coverage above the declared release floor and preserve escalation.", ), ) case = SafetyCase( contract.scope, contract.content_id, "support-answer-assistant-v7", hazards, evidence, "risk-reviewer-b", True, ) return contract, case def main(): report = audit_safety_case(*illustrative_fixture()) print("example=illustrative_only") print(f"decision={report.decision}") print(f"maximum_residual_risk={report.maximum_residual_risk}") print(f"hazard_conditions={len(report.launch_conditions)}") print(f"violations={len(report.violations)}") print(f"claim={report.claim}") if __name__ == "__main__": main()