# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-confirmation-biased-retrieval-from-amplifying-initial-errors.md
# Remnant Warrant:    https://corpusfetch.com/c/what-invariants-prevent-confirmation-biased-retrieval-from-amplifying-initial-errors.warrant.json
# Markdown SHA-256:   0cd200cb63f60dee135bfc4d40b030307c82644b62eef97c0fab5808f86e6ed9
# Code AST SHA-256:   bc84a1fedd0ae15492c40fb537bf0fea0a5f0f83f3babf7e3add21d6924c6323
# Verified Classes:   RetrievalBiasVerdict, RetrievalRound, RetrievalSession, RetrievalBiasPolicy, RetrievalBiasManifest, ConfirmationBiasRetrievalGuard
# Defensive Guards:   0 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class RetrievalBiasVerdict(Enum):
    NO_ITERATIVE_RETRIEVAL = auto()
    INDEPENDENT_RETRIEVAL = auto()
    CONFIRMATION_BIAS_SUSPECTED = auto()
    CONSENSUS_FABRICATION_DETECTED = auto()

@dataclass(frozen=True)
class RetrievalRound:
    round_id: str
    query_neutral: bool
    distinct_sources: frozenset[str]
    disconfirming_probe_run: bool
    context_terms_used: frozenset[str]

@dataclass(frozen=True)
class RetrievalSession:
    session_id: str
    rounds: tuple[RetrievalRound, ...]
    is_iterative: bool

@dataclass(frozen=True)
class RetrievalBiasPolicy:
    min_source_diversity: int
    require_disconfirming_for_critical: bool
    is_critical: bool

@dataclass(frozen=True)
class RetrievalBiasManifest:
    verdict: RetrievalBiasVerdict
    directive: str
    manifest_digest: str

class ConfirmationBiasRetrievalGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_session(
        cls, session: RetrievalSession, policy: RetrievalBiasPolicy
    ) -> RetrievalBiasManifest:
        
        # Gate 1: Not iterative
        if not session.is_iterative or len(session.rounds) <= 1:
            digest = cls._hash(json.dumps({
                "session": session.session_id,
                "type": "SINGLE_SHOT"
            }, sort_keys=True))
            return RetrievalBiasManifest(
                verdict=RetrievalBiasVerdict.NO_ITERATIVE_RETRIEVAL,
                directive=f"PROCEED: Session {session.session_id} not iterative.",
                manifest_digest=digest
            )

        # Gate 2: Disconfirmation requirement
        if policy.is_critical and policy.require_disconfirming_for_critical:
            no_probe = [r.round_id for r in session.rounds if not r.disconfirming_probe_run]
            if len(no_probe) == len(session.rounds):
                return RetrievalBiasManifest(
                    verdict=RetrievalBiasVerdict.CONFIRMATION_BIAS_SUSPECTED,
                    directive=f"HALT: Critical retrieval {session.session_id} lacks disconfirming probes.",
                    manifest_digest=cls._hash(f"NO_DISCONFIRM:{session.session_id}")
                )

        # Gate 3: Source diversity
        all_sources: set[str] = set()
        for round_data in session.rounds:
            all_sources.update(round_data.distinct_sources)

        if len(all_sources) < policy.min_source_diversity:
            return RetrievalBiasManifest(
                verdict=RetrievalBiasVerdict.CONSENSUS_FABRICATION_DETECTED,
                directive=f"FLAG: {len(session.rounds)} rounds but only {len(all_sources)} distinct sources.",
                manifest_digest=cls._hash(f"FABRICATION:{session.session_id}")
            )

        # Gate 4: Query framing
        for round_data in session.rounds:
            if not round_data.query_neutral:
                return RetrievalBiasManifest(
                    verdict=RetrievalBiasVerdict.CONFIRMATION_BIAS_SUSPECTED,
                    directive=f"FLAG: Round {round_data.round_id} has non-neutral framing.",
                    manifest_digest=cls._hash(f"FRAMED:{round_data.round_id}")
                )

        # Gate 5: Independent
        digest = cls._hash(json.dumps({
            "session": session.session_id,
            "rounds": str(len(session.rounds)),
            "sources": tuple(sorted(all_sources))
        }, sort_keys=True))
        
        return RetrievalBiasManifest(
            verdict=RetrievalBiasVerdict.INDEPENDENT_RETRIEVAL,
            directive=f"PROCEED: {len(session.rounds)} rounds across {len(all_sources)} sources.",
            manifest_digest=digest
        )