Skills › BMAD Method › bmad-eval › eval_common.py
eval_common.py
319 lines · skills/bmad-eval/scripts/eval_common.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# ///
5"""What run_evals.py and run_triggers.py share: the harness, the clean room, containment, run folders.
7The harness is the agent CLI the evals run through, described by four keys the model that first
8runs an eval records in this skill's customization (`[workflow.harness]`):
10 command argv for one non-interactive run; "{prompt}" is the input, "{cwd}" the workspace.
11 skill_dir the folder under the workspace the CLI reads skills from.
12 env env vars for the run: a value is set as given, with "~" meaning the fresh HOME;
13 an empty value forwards the host's value when it has one.
14 home_files files or folders under the host home brought into the fresh HOME at the same
15 path, so a CLI logged in through a file or keychain stays logged in.
17The runners read them through the project's customization resolver, or from `--harness <json>`
18for a project without BMad. A run happens in a clean room: a temporary folder outside the
19project holding the workspace and the fresh HOME, so the project's instruction files and
20installed skills are not ancestors of the run. The workspace is copied back afterwards.
21"""
23from __future__ import annotations
25import json
26import os
27import re
28import shutil
29import subprocess
30import sys
31import tempfile
32import time
33from collections.abc import Callable, Mapping
34from datetime import UTC, datetime
35from pathlib import Path
37SKILL_ROOT = Path(__file__).resolve().parents[1]
38HARNESS_KEY = "workflow.harness"
39DEFAULT_SKILL_DIR = ".agents/skills"
40SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]+")
41# A subprocess on Windows cannot start without these; nothing else crosses uninvited.
42PLATFORM_ENV = ("SYSTEMROOT", "COMSPEC", "PATHEXT", "TEMP", "TMP") if sys.platform == "win32" else ()
45def utc_now_iso() -> str:
46 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
49def write_json(path: Path, data: object) -> None:
50 path.parent.mkdir(parents=True, exist_ok=True)
51 path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
54def read_json(path: Path) -> object:
55 return json.loads(path.read_text(encoding="utf-8"))
58def find_project_root(start: Path) -> Path | None:
59 """Nearest ancestor holding `_bmad/`, else the nearest holding `.git`."""
60 git_root = None
61 current = start.resolve()
62 while True:
63 if (current / "_bmad").is_dir():
64 return current
65 if git_root is None and (current / ".git").exists():
66 git_root = current
67 if current.parent == current:
68 return git_root
69 current = current.parent
72# --- harness -----------------------------------------------------------------
75def validate_harness(harness: object) -> dict:
76 if not isinstance(harness, dict):
77 raise ValueError("harness must be a table")
78 command = harness.get("command")
79 if not isinstance(command, list) or not command or not all(isinstance(t, str) for t in command):
80 raise ValueError("harness.command must be a non-empty list of strings")
81 if not any("{prompt}" in t for t in command):
82 raise ValueError("harness.command needs a {prompt} token")
83 if not isinstance(harness.get("skill_dir", DEFAULT_SKILL_DIR), str):
84 raise ValueError("harness.skill_dir must be a string")
85 env = harness.get("env", {})
86 if not isinstance(env, dict) or not all(isinstance(v, str) for v in env.values()):
87 raise ValueError('harness.env is a table of var = "value" ("" forwards the host value, "~" is the fresh HOME)')
88 if not isinstance(harness.get("home_files", []), list):
89 raise ValueError("harness.home_files must be a list")
90 return harness
93def resolve_harness(project_root: Path | None, explicit: Path | None) -> tuple[dict | None, str]:
94 """The harness table and where it came from; (None, why) when there is none.
96 The runner never reads the override files itself: the project's resolver merges the layers.
97 """
98 if explicit is not None:
99 if not explicit.is_file():
100 return None, f"harness file not found: {explicit}"
101 return validate_harness(read_json(explicit)), str(explicit)
102 if project_root is None:
103 return None, "no project root"
104 resolver = project_root / "_bmad" / "scripts" / "resolve_customization.py"
105 if not resolver.is_file():
106 return None, "BMad is not set up in this project; pass --harness"
107 argv = [sys.executable, str(resolver), "--skill", str(SKILL_ROOT), "--project-root", str(project_root)]
108 proc = subprocess.run([*argv, "--key", HARNESS_KEY], capture_output=True, text=True)
109 if proc.returncode != 0:
110 return None, f"customization resolver failed: {proc.stderr.strip()[-300:]}"
111 harness = json.loads(proc.stdout or "{}").get(HARNESS_KEY)
112 if not isinstance(harness, dict) or not harness.get("command"):
113 return None, "no harness recorded in bmad-eval's customization"
114 return validate_harness(harness), f"customization {HARNESS_KEY}"
117def build_argv(harness: Mapping, prompt: str, cwd: str) -> list[str]:
118 """The command with its placeholders filled."""
119 return [
120 str(t).replace("{prompt}", prompt).replace("{query}", prompt).replace("{cwd}", cwd) for t in harness["command"]
121 ]
124def build_case_env(harness: Mapping | None, home_dir: Path, host_env: Mapping[str, str]) -> dict[str, str]:
125 """The subprocess environment, built from scratch, never from os.environ.
127 PATH, the fresh HOME (and USERPROFILE on Windows), then the harness's `env`: a value is set
128 as given with "~" expanded to the fresh HOME; an empty value forwards the host's value when
129 the host has one set non-empty, and is left out otherwise, since an empty credential would
130 override the CLI's own login.
131 """
132 env = {"PATH": host_env.get("PATH", ""), "HOME": str(home_dir)}
133 if sys.platform == "win32":
134 env["USERPROFILE"] = str(home_dir)
135 for name in PLATFORM_ENV:
136 if name in host_env:
137 env[name] = host_env[name]
138 for name, value in ((harness or {}).get("env") or {}).items():
139 if value == "":
140 if host_env.get(str(name)):
141 env[str(name)] = host_env[str(name)]
142 else:
143 env[str(name)] = expand_home(str(value), home_dir)
144 return env
147def expand_home(value: str, home_dir: Path) -> str:
148 if value == "~":
149 return str(home_dir)
150 if value.startswith("~/"):
151 return str(home_dir / value[2:])
152 return value
155def make_home(harness: Mapping | None, room: Path) -> Path:
156 """The fresh HOME for one run: folders named by `env` values made, `home_files` brought over.
158 `home_files` are paths relative to the host home (`.codex/auth.json`, `Library/Keychains`),
159 placed at the same path inside the fresh HOME: a file is copied, a folder is linked.
160 """
161 harness = harness or {}
162 home = room / ".home"
163 home.mkdir(parents=True, exist_ok=True)
164 for value in (harness.get("env") or {}).values():
165 if str(value).startswith("~/"):
166 contained(home, str(value)[2:]).mkdir(parents=True, exist_ok=True)
167 host_home = Path.home()
168 for entry in harness.get("home_files") or []:
169 rel = str(entry).removeprefix("~/").removeprefix("~")
170 src = host_home / rel
171 dest = contained(home, rel)
172 if dest.exists() or dest.is_symlink():
173 continue
174 dest.parent.mkdir(parents=True, exist_ok=True)
175 if src.is_dir():
176 dest.symlink_to(src, target_is_directory=True)
177 elif src.is_file():
178 shutil.copy2(src, dest)
179 else:
180 print(f"Warning: home_files entry not found on this host: {src}", file=sys.stderr)
181 return home
184# --- the clean room ----------------------------------------------------------
187def run_in_clean_room(
188 harness: Mapping, case_dir: Path, prompt: str, timeout: int, stage: Callable[[Path], None]
189) -> dict:
190 """One harness run in a temporary room outside the project; the workspace comes back to case_dir/cwd.
192 `stage(cwd)` fills the workspace before the run. The room holds the workspace, made a git
193 root so harnesses that look for a project boundary stop there, and the fresh HOME beside it.
194 Leaves prompt.txt, transcript.jsonl (stdout), stderr.txt and cwd/ in case_dir. Returns
195 status ("ok", "error", "timeout", "harness-missing"), return_code, elapsed_s, stdout, stderr.
196 """
197 case_dir.mkdir(parents=True, exist_ok=True)
198 (case_dir / "prompt.txt").write_text(prompt, encoding="utf-8")
199 with tempfile.TemporaryDirectory(prefix="bmad-eval-", ignore_cleanup_errors=True) as tmp:
200 room = Path(tmp)
201 cwd = room / "cwd"
202 cwd.mkdir()
203 stage(cwd)
204 # Drop GIT_DIR and the rest: a git hook exports them, and git init would rewrite that repository's config.
205 git_env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")}
206 subprocess.run(["git", "init", "-q", str(cwd)], capture_output=True, env=git_env)
207 env = build_case_env(harness, make_home(harness, room), os.environ)
208 argv = build_argv(harness, prompt, str(cwd))
209 status, return_code, stdout, stderr = "ok", 0, b"", b""
210 start = time.time()
211 try:
212 proc = subprocess.run(
213 argv, capture_output=True, stdin=subprocess.DEVNULL, cwd=str(cwd), env=env, timeout=timeout
214 )
215 stdout, stderr, return_code = proc.stdout or b"", proc.stderr or b"", proc.returncode
216 if return_code != 0:
217 status = "error"
218 except FileNotFoundError as e:
219 status, return_code, stderr = "harness-missing", -1, f"command not found: {e}".encode()
220 except subprocess.TimeoutExpired as e:
221 status, return_code = "timeout", -1
222 stdout, stderr = e.stdout or b"", (e.stderr or b"") + f"\nTIMEOUT after {timeout}s".encode()
223 elapsed = time.time() - start
224 shutil.copytree(cwd, case_dir / "cwd", symlinks=True, ignore=shutil.ignore_patterns(".git"), dirs_exist_ok=True)
225 stdout_text = stdout.decode("utf-8", errors="replace")
226 stderr_text = stderr.decode("utf-8", errors="replace")
227 (case_dir / "transcript.jsonl").write_text(stdout_text, encoding="utf-8")
228 (case_dir / "stderr.txt").write_text(stderr_text, encoding="utf-8")
229 return {
230 "status": status,
231 "return_code": return_code,
232 "elapsed_s": round(elapsed, 3),
233 "stdout": stdout_text,
234 "stderr": stderr_text,
235 }
238def account_transcript(transcript_text: str) -> dict:
239 """Best-effort usage from what the harness printed.
241 When the output is line-delimited JSON, per-message usage on assistant events is summed,
242 a turn-level usage block on any event overrides the sum, and tool calls are counted;
243 anything else degrades to zero counts with `tokens_reported` false, and elapsed time is
244 the metric.
245 """
246 input_tokens = 0
247 output_tokens = 0
248 total_steps = 0
249 tool_calls: dict[str, int] = {}
250 found_usage = False
251 for raw in transcript_text.splitlines():
252 raw = raw.strip()
253 if not raw:
254 continue
255 try:
256 evt = json.loads(raw)
257 except json.JSONDecodeError:
258 continue
259 if not isinstance(evt, dict):
260 continue
261 if evt.get("type") == "assistant":
262 total_steps += 1
263 msg = evt.get("message", {})
264 usage = msg.get("usage") if isinstance(msg, dict) else None
265 if isinstance(usage, dict):
266 found_usage = True
267 input_tokens += int(usage.get("input_tokens", 0) or 0)
268 output_tokens += int(usage.get("output_tokens", 0) or 0)
269 for item in msg.get("content", []) if isinstance(msg, dict) else []:
270 if isinstance(item, dict) and item.get("type") == "tool_use":
271 name = item.get("name", "?")
272 tool_calls[name] = tool_calls.get(name, 0) + 1
273 elif isinstance(evt.get("usage"), dict):
274 # A turn-level usage block is authoritative over the running sum.
275 usage = evt["usage"]
276 found_usage = True
277 input_tokens = int(usage.get("input_tokens", input_tokens) or input_tokens)
278 output_tokens = int(usage.get("output_tokens", output_tokens) or output_tokens)
279 return {
280 "input_tokens": input_tokens,
281 "output_tokens": output_tokens,
282 "total_tokens": input_tokens + output_tokens,
283 "tokens_reported": found_usage,
284 "total_steps": total_steps,
285 "tool_calls": tool_calls,
286 "total_tool_calls": sum(tool_calls.values()),
287 }
290# --- containment and run folders ----------------------------------------------
293def safe_name(value: object) -> str:
294 """A folder name from a case id: no separators, no `..`, never empty."""
295 name = SAFE_NAME_RE.sub("_", str(value)).strip(".")
296 return name or "unnamed"
299def contained(root: Path, rel: str) -> Path:
300 """`root / rel` when it stays inside root; ValueError when it would escape."""
301 root = root.resolve()
302 target = (root / rel).resolve()
303 if target != root and root not in target.parents:
304 raise ValueError(f"path escapes the workspace: {rel}")
305 return target
308def make_run_dir(output_dir: Path, label: str) -> tuple[str, Path]:
309 """A new run folder. A second run in the same second gets a suffix, never the same folder."""
310 stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
311 for n in range(1, 1000):
312 run_id = f"{stamp}-{label}" if n == 1 else f"{stamp}-{label}-{n}"
313 run_dir = (output_dir / run_id).resolve()
314 try:
315 run_dir.mkdir(parents=True, exist_ok=False)
316 except FileExistsError:
317 continue
318 return run_id, run_dir
319 raise RuntimeError(f"could not create a run folder under {output_dir}")