shared_resources/benchmark/run_benchmark.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["torch==2.12.1", "numpy==2.5.0"]
# ///
"""One Layer Deeper — Agent Collab benchmark wrapper.

Runs a `submission.py` through the PINNED upstream evaluator on one of this
collab's manifests, then turns the evaluator's `RESULT_JSON=` line into the
`depth_score` / `max_t` / `ood_depth_score` triple the leaderboard wants, plus
a ready-to-paste result frontmatter block.

This wrapper owns nothing that affects a score. It clones upstream at a fixed
commit, hands it an evaluator-owned manifest and your file, and does arithmetic
on the numbers that come back. See PINNED_UPSTREAM.md and README.md.

    python run_benchmark.py --config smoke  --submission submission.py
    python run_benchmark.py --config medium --submission submission.py \
        --submission-uri hf://buckets/agent-collaborations/deeper-me/subs/v3/submission.py

`--config medium` is the ranked configuration: 600 training seconds, seed 74,
24GB-class GPU. Everything else is practice.
"""

from __future__ import annotations

import argparse
from dataclasses import dataclass
import hashlib
import importlib.metadata
import importlib.util
import json
import os
from pathlib import Path
import platform
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile
import time
import urllib.request


# ── The pin. Do not bump casually; see PINNED_UPSTREAM.md. ──────────────────
UPSTREAM_REPO = "https://github.com/tilde-research/one-layer-deeper.git"
UPSTREAM_COMMIT = "e32c2f985f8ed4107c96d00271448777954ecc0c"
UPSTREAM_TARBALL = (
    "https://codeload.github.com/tilde-research/one-layer-deeper/tar.gz/"
    + UPSTREAM_COMMIT
)
PINNED_TORCH = "2.12.1"

CONFIGS = {
    "smoke": "manifests/deeper_smoke_cpu.json",
    "easy": "manifests/deeper_easy.json",
    "medium": "manifests/deeper_medium.json",
}
RANKED_CONFIG = "medium"
RANKED_HARDWARE = ("a10g-small", "l4x1")

# A score of exactly 0 is unpublishable (the backend requires a positive
# score), so a total failure lands on epsilon instead. The fractional part of a
# depth score is always < 1, so epsilon can never reorder anything.
SCORE_EPSILON = 1e-6
# Six decimals: the finest rung granularity in any of our configs is 1/768,
# so this is exact for every value the evaluator can produce, and it keeps the
# emitted scalar in plain decimal form (PyYAML does not read `1e-06` as a
# float).
SCORE_DECIMALS = 6


class BenchmarkError(RuntimeError):
    """Anything that should stop the run with a readable message."""


# ── upstream checkout ───────────────────────────────────────────────────────


def _git(*args: str, cwd: Path | None = None) -> str:
    result = subprocess.run(
        ["git", *args],
        cwd=None if cwd is None else str(cwd),
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise BenchmarkError(
            f"git {' '.join(args)} failed:\n{result.stderr.strip()}"
        )
    return result.stdout.strip()


def _clone_pinned(target: Path) -> None:
    print(f"[upstream] cloning {UPSTREAM_REPO} @ {UPSTREAM_COMMIT[:12]}", flush=True)
    target.parent.mkdir(parents=True, exist_ok=True)
    _git("clone", "--quiet", UPSTREAM_REPO, str(target))
    _git("checkout", "--quiet", "--detach", UPSTREAM_COMMIT, cwd=target)


def _download_pinned(target: Path) -> None:
    """git-less fallback: GitHub resolves the commit SHA, so this is pinned too."""

    print(f"[upstream] downloading tarball @ {UPSTREAM_COMMIT[:12]}", flush=True)
    target.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="deeper-upstream-") as scratch:
        archive = Path(scratch) / "upstream.tar.gz"
        with urllib.request.urlopen(UPSTREAM_TARBALL, timeout=120) as response:
            archive.write_bytes(response.read())
        with tarfile.open(archive) as handle:
            roots = {name.split("/", 1)[0] for name in handle.getnames()}
            if len(roots) != 1:
                raise BenchmarkError("unexpected upstream tarball layout")
            root = roots.pop()
            if UPSTREAM_COMMIT[:7] not in root:
                raise BenchmarkError(
                    f"tarball root {root!r} does not carry the pinned commit"
                )
            handle.extractall(scratch, filter="data")
        shutil.move(str(Path(scratch) / root), str(target))
    (target / ".pinned_commit").write_text(UPSTREAM_COMMIT + "\n", encoding="utf-8")


