Blame for analysis/error_analysis.py

"""Heuristic failure-mode tagging for a run's traces (no gold answers needed).

Reads ``runs/traces/<run_id>/*.json`` (schema: gaia_agent/trace.py) plus the run's rows in
``runs/answers.jsonl`` (for questions that timed out before a trace was written), tags each
task with observable failure modes, writes ``runs/error_analysis.json`` and prints a table.

The tags describe *how the agent behaved*, not whether the answer was right: this runs on
the 20 course questions, whose answers we never see. It is the open-coding pass of an error
analysis; the counts tell you which tool or prompt fix to try first.

Tags
----
hit_max_steps              ran out of steps (AgentMaxStepsError / "Reached max steps")
tool_error                 a tool returned or raised an error; also ``tool_error:<tool>``
code_execution_error       the agent's own Python failed (not inside a tool call)
parse_error                model output had no parsable code block
rate_limited               the *model* call hit a 429 / quota (tool-side 429s are tool errors)
timeout                    the question hit the runner's wall-clock limit, or a model timeout
empty_or_refusal_answer    no answer, or a refusal / "unknown" style answer
answer_format_suspicious   answer has units, a sentence, "FINAL ANSWER", quotes, or is long
blocked_by_guard           the answer-leak guard refused a query or URL
no_tool_used               finished without calling a single tool
repeated_identical_tool_call  same tool + same arguments 3+ times, or identical code twice in a row
paging_loop                same tool on the same first argument 4+ times with other args varying
vision_used                an image / video-frame question went to the vision model
no_trace                   only an answers.jsonl row exists (usually a timeout)

Usage: ``python -m analysis.error_analysis [--run-id ID] [--runs-dir runs] [--max-steps 12]``
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

PROJECT_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_RUNS_DIR = PROJECT_ROOT / "runs"

TAG_DESCRIPTIONS: dict[str, str] = {
    "hit_max_steps": "Ran out of steps before a final answer",
    "tool_error": "A tool returned or raised an error",
    "code_execution_error": "The agent's own Python code failed",
    "parse_error": "Model output had no parsable code block",
    "rate_limited": "A model call hit a 429 / quota",
    "timeout": "Wall-clock or model timeout",
    "empty_or_refusal_answer": "Empty, refusal or 'unknown' answer",
    "answer_format_suspicious": "Answer shape likely to fail exact match",
    "blocked_by_guard": "Answer-leak guard refused a query/URL",
    "no_tool_used": "Answered without calling any tool",
    "repeated_identical_tool_call": "Looped on an identical tool call",
    "paging_loop": "Same tool on the same target 4+ times (paging/re-reading)",
    "vision_used": "Used the vision model",
    "no_trace": "No trace file (answers.jsonl row only)",
}

# Tools the agent may expose (gaia_agent/tools_web.py + tools_media.py). Names seen in a run's
# own traces are added at runtime, so a renamed tool is still recognised.
DEFAULT_TOOL_NAMES = {
    "web_search",
    "visit_webpage",
    "wikipedia_search",
    "wikipedia_page",
    "wikipedia_revision",
    "wikipedia_page_info",
    "read_file",
    "run_python_file",
    "transcribe_audio",
    "image_question",
    "youtube_transcript",
    "youtube_audio_transcript",
    "youtube_frames_question",
}
VISION_TOOLS = {"image_question", "youtube_frames_question"}
NON_TOOLS = {"final_answer", "python_interpreter"}

# --------------------------------------------------------------------------- patterns

GUARD_RE = re.compile(r"BLOCKED by the answer-leak guard|\b(query|URL) refused\b", re.I)
# "429" only as a standalone token: a bare substring match also hits UUIDs and byte counts.
HTTP_429_RE = re.compile(r"(?<![0-9A-Za-z_-])429(?![0-9A-Za-z_-])")
RATE_RE = re.compile(
    r"rate[ _-]?limit|too many requests|RemoteQuotaExhausted|BudgetExceeded|quota (?:exceeded|exhausted)"
    r"|daily (?:request )?cap|resource[_ ]exhausted",
    re.I,
)
TIMEOUT_RE = re.compile(r"\btime(?:d)?[ _-]?out\b|timeouterror|APITimeoutError|deadline exceeded", re.I)
MAX_STEPS_RE = re.compile(r"reached max steps|AgentMaxStepsError|max(?:imum)? (?:number of )?steps", re.I)
PARSE_RE = re.compile(r"Error in code parsing|regex pattern .* was not found|AgentParsingError", re.I | re.S)
EXEC_RE = re.compile(r"Code execution failed|InterpreterError|AgentExecutionError", re.I)
FAILED_LINE_RE = re.compile(r"Code execution failed at line '(?P<line>.*?)' due to", re.S)
TOOL_EXC_RE = re.compile(r"Error (?:when )?executing tool '?(?P<tool>[A-Za-z_][A-Za-z0-9_]*)'?")
TOOL_ERROR_LINE_RE = re.compile(r"^\s*(?:ERROR|Error)\b[:\s]", re.M)
MODEL_ERROR_TYPES = {"AgentGenerationError"}

# Bare "unknown"/"none" style answers must match the whole answer ("Unknown Pleasures" is a
# legitimate album title); refusal phrases only need to open the answer.
REFUSAL_FULL_RE = re.compile(r"^(?:unknown|n/?a|none|null|nil|not (?:found|available|known|applicable)|no answer|\?+)$", re.I)
REFUSAL_PREFIX_RE = re.compile(
    r"^(?:cannot (?:be )?determined?|unable to (?:determine|find|answer)"
    r"|i(?: can ?not| cannot| can't| am unable|'m unable| do not know| don't know| could not| couldn't)"
    r"|i'?m sorry|sorry\b|insufficient (?:information|data)|it is not possible|there is no (?:way|information))",
    re.I,
)
UNIT_RE = re.compile(
    r"(?:^[$€£¥])|(?:\d\s*%$)|(?:\b\d[\d.,]*\s*(?:km|kg|mg|g|m|cm|mm|mi|miles?|lbs?|pounds?|ft|feet|inch(?:es)?|"
    r"meters?|metres?|seconds?|sec|s|minutes?|mins?|hours?|hrs?|h|days?|years?|yrs?|usd|eur|dollars?|euros?|"
    r"percent|°c|°f|k?w|mph|kph)\b)",
    re.I,
)
THOUSANDS_RE = re.compile(r"^\d{1,3}(?:,\d{3})+(?:\.\d+)?$")
SENTENCE_VERB_RE = re.compile(r"\b(?:is|are|was|were|has|have|had|the answer|answer is|would be|equals)\b", re.I)


# --------------------------------------------------------------------------- helpers


def _walk(steps: Iterable[dict]) -> Iterable[dict]:
    """Every step, including steps of sub-agents nested under an action step."""
    for s in steps or []:
        yield s
        for run in s.get("sub_agents") or []:
            yield from _walk(run.get("steps") or [])


def _short(text: Any, n: int = 140) -> str:
    s = re.sub(r"\s+", " ", str(text or "")).strip()
    return s if len(s) <= n else s[: n - 1] + "…"


def _tool_names_in_code(code: str | None, known: set[str]) -> list[str]:
    if not code:
        return []
    return [m.group(1) for m in re.finditer(r"\b([A-Za-z_][A-Za-z0-9_]*)\s*\(", code) if m.group(1) in known]


def _calls_in_code(code: str | None, known: set[str]) -> list[dict]:
    """Calls to known tools in a code block, with their argument source text as the key.

    Uses ``ast`` so ``visit_webpage(url_a)`` and ``visit_webpage(url_b)`` stay distinct; if the
    code does not parse, falls back to bare names with ``arguments=None`` (never counted as
    repeats, since identical names alone prove nothing).
    """
    if not code:
        return []
    import ast

    try:
        tree = ast.parse(code)
    except SyntaxError:
        return [{"name": n, "arguments": None} for n in _tool_names_in_code(code, known)]
    calls = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in known:
            args = [ast.unparse(a) for a in node.args] + [f"{k.arg}={ast.unparse(k.value)}" for k in node.keywords]
            first = args[0].split("=", 1)[-1] if args else None
            calls.append({"name": node.func.id, "arguments": ", ".join(args), "first": first, "_lineno": node.lineno})
    calls.sort(key=lambda c: c.pop("_lineno"))
    return calls


def _step_tools(step: dict, known: set[str]) -> list[dict]:
    """Tool calls of one step: the trace's recovered calls, else calls parsed from the code."""
    calls = [c for c in step.get("tool_calls") or [] if c.get("name") and c["name"] not in NON_TOOLS]
    if calls:
        return calls
    return _calls_in_code(step.get("code"), known)


