headroom/tests/test_compression_fidelity_regression.py

"""Offline fidelity regression gate (recall-based, zero-model).

Compresses vendored golden tool-output fixtures through SmartCrusher's lossy
path and asserts that the evidence a model needs to answer each case's question
survives compression. Scoring is pure stdlib (``headroom.evals.metrics``) — no
ML model, no network, no API keys — so this runs in the standard ``[dev]`` CI
shard as a blocking PR check.

A failure here means a code change made lossy compression silently drop
information that answers a known question. Fixtures and the committed baseline
are generated by ``tests/fixtures/fidelity_golden/_generate.py``.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from headroom.evals.metrics import compute_information_recall
from headroom.transforms.smart_crusher import SmartCrusherConfig, smart_crush_tool_output

FIXTURE_DIR = Path(__file__).parent / "fixtures" / "fidelity_golden"
CASES: list[dict] = json.loads((FIXTURE_DIR / "cases.json").read_text())
BASELINE: dict = json.loads((FIXTURE_DIR / "baseline.json").read_text())


def _compress(case: dict) -> tuple[str, str]:
    """Compress a case's tool output via the lossy SmartCrusher path (no model)."""
    original = json.dumps(case["content"])
    cfg = SmartCrusherConfig(max_items_after_crush=case["compress"]["max_items_after_crush"])
    crushed, _modified, _info = smart_crush_tool_output(
        original, cfg, with_compaction=case["compress"]["with_compaction"]
    )
    return original, crushed


@pytest.mark.parametrize("case", CASES, ids=[c["id"] for c in CASES])
def test_critical_evidence_survives_compression(case: dict) -> None:
    """Every ``answer_evidence`` string MUST survive lossy compression (recall == 1.0).

    Critical evidence lives in error/anomaly rows, which SmartCrusher formally
    guarantees to retain (see ``tests/test_quality_retention.py``).
    """
    original, crushed = _compress(case)
    result = compute_information_recall(original, crushed, case["answer_evidence"])
    assert result["recall"] == 1.0, (
        f"FIDELITY REGRESSION in '{case['id']}': compression dropped evidence "
        f"needed to answer {case['question']!r}. Lost: {result['facts_lost']}"
    )


def test_aggregate_recall_not_regressed() -> None:
    """Mean recall over all evidence must not fall below the committed baseline.

    Catches softer regressions (e.g. relevant-but-non-critical context being
    dropped more aggressively) that the per-case critical gate would not.
    """
    recalls = []
    for case in CASES:
        original, crushed = _compress(case)
        probes = case["answer_evidence"] + case["supporting_facts"]
        recalls.append(compute_information_recall(original, crushed, probes)["recall"])

    mean_recall = sum(recalls) / len(recalls)
    floor = BASELINE["aggregate_recall"] - BASELINE["tolerance"]
    assert mean_recall >= floor, (
        f"FIDELITY REGRESSION: mean recall {mean_recall:.4f} fell below baseline "
        f"floor {floor:.4f} (baseline {BASELINE['aggregate_recall']} - tolerance "
        f"{BASELINE['tolerance']}). If this drop is intended, regenerate the baseline."
    )