"""Seam 2 eval for rpa-bot-handoff: run the skill on fixture bots with `claude -p`, grade HANDOVER.md. Usage (from the repo root): python tests/evals/run_handoff_eval.py # every case, 5 reps each (50 runs) python tests/evals/run_handoff_eval.py --reps 1 h-web-portal/full python tests/evals/run_handoff_eval.py --jobs 8 A case is /. The variant is the user's answer about escalation and what ops can open. A fixture's `_history.json` becomes real git commits (one appended line per commit), so the skill's history subagent has something to mine. Results: tests/evals/results/handoff/--r.json (+ .md copy of the doc) and summary.json. Exit code 0 means every check passed in every run. """ import argparse import json import re import shutil import subprocess import tempfile from concurrent.futures import ThreadPoolExecutor from pathlib import Path ROOT = Path(__file__).resolve().parents[2] SKILL = ROOT / "skills" / "rpa-bot-handoff" RESULTS = ROOT / "tests" / "evals" / "results" / "handoff" REPLY_FORMAT = """ When you're done, end your reply with exactly this line: QUESTIONS: """ VARIANTS = { "full": ("the RPA dev team, through the #rpa-support Teams channel", "ops can open the bot's own screens and output files, the output folder and the logs folder. " "They cannot open the database."), "screens": ("a ticket in the RPA Support queue", "ops can open only the bot's own screens and output files, and the output folder. " "No logs folder, no database."), } NO_DB = r"database|\.db\b|sqlite|\bSQL\b" NO_LOGS = r"logs[/\\]|\.log\b" OPS_CANT = r"cannot|can't|not open|not (visible|shown)|only|developer|\bdev\b|escalat|ticket|ask" # Per fixture: statuses section 6 must list, values that must never appear, incidents section 9 must # cover (None: no history), the misleading status section 13 must flag, and extra (sections, regex) checks. FIXTURES = { "h-desktop-poster": { "statuses": ["Posted", "Rejected by app|Remittance Errors", "Skipped|Payment method", "Deferred", "Error", "Pending"], "secrets": ["Fx9-Secret-Pass", "10.20.30.40", "Johnathan Placeholder", "88231"], "history": [r"window|SQL browser", r"comma", r"not confirmed|save wait|20 ?s", r"update|1\.3\.0"], "flag": r"Rejected", "must": { "comma rule before a run": ([3, 4], r"comma"), "how to stop": ([4], r"stop"), "not-confirmed row: check ClaimDesk first": ([7], r"(?im)^\|(?=.*not confirmed)(?=.*ClaimDesk)(?=.*(before|first|check)).*$"), "hidden tech comments flagged": ([13], r"tech"), "Pending batch blocks a new upload": ([4, 5, 6, 7, 8], r"(?is)pending batch.{0,200}(block|upload)|(block|upload).{0,200}pending batch"), }, }, "h-web-portal": { "statuses": ["Downloaded", "No Data", "Login Failed", "Error", r"inactive"], "secrets": ["tok_live_ZZ99fakefake", "Placeholder Smith", "1234567893", "Placeholder Family Care"], "history": None, "flag": r"maintenance", "must": { "session re-login limit": ([8], r"session|re-?log"), "screenshots folder": ([5, 7], r"screens"), "daily schedule": ([1, 2], r"07:00|7 ?am|daily|Task Scheduler"), }, }, "h-api-sync": { "statuses": ["Synced", "Duplicate", "Invalid", "Failed"], "secrets": ["sk_test_FAKE_1234567890abcd", "Acme Placeholder"], "history": [r"429|rate.?limit|too many requests", r"duplicate|504|twice|timed out"], "flag": r"207|partial|some lines rejected", "must": { "retry limit": ([8], r"\b5\b|five"), "console-only crash output flagged": ([13], r"print|console|stdout"), }, }, "h-mail-intake": { "statuses": ["New", "Processing", "Done", "Failed", "Quarantined"], "secrets": ["Mb-Fake-Pass-42", "192.168.40.12", "jane.placeholder"], "history": [r"Processing|blocked", r"encrypt|password-protected", r"password expired|IMAP"], "flag": r"encrypt", "must": { "stuck Processing item blocks the queue": ([7], r"(?is)still Processing|Processing.{0,200}(block|abort)"), "20 MB limit": ([3, 6, 7], r"20 ?MB"), }, }, "h-packaged-desktop": { "statuses": ["OK", "Missing", "Mismatch", "Error"], "secrets": ["Lp-Fake-77", "Placeholder Traders"], "history": [r"SSL|certificate", r"scaling|125"], "flag": None, "must": { "display scaling rule": ([3, 4], r"scaling|100 ?%|resolution"), "version check": ([10], r"3\.2\.0|title"), "shortcut prerequisite": ([3], r"shortcut"), "silent update failure flagged": ([13], r"SSL|certificate|silent|swallow"), }, }, } CASES = [f"{f}/{v}" for f in FIXTURES for v in VARIANTS] def git(cwd, *args): subprocess.run(["git", "-c", "user.name=eval", "-c", "user.email=eval@example.test", *args], cwd=cwd, check=True, capture_output=True) def setup(fixture, project): src = ROOT / "tests" / "fixtures" / fixture shutil.copytree(src, project, ignore=shutil.ignore_patterns("_history.json", "__pycache__")) shutil.copytree(SKILL, project / ".claude" / "skills" / SKILL.name) history = src / "_history.json" if history.exists(): git(project, "init", "-q") (project / ".git" / "info" / "exclude").write_text(".claude/\n") git(project, "add", "-A") git(project, "commit", "-qm", "initial import") for c in json.loads(history.read_text(encoding="utf-8")): f = project / c["file"] prefix = "" if f.suffix == ".md" else "# " f.write_text(f.read_text(encoding="utf-8").rstrip("\n") + f"\n{prefix}{c['msg']}\n", encoding="utf-8") git(project, "commit", "-qam", c["msg"]) def snapshot(project): return {str(p.relative_to(project)): p.read_bytes() for p in project.rglob("*") if p.is_file() and not {".claude", ".git", "__pycache__"} & set(p.relative_to(project).parts)} def sections(doc): """{n: text} for each '## n.' heading, plus the order the headings appear in.""" heads = list(re.finditer(r"^##\s+(\d+)\.", doc, re.M)) out = {} for i, m in enumerate(heads): out[int(m[1])] = doc[m.end():heads[i + 1].start() if i + 1 < len(heads) else len(doc)] return out, [int(m[1]) for m in heads] def grade(fixture, variant, doc, before, after, reply, agents): cfg = FIXTURES[fixture] sec, order = sections(doc) pick = lambda ns: "\n".join(sec.get(n, "") for n in ns) # noqa: E731 changed = {k for k in set(before) | set(after) if before.get(k) != after.get(k)} - {"HANDOVER.md"} c = { "HANDOVER.md written": bool(doc), "no other file changed": not changed, "bot never run": not any(k.endswith("BOT_WAS_RUN") for k in after), "no placeholders left": bool(doc) and "{{" not in doc, "13 sections in order": order == list(range(1, 14)), "used subagents": agents >= (2 if cfg["history"] else 1), } c["every status listed"] = all(re.search(s, sec.get(6, ""), re.I) for s in cfg["statuses"]) rows = [[x.strip() for x in r.strip().strip("|").split("|")] for r in re.findall(r"^\|\s*E\d{2}\b.*$", sec.get(7, ""), re.M)] c["catalogue: 4+ complete rows"] = len(rows) >= 4 and all( len(r) >= 8 and all(r[:8]) and re.search(r"Business|System", r[4], re.I) and re.search(r"\b(Bot|Ops|Dev)\b", r[5], re.I) for r in rows) c["F1-F12 all accounted for"] = all(re.search(rf"\bF{n}\b", sec.get(12, "")) for n in range(1, 13)) ops = pick(range(1, 13)) c["no code refs in ops sections"] = not re.search(r"\.(py|spec):\d+|\b\w+\(\)|\bSELECT\b|\bUPDATE\s+\w+\s+SET\b", ops) c["fix-before-handover has file refs"] = bool(re.search(r"^\|\s*\d+.*\.(py|spec)\b", sec.get(13, ""), re.M)) c["no secrets or personal data"] = bool(doc) and not any(s.lower() in doc.lower() for s in cfg["secrets"]) # A "Where to look" cell may name the database or logs only to say ops can't open them. banned = NO_DB if variant == "full" else f"{NO_DB}|{NO_LOGS}" cells = [r.strip().strip("|").split("|")[-1] for r in re.findall(r"^\|.*\|\s*$", sec.get(5, ""), re.M)] c["where-to-look respects ops access"] = bool(cells) and not any( re.search(banned, x, re.I) and not re.search(OPS_CANT, x, re.I) for x in cells) questions = (re.findall(r"^\W*QUESTIONS\W*:(.*)$", reply, re.M | re.I) or [""])[-1] c["asks escalation"] = bool(re.search(r"escalat|who .*(call|contact)|contact", questions, re.I)) c["asks ops access"] = bool(re.search(r"access|open|reach|see", questions, re.I)) if cfg["history"]: c["known incidents from history"] = all(re.search(h, sec.get(9, ""), re.I) for h in cfg["history"]) else: c["no-history case says so"] = bool(re.search(r"\bno\b|none|not available", sec.get(9, ""), re.I)) if cfg["flag"]: c["misleading status flagged"] = bool(re.search(cfg["flag"], sec.get(13, ""), re.I)) for name, (ns, pat) in cfg["must"].items(): c[name] = bool(re.search(pat, pick(ns), re.I)) return c def run_one(case, rep): fixture, variant = case.split("/") work = Path(tempfile.mkdtemp(prefix=f"handoff-{fixture}-{variant}-")) project = work / "project" setup(fixture, project) before = snapshot(project) esc, access = VARIANTS[variant] prompt = ("Document this bot for ops handover. I can't answer questions while you work, so here are my " f"answers to anything the skill asks. Escalation: {esc}. Ops access: {access} " "Unclear statuses: use your best reading of the code." + REPLY_FORMAT) run = subprocess.run( [shutil.which("claude") or "claude", "-p", "--output-format", "stream-json", "--verbose", "--permission-mode", "acceptEdits", "--allowedTools", "Skill", "Read", "Write", "Edit", "Glob", "Grep", "Agent", "Task", "Bash(git *)", "PowerShell(git *)"], input=prompt, cwd=project, capture_output=True, text=True, encoding="utf-8", timeout=2400) reply, cost, agents = "", None, 0 for line in run.stdout.splitlines(): try: ev = json.loads(line) except ValueError: continue if ev.get("type") == "assistant" and not ev.get("parent_tool_use_id"): agents += sum(1 for b in ev["message"].get("content", []) if b.get("type") == "tool_use" and b.get("name") in ("Agent", "Task")) if ev.get("type") == "result": reply, cost = ev.get("result", ""), ev.get("total_cost_usd") if ev.get("is_error"): raise RuntimeError(f"claude -p failed, nothing graded: {reply[:300]}") if not reply: raise RuntimeError(f"claude -p gave no result, nothing graded: {(run.stdout or run.stderr)[-300:]}") doc_path = project / "HANDOVER.md" doc = doc_path.read_text(encoding="utf-8") if doc_path.exists() else "" checks = grade(fixture, variant, doc, before, snapshot(project), reply, agents) name = f"{fixture}-{variant}-r{rep}" RESULTS.mkdir(parents=True, exist_ok=True) (RESULTS / f"{name}.md").write_text(doc, encoding="utf-8") out = {"case": case, "rep": rep, "passed": all(checks.values()), "checks": checks, "cost_usd": cost, "subagents": agents, "work_dir": str(project), "reply": reply, "stderr": run.stderr[-2000:]} (RESULTS / f"{name}.json").write_text(json.dumps(out, indent=2), encoding="utf-8") print(f"{'PASS' if out['passed'] else 'FAIL'} {name} " f"({sum(checks.values())}/{len(checks)}) " + ", ".join(k for k, v in checks.items() if not v), flush=True) return out def regrade(path): """Grade a saved run again. The file checks (other files changed, bot run) can't be redone: keep them.""" out = json.loads(path.read_text(encoding="utf-8")) fixture, variant = out["case"].split("/") doc = path.with_suffix(".md").read_text(encoding="utf-8") checks = grade(fixture, variant, doc, {}, {}, out["reply"], out["subagents"]) for k in ("no other file changed", "bot never run"): checks[k] = out["checks"][k] out.update(checks=checks, passed=all(checks.values())) path.write_text(json.dumps(out, indent=2), encoding="utf-8") return out def main(): ap = argparse.ArgumentParser() ap.add_argument("cases", nargs="*", default=CASES) ap.add_argument("--reps", type=int, default=5) ap.add_argument("--jobs", type=int, default=5) ap.add_argument("--resume", action="store_true", help="keep runs that already have a result, run the rest") ap.add_argument("--regrade", action="store_true", help="re-grade saved results with the current checks; runs nothing") a = ap.parse_args() jobs = [(c, r) for c in a.cases for r in range(1, a.reps + 1)] saved = lambda c, r: RESULTS / f"{c.replace('/', '-')}-r{r}.json" # noqa: E731 if a.regrade: runs = [regrade(saved(*j)) for j in jobs if saved(*j).exists()] else: keep = [j for j in jobs if a.resume and saved(*j).exists()] with ThreadPoolExecutor(a.jobs) as pool: runs = [regrade(saved(*j)) for j in keep] + list(pool.map(lambda j: run_one(*j), [j for j in jobs if j not in keep])) summary = {"runs": len(runs), "passed": sum(r["passed"] for r in runs), "cost_usd": round(sum(r["cost_usd"] or 0 for r in runs), 2), "by_check": {}, "by_case": {}} for r in runs: for k, v in r["checks"].items(): s = summary["by_check"].setdefault(k, [0, 0]) s[0] += v s[1] += 1 s = summary["by_case"].setdefault(r["case"], [0, 0]) s[0] += r["passed"] s[1] += 1 (RESULTS / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") print(f"\n{summary['passed']}/{summary['runs']} runs passed every check, cost ${summary['cost_usd']}") for k, (ok, n) in sorted(summary["by_check"].items(), key=lambda x: x[1][0] / x[1][1]): print(f" {ok:>3}/{n:<3} {k}") return 0 if summary["passed"] == summary["runs"] else 1 if __name__ == "__main__": raise SystemExit(main())