Jyotisha/scripts/research/sealed_holdout_rerun.py

#!/usr/bin/env python3
"""Freeze and replay the exposed v3 corpus; never claim a fresh blind holdout."""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from scripts.active_rectification_event_engine import AYANAMSA, NODE_MODE
from scripts.minute_rectification_blind_eval import (
    _candidate_moments, _clock_distance, _opaque_winner, _request,
    implementation_sha256, summarize_trials,
)
from scripts.minute_rectification_fact_blind_eval_v4 import _would_confirm
from scripts.minute_rectification_fact_ranker_v4 import (
    ALGORITHM_VERSION, rank_fact_rows, score_fact_ranker_v4,
)
from scripts.minute_rectification_feature_facts_v4 import build_feature_fact_rows
from scripts.minute_rectification_holdout_validator import validate

DATASET = ROOT / "references/real_case_calibration/minute_rectification_holdout_v3.json"
FREEZE = ROOT / "docs/research/sealed_holdout_rerun_2026_09_20.freeze.json"
REPORT = ROOT / "docs/research/sealed_holdout_rerun_2026_09_20.json"


def file_sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def opaque_order(benchmark_id: str, case_id: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Extend the existing opaque winner to a total, truth-independent ranking."""
    return sorted(rows, key=lambda row: (
        -row["score"],
        hashlib.sha256(f"{benchmark_id}:{case_id}:{row['time']}".encode()).hexdigest(),
    ))


def freeze_record(dataset: Path = DATASET) -> dict[str, Any]:
    manifest = json.loads(dataset.read_text(encoding="utf-8"))
    files = manifest["frozen_scoring"]["files"]
    return {
        "record_version": "exposed-v3-fixed-protocol-rerun-v1",
        "frozen_at_utc": datetime.now(timezone.utc).isoformat(),
        "dataset_path": dataset.relative_to(ROOT).as_posix(),
        "dataset_sha256": file_sha256(dataset),
        "algorithm_version": ALGORITHM_VERSION,
        "implementation_sha256": implementation_sha256(files),
        "files": files,
        "historical_frozen_sha256": manifest["frozen_scoring"]["implementation_sha256"],
        "evaluator_sha256": file_sha256(Path(__file__)),
        "ayanamsa": AYANAMSA,
        "node_mode": NODE_MODE,
        "candidate_radius_minutes": sorted({case["candidate_radius_minutes"] for case in manifest["cases"]}),
        "minute_step": 1,
        "release_metrics": manifest["release_metrics"],
        "results_previously_seen": True,
        "official_valid_independent_blind": False,
        "must_not_use_for_tuning": True,
        "tie_breaker": manifest["frozen_scoring"]["tie_breaker"],
        "metric_rank_definition": "competition_rank_1_plus_strictly_higher_scores_legacy_protocol",
        "extra_metric_rank_definition": "score_desc_then_existing_opaque_sha256_total_order",
    }


def run(freeze_path: Path = FREEZE, dataset: Path = DATASET) -> dict[str, Any]:
    frozen = json.loads(freeze_path.read_text(encoding="utf-8"))
    actual = freeze_record(dataset)
    for key in actual:
        if key != "frozen_at_utc" and actual[key] != frozen.get(key):
            raise ValueError(f"frozen_record_mismatch:{key}")
    manifest = json.loads(dataset.read_text(encoding="utf-8"))
    validation = validate(dataset)
    invalid = validation["invalid_cases"]
    trials = []
    for index, case in enumerate(manifest["cases"], 1):
        if case["case_id"] in invalid:
            continue
        request = _request(case, case["events"])
        candidates = _candidate_moments(case)
        facts = build_feature_fact_rows(request, candidates=candidates)
        rows, _ = rank_fact_rows(facts, request["events"])
        result = score_fact_ranker_v4(facts, request["events"])
        predicted = _opaque_winner(manifest["benchmark_id"], case["case_id"], rows)
        ordered = opaque_order(manifest["benchmark_id"], case["case_id"], rows)
        sparse_request = _request(case, case["events"][:1])
        sparse_facts = build_feature_fact_rows(sparse_request, candidates=candidates)
        sparse_result = score_fact_ranker_v4(sparse_facts, sparse_request["events"])
        # Truth is revealed only after both full and sparse ranking/decisions.
        truth = case["birth"]["time"]
        truth_score = next(row["score"] for row in rows if row["time"] == truth)
        would_confirm = _would_confirm(result)
        trials.append({
            "case_ordinal": index,
            "candidate_count": len(rows),
            "event_count": len(request["events"]),
            "true_rank": 1 + sum(row["score"] > truth_score for row in rows),
            "opaque_true_rank": next(i for i, row in enumerate(ordered, 1) if row["time"] == truth),
            "minute_error": _clock_distance(predicted, truth),
            "would_confirm": would_confirm,
            "false_confirmation": would_confirm and predicted != truth,
            "insufficient_evidence_rejected": not _would_confirm(sparse_result),
            "full_trial_reasons": result["reasons"],
            "sparse_trial_reasons": sparse_result["reasons"],
        })
    aggregate = summarize_trials(trials, manifest["release_metrics"])
    count = len(trials)
    return {
        "scope": "fixed_protocol_previously_exposed_v3_rerun",
        "evaluated_on": datetime.now(timezone.utc).date().isoformat(),
        "frozen_record": frozen,
        "implementation_hash_matches_at_replay": True,
        "dataset_hash_matches_at_replay": True,
        "source_audit_status": manifest["source_audit_status"],
        "validation_status": validation["status"],
        "valid_public_aa_cases": validation["valid_public_aa_cases"],
        "excluded_cases": invalid,
        "trial_count": count,
        "trials": trials,
        **aggregate,
        "opaque_exact_top_1_rate": sum(row["opaque_true_rank"] == 1 for row in trials) / count if count else None,
        "opaque_exact_top_3_rate": sum(row["opaque_true_rank"] <= 3 for row in trials) / count if count else None,
        "official_valid_independent_blind": False,
        "official_blind_trial_count": 0,
        "is_blind_evaluation": False,
        "truth_hidden_from_ranker": True,
        "results_previously_seen": True,
        "verified_minute_claim_allowed": False,
        "status": "blocked_independent_blind_evidence",
        "boundary": "Three-event low-information protocol, not a mathematical accuracy lower bound and not representative of real sessions. Historical v3/v4 exposure cannot be undone by freezing today's scorer. No tuning or release claims.",
    }


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--freeze", action="store_true", help="Create a new record before replay; never overwrite an existing record")
    parser.add_argument("--freeze-path", type=Path, default=FREEZE)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()
    if args.freeze:
        with args.freeze_path.open("x", encoding="utf-8", newline="\n") as handle:
            json.dump(freeze_record(), handle, ensure_ascii=False, indent=2)
            handle.write("\n")
        print(args.freeze_path.relative_to(ROOT).as_posix())
    else:
        print(json.dumps(run(args.freeze_path), ensure_ascii=False, indent=2))