Jyotisha/tests/test_minute_rectification_holdout_intake.py

import hashlib
import json
from copy import deepcopy
from pathlib import Path

from scripts.minute_rectification_holdout_intake import DEFAULT_INTAKE, append_case
from scripts.minute_rectification_holdout_validator import (
    DEFAULT_MANIFEST,
    FRESH_HUMAN_REVIEW_KIND,
    MIGRATED_HUMAN_REVIEW_KIND,
    case_errors,
)

V3_HOLDOUT = Path(__file__).resolve().parents[1] / "references" / "real_case_calibration" / "minute_rectification_holdout_v3.json"
EXPECTED_V4_INTAKE_CASE_IDS = {
    "barack_obama_1961_aa_v4_holdout",
    "paul_ryan_1970_aa_v4_holdout",
    "sean_lennon_1975_aa_v4_holdout",
    "kurt_cobain_1967_aa_v4_holdout",
}


def _reviewed_case() -> dict:
    case = deepcopy(json.loads(DEFAULT_MANIFEST.read_text(encoding="utf-8"))["cases"][0])
    case["case_id"] = "new-reviewed-case"
    case["adjudicator"] = "independent-reviewer"
    case["independent_human_reviewed"] = True
    case["human_review_kind"] = FRESH_HUMAN_REVIEW_KIND
    case["frozen_before_scoring"] = True
    case["false_minute_commitments"] = [
        {
            "offset_minutes": offset,
            "commitment_hash": hashlib.sha256(f"control:{offset}".encode()).hexdigest(),
        }
        for offset in case["false_minute_offsets"]
    ]
    return case


def _queue(path: Path) -> None:
    path.write_text(json.dumps({
        "schema_version": "minute-rectification-holdout-v4-intake",
        "minimum_gate": {
            "events_per_case": 3,
            "domains_per_case": 2,
            "independent_event_sources_per_case": 2,
            "negative_minutes_per_case": 4,
            "day_precision_events_per_case": 3,
        },
        "cases": [],
    }), encoding="utf-8")


def test_intake_appends_reviewed_case_but_keeps_release_blocked(tmp_path: Path) -> None:
    path = tmp_path / "intake.json"
    _queue(path)

    report = append_case(path, _reviewed_case())
    data = json.loads(path.read_text(encoding="utf-8"))

    assert report["appended"] is True
    assert report["verified_minute_claim_allowed"] is False
    assert data["production_tuning_allowed"] is False
    assert data["verified_minute_claim_allowed"] is False
    assert data["cases"][0]["ingested_at"].endswith("Z")


def test_intake_rejects_boolean_review_without_fresh_biography_audit(tmp_path: Path) -> None:
    path = tmp_path / "intake.json"
    _queue(path)
    case = _reviewed_case()
    case["human_review_kind"] = MIGRATED_HUMAN_REVIEW_KIND

    report = append_case(path, case)

    assert report["appended"] is False
    assert "independent_review_requires_fresh_biography_audit" in report["errors"]
    assert json.loads(path.read_text(encoding="utf-8"))["cases"] == []


def test_intake_rejects_missing_review_and_commitments(tmp_path: Path) -> None:
    path = tmp_path / "intake.json"
    _queue(path)
    case = _reviewed_case()
    case["independent_human_reviewed"] = False
    case["false_minute_commitments"] = []

    report = append_case(path, case)

    assert report["appended"] is False
    assert "independent_review_not_attested" in report["errors"]
    assert "false_minute_commitments_do_not_match_offsets" in report["errors"]