def _is_vision_tool(name: str) -> bool:
    # Token match, not substring: "wikipedia_revision" contains "vision".
    return name in VISION_TOOLS or bool(re.search(r"(?:^|_)(?:vision|vlm)(?:_|$)", name))


def _first_arg(call: dict) -> str | None:
    if call.get("first") is not None:
        return str(call["first"])
    args = call.get("arguments")
    if isinstance(args, dict) and args:
        return str(next(iter(args.values())))
    if isinstance(args, (list, tuple)) and args:
        return str(args[0])
    return None


def _args_key(args: Any) -> str:
    try:
        return json.dumps(args, sort_keys=True, default=str)
    except Exception:
        return str(args)


def format_issues(answer: Any) -> list[str]:
    """Reasons an answer string looks wrong for GAIA's exact-match grader (empty list = fine)."""
    if answer is None:
        return []
    a = str(answer).strip()
    if not a:
        return []
    reasons = []
    if re.search(r"final\s*answer", a, re.I):
        reasons.append('contains "FINAL ANSWER"')
    if "\n" in a:
        reasons.append("multi-line")
    items = [x.strip() for x in a.split(",")] if "," in a and not THOUSANDS_RE.match(a) else [a]
    longest_item_words = max(len(x.split()) for x in items)
    if len(a) > 200 or longest_item_words > 12:
        reasons.append(f"very long ({len(a)} chars)")
    if any(UNIT_RE.search(x) for x in items):
        reasons.append("has units/currency/percent")
    if THOUSANDS_RE.match(a):
        reasons.append("thousands separator")
    words = a.split()
    if (a.endswith(".") and len(words) >= 4) or (len(words) >= 6 and SENTENCE_VERB_RE.search(a) and "," not in a):
        reasons.append("reads like a sentence")
    if len(a) >= 2 and a[0] in "\"'“‘`" and a[-1] in "\"'”’`":
        reasons.append("wrapped in quotes")
    if a.startswith(("**", "__")) or a.startswith("```"):
        reasons.append("markdown formatting")
    return reasons


