native_claim_audit.py

#!/usr/bin/env python3
"""Deterministic core audit reused across the six live FFOLayer claims.

The rate panel executes the paper's perturbed first-order construction twice
at seven tolerances.  The benchmark report structures measurements produced by
the unmodified released entrypoints; it deliberately retains the backward-only
counterexample used to falsify registered claim 4.  The other claim-matched
audits are persisted separately and checked by ``validate_evidence.py``.
"""

from __future__ import annotations

import argparse
import csv
import json
import math
from pathlib import Path

import numpy as np


ROOT = Path(__file__).resolve().parent
COMMIT = "28905f3e1750fca5b8918954d5d2ea5bed0cbacc"


def write_json(path: Path, payload: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def rate_panel(output_dir: Path) -> dict:
    epsilons = (1e-1, 5e-2, 2e-2, 1e-2, 5e-3, 2e-3, 1e-3)
    repetitions = (
        (np.diag([1.0, 4.0]), np.array([0.7, -1.1])),
        (np.diag([0.8, 3.2]), np.array([-0.6, 0.9])),
    )
    rows: list[dict] = []
    for repetition, (matrix, outer_gradient) in enumerate(repetitions):
        for epsilon in epsilons:
            delta = epsilon
            system = matrix + delta * np.eye(2)
            rhs = matrix @ outer_gradient
            step = 1.0 / np.linalg.eigvalsh(system).max()
            y = np.zeros(2)
            evaluations = 0
            while np.linalg.norm(system @ y - rhs) > delta * delta:
                y -= step * (system @ y - rhs)
                evaluations += 1
                if evaluations > 100_000:
                    raise RuntimeError("first-order perturbed solve did not converge")
            estimate = matrix @ (outer_gradient - y) / delta
            error = float(np.linalg.norm(estimate - outer_gradient))
            rows.append(
                {
                    "repetition": repetition,
                    "epsilon": epsilon,
                    "inverse_epsilon_scale": int(round(1.0 / epsilon)),
                    "gradient_oracle_evaluations": evaluations,
                    "residual_norm": float(np.linalg.norm(system @ y - rhs)),
                    "hypergradient_l2_error": error,
                    "error_le_2epsilon": error <= 2.0 * epsilon,
                }
            )

    csv_path = output_dir / "claim1_rate_repetitions.csv"
    with csv_path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)

    per_scale = []
    for epsilon in epsilons:
        subset = [row for row in rows if row["epsilon"] == epsilon]
        per_scale.append(
            {
                "epsilon": epsilon,
                "inverse_epsilon_scale": int(round(1.0 / epsilon)),
                "repetitions": len(subset),
                "mean_gradient_oracle_evaluations": float(
                    np.mean([row["gradient_oracle_evaluations"] for row in subset])
                ),
                "max_hypergradient_l2_error": max(
                    row["hypergradient_l2_error"] for row in subset
                ),
            }
        )
    logs = np.log([row["inverse_epsilon_scale"] for row in per_scale])
    means = np.array([row["mean_gradient_oracle_evaluations"] for row in per_scale])
    slope, intercept = np.polyfit(logs, means, 1)
    fitted = slope * logs + intercept
    r_squared = 1.0 - float(np.sum((means - fitted) ** 2) / np.sum((means - means.mean()) ** 2))

    # Destructive control: removing strong convexity leaves one direction
    # unidentifiable.  The same estimator must then fail the 2*epsilon target.
    epsilon = 1e-3
    singular = np.diag([0.0, 4.0])
    outer_gradient = np.array([0.7, -1.1])
    system = singular + epsilon * np.eye(2)
    rhs = singular @ outer_gradient
    step = 1.0 / np.linalg.eigvalsh(system).max()
    y = np.zeros(2)
    evaluations = 0
    while np.linalg.norm(system @ y - rhs) > epsilon * epsilon:
        y -= step * (system @ y - rhs)
        evaluations += 1
    estimate = singular @ (outer_gradient - y) / epsilon
    singular_error = float(np.linalg.norm(estimate - outer_gradient))

    summary = {
        "official_repository_commit": COMMIT,
        "mechanism": "paper perturbed lower solve using gradient evaluations only",
        "scales": [row["inverse_epsilon_scale"] for row in per_scale],
        "repetitions_per_scale": 2,
        "rows": len(rows),
        "all_errors_le_2epsilon": all(row["error_le_2epsilon"] for row in rows),
        "oracle_evaluations_vs_log_inverse_epsilon_slope": float(slope),
        "oracle_evaluations_vs_log_inverse_epsilon_intercept": float(intercept),
        "oracle_evaluations_log_fit_r_squared": r_squared,
        "per_scale": per_scale,
        "destructive_control": {
            "change": "set one lower-Hessian eigenvalue to zero, violating strong convexity",
            "epsilon": epsilon,
            "gradient_oracle_evaluations": evaluations,
            "hypergradient_l2_error": singular_error,
            "error_le_2epsilon": singular_error <= 2.0 * epsilon,
            "control_triggered": singular_error > 2.0 * epsilon,
        },
    }
    write_json(output_dir / "claim1_rate_summary.json", summary)
    return summary


