Agent Eval Harness skill for Claude Code

#!/usr/bin/env python3
"""Run a labeled eval set against headless Claude Code and score it.

Cases file (JSON list or JSONL), one object per case:
  {"id": "triggers-on-plain-request",
   "prompt": "Write a commit message for: renamed getUser to fetchUser",
   "tags": ["trigger"],
   "fixture": "fixtures/small-repo",          # optional dir copied into a temp cwd
   "max_turns": 8,                              # optional, default 10
   "graders": [
     {"type": "regex", "pattern": "^(feat|fix|refactor)(\\(.+\\))?: ", "flags": "m"},
     {"type": "not_contains", "text": "As an AI"},
     {"type": "tool_used", "name": "Skill", "input_regex": "commit-craft"},
     {"type": "tool_not_used", "name": "Bash", "input_regex": "git push"},
     {"type": "file_contains", "path": "CHANGELOG.md", "pattern": "fetchUser"},
     {"type": "max_cost_usd", "value": 0.25},
     {"type": "max_turns", "value": 6}
   ]}

Grader types: contains, not_contains, regex, not_regex, tool_used, tool_not_used,
file_exists, file_contains, max_cost_usd, max_turns, no_error.

Usage:
  python3 run_evals.py cases.jsonl [--runs 3] [--workers 2] [--out evals/results]
        [--baseline evals/results/<old>/summary.json] [--max-drop 0.10]
        [--tag trigger] [--case <id>] [--claude-args "--model sonnet"]
        [--cmd "python3 fake_agent.py"]   # replace the agent command (testing)

The default agent command is:
  claude -p --output-format stream-json --verbose --no-session-persistence
         --max-turns <max_turns> <claude-args>
with the prompt on stdin. Exit 0 when every case meets --threshold (default 1.0) and
no regression beyond --max-drop vs the baseline; 1 otherwise; 2 on usage errors.
"""
import argparse
import json
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


def load_cases(path: Path):
    text = path.read_text(encoding="utf-8").strip()
    if text.startswith("["):
        return json.loads(text)
    return [json.loads(line) for line in text.splitlines() if line.strip() and not line.startswith("#")]


def parse_output(stdout: str):
    """Accept stream-json lines, a single json result object, or plain text."""
    out = {"result": "", "tools": [], "cost": None, "turns": None, "is_error": False}
    events = []
    for line in stdout.splitlines():
        line = line.strip()
        if not line.startswith("{"):
            continue
        try:
            events.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    if not events:
        out["result"] = stdout
        return out
    for ev in events:
        if ev.get("type") == "assistant":
            for block in (ev.get("message") or {}).get("content") or []:
                if isinstance(block, dict) and block.get("type") == "tool_use":
                    out["tools"].append({"name": block.get("name", ""), "input": json.dumps(block.get("input", {}))})
        if ev.get("type") == "result":
            out["result"] = ev.get("result") or ""
            out["cost"] = ev.get("total_cost_usd")
            out["turns"] = ev.get("num_turns")
            out["is_error"] = bool(ev.get("is_error")) or ev.get("subtype", "success") != "success"
    return out


def grade(g, run, cwd: Path):
    t, text = g["type"], run["result"]
    flags = 0
    for ch in g.get("flags", ""):
        flags |= {"i": re.I, "m": re.M, "s": re.S}.get(ch, 0)

    def tool_hit():
        return any(
            tool["name"] == g["name"] and re.search(g.get("input_regex", ""), tool["input"])
            for tool in run["tools"]
        )

    if t == "contains":
        return g["text"] in text
    if t == "not_contains":
        return g["text"] not in text
    if t == "regex":
        return re.search(g["pattern"], text, flags) is not None
    if t == "not_regex":
        return re.search(g["pattern"], text, flags) is None
    if t == "tool_used":
        return tool_hit()
    if t == "tool_not_used":
        return not tool_hit()
    if t == "file_exists":
        return (cwd / g["path"]).exists()
    if t == "file_contains":
        p = cwd / g["path"]
        return p.is_file() and re.search(g["pattern"], p.read_text(encoding="utf-8", errors="replace"), flags) is not None
    if t == "max_cost_usd":
        return run["cost"] is not None and run["cost"] <= g["value"]
    if t == "max_turns":
        return run["turns"] is not None and run["turns"] <= g["value"]
    if t == "no_error":
        return not run["is_error"]
    raise ValueError(f"unknown grader type: {t}")