def is_empty_or_refusal(answer: Any) -> str | None:
    if answer is None:
        return "no final answer"
    a = str(answer).strip().strip(".").strip()
    if not a:
        return "empty answer"
    if REFUSAL_FULL_RE.match(a) or REFUSAL_PREFIX_RE.match(a):
        return f"refusal-like answer {_short(a, 60)!r}"
    return None


# --------------------------------------------------------------------------- core


def analyze_trace(
    trace: dict,
    *,
    max_steps: int | None = None,
    answer_row: dict | None = None,
    known_tools: set[str] | None = None,
) -> dict:
    """Tag one trace. Returns {task_id, tags, evidence, details}."""
    known = set(DEFAULT_TOOL_NAMES) | set(known_tools or ())
    for name in (trace.get("totals") or {}).get("tool_calls_by_name", {}) or {}:
        if name not in NON_TOOLS:
            known.add(name)

    details: dict[str, list[str]] = {}

    def tag(name: str, ev: str) -> None:
        lst = details.setdefault(name, [])
        if ev not in lst and len(lst) < 4:
            lst.append(ev)

    steps = list(_walk(trace.get("steps") or []))
    top_actions = [s for s in (trace.get("steps") or []) if s.get("type") == "action"]
    run_error = trace.get("error") or (answer_row or {}).get("error")
    run_error_type = trace.get("error_type")

    # ---- run-level errors
    if run_error:
        if TIMEOUT_RE.search(str(run_error)) or (run_error_type or "").lower().endswith("timeouterror"):
            tag("timeout", f"run error: {_short(run_error, 100)}")
        if HTTP_429_RE.search(str(run_error)) or RATE_RE.search(str(run_error)) or (
            run_error_type in ("BudgetExceeded", "RemoteQuotaExhausted")
        ):
            tag("rate_limited", f"run error: {_short(run_error, 100)}")
        if MAX_STEPS_RE.search(str(run_error)) or run_error_type == "AgentMaxStepsError":
            tag("hit_max_steps", f"run error: {_short(run_error, 80)}")

    # ---- per-step
    call_counter: Counter = Counter()
    target_counter: Counter = Counter()
    call_first_step: dict[str, Any] = {}
    prev_code = None
    vision_steps: list[str] = []
    any_tool = False
    for s in steps:
        idx = s.get("index")
        where = f"step {idx}" + (f" ({s['agent']})" if s.get("agent") else "")
        err = s.get("error") or ""
        etype = s.get("error_type") or ""
        obs = s.get("observation") or ""
        tools = _step_tools(s, known) if s.get("type") == "action" else []
        tool_names = [t["name"] for t in tools]
        if tools:
            any_tool = True

        if etype == "AgentMaxStepsError" or (err and MAX_STEPS_RE.search(err)):
            tag("hit_max_steps", f"{where}: {_short(err, 80)}")

        if err:
            model_side = etype in MODEL_ERROR_TYPES or err.startswith("Error in generating model output")
            if model_side and (HTTP_429_RE.search(err) or RATE_RE.search(err)):
                tag("rate_limited", f"{where}: {_short(err, 100)}")
            if model_side and TIMEOUT_RE.search(err):
                tag("timeout", f"{where}: model call timed out")
            if etype == "AgentParsingError" or PARSE_RE.search(err):
                tag("parse_error", f"{where}: {_short(err, 90)}")
            elif etype in ("AgentToolExecutionError", "AgentToolCallError") or TOOL_EXC_RE.search(err):
                m = TOOL_EXC_RE.search(err)
                name = m.group("tool") if m else (tool_names[0] if len(set(tool_names)) == 1 else "unattributed")
                tag(f"tool_error:{name}", f"{where}: {_short(err, 110)}")
            elif etype == "AgentExecutionError" or EXEC_RE.search(err):
                failing = FAILED_LINE_RE.search(err)
                in_tool = [n for n in _tool_names_in_code(failing.group("line"), known)] if failing else []
                if GUARD_RE.search(err):
                    tag("blocked_by_guard", f"{where}: {_short(err, 90)}")
                elif in_tool:
                    tag(f"tool_error:{in_tool[0]}", f"{where}: raised in {in_tool[0]}(): {_short(err, 90)}")
                elif not model_side:
                    tag("code_execution_error", f"{where}: {_short(err, 110)}")

        if GUARD_RE.search(obs) or GUARD_RE.search(err):
            tag("blocked_by_guard", f"{where}: guard refused a {'query' if 'query refused' in (obs + err) else 'URL/result'}")

        # Tools in this codebase report failures as returned strings starting with "ERROR".
        for m in TOOL_ERROR_LINE_RE.finditer(obs):
            line = obs[m.start() : obs.find("\n", m.start()) if obs.find("\n", m.start()) != -1 else len(obs)]
            if GUARD_RE.search(line):
                continue
            named = [n for n in set(tool_names) if n in line]
            if named:
                name = named[0]
            elif len(set(tool_names)) == 1:
                name = tool_names[0]
            else:
                name = "unattributed"
            tag(f"tool_error:{name}", f"{where}: {_short(line, 110)}")

        for t in tools:
            first = _first_arg(t)
            if first is not None:
                target_counter[f"{t['name']}|{first}"] += 1
            if t.get("arguments") is not None:
                key = f"{t['name']}({_args_key(t.get('arguments'))})"
                call_counter[key] += 1
                call_first_step.setdefault(key, idx)
            if _is_vision_tool(t["name"]):
                vision_steps.append(f"{where}: {t['name']}")

        code = (s.get("code") or "").strip()
        if s.get("type") == "action" and code and prev_code is not None and code == prev_code and tools:
            tag("repeated_identical_tool_call", f"{where}: identical code to the previous step")
        if s.get("type") == "action":
            prev_code = code or None

    for key, n in call_counter.items():
        if n >= 3:
            tag("repeated_identical_tool_call", f"{_short(key, 100)} x{n}")
    for key, n in target_counter.items():
        if n >= 4 and not any(k.startswith(key.split("|", 1)[0] + "(") and c >= 3 for k, c in call_counter.items()):
            name, target = key.split("|", 1)
            tag("paging_loop", f"{name} on {_short(target, 80)} x{n}")

    for v in vision_steps:
        tag("vision_used", v)

    ran_code = [s for s in top_actions if (s.get("code") or "").strip()]
    # Reading an attachment directly in code (pandas, open()) is real work, not "answered from memory".
    read_direct = any(re.search(r"\b(read_csv|read_excel|read_json|open\()", s.get("code") or "") for s in ran_code)
    if ran_code and not any_tool and not read_direct:  # parse failures that never ran code are not "answered without tools"
        tag("no_tool_used", f"{len(ran_code)} code step(s), no tool calls")

    if max_steps is not None and len(top_actions) >= max_steps and "hit_max_steps" not in details:
        if not any(s.get("is_final_answer") for s in top_actions):
            tag("hit_max_steps", f"{len(top_actions)} action steps >= max_steps {max_steps} without final_answer")

    # ---- the answer
    final = trace.get("final_answer")
    if final is None and answer_row is not None:
        final = answer_row.get("answer")
    empty = is_empty_or_refusal(final)
    if empty:
        tag("empty_or_refusal_answer", empty)
    else:
        issues = format_issues(final)
        raw = (answer_row or {}).get("raw")
        if not issues and raw is not None and raw != final:
            raw_issues = format_issues(raw)
            if raw_issues:  # the normalizer cleaned it, but the model's habit is worth knowing
                details.setdefault("_notes", []).append(f"raw answer had: {', '.join(raw_issues)}")
        if issues:
            tag("answer_format_suspicious", f"{', '.join(issues)}: {_short(final, 70)!r}")

    # Collapse tool_error:<name> into the generic tag as well.
    tool_tags = sorted(k for k in details if k.startswith("tool_error:"))
    if tool_tags:
        details["tool_error"] = [k.split(":", 1)[1] for k in tool_tags]

    ordered = [t for t in TAG_DESCRIPTIONS if t in details] + tool_tags
    evidence = "; ".join(f"{t}: {details[t][0]}" for t in ordered if t != "tool_error")
    return {
        "task_id": trace.get("task_id"),
        "tags": ordered,
        "evidence": evidence,
        "details": {k: v for k, v in details.items()},
        "final_answer": None if final is None else _short(final, 120),
        "steps": len(top_actions),
        "seconds": trace.get("seconds"),
    }


