#!/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_midnight_anchor_2026_09_21.freeze.json" REPORT = ROOT / "docs/research/sealed_holdout_rerun_midnight_anchor_2026_09_21.json" LEGACY_REPORT = ROOT / "docs/research/sealed_holdout_rerun_2026_09_20.json" ARCHIVE = ROOT / "docs/research/history/rectification_pre_cross_midnight_2026_09_20" PRODUCTION_FILES = [ "scripts/rectification/scoring_service.py", "scripts/rectification/dasha_transition_proximity.py", "scripts/rectification/candidate_contrast.py", "scripts/rectification/case_holdout.py", "scripts/rectification/contracts.py", "scripts/rectification/event_probes.py", "scripts/rectification/candidate_window.py", "scripts/rectification/decision_policy.py", "scripts/rectification/refinement_packet.py", "scripts/rectification/api_service.py", ] RESEARCH_FILES = [ "scripts/research/reported_offset_sweep.py", "scripts/research/cluster_width_lib.py", "scripts/research/probe_supply_after_six.py", "scripts/research/sealed_holdout_rerun.py", "scripts/minute_rectification_blind_eval.py", "scripts/minute_rectification_fact_blind_eval_v4.py", "scripts/minute_rectification_holdout_validator.py", ] 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 implementation_identity(dataset: Path = DATASET) -> dict[str, Any]: """Bind legacy scorer, changed production files and both replay adapters. The production identity is contextual for the shadow rerun, not a claim that the shadow scorer executes transition proximity. These are explicit audited file sets, not a transitive dependency or environment lock. """ legacy_files = json.loads(dataset.read_text(encoding="utf-8"))["frozen_scoring"]["files"] production_files = sorted(set(legacy_files + PRODUCTION_FILES)) return { "historical_artifacts_manifest_sha256": file_sha256(ARCHIVE / "manifest.json"), "production_scoring_files": production_files, "production_scoring_sha256": implementation_sha256(production_files), "research_files": RESEARCH_FILES, "research_implementation_sha256": implementation_sha256(RESEARCH_FILES), "file_sha256": {path: file_sha256(ROOT / path) for path in sorted(set(production_files + RESEARCH_FILES))}, "hash_scope": "explicit_identity_file_sets_not_a_transitive_dependency_lock", } def verify_frozen_record(frozen: dict[str, Any], actual: dict[str, Any]) -> None: for key in actual: if key != "frozen_at_utc" and actual[key] != frozen.get(key): raise ValueError(f"frozen_record_mismatch:{key}") if set(frozen) != set(actual): raise ValueError("frozen_record_mismatch:keys") def historical_comparison(report_path: Path, trials: list[dict[str, Any]], keys: tuple[str, ...]) -> dict[str, Any]: """Keep every before/after row, including unchanged rows and failures.""" history_path = ARCHIVE / report_path.relative_to(ROOT) archive_manifest = json.loads((ARCHIVE / "manifest.json").read_text(encoding="utf-8")) entry = next(item for item in archive_manifest["files"] if item["source_path"] == report_path.relative_to(ROOT).as_posix()) if file_sha256(history_path) != entry["sha256"]: raise ValueError("historical_report_bytes_changed") previous = json.loads(history_path.read_text(encoding="utf-8")) identity = lambda row: tuple(row[key] for key in keys) old = {identity(row): row for row in previous["trials"]} new = {identity(row): row for row in trials} if len(old) != len(previous["trials"]) or len(new) != len(trials) or old.keys() != new.keys(): raise ValueError("historical_comparison_trial_identity_mismatch") comparisons = [{ **{key: row[key] for key in keys}, "before": old[identity(row)], "after": row, "changed_fields": sorted(key for key in set(row) | set(old[identity(row)]) if row.get(key) != old[identity(row)].get(key)), } for row in trials] return { "source_report": history_path.relative_to(ROOT).as_posix(), "source_report_sha256": file_sha256(history_path), "reason": "BUG-981 new production identity and native single-matrix sweep; prior grouped sweep/shadow metrics not presumed incorrect", "trial_count": len(comparisons), "changed_trial_count": sum(bool(row["changed_fields"]) for row in comparisons), "trials": comparisons, } 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-cross-midnight-rerun-v2", "extended_identity": implementation_identity(dataset), "production_identity_scope": "context_only_shadow_scorer_does_not_call_transition_proximity", "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) verify_frozen_record(frozen, actual) replay_started_at = datetime.now(timezone.utc).isoformat() 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"], }) verify_frozen_record(frozen, freeze_record(dataset)) aggregate = summarize_trials(trials, manifest["release_metrics"]) count = len(trials) return { "scope": "fixed_protocol_previously_exposed_v3_rerun", "replay_started_at_utc": replay_started_at, "replay_finished_at_utc": datetime.now(timezone.utc).isoformat(), "historical_comparison": historical_comparison(LEGACY_REPORT, trials, ("case_ordinal",)), "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))