def benchmark_report(output_dir: Path) -> dict:
    # These values are emitted by the unmodified source_current entrypoint at
    # ydim=800, batch_size=200, one epoch, seed=3 on CPU.  Precision is exactly
    # the precision retained from the run transcript.
    rows = [
        {
            "method": "ffocp_eq",
            "test_df_loss": -0.4995159513,
            "forward_seconds": 54.6145,
            "backward_seconds": 51.7374,
        },
        {
            "method": "qpth",
            "test_df_loss": -0.4996193552,
            "forward_seconds": 512.8602,
            "backward_seconds": 45.7594,
        },
    ]
    for row in rows:
        row["total_seconds"] = row["forward_seconds"] + row["backward_seconds"]
    ffo, qpth = rows
    total_speedup = qpth["total_seconds"] / ffo["total_seconds"]
    backward_speedup = qpth["backward_seconds"] / ffo["backward_seconds"]
    endpoint_gap = abs(ffo["test_df_loss"] - qpth["test_df_loss"])

    csv_path = output_dir / "claim2_native_synthetic_benchmark.csv"
    with csv_path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)

    report = {
        "official_repository_commit": COMMIT,
        "entrypoint": "synthetic_task/main_synthetic.py",
        "command_scope": {
            "ydim": 800,
            "batch_size": 200,
            "epochs": 1,
            "seed": 3,
            "device": "cpu",
            "training_samples": 1600,
            "test_samples": 400,
        },
        "measurements": rows,
        "total_computation_speedup_ffolayer_over_qpth": total_speedup,
        "absolute_test_df_loss_gap": endpoint_gap,
        "similar_endpoint_threshold": 5e-4,
        "similar_endpoint": endpoint_gap <= 5e-4,
        "substantially_faster_total_threshold": 2.0,
        "substantially_faster_total": total_speedup >= 2.0,
        "destructive_boundary_control": {
            "scope": "backward phase only rather than total computation",
            "qpth_over_ffolayer_speedup": backward_speedup,
            "ffolayer_faster_on_backward_only": backward_speedup > 1.0,
            "control_triggered": backward_speedup < 1.0,
            "interpretation": "The direct CPU run supports total computation, not a backward-only speed claim.",
        },
        "measurement_precision": "losses retained to 10 decimal places and timings to 4 decimal places from the official entrypoint transcript",
    }
    write_json(output_dir / "claim2_native_synthetic_benchmark.json", report)
    return report


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", type=Path, default=ROOT / "outputs")
    args = parser.parse_args()
    args.output_dir.mkdir(parents=True, exist_ok=True)
    rate = rate_panel(args.output_dir)
    benchmark = benchmark_report(args.output_dir)
    status = {
        "status": "PASS",
        "claim1_rate_rows": rate["rows"],
        "claim1_control_triggered": rate["destructive_control"]["control_triggered"],
        "claim2_total_speedup": benchmark["total_computation_speedup_ffolayer_over_qpth"],
        "claim2_similar_endpoint": benchmark["similar_endpoint"],
        "claim2_control_triggered": benchmark["destructive_boundary_control"]["control_triggered"],
    }
    print(json.dumps(status, sort_keys=True))


if __name__ == "__main__":
    main()