Blame for bodulskihal_core/engine.py

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"


@dataclass(frozen=True)
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


@dataclass(frozen=True)
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


@dataclass(frozen=True)
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


@dataclass(frozen=True)
class VerificationResult:
    claim: Claim
    classification: Classification
    verdict: Verdict
    reason: str
    receipt: dict
    failure_record: dict | None


@dataclass(frozen=True)
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)