Release portable deterministic core 1.1.0: frozen formats, offline recovery and regression checks
ecad52f verified | from __future__ import annotations | |
| import hashlib | |
| import json | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from enum import Enum | |
| from typing import Iterable, Mapping, Sequence | |
| from .runtime_provenance import RuntimeProvenance, current_runtime_provenance | |
| ALGORITHM_VERSION = "2.1.2" | |
| class ClaimType(str, Enum): | |
| UNCLASSIFIED = "unclassified" | |
| OBSERVATION = "observation" | |
| INFERENCE = "inference" | |
| ESTIMATE = "estimate" | |
| ACTION_STATE = "action_state" | |
| RISK_SENSITIVE = "risk_sensitive" | |
| class Classification(str, Enum): | |
| FACT = "FACT" | |
| INFERENCE = "INFERENCE" | |
| ESTIMATE = "ESTIMATE" | |
| UNKNOWN = "UNKNOWN" | |
| REFUTED = "REFUTED" | |
| CONFLICT = "CONFLICT" | |
| class Verdict(str, Enum): | |
| APPROVE = "APPROVE" | |
| BLOCK = "BLOCK" | |
| ESCALATE = "ESCALATE" | |
| class RiskLevel(str, Enum): | |
| LOW = "low" | |
| MEDIUM = "medium" | |
| HIGH = "high" | |
| class Directness(str, Enum): | |
| DIRECT = "direct" | |
| INDIRECT = "indirect" | |
| class SourceRole(str, Enum): | |
| EXECUTOR = "executor" | |
| VERIFIER = "verifier" | |
| EXTERNAL = "external" | |
| class EvidenceOrigin(str, Enum): | |
| EXTERNAL_SOURCE = "external_source" | |
| TOOL_OBSERVATION = "tool_observation" | |
| COMPUTED_RESULT = "computed_result" | |
| MEMORY = "memory" | |
| USER_ASSERTION = "user_assertion" | |
| USER_SELF_REPORT = "user_self_report" | |
| USER_PREFERENCE = "user_preference" | |
| USER_INSTRUCTION = "user_instruction" | |
| class AuthorityScope(str, Enum): | |
| EXTERNAL_FACT = "external_fact" | |
| SYSTEM_STATE = "system_state" | |
| USER_SELF_REPORT = "user_self_report" | |
| USER_PREFERENCE = "user_preference" | |
| USER_INSTRUCTION = "user_instruction" | |
| class AnswerabilityStatus(str, Enum): | |
| ANSWERABLE = "answerable" | |
| PARTIALLY_ANSWERABLE = "partially_answerable" | |
| UNDERSPECIFIED = "underspecified" | |
| FALSE_PREMISE = "false_premise" | |
| CONTRADICTORY_INPUT = "contradictory_input" | |
| NEEDS_CLARIFICATION = "needs_clarification" | |
| OUT_OF_SCOPE = "out_of_scope" | |
| class Claim: | |
| claim_id: str | |
| text: str | |
| claim_type: ClaimType | |
| risk: RiskLevel = RiskLevel.LOW | |
| consequential: bool = True | |
| authority_scope: AuthorityScope = AuthorityScope.EXTERNAL_FACT | |
| answerability: AnswerabilityStatus = AnswerabilityStatus.ANSWERABLE | |
| class Evidence: | |
| source_id: str | |
| supports: bool | |
| observed_at: datetime | |
| directness: Directness | |
| post_action: bool = False | |
| eligible: bool = True | |
| evidence_id: str | None = None | |
| source_role: SourceRole = SourceRole.EXTERNAL | |
| source_ref: str | None = None | |
| independence_group: str | None = None | |
| content_sha256: str | None = None | |
| origin: EvidenceOrigin = EvidenceOrigin.EXTERNAL_SOURCE | |
| authority_scope: AuthorityScope = AuthorityScope.EXTERNAL_FACT | |
| class Policy: | |
| version: str | |
| max_age_seconds: int | |
| reject_future_evidence: bool = True | |
| require_independent_action_evidence: bool = True | |
| medium_risk_min_independent_sources: int = 2 | |
| require_runtime_provenance: bool = False | |
| class VerificationResult: | |
| claim: Claim | |
| classification: Classification | |
| verdict: Verdict | |
| reason: str | |
| receipt: dict | |
| failure_record: dict | None | |
| class CompoundResult: | |
| results: tuple[VerificationResult, ...] | |
| verdict: Verdict | |
| def _is_aware(value: datetime) -> bool: | |
| return value.tzinfo is not None and value.utcoffset() is not None | |
| def _effective_evidence_id(item: Evidence) -> str: | |
| return item.evidence_id or item.source_id | |
| def _validate_human_approval(value: bool, *, field_name: str = "human_approved") -> None: | |
| if type(value) is not bool: | |
| raise TypeError(f"{field_name} must be bool") | |
| def _validate_non_empty_string(value: object, field_name: str) -> None: | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ValueError(f"{field_name} must be non-empty string") | |
| def _validate_provenance_inputs( | |
| claim: Claim, evidence: Sequence[Evidence], policy: Policy | |
| ) -> None: | |
| _validate_non_empty_string(claim.claim_id, "claim_id") | |
| _validate_non_empty_string(claim.text, "claim.text") | |
| _validate_non_empty_string(policy.version, "policy.version") | |
| if not isinstance(claim.authority_scope, AuthorityScope): | |
| raise TypeError("claim.authority_scope must be AuthorityScope") | |
| if not isinstance(claim.answerability, AnswerabilityStatus): | |
| raise TypeError("claim.answerability must be AnswerabilityStatus") | |
| for item in evidence: | |
| _validate_non_empty_string(item.source_id, "evidence.source_id") | |
| if not isinstance(item.origin, EvidenceOrigin): | |
| raise TypeError("evidence.origin must be EvidenceOrigin") | |
| if not isinstance(item.authority_scope, AuthorityScope): | |
| raise TypeError("evidence.authority_scope must be AuthorityScope") | |
| def _validate_common_inputs( | |
| evidence: Sequence[Evidence], *, policy: Policy, evaluation_time: datetime | |
| ) -> None: | |
| if not _is_aware(evaluation_time): | |
| raise ValueError("evaluation_time must be timezone-aware") | |
| if policy.max_age_seconds < 0: | |
| raise ValueError("max_age_seconds must be non-negative") | |
| if policy.medium_risk_min_independent_sources < 1: | |
| raise ValueError("medium_risk_min_independent_sources must be at least 1") | |
| if type(policy.require_runtime_provenance) is not bool: | |
| raise TypeError("require_runtime_provenance must be bool") | |
| for item in evidence: | |
| if not _is_aware(item.observed_at): | |
| raise ValueError("observed_at must be timezone-aware") | |
| if item.evidence_id is not None and not item.evidence_id.strip(): | |
| raise ValueError("evidence_id must be non-empty when supplied") | |
| if item.source_ref is not None and not item.source_ref.strip(): | |
| raise ValueError("source_ref must be non-empty when supplied") | |
| if item.independence_group is not None and not item.independence_group.strip(): | |
| raise ValueError("independence_group must be non-empty when supplied") | |
| if item.content_sha256 is not None: | |
| digest = item.content_sha256.lower() | |
| if len(digest) != 64 or any(ch not in "0123456789abcdef" for ch in digest): | |
| raise ValueError("content_sha256 must be a 64-character hexadecimal SHA-256") | |
| def _validate_evidence_gate_types(evidence: Sequence[Evidence]) -> None: | |
| for item in evidence: | |
| if type(item.supports) is not bool: | |
| raise TypeError("evidence.supports must be bool") | |
| if type(item.post_action) is not bool: | |
| raise TypeError("evidence.post_action must be bool") | |
| if type(item.eligible) is not bool: | |
| raise TypeError("evidence.eligible must be bool") | |
| def _validate_evidence_identity(evidence: Sequence[Evidence]) -> None: | |
| ids = [_effective_evidence_id(item) for item in evidence] | |
| if len(ids) != len(set(ids)): | |
| raise ValueError( | |
| "duplicate evidence_id; duplicate source_id fallback identity is not allowed" | |
| ) | |
| def _is_fresh(item: Evidence, *, policy: Policy, evaluation_time: datetime) -> bool: | |
| age = (evaluation_time - item.observed_at).total_seconds() | |
| if policy.reject_future_evidence and age < 0: | |
| return False | |
| return 0 <= age <= policy.max_age_seconds | |
| def _eligible_evidence( | |
| evidence: Sequence[Evidence], *, policy: Policy, evaluation_time: datetime | |
| ) -> tuple[Evidence, ...]: | |
| return tuple( | |
| item | |
| for item in evidence | |
| if item.eligible and _is_fresh(item, policy=policy, evaluation_time=evaluation_time) | |
| ) | |
| _USER_ORIGINS = { | |
| EvidenceOrigin.USER_ASSERTION, | |
| EvidenceOrigin.USER_SELF_REPORT, | |
| EvidenceOrigin.USER_PREFERENCE, | |
| EvidenceOrigin.USER_INSTRUCTION, | |
| } | |
| def _authority_allows(claim: Claim, item: Evidence) -> bool: | |
| if item.authority_scope is not claim.authority_scope: | |
| return False | |
| if claim.authority_scope in {AuthorityScope.EXTERNAL_FACT, AuthorityScope.SYSTEM_STATE}: | |
| return item.origin in { | |
| EvidenceOrigin.EXTERNAL_SOURCE, | |
| EvidenceOrigin.TOOL_OBSERVATION, | |
| EvidenceOrigin.COMPUTED_RESULT, | |
| } | |
| if claim.authority_scope is AuthorityScope.USER_PREFERENCE: | |
| return item.origin in { | |
| EvidenceOrigin.USER_PREFERENCE, | |
| EvidenceOrigin.USER_SELF_REPORT, | |
| EvidenceOrigin.TOOL_OBSERVATION, | |
| } | |
| if claim.authority_scope is AuthorityScope.USER_INSTRUCTION: | |
| return item.origin in { | |
| EvidenceOrigin.USER_INSTRUCTION, | |
| EvidenceOrigin.TOOL_OBSERVATION, | |
| } | |
| if claim.authority_scope is AuthorityScope.USER_SELF_REPORT: | |
| return item.origin in { | |
| EvidenceOrigin.USER_SELF_REPORT, | |
| EvidenceOrigin.TOOL_OBSERVATION, | |
| } | |
| return False | |
| def _authorized_evidence( | |
| claim: Claim, evidence: Sequence[Evidence] | |
| ) -> tuple[Evidence, ...]: | |
| return tuple(item for item in evidence if _authority_allows(claim, item)) | |
| def _classify(claim: Claim, eligible: Sequence[Evidence]) -> Classification: | |
| if claim.answerability is AnswerabilityStatus.CONTRADICTORY_INPUT: | |
| return Classification.CONFLICT | |
| if claim.answerability is not AnswerabilityStatus.ANSWERABLE: | |
| return Classification.UNKNOWN | |
| if claim.claim_type is ClaimType.UNCLASSIFIED: | |
| return Classification.UNKNOWN | |
| supporting = tuple(item for item in eligible if item.supports) | |
| contradicting = tuple(item for item in eligible if not item.supports) | |
| if supporting and contradicting: | |
| return Classification.CONFLICT | |
| if contradicting: | |
| return Classification.REFUTED | |
| if not supporting: | |
| return Classification.UNKNOWN | |
| if claim.claim_type is ClaimType.INFERENCE: | |
| return Classification.INFERENCE | |
| if claim.claim_type is ClaimType.ESTIMATE: | |
| return Classification.ESTIMATE | |
| if not any(item.directness is Directness.DIRECT for item in supporting): | |
| return Classification.INFERENCE | |
| return Classification.FACT | |
| def _has_direct_post_action_evidence(eligible: Sequence[Evidence]) -> bool: | |
| return any( | |
| item.supports | |
| and item.post_action | |
| and item.directness is Directness.DIRECT | |
| for item in eligible | |
| ) | |
| def _has_independent_post_action_evidence(eligible: Sequence[Evidence]) -> bool: | |
| return any( | |
| item.supports | |
| and item.post_action | |
| and item.directness is Directness.DIRECT | |
| and item.source_role in {SourceRole.VERIFIER, SourceRole.EXTERNAL} | |
| for item in eligible | |
| ) | |
| def _independent_supporting_source_count(eligible: Sequence[Evidence]) -> int: | |
| # Extra metadata may join known identities, never split the same provenance. | |
| # Connected components preserve overlap transitively, regardless of input order. | |
| parents: list[int] = [] | |
| seen: dict[tuple[str, str], int] = {} | |
| def root(index: int) -> int: | |
| while parents[index] != index: | |
| parents[index] = parents[parents[index]] | |
| index = parents[index] | |
| return index | |
| for item in eligible: | |
| if not ( | |
| item.supports | |
| and item.directness is Directness.DIRECT | |
| and item.source_role in {SourceRole.VERIFIER, SourceRole.EXTERNAL} | |
| ): | |
| continue | |
| index = len(parents) | |
| parents.append(index) | |
| identities = [("source", item.source_id.strip())] | |
| if item.source_ref is not None: | |
| identities.append(("ref", item.source_ref.strip())) | |
| if item.independence_group is not None: | |
| identities.append(("group", item.independence_group.strip())) | |
| if item.content_sha256 is not None: | |
| identities.append(("content", item.content_sha256.lower())) | |
| for identity in identities: | |
| previous = seen.setdefault(identity, index) | |
| parents[root(index)] = root(previous) | |
| return len({root(index) for index in range(len(parents))}) | |
| def _decide_verdict( | |
| claim: Claim, | |
| classification: Classification, | |
| eligible: Sequence[Evidence], | |
| *, | |
| policy: Policy, | |
| human_approved: bool, | |
| authority_rejected: bool = False, | |
| runtime_provenance_verified: bool = False, | |
| ) -> tuple[Verdict, str]: | |
| if policy.require_runtime_provenance and not runtime_provenance_verified: | |
| return Verdict.BLOCK, "runtime_provenance_unverified" | |
| if claim.answerability is AnswerabilityStatus.NEEDS_CLARIFICATION: | |
| return Verdict.ESCALATE, "answerability_needs_clarification" | |
| if claim.answerability is not AnswerabilityStatus.ANSWERABLE: | |
| return Verdict.BLOCK, f"answerability_{claim.answerability.value}" | |
| if claim.claim_type is ClaimType.UNCLASSIFIED: | |
| return Verdict.BLOCK, "unclassified_claim" | |
| if classification is Classification.CONFLICT: | |
| return Verdict.BLOCK, "eligible_contradiction" | |
| if classification is Classification.REFUTED: | |
| return Verdict.BLOCK, "claim_refuted_by_eligible_evidence" | |
| if classification is Classification.UNKNOWN: | |
| if authority_rejected: | |
| return Verdict.BLOCK, "insufficient_authorized_evidence" | |
| return Verdict.BLOCK, "insufficient_eligible_evidence" | |
| if classification is Classification.INFERENCE: | |
| return Verdict.BLOCK, "inference_not_fact" | |
| if classification is Classification.ESTIMATE: | |
| return Verdict.BLOCK, "estimate_not_fact" | |
| if claim.claim_type is ClaimType.ACTION_STATE: | |
| if not _has_direct_post_action_evidence(eligible): | |
| return Verdict.BLOCK, "missing_post_action_final_state_evidence" | |
| if ( | |
| policy.require_independent_action_evidence | |
| and not _has_independent_post_action_evidence(eligible) | |
| ): | |
| return Verdict.BLOCK, "missing_independent_post_action_evidence" | |
| if ( | |
| claim.risk is RiskLevel.MEDIUM | |
| and _independent_supporting_source_count(eligible) | |
| < policy.medium_risk_min_independent_sources | |
| ): | |
| return Verdict.ESCALATE, "additional_independent_evidence_required" | |
| if claim.risk is RiskLevel.HIGH and not human_approved: | |
| return Verdict.ESCALATE, "human_approval_required" | |
| return Verdict.APPROVE, "fact_with_required_gates" | |
| def _iso(value: datetime) -> str: | |
| return value.isoformat() | |
| def _make_failure_record(claim: Claim, verdict: Verdict, reason: str) -> dict | None: | |
| if verdict is Verdict.APPROVE: | |
| return None | |
| return { | |
| "record_type": "expected_gate", | |
| "claim_id": claim.claim_id, | |
| "failure_mode": reason, | |
| "root_cause": "verification_gate_not_satisfied", | |
| "guardrail": f"preserve_{verdict.value.lower()}_until_new_eligible_evidence_or_required_approval", | |
| "regression_test_required": False, | |
| "version_change_required": False, | |
| } | |
| def make_failure_evolution_record( | |
| *, | |
| failure_mode: str, | |
| root_cause: str, | |
| guardrail: str, | |
| regression_test: str, | |
| ) -> dict: | |
| fields = { | |
| "failure_mode": failure_mode, | |
| "root_cause": root_cause, | |
| "guardrail": guardrail, | |
| "regression_test": regression_test, | |
| } | |
| for name, value in fields.items(): | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ValueError(f"{name} must be non-empty") | |
| return { | |
| "record_type": "failure_evolution", | |
| "failure_mode": failure_mode.strip(), | |
| "root_cause": root_cause.strip(), | |
| "guardrail": guardrail.strip(), | |
| "regression_test": regression_test.strip(), | |
| "regression_test_required": True, | |
| "version_change_required": True, | |
| } | |
| def canonical_receipt_sha256(receipt: Mapping) -> str: | |
| payload = { | |
| key: value | |
| for key, value in receipt.items() | |
| if key != "receipt_sha256" | |
| } | |
| canonical = json.dumps( | |
| payload, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| ensure_ascii=False, | |
| ).encode("utf-8") | |
| return hashlib.sha256(canonical).hexdigest() | |
| def _make_receipt( | |
| *, | |
| claim: Claim, | |
| policy: Policy, | |
| evaluation_time: datetime, | |
| all_evidence: Sequence[Evidence], | |
| eligible: Sequence[Evidence], | |
| classification: Classification, | |
| verdict: Verdict, | |
| reason: str, | |
| human_approved: bool, | |
| authority_rejected: Sequence[Evidence] = (), | |
| runtime_provenance: RuntimeProvenance | None = None, | |
| ) -> dict: | |
| eligible_ids = {_effective_evidence_id(item) for item in eligible} | |
| sorted_all = sorted( | |
| all_evidence, | |
| key=lambda item: (_effective_evidence_id(item), item.source_id), | |
| ) | |
| supporting_sources = sorted(item.source_id for item in eligible if item.supports) | |
| contradicting_sources = sorted(item.source_id for item in eligible if not item.supports) | |
| supporting_evidence_ids = sorted( | |
| _effective_evidence_id(item) for item in eligible if item.supports | |
| ) | |
| contradicting_evidence_ids = sorted( | |
| _effective_evidence_id(item) for item in eligible if not item.supports | |
| ) | |
| if claim.risk is RiskLevel.HIGH: | |
| human_gate = "approved" if human_approved else "required" | |
| else: | |
| human_gate = "not_required" | |
| receipt = { | |
| "algorithm_version": ALGORITHM_VERSION, | |
| "policy_version": policy.version, | |
| "evaluation_time": _iso(evaluation_time), | |
| "claim": { | |
| "id": claim.claim_id, | |
| "text": claim.text, | |
| "type": claim.claim_type.value, | |
| "risk": claim.risk.value, | |
| "consequential": claim.consequential, | |
| "authority_scope": claim.authority_scope.value, | |
| "answerability": claim.answerability.value, | |
| }, | |
| "classification": classification.value, | |
| "verdict": verdict.value, | |
| "reason": reason, | |
| "human_approved": human_approved, | |
| "runtime_provenance": ( | |
| runtime_provenance.as_dict() | |
| if runtime_provenance is not None | |
| else current_runtime_provenance().as_dict() | |
| ), | |
| "gates": { | |
| "eligible_source_count": len(eligible), | |
| "supporting_sources": supporting_sources, | |
| "contradicting_sources": contradicting_sources, | |
| "supporting_evidence_ids": supporting_evidence_ids, | |
| "contradicting_evidence_ids": contradicting_evidence_ids, | |
| "contradiction_count": len(contradicting_evidence_ids), | |
| "authority_rejected_evidence_ids": sorted( | |
| _effective_evidence_id(item) for item in authority_rejected | |
| ), | |
| "post_action_verified": ( | |
| _has_direct_post_action_evidence(eligible) | |
| if claim.claim_type is ClaimType.ACTION_STATE | |
| else None | |
| ), | |
| "independent_post_action_verified": ( | |
| _has_independent_post_action_evidence(eligible) | |
| if claim.claim_type is ClaimType.ACTION_STATE | |
| else None | |
| ), | |
| "independent_supporting_source_count": _independent_supporting_source_count(eligible), | |
| "independent_supporting_group_count": _independent_supporting_source_count(eligible), | |
| "medium_risk_required_sources": ( | |
| policy.medium_risk_min_independent_sources | |
| if claim.risk is RiskLevel.MEDIUM | |
| else None | |
| ), | |
| "human_gate": human_gate, | |
| "runtime_provenance_required": policy.require_runtime_provenance, | |
| }, | |
| "evidence": [ | |
| { | |
| "evidence_id": _effective_evidence_id(item), | |
| "source_id": item.source_id, | |
| "source_role": item.source_role.value, | |
| "source_ref": item.source_ref, | |
| "independence_group": item.independence_group, | |
| "content_sha256": ( | |
| item.content_sha256.lower() | |
| if item.content_sha256 is not None | |
| else None | |
| ), | |
| "origin": item.origin.value, | |
| "authority_scope": item.authority_scope.value, | |
| "supports": item.supports, | |
| "observed_at": _iso(item.observed_at), | |
| "directness": item.directness.value, | |
| "post_action": item.post_action, | |
| "caller_eligible": item.eligible, | |
| "eligible_at_evaluation": _effective_evidence_id(item) in eligible_ids, | |
| } | |
| for item in sorted_all | |
| ], | |
| } | |
| receipt["receipt_sha256"] = canonical_receipt_sha256(receipt) | |
| return receipt | |
| def verify_claim( | |
| claim: Claim, | |
| evidence: Iterable[Evidence], | |
| *, | |
| policy: Policy, | |
| evaluation_time: datetime, | |
| human_approved: bool = False, | |
| ) -> VerificationResult: | |
| _validate_human_approval(human_approved) | |
| evidence_seq = tuple(evidence) | |
| _validate_provenance_inputs(claim, evidence_seq, policy) | |
| _validate_evidence_gate_types(evidence_seq) | |
| _validate_evidence_identity(evidence_seq) | |
| _validate_common_inputs(evidence_seq, policy=policy, evaluation_time=evaluation_time) | |
| fresh_eligible = _eligible_evidence( | |
| evidence_seq, | |
| policy=policy, | |
| evaluation_time=evaluation_time, | |
| ) | |
| eligible = _authorized_evidence(claim, fresh_eligible) | |
| authority_rejected = tuple( | |
| item for item in fresh_eligible if item not in eligible | |
| ) | |
| runtime_provenance = current_runtime_provenance() | |
| classification = _classify(claim, eligible) | |
| verdict, reason = _decide_verdict( | |
| claim, | |
| classification, | |
| eligible, | |
| policy=policy, | |
| human_approved=human_approved, | |
| authority_rejected=bool(authority_rejected), | |
| runtime_provenance_verified=runtime_provenance.verified, | |
| ) | |
| receipt = _make_receipt( | |
| claim=claim, | |
| policy=policy, | |
| evaluation_time=evaluation_time, | |
| all_evidence=evidence_seq, | |
| eligible=eligible, | |
| classification=classification, | |
| verdict=verdict, | |
| reason=reason, | |
| human_approved=human_approved, | |
| authority_rejected=authority_rejected, | |
| runtime_provenance=runtime_provenance, | |
| ) | |
| failure_record = _make_failure_record(claim, verdict, reason) | |
| return VerificationResult( | |
| claim=claim, | |
| classification=classification, | |
| verdict=verdict, | |
| reason=reason, | |
| receipt=receipt, | |
| failure_record=failure_record, | |
| ) | |
| def verify_compound( | |
| claims: Sequence[Claim], | |
| evidence_by_claim: Mapping[str, Iterable[Evidence]], | |
| *, | |
| policy: Policy, | |
| evaluation_time: datetime, | |
| human_approvals: Mapping[str, bool] | None = None, | |
| ) -> CompoundResult: | |
| if not claims: | |
| raise ValueError("at least one claim is required") | |
| claim_ids = [claim.claim_id for claim in claims] | |
| if len(claim_ids) != len(set(claim_ids)): | |
| raise ValueError("duplicate claim_id") | |
| approvals = human_approvals or {} | |
| for approval_claim_id, approval in approvals.items(): | |
| _validate_human_approval( | |
| approval, | |
| field_name=f"human approval for {approval_claim_id}", | |
| ) | |
| results = tuple( | |
| verify_claim( | |
| claim, | |
| evidence_by_claim.get(claim.claim_id, ()), | |
| policy=policy, | |
| evaluation_time=evaluation_time, | |
| human_approved=approvals.get(claim.claim_id, False), | |
| ) | |
| for claim in claims | |
| ) | |
| consequential = [result for result in results if result.claim.consequential] | |
| if any(result.verdict is Verdict.ESCALATE for result in consequential): | |
| overall = Verdict.ESCALATE | |
| elif any(result.verdict is not Verdict.APPROVE for result in consequential): | |
| overall = Verdict.BLOCK | |
| else: | |
| overall = Verdict.APPROVE | |
| return CompoundResult(results=results, verdict=overall) |