overlay/scripts/benchmark_contract.py

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

import json
from pathlib import Path
from typing import Any


def _require_path(payload: dict[str, Any], path: str) -> None:
    current: Any = payload
    for part in path.split('.'):
        if not isinstance(current, dict) or part not in current:
            raise ValueError(f"missing required field: {path}")
        current = current[part]


def validate_benchmark_contract(payload: dict[str, Any]) -> None:
    for field in [
        "cycle_id",
        "hardware_class",
        "seeds",
        "budget_modes",
        "coding_benchmarks.fast_iteration",
        "coding_benchmarks.milestone",
        "reasoning_benchmarks.fast_iteration",
        "reasoning_benchmarks.milestone",
        "variants.hydra_full",
        "variants.baseline_mamba_matched",
    ]:
        _require_path(payload, field)

    for section in [
        payload["coding_benchmarks"]["fast_iteration"],
        payload["coding_benchmarks"]["milestone"],
        payload["reasoning_benchmarks"]["fast_iteration"],
        payload["reasoning_benchmarks"]["milestone"],
    ]:
        if "name" not in section or "primary_metric" not in section or "decode" not in section:
            raise ValueError("benchmark sections require name, primary_metric, and decode")

    if not isinstance(payload["seeds"], list) or len(payload["seeds"]) < 3:
        raise ValueError("seeds must contain at least three values")

    if payload["variants"]["hydra_full"].get("status") != "runnable_now":
        raise ValueError("hydra_full must be runnable_now")

    if payload["variants"]["baseline_mamba_matched"].get("status") != "runnable_now":
        raise ValueError("baseline_mamba_matched must be runnable_now")


def load_benchmark_contract(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("benchmark contract must be a JSON object")
    validate_benchmark_contract(payload)
    return payload


def main() -> int:
    path = Path("artifacts/cycle_1_execution_freeze.json")
    payload = load_benchmark_contract(path)
    print(json.dumps(payload, indent=2, sort_keys=True))
    return 0


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