from __future__ import annotations import json from pathlib import Path import pytest from scripts.benchmark.deermem_eviction.config import load_evaluation_config from scripts.benchmark.deermem_eviction.policy import PolicyResult from scripts.benchmark.deermem_eviction.pool import PreparedCase from scripts.benchmark.deermem_eviction.report import AnswerRowIntegrityError, collect_answer_rows, compute_qa_statistics, grade_answer_rows, summarize_qa_rows, write_qa_report from scripts.benchmark.deermem_eviction.runner import response_path from scripts.benchmark.deermem_eviction.stats import exact_mcnemar, paired_bootstrap_difference EVAL_ROOT = Path(__file__).parents[1] / "scripts" / "benchmark" / "deermem_eviction" def _load_config(): return load_evaluation_config(EVAL_ROOT / "configs" / "pr4789-reproduction-v1.yaml") def _case(case_id: str, *, source: str, scenario: str, answer: str) -> PreparedCase: config = _load_config() facts = [{"id": f"{case_id}-fact", "content": "Secret memory content.", "category": "context", "confidence": 0.9, "createdAt": "2026-02-14T00:00:00Z", "source": "synthetic"}] return PreparedCase( case_id=case_id, source=source, # type: ignore[arg-type] scenario=scenario, # type: ignore[arg-type] question_type="synthetic-shape", question="Secret question text?", answer=answer, question_date=None, evaluation_time=config.evaluation_time, facts=facts, usage={}, support_fact_ids=(f"{case_id}-fact",), ) def _policy_result(case: PreparedCase, policy: str) -> PolicyResult: return PolicyResult( case_id=case.case_id, source=case.source, scenario=case.scenario, question_type=case.question_type, policy=policy, # type: ignore[arg-type] capacity=7, support_fact_ids=case.support_fact_ids, kept_fact_ids=case.support_fact_ids, evicted=(), scores={}, support_all_retained=True, support_recall=1.0, reserved_correction_slots=0, ) def _write_response(output_dir: Path, case: PreparedCase, policy: str, prediction: str, *, capacity: int = 7) -> None: row_id = f"{case.case_id}__{policy}" path = response_path(output_dir, row_id) path.parent.mkdir(parents=True, exist_ok=True) row = { "schema_version": 1, "row_id": row_id, "case_id": case.case_id, "source": case.source, "scenario": case.scenario, "policy": policy, "capacity": capacity, "kept_fact_ids": list(case.support_fact_ids), "prediction": prediction, "attempts": 1, "request_fingerprint": "f" * 64, "response_model": "deepseek-v4-flash", "usage": {"prompt_tokens": 10, "completion_tokens": 2}, "created_at": "2026-08-17T00:00:00Z", } path.write_text(json.dumps(row) + "\n", encoding="utf-8") def _setup(tmp_path: Path): official = _case("case-off", source="longmemeval", scenario="access_help", answer="Lyon-Reference-Secret") synthetic = _case("case-syn", source="synthetic", scenario="correction_reserve", answer="NO") cases = [official, synthetic] _write_response(tmp_path, official, "confidence", "Paris") _write_response(tmp_path, official, "hybrid-v1", "Lyon-Reference-Secret") _write_response(tmp_path, synthetic, "confidence", "NO") _write_response(tmp_path, synthetic, "hybrid-v1", "NO") results = {f"{case.case_id}__{policy}": _policy_result(case, policy) for case in cases for policy in ("confidence", "hybrid-v1")} fingerprints = {row_id: "f" * 64 for row_id in results} return cases, results, fingerprints def test_exact_mcnemar_matches_the_two_sided_exact_binomial() -> None: concordant = exact_mcnemar([(True, True), (False, False)]) assert concordant.p_value == 1.0 one_sided_shift = exact_mcnemar([(False, True)] * 5) assert one_sided_shift.only_second_correct == 5 assert one_sided_shift.p_value == pytest.approx(2 * 0.5**5) mixed = exact_mcnemar([(False, True)] * 8 + [(True, False)] + [(True, True)] * 3) assert (mixed.only_first_correct, mixed.only_second_correct, mixed.both_correct) == (1, 8, 3) assert mixed.p_value == pytest.approx(2 * (0.5**9) * (1 + 9), rel=1e-12) def test_paired_bootstrap_is_seed_deterministic_and_signed_second_minus_first() -> None: config = _load_config() pairs = [(False, True)] * 4 + [(True, True)] * 4 first = paired_bootstrap_difference(pairs, seed=config.statistics.bootstrap_seed, iterations=1000, alpha=config.statistics.alpha) second = paired_bootstrap_difference(pairs, seed=config.statistics.bootstrap_seed, iterations=1000, alpha=config.statistics.alpha) assert first == second assert first.mean_difference == pytest.approx(0.5) degenerate = paired_bootstrap_difference([(False, True)] * 3, seed=1, iterations=100, alpha=0.05) assert (degenerate.mean_difference, degenerate.lower, degenerate.upper) == (1.0, 1.0, 1.0) def test_grading_is_blind_and_joined_by_row_id(tmp_path: Path) -> None: cases, results, fingerprints = _setup(tmp_path) rows = collect_answer_rows(tmp_path, cases) graded = grade_answer_rows(cases, results, rows, expected_fingerprints=fingerprints) by_row = {row["row_id"]: row for row in graded} assert not by_row["case-off__confidence"]["grade_correct"] assert by_row["case-off__hybrid-v1"]["grade_correct"] assert by_row["case-syn__confidence"]["grade_correct"] assert by_row["case-syn__hybrid-v1"]["grade_correct"] assert all(row["grader_version"] == "deterministic-overlap-v1" for row in graded) def test_collect_and_integrity_checks_reject_incomplete_or_tampered_rows(tmp_path: Path) -> None: cases, results, fingerprints = _setup(tmp_path) response_path(tmp_path, "case-syn__hybrid-v1").unlink() with pytest.raises(AnswerRowIntegrityError, match="case-syn__hybrid-v1"): collect_answer_rows(tmp_path, cases) _write_response(tmp_path, cases[1], "hybrid-v1", "NO", capacity=5) rows = collect_answer_rows(tmp_path, cases) with pytest.raises(AnswerRowIntegrityError, match="capacity/policy"): grade_answer_rows(cases, results, rows, expected_fingerprints=fingerprints) _write_response(tmp_path, cases[1], "hybrid-v1", "NO") rows = collect_answer_rows(tmp_path, cases) tampered = dict(results) tampered["case-syn__hybrid-v1"] = _policy_result(_case("case-syn", source="synthetic", scenario="correction_reserve", answer="NO"), "hybrid-v1") object.__setattr__(tampered["case-syn__hybrid-v1"], "kept_fact_ids", ("other-fact",)) with pytest.raises(AnswerRowIntegrityError, match="kept facts"): grade_answer_rows(cases, tampered, rows, expected_fingerprints=fingerprints) stale = dict(fingerprints) stale["case-syn__hybrid-v1"] = "0" * 64 with pytest.raises(AnswerRowIntegrityError, match="request fingerprint"): grade_answer_rows(cases, results, rows, expected_fingerprints=stale) def test_grading_rejects_a_row_reassigned_to_another_valid_case(tmp_path: Path) -> None: cases, results, fingerprints = _setup(tmp_path) path = response_path(tmp_path, "case-off__hybrid-v1") original = json.loads(path.read_text(encoding="utf-8")) for field, value in (("case_id", "case-syn"), ("source", "synthetic"), ("scenario", "correction_reserve"), ("row_id", "case-syn__hybrid-v1")): path.write_text(json.dumps(dict(original, **{field: value})) + "\n", encoding="utf-8") rows = collect_answer_rows(tmp_path, cases) with pytest.raises(AnswerRowIntegrityError, match="case-off__hybrid-v1.*case identity"): grade_answer_rows(cases, results, rows, expected_fingerprints=fingerprints) def test_summary_and_statistics_keep_suites_separate(tmp_path: Path) -> None: cases, results, fingerprints = _setup(tmp_path) config = _load_config() graded = grade_answer_rows(cases, results, collect_answer_rows(tmp_path, cases), expected_fingerprints=fingerprints) summary = summarize_qa_rows(graded) assert {(group["source"], group["scenario"], group["policy"]): group["accuracy"] for group in summary} == { ("longmemeval", "access_help", "confidence"): 0.0, ("longmemeval", "access_help", "hybrid-v1"): 1.0, ("synthetic", "correction_reserve", "confidence"): 1.0, ("synthetic", "correction_reserve", "hybrid-v1"): 1.0, } statistics = compute_qa_statistics(graded, config) assert statistics["suites"]["official"]["cases"] == 1 assert statistics["suites"]["synthetic"]["cases"] == 1 assert statistics["suites"]["overall"]["cases"] == 2 assert statistics["suites"]["official"]["mcnemar"]["only_second_correct"] == 1 assert statistics["suites"]["official"]["bootstrap"]["seed"] == config.statistics.bootstrap_seed def test_written_report_redacts_dataset_text_and_refuses_overwrite(tmp_path: Path) -> None: cases, results, fingerprints = _setup(tmp_path) config = _load_config() graded = grade_answer_rows(cases, results, collect_answer_rows(tmp_path, cases), expected_fingerprints=fingerprints) write_qa_report(tmp_path, graded=graded, summary=summarize_qa_rows(graded), statistics=compute_qa_statistics(graded, config), config=config) rows_text = (tmp_path / "qa.rows.jsonl").read_text(encoding="utf-8") assert "Secret question text" not in rows_text assert "Secret memory content" not in rows_text confidence_official = json.loads(next(line for line in rows_text.splitlines() if '"case-off__confidence"' in line)) assert "Lyon-Reference-Secret" not in json.dumps(confidence_official) assert (tmp_path / "qa.summary.json").exists() assert (tmp_path / "qa.stats.json").exists() with pytest.raises(FileExistsError): write_qa_report(tmp_path, graded=graded, summary=summarize_qa_rows(graded), statistics=compute_qa_statistics(graded, config), config=config)