Hallucination Detection & Mitigation

from collections import Counter
from collections.abc import Mapping
from dataclasses import dataclass, replace
from enum import Enum
from types import MappingProxyType

ALLOWED_FIELDS = frozenset({
    "service", "status", "last_event", "last_event_at", "recovery_eta",
})
CANONICAL_STATUSES = frozenset({"investigating", "mitigated", "resolved"})

def require_text(value: object, name: str) -> None:
    if type(value) is not str or not value.strip():
        raise ValueError(f"{name} must be a nonempty string")

@dataclass(frozen=True)
class Evidence:
    source_id: str
    version: str
    facts: Mapping[str, str]

    def __post_init__(self) -> None:
        require_text(self.source_id, "source_id")
        require_text(self.version, "version")
        if not isinstance(self.facts, Mapping):
            raise ValueError("facts must be a field mapping")
        copied = dict(self.facts)
        if not copied or not set(copied) <= ALLOWED_FIELDS:
            raise ValueError("facts must use the declared incident schema")
        for value in copied.values():
            require_text(value, "fact value")
        if "status" in copied and copied["status"] not in CANONICAL_STATUSES:
            raise ValueError("unknown canonical incident status")
        object.__setattr__(self, "facts", MappingProxyType(copied))

@dataclass(frozen=True)
class Claim:
    claim_id: str
    text: str
    field: str
    value: str
    citation_id: str
    citation_version: str

    def __post_init__(self) -> None:
        for name in ("claim_id", "text", "field", "value", "citation_id", "citation_version"):
            require_text(getattr(self, name), name)
        if self.field not in ALLOWED_FIELDS:
            raise ValueError("claim field is outside the declared incident schema")

incident = Evidence(
    source_id="incident-INC-48291",
    version="runbook-feed/2026-05-27T10:00:00Z",
    facts={
        "service": "embedding-api",
        "status": "investigating",
        "last_event": "database failover completed",
        "last_event_at": "May 26 at 08:14 UTC",
    },
)
sources = MappingProxyType({incident.source_id: incident})

draft_claims = [
    Claim("service", "Service: embedding-api.", "service", "embedding-api", incident.source_id, incident.version),
    Claim("event", "Last event: database failover completed.", "last_event", "database failover completed", incident.source_id, incident.version),
    Claim("event_time", "Event time: May 26 at 08:14 UTC.", "last_event_at", "May 26 at 08:14 UTC", incident.source_id, incident.version),
    Claim("eta", "Full recovery is expected by 14:30 UTC.", "recovery_eta", "14:30 UTC", incident.source_id, incident.version),
]

print(f"Evidence version: {incident.version}")
print(f"Available facts: {sorted(incident.facts)}")
print(f"Draft factual claims: {len(draft_claims)}")

class Verdict(str, Enum):
    SUPPORTED = "supported"
    NOT_SUPPORTED = "not_supported"
    CONTRADICTED = "contradicted"
    NO_SOURCE = "no_source"
    VERSION_MISMATCH = "version_mismatch"

@dataclass(frozen=True)
class Verification:
    claim: Claim
    verdict: Verdict
    evidence_version: str | None

def verify_claim(claim: Claim) -> Verification:
    if type(claim) is not Claim:
        raise ValueError("verification requires a validated Claim")
    source = sources.get(claim.citation_id)
    if source is None:
        return Verification(claim, Verdict.NO_SOURCE, None)
    if claim.citation_version != source.version:
        return Verification(claim, Verdict.VERSION_MISMATCH, source.version)
    expected = source.facts.get(claim.field)
    if expected is None:
        return Verification(claim, Verdict.NOT_SUPPORTED, source.version)
    if expected != claim.value:
        # Different canonical, mutually exclusive statuses contradict this snapshot.
        # Other string differences may just be paraphrases or alternate formatting.
        verdict = (
            Verdict.CONTRADICTED
            if claim.field == "status" and claim.value in CANONICAL_STATUSES
            else Verdict.NOT_SUPPORTED
        )
        return Verification(claim, verdict, source.version)
    return Verification(claim, Verdict.SUPPORTED, source.version)

