bundle/v0/scripts/validate_hypergradient.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any

for variable in (
    "OMP_NUM_THREADS",
    "MKL_NUM_THREADS",
    "OPENBLAS_NUM_THREADS",
    "NUMEXPR_NUM_THREADS",
):
    os.environ[variable] = "1"

import numpy as np


LANE_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(LANE_ROOT / "src"))

from loss_aware_dro_repro.core import (  # noqa: E402
    load_json,
    load_plan,
    plan_hash,
    sha256_value,
)
from loss_aware_dro_repro.datasets import generate_dataset  # noqa: E402
from loss_aware_dro_repro.hypergradient_validation import (  # noqa: E402
    _validate_config,
    build_hypergradient_validation,
    hypergradient_git_snapshot,
    hypergradient_live_file_bindings,
    hypergradient_source_tree_hash,
    locked_regeneration_environment,
)
from loss_aware_dro_repro.matrix import expand_tasks  # noqa: E402
from loss_aware_dro_repro.residuals import conic_residual_maximum  # noqa: E402


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


def _inside(path: Path, root: Path) -> bool:
    path = path.resolve()
    root = root.resolve()
    return path == root or root in path.parents


def _validate_sidecar(artifact_path: Path, errors: list[str]) -> None:
    sidecar_path = artifact_path.with_suffix(".sha256")
    try:
        raw = sidecar_path.read_text(encoding="ascii")
    except (OSError, UnicodeError) as exc:
        errors.append(f"adjacent SHA-256 sidecar is missing or unreadable: {exc}")
        return
    expected = f"{_hash_file(artifact_path)}  {artifact_path.name}\n"
    if raw != expected:
        errors.append("adjacent SHA-256 sidecar does not match the artifact bytes and filename")


def _finite_matrix(raw: Any, dimension: int) -> np.ndarray | None:
    try:
        matrix = np.asarray(raw, dtype=np.float64)
    except (TypeError, ValueError):
        return None
    if matrix.shape != (dimension, dimension) or not np.all(np.isfinite(matrix)):
        return None
    if not np.array_equal(matrix, np.tril(matrix)):
        return None
    return matrix


def _relative_error(analytic: np.ndarray, numerical: np.ndarray) -> float:
    denominator = max(
        float(np.linalg.norm(analytic, ord="fro")),
        float(np.linalg.norm(numerical, ord="fro")),
        1e-12,
    )
    return float(np.linalg.norm(analytic - numerical, ord="fro") / denominator)


def _validate_diagnostics(
    value: Any,
    *,
    solver_threshold: float,
    transport_threshold: float,
    label: str,
    errors: list[str],
) -> None:
    if isinstance(value, list):
        for index, child in enumerate(value):
            _validate_diagnostics(
                child,
                solver_threshold=solver_threshold,
                transport_threshold=transport_threshold,
                label=f"{label}[{index}]",
                errors=errors,
            )
        return
    if not isinstance(value, dict):
        return
    kind = value.get("kind")
    if kind == "conic":
        residuals = value.get("residuals")
        try:
            maximum = conic_residual_maximum(residuals)
        except Exception as exc:
            errors.append(f"{label}: invalid conic residual map: {exc}")
        else:
            if value.get("status") not in {"optimal", "optimal_inaccurate"}:
                errors.append(f"{label}: unaccepted conic status")
            if value.get("threshold") != solver_threshold:
                errors.append(f"{label}: conic threshold mismatch")
            if not math.isclose(
                float(value.get("residual_maximum", math.inf)), maximum, rel_tol=0.0, abs_tol=0.0
            ):
                errors.append(f"{label}: conic residual maximum is not reproducible")
            if maximum > solver_threshold or value.get("passed") is not True:
                errors.append(f"{label}: conic residual contract failed")
    elif kind == "transport":
        residuals = value.get("residuals")
        if not isinstance(residuals, dict) or set(residuals) != {
            "marginal_row",
            "marginal_column",
            "dual_feasibility_relative",
            "complementary_slackness_relative",
            "duality_gap_relative",
        }:
            errors.append(f"{label}: invalid transport residual map")
        else:
            try:
                numbers = [float(item) for item in residuals.values()]
            except (TypeError, ValueError):
                numbers = [math.inf]
            if any(not math.isfinite(item) or item < 0 for item in numbers):
                errors.append(f"{label}: non-finite or negative transport residual")
            maximum = max(numbers)
            if value.get("threshold") != transport_threshold:
                errors.append(f"{label}: transport threshold mismatch")
            if not math.isclose(
                float(value.get("residual_maximum", math.inf)), maximum, rel_tol=0.0, abs_tol=0.0
            ):
                errors.append(f"{label}: transport residual maximum is not reproducible")
            if (
                maximum > transport_threshold
                or value.get("solver_result_code") != 1
                or value.get("solver_warning") is not None
                or value.get("passed") is not True
            ):
                errors.append(f"{label}: transport residual contract failed")
    for key, child in value.items():
        if key not in {"residuals"}:
            _validate_diagnostics(
                child,
                solver_threshold=solver_threshold,
                transport_threshold=transport_threshold,
                label=f"{label}.{key}",
                errors=errors,
            )


