RAGcore-Public/tests/unit/retrieval/answerability/test_classifier.py

"""AnswerabilityClassifier: deterministic, content-free, non-probabilistic decisions."""

from __future__ import annotations

from uuid import UUID, uuid4

import pytest

from ragcore.retrieval.answerability import (
    MAX_EVIDENCE_ITEMS,
    AnswerabilityClassifier,
    AnswerabilityDecision,
    AnswerabilityReason,
    AnswerabilityStatus,
    AnswerabilityThresholds,
    EvidenceSignal,
)


def make_signal(**changes: object) -> EvidenceSignal:
    value: dict[str, object] = {
        "document_id": uuid4(),
        "source_id": uuid4(),
        "trust_level": "source",
        "fused_score": 0.05,
        "rerank_score": 0.8,
        "covers_distinct_facet": True,
        "is_fresh": True,
        "exact_identifier_match": False,
    }
    value.update(changes)
    return EvidenceSignal(**value)


def default_thresholds(**changes: object) -> AnswerabilityThresholds:
    value: dict[str, object] = {"version": "answerability-v1"}
    value.update(changes)
    return AnswerabilityThresholds(**value)


def test_sufficient_diverse_evidence_is_answerable() -> None:
    evidence = (make_signal(), make_signal())
    decision = AnswerabilityClassifier(default_thresholds()).classify(evidence)

    assert decision.status is AnswerabilityStatus.ANSWERABLE
    assert decision.reasons == (AnswerabilityReason.SUFFICIENT_EVIDENCE,)
    assert decision.evidence_count == 2
    assert decision.document_diversity == 2
    assert decision.source_independence == 2
    assert decision.facet_coverage_ratio == 1.0
    assert decision.thresholds_version == "answerability-v1"


def test_empty_evidence_is_not_answerable() -> None:
    decision = AnswerabilityClassifier(default_thresholds()).classify(())

    assert decision.status is AnswerabilityStatus.NOT_ANSWERABLE
    assert AnswerabilityReason.INSUFFICIENT_EVIDENCE in decision.reasons
    assert decision.evidence_count == 0
    assert decision.document_diversity == 0


def test_insufficient_evidence_count_is_not_answerable() -> None:
    decision = AnswerabilityClassifier(default_thresholds()).classify((make_signal(),))

    assert decision.status is AnswerabilityStatus.NOT_ANSWERABLE
    assert AnswerabilityReason.INSUFFICIENT_EVIDENCE in decision.reasons


def test_low_document_diversity_is_not_answerable() -> None:
    document_id = uuid4()
    evidence = (
        make_signal(document_id=document_id, source_id=uuid4()),
        make_signal(document_id=document_id, source_id=uuid4()),
    )
    decision = AnswerabilityClassifier(default_thresholds()).classify(evidence)

    assert decision.status is AnswerabilityStatus.NOT_ANSWERABLE
    assert AnswerabilityReason.LOW_DOCUMENT_DIVERSITY in decision.reasons
    assert decision.document_diversity == 1


def test_low_source_independence_is_not_answerable() -> None:
    source_id = uuid4()
    evidence = (
        make_signal(document_id=uuid4(), source_id=source_id),
        make_signal(document_id=uuid4(), source_id=source_id),
    )
    decision = AnswerabilityClassifier(default_thresholds()).classify(evidence)

    assert decision.status is AnswerabilityStatus.NOT_ANSWERABLE
    assert AnswerabilityReason.LOW_SOURCE_INDEPENDENCE in decision.reasons
    assert decision.source_independence == 1


def test_below_rerank_threshold_is_not_answerable_when_gating_enabled() -> None:
    thresholds = default_thresholds(min_rerank_score=0.5)
    evidence = (
        make_signal(rerank_score=0.1),
        make_signal(rerank_score=0.2),
    )
    decision = AnswerabilityClassifier(thresholds).classify(evidence)

    assert decision.status is AnswerabilityStatus.NOT_ANSWERABLE
    assert AnswerabilityReason.BELOW_RERANK_THRESHOLD in decision.reasons


def test_rerank_gating_disabled_by_default_ignores_low_or_missing_scores() -> None:
    evidence = (
        make_signal(rerank_score=None),
        make_signal(rerank_score=None),
    )
    decision = AnswerabilityClassifier(default_thresholds()).classify(evidence)

    assert decision.status is AnswerabilityStatus.ANSWERABLE


