Jyotisha/tests/test_sealed_holdout_contract_freshness.py

from __future__ import annotations

import json
from pathlib import Path

import pytest

from scripts.minute_rectification_blind_eval import implementation_sha256, summarize_trials
from scripts.rectification.sealed_holdout import holdout_passed, load_sealed_minute_holdout
from scripts.research.sealed_holdout_rerun import (
    ARCHIVE, DATASET, FREEZE, REPORT, LEGACY_REPORT, PRODUCTION_FILES, file_sha256,
    freeze_record, historical_comparison, implementation_identity, run,
)

ROOT = Path(__file__).resolve().parents[1]


def read(path):
    return json.loads(path.read_text(encoding="utf-8"))


def test_contract_tracks_actual_current_scorer_and_dataset_audit():
    dataset = read(DATASET)
    contract = read(ROOT / "references/rectification_sealed_holdout.v1.json")
    actual_hash = implementation_sha256(dataset["frozen_scoring"]["files"])
    assert contract["current_tree_scorer"]["implementation_sha256"] == actual_hash
    assert contract["current_tree_scorer"]["extended_identity"] == implementation_identity()
    historical_contract = read(ARCHIVE / "references/rectification_sealed_holdout.v1.json")
    runtime_keys = ("status", "valid_public_aa_cases", "required_cases", "top_1_rate", "confirmation_coverage_rate", "sealed_benchmark_id")
    for key in runtime_keys:
        assert type(contract[key]) is type(historical_contract[key])
        assert contract[key] == historical_contract[key]
    assert contract["source_audit_status"] == dataset["source_audit_status"]
    assert contract["evaluated_on"] == read(REPORT)["evaluated_on"]
    assert contract["status"] == "not_ready"
    assert contract["confirmation_coverage_rate"] == 0.0
    assert holdout_passed(load_sealed_minute_holdout()) is False


def test_frozen_record_matches_dataset_scorer_and_evaluator_bytes():
    frozen = read(FREEZE)
    actual = freeze_record()
    assert {key: value for key, value in frozen.items() if key != "frozen_at_utc"} == {
        key: value for key, value in actual.items() if key != "frozen_at_utc"
    }
    assert frozen["dataset_sha256"] == file_sha256(DATASET)
    assert len(frozen["files"]) == 12
    assert read(DATASET)["frozen_scoring"]["implementation_sha256"] == frozen["historical_frozen_sha256"]
    assert frozen["extended_identity"] == implementation_identity()
    assert set(PRODUCTION_FILES) <= set(frozen["extended_identity"]["production_scoring_files"])
    assert {"scripts/rectification/scoring_service.py", "scripts/rectification/dasha_transition_proximity.py"} <= set(PRODUCTION_FILES)
    for path, digest in frozen["extended_identity"]["file_sha256"].items():
        assert digest == file_sha256(ROOT / path)


def test_fixed_protocol_rerun_is_auditable_but_never_independent_blind():
    report = read(REPORT)
    contract = read(ROOT / "references/rectification_sealed_holdout.v1.json")
    scorer = contract["current_tree_scorer"]
    rerun = contract["current_tree_fixed_protocol_rerun"]
    assert report["frozen_record"] == read(FREEZE)
    assert report["frozen_record"]["frozen_at_utc"] <= report["replay_started_at_utc"] <= report["replay_finished_at_utc"]
    assert scorer["extended_identity"] == report["frozen_record"]["extended_identity"]
    assert rerun["extended_identity"] == scorer["extended_identity"]
    assert rerun["freeze_record_path"] == FREEZE.relative_to(ROOT).as_posix()
    assert report["historical_comparison"] == historical_comparison(LEGACY_REPORT, report["trials"], ("case_ordinal",))
    assert report["trial_count"] == len(report["trials"]) == 20
    assert report["excluded_cases"] == []
    aggregate = summarize_trials(report["trials"], read(DATASET)["release_metrics"])
    assert report["metrics"] == aggregate["metrics"] == scorer["metrics"]
    assert report["metric_gates_passed"] == aggregate["metric_gates_passed"]
    assert scorer["source_report"] == REPORT.relative_to(ROOT).as_posix()
    assert scorer["implementation_sha256"] == report["frozen_record"]["implementation_sha256"]
    assert scorer["fixed_protocol_rerun_trial_count"] == report["trial_count"]
    assert scorer["fixed_protocol_rerun_hash_matches"] is True
    assert scorer["official_eval_trial_count"] == report["official_blind_trial_count"] == 0
    assert scorer["official_eval_implementation_hash_matches"] is False
    for item in (rerun, report):
        assert item["official_valid_independent_blind"] is False
        assert item["is_blind_evaluation"] is False
        assert item["truth_hidden_from_ranker"] is True
        assert item["results_previously_seen"] is True
        assert item["verified_minute_claim_allowed"] is False
    assert rerun["must_not_claim_as_release_metrics"] is True
    assert all(row["event_count"] == 3 for row in report["trials"])
    # Never persist actual candidate/truth times or coordinates in report artifacts.
    for row in report["trials"]:
        assert not ({"predicted_time", "published_truth_revealed_after_ranking", "latitude", "longitude"} & row.keys())


