validate_evidence.py

#!/usr/bin/env python3
"""Fail-closed validator for the six-claim FFOLayer reproduction."""

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
COMMIT = "28905f3e1750fca5b8918954d5d2ea5bed0cbacc"
TREE = "f236d623acd0a089adebafd61c7c239434c9e6b2"
SOURCE_FILES = 93
SOURCE_MANIFEST = "8ca8beef7468dacb0e6a91d4a38e28dafd5256f6b44f680dc65ad27a3678f9c6"
PDF_SHA = "41245b95365c2ffede396f2cb48071f3bf5b749039124af5c804830994e0a14c"
TAR_SHA = "043f3bd94fa18e7cf62f311a3090f1213c81d5c5d65059490d514d1d1a77db13"
LPGD_ERROR = "Unsupported mode lpgd; the supported modes are 'dense', 'lsqr' and 'lsmr'"
CLAIMS = [
    "FFOLayer computes an ε-approximate hypergradient using an active-set Lagrangian oracle that requires no Hessian evaluations, achieving Õ(1) first-order oracle calls per hypergradient estimate (Section 4.2, Algorithm 1).",
    "Theorem 4.1 proves that the 'ghost bilevel optimization' reformulation, which treats active constraints as equalities, preserves the accuracy of the hypergradient computed at the original constrained-optimization solution (Section 4.1, Theorem 4.1).",
    "For constrained bilevel optimization, the method achieves an oracle complexity of Õ(δ⁻¹ε⁻³), matching best-known rates for non-smooth non-convex optimization, while extending prior guarantees from linear to general convex constraints (Section 4, complexity analysis).",
    "On synthetic decision-focused-learning QP tasks and 9×9 Sudoku constraint-learning tasks formulated as linear programs, FFOLayer matches the convergence of exact differentiable-optimization solvers CvxpyLayer and qpth while using a substantially faster backward pass (Experiments section, synthetic QP and Sudoku benchmarks).",
    "FFOLayer's PyTorch implementation is objective-agnostic, exposing task-loss influence via a single detached gradient coefficient c := detach(dF/dy*), allowing users to substitute it for CvxpyLayer with minimal code changes (Section on practical implementation).",
    "FFOLayer outperforms the gradient-unrolling baseline LPGD in the reported experiments while eliminating the cubic-complexity Hessian inversion required by standard implicit differentiation (Experiments section, comparison with LPGD).",
]


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


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def load(relative: str) -> object:
    return json.loads((ROOT / relative).read_text(encoding="utf-8"))


def source_manifest() -> tuple[int, str]:
    source = ROOT / "source_current"
    files = sorted(
        path for path in source.rglob("*")
        if path.is_file() and "__pycache__" not in path.parts and path.suffix != ".pyc"
    )
    lines = [f"{sha256(path)}  {path.relative_to(source).as_posix()}" for path in files]
    digest = hashlib.sha256(("\n".join(lines) + "\n").encode()).hexdigest()
    return len(files), digest


def compare_generated(directory: Path) -> None:
    names = {
        "claim1_rate_repetitions.csv",
        "claim1_rate_summary.json",
        "claim2_native_synthetic_benchmark.csv",
        "claim2_native_synthetic_benchmark.json",
    }
    actual = {path.name for path in directory.iterdir() if path.is_file()}
    need(actual == names, f"replay output set changed: {sorted(actual)}")
    for name in sorted(names):
        need(
            (directory / name).read_bytes() == (ROOT / "outputs" / name).read_bytes(),
            f"replay mismatch: {name}",
        )


def validate_manifest() -> int:
    manifest = ROOT / "BUNDLE_SHA256SUMS.txt"
    need(manifest.is_file(), "missing recursive manifest")
    lines = 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 path.suffix != ".pyc"
    )
    paths = []
    for line in lines:
        digest, relative = line.split("  ", 1)
        paths.append(relative)
        need(sha256(ROOT / relative) == digest, f"manifest mismatch: {relative}")
    need(paths == expected_paths, "manifest path set mismatch")
    return len(lines)