def _checkout_commit(repo: Path) -> str:
    """Return the commit a checkout is actually sitting on."""

    marker = repo / ".pinned_commit"
    if not (repo / ".git").exists() and marker.exists():
        return marker.read_text(encoding="utf-8").strip()
    return _git("rev-parse", "HEAD", cwd=repo)


def prepare_upstream(work_dir: Path, upstream_dir: Path | None) -> Path:
    """Return a checkout of the pinned upstream commit, or fail loudly."""

    if upstream_dir is not None:
        repo = upstream_dir.resolve()
        if not (repo / "benchmark" / "runner.py").is_file():
            raise BenchmarkError(f"{repo} is not a one-layer-deeper checkout")
        commit = _checkout_commit(repo)
        if commit != UPSTREAM_COMMIT:
            raise BenchmarkError(
                f"--upstream-dir sits on {commit[:12]}, not the pinned "
                f"{UPSTREAM_COMMIT[:12]}. Run:\n"
                f"  git -C {repo} fetch origin {UPSTREAM_COMMIT}\n"
                f"  git -C {repo} checkout --detach {UPSTREAM_COMMIT}"
            )
        print(f"[upstream] reusing {repo} @ {commit[:12]}", flush=True)
        return repo

    repo = (work_dir / "one-layer-deeper").resolve()
    if repo.exists() and _checkout_commit(repo) == UPSTREAM_COMMIT:
        print(f"[upstream] reusing {repo} @ {UPSTREAM_COMMIT[:12]}", flush=True)
        return repo
    if repo.exists():
        shutil.rmtree(repo)
    if shutil.which("git"):
        _clone_pinned(repo)
    else:
        _download_pinned(repo)
    commit = _checkout_commit(repo)
    if commit != UPSTREAM_COMMIT:
        raise BenchmarkError(
            f"upstream checkout landed on {commit[:12]}, expected "
            f"{UPSTREAM_COMMIT[:12]}"
        )
    return repo


# ── environment and data integrity ──────────────────────────────────────────


def check_runtime_dependencies() -> None:
    missing = [
        name for name in ("torch", "numpy") if importlib.util.find_spec(name) is None
    ]
    if missing:
        raise BenchmarkError(
            f"missing evaluator dependencies: {', '.join(missing)}.\n"
            "This script carries them as PEP 723 metadata, so the simplest fix "
            "is to let uv install them:\n"
            f"  uv run {Path(__file__).name} --help\n"
            "or install them yourself:\n"
            f"  {sys.executable} -m pip install torch=={PINNED_TORCH} numpy"
        )
    try:
        version = importlib.metadata.version("torch")
    except importlib.metadata.PackageNotFoundError:
        return
    if version != PINNED_TORCH:
        print(
            f"[warn] torch {version} != upstream's pinned {PINNED_TORCH}; "
            "results stay valid but may not be bit-comparable",
            flush=True,
        )


def verify_dataset_checksums(kit_root: Path, data_root: str) -> int:
    """Check the shipped dataset bytes for one config. Bit-identical or bust."""

    manifest = kit_root / "datasets" / "CHECKSUMS.sha256"
    if not manifest.is_file():
        raise BenchmarkError(f"missing dataset checksum manifest: {manifest}")
    prefix = data_root.rstrip("/") + "/"
    checked = 0
    for line in manifest.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        expected, _, relative = line.partition("  ")
        relative = relative.strip()
        if not relative.startswith(prefix):
            continue
        path = kit_root / relative
        if not path.is_file():
            raise BenchmarkError(f"dataset file missing: {path}")
        digest = hashlib.sha256()
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1 << 20), b""):
                digest.update(chunk)
        if digest.hexdigest() != expected:
            raise BenchmarkError(
                f"dataset file {relative} does not match CHECKSUMS.sha256 — "
                "re-sync shared_resources/benchmark/datasets/ before scoring"
            )
        checked += 1
    if checked == 0:
        raise BenchmarkError(f"no checksums recorded for {data_root}")
    return checked


def validate_submission_source(upstream: Path, submission: Path) -> None:
    """Run upstream's own AST-level source policy over the file."""

    if submission.name != "submission.py":
        print(
            f"[warn] upstream requires the file to be named submission.py, "
            f"not {submission.name} — rename it before forwarding a result",
            flush=True,
        )
    program = (
        "import os, sys\n"
        "sys.path.insert(0, sys.argv[1])\n"
        "from submission_validation import validate_submission_source\n"
        "path = sys.argv[2]\n"
        "source = open(path, encoding='utf-8').read()\n"
        "validate_submission_source(\n"
        "    os.path.basename(path), source, 256 * 1024, required_filename=None\n"
        ")\n"
        "print('SUBMISSION_SOURCE_OK')\n"
    )
    result = subprocess.run(
        [sys.executable, "-c", program, str(upstream), str(submission)],
        capture_output=True,
        text=True,
    )
    if "SUBMISSION_SOURCE_OK" not in result.stdout:
        detail = (result.stderr.strip().splitlines() or ["unknown error"])[-1]
        raise BenchmarkError(f"submission rejected by upstream validation: {detail}")
    size = submission.stat().st_size
    print(
        f"[validate] {submission.name} passes upstream source policy "
        f"({size:,} bytes / 262,144 limit)",
        flush=True,
    )