draft_verdicts = [verify_claim(claim) for claim in draft_claims]
resolved_claim = Claim(
    "resolved",
    "Incident INC-48291 is resolved.",
    "status",
    "resolved",
    incident.source_id,
    incident.version,
)

assert [item.verdict for item in draft_verdicts] == [
    Verdict.SUPPORTED,
    Verdict.SUPPORTED,
    Verdict.SUPPORTED,
    Verdict.NOT_SUPPORTED,
]
assert verify_claim(resolved_claim).verdict == Verdict.CONTRADICTED

for result in draft_verdicts + [verify_claim(resolved_claim)]:
    print(f"{result.claim.claim_id:10} {result.verdict.value:15} {result.claim.text}")

def checked_claims(claims: list[Claim] | tuple[Claim, ...]) -> tuple[Claim, ...]:
    if type(claims) not in (list, tuple) or any(type(item) is not Claim for item in claims):
        raise ValueError("claims must be a sequence of validated Claim records")
    copied = tuple(claims)
    if len({item.claim_id for item in copied}) != len(copied):
        raise ValueError("claim IDs must be unique")
    if len({item.field for item in copied}) != len(copied):
        raise ValueError("this single-record grammar permits one claim per field")
    return copied

TEMPLATES = MappingProxyType({
    "service": "Service: {value}.",
    "status": "Recorded status: {value}.",
    "last_event": "Last event: {value}.",
    "last_event_at": "Event time: {value}.",
    "recovery_eta": "Recorded recovery estimate: {value}.",
})

def render_supported(claim: Claim) -> str:
    if verify_claim(claim).verdict != Verdict.SUPPORTED:
        raise ValueError("only verified fields may be rendered")
    return TEMPLATES[claim.field].format(value=claim.value)

def bounded_fallback(failures: list[Verification]) -> str:
    if any(item.verdict == Verdict.CONTRADICTED for item in failures):
        return "The draft conflicts with the incident record. Please review the latest runbook state."
    if any(item.verdict == Verdict.NO_SOURCE for item in failures):
        return "Incident details are unavailable right now."
    if any(item.verdict == Verdict.VERSION_MISMATCH for item in failures):
        return "The cited version is not the admitted snapshot."
    if all(item.claim.field == "recovery_eta" for item in failures):
        return "The admitted record provides no support for that recovery estimate."
    return "Some requested details are not supported by the admitted record."

def safe_answer(claims: list[Claim] | tuple[Claim, ...]) -> dict[str, object]:
    claims = checked_claims(claims)
    if not claims:
        return {"route": "abstain", "answer": "No verifiable claims were supplied.", "blocked_claims": []}
    verdicts = [verify_claim(claim) for claim in claims]
    failures = [item for item in verdicts if item.verdict != Verdict.SUPPORTED]
    if any(item.verdict != Verdict.NOT_SUPPORTED for item in failures):
        return {
            "route": "abstain", "answer": bounded_fallback(failures),
            "blocked_claims": [item.claim.claim_id for item in failures],
        }
    supported_text = " ".join(
        render_supported(item.claim) for item in verdicts if item.verdict == Verdict.SUPPORTED
    )
    if failures:
        return {
            "route": "shorten" if supported_text else "abstain",
            "answer": " ".join(part for part in [supported_text, bounded_fallback(failures)] if part),
            "blocked_claims": [item.claim.claim_id for item in failures],
        }
    return {
        "route": "serve",
        "answer": supported_text,
        "blocked_claims": [],
    }

decision = safe_answer(draft_claims)
conflict_decision = safe_answer([resolved_claim])

assert decision["route"] == "shorten"
assert decision["blocked_claims"] == ["eta"]
assert "14:30" not in str(decision["answer"])
assert conflict_decision["route"] == "abstain"
assert "resolved" not in str(conflict_decision["answer"])

print(f"Route: {decision['route']}")
print(f"Blocked claims: {decision['blocked_claims']}")
print(f"Served answer: {decision['answer']}")
print(f"Contradicted-only route: {conflict_decision['route']}")

smuggled = replace(draft_claims[0], text="Service: embedding-api. Recovery by 14:30 UTC.")
old_version = replace(draft_claims[0], citation_version="runbook-feed/old")
paraphrase = replace(draft_claims[1], value="database failover finished")

