bundle/v0/scripts/run_hypergradient_validation.py

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

import argparse
import hashlib
import os
import sys
from pathlib import Path


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

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

from loss_aware_dro_repro.hypergradient_validation import (  # noqa: E402
    write_hypergradient_validation,
)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Run bounded, component-level hypergradient validation."
    )
    parser.add_argument(
        "--config",
        type=Path,
        default=LANE_ROOT / "configs" / "hypergradient_validation_v1.json",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=LANE_ROOT
        / ".openresearch"
        / "artifacts"
        / "validation"
        / "hypergradient.json",
    )
    args = parser.parse_args()
    artifact = write_hypergradient_validation(args.config, args.output)
    digest = hashlib.sha256(args.output.resolve().read_bytes()).hexdigest()
    sidecar = args.output.resolve().with_suffix(".sha256")
    print(f"artifact={args.output.resolve()}")
    print(f"sidecar={sidecar}")
    print(f"sha256={digest}")
    print(f"evidence_payload_hash={artifact['evidence_payload_hash']}")
    print(f"routes={len(artifact['evidence_payload']['routes'])}")
    print(f"failures={len(artifact['evidence_payload']['failures'])}")
    print(f"all_pass={artifact['evidence_payload']['all_pass']}")
    return 0 if artifact["evidence_payload"]["all_pass"] else 2


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