# ── manifest ────────────────────────────────────────────────────────────────


def build_effective_manifest(
    kit_root: Path,
    config: str,
    scratch: Path,
    *,
    train_seconds: float | None,
    device: str | None,
    manifest_path: Path | None = None,
) -> tuple[Path, dict, list[str]]:
    """Resolve data_root to an absolute path and apply any debug overrides."""

    source = manifest_path if manifest_path is not None else kit_root / CONFIGS[config]
    if not source.is_file():
        raise BenchmarkError(f"missing manifest: {source}")
    manifest = json.loads(source.read_text(encoding="utf-8"))

    data_root = manifest["data"]["data_root"]
    if data_root is None:
        raise BenchmarkError("collab manifests must reference a shipped dataset")
    checked = verify_dataset_checksums(kit_root, data_root)
    print(f"[data] {checked} files verified against CHECKSUMS.sha256", flush=True)
    manifest["data"]["data_root"] = str((kit_root / data_root).resolve())

    overrides: list[str] = []
    if train_seconds is not None:
        overrides.append(
            f"train_seconds={train_seconds:g}"
            f" (config default {manifest['runtime']['total_training_time_seconds']:g})"
        )
        manifest["runtime"]["total_training_time_seconds"] = float(train_seconds)
    if device is not None:
        overrides.append(
            f"device={device} (config default {manifest['runtime']['device']})"
        )
        manifest["runtime"]["device"] = device
        if not device.startswith("cuda"):
            # bfloat16 autocast on CPU is slow enough to starve both budgets,
            # so a CPU debug run gets float32 and no autocast. One more reason
            # this path can never be ranked.
            manifest["data"]["pin_memory"] = False
            manifest["runtime"]["dtype"] = "float32"
            manifest["runtime"]["amp"] = False
            overrides.append("dtype=float32, amp=false (implied by a non-CUDA device)")

    path = scratch / f"effective_{config}.json"
    path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
    return path, manifest, overrides


# ── the evaluator ───────────────────────────────────────────────────────────


def run_evaluator(upstream: Path, manifest: Path, submission: Path) -> dict:
    """Run upstream's runner, echo its log, and return the parsed RESULT_JSON."""

    command = [
        sys.executable,
        "-m",
        "benchmark.runner",
        "--manifest",
        str(manifest),
        "--submission-file",
        str(submission),
    ]
    print(f"[run] {' '.join(command)}", flush=True)
    print(f"[run] cwd={upstream}", flush=True)
    print("-" * 72, flush=True)
    started = time.monotonic()
    environment = dict(os.environ)
    environment["PYTHONUNBUFFERED"] = "1"
    process = subprocess.Popen(
        command,
        cwd=str(upstream),
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
        env=environment,
    )
    result_line: str | None = None
    tail: list[str] = []
    assert process.stdout is not None
    for line in process.stdout:
        line = line.rstrip("\n")
        print(line, flush=True)
        tail.append(line)
        del tail[:-40]
        if line.startswith("RESULT_JSON="):
            result_line = line
    code = process.wait()
    print("-" * 72, flush=True)
    print(f"[run] evaluator exited {code} after {time.monotonic() - started:.1f}s", flush=True)
    if code != 0:
        message = "the evaluator failed. Last lines:\n  " + "\n  ".join(tail)
        if any("time budget" in line for line in tail):
            message += (
                "\n\nThe evaluation budget is always half the training budget, so "
                "shortening training with --train-seconds shortens evaluation "
                "too. Use --config smoke for fast iteration instead of a "
                "truncated GPU config."
            )
        raise BenchmarkError(message)
    if result_line is None:
        raise BenchmarkError("the evaluator printed no RESULT_JSON= line")
    return json.loads(result_line[len("RESULT_JSON=") :])


# ── scoring ─────────────────────────────────────────────────────────────────


@dataclass(frozen=True)
class ProfileScore:
    """One depth profile of one seed, collapsed to the leaderboard scalar."""

    certified_time_steps: int
    next_rung_time_steps: int | None
    next_rung_exact_accuracy: float
    complete: bool

    @property
    def score(self) -> float:
        return self.certified_time_steps + self.next_rung_exact_accuracy


