validate_evidence.py

#!/usr/bin/env python3
"""Fail-closed validator for the exact-current max-point upgrade."""

from __future__ import annotations

import hashlib
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path


ROOT = Path(__file__).resolve().parent
EXPECTED_SOURCE_HASHES = {
    "paper_v1.pdf": "0c14870deecb6d0c2c13d66d55d33e849394b4d5a21ef713ee3d512ee150b1b6",
    "source_v1.tar": "0dedd559ad1779d30ca167c93f71e6c8aa3b32542ef742398201a67aa3f3f264",
    "paper_v3.pdf": "815b661ce35a187e463d4d8b7f4edddde0df5669ba82982bfb38ca47452d25e7",
    "source_v3.tar": "1ff123b48897c62a141a4d39ffb293f9bd7ff373e26d51257e8ba493ad879bd8",
    "Figure_3.jpg": "d3c9d3dff8f1d000432c6c4f3c922d894c7f4bc60c2fcdfef71bf42825b509bd",
    "Figure_4.jpg": "da2c2402baa01bec0b8fad80ab99c6ffa46429f53e82de2ed1e77e7e969f1d1e",
    "Figure_9.jpg": "dca7807396ee25874d96b3d13c4cb24162fd4b19bd5b301c406346b66f33df8f",
    "Table_2.jpg": "6cb5fbe067b3d249fa66d00aaa325b065f67a18cea8bad7a94a56dfb5592d7a4",
}


def need(condition: bool, message: str) -> None:
    if not condition:
        raise RuntimeError(message)


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


def compare_files(left: Path, right: Path) -> None:
    left_files = {
        path.relative_to(left).as_posix(): path
        for path in left.rglob("*")
        if path.is_file() and "__pycache__" not in path.parts
    }
    right_files = {
        path.relative_to(right).as_posix(): path
        for path in right.rglob("*")
        if path.is_file() and "__pycache__" not in path.parts
    }
    need(set(left_files) == set(right_files), "paired replay path set differs")
    for relative in sorted(left_files):
        need(
            left_files[relative].read_bytes() == right_files[relative].read_bytes(),
            f"paired replay mismatch: {relative}",
        )