print("Smuggled text rendered:", safe_answer([smuggled])["answer"])
print("Old citation:", verify_claim(old_version).verdict.value)
print("Paraphrase:", verify_claim(paraphrase).verdict.value)
print("Empty draft:", safe_answer([])["route"])
print("Mixed contradiction:", safe_answer([draft_claims[0], resolved_claim])["route"])
assert "14:30" not in safe_answer([smuggled])["answer"]
assert verify_claim(old_version).verdict == Verdict.VERSION_MISMATCH
assert verify_claim(paraphrase).verdict == Verdict.NOT_SUPPORTED
assert safe_answer([])["route"] == "abstain"
assert safe_answer([draft_claims[0], resolved_claim])["route"] == "abstain"

@dataclass(frozen=True)
class AnswerCase:
    case_id: str
    claims: tuple[Claim, ...]

    def __post_init__(self) -> None:
        require_text(self.case_id, "case_id")
        object.__setattr__(self, "claims", checked_claims(self.claims))

cases = [
    AnswerCase("clean_update", draft_claims[:3]),
    AnswerCase("invented_eta", draft_claims),
    AnswerCase("wrong_status", [resolved_claim]),
    AnswerCase(
        "unadmitted_source",
        [Claim("service", "Service: embedding-api.", "service", "embedding-api", "missing-feed", incident.version)],
    ),
]

CASE_MANIFEST = MappingProxyType({case.case_id: case for case in cases})
assert len(CASE_MANIFEST) == len(cases)
cases = tuple(cases)

def has_unsafe_claim(case: AnswerCase) -> bool:
    return any(verify_claim(claim).verdict != Verdict.SUPPORTED for claim in case.claims)

baseline_served_unsafe = sum(has_unsafe_claim(case) for case in cases)
verdict_counts = Counter(
    verify_claim(claim).verdict.value
    for case in cases
    for claim in case.claims
)
claim_support_rate = verdict_counts["supported"] / sum(verdict_counts.values())

print(f"Baseline unsafe serves if all drafts ship: {baseline_served_unsafe}/{len(cases)}")
print(f"Claim verdict counts: {dict(verdict_counts)}")
print(f"Claim support rate: {claim_support_rate:.1%}")
assert baseline_served_unsafe == 3
assert claim_support_rate == 2 / 3

from math import log

def disagreement_rate(values: list[str]) -> float:
    if type(values) not in (list, tuple) or not values:
        raise ValueError("at least one sample is required")
    for value in values:
        require_text(value, "sample")
    most_common_count = Counter(values).most_common(1)[0][1]
    return 1 - most_common_count / len(values)

def empirical_entropy(cluster_labels: list[str]) -> float:
    if type(cluster_labels) not in (list, tuple) or not cluster_labels:
        raise ValueError("at least one clustered sample is required")
    for label in cluster_labels:
        require_text(label, "cluster label")
    probabilities = [count / len(cluster_labels) for count in Counter(cluster_labels).values()]
    return -sum(p * log(p) for p in probabilities)

unstable_eta_samples = ["14:30 UTC", "14:45 UTC", "14:30 UTC", "15:00 UTC"]
stable_unsupported_eta_samples = ["14:30 UTC", "14:30 UTC", "14:30 UTC", "14:30 UTC"]
eta_verdict = verify_claim(draft_claims[-1]).verdict

print(f"Unstable recovery estimate disagreement: {disagreement_rate(unstable_eta_samples):.2f}")
print(f"Repeated recovery estimate disagreement: {disagreement_rate(stable_unsupported_eta_samples):.2f}")
print(f"Repeated recovery estimate evidence verdict: {eta_verdict.value}")
print(f"Three-cluster empirical entropy: {empirical_entropy(unstable_eta_samples):.3f} nats")
# These two surface forms express the same time; cluster them manually here.
surface_forms = ["14:30 UTC", "2:30 PM UTC"] * 2
meaning_labels = ["14:30 UTC"] * 4
print(f"Paraphrase string disagreement: {disagreement_rate(surface_forms):.2f}")
print(f"Paraphrase meaning entropy: {abs(empirical_entropy(meaning_labels)):.3f} nats")