def score_profile(rungs: list[dict]) -> ProfileScore:
    """certified Max T + exact accuracy at the first uncertified rung.

    A rung is certified only when it and every lower rung scored 100% exact
    accuracy, so the walk stops at the first rung that is not certified — that
    rung supplies the fractional term. A rung the evaluator could not finish
    inside its eval budget (`not_completed`) contributes 0.0.
    """

    certified = 0
    for rung in sorted(rungs, key=lambda rung: int(rung["time_steps"])):
        if rung.get("status") == "certified":
            certified = int(rung["time_steps"])
            continue
        accuracy = rung.get("exact_accuracy")
        return ProfileScore(
            certified_time_steps=certified,
            next_rung_time_steps=int(rung["time_steps"]),
            next_rung_exact_accuracy=0.0 if accuracy is None else float(accuracy),
            complete=False,
        )
    return ProfileScore(
        certified_time_steps=certified,
        next_rung_time_steps=None,
        next_rung_exact_accuracy=0.0,
        complete=bool(rungs),
    )


def score_result(result: dict) -> dict:
    """Collapse a RESULT_JSON payload to the collab's leaderboard fields.

    Multi-seed runs take the minimum across seeds. `certified + accuracy` with
    accuracy < 1 is a total order identical to lexicographic
    `(certified, accuracy)`, so the minimum composite is also the minimum
    certified rung — the same rule upstream applies to Max T.
    """

    seeds = result.get("seeds") or []
    if not seeds:
        raise BenchmarkError("RESULT_JSON carries no per-seed results")

    per_seed = []
    for seed_result in seeds:
        profile = seed_result.get("depth_profile") or {}
        per_seed.append(
            {
                "seed": seed_result.get("seed"),
                "seen_n": score_profile(profile.get("rungs") or []),
                "ood_n": score_profile(profile.get("ood_n_rungs") or []),
                "reported_certified": profile.get("max_certified_time_steps"),
                "reported_ood_certified": profile.get(
                    "ood_n_max_certified_time_steps"
                ),
            }
        )

    warnings: list[str] = []
    for entry in per_seed:
        for key, reported_key, label in (
            ("seen_n", "reported_certified", "Max T"),
            ("ood_n", "reported_ood_certified", "OOD N Max T"),
        ):
            reported = entry[reported_key] or 0
            if entry[key].certified_time_steps != reported:
                warnings.append(
                    f"seed {entry['seed']}: computed {label}="
                    f"{entry[key].certified_time_steps} but the evaluator "
                    f"reported {reported}"
                )

    worst_seen = min(per_seed, key=lambda entry: entry["seen_n"].score)
    worst_ood = min(per_seed, key=lambda entry: entry["ood_n"].score)
    if not worst_seen["seen_n"].complete and worst_seen["seen_n"].next_rung_time_steps is None:
        warnings.append(
            "this manifest exposed no depth ladder — depth_score is meaningless"
        )

    return {
        "depth_score": max(worst_seen["seen_n"].score, SCORE_EPSILON),
        "max_t": worst_seen["seen_n"].certified_time_steps,
        "ood_depth_score": max(worst_ood["ood_n"].score, SCORE_EPSILON),
        "ood_max_t": worst_ood["ood_n"].certified_time_steps,
        "seen_n": worst_seen["seen_n"],
        "ood_n": worst_ood["ood_n"],
        "per_seed": per_seed,
        "warnings": warnings,
    }


# ── reporting ───────────────────────────────────────────────────────────────


def format_score(value: float) -> str:
    return f"{value:.{SCORE_DECIMALS}f}"


# YAML indicator characters, and the words a plain scalar would be coerced from.
_UNSAFE_FIRST = set("-?:,[]{}#&*!|>'\"%@`")
_COERCED = {"true", "false", "yes", "no", "on", "off", "null", "~", ""}
_NUMERIC = re.compile(r"^[-+.\d]")


def yaml_scalar(value: str) -> str:
    """Emit a string as a plain scalar when YAML would read it back verbatim."""

    plain = (
        bool(value)
        and value == value.strip()
        and "\n" not in value
        and value[0] not in _UNSAFE_FIRST
        and ": " not in value
        and " #" not in value
        and not value.endswith(":")
        and value.lower() not in _COERCED
        and not _NUMERIC.match(value)
    )
    if plain:
        return value
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def detect_hardware() -> str:
    accelerator = os.environ.get("ACCELERATOR", "").strip()
    if accelerator:
        return accelerator
    try:
        import torch

        if torch.cuda.is_available():
            return torch.cuda.get_device_name(0)
    except Exception:  # noqa: BLE001 - reporting only, never fatal
        pass
    return f"cpu-{platform.machine()}"


