lpgd_release_compatibility_audit.py

#!/usr/bin/env python3
"""Audit the released LPGD path and its valid-mode control.

The paper's repository registers LPGD through its local CvxpyLayer wrapper.
That wrapper passes ``mode='lpgd'`` to the declared ``diffcp`` dependency.
The released dependency rejects that mode.  This audit preserves the exact
full 9x9 failure and reproduces the compatibility split on the smallest
released Sudoku instance without altering either implementation.
"""

from __future__ import annotations

import hashlib
import importlib.metadata
import json
import sys
import warnings
from pathlib import Path

import torch


ROOT = Path(__file__).resolve().parent
SOURCE = ROOT / "source_current"
OUTPUT = ROOT / "outputs" / "claim6_lpgd_release_failure.json"
FULL_LOG = ROOT / "sudoku_results_8" / "lpgd" / "central_failures.log"
FULL_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"
ERROR = "Unsupported mode lpgd; the supported modes are 'dense', 'lsqr' and 'lsmr'"


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


def run_control(method: str) -> dict:
    sudoku = SOURCE / "sudoku"
    sys.path.insert(0, str(sudoku))
    from models_sudoku import SingleOptLayerSudoku  # noqa: PLC0415

    torch.manual_seed(3)
    x = torch.zeros((1, 4, 4, 4), dtype=torch.float32)
    model = SingleOptLayerSudoku(
        2,
        learnable_parts=["eq"],
        layer_type=method,
        batch_size=1,
    ).to("cpu")
    captured: list[str] = []
    try:
        with warnings.catch_warnings(record=True) as seen:
            warnings.simplefilter("always")
            y = model(x)
        captured = sorted({str(item.message) for item in seen})
        return {
            "method": method,
            "status": "pass",
            "output_shape": list(y.shape),
            "all_finite": bool(torch.isfinite(y).all()),
            "warnings": captured,
        }
    except Exception as error:  # exact exception is the measured result
        return {
            "method": method,
            "status": "fail",
            "exception_type": type(error).__name__,
            "exception": str(error),
            "warnings": captured,
        }
    finally:
        sys.path.remove(str(sudoku))


def main() -> None:
    log = FULL_LOG.read_text(encoding="utf-8")
    full_rows = FULL_CSV.read_text(encoding="utf-8").splitlines()
    step_rows = STEP_CSV.read_text(encoding="utf-8").splitlines()
    if ERROR not in log:
        raise RuntimeError("full released LPGD failure is absent from the native log")
    if len(full_rows) != 1 or len(step_rows) != 1:
        raise RuntimeError("full LPGD run unexpectedly completed a training record")

    utils = SOURCE / "baselines" / "cvxpylayers_local" / "utils.py"
    utils_text = utils.read_text(encoding="utf-8")
    if "# import diffcp_lpgd" not in utils_text:
        raise RuntimeError("pinned commented LPGD-fork import changed")
    if "mode='lpgd'" not in utils_text:
        raise RuntimeError("pinned LPGD mode branch changed")

    valid = run_control("cvxpylayer")
    invalid = run_control("lpgd")
    if valid.get("status") != "pass" or valid.get("all_finite") is not True:
        raise RuntimeError("valid-mode released control did not produce a finite solution")
    if invalid.get("status") != "fail" or invalid.get("exception") != ERROR:
        raise RuntimeError("released LPGD compatibility failure did not reproduce exactly")

    result = {
        "schema_version": 1,
        "source_lock": {
            "repository": "GT-KOALA/FFOLayer",
            "commit": "28905f3e1750fca5b8918954d5d2ea5bed0cbacc",
            "tree": "f236d623acd0a089adebafd61c7c239434c9e6b2",
            "utils_sha256": sha256(utils),
            "models_sudoku_sha256": sha256(SOURCE / "sudoku" / "models_sudoku.py"),
            "main_sudoku_sha256": sha256(SOURCE / "sudoku" / "main_sudoku.py"),
        },
        "environment": {
            "python": ".".join(map(str, sys.version_info[:3])),
            "torch": importlib.metadata.version("torch"),
            "cvxpy": importlib.metadata.version("cvxpy"),
            "diffcp": importlib.metadata.version("diffcp"),
            "scs": importlib.metadata.version("scs"),
        },
        "full_native_9x9_attempt": {
            "command": "python sudoku/main_sudoku.py --method lpgd --n 3 --epochs 1 --batch_size 8 --seed 3 --device cpu",
            "dataset": "released 10,000-puzzle 9x9 Sudoku dataset (9,000 train; 1,000 test)",
            "train_batches_requested": 1125,
            "completed_train_records": 0,
            "failure_phase": "first forward pass of training batch 0",
            "exception_type": "ValueError",
            "exception": ERROR,
            "failure_log_sha256": sha256(FULL_LOG),
            "epoch_csv_sha256": sha256(FULL_CSV),
            "step_csv_sha256": sha256(STEP_CSV),
        },
        "released_micro_control": {
            "instance": "released n=2 Sudoku layer, batch=1, seed=3, all-zero puzzle tensor",
            "valid_diffcp_mode_path": valid,
            "registered_lpgd_mode_path": invalid,
        },
        "source_mechanism": {
            "active_import": "import diffcp",
            "inactive_import": "# import diffcp_lpgd",
            "registered_call": "diffcp.solve_and_derivative_batch(..., mode='lpgd', derivative_kwargs={'tau': 1e-4, 'rho': 0.1})",
        },
        "literal_result": "The released LPGD path cannot execute the registered comparison under the repository's declared diffcp dependency; the same released problem succeeds through the supported cvxpylayer/lsqr control.",
        "verdict": "falsified_as_literally_registered",
        "scope": "This is a release-compatibility falsification, not evidence that a separately patched or unpublished LPGD fork cannot outperform FFOLayer.",
    }
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(json.dumps(result, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()