Release portable deterministic core 1.1.0: frozen formats, offline recovery and regression checks
ecad52f verified | from __future__ import annotations | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from enum import Enum | |
| from typing import Iterable | |
| from .engine import ( | |
| Claim, | |
| Classification, | |
| Evidence, | |
| Policy, | |
| VerificationResult, | |
| Verdict, | |
| verify_claim, | |
| ) | |
| GUARD_VERSION = "2.0.0" | |
| class Realism(str, Enum): | |
| POSSIBLE = "possible" | |
| UNCERTAIN = "uncertain" | |
| IMPOSSIBLE = "impossible" | |
| class GuardStage(str, Enum): | |
| SELF_CHECK = "self_check" | |
| VERIFICATION = "verification" | |
| class UnknownSeverity(str, Enum): | |
| LOW = "low" | |
| MEDIUM = "medium" | |
| HIGH = "high" | |
| CRITICAL = "critical" | |
| class UnknownRecord: | |
| description: str | |
| severity: UnknownSeverity = UnknownSeverity.MEDIUM | |
| reviewed: bool = False | |
| resolved: bool = False | |
| blocking: bool = False | |
| resolution_reference: str | None = None | |
| class SelfCheck: | |
| realism: Realism | |
| known: tuple[str, ...] | |
| assumptions: tuple[str, ...] | |
| unknowns: tuple[str, ...] | |
| assumptions_reviewed: bool | |
| unknowns_reviewed: bool | |
| unknown_records: tuple[UnknownRecord, ...] = () | |
| class GuardResult: | |
| stage: GuardStage | |
| classification: Classification | |
| verdict: Verdict | |
| reason: str | |
| verification: VerificationResult | None | |
| receipt: dict | |
| def _validate_self_check(check: SelfCheck) -> None: | |
| if not isinstance(check, SelfCheck): | |
| raise TypeError("self_check must be SelfCheck") | |
| if not isinstance(check.realism, Realism): | |
| raise TypeError("self_check.realism must be Realism") | |
| if type(check.assumptions_reviewed) is not bool: | |
| raise TypeError("assumptions_reviewed must be bool") | |
| if type(check.unknowns_reviewed) is not bool: | |
| raise TypeError("unknowns_reviewed must be bool") | |
| for field_name, values in ( | |
| ("known", check.known), | |
| ("assumptions", check.assumptions), | |
| ("unknowns", check.unknowns), | |
| ): | |
| if not isinstance(values, tuple): | |
| raise TypeError(f"{field_name} must be tuple") | |
| if any(not isinstance(item, str) or not item.strip() for item in values): | |
| raise ValueError(f"{field_name} items must be non-empty strings") | |
| if not isinstance(check.unknown_records, tuple): | |
| raise TypeError("unknown_records must be tuple") | |
| for item in check.unknown_records: | |
| if not isinstance(item, UnknownRecord): | |
| raise TypeError("unknown_records must contain UnknownRecord values") | |
| if not isinstance(item.description, str) or not item.description.strip(): | |
| raise ValueError("unknown record description must be non-empty") | |
| if not isinstance(item.severity, UnknownSeverity): | |
| raise TypeError("unknown record severity must be UnknownSeverity") | |
| if type(item.reviewed) is not bool: | |
| raise TypeError("unknown record reviewed must be bool") | |
| if type(item.resolved) is not bool: | |
| raise TypeError("unknown record resolved must be bool") | |
| if type(item.blocking) is not bool: | |
| raise TypeError("unknown record blocking must be bool") | |
| if item.resolution_reference is not None and ( | |
| not isinstance(item.resolution_reference, str) | |
| or not item.resolution_reference.strip() | |
| ): | |
| raise ValueError("resolution_reference must be non-empty when supplied") | |
| if item.resolved and (item.blocking or item.severity is UnknownSeverity.CRITICAL): | |
| if item.resolution_reference is None: | |
| raise ValueError( | |
| "resolved blocking or critical unknown requires resolution_reference" | |
| ) | |
| def _self_check_payload(check: SelfCheck) -> dict: | |
| return { | |
| "realism": check.realism.value, | |
| "known": list(check.known), | |
| "assumptions": list(check.assumptions), | |
| "unknowns": list(check.unknowns), | |
| "assumptions_reviewed": check.assumptions_reviewed, | |
| "unknowns_reviewed": check.unknowns_reviewed, | |
| "unknown_records": [ | |
| { | |
| "description": item.description, | |
| "severity": item.severity.value, | |
| "reviewed": item.reviewed, | |
| "resolved": item.resolved, | |
| "blocking": item.blocking, | |
| "resolution_reference": item.resolution_reference, | |
| } | |
| for item in check.unknown_records | |
| ], | |
| } | |
| def _self_check_result( | |
| *, | |
| claim: Claim, | |
| check: SelfCheck, | |
| verdict: Verdict, | |
| reason: str, | |
| ) -> GuardResult: | |
| receipt = { | |
| "guard_version": GUARD_VERSION, | |
| "claim_id": claim.claim_id, | |
| "stage": GuardStage.SELF_CHECK.value, | |
| "self_check": _self_check_payload(check), | |
| "verification": None, | |
| "classification": Classification.UNKNOWN.value, | |
| "verdict": verdict.value, | |
| "reason": reason, | |
| } | |
| return GuardResult( | |
| stage=GuardStage.SELF_CHECK, | |
| classification=Classification.UNKNOWN, | |
| verdict=verdict, | |
| reason=reason, | |
| verification=None, | |
| receipt=receipt, | |
| ) | |
| def run_guard_loop( | |
| *, | |
| claim: Claim, | |
| evidence: Iterable[Evidence], | |
| self_check: SelfCheck, | |
| policy: Policy, | |
| evaluation_time: datetime, | |
| human_approved: bool = False, | |
| ) -> GuardResult: | |
| _validate_self_check(self_check) | |
| if self_check.realism is Realism.IMPOSSIBLE: | |
| return _self_check_result( | |
| claim=claim, | |
| check=self_check, | |
| verdict=Verdict.BLOCK, | |
| reason="self_check_impossible", | |
| ) | |
| if self_check.realism is Realism.UNCERTAIN: | |
| return _self_check_result( | |
| claim=claim, | |
| check=self_check, | |
| verdict=Verdict.ESCALATE, | |
| reason="self_check_uncertain", | |
| ) | |
| unresolved_blocking = tuple( | |
| item | |
| for item in self_check.unknown_records | |
| if not item.resolved | |
| and (item.blocking or item.severity is UnknownSeverity.CRITICAL) | |
| ) | |
| if unresolved_blocking: | |
| return _self_check_result( | |
| claim=claim, | |
| check=self_check, | |
| verdict=Verdict.ESCALATE, | |
| reason="blocking_unknown_unresolved", | |
| ) | |
| if not self_check.assumptions_reviewed: | |
| return _self_check_result( | |
| claim=claim, | |
| check=self_check, | |
| verdict=Verdict.ESCALATE, | |
| reason="assumptions_not_reviewed", | |
| ) | |
| if not self_check.unknowns_reviewed: | |
| return _self_check_result( | |
| claim=claim, | |
| check=self_check, | |
| verdict=Verdict.ESCALATE, | |
| reason="unknowns_not_reviewed", | |
| ) | |
| verification = verify_claim( | |
| claim, | |
| evidence, | |
| policy=policy, | |
| evaluation_time=evaluation_time, | |
| human_approved=human_approved, | |
| ) | |
| receipt = { | |
| "guard_version": GUARD_VERSION, | |
| "claim_id": claim.claim_id, | |
| "stage": GuardStage.VERIFICATION.value, | |
| "self_check": _self_check_payload(self_check), | |
| "verification": { | |
| "classification": verification.classification.value, | |
| "verdict": verification.verdict.value, | |
| "reason": verification.reason, | |
| "receipt": verification.receipt, | |
| }, | |
| "classification": verification.classification.value, | |
| "verdict": verification.verdict.value, | |
| "reason": verification.reason, | |
| } | |
| return GuardResult( | |
| stage=GuardStage.VERIFICATION, | |
| classification=verification.classification, | |
| verdict=verification.verdict, | |
| reason=verification.reason, | |
| verification=verification, | |
| receipt=receipt, | |
| ) |