def describe_profile(label: str, profile: ProfileScore) -> str:
    if profile.next_rung_time_steps is None:
        if not profile.complete:
            return f"{label}: no depth ladder in this manifest"
        return f"{label}: certified the whole ladder (T={profile.certified_time_steps})"
    return (
        f"{label}: certified T={profile.certified_time_steps}, "
        f"exact accuracy {profile.next_rung_exact_accuracy:.4f} at the first "
        f"uncertified rung T={profile.next_rung_time_steps}"
    )


def print_summary(
    *,
    scores: dict,
    result: dict,
    config: str,
    manifest: dict,
    overrides: list[str],
    hardware: str,
    elapsed: float,
) -> None:
    print()
    print("=" * 72)
    print("One Layer Deeper — Agent Collab benchmark summary")
    print("=" * 72)
    print(f"  config              {config}"
          f"{'  (RANKED)' if config == RANKED_CONFIG and not overrides else ''}")
    print(f"  manifest            {manifest['name']}")
    print(f"  upstream commit     {UPSTREAM_COMMIT}")
    print(f"  training seconds    {manifest['runtime']['total_training_time_seconds']:g}")
    print(f"  seeds               {manifest['runtime']['seeds']}")
    print(f"  device / dtype      {manifest['runtime']['device']} / "
          f"{manifest['runtime']['dtype']} (amp={manifest['runtime']['amp']})")
    print(f"  hardware            {hardware}")
    print(f"  wall clock          {elapsed:.1f}s")
    print()
    print(f"  {describe_profile('in-distribution', scores['seen_n'])}")
    print(f"  {describe_profile('out-of-dist. N ', scores['ood_n'])}")
    print()
    print(f"  depth_score         {format_score(scores['depth_score'])}")
    print(f"  max_t               {scores['max_t']}")
    print(f"  ood_depth_score     {format_score(scores['ood_depth_score'])}")
    mean_accuracy = (result.get("score") or {}).get("mean_exact_accuracy")
    if mean_accuracy is not None:
        print(f"  (upstream mean_exact_accuracy on scored splits: {mean_accuracy:.6f})")
    for seed_result in result.get("seeds", []):
        print(
            f"  seed {seed_result.get('seed')}: "
            f"{seed_result.get('completed_training_steps')} steps, "
            f"final train loss {seed_result.get('final_train_loss')}, "
            f"{seed_result.get('model_state_elements'):,} model-state elements"
        )
    for warning in scores["warnings"]:
        print(f"  [warn] {warning}")
    if overrides:
        print()
        print("  NOT RANKED — manifest overrides applied:")
        for override in overrides:
            print(f"    - {override}")
    elif config != RANKED_CONFIG:
        print()
        print(f"  NOT RANKED — practice config; ranked runs use --config {RANKED_CONFIG}.")
    elif hardware not in RANKED_HARDWARE:
        print()
        print(
            f"  [warn] hardware {hardware!r} is not one of the ranked "
            f"24GB-class flavors {list(RANKED_HARDWARE)} — report what you "
            "actually used and expect organizers to check."
        )
    print("=" * 72)


def frontmatter_block(
    *,
    scores: dict,
    manifest: dict,
    config: str,
    overrides: list[str],
    hardware: str,
    method: str,
    submission_uri: str,
    description: str | None,
) -> str:
    seeds = manifest["runtime"]["seeds"]
    seed_field = seeds[0] if len(seeds) == 1 else json.dumps(seeds)
    if description is None:
        description = (
            f"{method} on {manifest['name']} "
            f"({manifest['runtime']['total_training_time_seconds']:g} training "
            f"seconds, seed {seed_field}, {hardware}): "
            f"certified T={scores['max_t']}"
        )
        if scores["seen_n"].next_rung_time_steps is not None:
            description += (
                f", exact accuracy "
                f"{scores['seen_n'].next_rung_exact_accuracy:.4f} at "
                f"T={scores['seen_n'].next_rung_time_steps}"
            )
        description += f"; OOD-N certified T={scores['ood_max_t']}."
    if overrides or config != RANKED_CONFIG:
        description = f"[not ranked: config={config}" + (
            f"; overrides {', '.join(overrides)}] " if overrides else "] "
        ) + description

    lines = [
        "---",
        f"depth_score: {format_score(scores['depth_score'])}",
        f"method: {yaml_scalar(method)}",
        "status: agent-run",
        f"description: {yaml_scalar(description)}",
        f"max_t: {scores['max_t']}",
        f"ood_depth_score: {format_score(scores['ood_depth_score'])}",
        f"hardware: {yaml_scalar(hardware)}",
        f"seed: {seed_field}",
        f"submission: {yaml_scalar(submission_uri)}",
        "---",
    ]
    return "\n".join(lines)