def _validate_component(
    component: dict[str, Any],
    *,
    dimension: int,
    config: dict[str, Any],
    label: str,
    errors: list[str],
) -> None:
    analytic = _finite_matrix(component.get("analytic"), dimension)
    point = _finite_matrix(component.get("evaluation_point"), dimension)
    if analytic is None:
        errors.append(f"{label}: analytic gradient is not a finite lower-triangular matrix")
    if point is None:
        errors.append(f"{label}: evaluation point is not a finite lower-triangular matrix")
    elif component.get("evaluation_point_hash") != sha256_value(point.tolist()):
        errors.append(f"{label}: evaluation point hash mismatch")
    steps = component.get("steps")
    if not isinstance(steps, list) or [row.get("step") for row in steps] != [
        float(value) for value in config["central_difference_steps"]
    ]:
        errors.append(f"{label}: central-difference step grid mismatch")
        return
    coordinates = [[row, column] for row in range(dimension) for column in range(row + 1)]
    passing_count = 0
    for step_index, step_row in enumerate(steps):
        step_label = f"{label}.steps[{step_index}]"
        numerical = _finite_matrix(step_row.get("finite_difference"), dimension)
        evaluations = step_row.get("evaluations")
        reconstructed = np.zeros((dimension, dimension), dtype=np.float64)
        reconstruction_complete = True
        if not isinstance(evaluations, list) or [row.get("coordinate") for row in evaluations] != coordinates:
            errors.append(f"{step_label}: finite-difference evaluation coordinates are not count-closed")
            reconstruction_complete = False
        else:
            for coordinate_index, evaluation in enumerate(evaluations):
                side_values: dict[str, float] = {}
                for side in ("plus", "minus"):
                    side_row = evaluation.get(side)
                    if not isinstance(side_row, dict) or "failure" in side_row:
                        errors.append(
                            f"{step_label}.evaluations[{coordinate_index}].{side}: missing successful scalar evaluation"
                        )
                        reconstruction_complete = False
                        continue
                    try:
                        scalar = float(side_row.get("value"))
                    except (TypeError, ValueError):
                        scalar = math.nan
                    if not math.isfinite(scalar):
                        errors.append(
                            f"{step_label}.evaluations[{coordinate_index}].{side}: scalar is non-finite"
                        )
                        reconstruction_complete = False
                    else:
                        side_values[side] = scalar
                    _validate_diagnostics(
                        side_row.get("diagnostics"),
                        solver_threshold=config["solver_residual_max"],
                        transport_threshold=config["transport_residual_max"],
                        label=f"{step_label}.evaluations[{coordinate_index}].{side}.diagnostics",
                        errors=errors,
                    )
                if set(side_values) == {"plus", "minus"}:
                    row, column = coordinates[coordinate_index]
                    reconstructed[row, column] = (
                        side_values["plus"] - side_values["minus"]
                    ) / (2.0 * float(step_row["step"]))
                else:
                    reconstruction_complete = False
        if numerical is not None and reconstruction_complete and not np.array_equal(
            numerical, reconstructed
        ):
            errors.append(
                f"{step_label}: finite-difference matrix does not equal its retained scalar evaluations"
            )
        if analytic is None or numerical is None:
            errors.append(f"{step_label}: numerical gradient is invalid")
            continue
        relative = _relative_error(analytic, numerical)
        maximum_absolute = float(np.max(np.abs(analytic - numerical), initial=0.0))
        expected_pass = bool(
            relative <= config["relative_error_threshold"]
            or maximum_absolute <= config["absolute_error_threshold"]
        )
        if not math.isclose(
            float(step_row.get("relative_error", math.inf)), relative, rel_tol=1e-15, abs_tol=1e-15
        ):
            errors.append(f"{step_label}: relative error is not reproducible")
        if not math.isclose(
            float(step_row.get("max_absolute_error", math.inf)),
            maximum_absolute,
            rel_tol=1e-15,
            abs_tol=1e-15,
        ):
            errors.append(f"{step_label}: maximum absolute error is not reproducible")
        if step_row.get("passed") is not expected_pass:
            errors.append(f"{step_label}: pass flag disagrees with frozen thresholds")
        passing_count += int(expected_pass)
    primary_index = config["central_difference_steps"].index(config["primary_step"])
    expected_component_pass = bool(
        steps[primary_index].get("passed") is True
        and passing_count >= config["step_pass_policy"]["minimum_passing_steps"]
    )
    if component.get("primary_step") != config["primary_step"]:
        errors.append(f"{label}: primary step mismatch")
    if component.get("passing_step_count") != passing_count:
        errors.append(f"{label}: passing step count mismatch")
    if component.get("passed") is not expected_component_pass:
        errors.append(f"{label}: component pass flag mismatch")
    _validate_diagnostics(
        component.get("analytic_diagnostics"),
        solver_threshold=config["solver_residual_max"],
        transport_threshold=config["transport_residual_max"],
        label=f"{label}.analytic_diagnostics",
        errors=errors,
    )


