cmos/scripts/triage.py

"""Triage canary failures from the most recent loop run using an LLM judge.

Reads ``logs/runs.jsonl``, finds the latest run, identifies exemplars whose
``exact_match`` is False (canary failures), calls the judge on each, and
appends verdicts to ``logs/triage.jsonl``.

The judge output is ADVISORY — it never feeds into the loop scalar.
Verdicts are for a human to review ("is this a variant or a real
regression?") after the fact.

Usage:
    uv run python scripts/triage.py
    uv run python scripts/triage.py --run-index -1     # last run (default)
    uv run python scripts/triage.py --run-index -2     # second-last
    uv run python scripts/triage.py --dry-run          # no API calls

If there are no canary failures in the target run, the script prints a
note and exits 0 without calling the judge.
"""

from __future__ import annotations

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

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from harness.judge import JUDGE_MODEL, Verdict, judge  # noqa: E402
from harness.score import load_exemplars  # noqa: E402


def _dry_run_judge(expected: str, candidate: str, cmos_type: str) -> Verdict:
    return Verdict(label="unclear", reasoning="(dry run — judge not called)")


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--runs-log", default="logs/runs.jsonl")
    ap.add_argument("--triage-log", default="logs/triage.jsonl")
    ap.add_argument("--exemplars", default="exemplars")
    ap.add_argument("--run-index", type=int, default=-1, help="which run to triage")
    ap.add_argument("--dry-run", action="store_true", help="do not call the judge")
    args = ap.parse_args()

    runs_path = Path(args.runs_log)
    if not runs_path.exists():
        print(f"no runs log at {runs_path}", file=sys.stderr)
        return 2

    with runs_path.open() as fh:
        runs = [json.loads(line) for line in fh if line.strip()]
    if not runs:
        print("runs log is empty", file=sys.stderr)
        return 2
    try:
        run = runs[args.run_index]
    except IndexError:
        print(f"run index {args.run_index} out of range (have {len(runs)} runs)", file=sys.stderr)
        return 2

    # Map exemplar name → canonical expected_bibliography and type.
    ex_map = {ex.name: ex for ex in load_exemplars(Path(args.exemplars))}

    failures = [r for r in run["per_exemplar"] if r["canary"] and not r["exact_match"]]
    if not failures:
        print(f"no canary failures in run at {run['timestamp']} — nothing to triage")
        return 0

    judge_fn = _dry_run_judge if args.dry_run else judge

    print(f"triaging {len(failures)} canary failure(s) from run {run['timestamp']}")
    print(f"judge model: {'(dry run)' if args.dry_run else JUDGE_MODEL}")
    print()

    triage_path = Path(args.triage_log)
    triage_path.parent.mkdir(parents=True, exist_ok=True)
    with triage_path.open("a") as out:
        for r in failures:
            name = r["name"]
            ex = ex_map.get(name)
            if ex is None:
                print(f"  [{name}] exemplar not found on disk, skipping")
                continue
            expected = ex.expected_bibliography
            candidate = r["candidate"]
            verdict = judge_fn(expected, candidate, ex.type)

            entry = {
                "triage_timestamp": datetime.now(timezone.utc).isoformat(),
                "run_timestamp": run["timestamp"],
                "exemplar": name,
                "cmos_type": ex.type,
                "expected": expected,
                "candidate": candidate,
                "judge_label": verdict.label,
                "judge_reasoning": verdict.reasoning,
                "judge_model": None if args.dry_run else JUDGE_MODEL,
            }
            out.write(json.dumps(entry) + "\n")

            icon = {
                "regression": "!!",
                "variant": "ok",
                "unclear": "??",
            }.get(verdict.label, "??")
            print(f"  [{icon}] {name}: {verdict.label}")
            if verdict.reasoning:
                print(f"       {verdict.reasoning}")

    print()
    print(f"verdicts appended to {triage_path}")
    return 0


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