def analyze_answer_row(row: dict) -> dict:
    """Tag a task that has no trace file, from its answers.jsonl row alone."""
    pseudo = {
        "task_id": row.get("task_id"),
        "steps": [],
        "final_answer": row.get("answer"),
        "error": row.get("error"),
        "seconds": row.get("seconds"),
    }
    item = analyze_trace(pseudo, answer_row=row)
    item["tags"] = ["no_trace"] + item["tags"]
    item["details"]["no_trace"] = ["only an answers.jsonl row"]
    item["evidence"] = ("no_trace: answers.jsonl row only; " + item["evidence"]).rstrip("; ")
    item["steps"] = row.get("steps")
    return item


# --------------------------------------------------------------------------- run level


def load_run_traces(run_id: str, runs_dir: Path) -> list[dict]:
    folder = Path(runs_dir) / "traces" / run_id
    traces = []
    if not folder.is_dir():
        return traces
    for p in sorted(folder.glob("*.json")):
        if p.name.startswith("."):
            continue
        try:
            traces.append(json.loads(p.read_text(encoding="utf-8")))
        except (OSError, json.JSONDecodeError) as e:
            print(f"warning: skipping unreadable trace {p.name}: {e}", file=sys.stderr)
    return traces


def load_answer_rows(run_id: str, runs_dir: Path) -> dict[str, dict]:
    """Latest answers.jsonl row per task for this run (rows without run_id are ignored)."""
    path = Path(runs_dir) / "answers.jsonl"
    rows: dict[str, dict] = {}
    if not path.is_file():
        return rows
    for line in path.read_text(encoding="utf-8").splitlines():
        try:
            r = json.loads(line)
        except json.JSONDecodeError:
            continue
        if r.get("run_id") == run_id and r.get("task_id"):
            rows[r["task_id"]] = r
    return rows