def run_once(case, args, run_idx: int, base_dir: Path):
    tmp = Path(tempfile.mkdtemp(prefix=f"eval-{case['id']}-"))
    try:
        if case.get("fixture"):
            shutil.copytree(base_dir / case["fixture"], tmp, dirs_exist_ok=True)
        if args.cmd:
            cmd = shlex.split(args.cmd)
        else:
            cmd = ["claude", "-p", "--output-format", "stream-json", "--verbose",
                   "--no-session-persistence", "--max-turns", str(case.get("max_turns", 10))]
            cmd += shlex.split(args.claude_args)
        start = time.time()
        try:
            proc = subprocess.run(cmd, input=case["prompt"], capture_output=True, text=True,
                                  cwd=tmp, timeout=args.timeout)
            stdout, rc = proc.stdout, proc.returncode
            err = proc.stderr[-500:] if proc.returncode else ""
        except subprocess.TimeoutExpired:
            stdout, rc, err = "", -1, f"timeout after {args.timeout}s"
        run = parse_output(stdout)
        run["is_error"] = run["is_error"] or rc != 0
        results = []
        for g in case["graders"]:
            try:
                ok = bool(grade(g, run, tmp))
            except Exception as exc:  # grader bug counts as failure, reported
                ok, err = False, f"grader error: {exc}"
            results.append({"grader": g["type"], "detail": g.get("pattern") or g.get("text") or g.get("name") or g.get("value"), "pass": ok})
        weights = [g.get("weight", 1) for g in case["graders"]]
        score = sum(w for w, r in zip(weights, results) if r["pass"]) / max(sum(weights), 1)
        return {"case": case["id"], "run": run_idx, "score": score, "graders": results,
                "cost": run["cost"], "turns": run["turns"], "seconds": round(time.time() - start, 1),
                "error": err, "result_head": run["result"][:400], "tools": [t["name"] for t in run["tools"]]}
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("cases", type=Path)
    ap.add_argument("--runs", type=int, default=3)
    ap.add_argument("--workers", type=int, default=2)
    ap.add_argument("--threshold", type=float, default=1.0)
    ap.add_argument("--timeout", type=int, default=600)
    ap.add_argument("--out", type=Path, default=Path("evals/results"))
    ap.add_argument("--baseline", type=Path)
    ap.add_argument("--max-drop", type=float, default=0.10)
    ap.add_argument("--tag")
    ap.add_argument("--case")
    ap.add_argument("--claude-args", default="")
    ap.add_argument("--cmd", help="override agent command (prompt is sent on stdin)")
    a = ap.parse_args()
    if not a.cases.is_file():
        print(f"error: {a.cases} not found", file=sys.stderr)
        return 2
    cases = load_cases(a.cases)
    if a.tag:
        cases = [c for c in cases if a.tag in c.get("tags", [])]
    if a.case:
        cases = [c for c in cases if c["id"] == a.case]
    if not cases:
        print("error: no cases selected", file=sys.stderr)
        return 2
    ids = [c["id"] for c in cases]
    if len(ids) != len(set(ids)):
        print("error: duplicate case ids", file=sys.stderr)
        return 2

    jobs = [(c, r) for c in cases for r in range(a.runs)]
    with ThreadPoolExecutor(max_workers=a.workers) as pool:
        runs = list(pool.map(lambda j: run_once(j[0], a, j[1], a.cases.parent), jobs))

    stamp = time.strftime("%Y%m%dT%H%M%S") + f"-{int(time.time() * 1000) % 1000:03d}"
    out_dir = a.out / stamp
    out_dir.mkdir(parents=True, exist_ok=True)
    with open(out_dir / "runs.jsonl", "w", encoding="utf-8") as fh:
        for r in runs:
            fh.write(json.dumps(r) + "\n")

    summary = {"created": stamp, "runs_per_case": a.runs, "cases": {}}
    for c in cases:
        rs = [r for r in runs if r["case"] == c["id"]]
        costs = [r["cost"] for r in rs if r["cost"] is not None]
        summary["cases"][c["id"]] = {
            "score": round(sum(r["score"] for r in rs) / len(rs), 3),
            "pass_runs": sum(1 for r in rs if r["score"] >= 1.0),
            "mean_cost": round(sum(costs) / len(costs), 4) if costs else None,
            "failing_graders": sorted({f"{g['grader']}:{g['detail']}" for r in rs for g in r["graders"] if not g["pass"]}),
            "errors": sorted({r["error"] for r in rs if r["error"]}),
        }
    scores = [v["score"] for v in summary["cases"].values()]
    summary["mean_score"] = round(sum(scores) / len(scores), 3)
    (out_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")

    base = json.loads(a.baseline.read_text(encoding="utf-8"))["cases"] if a.baseline else {}
    failed = False
    print(f"{'CASE':40} {'SCORE':>6} {'PASS':>6} {'BASE':>6} {'DELTA':>7}  NOTES")
    for cid, v in summary["cases"].items():
        b = base.get(cid, {}).get("score")
        delta = None if b is None else round(v["score"] - b, 3)
        notes = []
        if v["score"] < a.threshold:
            failed = True
            notes.append("below threshold")
        if delta is not None and delta < -a.max_drop:
            failed = True
            notes.append("REGRESSION")
        if v["failing_graders"]:
            notes.append("fails: " + "; ".join(v["failing_graders"])[:80])
        if v["errors"]:
            notes.append("error: " + v["errors"][0][:60])
        print(f"{cid[:40]:40} {v['score']:6.2f} {v['pass_runs']:>3}/{a.runs:<2} "
              f"{'' if b is None else f'{b:.2f}':>6} {'' if delta is None else f'{delta:+.2f}':>7}  {' | '.join(notes)}")
    print(f"\nmean score {summary['mean_score']:.3f} over {len(cases)} case(s) x {a.runs} run(s)")
    print(f"results: {out_dir}/summary.json")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())