def main() -> None:
    claims = load("official_claims.json")
    logbook = load("logbook.json")
    matrix = load("EVIDENCE_MATRIX.json")
    pin = load("SOURCE_PIN.json")
    need(claims == CLAIMS, "official claims drift")
    need(logbook["claims"] == CLAIMS, "logbook claims drift")
    need(matrix["paper_id"] == "jJur8Fq7IK", "matrix paper mismatch")
    need([row["literal_claim"] for row in matrix["claims"]] == CLAIMS, "matrix claims drift")
    gate = matrix["release_quality_gate"]
    need(gate["semantic_quality_gate_version"] == 4, "semantic gate version changed")
    need(gate["registered_claims"] == 6, "claim count changed")
    need(gate["supported_by_independent_evidence"] == 6, "support count changed")
    need(gate["literal_falsifications"] == 2, "literal falsification census changed")
    need(gate["direct_rate_claims"] == 2, "direct rate census changed")
    need(gate["expected_verified_points"] == 12, "expected points changed")
    need(pin["commit"] == COMMIT and pin["git_tree"] == TREE, "source pin drift")
    count, digest = source_manifest()
    need(count == SOURCE_FILES and digest == SOURCE_MANIFEST, "official source snapshot drift")
    need(sha256(ROOT / "source_paper/arxiv-2512.02494.pdf") == PDF_SHA, "PDF hash drift")
    need(sha256(ROOT / "source_paper/arxiv-2512.02494.tar") == TAR_SHA, "source tar hash drift")

    rate = load("outputs/claim1_rate_summary.json")
    need(rate["rows"] == 14 and rate["repetitions_per_scale"] == 2, "claim-1 rate census changed")
    need(rate["scales"] == [10, 20, 50, 100, 200, 500, 1000], "claim-1 scales changed")
    need(rate["all_errors_le_2epsilon"] is True, "claim-1 accuracy failed")
    need(rate["oracle_evaluations_log_fit_r_squared"] >= 0.999, "claim-1 log rate fit failed")
    need(rate["destructive_control"]["control_triggered"] is True, "claim-1 control failed")

    theory = load("outputs/theory_and_active_set.json")
    ghost = theory["claim_2_ghost_active_set"]
    need(ghost["max_regular_point_absolute_error"] <= 4e-11, "claim-2 ghost identity failed")
    boundary = ghost["destructive_boundary_control"]
    need(boundary["differentiability_assumption_violated"] is True, "claim-2 control failed")
    need(abs(boundary["left_derivative"] - boundary["right_derivative"]) >= 0.24, "claim-2 boundary did not separate")

    complexity = load("outputs/claim3_general_convex_scaling.json")
    need(complexity["rows"] == 8 and complexity["repetitions_per_scale"] == 2, "claim-3 rate census changed")
    need(complexity["scales"] == [10, 20, 50, 100], "claim-3 scales changed")
    need(complexity["all_finite"] is True, "claim-3 non-finite gradient")
    need(complexity["all_soc_constraints_active_at_solver_tolerance"] is True, "claim-3 SOC activity failed")
    need(complexity["measured_exponent_no_worse_than_claimed_upper_bound"] is True, "claim-3 measured rate failed")
    need(complexity["solver_iteration_log_log_r_squared"] >= 0.98, "claim-3 rate fit failed")
    need(complexity["max_relative_hypergradient_error"] <= 0.04, "claim-3 gradient error failed")

    static = load("outputs/implementation_static_audit.json")
    need(static["official_repo_commit"] == COMMIT, "static audit pin drift")
    need(static["hessian_tokens_total"] == 0, "Hessian token appeared")
    need(static["explicit_inverse_calls_total"] == 0, "explicit inverse appeared")
    for name in ("box_qp", "nonnegative_budget_qp", "soc_qp"):
        report = load(f"outputs/objective_agnostic_{name}.json")
        need(len(report["comparisons"]) == 3, f"{name} objective census changed")
        need(report["all_finite"] is True, f"{name} non-finite gradient")
        need(report["max_relative_l2_error"] <= 2e-4, f"{name} gradient error failed")
        need(report["min_cosine_similarity"] >= 0.999999, f"{name} gradient cosine failed")
    objective_control = load("outputs/objective_agnostic_failure_control.json")
    need(objective_control["returncode"] == 0, "claim-5 active-boundary control failed")

    benchmark = load("outputs/claim2_native_synthetic_benchmark.json")
    need(benchmark["official_repository_commit"] == COMMIT, "benchmark pin drift")
    need(benchmark["total_computation_speedup_ffolayer_over_qpth"] >= 5.0, "native total speed failed")
    need(benchmark["absolute_test_df_loss_gap"] <= 5e-4, "native endpoint similarity failed")
    need(benchmark["destructive_boundary_control"]["control_triggered"] is True, "claim-4 falsification failed")
    need(benchmark["measurements"][0]["backward_seconds"] > benchmark["measurements"][1]["backward_seconds"], "claim-4 backward ordering changed")
    cvx = load("outputs/claim2_cvxpylayer_native_batch.json")
    need(cvx["batch_shape"] == [8, 800] and cvx["finite_gradient"] is True, "CvxpyLayer batch failed")
    sudoku = load("outputs/claim2_ffolayer_sudoku_epoch.json")
    need(sudoku["training_samples"] == 9000 and sudoku["test_samples"] == 1000, "Sudoku scale changed")
    need(sudoku["train_batches"] == 1125 and sudoku["test_loss"] < 0.1, "Sudoku execution failed")

    lpgd = load("outputs/claim6_lpgd_release_failure.json")
    full = lpgd["full_native_9x9_attempt"]
    micro = lpgd["released_micro_control"]
    need(lpgd["verdict"] == "falsified_as_literally_registered", "claim-6 verdict changed")
    need(full["train_batches_requested"] == 1125 and full["completed_train_records"] == 0, "claim-6 native scale changed")
    need(full["exception"] == LPGD_ERROR, "claim-6 full failure changed")
    need(micro["registered_lpgd_mode_path"]["exception"] == LPGD_ERROR, "claim-6 micro failure changed")
    need(micro["valid_diffcp_mode_path"]["status"] == "pass", "claim-6 valid-mode control failed")
    native_log = ROOT / "sudoku_results_8/lpgd/central_failures.log"
    epoch_csv = ROOT / "sudoku_results_8/lpgd/lpgd_n3_lr0.1_seed3_20260727_022547.csv"
    step_csv = ROOT / "sudoku_results_8/lpgd_steps/lpgd_n3_lr0.1_seed3_20260727_022547.csv"
    need(sha256(native_log) == full["failure_log_sha256"], "claim-6 native log drift")
    need(sha256(epoch_csv) == full["epoch_csv_sha256"], "claim-6 epoch record drift")
    need(sha256(step_csv) == full["step_csv_sha256"], "claim-6 step record drift")
    utils_text = (ROOT / "source_current/baselines/cvxpylayers_local/utils.py").read_text(encoding="utf-8")
    need("# import diffcp_lpgd" in utils_text and "mode='lpgd'" in utils_text, "claim-6 source mechanism drift")

    for row in matrix["claims"]:
        for key in ("oracle_artifacts", "control_artifacts", "independent_evidence", "executed_outputs"):
            for relative in row[key]:
                path = ROOT / relative
                need(path.is_file() and path.stat().st_size > 0, f"missing evidence: {relative}")

    environment = {
        **os.environ,
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTHONHASHSEED": "0",
        "PYTHONWARNINGS": "error",
    }
    replay_hashes = []
    for repetition in range(2):
        with tempfile.TemporaryDirectory(prefix=f"ffolayer-replay-{repetition}-") as temporary:
            output = Path(temporary) / "outputs"
            subprocess.run(
                [sys.executable, str(ROOT / "native_claim_audit.py"), "--output-dir", str(output)],
                cwd=ROOT,
                env=environment,
                check=True,
                stdout=subprocess.DEVNULL,
                timeout=120,
            )
            compare_generated(output)
            replay_hashes.append(sha256(output / "claim1_rate_summary.json"))
    need(len(set(replay_hashes)) == 1, "paired replay hashes differ")

    entries = validate_manifest()
    print(json.dumps({
        "status": "PASS",
        "claims": "6/6",
        "expected_points": "12/12",
        "literal_falsifications": 2,
        "native_program_objective_pairs": 9,
        "rate_trials": 22,
        "paired_warning_strict_replays": 2,
        "manifest_entries": entries,
    }, sort_keys=True))


if __name__ == "__main__":
    main()