def test_default_intake_is_non_production_and_contains_day_precision_v3_cases() -> None:
    data = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))
    gate = data["minimum_gate"]

    assert {case["case_id"] for case in data["cases"]} == EXPECTED_V4_INTAKE_CASE_IDS
    assert len(data["cases"]) == 4
    assert data["production_tuning_allowed"] is False
    assert data["verified_minute_claim_allowed"] is False
    assert data["status"] == "exposed_awaiting_human_rereview"
    assert data["blind_holdout_eligible_case_count"] == 0
    assert data["exposed_awaiting_human_rereview_case_count"] == 4
    for case in data["cases"]:
        assert case["prior_score_exposures"]
        assert case["human_review_kind"] == MIGRATED_HUMAN_REVIEW_KIND
        assert case["independent_human_reviewed"] is False
        assert case["frozen_before_scoring"] is False
        assert case["holdout_partition"] == "exposed_awaiting_human_rereview"
        errors = case_errors(case, gate, require_review_safeguards=True)
        assert "results_already_seen_cannot_be_blind_holdout" in errors
        assert "migrated_review_does_not_satisfy_safeguard" in errors
        assert "fresh_biography_audit_not_attested" in errors
    eligible = [
        case["case_id"]
        for case in data["cases"]
        if not case_errors(case, gate, require_review_safeguards=True)
    ]
    assert eligible == []


def test_exposed_case_cannot_pass_blind_holdout_safeguards() -> None:
    case = _reviewed_case()
    case["prior_score_exposures"] = [
        {
            "evaluation_kind": "v3_blind_replay",
            "report_path": "references/real_case_calibration/minute_rectification_holdout_v3_report.json",
        },
        {
            "evaluation_kind": "post_audit_sidecar_diagnostic",
            "report_path": "references/real_case_calibration/minute_rectification_holdout_v3_post_audit_diagnostic_report.json",
        },
    ]
    gate = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))["minimum_gate"]
    errors = case_errors(case, gate, require_review_safeguards=True)
    assert "results_already_seen_cannot_be_blind_holdout" in errors


def test_migrated_human_review_does_not_satisfy_safeguards() -> None:
    case = _reviewed_case()
    case["human_review_kind"] = MIGRATED_HUMAN_REVIEW_KIND
    case["adjudicator"] = "v3_sealed_set_post_audit_migration"
    gate = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))["minimum_gate"]
    errors = case_errors(case, gate, require_review_safeguards=True)
    assert "migrated_review_does_not_satisfy_safeguard" in errors
    assert "fresh_biography_audit_not_attested" in errors


def test_fresh_unseen_reviewed_case_still_passes_safeguards() -> None:
    case = _reviewed_case()
    assert "prior_score_exposures" not in case
    gate = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))["minimum_gate"]
    assert case_errors(case, gate, require_review_safeguards=True) == []


def test_year_precision_v3_cases_are_rejected_for_insufficient_day_precision() -> None:
    v3 = json.loads(V3_HOLDOUT.read_text(encoding="utf-8"))
    gate = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))["minimum_gate"]
    rejected: dict[str, list[str]] = {}
    for case in v3["cases"]:
        payload = deepcopy(case)
        payload.update({
            "adjudicator": "v3_sealed_set_post_audit_migration",
            "human_review_kind": MIGRATED_HUMAN_REVIEW_KIND,
            "independent_human_reviewed": True,
            "frozen_before_scoring": True,
            "false_minute_commitments": [
                {
                    "offset_minutes": offset,
                    "commitment_hash": hashlib.sha256(f"control:{offset}".encode()).hexdigest(),
                }
                for offset in case["false_minute_offsets"]
            ],
        })
        errors = case_errors(payload, gate, require_review_safeguards=True)
        if errors:
            rejected[str(case["case_id"])] = errors

    assert "insufficient_day_precision_events" in rejected["john_robbins_1947_aa_v4_holdout"]
    assert "insufficient_day_precision_events" in rejected["iwao_takamoto_1925_aa_v4_holdout"]
    assert "insufficient_day_precision_events" in rejected["angelina_jolie_1975_aa_v4_holdout"]
    for case_id in EXPECTED_V4_INTAKE_CASE_IDS:
        assert "migrated_review_does_not_satisfy_safeguard" in rejected[case_id]
        assert "fresh_biography_audit_not_attested" in rejected[case_id]
        assert "insufficient_day_precision_events" not in rejected[case_id]
    assert all("insufficient_day_precision_events" in errors for case_id, errors in rejected.items() if case_id not in EXPECTED_V4_INTAKE_CASE_IDS)