assert disagreement_rate(stable_unsupported_eta_samples) == 0.0
assert eta_verdict == Verdict.NOT_SUPPORTED
assert empirical_entropy(meaning_labels) == 0.0

def route_with_signals(case: AnswerCase, sampled_statuses: list[str]) -> str:
    evidence_route = safe_answer(case.claims)["route"]
    if evidence_route == "abstain":
        return "abstain_evidence_failure"
    if evidence_route == "shorten":
        return "shorten_unsupported_detail"
    if not sampled_statuses:
        return "review_unstable_generation"
    if disagreement_rate(sampled_statuses) > 0.25:
        return "review_unstable_generation"
    return "serve_supported_answer"

clean_case = cases[0]
eta_case = cases[1]

routes = {
    "clean_stable": route_with_signals(clean_case, ["investigating"] * 4),
    "clean_unstable": route_with_signals(
        clean_case, ["investigating", "investigating", "mitigated", "resolved"]
    ),
    "eta_repeated": route_with_signals(eta_case, stable_unsupported_eta_samples),
}

for name, route_name in routes.items():
    print(f"{name:14} -> {route_name}")

assert routes["clean_stable"] == "serve_supported_answer"
assert routes["clean_unstable"] == "review_unstable_generation"
assert routes["eta_repeated"] == "shorten_unsupported_detail"

def cited_sentence(claim: Claim) -> str:
    result = verify_claim(claim)
    if result.verdict != Verdict.SUPPORTED:
        raise ValueError(f"cannot cite {claim.claim_id}: {result.verdict.value}")
    return f"{render_supported(claim)} [{claim.citation_id}@{result.evidence_version}]"

served_sentences = [cited_sentence(claim) for claim in draft_claims[:3]]
served_answer = " ".join(served_sentences)

try:
    cited_sentence(draft_claims[-1])
except ValueError as error:
    blocked_citation = str(error)

print(served_answer)
print(blocked_citation)
assert "14:30" not in served_answer
assert "not_supported" in blocked_citation

from math import exp, isclose

choices = ("investigating", "resolved", "unknown")
with_context = (0.70, 0.20, 0.10)
without_context = (0.10, 0.80, 0.10)
raw_difference = [pc - p0 for pc, p0 in zip(with_context, without_context)]

# Log probabilities are valid logits up to an additive normalization constant.
alpha = 1.0
contrast_scores = [
    (1 + alpha) * log(pc) - alpha * log(p0)
    for pc, p0 in zip(with_context, without_context)
]
weights = [exp(score - max(contrast_scores)) for score in contrast_scores]
cad_probabilities = [weight / sum(weights) for weight in weights]

print("Raw difference:", [round(value, 2) for value in raw_difference])
for choice, probability in zip(choices, cad_probabilities):
    print(f"CAD {choice:13} {probability:.3f}")
print(f"Normalized mass: {sum(cad_probabilities):.3f}")
assert any(value < 0 for value in raw_difference)
assert isclose(sum(cad_probabilities), 1.0)
assert cad_probabilities[0] > with_context[0]

@dataclass(frozen=True)
class RunTrace:
    request_id: str
    case: AnswerCase
    evidence_version: str | None
    sampled_statuses: tuple[str, ...]

    def __post_init__(self) -> None:
        require_text(self.request_id, "request_id")
        if type(self.case) is not AnswerCase:
            raise ValueError("trace requires a validated answer case")
        if self.evidence_version is not None:
            require_text(self.evidence_version, "trace evidence version")
            if self.evidence_version != incident.version:
                raise ValueError("trace version must match this lab's host snapshot")
        if type(self.sampled_statuses) not in (list, tuple):
            raise ValueError("sampled outputs must be a sequence")
        for value in self.sampled_statuses:
            require_text(value, "sampled output")
        object.__setattr__(self, "sampled_statuses", tuple(self.sampled_statuses))

def trace_verifications(trace: RunTrace) -> list[Verification]:
    if trace.evidence_version is None:
        return [Verification(claim, Verdict.NO_SOURCE, None) for claim in trace.case.claims]
    return [verify_claim(claim) for claim in trace.case.claims]

