cli.py

"""Command line entry point for RCAgentBench.

python cli.py scenarios              inspect the built traces
python cli.py benchmark --runs 3     full comparison, needs an API key
python cli.py benchmark --heuristic  the rule-based floor, no API key
python cli.py report                 turn the latest results into markdown
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

from agent.providers import PROVIDERS
from evaluation.cost import describe_models, format_cost
from evaluation.evaluator import Evaluator, FatalRunError
from evaluation.report import render_markdown_report, render_report
from traces.builder import ScenarioError
from traces.store import TraceStore

DEFAULT_SCENARIOS = "scenarios"
DEFAULT_ANTHROPIC_MODEL = "claude-opus-5"
RESULTS_DIR = Path("results")
ANOMALY_FACTOR = 3.0


def cmd_scenarios(args: argparse.Namespace) -> int:
    """Print each scenario's assembled trace."""
    store = TraceStore.from_directory(args.scenarios)
    print(f"{len(store)} scenarios in {args.scenarios}\n")

    for scenario in store.scenarios:
        trace = scenario.trace
        print(f"{scenario.name}  [{scenario.difficulty}]")
        print(f"  trace_id  {scenario.trace_id}")
        print(f"  duration  {trace.duration_ms:.0f}ms across {len(trace.spans)} spans")
        print(
            f"  expected  {scenario.expectation.service} "
            f"({scenario.expectation.fault_type})"
        )
        for span in trace.spans:
            baseline = trace.baselines.get(span.service, 0.0)
            flag = ""
            if baseline and span.self_ms / baseline >= ANOMALY_FACTOR:
                flag = f"  <- {span.self_ms / baseline:.0f}x baseline"
            if span.is_error:
                code = f" {span.status_code}" if span.status_code else " (propagated)"
                flag += f"  <- error{code}"
            indent = "    " + "  " * span.depth
            print(
                f"{indent}{span.service}.{span.operation}  "
                f"{span.duration_ms:.0f}ms total / {span.self_ms:.0f}ms self{flag}"
            )
        print()
    return 0


def _default_output(runs: int, heuristic_only: bool) -> Path:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M")
    kind = "heuristic" if heuristic_only else "benchmark"
    suffix = f"-x{runs}" if runs > 1 else ""
    return RESULTS_DIR / f"{kind}-{stamp}{suffix}.json"


def cmd_benchmark(args: argparse.Namespace) -> int:
    """Run the evaluation and write a report."""
    evaluator = Evaluator.from_directory(args.scenarios)

    if args.heuristic:
        tool_sets = ["heuristic"]
    elif args.no_baseline:
        tool_sets = list(args.tool_sets)
    else:
        # The floor runs alongside the model by default. A model score means
        # little without knowing what simple rules already achieve.
        tool_sets = ["heuristic", *args.tool_sets]

    # Each provider names its models differently, so an unset --model picks
    # that provider's default rather than an Anthropic model id it would reject.
    if args.provider != "anthropic" and args.model == DEFAULT_ANTHROPIC_MODEL:
        args.model = PROVIDERS[args.provider].default_model

    scenarios = list(evaluator.store.scenarios)
    if args.limit:
        scenarios = scenarios[: args.limit]

    output = Path(args.output) if args.output else _default_output(args.runs, args.heuristic)
    uses_model = any(ts != "heuristic" for ts in tool_sets)

    total = (
        len(scenarios)
        * args.runs
        * sum(1 if ts == "heuristic" else len(args.styles) for ts in tool_sets)
    )
    print(
        f"{len(scenarios)} scenarios x {args.runs} "
        f"pass{'es' if args.runs != 1 else ''} = {total} evaluations"
    )

    if uses_model:
        suffix = f", effort={args.effort}" if args.effort else ""
        print(f"Provider: {args.provider}  Model: {args.model}{suffix}")
        if args.provider == "anthropic":
            print(describe_models([args.model]))
            print(
                "Cost accumulates below as runs complete. Estimates use list "
                "price; your rate may be lower."
            )
        else:
            provider = PROVIDERS[args.provider]
            if provider.notes:
                print(f"  {provider.notes}")
            if provider.env_var:
                print(f"  reads the key from {provider.env_var}")

    # Checkpointing is on by default for model-backed runs. Losing completed
    # work to a transient error would mean paying for it twice.
    checkpoint = Path(args.checkpoint) if args.checkpoint else output.with_suffix(".jsonl")
    if not args.no_checkpoint:
        recovered = evaluator.use_checkpoint(checkpoint)
        if recovered:
            print(f"Resuming from {checkpoint}: {recovered} runs already complete")
        else:
            print(f"Checkpointing to {checkpoint}")
    print()

    try:
        evaluator.run(
            tool_sets=tool_sets,
            instruction_styles=args.styles,
            model=args.model,
            runs=args.runs,
            scenarios=scenarios,
            effort=args.effort,
            provider=args.provider,
        )
    except FatalRunError as exc:
        print(f"\nStopping: {exc}", file=sys.stderr)
        print(
            "This failure would repeat on every remaining call, so the run "
            "stopped rather than spending more.",
            file=sys.stderr,
        )
        if not args.no_checkpoint:
            print(f"Completed runs are safe in {checkpoint}.", file=sys.stderr)
        return 2

    payload = evaluator.save(output)
    print()
    print(render_report(payload))

    spend = evaluator.spend_so_far()
    if spend is not None:
        print(f"Estimated cost at list price: {format_cost(spend)}\n")

    print(f"Results written to {output}")

    if not args.no_report:
        markdown = render_markdown_report(payload, source=output.as_posix())
        report_path = RESULTS_DIR / "report.md"
        report_path.write_text(markdown, encoding="utf-8")
        print(f"Markdown report written to {report_path}")

    return 0 if payload["summary"]["runs"] else 1


