"""Standalone CLI for the benchmark framework. Invoke from the opensre repo root with: uv run python -m tests.benchmarks._framework.cli [args] Subcommands: list Show available adapters and their metric schemas validate Load + lint a config; exit non-zero if dishonest run [--dev] Load config, instantiate adapter, run benchmark run-stub Same as run but uses a fake LLM (no API cost) — useful for testing the wiring The CLI is deliberately standalone — not a subcommand of opensre's main CLI — so the framework stays decoupled from opensre's CLI dispatcher. A future ``opensre bench`` subcommand can wrap this if user-facing surfacing is needed. Exit codes: 0 success 1 config lint failed (anti-pattern) 2 integrity gate blocked the run / report 3 cost budget exceeded mid-run 4 no adapter for ``config.benchmark`` 5 pre-flight failed for some other reason """ from __future__ import annotations import argparse import sys from pathlib import Path from tests.benchmarks._framework.adapters import BenchmarkAdapter from tests.benchmarks._framework.config import ( load_config, validate_config_or_raise, ) from tests.benchmarks._framework.cost import CostBudgetExceeded from tests.benchmarks._framework.integrity import IntegrityViolation # --------------------------------------------------------------------------- # # Exit codes # # # # Stable contract: external tooling (ECS task definitions, CI conditionals, # # wrapper scripts) may key off these. Document changes prominently. The # # values match the module docstring at top. # # --------------------------------------------------------------------------- # EXIT_OK = 0 EXIT_USAGE_OR_INPUT = 1 # Bad path, missing file, config lint failure EXIT_INTEGRITY_VIOLATION = 2 # IntegrityGuard rejected the config or report EXIT_BUDGET_EXCEEDED = 3 # CostBudgetExceeded mid-run OR outcome.aborted EXIT_UNKNOWN_ADAPTER = 4 # No adapter registered for config.benchmark EXIT_PREFLIGHT_ERROR = 5 # Uncaught exception during run setup from tests.benchmarks._framework.reporting import render_report_dir from tests.benchmarks._framework.runner import BenchmarkRunner # --------------------------------------------------------------------------- # # Adapter registry # # --------------------------------------------------------------------------- # def _build_adapter(name: str) -> BenchmarkAdapter: """Map ``config.benchmark`` to an adapter instance via the registry. The framework's ``adapter_registry`` (see ``_framework/adapters.py``) is the single source of truth for which adapters exist. This wrapper bootstraps known adapters on first call and then delegates. """ from tests.benchmarks._framework.adapters import ( build_adapter, ensure_known_adapters_registered, ) ensure_known_adapters_registered() return build_adapter(name) def _known_adapters() -> list[str]: """Adapters this CLI knows how to construct, via the registry.""" from tests.benchmarks._framework.adapters import ( ensure_known_adapters_registered, known_adapters, ) ensure_known_adapters_registered() return known_adapters() # --------------------------------------------------------------------------- # # Subcommands # # --------------------------------------------------------------------------- # def _cmd_list(_args: argparse.Namespace) -> int: print("Adapters known to this CLI:") for name in _known_adapters(): try: adapter = _build_adapter(name) except Exception as exc: print(f" - {name} (failed to construct: {exc})") continue schema = adapter.metric_schema() completeness = schema.validate_completeness() status = "✓ ready" if not completeness else f"⚠ {len(completeness)} issue(s)" print(f" - {name} v{adapter.version} ({len(schema.all_metrics())} metrics, {status})") if completeness: for err in completeness: print(f" - {err}") return EXIT_OK def _cmd_validate(args: argparse.Namespace) -> int: path = Path(args.config) if not path.exists(): print(f" ✗ {path} does not exist", file=sys.stderr) return EXIT_USAGE_OR_INPUT try: config = validate_config_or_raise(path) except (FileNotFoundError, ValueError) as exc: print(f" ✗ {path}\n{exc}", file=sys.stderr) return EXIT_USAGE_OR_INPUT print(f" ✓ {path}") print(f" benchmark: {config.benchmark}") print(f" modes: {config.modes}") print(f" llms ({len(config.llms)}): {config.llms}") print(f" runs_per_case: {config.runs_per_case}") print(f" workers: {config.workers}") print(f" cost_budget_usd: ${config.cost_budget_usd:.2f}") print(f" output_dir: {config.output_dir}") if config.pre_registration_path: print(f" pre_registration_path: {config.pre_registration_path}") return EXIT_OK def _cmd_run(args: argparse.Namespace) -> int: path = Path(args.config) try: config = load_config(path) except FileNotFoundError as exc: print(f" ✗ {exc}", file=sys.stderr) return EXIT_USAGE_OR_INPUT if not args.dev: # Production runs MUST pass the lint pre-check lint_errors = config.lint() if lint_errors: print(" ✗ Config failed integrity lint:", file=sys.stderr) for err in lint_errors: print(f" - {err}", file=sys.stderr) return EXIT_USAGE_OR_INPUT try: adapter = _build_adapter(config.benchmark) except KeyError: print( f" ✗ no adapter registered for benchmark={config.benchmark!r}. " f"Known: {_known_adapters()}", file=sys.stderr, ) return EXIT_UNKNOWN_ADAPTER # Strategy-pattern hand-off: each adapter owns its own config-knob # handling via ``apply_config_overrides``. The framework no longer # needs to know which adapter honors which fields. Default (base # BenchmarkAdapter.apply_config_overrides) is a no-op for adapters # without runtime knobs. adapter.apply_config_overrides(config) runner = BenchmarkRunner(config=config, adapter=adapter, config_path=path) try: outcome = runner.run_without_integrity() if args.dev else runner.run() except IntegrityViolation as v: print(f" ✗ Integrity gate blocked the run:\n{v}", file=sys.stderr) return EXIT_INTEGRITY_VIOLATION except CostBudgetExceeded as exc: # Defensive: BenchmarkRunner.run() normally catches CostBudgetExceeded # internally and returns RunOutcome(aborted=True). This except remains # for direct callers or future code paths that bypass that handling. print(f" ✗ Cost budget exceeded mid-run: {exc}", file=sys.stderr) return EXIT_BUDGET_EXCEEDED except Exception as exc: print(f" ✗ Pre-flight failed: {type(exc).__name__}: {exc}", file=sys.stderr) return EXIT_PREFLIGHT_ERROR print() print(f" ✓ Run complete: {len(outcome.cells)} cell(s), aborted={outcome.aborted}") print(f" ✓ run_id: {outcome.report.run_id}") print(f" ✓ artifacts: {outcome.report.raw_artifacts_dir}") if outcome.abort_reason: print(f" ⚠ abort reason: {outcome.abort_reason}", file=sys.stderr) # Aborted runs (e.g. budget overrun caught inside the runner) must NOT # report success — ECS / CI determine task success from the exit code, # and a halted run that exits 0 is silently lost. Return the same code # as the CostBudgetExceeded path above so wrapping tooling can treat # both as a single class of failure. if outcome.aborted: return EXIT_BUDGET_EXCEEDED return EXIT_OK def _cmd_report(args: argparse.Namespace) -> int: run_dir = Path(args.run_dir) if not run_dir.exists(): print(f" ✗ {run_dir} does not exist", file=sys.stderr) return EXIT_USAGE_OR_INPUT formats = [f.strip() for f in args.format.split(",")] if args.format else None try: rendered = render_report_dir(run_dir, formats=formats) except FileNotFoundError as exc: print(f" ✗ {exc}", file=sys.stderr) return EXIT_USAGE_OR_INPUT for fmt, path in rendered.items(): print(f" ✓ {fmt}: {path} ({path.stat().st_size:,} bytes)") return EXIT_OK # --------------------------------------------------------------------------- # # Entry point # # --------------------------------------------------------------------------- # def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="bench", description="Standalone CLI for the opensre benchmark framework.", ) sub = parser.add_subparsers(dest="cmd", required=True, metavar="command") p_list = sub.add_parser("list", help="Show available adapters.") p_list.set_defaults(func=_cmd_list) p_validate = sub.add_parser("validate", help="Load + lint a config; exit non-zero on failure.") p_validate.add_argument("config", help="Path to YAML config.") p_validate.set_defaults(func=_cmd_validate) p_run = sub.add_parser("run", help="Run a benchmark from a YAML config.") p_run.add_argument("config", help="Path to YAML config.") p_run.add_argument( "--dev", action="store_true", help=( "DEVELOPMENT ONLY: skip integrity gates. Results stamped with " "dev_mode=True (run_id prefix) so they can't be silently promoted." ), ) p_run.set_defaults(func=_cmd_run) p_report = sub.add_parser( "report", help="Re-render report.md + report.html from a finished run's report.json.", ) p_report.add_argument("run_dir", help="Directory containing report.json + cases/") p_report.add_argument( "--format", default="markdown,html", help="Comma-separated subset of {markdown,html}. Default: both.", ) p_report.set_defaults(func=_cmd_report) return parser def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) return int(args.func(args)) if __name__ == "__main__": sys.exit(main())