def first_failed_stage(trace: RunTrace) -> str:
    if trace.evidence_version is None:
        return "evidence_admission"
    if any(item.verdict != Verdict.SUPPORTED for item in trace_verifications(trace)):
        return "claim_generation"
    if not trace.case.claims:
        return "claim_generation"
    if not trace.sampled_statuses:
        return "stability_check_incomplete"
    if disagreement_rate(trace.sampled_statuses) > 0.25:
        return "generation_stability"
    return "passed"

traces = [
    RunTrace("req-clean", cases[0], incident.version, ["investigating"] * 4),
    RunTrace("req-eta", cases[1], incident.version, stable_unsupported_eta_samples),
    RunTrace("req-missing", cases[3], None, ["unknown"] * 4),
    RunTrace("req-vary", cases[0], incident.version, ["investigating", "resolved", "investigating", "mitigated"]),
]

for trace in traces:
    print(f"{trace.request_id:11} -> {first_failed_stage(trace)}")

assert [first_failed_stage(trace) for trace in traces] == [
    "passed",
    "claim_generation",
    "evidence_admission",
    "generation_stability",
]

def baseline_router(case: AnswerCase) -> dict[str, object]:
    return {"route": "serve", "answer": " ".join(claim.text for claim in case.claims)}

def product_router(case: AnswerCase) -> dict[str, object]:
    return safe_answer(case.claims)

# Hand-authored accepted payloads for this grammar, not an annotation study.
GOLD_UPDATE = (
    "Service: embedding-api. Last event: database failover completed. "
    "Event time: May 26 at 08:14 UTC."
)
GOLD_ANSWERS = MappingProxyType({
    "clean_update": frozenset({GOLD_UPDATE}),
    "invented_eta": frozenset({
        GOLD_UPDATE + " The admitted record provides no support for that recovery estimate."
    }),
    "wrong_status": frozenset(),
    "unadmitted_source": frozenset(),
})
CLEAN_CASE_IDS = frozenset({"clean_update"})
ABSTENTION_TEXTS = frozenset({
    "No verifiable claims were supplied.",
    "The draft conflicts with the incident record. Please review the latest runbook state.",
    "Incident details are unavailable right now.",
    "The cited version is not the admitted snapshot.",
    "The admitted record provides no support for that recovery estimate.",
    "Some requested details are not supported by the admitted record.",
    "Unable to provide a grounded update.",
})

def score_router(router) -> dict[str, float | int | None]:
    # Evaluate every declared case, not a caller-filtered subset.
    suite_cases = tuple(CASE_MANIFEST.values())
    responses = [(case, router(case)) for case in suite_cases]
    for _, response in responses:
        if type(response) is not dict or not {"route", "answer"} <= set(response):
            raise ValueError("router must return route and answer")
        if type(response["answer"]) is not str:
            raise ValueError("answer payload must be a string")
        if type(response["route"]) is not str or response["route"] not in {
            "serve", "shorten", "abstain", "review",
        }:
            raise ValueError("unknown response route")
    # Grade the actual payload. Merely labeling a leaked answer "shorten" is not safe.
    served = [(case, response) for case, response in responses
              if response["answer"] and response["answer"] not in ABSTENTION_TEXTS]
    unsafe_serves = sum(response["answer"] not in GOLD_ANSWERS[case.case_id]
                        for case, response in served)
    supported_serves = sum(case.case_id in CLEAN_CASE_IDS
                           and response["answer"] in GOLD_ANSWERS[case.case_id]
                           for case, response in served)
    return {
        "served": len(served),
        "unsafe_serves": unsafe_serves,
        "unsafe_serve_rate": unsafe_serves / len(served) if served else None,
        "supported_coverage": supported_serves / len(CLEAN_CASE_IDS),
        "response_coverage": len(served) / len(suite_cases),
        "abstentions": len(suite_cases) - len(served),
        "shortens": sum(response["route"] == "shorten" for _, response in responses),
    }

baseline_metrics = score_router(baseline_router)
candidate_metrics = score_router(product_router)