def validate(
    artifact_path: Path,
    config_path: Path,
    *,
    recompute: bool = True,
) -> list[str]:
    errors: list[str] = []
    artifact_path = artifact_path.resolve()
    config_path = config_path.resolve()
    try:
        artifact = json.loads(artifact_path.read_text(encoding="utf-8"))
        config = load_json(config_path)
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        return [f"cannot read hypergradient artifact/config: {exc}"]
    _validate_sidecar(artifact_path, errors)
    if artifact.get("schema_version") != 1 or config.get("schema_version") != 1:
        errors.append("unsupported artifact or config schema")
    try:
        _validate_config(config)
    except Exception as exc:
        errors.append(f"invalid frozen hypergradient config: {exc}")
    if artifact.get("validation_id") != config.get("validation_id"):
        errors.append("validation id mismatch")
    if artifact.get("evidence_scale") != "BOUNDED_COMPONENT_VALIDATION_NOT_PAPER_SCALE":
        errors.append("artifact evidence scale is not the bounded validation class")
    if artifact.get("claim_eligible") is not False:
        errors.append("bounded hypergradient artifact must not be claim eligible")
    if artifact.get("scientific_verdicts") != {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"}:
        errors.append("scientific verdicts must remain HOLD")
    if artifact.get("authority") != config.get("authority") or any(
        artifact.get("authority", {}).values()
    ):
        errors.append("artifact authority must match the all-denied config")

    bindings = artifact.get("bindings", {})
    config_binding = bindings.get("config", {})
    if config_binding.get("path") != config_path.relative_to(LANE_ROOT).as_posix():
        errors.append("config path binding mismatch")
    if config_binding.get("byte_sha256") != _hash_file(config_path):
        errors.append("config byte hash mismatch")
    if config_binding.get("canonical_hash") != sha256_value(config):
        errors.append("config canonical hash mismatch")
    plan_path = (LANE_ROOT / config["paper_plan"]).resolve()
    if not _inside(plan_path, LANE_ROOT) or not plan_path.is_file():
        errors.append("paper plan path is missing or escaping")
        plan = {}
    else:
        plan = load_plan(plan_path)
        plan_binding = bindings.get("plan", {})
        if plan_binding.get("path") != plan_path.relative_to(LANE_ROOT).as_posix():
            errors.append("plan path binding mismatch")
        if plan_binding.get("byte_sha256") != _hash_file(plan_path):
            errors.append("plan byte hash mismatch")
        if plan_binding.get("canonical_hash") != plan_hash(plan):
            errors.append("plan canonical hash mismatch")
        if bindings.get("reference_source_commit") != plan["reference_source"]["commit"]:
            errors.append("reference source commit mismatch")
    if bindings.get("source_tree_hash") != hypergradient_source_tree_hash(config_path):
        errors.append("live scientific source/input tree differs from artifact")
    commit = bindings.get("base_commit")
    if not isinstance(commit, str) or re.fullmatch(r"[0-9a-f]{40}", commit) is None:
        errors.append("base commit is missing or malformed")
    else:
        result = subprocess.run(
            ["git", "cat-file", "-e", f"{commit}^{{commit}}"],
            cwd=LANE_ROOT,
            capture_output=True,
            text=True,
            check=False,
        )
        if result.returncode != 0:
            errors.append("bound base commit does not exist in this repository")
        else:
            try:
                base_tree, committed_files = hypergradient_git_snapshot(commit, config_path)
                live_files = hypergradient_live_file_bindings(config_path)
            except Exception as exc:
                errors.append(f"cannot prove bound scientific files against Git objects: {exc}")
            else:
                if bindings.get("base_tree") != base_tree:
                    errors.append("bound base tree does not match the base commit Git tree")
                if bindings.get("scientific_files") != committed_files:
                    errors.append("bound scientific file manifest differs from base-commit Git blobs")
                committed_sha256 = {
                    path: {"byte_sha256": row["byte_sha256"]}
                    for path, row in committed_files.items()
                }
                if committed_sha256 != live_files:
                    errors.append("live scientific files differ from the base-commit Git blobs")
    if bindings.get("working_tree_clean") is not True:
        errors.append("artifact was not generated from a clean paper lane")

    expected_source_configs: dict[str, Any] = {}
    for route_config in config["routes"]:
        path = (LANE_ROOT / route_config["source_config"]).resolve()
        source_config = load_json(path)
        expected_source_configs[path.relative_to(LANE_ROOT).as_posix()] = {
            "byte_sha256": _hash_file(path),
            "canonical_hash": sha256_value(source_config),
        }
    if bindings.get("source_configs") != expected_source_configs:
        errors.append("source config bindings mismatch")
    environment_path = LANE_ROOT / "environment" / "scientific-freeze.txt"
    if bindings.get("environment_lock") != {
        "path": "environment/scientific-freeze.txt",
        "byte_sha256": _hash_file(environment_path),
    }:
        errors.append("environment lock binding mismatch")
    direct_path = LANE_ROOT / "environment" / "requirements.lock"
    if bindings.get("direct_requirements") != {
        "path": "environment/requirements.lock",
        "byte_sha256": _hash_file(direct_path),
    }:
        errors.append("direct requirements binding mismatch")
    artifact_environment = artifact.get("environment")
    if not isinstance(artifact_environment, dict):
        errors.append("locked regeneration environment receipt is missing")
    else:
        locks = artifact_environment.get("locks")
        if locks != {
            "transitive": bindings.get("environment_lock"),
            "direct": bindings.get("direct_requirements"),
        }:
            errors.append("environment receipt lock bindings differ from artifact bindings")
        packages = artifact_environment.get("packages")
        if not isinstance(packages, dict) or set(packages) != {
            "numpy", "scipy", "cvxpy", "clarabel", "POT"
        }:
            errors.append("environment receipt omits a canonical regeneration dependency")
        if artifact_environment.get("threads") != {
            "OMP_NUM_THREADS": "1",
            "MKL_NUM_THREADS": "1",
            "OPENBLAS_NUM_THREADS": "1",
            "NUMEXPR_NUM_THREADS": "1",
        }:
            errors.append("environment receipt does not prove the one-thread contract")
        if not isinstance(artifact_environment.get("cpu_model"), str) or not artifact_environment[
            "cpu_model"
        ].strip():
            errors.append("environment receipt omits the CPU model")
        blas = artifact_environment.get("blas")
        if not isinstance(blas, dict) or not isinstance(blas.get("name"), str) or not blas["name"]:
            errors.append("environment receipt omits the BLAS implementation")

    evidence = artifact.get("evidence_payload", {})
    if artifact.get("evidence_payload_hash") != sha256_value(evidence):
        errors.append("evidence payload hash mismatch")
    routes = evidence.get("routes")
    if not isinstance(routes, list) or [row.get("route_id") for row in routes] != [
        row["route_id"] for row in config["routes"]
    ]:
        errors.append("route list/order differs from frozen config")
        routes = []
    tasks = {task["task_id"]: task for task in expand_tasks(plan)} if plan else {}
    expected_datasets: dict[str, Any] = {}
    for route_config, route_result in zip(config["routes"], routes):
        task = tasks.get(route_config["task_selector"])
        if task is None:
            errors.append(f"{route_config['route_id']}: task missing from plan")
            continue
        samples, metadata = generate_dataset(task)
        expected_datasets[route_config["route_id"]] = {
            "task_id": task["task_id"],
            "task_hash": task["task_hash"],
            "fingerprint": metadata["fingerprint"],
            "shape": list(samples.shape),
            "seeds": task["seeds"],
        }
        label = route_config["route_id"]
        if route_result.get("route_family") != route_config["route_family"]:
            errors.append(f"{label}: route family mismatch")
        if route_result.get("implementation") != route_config["implementation"]:
            errors.append(f"{label}: implementation mismatch")
        if route_result.get("task_id") != task["task_id"] or route_result.get("task_hash") != task["task_hash"]:
            errors.append(f"{label}: task identity mismatch")
        components = route_result.get("components")
        required = list(config["required_components"])
        required.extend(config.get("additional_required_components", {}).get(route_config["route_family"], []))
        if not isinstance(components, dict) or set(components) != set(required):
            errors.append(f"{label}: component set is not closed")
            continue
        for component_name in required:
            _validate_component(
                components[component_name],
                dimension=samples.shape[1],
                config=config,
                label=f"{label}.{component_name}",
                errors=errors,
            )
        active_component_names = (
            "active_lower_value",
            "smoothed_violation",
            "active_coverage_penalty",
            "total_hypergradient",
        )
        active_points = [
            components[name].get("evaluation_point") for name in active_component_names
        ]
        if any(point != active_points[0] for point in active_points[1:]):
            errors.append(f"{label}: active component evaluation points differ")
        active_lower_gradient = _finite_matrix(
            components["active_lower_value"].get("analytic"), samples.shape[1]
        )
        violation_gradient = _finite_matrix(
            components["smoothed_violation"].get("analytic"), samples.shape[1]
        )
        penalty_gradient = _finite_matrix(
            components["active_coverage_penalty"].get("analytic"), samples.shape[1]
        )
        total_gradient = _finite_matrix(
            components["total_hypergradient"].get("analytic"), samples.shape[1]
        )
        source_config = load_json((LANE_ROOT / route_config["source_config"]).resolve())
        penalty_diagnostics = components["active_coverage_penalty"].get(
            "analytic_diagnostics", {}
        )
        total_diagnostics = components["total_hypergradient"].get(
            "analytic_diagnostics", {}
        )
        try:
            violation = float(penalty_diagnostics["violation"])
            penalty = float(penalty_diagnostics["penalty"])
            lower_value = float(total_diagnostics["lower_value"])
            total_value = float(total_diagnostics["total"])
        except (KeyError, TypeError, ValueError):
            errors.append(f"{label}: active scalar composition receipt is incomplete")
        else:
            expected_penalty_value = float(
                source_config["coverage_penalty_lambda"] * max(violation, 0.0) ** 2
            )
            if not math.isclose(penalty, expected_penalty_value, rel_tol=1e-15, abs_tol=1e-15):
                errors.append(f"{label}: active penalty scalar composition mismatch")
            if not math.isclose(total_value, lower_value + penalty, rel_tol=1e-15, abs_tol=1e-15):
                errors.append(f"{label}: total scalar composition mismatch")
            if total_diagnostics.get("penalty") != penalty:
                errors.append(f"{label}: total receipt penalty differs from penalty component")
            if violation_gradient is not None and penalty_gradient is not None:
                expected_penalty_gradient = np.tril(
                    2.0
                    * source_config["coverage_penalty_lambda"]
                    * max(violation, 0.0)
                    * violation_gradient
                )
                if not np.allclose(
                    penalty_gradient,
                    expected_penalty_gradient,
                    rtol=0.0,
                    atol=1e-15,
                ):
                    errors.append(f"{label}: active penalty gradient chain rule mismatch")
            if (
                active_lower_gradient is not None
                and penalty_gradient is not None
                and total_gradient is not None
                and not np.allclose(
                    total_gradient,
                    active_lower_gradient + penalty_gradient,
                    rtol=0.0,
                    atol=1e-15,
                )
            ):
                errors.append(f"{label}: total hypergradient composition mismatch")
        consistency = route_result.get("post_solve_square_consistency")
        consistency_pass = True
        if route_config["route_family"] == "squared_regression":
            if not isinstance(consistency, dict):
                errors.append(f"{label}: post-solve square consistency receipt is missing")
                consistency_pass = False
            else:
                root = float(consistency.get("root_objective", math.nan))
                reported = float(consistency.get("reported_objective", math.nan))
                expected_squared = root**2
                objective_error = abs(reported - expected_squared)
                root_gradient = _finite_matrix(
                    components["root_lower_value"].get("analytic"), samples.shape[1]
                )
                squared_gradient = _finite_matrix(
                    components["lower_value"].get("analytic"), samples.shape[1]
                )
                gradient_error = (
                    math.inf
                    if root_gradient is None or squared_gradient is None
                    else float(
                        np.max(
                            np.abs(squared_gradient - 2.0 * root * root_gradient),
                            initial=0.0,
                        )
                    )
                )
                consistency_pass = bool(
                    consistency.get("post_solve_square") is True
                    and math.isfinite(root)
                    and math.isfinite(reported)
                    and objective_error <= 1e-12
                    and gradient_error <= 1e-12
                )
                if not math.isclose(
                    float(consistency.get("expected_squared_objective", math.inf)),
                    expected_squared,
                    rel_tol=0.0,
                    abs_tol=0.0,
                ):
                    errors.append(f"{label}: squared objective identity is not reproducible")
                if not math.isclose(
                    float(consistency.get("objective_absolute_error", math.inf)),
                    objective_error,
                    rel_tol=0.0,
                    abs_tol=0.0,
                ):
                    errors.append(f"{label}: squared objective error is not reproducible")
                if not math.isclose(
                    float(consistency.get("gradient_max_absolute_error", math.inf)),
                    gradient_error,
                    rel_tol=0.0,
                    abs_tol=0.0,
                ):
                    errors.append(f"{label}: post-solve chain-rule error is not reproducible")
                if consistency.get("passed") is not consistency_pass:
                    errors.append(f"{label}: post-solve square pass flag mismatch")
        elif consistency is not None:
            errors.append(f"{label}: unexpected post-solve square receipt")
            consistency_pass = False
        expected_route_pass = bool(
            all(components[name].get("passed") is True for name in required)
            and consistency_pass
        )
        if route_result.get("passed") is not expected_route_pass:
            errors.append(f"{label}: route pass flag mismatch")
    if bindings.get("datasets") != expected_datasets:
        errors.append("dataset bindings mismatch")
    expected_families_present = {
        row.get("route_family") for row in routes
    } == set(config["required_route_families"])
    if evidence.get("all_required_route_families_present") is not expected_families_present:
        errors.append("required-route-family coverage flag mismatch")
    failures = evidence.get("failures")
    if not isinstance(failures, list):
        errors.append("failure ledger is missing")
        failures = []
    expected_all_pass = bool(
        not failures
        and routes
        and all(row.get("passed") is True for row in routes)
        and expected_families_present
    )
    if evidence.get("all_pass") is not expected_all_pass:
        errors.append("global pass flag mismatch")
    if not expected_all_pass:
        errors.append("hypergradient evidence contains a failed or missing required check")

    method = artifact.get("method", {})
    expected_method = {
        "finite_difference": "lower_triangular_central_difference",
        "steps": [float(value) for value in config["central_difference_steps"]],
        "primary_step": float(config["primary_step"]),
        "relative_error_threshold": config["relative_error_threshold"],
        "absolute_error_threshold": config["absolute_error_threshold"],
        "step_pass_policy": config["step_pass_policy"],
        "solver_residual_max": config["solver_residual_max"],
        "transport_residual_max": config["transport_residual_max"],
    }
    if method != expected_method:
        errors.append("finite-difference method/threshold contract mismatch")
    if recompute and not errors:
        try:
            observed_environment = locked_regeneration_environment()
        except Exception as exc:
            errors.append(f"locked regeneration environment is unavailable: {exc}")
        else:
            if observed_environment["packages"] != artifact_environment.get("packages"):
                errors.append("regeneration package versions differ from the sealed artifact")
            if observed_environment["threads"] != artifact_environment.get("threads"):
                errors.append("regeneration thread environment differs from the sealed artifact")
        if not errors:
            regenerated = build_hypergradient_validation(config_path, require_clean=False)
            if regenerated["evidence_payload_hash"] != artifact.get("evidence_payload_hash"):
                errors.append("independent regeneration evidence hash mismatch")
            if regenerated["evidence_payload"] != evidence:
                errors.append("independent regeneration evidence payload differs")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Validate and independently regenerate hypergradient evidence."
    )
    parser.add_argument(
        "artifact",
        type=Path,
        nargs="?",
        default=LANE_ROOT
        / ".openresearch"
        / "artifacts"
        / "validation"
        / "hypergradient.json",
    )
    parser.add_argument(
        "--config",
        type=Path,
        default=LANE_ROOT / "configs" / "hypergradient_validation_v1.json",
    )
    parser.add_argument(
        "--structural-only",
        action="store_true",
        help="Skip scientific regeneration. This mode is for adversarial unit tests only.",
    )
    args = parser.parse_args()
    errors = validate(
        args.artifact,
        args.config,
        recompute=not args.structural_only,
    )
    if errors:
        print("HYPERGRADIENT VALIDATION INVALID")
        for error in errors:
            print(f"- {error}")
        return 2
    if args.structural_only:
        print(
            "HYPERGRADIENT STRUCTURAL VALIDATION PASS: artifact sidecar, Git-object "
            "lineage, retained scalar evaluations, step grid, and residual contracts agree; "
            "scientific regeneration was NOT run; C1-C3 remain HOLD"
        )
    else:
        print(
            "HYPERGRADIENT VALIDATION PASS: the documented locked environment, all route "
            "families, analytic components, step grid, residuals, Git-object lineage, "
            "adjacent sidecar, and independent regeneration agree; C1-C3 remain HOLD"
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())