# ── self-test ───────────────────────────────────────────────────────────────


def _fixture(rungs, ood_rungs=None, seed=74):
    """Build the depth_profile shape the evaluator emits, for one seed."""

    def build(entries):
        built = []
        solved_prefix = True
        for time_steps, accuracy in entries:
            if accuracy is None:
                built.append(
                    {
                        "time_steps": time_steps,
                        "status": "not_completed",
                        "exact_accuracy": None,
                    }
                )
                break
            solved = accuracy == 1.0
            solved_prefix = solved_prefix and solved
            built.append(
                {
                    "time_steps": time_steps,
                    "status": (
                        "certified"
                        if solved_prefix
                        else "passed_uncertified"
                        if solved
                        else "failed"
                    ),
                    "exact_accuracy": accuracy,
                }
            )
        return built

    seen = build(rungs)
    ood = build(ood_rungs if ood_rungs is not None else rungs)
    certified = [r["time_steps"] for r in seen if r["status"] == "certified"]
    ood_certified = [r["time_steps"] for r in ood if r["status"] == "certified"]
    return {
        "seed": seed,
        "depth_profile": {
            "ladder": [r["time_steps"] for r in seen],
            "max_certified_time_steps": certified[-1] if certified else None,
            "rungs": seen,
            "ood_n_rungs": ood,
            "ood_n_max_certified_time_steps": (
                ood_certified[-1] if ood_certified else None
            ),
        },
    }