for name, metrics in [("baseline", baseline_metrics), ("candidate", candidate_metrics)]:
    print(
        f"{name:9} unsafe_rate={metrics['unsafe_serve_rate']:.1%} "
        f"clean_coverage={metrics['supported_coverage']:.1%} "
        f"response_coverage={metrics['response_coverage']:.1%} "
        f"shortens={metrics['shortens']} abstentions={metrics['abstentions']}"
    )

assert baseline_metrics["unsafe_serve_rate"] == 0.75
assert candidate_metrics["unsafe_serve_rate"] == 0.0
assert candidate_metrics["supported_coverage"] == 1.0
assert candidate_metrics["shortens"] == 1
assert candidate_metrics["abstentions"] == 2

# Adversarial checks on the scorer itself.
leaky_metrics = score_router(lambda case: {**baseline_router(case), "route": "shorten"})
silent_metrics = score_router(lambda case: {
    "route": "abstain", "answer": "Unable to provide a grounded update."
})
print(f"leaky shorten unsafe_rate={leaky_metrics['unsafe_serve_rate']:.1%}")
print(f"abstain-all unsafe_rate={silent_metrics['unsafe_serve_rate']} (undefined), coverage=0%")
assert leaky_metrics["unsafe_serve_rate"] == 0.75
assert silent_metrics["unsafe_serve_rate"] is None
assert silent_metrics["response_coverage"] == 0

@dataclass(frozen=True)
class MonitorEvent:
    request_id: str
    case_id: str
    evidence_version: str | None
    route: str
    first_failed_stage: str
    verdict_counts: Mapping[str, int]
    claim_verdicts: tuple[tuple[str, str], ...]

def trace_route(trace: RunTrace) -> str:
    if trace.evidence_version is None:
        return "abstain"
    signal_route = route_with_signals(trace.case, trace.sampled_statuses)
    return {
        "abstain_evidence_failure": "abstain",
        "shorten_unsupported_detail": "shorten",
        "review_unstable_generation": "review",
        "serve_supported_answer": "serve",
    }[signal_route]

def monitor_event(trace: RunTrace) -> MonitorEvent:
    results = trace_verifications(trace)
    counts = Counter(item.verdict.value for item in results)
    return MonitorEvent(
        request_id=trace.request_id,
        case_id=trace.case.case_id,
        evidence_version=trace.evidence_version,
        route=trace_route(trace),
        first_failed_stage=first_failed_stage(trace),
        verdict_counts=MappingProxyType(dict(counts)),
        claim_verdicts=tuple((item.claim.claim_id, item.verdict.value) for item in results),
    )

REQUIRED_REQUIREMENTS = (
    "known_unsafe_claims_blocked_and_clean_case_kept",
    "source_version_logged_when_admitted",
    "representative_holdout_evaluated_with_audited_grader",
    "monitoring_alert_owner_assigned",
)

def promotion_decision(records: dict[str, bool]) -> tuple[str, list[str]]:
    if type(records) is not dict or set(records) != set(REQUIRED_REQUIREMENTS):
        raise ValueError("every declared release requirement must be present")
    if any(type(value) is not bool for value in records.values()):
        raise ValueError("release results must be booleans")
    failures = [name for name in REQUIRED_REQUIREMENTS if not records[name]]
    return ("BLOCKED" if failures else "APPROVED"), failures

events = [monitor_event(trace) for trace in traces]
requirements = {
    "known_unsafe_claims_blocked_and_clean_case_kept": (
        candidate_metrics["unsafe_serves"] == 0
        and candidate_metrics["supported_coverage"] == 1.0
    ),
    "source_version_logged_when_admitted": all(
        event.evidence_version is not None
        for event in events
        if event.first_failed_stage != "evidence_admission"
    ),
    "representative_holdout_evaluated_with_audited_grader": False,
    "monitoring_alert_owner_assigned": False,
}
decision, failed_requirements = promotion_decision(requirements)

print(f"Candidate promotion: {decision}")
for name in failed_requirements:
    print(f"  missing: {name}")
print(f"Example failed stage: {events[1].first_failed_stage}")
print(f"Example review route: {events[3].route}")

assert decision == "BLOCKED"
assert [event.route for event in events] == ["serve", "shorten", "abstain", "review"]
assert [event.evidence_version for event in events] == [
    incident.version,
    incident.version,
    None,
    incident.version,
]