def test_partial_facet_coverage_is_partially_answerable() -> None:
    evidence = (
        make_signal(covers_distinct_facet=True),
        make_signal(covers_distinct_facet=False),
    )
    decision = AnswerabilityClassifier(default_thresholds()).classify(
        evidence, total_query_facets=2
    )

    assert decision.status is AnswerabilityStatus.PARTIALLY_ANSWERABLE
    assert AnswerabilityReason.PARTIAL_FACET_COVERAGE in decision.reasons
    assert decision.facet_coverage_ratio == 0.5


def test_conflicting_evidence_fixture_is_classified_despite_otherwise_meeting_thresholds() -> None:
    first_document, second_document = uuid4(), uuid4()
    evidence = (
        make_signal(
            document_id=first_document,
            source_id=uuid4(),
            conflicts_with_documents=frozenset({second_document}),
        ),
        make_signal(document_id=second_document, source_id=uuid4()),
    )
    decision = AnswerabilityClassifier(default_thresholds()).classify(evidence)

    assert decision.status is AnswerabilityStatus.CONFLICTING_EVIDENCE
    assert decision.reasons == (AnswerabilityReason.CONFLICTING_SOURCES,)


def test_conflict_pointing_outside_the_current_evidence_set_does_not_trigger() -> None:
    evidence = (make_signal(conflicts_with_documents=frozenset({uuid4()})), make_signal())
    decision = AnswerabilityClassifier(default_thresholds()).classify(evidence)

    assert decision.status is AnswerabilityStatus.ANSWERABLE


def test_decision_never_carries_a_probability_or_confidence_field() -> None:
    decision = AnswerabilityClassifier(default_thresholds()).classify(
        (make_signal(), make_signal())
    )
    field_names = {field for field in decision.__dataclass_fields__}

    assert "confidence" not in field_names
    assert "probability" not in field_names
    assert all(isinstance(reason, AnswerabilityReason) for reason in decision.reasons)


def test_thresholds_version_is_validated_and_propagated() -> None:
    with pytest.raises(ValueError, match="version"):
        AnswerabilityThresholds(version="Not A Safe Version!")

    decision = AnswerabilityClassifier(default_thresholds(version="answerability-v2")).classify(
        (make_signal(), make_signal())
    )
    assert decision.thresholds_version == "answerability-v2"


def test_thresholds_are_bounded() -> None:
    with pytest.raises(ValueError, match="min_evidence_count"):
        default_thresholds(min_evidence_count=0)
    with pytest.raises(ValueError, match="min_facet_coverage_ratio"):
        default_thresholds(min_facet_coverage_ratio=0.0)
    with pytest.raises(ValueError, match="min_facet_coverage_ratio"):
        default_thresholds(min_facet_coverage_ratio=1.5)


def test_evidence_input_is_bounded() -> None:
    classifier = AnswerabilityClassifier(default_thresholds())
    with pytest.raises(ValueError, match="bounded limit"):
        classifier.classify(tuple(make_signal() for _ in range(MAX_EVIDENCE_ITEMS + 1)))
    with pytest.raises(ValueError, match="total_query_facets"):
        classifier.classify((make_signal(),), total_query_facets=0)


def test_evidence_signal_rejects_self_conflict_and_nil_ids() -> None:
    document_id = uuid4()
    with pytest.raises(ValueError, match="own document"):
        make_signal(document_id=document_id, conflicts_with_documents=frozenset({document_id}))
    with pytest.raises(ValueError, match="document_id"):
        make_signal(document_id=UUID(int=0))


def test_evidence_signal_rejects_nonfinite_scores() -> None:
    with pytest.raises(ValueError, match="fused_score"):
        make_signal(fused_score=float("nan"))
    with pytest.raises(ValueError, match="rerank_score"):
        make_signal(rerank_score=float("inf"))


def test_classification_is_deterministic_given_the_same_input() -> None:
    evidence = (make_signal(), make_signal(), make_signal(covers_distinct_facet=False))
    classifier = AnswerabilityClassifier(default_thresholds())

    first = classifier.classify(evidence, total_query_facets=2)
    second = classifier.classify(evidence, total_query_facets=2)

    assert first == second
    assert isinstance(first, AnswerabilityDecision)