def latest_run_id(runs_dir: Path) -> str | None:
    folder = Path(runs_dir) / "traces"
    if not folder.is_dir():
        return None
    dirs = [d for d in folder.iterdir() if d.is_dir() and any(d.glob("*.json"))]
    return max(dirs, key=lambda d: d.stat().st_mtime).name if dirs else None


def analyze_run(run_id: str, runs_dir: Path | str = DEFAULT_RUNS_DIR, max_steps: int | None = None) -> dict:
    runs_dir = Path(runs_dir)
    traces = load_run_traces(run_id, runs_dir)
    rows = load_answer_rows(run_id, runs_dir)
    known: set[str] = set()
    for t in traces:
        known.update(k for k in ((t.get("totals") or {}).get("tool_calls_by_name") or {}) if k not in NON_TOOLS)
    items = []
    seen = set()
    for t in traces:
        tid = t.get("task_id")
        seen.add(tid)
        items.append(analyze_trace(t, max_steps=max_steps, answer_row=rows.get(tid), known_tools=known))
    for tid, row in rows.items():
        if tid not in seen:
            items.append(analyze_answer_row(row))
    counts: Counter = Counter()
    for it in items:
        counts.update(it["tags"])
    return {
        "run_id": run_id,
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "n_tasks": len(items),
        "max_steps": max_steps,
        "items": items,
        "counts": dict(sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))),
        "clean_tasks": [it["task_id"] for it in items if not [t for t in it["tags"] if t != "vision_used"]],
        "tag_descriptions": TAG_DESCRIPTIONS,
        "note": "Heuristic behaviour tags from traces; no gold answers are used.",
    }


