bundle/v0/tests/test_hypergradient_validation.py

from __future__ import annotations

import importlib.util
import hashlib
import json
import os
import subprocess
import sys
from pathlib import Path

import numpy as np
import pytest

from loss_aware_dro_repro.core import ContractError, canonical_bytes, load_json, sha256_value
from loss_aware_dro_repro.hypergradient_validation import (
    _central_difference,
    _validate_config,
    build_hypergradient_validation,
    hypergradient_git_snapshot,
)


LANE = Path(__file__).resolve().parents[1]
CONFIG = LANE / "configs" / "hypergradient_validation_v1.json"
SPEC = importlib.util.spec_from_file_location(
    "hypergradient_validator", LANE / "scripts" / "validate_hypergradient.py"
)
assert SPEC and SPEC.loader
VALIDATOR = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(VALIDATOR)


@pytest.fixture(scope="module")
def artifact() -> dict:
    result = build_hypergradient_validation(CONFIG, require_clean=False)
    # Unit-test execution necessarily sees the new test files as dirty. The live
    # artifact runner is separately required to establish a clean snapshot.
    result["bindings"]["working_tree_clean"] = True
    return result


@pytest.fixture()
def artifact_path(tmp_path: Path, artifact: dict) -> Path:
    path = tmp_path / "hypergradient.json"
    _write_artifact(path, artifact)
    return path


def _write_artifact(path: Path, payload: dict) -> None:
    raw = canonical_bytes(payload) + b"\n"
    path.write_bytes(raw)
    path.with_suffix(".sha256").write_text(
        f"{hashlib.sha256(raw).hexdigest()}  {path.name}\n",
        encoding="ascii",
        newline="\n",
    )


def test_real_component_validation_covers_every_route_and_passes(artifact: dict):
    evidence = artifact["evidence_payload"]
    assert evidence["failures"] == []
    assert evidence["all_pass"] is True
    assert {row["route_family"] for row in evidence["routes"]} == {
        "gaussian_portfolio",
        "empirical_w1_portfolio",
        "absolute_regression",
        "squared_regression",
    }
    for route in evidence["routes"]:
        assert {
            "distance",
            "lower_value",
            "active_lower_value",
            "smoothed_violation",
            "active_coverage_penalty",
            "total_hypergradient",
        }.issubset(route["components"])
        assert all(component["passed"] for component in route["components"].values())
        assert all(
            component["passing_step_count"] == 3
            for component in route["components"].values()
        )
    squared = next(row for row in evidence["routes"] if row["route_family"] == "squared_regression")
    assert squared["components"]["root_lower_value"]["passed"] is True


def test_validator_independently_regenerates_exact_evidence(artifact_path: Path):
    assert VALIDATOR.validate(artifact_path, CONFIG, recompute=True) == []


def test_validator_rejects_payload_tamper_even_without_recomputation(
    artifact_path: Path,
):
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    payload["evidence_payload"]["routes"][0]["components"]["distance"]["analytic"][0][0] += 1.0
    _write_artifact(artifact_path, payload)
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert "evidence payload hash mismatch" in errors
    assert any("relative error is not reproducible" in error for error in errors)


def test_validator_rejects_rehashed_residual_forgery(artifact_path: Path):
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    diagnostics = payload["evidence_payload"]["routes"][0]["components"]["lower_value"][
        "analytic_diagnostics"
    ]
    diagnostics["residuals"]["solver_native_primal"] = 1.0
    payload["evidence_payload_hash"] = sha256_value(payload["evidence_payload"])
    _write_artifact(artifact_path, payload)
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert any("conic residual maximum is not reproducible" in error for error in errors)
    assert any("conic residual contract failed" in error for error in errors)


