arborist/bench/batteries/runner.py

"""CLI entry: ``python -m bench.batteries.runner``.

Runs one or more battery sub-batteries against their fixture files
and emits a JSON result on stdout (or to ``--out`` if provided).

Usage examples
--------------

::

    # Run 5S Syntax + Semantics with the seed fixtures.
    python -m bench.batteries.runner \\
        --battery 5s --sub syntax --fixtures bench/fixtures/5s/syntax-v1.jsonl

    python -m bench.batteries.runner \\
        --battery 5s --sub semantics \\
        --fixtures bench/fixtures/5s/semantics-v1.jsonl

    # Run 5T Transfer.
    python -m bench.batteries.runner \\
        --battery 5t --sub transfer \\
        --fixtures bench/fixtures/5t/transfer-v1.jsonl

    # All Phase-1a fixtures, written to a results file.
    python -m bench.batteries.runner --all --out bench/results/run.json
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import asdict
from pathlib import Path
from typing import Optional

from bench.batteries import b_5f, b_5r, b_5s, b_5t
from bench.batteries.base import BatteryResult


_BATTERIES = {
    "5s": b_5s.SUB_BATTERIES,
    "5t": b_5t.SUB_BATTERIES,
    "5f": b_5f.SUB_BATTERIES,
    "5r": b_5r.SUB_BATTERIES,
}

_DEFAULT_FIXTURES = {
    # 5S — Phase 1a
    ("5s", "syntax"): "bench/fixtures/5s/syntax-v1.jsonl",
    ("5s", "semantics"): "bench/fixtures/5s/semantics-v1.jsonl",
    # 5S — Phase 1b (ticket #000023)
    ("5s", "syllogism"): "bench/fixtures/5s/syllogism-v1.jsonl",
    ("5s", "synthesis"): "bench/fixtures/5s/synthesis-v1.jsonl",
    ("5s", "semiotics"): "bench/fixtures/5s/semiotics-v1.jsonl",
    # 5T — Phase 1a (legacy SQD vocabulary)
    ("5t", "transfer"): "bench/fixtures/5t/transfer-v1.jsonl",
    # 5T — Phase 1b (Dav1DPrometheus vocabulary, ticket #000024)
    ("5t", "transfer-learning"): "bench/fixtures/5t/transfer-learning-v2.jsonl",
    ("5t", "triangulation"): "bench/fixtures/5t/triangulation-v1.jsonl",
    ("5t", "truthtables"): "bench/fixtures/5t/truthtables-v1.jsonl",
    ("5t", "transitivity"): "bench/fixtures/5t/transitivity-v1.jsonl",
    ("5t", "time"): "bench/fixtures/5t/time-v1.jsonl",
    # 5F — Phase 1a (ticket #000025)
    ("5f", "function"): "bench/fixtures/5f/function-v1.jsonl",
    ("5f", "finetuning"): "bench/fixtures/5f/finetuning-v1.jsonl",
    ("5f", "falsification"): "bench/fixtures/5f/falsification-v1.jsonl",
    ("5f", "formulate"): "bench/fixtures/5f/formulate-v1.jsonl",
    ("5f", "feedback-loop"): "bench/fixtures/5f/feedback-loop-v1.jsonl",
    # 5R — Phase 2 of #000021 (ticket #000021 §7 Phase 2; depends on
    # SelfModel + memory_root, both landed)
    ("5r", "react"): "bench/fixtures/5r/react-v1.jsonl",
    ("5r", "rearrange"): "bench/fixtures/5r/rearrange-v1.jsonl",
    ("5r", "restore"): "bench/fixtures/5r/restore-v1.jsonl",
    ("5r", "replicate"): "bench/fixtures/5r/replicate-v1.jsonl",
    ("5r", "resonate"): "bench/fixtures/5r/resonate-v1.jsonl",
}


def _run_one(battery: str, sub: str, fixtures: Path) -> BatteryResult:
    if battery not in _BATTERIES:
        raise SystemExit(f"unknown battery {battery!r}")
    if sub not in _BATTERIES[battery]:
        raise SystemExit(f"unknown sub-battery {sub!r} for {battery}")
    return _BATTERIES[battery][sub](fixtures)


def _emit(results: list[BatteryResult], out: Optional[str]) -> None:
    payload = {
        "schema_version": "bench-result-v1",
        "results": [asdict(r) for r in results],
    }
    encoded = json.dumps(payload, indent=2, ensure_ascii=False, default=str)
    if out:
        Path(out).parent.mkdir(parents=True, exist_ok=True)
        Path(out).write_text(encoded + "\n", encoding="utf-8")
        print(f"wrote {out}")
    else:
        print(encoded)


def main(argv: Optional[list[str]] = None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--battery", choices=sorted(_BATTERIES), default=None)
    p.add_argument("--sub", default=None, help="sub-battery name")
    p.add_argument("--fixtures", default=None, help="fixture JSONL path")
    p.add_argument(
        "--all",
        dest="run_all",
        action="store_true",
        help="run all Phase-1a sub-batteries with default fixtures",
    )
    p.add_argument("--out", default=None, help="write JSON to this path")
    args = p.parse_args(argv)

    results: list[BatteryResult] = []
    if args.run_all:
        for (battery, sub), fixture_path in _DEFAULT_FIXTURES.items():
            results.append(_run_one(battery, sub, Path(fixture_path)))
    else:
        if not args.battery or not args.sub:
            p.error("--battery and --sub are required unless --all is given")
        if args.fixtures:
            fixtures = Path(args.fixtures)
        else:
            key = (args.battery, args.sub)
            if key not in _DEFAULT_FIXTURES:
                p.error(
                    f"no default fixture for ({args.battery},{args.sub}); "
                    "pass --fixtures explicitly"
                )
            fixtures = Path(_DEFAULT_FIXTURES[key])
        results.append(_run_one(args.battery, args.sub, fixtures))

    _emit(results, args.out)
    failed = sum(r.fail_count for r in results)
    return 1 if failed else 0


if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())