def write_report(report: dict, out: Path) -> Path:
    out.parent.mkdir(parents=True, exist_ok=True)
    tmp = out.with_name(f".{out.name}.tmp")
    tmp.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    os.replace(tmp, out)
    return out


_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-", re.I)


def _short_id(tid: Any) -> str:
    s = str(tid)
    return s[:8] if _UUID_RE.match(s) else s[:24]


def format_table(report: dict) -> str:
    n = report["n_tasks"] or 1
    lines = [f"Error analysis: run {report['run_id']} ({report['n_tasks']} tasks)", ""]
    header = f"{'tag':38s} {'n':>3s} {'%':>5s}  examples"
    lines += [header, "-" * len(header)]
    examples: dict[str, list[str]] = {}
    for it in report["items"]:
        for t in it["tags"]:
            examples.setdefault(t, []).append(_short_id(it["task_id"]))
    for tag, c in report["counts"].items():
        ex = ", ".join(examples.get(tag, [])[:3]) + (" …" if len(examples.get(tag, [])) > 3 else "")
        lines.append(f"{tag:38s} {c:3d} {100 * c / n:4.0f}%  {ex}")
    if not report["counts"]:
        lines.append("(no failure tags)")
    lines += ["", f"clean (no failure tag): {len(report['clean_tasks'])}/{report['n_tasks']}", ""]
    lines.append(f"{'task':24s} {'steps':>5s}  tags")
    for it in report["items"]:
        tags = ", ".join(it["tags"]) or "-"
        lines.append(f"{_short_id(it['task_id']):24s} {str(it.get('steps') or '-'):>5s}  {tags}")
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Tag failure modes in a run's traces (no gold needed).")
    ap.add_argument("--run-id", default=os.environ.get("GAIA_RUN_ID"))
    ap.add_argument("--runs-dir", default=str(DEFAULT_RUNS_DIR))
    ap.add_argument("--max-steps", type=int, default=None, help="agent max_steps, to infer hit_max_steps")
    ap.add_argument("--out", default=None, help="output path (default <runs-dir>/error_analysis.json)")
    ap.add_argument("--json", action="store_true", help="print the JSON instead of the table")
    args = ap.parse_args(argv)
    runs_dir = Path(args.runs_dir)
    run_id = args.run_id or latest_run_id(runs_dir)
    if not run_id:
        print(f"No traces found under {runs_dir / 'traces'}; pass --run-id.", file=sys.stderr)
        return 2
    report = analyze_run(run_id, runs_dir, args.max_steps)
    if not report["items"]:
        print(f"No traces or answers for run {run_id!r} under {runs_dir}.", file=sys.stderr)
        return 2
    out = write_report(report, Path(args.out) if args.out else runs_dir / "error_analysis.json")
    print(json.dumps(report, indent=2) if args.json else format_table(report))
    print(f"\nwrote {out}")
    return 0


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