def test_validator_rejects_rehashed_numerical_matrix_without_scalar_lineage(
    artifact_path: Path,
):
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    step = payload["evidence_payload"]["routes"][0]["components"]["distance"]["steps"][0]
    step["finite_difference"][0][0] += 0.01
    analytic = np.asarray(
        payload["evidence_payload"]["routes"][0]["components"]["distance"]["analytic"]
    )
    numerical = np.asarray(step["finite_difference"])
    denominator = max(
        float(np.linalg.norm(analytic, ord="fro")),
        float(np.linalg.norm(numerical, ord="fro")),
        1e-12,
    )
    step["relative_error"] = float(
        np.linalg.norm(analytic - numerical, ord="fro") / denominator
    )
    step["max_absolute_error"] = float(
        np.max(np.abs(analytic - numerical))
    )
    step["passed"] = False
    payload["evidence_payload_hash"] = sha256_value(payload["evidence_payload"])
    _write_artifact(artifact_path, payload)
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert any("does not equal its retained scalar evaluations" in error for error in errors)


def test_validator_rejects_rehashed_missing_component(artifact_path: Path):
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    del payload["evidence_payload"]["routes"][2]["components"]["smoothed_violation"]
    payload["evidence_payload_hash"] = sha256_value(payload["evidence_payload"])
    _write_artifact(artifact_path, payload)
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert "empirical_w1_portfolio: component set is not closed" in errors


def test_validator_rejects_source_tree_rebinding(artifact_path: Path):
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    payload["bindings"]["source_tree_hash"] = "sha256:" + "0" * 64
    _write_artifact(artifact_path, payload)
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert "live scientific source/input tree differs from artifact" in errors


def test_validator_rejects_valid_but_older_commit_rebinding(artifact_path: Path):
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    current = payload["bindings"]["base_commit"]
    candidates = subprocess.check_output(
        ["git", "rev-list", f"{current}^"],
        cwd=LANE,
        text=True,
    ).splitlines()
    older = None
    for candidate in candidates:
        try:
            _, files = hypergradient_git_snapshot(candidate, CONFIG)
        except ContractError:
            continue
        if files != payload["bindings"]["scientific_files"]:
            older = candidate
            break
    assert older is not None, "test history must contain an older, scientifically different commit"
    payload["bindings"]["base_commit"] = older
    _write_artifact(artifact_path, payload)
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert any(
        "base tree does not match" in error
        or "scientific file manifest differs" in error
        or "live scientific files differ" in error
        for error in errors
    )


def test_validator_rejects_bogus_adjacent_sidecar(artifact_path: Path):
    artifact_path.with_suffix(".sha256").write_text(
        f"{'0' * 64}  {artifact_path.name}\n",
        encoding="ascii",
        newline="\n",
    )
    errors = VALIDATOR.validate(artifact_path, CONFIG, recompute=False)
    assert "adjacent SHA-256 sidecar does not match the artifact bytes and filename" in errors


def test_structural_only_reports_that_regeneration_was_not_run(artifact_path: Path):
    environment = {**os.environ, **{
        "OMP_NUM_THREADS": "1",
        "MKL_NUM_THREADS": "1",
        "OPENBLAS_NUM_THREADS": "1",
        "NUMEXPR_NUM_THREADS": "1",
    }}
    result = subprocess.run(
        [
            sys.executable,
            str(LANE / "scripts" / "validate_hypergradient.py"),
            str(artifact_path),
            "--config",
            str(CONFIG),
            "--structural-only",
        ],
        cwd=LANE,
        env=environment,
        text=True,
        capture_output=True,
        check=False,
    )
    assert result.returncode == 0, result.stdout + result.stderr
    assert "scientific regeneration was NOT run" in result.stdout
    assert "independent regeneration agree" not in result.stdout


def test_finite_difference_retains_each_side_failure():
    failures: list[dict] = []

    def broken(_):
        raise RuntimeError("deliberate solver failure")

    numerical, evaluations = _central_difference(
        broken,
        point=np.eye(2),
        step=1e-5,
        route_id="route",
        component="component",
        failures=failures,
    )
    assert numerical is None
    assert len(evaluations) == 3
    assert len(failures) == 6
    assert {(row["coordinate"][0], row["coordinate"][1]) for row in failures} == {
        (0, 0),
        (1, 0),
        (1, 1),
    }
    assert {row["side"] for row in failures} == {"plus", "minus"}


def test_config_refuses_route_family_omission():
    config = load_json(CONFIG)
    config["routes"] = config["routes"][:-1]
    with pytest.raises(ContractError, match="cover exactly"):
        _validate_config(config)