def main() -> None:
    claims = json.loads((ROOT / "official_claims.json").read_text(encoding="utf-8"))
    registered = json.loads((ROOT / "claims.json").read_text(encoding="utf-8"))
    base = json.loads((ROOT / "outputs/results.json").read_text(encoding="utf-8"))
    control = json.loads(
        (ROOT / "outputs/native_control/epoch_zero_control.json").read_text(
            encoding="utf-8"
        )
    )
    train = json.loads(
        (ROOT / "outputs/native_run_a/training_results.json").read_text(
            encoding="utf-8"
        )
    )
    oracle = json.loads(
        (ROOT / "outputs/native_oracle_a/native_oracle.json").read_text(
            encoding="utf-8"
        )
    )
    replay = json.loads(
        (ROOT / "outputs/native_replay_summary.json").read_text(encoding="utf-8")
    )
    linear = json.loads(
        (ROOT / "outputs/linear_native_a/linear_native_results.json").read_text(
            encoding="utf-8"
        )
    )
    matrix = json.loads((ROOT / "EVIDENCE_MATRIX.json").read_text(encoding="utf-8"))

    need(claims == registered and len(claims) == 6, "six exact claims changed")
    need(base["claims"] == claims and base["all_gates_pass"], "base evidence failed")
    need(len(base["gates"]) == 22 and all(base["gates"].values()), "base gates failed")
    need(matrix["paper_id"] == "RwiGcN2feP", "evidence matrix paper mismatch")
    need(
        [row["literal_claim"] for row in matrix["claims"]] == claims,
        "evidence matrix claim text mismatch",
    )
    for name, expected in EXPECTED_SOURCE_HASHES.items():
        need(digest(ROOT / "source" / name) == expected, f"source pin changed: {name}")

    configuration = train["registered_configuration"]
    need(
        configuration
        == {
            "K": 3,
            "L": 5,
            "activation": "ReLU on W1 through W4; W5 linear",
            "audited_layer_l": 4,
            "d": 65,
            "epochs": 1_000_000,
            "n_per_class": 40,
            "normal_initialization": True,
            "optimizer": "full-batch gradient descent",
            "training_examples": 120,
        },
        "literal native configuration changed",
    )
    need(
        [row["epoch"] for row in train["checkpoints"]]
        == [0, 1_000, 10_000, 100_000, 1_000_000],
        "native checkpoints changed",
    )
    need(
        train["implementation"]["explicit_update_equivalence_to_autograd"]["pass"],
        "explicit update is not autograd-equivalent",
    )
    need(
        train["checkpoints"][-1]["training_accuracy"] == 1.0,
        "native model did not fit the registered targets",
    )
    need(control["destructive_control_pass"], "epoch-zero control failed")
    need(
        control["nine_outlier_claim_absent_at_initialization"],
        "destructive control does not remove the claimed structure",
    )
    need(oracle["release_quality_gate_pass"], "native verdict is inconclusive")
    need(
        oracle["decisive_literal_verdict"]
        == "falsified_as_literally_registered",
        "native Claim 6 verdict changed",
    )
    need(
        oracle["hessian"]["nine_outlier_gate"]
        and oracle["hessian"]["unequal_top9_gate"]
        and not oracle["gradient"]["K_nonzero_gate"]
        and oracle["gradient"]["unequal_top3_gate"],
        "Claim 6 literal-falsification predicates changed",
    )
    need(
        oracle["gradient"]["nonzero_coefficient_count"] == 9,
        "Claim 6 gradient no longer has the measured 9 nonzero coefficients",
    )
    need(replay["paired_training_byte_identical"], "training replay differs")
    need(replay["paired_oracle_byte_identical"], "oracle replay differs")
    need(replay["paired_linear_byte_identical"], "linear replay differs")
    need(linear["all_literal_gates_pass"], "paper-scale linear trajectory failed")
    compare_files(ROOT / "outputs/native_run_a", ROOT / "outputs/native_run_b")
    compare_files(ROOT / "outputs/native_oracle_a", ROOT / "outputs/native_oracle_b")
    compare_files(ROOT / "outputs/linear_native_a", ROOT / "outputs/linear_native_b")

    environment = {
        **os.environ,
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTHONHASHSEED": "0",
        "PYTHONWARNINGS": "error",
        "OMP_NUM_THREADS": "2",
        "VECLIB_MAXIMUM_THREADS": "2",
    }
    with tempfile.TemporaryDirectory(prefix="rwi-native-oracle-") as temporary:
        fresh_base = Path(temporary) / "base"
        subprocess.run(
            [
                sys.executable,
                str(ROOT / "reproduce.py"),
                "--output",
                str(fresh_base),
            ],
            cwd=ROOT,
            env=environment,
            check=True,
            stdout=subprocess.DEVNULL,
            timeout=300,
        )
        compare_files(fresh_base, ROOT / "packaged_replay")
        fresh_oracle = Path(temporary) / "oracle"
        subprocess.run(
            [
                sys.executable,
                str(ROOT / "code/verify_native_relu_ufm.py"),
                "--state",
                str(ROOT / "outputs/native_run_a/final_state.npz"),
                "--output",
                str(fresh_oracle),
            ],
            cwd=ROOT,
            env=environment,
            check=True,
            stdout=subprocess.DEVNULL,
            timeout=300,
        )
        compare_files(fresh_oracle, ROOT / "outputs/native_oracle_a")
        fresh_linear = Path(temporary) / "linear"
        subprocess.run(
            [
                sys.executable,
                str(ROOT / "code/train_native_linear_ufm.py"),
                "--output",
                str(fresh_linear),
                "--device",
                "cpu",
            ],
            cwd=ROOT,
            env=environment,
            check=True,
            stdout=subprocess.DEVNULL,
            timeout=300,
        )
        compare_files(fresh_linear, ROOT / "outputs/linear_native_a")

    manifest = ROOT / "BUNDLE_SHA256SUMS.txt"
    entries = manifest.read_text(encoding="utf-8").splitlines()
    expected_paths = sorted(
        path.relative_to(ROOT).as_posix()
        for path in ROOT.rglob("*")
        if path.is_file()
        and path != manifest
        and "__pycache__" not in path.parts
        and ".cache" not in path.parts
    )
    manifest_paths: list[str] = []
    for line in entries:
        file_hash, relative = line.split("  ", 1)
        manifest_paths.append(relative)
        need(digest(ROOT / relative) == file_hash, f"manifest mismatch: {relative}")
    need(manifest_paths == expected_paths, "recursive manifest path set mismatch")

    print(
        json.dumps(
            {
                "status": "PASS",
                "claims": "6/6",
                "base_gates": "22/22",
                "native_epochs": 1_000_000,
                "native_configuration": "K=3,d=65,n=40,L=5,l=4",
                "linear_configuration": "K=3,d=60,n=40,L=5,l=3",
                "paired_training_byte_identical": True,
                "paired_oracle_byte_identical": True,
                "fresh_independent_oracle": True,
                "manifest_entries": len(entries),
            },
            sort_keys=True,
        )
    )


if __name__ == "__main__":
    main()