def test_changed_frozen_identity_fails_before_any_replay(tmp_path, monkeypatch):
    frozen = read(FREEZE)
    frozen["implementation_sha256"] = "0" * 64
    path = tmp_path / "bad-freeze.json"
    path.write_text(json.dumps(frozen), encoding="utf-8")
    def unexpected(*args, **kwargs):
        raise AssertionError("must reject identity before scoring")
    monkeypatch.setattr("scripts.research.sealed_holdout_rerun.build_feature_fact_rows", unexpected)
    with pytest.raises(ValueError, match="frozen_record_mismatch:implementation_sha256"):
        run(path)


@pytest.mark.parametrize("path", ["scripts/rectification/scoring_service.py", "scripts/rectification/dasha_transition_proximity.py"])
def test_extended_identity_drift_rejected_even_when_legacy_hash_unchanged(tmp_path, monkeypatch, path):
    from scripts.research import sealed_holdout_rerun as replay
    frozen = replay.freeze_record()
    freeze_path = tmp_path / "extended-freeze.json"
    freeze_path.write_text(json.dumps(frozen), encoding="utf-8")
    original = replay.implementation_identity
    def changed(dataset=DATASET):
        actual = original(dataset)
        actual["file_sha256"][path] = "0" * 64
        return actual
    monkeypatch.setattr(replay, "implementation_identity", changed)
    def unexpected(*args, **kwargs):
        raise AssertionError("extended drift must reject before shadow scoring")
    monkeypatch.setattr(replay, "build_feature_fact_rows", unexpected)
    assert frozen["implementation_sha256"] == replay.freeze_record()["implementation_sha256"]
    with pytest.raises(ValueError, match="frozen_record_mismatch:extended_identity"):
        replay.run(freeze_path)


def test_historical_artifacts_are_byte_preserved_not_refreshed():
    manifest = read(ARCHIVE / "manifest.json")
    assert file_sha256(ARCHIVE / "manifest.json") == "6102a26a840be207b5858b3a4c0509274468ae9071e86d308cbbac7371d6864e"
    assert len(manifest["files"]) == 8
    for record in manifest["files"]:
        archived = ROOT / record["archive_path"]
        assert archived.stat().st_size == record["size_bytes"]
        assert file_sha256(archived) == record["sha256"]
    old_freeze = ROOT / "docs/research/sealed_holdout_rerun_2026_09_20.freeze.json"
    assert old_freeze.read_bytes() == (ARCHIVE / old_freeze.relative_to(ROOT)).read_bytes()
    assert file_sha256(old_freeze) == "d17651cb50224acdd1af7a4692c777ed169b216a962a0675371254ce561d0921"
    assert FREEZE != old_freeze
    for name, digest in (
        ("reported_offset_2026_09_20.json", "9878f2b50c957a470fafcb2ed0a9eb16405c3f6fa422ea50f467e84b0189ace1"),
        ("sealed_holdout_rerun_2026_09_20.json", "40df220bfcb51a683b31fbd626896e47a5aed7ee4f30ff669543691f7ce53c1e"),
    ):
        assert file_sha256(ARCHIVE / "docs/research" / name) == digest