Skills › BMAD Method › bmad-eval › run_evals.py
run_evals.py
363 lines · skills/bmad-eval/scripts/run_evals.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# ///
5"""Run eval cases through the recorded harness.
7A case is `input + rubric + optional state_prefix + optional files`. For each case and
8config the runner makes a clean room outside the project, copies the skill under test into
9the folder the harness reads skills from, stages the fixtures, runs the harness command once
10from the workspace, and records what it printed with timing and token usage. Grading happens
11elsewhere, from the transcript and artifacts left behind.
13Modes (--mode) decide which configs each case runs under:
15 quality : one config, "skill", the skill staged in the workspace.
16 baseline : "skill" and "bare" (nothing staged), same input, so the bare-model floor is
17 measured under identical conditions.
18 variant : "skill" (--skill-path) and "variant" (--variant-path).
20Run layout: <run-dir>/<config>/<case-id>/ (plus /run-N/ when --runs > 1), each holding
21prompt.txt, transcript.jsonl, stderr.txt, timing.json and cwd/ (the workspace after the run),
22so `aggregate_benchmark.py --baseline <run-dir>/bare --variant <run-dir>/skill` compares
23configs from the timing.json files.
25Harness: `[workflow.harness]` in this skill's customization, read through the project's
26resolver; `--harness <json>` with the same keys for a project without BMad. The keys and the
27clean room are described in eval_common.py. Without a harness the runner stages every case,
28records each as skipped and exits 3.
30Usage:
31 uv run run_evals.py --cases CASES.json --skill-path SKILL_DIR --output-dir DIR
32 [--mode quality|baseline|variant] [--variant-path SKILL_DIR] [--project-root DIR]
33 [--harness HARNESS.json] [--case-ids A1,B3] [--runs N] [--timeout SECS]
34 [--workers N] [--label NAME] [--quiet]
36CASES.json is a list of cases or {"cases": [...]}. Each case:
37 {"id": "...", "input": "...", "rubric": [...], "state_prefix": "..."?, "files": ["..."]?}
39Exit 0 when every run completed, 1 when any run errored or timed out or the command was
40not found, 2 on a usage error, 3 when no harness is recorded.
41"""
43from __future__ import annotations
45import argparse
46import json
47import shutil
48import sys
49from concurrent.futures import ThreadPoolExecutor, as_completed
50from pathlib import Path
52from eval_common import (
53 DEFAULT_SKILL_DIR,
54 account_transcript,
55 contained,
56 find_project_root,
57 make_run_dir,
58 read_json,
59 resolve_harness,
60 run_in_clean_room,
61 safe_name,
62 utc_now_iso,
63 write_json,
64)
66FAILURE_STATUSES = ("error", "timeout", "exception", "harness-missing")
68# --- staging: skill under test + fixtures ------------------------------------
71def stage_skill(skill_path: Path, cwd: Path, skills_subdir: str) -> Path:
72 """Copy the skill where the harness discovers skills inside the workspace.
74 A copy, not a symlink: a run that edits its staged skill must not touch the source or
75 the other cases.
76 """
77 dest = cwd / skills_subdir / skill_path.name
78 shutil.copytree(skill_path, dest, dirs_exist_ok=True, ignore=shutil.ignore_patterns("__pycache__", ".git"))
79 return dest
82def resolve_fixtures(files: list, project_root: Path, cases_dir: Path) -> list[tuple[Path, str]]:
83 """Map each `files` entry to (source, dest-relative-path).
85 The entry's own relative path is preserved inside the workspace, so a bare filename lands
86 at its root and a nested path keeps its directory structure, matching the path the case
87 input references.
88 """
89 out: list[tuple[Path, str]] = []
90 for entry in files or []:
91 entry = str(entry)
92 for candidate in (
93 (project_root / entry).resolve(),
94 (cases_dir / entry).resolve(),
95 Path(entry).resolve(),
96 ):
97 if candidate.is_file():
98 out.append((candidate, entry))
99 break
100 else:
101 print(f"Warning: fixture not found: {entry}", file=sys.stderr)
102 return out
105def stage_fixtures(fixtures: list[tuple[Path, str]], cwd: Path) -> None:
106 for src, dest_rel in fixtures:
107 dest = contained(cwd, dest_rel)
108 dest.parent.mkdir(parents=True, exist_ok=True)
109 shutil.copy2(src, dest)
112def compose_prompt(case: dict) -> str:
113 """Prepend the state_prefix, a bracketed prime that places the skill mid-workflow."""
114 input_text = str(case.get("input", ""))
115 prefix = case.get("state_prefix")
116 if prefix:
117 return f"{str(prefix).rstrip()}\n\n{input_text}"
118 return input_text
121# --- per-case execution -----------------------------------------------------
124def run_case(
125 case: dict,
126 case_dir: Path,
127 run_dir: Path,
128 harness: dict | None,
129 timeout: int,
130 config: str,
131 skill_path: Path | None,
132 fixtures: list[tuple[Path, str]],
133) -> dict:
134 case_id = str(case.get("id", "unnamed"))
135 case_dir.mkdir(parents=True, exist_ok=True)
136 prompt = compose_prompt(case)
137 write_json(case_dir / "case.json", case)
139 def timing(status: str, **extra: object) -> None:
140 write_json(case_dir / "timing.json", {"case_id": case_id, "config": config, "status": status, **extra})
142 def result(status: str, **extra: object) -> dict:
143 return {
144 "case_id": case_id,
145 "config": config,
146 "status": status,
147 "case_dir": str(case_dir.relative_to(run_dir)),
148 **extra,
149 }
151 def stage(cwd: Path) -> None:
152 stage_fixtures(fixtures, cwd)
153 if skill_path is not None:
154 stage_skill(skill_path, cwd, (harness or {}).get("skill_dir", DEFAULT_SKILL_DIR))
156 if harness is None:
157 (case_dir / "prompt.txt").write_text(prompt, encoding="utf-8")
158 try:
159 stage(case_dir / "cwd")
160 except ValueError as e:
161 timing("error", reason=str(e), captured_at=utc_now_iso())
162 return result("error", reason=str(e))
163 timing("skipped", captured_at=utc_now_iso())
164 return result("skipped", reason="no harness recorded", prompt_chars=len(prompt))
166 try:
167 run = run_in_clean_room(harness, case_dir, prompt, timeout, stage)
168 except ValueError as e:
169 timing("error", reason=str(e), captured_at=utc_now_iso())
170 return result("error", reason=str(e))
172 accounting = account_transcript(run["stdout"])
173 timing(
174 run["status"],
175 elapsed_s=run["elapsed_s"],
176 return_code=run["return_code"],
177 input_tokens=accounting["input_tokens"],
178 output_tokens=accounting["output_tokens"],
179 total_tokens=accounting["total_tokens"],
180 tokens_reported=accounting["tokens_reported"],
181 total_steps=accounting["total_steps"],
182 total_tool_calls=accounting["total_tool_calls"],
183 captured_at=utc_now_iso(),
184 )
185 return result(
186 run["status"],
187 elapsed_s=run["elapsed_s"],
188 return_code=run["return_code"],
189 tokens=accounting["total_tokens"],
190 tool_calls=accounting["tool_calls"],
191 error_tail=run["stderr"][-2000:],
192 )
195# --- main -------------------------------------------------------------------
198def load_cases(cases_file: Path) -> list[dict]:
199 data = read_json(cases_file)
200 if isinstance(data, dict) and "cases" in data:
201 cases = data["cases"]
202 elif isinstance(data, list):
203 cases = data
204 else:
205 raise ValueError("cases file must be a list or {'cases': [...]}")
206 if not isinstance(cases, list):
207 raise ValueError("'cases' must be a list")
208 return cases
211def main(argv: list[str] | None = None) -> int:
212 p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
213 p.add_argument("--cases", required=True, type=Path)
214 p.add_argument("--skill-path", required=True, type=Path, help="the skill under test (holds SKILL.md)")
215 p.add_argument("--output-dir", required=True, type=Path)
216 p.add_argument("--mode", choices=("quality", "baseline", "variant"), default="quality")
217 p.add_argument("--variant-path", type=Path, default=None, help="variant mode: the stripped or prior skill")
218 p.add_argument(
219 "--project-root",
220 type=Path,
221 default=None,
222 help="holds _bmad/ and anchors fixture paths; found from the skill path when omitted",
223 )
224 p.add_argument("--harness", type=Path, default=None, help="harness JSON for a project without BMad")
225 p.add_argument("--case-ids", default=None, help="comma-separated subset of case ids to run")
226 p.add_argument("--runs", type=int, default=1, help="repeats per case per config for the variance benchmark")
227 p.add_argument("--timeout", type=int, default=600)
228 p.add_argument("--workers", type=int, default=4)
229 p.add_argument("--label", default="evals", help="label for the run id")
230 p.add_argument("--quiet", action="store_true")
231 args = p.parse_args(argv)
233 cases_file = args.cases.resolve()
234 if not cases_file.is_file():
235 print(f"cases file not found: {cases_file}", file=sys.stderr)
236 return 2
238 skill_path = args.skill_path.resolve()
239 if not (skill_path / "SKILL.md").is_file():
240 print(f"skill path has no SKILL.md: {skill_path}", file=sys.stderr)
241 return 2
243 if args.mode == "variant":
244 if args.variant_path is None:
245 print("--mode variant requires --variant-path", file=sys.stderr)
246 return 2
247 variant_path = args.variant_path.resolve()
248 if not (variant_path / "SKILL.md").is_file():
249 print(f"variant path has no SKILL.md: {variant_path}", file=sys.stderr)
250 return 2
251 else:
252 variant_path = None
254 project_root = args.project_root.resolve() if args.project_root else find_project_root(skill_path)
255 fixture_root = project_root or cases_file.parent
257 if args.mode == "baseline":
258 configs: list[tuple[str, Path | None]] = [("skill", skill_path), ("bare", None)]
259 elif args.mode == "variant":
260 configs = [("skill", skill_path), ("variant", variant_path)]
261 else:
262 configs = [("skill", skill_path)]
264 cases = load_cases(cases_file)
265 if args.case_ids:
266 wanted = {x.strip() for x in args.case_ids.split(",") if x.strip()}
267 cases = [c for c in cases if str(c.get("id")) in wanted]
269 try:
270 harness, harness_note = resolve_harness(project_root, args.harness)
271 except (ValueError, json.JSONDecodeError) as e:
272 print(f"harness invalid: {e}", file=sys.stderr)
273 return 2
275 run_id, run_dir = make_run_dir(args.output_dir, safe_name(args.label))
276 write_json(
277 run_dir / "run.json",
278 {
279 "run_id": run_id,
280 "cases_file": str(cases_file),
281 "skill_path": str(skill_path),
282 "variant_path": str(variant_path) if variant_path else None,
283 "mode": args.mode,
284 "configs": [name for name, _ in configs],
285 "runs_per_case": args.runs,
286 "harness": harness_note,
287 "command": (harness or {}).get("command"),
288 "started_at": utc_now_iso(),
289 "case_count": len(cases),
290 },
291 )
293 if not args.quiet:
294 if harness is None:
295 print(f"[run_evals] no harness ({harness_note}); staging cases only", file=sys.stderr)
296 else:
297 print(f"[run_evals] harness: {' '.join(harness['command'])}", file=sys.stderr)
298 print(
299 f"[run_evals] {len(cases)} cases x {len(configs)} configs x {args.runs} runs, "
300 f"mode={args.mode}, run_dir={run_dir}",
301 file=sys.stderr,
302 )
304 jobs: list[tuple[str, dict, Path, Path | None]] = []
305 for config_name, config_skill in configs:
306 for c in cases:
307 base = run_dir / config_name / safe_name(c.get("id", "unnamed"))
308 for i in range(max(1, args.runs)):
309 case_dir = base / f"run-{i + 1}" if args.runs > 1 else base
310 jobs.append((config_name, c, case_dir, config_skill))
312 results: list[dict] = []
313 with ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
314 fut_to_case = {
315 pool.submit(
316 run_case,
317 c,
318 case_dir,
319 run_dir,
320 harness,
321 int(c.get("timeout", args.timeout)),
322 config_name,
323 config_skill,
324 resolve_fixtures(c.get("files", []), fixture_root, cases_file.parent),
325 ): c
326 for config_name, c, case_dir, config_skill in jobs
327 }
328 for fut in as_completed(fut_to_case):
329 c = fut_to_case[fut]
330 try:
331 res = fut.result()
332 except Exception as e:
333 res = {"case_id": str(c.get("id")), "status": "exception", "reason": str(e)}
334 results.append(res)
335 if not args.quiet:
336 print(
337 f" [{res.get('status')}] {res.get('config', '?')}/{res.get('case_id')} "
338 f"({res.get('elapsed_s', 0)}s)",
339 file=sys.stderr,
340 )
342 failures = sum(1 for r in results if r.get("status") in FAILURE_STATUSES)
343 summary = {
344 "run_id": run_id,
345 "completed_at": utc_now_iso(),
346 "mode": args.mode,
347 "harness": harness_note,
348 "total": len(jobs),
349 "executed": sum(1 for r in results if r.get("status") == "ok"),
350 "skipped": sum(1 for r in results if r.get("status") == "skipped"),
351 "failures": failures,
352 "run_dir": str(run_dir),
353 "results": results,
354 }
355 write_json(run_dir / "execution-summary.json", summary)
356 print(json.dumps(summary, indent=2))
357 if harness is None:
358 return 3
359 return 1 if failures else 0
362if __name__ == "__main__":
363 sys.exit(main())