def selftest() -> int:
    """Hand-built RESULT_JSON fixtures for the scoring rules."""

    ladder = (1, 2, 4, 8, 16, 32, 64)
    failures: list[str] = []

    def check(name, actual, expected):
        ok = actual == expected
        print(f"  {'ok  ' if ok else 'FAIL'} {name}: {actual!r}")
        if not ok:
            failures.append(f"{name}: expected {expected!r}, got {actual!r}")

    # 1. certified prefix through T=2, 87% at the first uncertified rung.
    scores = score_result(
        {"seeds": [_fixture([(1, 1.0), (2, 1.0), (4, 0.87), (8, 0.1), (16, 0.0),
                             (32, 0.0), (64, 0.0)])]}
    )
    check("certified prefix -> 2 + 0.87", round(scores["depth_score"], 6), 2.87)
    check("certified prefix -> max_t", scores["max_t"], 2)

    # 2. a solved rung above a failed one never certifies.
    scores = score_result(
        {"seeds": [_fixture([(1, 1.0), (2, 0.5), (4, 1.0), (8, 1.0)])]}
    )
    check("gap in the prefix -> 1 + 0.5", round(scores["depth_score"], 6), 1.5)
    check("gap in the prefix -> max_t", scores["max_t"], 1)

    # 3. full ladder certified: no uncertified rung, so no fractional term.
    scores = score_result(
        {"seeds": [_fixture([(t, 1.0) for t in ladder])]}
    )
    check("whole ladder -> 64.0", scores["depth_score"], 64.0)
    check("whole ladder -> max_t", scores["max_t"], 64)

    # 4. rung the evaluator could not finish contributes 0.0 and truncates.
    scores = score_result(
        {"seeds": [_fixture([(1, 1.0), (2, 1.0), (4, 1.0), (8, None)])]}
    )
    check("not_completed rung -> 4 + 0.0", scores["depth_score"], 4.0)
    check("not_completed rung -> max_t", scores["max_t"], 4)

    # 5. first rung fails outright: epsilon, never exact zero.
    scores = score_result({"seeds": [_fixture([(1, 0.0), (2, 0.0)])]})
    check("total failure -> epsilon", scores["depth_score"], SCORE_EPSILON)
    check("total failure -> max_t", scores["max_t"], 0)
    check("epsilon is publishable", scores["depth_score"] > 0, True)
    check("epsilon renders as decimal", format_score(scores["depth_score"]), "0.000001")

    # 6. truncated ladder: the evaluator timed out before the first rung.
    scores = score_result({"seeds": [_fixture([(1, None)])]})
    check("immediate timeout -> epsilon", scores["depth_score"], SCORE_EPSILON)

    # 7. multi-seed takes the minimum, lexicographically on (max_t, accuracy).
    scores = score_result(
        {
            "seeds": [
                _fixture([(1, 1.0), (2, 1.0), (4, 1.0), (8, 0.0)], seed=1),
                _fixture([(1, 1.0), (2, 0.25), (4, 1.0)], seed=2),
            ]
        }
    )
    check("multi-seed min -> 1 + 0.25", round(scores["depth_score"], 6), 1.25)
    check("multi-seed min -> max_t", scores["max_t"], 1)
    scores = score_result(
        {
            "seeds": [
                _fixture([(1, 1.0), (2, 0.9)], seed=1),
                _fixture([(1, 1.0), (2, 0.1)], seed=2),
            ]
        }
    )
    check("multi-seed same max_t -> worst accuracy", round(scores["depth_score"], 6), 1.1)

    # 8. the two profiles are scored independently.
    scores = score_result(
        {
            "seeds": [
                _fixture(
                    [(1, 1.0), (2, 1.0), (4, 0.5)],
                    ood_rungs=[(1, 1.0), (2, 0.42), (4, 0.0)],
                )
            ]
        }
    )
    check("in-distribution profile", round(scores["depth_score"], 6), 2.5)
    check("ood profile", round(scores["ood_depth_score"], 6), 1.42)
    check("ood max_t", scores["ood_max_t"], 1)

    # 9. no depth ladder at all (e.g. upstream's built-in smoke dataset).
    scores = score_result({"seeds": [_fixture([])]})
    check("no ladder -> epsilon", scores["depth_score"], SCORE_EPSILON)
    check("no ladder -> warning", bool(scores["warnings"]), True)

    # 10. an evaluator/computed disagreement surfaces instead of hiding.
    broken = _fixture([(1, 1.0), (2, 0.5)])
    broken["depth_profile"]["max_certified_time_steps"] = 32
    check(
        "disagreement warns",
        any("reported" in w for w in score_result({"seeds": [broken]})["warnings"]),
        True,
    )

    # 11. the emitted frontmatter parses as YAML with a positive score.
    block = frontmatter_block(
        scores=score_result(
            {"seeds": [_fixture([(1, 1.0), (2, 1.0), (4, 0.87)])]}
        ),
        manifest={
            "name": "deeper-medium-mirror",
            "runtime": {"total_training_time_seconds": 600, "seeds": [74]},
        },
        config="medium",
        overrides=[],
        hardware="a10g-small",
        method="looped-transformer: v3",
        submission_uri="hf://buckets/agent-collaborations/deeper-me/v3/submission.py",
        description=None,
    )
    body = "\n".join(block.splitlines()[1:-1])
    try:
        import yaml  # type: ignore

        parsed = yaml.safe_load(body)
    except ImportError:
        parsed = None
        print("  skip pyyaml not installed; frontmatter parsed structurally only")
    if parsed is not None:
        check("frontmatter score is a float", isinstance(parsed["depth_score"], float), True)
        check("frontmatter score is positive", parsed["depth_score"] > 0, True)
        check("frontmatter max_t is an int", isinstance(parsed["max_t"], int), True)
        check("frontmatter seed is an int", isinstance(parsed["seed"], int), True)
        check("frontmatter keeps the colon in method", parsed["method"], "looped-transformer: v3")
        check(
            "frontmatter bucket uri round-trips",
            parsed["submission"],
            "hf://buckets/agent-collaborations/deeper-me/v3/submission.py",
        )
        check(
            "bucket uri needs no quotes",
            "submission: hf://buckets/agent-collaborations/deeper-me/v3/submission.py" in block,
            True,
        )
        for risky in ("123", "1e5", "yes", "null", "- dash", "trailing ", "a: b", "#hash"):
            round_tripped = yaml.safe_load(f"value: {yaml_scalar(risky)}")["value"]
            check(f"scalar {risky!r} round-trips as a string", round_tripped, risky)
        check(
            "frontmatter fields",
            sorted(parsed),
            sorted(
                [
                    "depth_score",
                    "method",
                    "status",
                    "description",
                    "max_t",
                    "ood_depth_score",
                    "hardware",
                    "seed",
                    "submission",
                ]
            ),
        )
    else:
        check("frontmatter line count", len(block.splitlines()), 11)

    print()
    if failures:
        print(f"selftest FAILED ({len(failures)}):")
        for failure in failures:
            print(f"  - {failure}")
        return 1
    print("selftest passed")
    return 0


