Skills › BMAD Method › bmad-eval › run_triggers.py
run_triggers.py
289 lines · skills/bmad-eval/scripts/run_triggers.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# ///
5"""Trigger evals: does a skill's description fire on each near-miss query?
7A trigger query is a should/should-not user message that shares keywords with the skill, so
8the description has to discriminate. For each query the runner stages a synthetic skill with
9the description under test where the harness reads skills, sends the query through the
10harness in a clean room, and checks whether the skill loaded. Each query runs several times
11(--runs-per-query) so the rate is stable, not a coin flip.
13Detection is a canary, the same on every harness. The synthetic skill's body asks the model
14to begin its reply with a token unique to that attempt; the token appearing in what the
15harness printed is the load. The skill's name and description never contain the token, so a
16harness that lists its discovered skills at startup cannot fake a hit.
18A query whose attempts did not all complete (command failed, timed out) is unmeasured, never
19passed: a should-not query with no completed attempts would otherwise pass at a rate of zero.
21Each attempt keeps prompt.txt, transcript.jsonl (what the harness printed), stderr.txt,
22timing.json and cwd/ (the workspace after the run, staged skill included) under
23<run-dir>/queries/qNNN-rN/. Harness and clean room are as in run_evals.py.
25Usage:
26 uv run run_triggers.py --skill-path SKILL_DIR --queries QUERIES.json --output-dir DIR
27 [--project-root DIR] [--harness HARNESS.json] [--runs-per-query N] [--threshold 0.5]
28 [--timeout SECS] [--workers N] [--quiet]
30QUERIES.json is a list of {"query": "...", "should_trigger": true|false}.
32Exit 0 when every query was measured, 1 when any attempt failed or the command was not
33found, 2 on a usage error, 3 when no harness is recorded. A query failing its threshold is a
34result, not an error.
35"""
37from __future__ import annotations
39import argparse
40import json
41import re
42import sys
43import uuid
44from concurrent.futures import ThreadPoolExecutor, as_completed
45from pathlib import Path
47from eval_common import (
48 DEFAULT_SKILL_DIR,
49 account_transcript,
50 find_project_root,
51 make_run_dir,
52 read_json,
53 resolve_harness,
54 run_in_clean_room,
55 utc_now_iso,
56 write_json,
57)
59CANARY_PREFIX = "TRIGGER-LOADED-"
62def unquote_yaml(value: str) -> str:
63 """A YAML scalar's text: single quotes doubled inside, double quotes with backslash escapes."""
64 value = value.strip()
65 if len(value) >= 2 and value[0] == value[-1] == "'":
66 return value[1:-1].replace("''", "'")
67 if len(value) >= 2 and value[0] == value[-1] == '"':
68 return value[1:-1].replace('\\"', '"').replace("\\n", "\n").replace("\\\\", "\\")
69 return value
72def parse_skill_md(skill_path: Path) -> tuple[str, str]:
73 """Return (name, description) from SKILL.md frontmatter, quotes and block indicators removed."""
74 text = (skill_path / "SKILL.md").read_text(encoding="utf-8")
75 m = re.match(r"^---\s*\n(.*?)\n---\s*\n", text, re.DOTALL)
76 if not m:
77 raise ValueError(f"SKILL.md at {skill_path} is missing frontmatter")
78 fields: dict[str, list[str]] = {}
79 current = None
80 for line in m.group(1).splitlines():
81 if line[:1] not in (" ", "\t") and ":" in line:
82 key, _, value = line.partition(":")
83 current = key.strip()
84 fields[current] = [] if value.strip() in ("|", ">", "|-", ">-") else [value.strip()]
85 elif current and line.strip():
86 fields[current].append(line.strip())
87 name = unquote_yaml(" ".join(fields.get("name", [])))
88 if not name:
89 raise ValueError(f"SKILL.md at {skill_path} has no name")
90 return name, unquote_yaml(" ".join(fields.get("description", [])))
93# --- synthetic skill and detection -------------------------------------------
96def write_synthetic_skill(skills_dir: Path, skill_name: str, description: str, token: str) -> str:
97 """Write a skill the harness can discover, carrying the description and the canary. Returns its name."""
98 clean_name = f"{skill_name}-trig-{token[len(CANARY_PREFIX) :]}"
99 root = skills_dir / clean_name
100 root.mkdir(parents=True, exist_ok=True)
101 indented = "\n ".join(description.split("\n"))
102 (root / "SKILL.md").write_text(
103 f"---\nname: {clean_name}\ndescription: |\n {indented}\n---\n\n"
104 f"# {skill_name}\n\nThis skill handles: {description}\n\n"
105 f"Begin your reply with the exact token `{token}`, then continue.\n",
106 encoding="utf-8",
107 )
108 return clean_name
111def detect_load(output: str, token: str) -> bool:
112 """Did the synthetic skill load? Its canary token in the output says so."""
113 return token in output
116# --- per-attempt execution ----------------------------------------------------
119def run_query_once(
120 query: str, skill_name: str, description: str, harness: dict, attempt_dir: Path, timeout: int
121) -> tuple[bool | None, str]:
122 """One attempt: (loaded, "") when it completed, (None, why) when it did not."""
123 token = CANARY_PREFIX + uuid.uuid4().hex[:8]
125 def stage(cwd: Path) -> None:
126 write_synthetic_skill(cwd / harness.get("skill_dir", DEFAULT_SKILL_DIR), skill_name, description, token)
128 run = run_in_clean_room(harness, attempt_dir, query, timeout, stage)
129 accounting = account_transcript(run["stdout"])
130 loaded = detect_load(run["stdout"], token) if run["status"] == "ok" else None
131 write_json(
132 attempt_dir / "timing.json",
133 {
134 "status": run["status"],
135 "elapsed_s": run["elapsed_s"],
136 "return_code": run["return_code"],
137 "loaded": loaded,
138 "total_tokens": accounting["total_tokens"],
139 "tokens_reported": accounting["tokens_reported"],
140 "captured_at": utc_now_iso(),
141 },
142 )
143 if run["status"] != "ok":
144 return None, f"{run['status']}: {run['stderr'][-500:].strip()}"
145 return loaded, ""
148# --- main -------------------------------------------------------------------
151def main(argv: list[str] | None = None) -> int:
152 p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
153 p.add_argument("--skill-path", required=True, type=Path)
154 p.add_argument("--queries", required=True, type=Path)
155 p.add_argument("--output-dir", required=True, type=Path)
156 p.add_argument(
157 "--project-root", type=Path, default=None, help="holds _bmad/; found from the skill path when omitted"
158 )
159 p.add_argument("--harness", type=Path, default=None, help="harness JSON for a project without BMad")
160 p.add_argument("--runs-per-query", type=int, default=3)
161 p.add_argument("--threshold", type=float, default=0.5)
162 p.add_argument("--timeout", type=int, default=180)
163 p.add_argument("--workers", type=int, default=4)
164 p.add_argument("--quiet", action="store_true")
165 args = p.parse_args(argv)
167 skill_path = args.skill_path.resolve()
168 queries_file = args.queries.resolve()
169 if not queries_file.is_file():
170 print(f"queries file not found: {queries_file}", file=sys.stderr)
171 return 2
173 skill_name, description = parse_skill_md(skill_path)
174 queries = read_json(queries_file)
175 if not isinstance(queries, list):
176 print("queries file must be a JSON list", file=sys.stderr)
177 return 2
179 project_root = args.project_root.resolve() if args.project_root else find_project_root(skill_path)
180 try:
181 harness, harness_note = resolve_harness(project_root, args.harness)
182 except (ValueError, json.JSONDecodeError) as e:
183 print(f"harness invalid: {e}", file=sys.stderr)
184 return 2
186 run_id, run_dir = make_run_dir(args.output_dir, f"{skill_name}-triggers")
187 (run_dir / "queries").mkdir(parents=True, exist_ok=True)
188 write_json(
189 run_dir / "run.json",
190 {
191 "run_id": run_id,
192 "skill_name": skill_name,
193 "description": description,
194 "harness": harness_note,
195 "command": (harness or {}).get("command"),
196 "started_at": utc_now_iso(),
197 "query_count": len(queries),
198 "runs_per_query": args.runs_per_query,
199 "threshold": args.threshold,
200 },
201 )
203 if harness is None:
204 if not args.quiet:
205 print(f"[run_triggers] no harness ({harness_note}); staging only", file=sys.stderr)
206 output = {
207 "run_id": run_id,
208 "completed_at": utc_now_iso(),
209 "skill_name": skill_name,
210 "description": description,
211 "status": "skipped",
212 "reason": "no harness recorded",
213 "results": [],
214 "summary": {"total": len(queries), "passed": 0, "failed": 0, "unmeasured": len(queries)},
215 }
216 write_json(run_dir / "triggers-result.json", output)
217 print(json.dumps(output, indent=2))
218 return 3
220 def run_one(idx: int, q: dict, run_idx: int) -> tuple[int, bool | None, str]:
221 attempt_dir = run_dir / "queries" / f"q{idx:03d}-r{run_idx}"
222 try:
223 loaded, error = run_query_once(q["query"], skill_name, description, harness, attempt_dir, args.timeout)
224 except (OSError, ValueError) as e:
225 loaded, error = None, str(e)
226 return idx, loaded, error
228 attempts: dict[int, list[bool]] = {idx: [] for idx in range(len(queries))}
229 errors: dict[int, list[str]] = {idx: [] for idx in range(len(queries))}
230 if not args.quiet:
231 print(f"[run_triggers] harness: {' '.join(harness['command'])}", file=sys.stderr)
232 print(f"[run_triggers] {len(queries)} queries x {args.runs_per_query} runs", file=sys.stderr)
234 with ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
235 futures = [
236 pool.submit(run_one, idx, q, run_idx)
237 for idx, q in enumerate(queries)
238 for run_idx in range(args.runs_per_query)
239 ]
240 for fut in as_completed(futures):
241 idx, loaded, error = fut.result()
242 if loaded is None:
243 errors[idx].append(error)
244 if not args.quiet:
245 print(f" attempt failed for query {idx}: {error}", file=sys.stderr)
246 else:
247 attempts[idx].append(loaded)
249 results = []
250 for idx, q in enumerate(queries):
251 runs = attempts[idx]
252 measured = len(runs) == args.runs_per_query
253 rate = (sum(runs) / len(runs)) if runs else 0.0
254 should = bool(q.get("should_trigger", True))
255 passed = ((rate >= args.threshold) if should else (rate < args.threshold)) if measured else None
256 results.append(
257 {
258 "query": q["query"],
259 "should_trigger": should,
260 "trigger_rate": round(rate, 3),
261 "triggers": int(sum(runs)),
262 "runs": len(runs),
263 "errors": errors[idx],
264 "pass": passed,
265 }
266 )
268 unmeasured = sum(1 for r in results if r["pass"] is None)
269 output = {
270 "run_id": run_id,
271 "completed_at": utc_now_iso(),
272 "skill_name": skill_name,
273 "description": description,
274 "harness": harness_note,
275 "results": results,
276 "summary": {
277 "total": len(results),
278 "passed": sum(1 for r in results if r["pass"] is True),
279 "failed": sum(1 for r in results if r["pass"] is False),
280 "unmeasured": unmeasured,
281 },
282 }
283 write_json(run_dir / "triggers-result.json", output)
284 print(json.dumps(output, indent=2))
285 return 1 if unmeasured else 0
288if __name__ == "__main__":
289 sys.exit(main())