#!/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()