# ── cli ─────────────────────────────────────────────────────────────────────


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--submission",
        help="path to your submission.py",
    )
    parser.add_argument(
        "--config",
        choices=sorted(CONFIGS),
        default=RANKED_CONFIG,
        help=f"benchmark configuration (default: {RANKED_CONFIG}, the ranked one)",
    )
    parser.add_argument(
        "--kit-root",
        type=Path,
        default=Path(__file__).resolve().parent,
        help="directory holding manifests/ and datasets/ (default: next to this script)",
    )
    parser.add_argument(
        "--manifest",
        type=Path,
        help="ORGANIZERS: run an out-of-tree manifest (e.g. the private eval "
        "set) instead of --config; data_root and checksums resolve against "
        "--kit-root. Never ranked.",
    )
    parser.add_argument(
        "--upstream-dir",
        type=Path,
        help="reuse an existing one-layer-deeper checkout instead of cloning",
    )
    parser.add_argument(
        "--work-dir",
        type=Path,
        default=Path(
            os.environ.get(
                "DEEPER_WORK_DIR", str(Path.home() / ".cache" / "one-layer-deeper-collab")
            )
        ),
        help="where to keep the pinned upstream clone",
    )
    parser.add_argument(
        "--method",
        help="method name for the result frontmatter (default: the file stem)",
    )
    parser.add_argument(
        "--submission-uri",
        default="hf://buckets/REPLACE_ME/submission.py",
        help="scratch-bucket path of the exact file you ran, for the frontmatter",
    )
    parser.add_argument(
        "--description",
        help="override the auto-generated frontmatter description",
    )
    parser.add_argument(
        "--hardware",
        help="override the detected hardware label (e.g. a10g-small, l4x1)",
    )
    parser.add_argument(
        "--train-seconds",
        type=float,
        help="DEBUG ONLY: shorten the training budget; marks the run not ranked",
    )
    parser.add_argument(
        "--device",
        help="DEBUG ONLY: override the manifest device; marks the run not ranked",
    )
    parser.add_argument(
        "--save-result",
        type=Path,
        help="write the raw RESULT_JSON plus computed scores to this path",
    )
    parser.add_argument(
        "--validate-only",
        action="store_true",
        help="check upstream pin, datasets, and submission source, then stop",
    )
    parser.add_argument(
        "--selftest",
        action="store_true",
        help="run the scoring unit checks and exit (no GPU, no submission needed)",
    )
    args = parser.parse_args(argv)
    if not args.selftest and not args.submission:
        parser.error("--submission is required (or use --selftest)")
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    if args.selftest:
        return selftest()

    kit_root = args.kit_root.resolve()
    submission = Path(args.submission).resolve()
    if not submission.is_file():
        raise BenchmarkError(f"no such submission file: {submission}")

    started = time.monotonic()
    check_runtime_dependencies()
    upstream = prepare_upstream(args.work_dir.expanduser(), args.upstream_dir)
    validate_submission_source(upstream, submission)

    config = args.config
    if args.manifest is not None:
        config = args.manifest.stem

    with tempfile.TemporaryDirectory(prefix="deeper-manifest-") as scratch:
        manifest_path, manifest, overrides = build_effective_manifest(
            kit_root,
            config,
            Path(scratch),
            train_seconds=args.train_seconds,
            device=args.device,
            manifest_path=None if args.manifest is None else args.manifest.resolve(),
        )
        if args.validate_only:
            print("[validate] pin, datasets, and submission source all OK")
            return 0
        result = run_evaluator(upstream, manifest_path, submission)

    scores = score_result(result)
    hardware = args.hardware or detect_hardware()
    method = args.method or submission.stem
    elapsed = time.monotonic() - started

    print_summary(
        scores=scores,
        result=result,
        config=config,
        manifest=manifest,
        overrides=overrides,
        hardware=hardware,
        elapsed=elapsed,
    )
    block = frontmatter_block(
        scores=scores,
        manifest=manifest,
        config=config,
        overrides=overrides,
        hardware=hardware,
        method=method,
        submission_uri=args.submission_uri,
        description=args.description,
    )
    print()
    print("Paste this into your result file (fill in `submission:` with the")
    print("scratch-bucket path of the exact file you just ran):")
    print()
    print(block)
    print()

    if args.save_result:
        args.save_result.parent.mkdir(parents=True, exist_ok=True)
        args.save_result.write_text(
            json.dumps(
                {
                    "upstream_commit": UPSTREAM_COMMIT,
                    "config": config,
                    "manifest": manifest,
                    "overrides": overrides,
                    "hardware": hardware,
                    "depth_score": scores["depth_score"],
                    "max_t": scores["max_t"],
                    "ood_depth_score": scores["ood_depth_score"],
                    "ood_max_t": scores["ood_max_t"],
                    "frontmatter": block,
                    "result_json": result,
                },
                indent=2,
                sort_keys=True,
            )
            + "\n",
            encoding="utf-8",
        )
        print(f"[save] wrote {args.save_result}")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except BenchmarkError as error:
        print(f"\nERROR: {error}", file=sys.stderr)
        sys.exit(2)
    except KeyboardInterrupt:
        print("\ninterrupted", file=sys.stderr)
        sys.exit(130)