def _latest_results() -> Path | None:
    candidates = [p for p in RESULTS_DIR.glob("*.json")]
    return max(candidates, key=lambda p: p.stat().st_mtime) if candidates else None


def cmd_report(args: argparse.Namespace) -> int:
    """Render a saved results file as markdown."""
    source = Path(args.input) if args.input else _latest_results()
    if source is None:
        print(
            f"no results found in {RESULTS_DIR}; run 'benchmark' first",
            file=sys.stderr,
        )
        return 1
    if not source.is_file():
        print(f"no such results file: {source}", file=sys.stderr)
        return 1

    payload = json.loads(source.read_text(encoding="utf-8"))
    markdown = render_markdown_report(payload, source=source.as_posix())

    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(markdown, encoding="utf-8")

    print(render_report(payload))
    print(f"Read {source}")
    print(f"Wrote {output}")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="rcagentbench",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--scenarios", default=DEFAULT_SCENARIOS, help="directory of scenario YAML files"
    )

    sub = parser.add_subparsers(dest="command", required=True)

    scenarios = sub.add_parser("scenarios", help="show the traces scenarios produce")
    scenarios.set_defaults(func=cmd_scenarios)

    bench = sub.add_parser("benchmark", help="evaluate agents against the scenarios")
    bench.add_argument(
        "--model",
        default=DEFAULT_ANTHROPIC_MODEL,
        help="model id; unset uses the chosen provider's default",
    )
    bench.add_argument(
        "--provider",
        default="anthropic",
        choices=["anthropic", *sorted(PROVIDERS)],
        help="anthropic (default) or an OpenAI-compatible endpoint",
    )
    bench.add_argument(
        "--effort",
        choices=["low", "medium", "high", "xhigh", "max"],
        help="thinking depth; Anthropic models only, ignored elsewhere",
    )
    bench.add_argument(
        "--limit",
        type=int,
        metavar="N",
        help="use only the first N scenarios, for a cheap pilot run",
    )
    bench.add_argument(
        "--checkpoint",
        help="JSONL file of completed runs (default: alongside --output)",
    )
    bench.add_argument(
        "--no-checkpoint",
        action="store_true",
        help="do not record or resume from completed runs",
    )
    bench.add_argument(
        "--runs",
        type=int,
        default=1,
        metavar="N",
        help="passes over the full matrix; more than one reports variance",
    )
    bench.add_argument(
        "--tool-sets",
        nargs="+",
        default=["granular", "analytical", "jaeger_mcp"],
        dest="tool_sets",
        help="granular, analytical, jaeger_mcp, or heuristic",
    )
    bench.add_argument("--styles", nargs="+", default=["strict", "exploratory"])
    bench.add_argument(
        "--heuristic",
        action="store_true",
        help="run only the rule-based floor; needs no API key",
    )
    bench.add_argument(
        "--no-baseline",
        action="store_true",
        help="skip the rule-based floor and run only model configurations",
    )
    bench.add_argument("--output", help="where to write JSON (default: timestamped)")
    bench.add_argument(
        "--no-report", action="store_true", help="skip writing results/report.md"
    )
    bench.set_defaults(func=cmd_benchmark)

    report = sub.add_parser("report", help="render saved results as markdown")
    report.add_argument("--input", help="results JSON (default: most recent)")
    report.add_argument("--output", default=str(RESULTS_DIR / "report.md"))
    report.set_defaults(func=cmd_report)

    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        return args.func(args)
    except ScenarioError as exc:
        print(f"scenario error: {exc}", file=sys.stderr)
        return 2
    except RuntimeError as exc:
        print(f"{exc}", file=sys.stderr)
        return 2
    except KeyboardInterrupt:
        print("\ninterrupted", file=sys.stderr)
        return 130


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