diff --git a/engine/hooks/repeat-error-stop/README.md b/engine/hooks/repeat-error-stop/README.md index cb9d5580..9bc51884 100644 --- a/engine/hooks/repeat-error-stop/README.md +++ b/engine/hooks/repeat-error-stop/README.md @@ -70,17 +70,26 @@ wrong-typed, or expired state means no block and no nudge. ## Backtest against real sessions -`backtest.py` replays Claude Code transcripts through the same `detect.py` -and reports, for every point the hook would have fired, how many identical -errors actually followed (thrash it would have cut) and whether the next real -run of that command succeeded anyway (a premature stop). +`detect.py:replay_blocks` replays Claude Code transcripts through the same +counting the hooks use, driven by the shared runner +`scripts/backtest_detector.py`. For every point the hook would have fired it +reports how many identical errors actually followed (`saved`, the thrash it +would have cut) and whether the next real run of that command succeeded +anyway (`next_try=ok`, a premature stop). ```sh python3 engine/hooks/repeat-error-stop/backtest.py ~/.claude/projects/ [...] REPEAT_ERROR_STOP_OBSERVED=0 python3 engine/hooks/repeat-error-stop/backtest.py ... python3 engine/hooks/repeat-error-stop/backtest.py --epochs 2 --expect fires=88 --expect later_identical_errors_saved=39 ~/.claude/projects/ [...] +python3 scripts/backtest_detector.py --detector engine/hooks/repeat-error-stop/detect.py:replay_blocks --unit rows ~/.claude/projects/ [...] +REPEAT_ERROR_STOP_OBSERVED=0 python3 scripts/backtest_detector.py --detector engine/hooks/repeat-error-stop/detect.py:replay_blocks --unit rows ... ``` +The knobs above (`REPEAT_ERROR_STOP_THRESHOLD`, `REPEAT_ERROR_STOP_OBSERVED`, +`REPEAT_ERROR_STOP_RESET_ON_EDIT`) apply to the replay too. `--json OUT` +writes every block; `--compare ` lists the blocks a change adds or +removes. + 286 sessions, 38.6k tool results, Aug 2–Sep 1 2026 (Invoker + catstack + two other repos): diff --git a/engine/hooks/repeat-error-stop/detect.py b/engine/hooks/repeat-error-stop/detect.py index e7d2c00e..dec77933 100644 --- a/engine/hooks/repeat-error-stop/detect.py +++ b/engine/hooks/repeat-error-stop/detect.py @@ -335,3 +335,94 @@ def handle_prompt(payload: dict) -> bool: return False reset_state(payload) return True + + +def _content_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join(b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text") + return "" + + +def is_human_prompt_row(row: dict) -> bool: + if row.get("type") != "user" or row.get("isMeta") or row.get("isSidechain"): + return False + content = (row.get("message") or {}).get("content") + if isinstance(content, list) and any(isinstance(b, dict) and b.get("type") == "tool_result" for b in content): + return False + text = _content_text(content).strip() + if not text or text.startswith("<") or text.startswith("[Request interrupted"): + return False + return not is_automated_prompt(text) + + +def _closed_blocks(blocks: list[dict]): + for blk in blocks: + yield blk["key"], blk["text"], {"tool": blk["tool"], "next_try": blk["next_try"], "saved": blk["saved"]} + + +def replay_blocks(rows, threshold: int = THRESHOLD): + """Rows detector for scripts/backtest_detector.py: every tool result of a + Claude Code transcript, replayed through the same counting as the hooks. + A hit is the result that trips the block, reported once its outcome is + known: saved = identical errors that followed it, next_try = what the next + real run of a blocked command did (ok means the block was premature).""" + pending: dict[str, dict] = {} + counts: dict[str, dict] = {} + open_blocks: list[dict] = [] + edit_epoch = 0 + for index, row in rows: + if is_human_prompt_row(row): + yield from _closed_blocks(open_blocks) + counts, open_blocks, edit_epoch = {}, [], 0 + continue + msg = row.get("message") or {} + if row.get("type") == "assistant": + for b in msg.get("content") or []: + if isinstance(b, dict) and b.get("type") == "tool_use": + pending[b.get("id")] = {"name": b.get("name"), "input": b.get("input") or {}} + continue + if row.get("type") != "user" or not isinstance(msg.get("content"), list): + continue + for offset, b in enumerate(msg["content"]): + if not (isinstance(b, dict) and b.get("type") == "tool_result"): + continue + call = pending.pop(b.get("tool_use_id"), None) + if not call: + continue + key = f"{index}.{offset}" + text = _content_text(b.get("content")) + if b.get("is_error"): + payload = {"hook_event_name": "PostToolUseFailure", "tool_name": call["name"], "tool_input": call["input"], "error": text or "tool failed"} + else: + payload = {"hook_event_name": "PostToolUse", "tool_name": call["name"], "tool_input": call["input"], "tool_response": text} + cmd = command_signature(payload) + failure = failure_text(payload) + sig = error_signature(failure, cmd) if failure is not None else None + for blk in open_blocks: + if cmd and cmd in blk["commands"] and blk["next_try"] == "none": + blk["next_try"] = "same" if (sig and sig[0] == blk["sig"]) else "ok" + if sig and sig[0] == blk["sig"]: + blk["saved"] += 1 + if sig is None: + if RESET_ON_EDIT and call["name"] in EDIT_TOOLS and not b.get("is_error"): + edit_epoch += 1 + yield key, text, None + continue + digest, sample = sig + entry = counts.setdefault(digest, {"count": 0, "commands": set(), "epoch": edit_epoch}) + if entry["epoch"] != edit_epoch: + entry.update(count=0, commands=set(), epoch=edit_epoch) + entry["count"] += 1 + if cmd: + entry["commands"].add(cmd) + if entry["count"] != threshold: + yield key, text, None + continue + command = str(call["input"].get("command") or "")[:160] + open_blocks.append({ + "key": key, "tool": call["name"], "sig": digest, "commands": set(entry["commands"]), + "text": f"{call['name']}: {command} -> {sample[:200]}", "saved": 0, "next_try": "none", + }) + yield from _closed_blocks(open_blocks) diff --git a/engine/hooks/repeat-error-stop/tests/test_hooks.py b/engine/hooks/repeat-error-stop/tests/test_hooks.py index f55892fa..064198c4 100644 --- a/engine/hooks/repeat-error-stop/tests/test_hooks.py +++ b/engine/hooks/repeat-error-stop/tests/test_hooks.py @@ -334,6 +334,53 @@ def test_malformed_stdin_fails_open(self): claude_pretooluse.main() +def call_row(tool_id: str, command: str) -> dict: + return {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "id": tool_id, "name": "Bash", "input": {"command": command}}]}} + + +def result_row(tool_id: str, text: str, is_error: bool) -> dict: + return {"type": "user", "message": {"content": [ + {"type": "tool_result", "tool_use_id": tool_id, "content": text, "is_error": is_error}]}} + + +def failing_runs(command: str, count: int, start: int = 0) -> list[dict]: + rows = [] + for i in range(start, start + count): + rows += [call_row(f"t{i}", command), result_row(f"t{i}", "Exit code 1\n" + TIMEOUT.format(name=f"n{i}"), True)] + return rows + + +class TestReplayBlocks(unittest.TestCase): + def replay(self, rows): + return list(detect.replay_blocks(enumerate(rows), threshold=3)) + + def test_replay_hits_third_identical_failure_and_counts_what_followed(self): + rows = failing_runs("pnpm test", 4) + rows += [call_row("ok", "pnpm test"), result_row("ok", PASS, False)] + units = self.replay(rows) + self.assertEqual(len(units), 5) + hits = [u for u in units if u[2]] + self.assertEqual(len(hits), 1) + key, text, verdict = hits[0] + self.assertEqual(key, "5.0") + self.assertIn("pnpm test", text) + self.assertEqual(verdict, {"tool": "Bash", "next_try": "same", "saved": 1}) + + def test_replay_marks_block_premature_when_next_run_succeeds(self): + rows = failing_runs("pnpm test", 3) + [call_row("ok", "pnpm test"), result_row("ok", PASS, False)] + verdicts = [u[2] for u in self.replay(rows) if u[2]] + self.assertEqual(verdicts, [{"tool": "Bash", "next_try": "ok", "saved": 0}]) + + def test_replay_human_prompt_resets_count_no_hit(self): + rows = failing_runs("pnpm test", 2) + rows.append({"type": "user", "message": {"role": "user", "content": "try the other branch"}}) + rows += failing_runs("pnpm test", 2, start=2) + units = self.replay(rows) + self.assertEqual(len(units), 4) + self.assertFalse(any(u[2] for u in units)) + + class TestInstallers(unittest.TestCase): def test_claude_installer_merges_once(self): base = {"hooks": {"Stop": [{"matcher": "*", "hooks": [{"type": "command", "command": "other"}]}]}} diff --git a/engine/hooks/wait-needs-wakeup/README.md b/engine/hooks/wait-needs-wakeup/README.md index c38ed39c..2919e877 100644 --- a/engine/hooks/wait-needs-wakeup/README.md +++ b/engine/hooks/wait-needs-wakeup/README.md @@ -39,16 +39,32 @@ poll." - `detect.py` -- loop / sleep / status-check patterns, wait-language and clock-ETA patterns, transcript wakeup state; `decide_pretooluse()`, - `decide_stop()`. + `decide_stop()`, and the two backtest entry points `pretooluse_reason()` + and `replay_stop()`. - `claude_pretooluse.py`, `claude_stop_check.py` -- Claude entrypoints. - `claude.hook.json` / `install_claude_hook.py` -- settings.json merge for both events (idempotent). -- `backtest.py` -- replay over a transcript (`backtest.py X.jsonl`) or over - the fixtures (`backtest.py --fixtures`); prints would-block counts. - `tests/fixtures/poll_commands_{fires,silent}.json`, `tests/fixtures/wait_replies_{fires,silent}.json` -- sanitized replays of the real commands and replies (fires) and their corrected forms (silent). - `tests/test_hooks.py` -- every fires fixture blocks, every silent fixture - passes, the backtest reproduces the counts. + passes, and the Stop replay agrees with the hook on every fixture. Tests: `python3 -m unittest discover -s engine/hooks/wait-needs-wakeup/tests -v` + +## Backtest against real sessions + +Both halves replay over local transcripts through the shared runner, +`scripts/backtest_detector.py`: + +```sh +python3 scripts/backtest_detector.py --detector engine/hooks/wait-needs-wakeup/detect.py:pretooluse_reason --unit tool --tool Bash X.jsonl +python3 scripts/backtest_detector.py --detector engine/hooks/wait-needs-wakeup/detect.py:replay_stop --unit rows X.jsonl +``` + +The first counts the Bash commands the PreToolUse half would block. The +second counts the turn-ending replies the Stop half would block; a wait +reply that already names an ETA and holds a wakeup is listed as a +near-miss. Leave out the path to scan the newest sessions (`--limit N`), +and add `--compare ` to see what a change newly blocks or lets +through. diff --git a/engine/hooks/wait-needs-wakeup/backtest.py b/engine/hooks/wait-needs-wakeup/backtest.py deleted file mode 100644 index f935f594..00000000 --- a/engine/hooks/wait-needs-wakeup/backtest.py +++ /dev/null @@ -1,128 +0,0 @@ -#!/usr/bin/env python3 -"""Replay the wait-needs-wakeup rules over a Claude Code transcript. - - python3 engine/hooks/wait-needs-wakeup/backtest.py TRANSCRIPT.jsonl - python3 engine/hooks/wait-needs-wakeup/backtest.py --fixtures - -Transcript mode prints how many Bash commands were foreground polls the -PreToolUse half would have blocked, and how many end-of-turn replies the -Stop half would have blocked (wait language with no ETA or no wakeup). -Fixture mode replays tests/fixtures/*.json and prints blocked/total per -file; the *_fires files are the incident, the *_silent files are the -corrected forms and must come out 0. -""" -from __future__ import annotations - -import json -import os -import sys - -HERE = os.path.dirname(os.path.abspath(__file__)) -sys.path.insert(0, HERE) - -import detect # noqa: E402 - - -def _final_reply_indexes(lines: list[dict]) -> list[int]: - """Assistant text lines that end a response: the next user/assistant - line is not a tool_result and not another assistant block.""" - out: list[int] = [] - for i, data in enumerate(lines): - if data.get("type") != "assistant" or not detect._text_content(data).strip(): - continue - nxt = next((d for d in lines[i + 1:] if d.get("type") in ("user", "assistant")), None) - if nxt is None: - out.append(i) - continue - if nxt.get("type") == "assistant": - continue - content = (nxt.get("message") or {}).get("content") - if isinstance(content, list) and any( - isinstance(b, dict) and b.get("type") == "tool_result" for b in content - ): - continue - out.append(i) - return out - - -def run_transcript(path: str) -> dict: - with open(path, encoding="utf-8") as handle: - lines = detect.parse_lines(handle) - report = { - "path": path, "bash_commands": 0, "poll_blocked": 0, "poll_details": [], - "final_replies": 0, "wait_replies": 0, "reply_blocked": 0, "reply_details": [], - } - for i, data in enumerate(lines): - for block in detect._tool_uses(data): - if block.get("name") != "Bash": - continue - inp = block.get("input") or {} - command = inp.get("command") - if not isinstance(command, str): - continue - report["bash_commands"] += 1 - reason = detect.classify_command(command, bool(inp.get("run_in_background"))) - if reason: - report["poll_blocked"] += 1 - report["poll_details"].append((i, reason, command.strip().splitlines()[0][:90])) - seen: set[str] = set() - for i in _final_reply_indexes(lines): - text = detect._text_content(lines[i]) - if text in seen: - continue - seen.add(text) - report["final_replies"] += 1 - if not detect.is_wait_reply(text): - continue - report["wait_replies"] += 1 - verdict = detect.decide_stop_from_lines(text, lines[: i + 1]) - if verdict: - report["reply_blocked"] += 1 - report["reply_details"].append((i, text.strip().replace("\n", " ")[:110])) - return report - - -def run_fixtures(fixtures_dir: str) -> dict: - counts: dict = {} - for name in sorted(os.listdir(fixtures_dir)): - if not name.endswith(".json"): - continue - with open(os.path.join(fixtures_dir, name), encoding="utf-8") as handle: - cases = json.load(handle) - blocked = 0 - for case in cases: - if "command" in case: - hit = detect.classify_command(case["command"], case.get("run_in_background", False)) - else: - lines = detect.parse_lines(json.dumps(line) for line in case.get("transcript", [])) - hit = detect.decide_stop_from_lines(case["reply"], lines) - blocked += 1 if hit else 0 - counts[name] = {"total": len(cases), "blocked": blocked} - return counts - - -def main(argv: list[str]) -> int: - if not argv or argv[0] in ("-h", "--help"): - print(__doc__) - return 2 - if argv[0] == "--fixtures": - counts = run_fixtures(os.path.join(HERE, "tests", "fixtures")) - for name, c in counts.items(): - print(f"{name}: {c['blocked']}/{c['total']} would block") - return 0 - report = run_transcript(argv[0]) - print(f"wait-needs-wakeup backtest: {report['path']}") - print(f"bash commands scanned: {report['bash_commands']}") - print(f" poll commands that would block (PreToolUse): {report['poll_blocked']}") - for line, reason, head in report["poll_details"]: - print(f" line {line}: {reason}: {head}") - print(f"final assistant replies scanned: {report['final_replies']}") - print(f" replies with wait language: {report['wait_replies']}") - print(f" replies that would block (Stop): {report['reply_blocked']}") - for line, head in report["reply_details"]: - print(f" line {line}: {head}") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/engine/hooks/wait-needs-wakeup/detect.py b/engine/hooks/wait-needs-wakeup/detect.py index 0159cc65..4616e4c0 100644 --- a/engine/hooks/wait-needs-wakeup/detect.py +++ b/engine/hooks/wait-needs-wakeup/detect.py @@ -129,14 +129,18 @@ def classify_command(command: str, run_in_background: bool = False) -> str | Non return None -def decide_pretooluse(payload: dict) -> str | None: +def pretooluse_reason(payload: dict) -> str | None: if payload.get("tool_name") not in (None, "Bash"): return None tool_input = payload.get("tool_input") or {} command = tool_input.get("command") if not isinstance(command, str): return None - reason = classify_command(command, bool(tool_input.get("run_in_background"))) + return classify_command(command, bool(tool_input.get("run_in_background"))) + + +def decide_pretooluse(payload: dict) -> str | None: + reason = pretooluse_reason(payload) if not reason: return None return PRETOOLUSE_MESSAGE.format(reason=reason) @@ -205,50 +209,93 @@ def _tool_uses(data: dict): yield block -def wakeup_state(lines: list[dict]) -> dict: - """What the transcript says about scheduled wakeups. +class WakeupTracker: + """What the transcript so far says about scheduled wakeups, fed one line at a time. scheduled_this_turn: a ScheduleWakeup / CronCreate call since the last human message. pending: a Monitor / Agent / run_in_background Bash whose task-notification has not arrived yet (it will wake the agent).""" - turn_start = 0 - for i, data in enumerate(lines): + + def __init__(self) -> None: + self.scheduled_this_turn = False + self.launched: dict[str, str] = {} + self.notified: set[str] = set() + + def feed(self, index: int, data: dict) -> None: if _is_human_user_line(data): - turn_start = i - launched: dict[str, str] = {} - notified: set[str] = set() - scheduled_this_turn = False - for i, data in enumerate(lines): + self.scheduled_this_turn = False content = data.get("content") if isinstance(data.get("content"), str) else _text_content(data) - for tid in TOOL_USE_ID_RE.findall(content or ""): - notified.add(tid) + self.notified.update(TOOL_USE_ID_RE.findall(content or "")) for block in _tool_uses(data): name = block.get("name") inp = block.get("input") or {} - if name in WAKEUP_TOOLS and i >= turn_start: - scheduled_this_turn = True + if name in WAKEUP_TOOLS: + self.scheduled_this_turn = True if name in TASK_TOOLS or (name == "Bash" and inp.get("run_in_background")): - launched[block.get("id") or f"line{i}"] = name - pending = [name for tid, name in launched.items() if tid not in notified] - return {"scheduled_this_turn": scheduled_this_turn, "pending": pending} + self.launched[block.get("id") or f"line{index}"] = name + + def state(self) -> dict: + pending = [name for tid, name in self.launched.items() if tid not in self.notified] + return {"scheduled_this_turn": self.scheduled_this_turn, "pending": pending} + + +def wakeup_state(lines: list[dict]) -> dict: + tracker = WakeupTracker() + for i, data in enumerate(lines): + tracker.feed(i, data) + return tracker.state() + + +def stop_gaps(message: str, state: dict) -> list[str]: + gaps = [] + if not (state["scheduled_this_turn"] or state["pending"]): + gaps.append("no wakeup is scheduled (no ScheduleWakeup / Monitor call, no background task still pending)") + if not has_clock_eta(message): + gaps.append("no clock-time ETA is named") + return gaps def decide_stop_from_lines(message: str, lines: list[dict]) -> str | None: if not is_wait_reply(message): return None - state = wakeup_state(lines) - has_wakeup = state["scheduled_this_turn"] or bool(state["pending"]) - eta = has_clock_eta(message) - if has_wakeup and eta: + gaps = stop_gaps(message, wakeup_state(lines)) + if not gaps: return None - gaps = [] - if not has_wakeup: - gaps.append("no wakeup is scheduled (no ScheduleWakeup / Monitor call, no background task still pending)") - if not eta: - gaps.append("no clock-time ETA is named") return STOP_MESSAGE.format(gap="; ".join(gaps)) +def _is_tool_result_line(data: dict) -> bool: + message = data.get("message") + content = message.get("content") if isinstance(message, dict) else None + return isinstance(content, list) and any( + isinstance(b, dict) and b.get("type") == "tool_result" for b in content + ) + + +def replay_stop(rows): + """Rows detector for scripts/backtest_detector.py: every turn-ending reply, + judged against the wakeup state at that point. A wait reply that names an + ETA and holds a wakeup is the near-miss.""" + tracker = WakeupTracker() + held = None + seen: set[str] = set() + for index, data in rows: + tracker.feed(index, data) + kind = data.get("type") + if kind not in ("user", "assistant"): + continue + if held is not None and kind == "user" and not _is_tool_result_line(data) and held[1] not in seen: + seen.add(held[1]) + yield held + held = None + text = _text_content(data) if kind == "assistant" else "" + if text.strip(): + gaps = stop_gaps(text, tracker.state()) if is_wait_reply(text) else None + held = (index, text, gaps or None, gaps == []) + if held is not None and held[1] not in seen: + yield held + + def decide_stop(payload: dict) -> str | None: """Return blocking feedback for the Stop event, or None to let the turn finish.""" if payload.get("stop_hook_active"): diff --git a/engine/hooks/wait-needs-wakeup/tests/test_hooks.py b/engine/hooks/wait-needs-wakeup/tests/test_hooks.py index ad5c3c3c..4e9bfdbb 100644 --- a/engine/hooks/wait-needs-wakeup/tests/test_hooks.py +++ b/engine/hooks/wait-needs-wakeup/tests/test_hooks.py @@ -26,7 +26,6 @@ FIXTURES = os.path.join(HOOK_DIR, "tests", "fixtures") sys.path.insert(0, HOOK_DIR) -import backtest # noqa: E402 import claude_pretooluse # noqa: E402 import claude_stop_check # noqa: E402 import detect # noqa: E402 @@ -196,36 +195,57 @@ def test_fails_open_on_garbage_stdin(self): self.assertEqual(err.getvalue(), "") -class TestBacktestReproducesTheIncident(unittest.TestCase): - def test_backtest_counts_fires_fixtures_as_blocked_and_silent_as_zero(self): - counts = backtest.run_fixtures(FIXTURES) - self.assertEqual(counts["poll_commands_fires.json"]["blocked"], counts["poll_commands_fires.json"]["total"]) - self.assertEqual(counts["poll_commands_silent.json"]["blocked"], 0) - self.assertEqual(counts["wait_replies_fires.json"]["blocked"], counts["wait_replies_fires.json"]["total"]) - self.assertEqual(counts["wait_replies_silent.json"]["blocked"], 0) +def assistant_text(text): + return {"type": "assistant", "message": {"role": "assistant", "content": [{"type": "text", "text": text}]}} - def test_backtest_flags_poll_and_wait_reply_in_a_synthetic_transcript(self): - lines = [ + +def replay(lines): + return list(detect.replay_stop(enumerate(lines))) + + +class TestBacktestDetectors(unittest.TestCase): + def test_pretooluse_reason_blocks_foreground_poll(self): + reason = detect.pretooluse_reason({ + "tool_name": "Bash", + "tool_input": {"command": "until grep -q '^exit=' out; do sleep 3; done"}, + }) + self.assertEqual(reason, "foreground until loop sleeps while checking a status") + + def test_pretooluse_reason_silent_on_other_tools(self): + self.assertIsNone(detect.pretooluse_reason({"tool_name": "Read", "tool_input": {"command": "sleep 90"}})) + + def test_replay_stop_blocks_final_wait_reply_without_eta_or_wakeup(self): + units = replay([ {"type": "user", "message": {"role": "user", "content": "land it"}}, + assistant_text("Checking the queue."), {"type": "assistant", "message": {"role": "assistant", "content": [ - {"type": "tool_use", "id": "t1", "name": "Bash", - "input": {"command": "until grep -q '^exit=' out; do sleep 3; done"}}]}}, + {"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": "gh pr view 1"}}]}}, {"type": "user", "message": {"role": "user", "content": [ - {"type": "tool_result", "tool_use_id": "t1", "content": "exit=0"}]}}, - {"type": "assistant", "message": {"role": "assistant", "content": [ - {"type": "text", "text": "A watcher will report all three. Nothing needed from you."}]}}, + {"type": "tool_result", "tool_use_id": "t1", "content": "OPEN"}]}}, + assistant_text("A watcher will report all three. Nothing needed from you."), {"type": "user", "message": {"role": "user", "content": "ok"}}, - ] - path = transcript_file(lines) - try: - report = backtest.run_transcript(path) - finally: - os.unlink(path) - self.assertEqual(report["bash_commands"], 1) - self.assertEqual(report["poll_blocked"], 1) - self.assertEqual(report["final_replies"], 1) - self.assertEqual(report["wait_replies"], 1) - self.assertEqual(report["reply_blocked"], 1) + ]) + self.assertEqual(len(units), 1) + key, text, gaps, near = units[0] + self.assertEqual(key, 4) + self.assertEqual(len(gaps), 2) + self.assertFalse(near) + + def test_replay_stop_counts_wait_reply_with_eta_and_wakeup_as_near_miss_not_hit(self): + units = replay([ + {"type": "user", "message": {"role": "user", "content": "land it"}}, + {"type": "assistant", "message": {"role": "assistant", "content": [ + {"type": "tool_use", "id": "w1", "name": "ScheduleWakeup", "input": {"delaySeconds": 600}}]}}, + assistant_text("Nothing needed from you until then; back at 07:26 UTC."), + ]) + self.assertEqual(units, [(2, "Nothing needed from you until then; back at 07:26 UTC.", None, True)]) + + def test_replay_stop_agrees_with_decide_stop_on_every_fixture(self): + for name, expect_hit in (("wait_replies_fires.json", True), ("wait_replies_silent.json", False)): + for case in load(name): + with self.subTest(fixture=name, label=case["label"]): + units = replay(list(case["transcript"]) + [assistant_text(case["reply"])]) + self.assertEqual(bool(units[-1][2]), expect_hit) if __name__ == "__main__": diff --git a/engine/skills/reflect/references/cost-audit.md b/engine/skills/reflect/references/cost-audit.md index 50eccb07..8faf2b75 100644 --- a/engine/skills/reflect/references/cost-audit.md +++ b/engine/skills/reflect/references/cost-audit.md @@ -30,7 +30,14 @@ Both `claude` and `omp` modes also emit `frustration-signals` — the mechanical They also emit `intervention-must-automate`: yes when a verbatim re-send fired, any intervention kind (`told-you`, `accusation`, `agent-blame`) appears ≥2 times, or ≥2 distinct intervention kinds appear in the session. One "I told you" is frustration only; the same class twice is FAIL and must route to `automate-me`. `/loop` polls and Stop-hook injection text are not the human complaining. -To re-run the reality check after tuning the detector, `python3 skills/reflect/scripts/backtest.py [--limit N] [--verbose]` sweeps the newest local Claude/OMP transcripts (or explicit paths) and prints one flagged/interruptions/kinds summary line per session — the committed, repeatable half of the backtest; the transcript data itself stays local. +To re-run the reality check after tuning the detector, use the shared runner `scripts/backtest_detector.py` from the repo root: + +``` +python3 scripts/backtest_detector.py --detector engine/skills/reflect/scripts/token_audit.py:replay_frustration --unit rows [--limit N] [--verbose] [paths...] +python3 scripts/backtest_detector.py --detector engine/skills/reflect/scripts/token_audit.py:replay_frustration --unit rows --compare main +``` + +It sweeps the newest N (default 5) local Claude/OMP transcripts, or explicit paths, and prints one hits/kinds summary line per session, then the totals, hit rate, and a sample of flagged messages. `--compare ` replays the same transcripts through the detector at that git revision and lists the messages the change newly flags and newly misses — what a widening change has to justify. The committed half is the runner and `replay_frustration()`; the transcript data itself stays local. ## Same-problem thrash diff --git a/engine/skills/reflect/scripts/backtest.py b/engine/skills/reflect/scripts/backtest.py deleted file mode 100755 index 7e74fff2..00000000 --- a/engine/skills/reflect/scripts/backtest.py +++ /dev/null @@ -1,113 +0,0 @@ -#!/usr/bin/env python3 -"""Frustration-signal backtest runner. - -Runs token_audit's frustration detector against REAL local transcripts and -prints one summary line per session. The transcripts stay local — the -fixtures policy in cost-audit.md forbids committing real user transcripts — -so this runner is the committed, repeatable half of the backtest and the -data is whatever exists on the current machine. - -This is how the 2026-08-17 detector change was validated: 13/58 flagged on -the motivating OMP session (matching the hand audit) and 3 genuine -"you are thrashing" accusation hits on a prior 124MB Claude session. - -Usage: - python3 skills/reflect/scripts/backtest.py # newest 5 sessions per tool - python3 skills/reflect/scripts/backtest.py a.jsonl b.jsonl # explicit files - python3 skills/reflect/scripts/backtest.py --limit 10 --verbose -""" -import glob -import io -import json -import os -import sys -from contextlib import redirect_stdout - -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -import token_audit # noqa: E402 - - -def sniff_mode(path, max_lines=50): - """Claude Code lines carry type user/assistant; OMP lines carry - type=message with a role inside. First recognizable line wins.""" - try: - with open(path) as f: - for i, line in enumerate(f): - if i >= max_lines: - break - try: - d = json.loads(line) - except json.JSONDecodeError: - continue - t = d.get("type") - if t in ("user", "assistant"): - return "claude" - if t == "message": - return "omp" - except OSError: - return None - return None - - -def discover(limit): - claude = glob.glob(os.path.expanduser("~/.claude/projects/*/*.jsonl")) - omp = glob.glob(os.path.expanduser("~/.omp/agent/sessions/**/*.jsonl"), recursive=True) - tmp_iso = glob.glob(os.path.join(os.environ.get("TMPDIR", "/tmp"), "omp-agent-iso/*/sessions/*/*.jsonl")) - omp = [p for p in omp + tmp_iso if "merge-clones" not in p and "--private-tmp--" not in p] - found = [] - for group in (claude, omp): - group.sort(key=os.path.getmtime, reverse=True) - found.extend(group[:limit]) - return found - - -def summarize(path, verbose=False): - mode = sniff_mode(path) - if mode is None: - return f"SKIP (unrecognized format) {path}" - audit = token_audit.audit_claude if mode == "claude" else token_audit.audit_omp - try: - with redirect_stdout(io.StringIO()): - result = audit(path) - except Exception as e: # per-file fail-soft: one broken transcript shouldn't kill the sweep - return f"ERROR {os.path.basename(path)}: {e}" - fr = result["frustration"] - kinds = ",".join(f"{k}:{v}" for k, v in sorted(fr["kinds"].items(), key=lambda kv: -kv[1])) or "-" - peak = f" peak={fr['peak_window'][0]}..{fr['peak_window'][1]}" if fr["peak_window"] else "" - line = ( - f"{mode:6} {os.path.basename(path)[:52]:52} " - f"flagged={fr['count']}/{fr['n_user_messages']} " - f"interruptions={fr['interruptions']} kinds={kinds}{peak}" - ) - if verbose and fr["flagged"]: - line += "\n" + "\n".join( - f" [{f['index']}] {f['ts']} {f['kinds']}: {f['excerpt'][:70]!r}" - for f in fr["flagged"] - ) - return line - - -def main(argv): - args = list(argv[1:]) - verbose = "--verbose" in args - args = [a for a in args if a != "--verbose"] - limit = 5 - if "--limit" in args: - i = args.index("--limit") - try: - limit = int(args[i + 1]) - except (IndexError, ValueError): - print("--limit requires an integer", file=sys.stderr) - return 1 - del args[i:i + 2] - paths = args or discover(limit) - if not paths: - print("no transcripts found", file=sys.stderr) - return 1 - for p in paths: - print(summarize(p, verbose=verbose)) - return 0 - - -if __name__ == "__main__": - sys.exit(main(sys.argv)) diff --git a/engine/skills/reflect/scripts/tests/test_backtest.py b/engine/skills/reflect/scripts/tests/test_backtest.py deleted file mode 100755 index c593fe87..00000000 --- a/engine/skills/reflect/scripts/tests/test_backtest.py +++ /dev/null @@ -1,134 +0,0 @@ -#!/usr/bin/env python3 -"""Unit tests for backtest.py (frustration-signal backtest runner). - -Run: python3 -m unittest discover -s skills/reflect/scripts/tests -v -Synthetic fixtures only — real transcripts never enter the repo. -""" -import io -import json -import os -import sys -import tempfile -import unittest -from contextlib import redirect_stdout - -SCRIPTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -sys.path.insert(0, SCRIPTS_DIR) - -import backtest # noqa: E402 - - -def write_jsonl(lines): - f = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) - for d in lines: - f.write(json.dumps(d) + "\n") - f.close() - return f.name - - -CLAUDE_LINES = [ - {"type": "user", "timestamp": "2026-08-18T02:30:00Z", - "message": {"role": "user", "content": "WHERE IS MY DIGITAL TWIN? WHAT THE FUCK IS GOING ON"}}, - {"type": "assistant", - "message": {"id": "m1", "model": "claude-sonnet-5", - "usage": {"input_tokens": 1, "output_tokens": 1}, - "content": [{"type": "text", "text": "ok"}]}}, -] - -OMP_LINES = [ - {"type": "message", "timestamp": "2026-08-18T02:30:00Z", - "message": {"role": "user", "content": [{"type": "text", "text": "calm question about the code"}]}}, - {"type": "message", - "message": {"role": "assistant", "usage": {"input": 1, "output": 1, "cacheRead": 0, "cacheWrite": 0}, - "content": [{"type": "text", "text": "ok"}]}}, -] - - -class TestSniffMode(unittest.TestCase): - def test_sniffs_claude(self): - path = write_jsonl(CLAUDE_LINES) - try: - self.assertEqual(backtest.sniff_mode(path), "claude") - finally: - os.unlink(path) - - def test_sniffs_omp(self): - path = write_jsonl(OMP_LINES) - try: - self.assertEqual(backtest.sniff_mode(path), "omp") - finally: - os.unlink(path) - - def test_garbage_is_unrecognized(self): - f = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) - f.write("not json at all\n{\"type\": \"other\"}\n") - f.close() - try: - self.assertIsNone(backtest.sniff_mode(f.name)) - finally: - os.unlink(f.name) - - def test_missing_file_is_none(self): - self.assertIsNone(backtest.sniff_mode("/nonexistent.jsonl")) - - -class TestSummarize(unittest.TestCase): - def test_claude_summary_counts_flagged(self): - path = write_jsonl(CLAUDE_LINES) - try: - line = backtest.summarize(path) - self.assertIn("claude", line) - self.assertIn("flagged=1/1", line) - self.assertIn("allcaps", line) - finally: - os.unlink(path) - - def test_omp_summary_zero_flagged(self): - path = write_jsonl(OMP_LINES) - try: - line = backtest.summarize(path) - self.assertIn("omp", line) - self.assertIn("flagged=0/1", line) - finally: - os.unlink(path) - - def test_verbose_includes_excerpts(self): - path = write_jsonl(CLAUDE_LINES) - try: - line = backtest.summarize(path, verbose=True) - self.assertIn("WHERE IS MY DIGITAL TWIN", line) - finally: - os.unlink(path) - - def test_unrecognized_file_is_skipped_not_fatal(self): - f = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) - f.write("garbage\n") - f.close() - try: - self.assertTrue(backtest.summarize(f.name).startswith("SKIP")) - finally: - os.unlink(f.name) - - -class TestMain(unittest.TestCase): - def test_explicit_paths_print_one_line_each(self): - p1 = write_jsonl(CLAUDE_LINES) - p2 = write_jsonl(OMP_LINES) - buf = io.StringIO() - try: - with redirect_stdout(buf): - code = backtest.main(["backtest.py", p1, p2]) - self.assertEqual(code, 0) - out = buf.getvalue().strip().splitlines() - self.assertEqual(len(out), 2) - finally: - os.unlink(p1) - os.unlink(p2) - - def test_bad_limit_errors(self): - code = backtest.main(["backtest.py", "--limit", "nope"]) - self.assertEqual(code, 1) - - -if __name__ == "__main__": - unittest.main() diff --git a/engine/skills/reflect/scripts/tests/test_token_audit.py b/engine/skills/reflect/scripts/tests/test_token_audit.py index 9944f3bf..7b9fcb06 100644 --- a/engine/skills/reflect/scripts/tests/test_token_audit.py +++ b/engine/skills/reflect/scripts/tests/test_token_audit.py @@ -1587,6 +1587,50 @@ def test_omp_counts_interruptions_and_writes_out_report(self): os.unlink(out) +class TestReplayFrustration(unittest.TestCase): + def replay(self, rows): + return list(token_audit.replay_frustration(enumerate(rows))) + + def test_replay_flags_claude_human_messages_like_the_audit(self): + rows = [ + {"type": "user", "timestamp": "2026-08-18T02:30:00Z", + "message": {"role": "user", "content": "WHERE IS MY DIGITAL TWIN? WHAT THE FUCK IS GOING ON"}}, + {"type": "user", "message": {"role": "user", "content": "you are thrashing"}}, + {"type": "user", "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t1", "content": "you are ignoring me"}]}}, + {"type": "user", "timestamp": "2026-08-18T02:31:00Z", + "message": {"role": "user", "content": "calm question about the code"}}, + ] + units = self.replay(rows) + self.assertEqual([u[0] for u in units], [0, 3]) + self.assertEqual(units[0][2], ["allcaps", "profanity"]) + self.assertIsNone(units[1][2]) + path = write_jsonl(rows) + try: + with redirect_stdout(io.StringIO()): + audited = token_audit.audit_claude(path, include_subagents=False)["frustration"] + finally: + os.unlink(path) + self.assertEqual((audited["count"], audited["n_user_messages"]), (1, len(units))) + + def test_replay_flags_verbatim_repeat_but_not_after_an_api_error(self): + ask = {"type": "user", "timestamp": "2026-08-18T02:30:00Z", + "message": {"role": "user", "content": "please rerun the migration now"}} + again = dict(ask, timestamp="2026-08-18T02:31:00Z") + api_error = {"type": "assistant", "isApiErrorMessage": True, "error": "authentication_failed", + "message": {"role": "assistant", "content": [{"type": "text", "text": "Login expired"}]}} + self.assertEqual(self.replay([ask, again])[1][2], ["verbatim-repeat"]) + self.assertIsNone(self.replay([ask, api_error, again])[1][2]) + + def test_replay_reads_omp_user_messages_and_stays_silent_on_calm_text(self): + rows = [ + {"type": "message", "timestamp": "2026-08-18T02:30:00Z", + "message": {"role": "user", "content": [{"type": "text", "text": "calm question about the code"}]}}, + omp_assistant_line([{"type": "text", "text": "ok"}]), + ] + self.assertEqual(self.replay(rows), [(0, "calm question about the code", None)]) + + class TestSelfRetractionFlag(unittest.TestCase): def test_self_retraction_flag_yes_on_admission(self): u = { diff --git a/engine/skills/reflect/scripts/token_audit.py b/engine/skills/reflect/scripts/token_audit.py index 0004d8f0..55ab8922 100644 --- a/engine/skills/reflect/scripts/token_audit.py +++ b/engine/skills/reflect/scripts/token_audit.py @@ -349,6 +349,65 @@ def intervention_must_automate(frustration): return yes, count, rationale +def _omp_user_text(row): + if row.get("type") != "message": + return "" + msg = row.get("message") or {} + if msg.get("role") != "user": + return "" + content = msg.get("content") + if isinstance(content, str): + return content + return "\n".join( + b.get("text", "") for b in (content or []) + if isinstance(b, dict) and b.get("type") == "text" + ) + + +def replay_frustration(rows): + """Rows detector for scripts/backtest_detector.py: every human message of a + Claude or OMP transcript with its frustration kinds, from the same + human-message filter and frustration_signals() the audit uses.""" + failed_turn_indices = [] + omp_msgs = [] + + def claude_rows(): + for position, (_, row) in enumerate(rows): + if _is_api_error_line(row): + failed_turn_indices.append(position) + text = _omp_user_text(row) + if text.strip(): + omp_msgs.append((position, row.get("timestamp"), text)) + yield row + + user_msgs = [ + (utterance.index, utterance.timestamp, utterance.text) + for utterance in transcript_provenance.direct_human_claude_rows( + claude_rows(), include_queue_operations=True, + ) + ] + omp_msgs + frustration = frustration_signals(user_msgs, failed_turn_indices=failed_turn_indices) + kinds = {f["index"]: f["kinds"] for f in frustration["flagged"]} + for index, _, text in user_msgs: + if text.strip(): + yield index, text, kinds.get(index) + + +def _load_wrong_check_detect(): + """Load engine/hooks/wrong-check-reflect/detect.py without polluting sys.path.""" + here = os.path.dirname(os.path.abspath(__file__)) + # scripts -> reflect -> skills -> engine -> repo + detect_path = os.path.normpath( + os.path.join(here, "..", "..", "..", "hooks", "wrong-check-reflect", "detect.py") + ) + spec = importlib.util.spec_from_file_location("wrong_check_reflect_detect", detect_path) + if spec is None or spec.loader is None: + return None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def self_retraction_hits(assistant_texts): """Assistant-only first-person wrong-check admissions, scanned offline. @@ -1225,14 +1284,7 @@ def audit_omp(path, out_path=None): if d.get("type") == "message": msg = d.get("message", {}) if msg.get("role") == "user": - content = msg.get("content") - if isinstance(content, str): - text = content - else: - text = "\n".join( - b.get("text", "") for b in (content or []) - if isinstance(b, dict) and b.get("type") == "text" - ) + text = _omp_user_text(d) if text.strip(): user_msgs.append((len(user_msgs), d.get("timestamp"), text)) if msg.get("role") == "toolResult": diff --git a/engine/skills/reflect/scripts/transcript_provenance.py b/engine/skills/reflect/scripts/transcript_provenance.py index d4c5bc38..54cc53f0 100644 --- a/engine/skills/reflect/scripts/transcript_provenance.py +++ b/engine/skills/reflect/scripts/transcript_provenance.py @@ -325,3 +325,14 @@ def direct_human_utterances( for utterance in extract_utterances(path, harness, include_queue_operations=include_queue_operations) if utterance.can_trigger_intervention ] + + +def direct_human_claude_rows( + rows: Iterable[dict[str, Any]], path: str = "", *, include_queue_operations: bool = False, +) -> list[HumanUtterance]: + """Same as direct_human_utterances for Claude, over rows streamed in file order.""" + return [ + utterance + for utterance in _claude_utterances(path, rows, include_queue_operations=include_queue_operations) + if utterance.can_trigger_intervention + ] diff --git a/scripts/backtest_detector.py b/scripts/backtest_detector.py new file mode 100644 index 00000000..d9fbc219 --- /dev/null +++ b/scripts/backtest_detector.py @@ -0,0 +1,578 @@ +#!/usr/bin/env python3 +"""Replay one detector over real local transcripts and report what it catches. + + python3 scripts/backtest_detector.py --detector PATH:CALLABLE [--unit UNIT] + [--limit N] [--verbose] [--json OUT] [TRANSCRIPT_OR_DIR ...] + python3 scripts/backtest_detector.py --detector PATH:CALLABLE --compare REF + python3 scripts/backtest_detector.py --detector PATH:CALLABLE --compare OTHER_PATH[:CALLABLE] + +Units, and the call made for each: + assistant CALLABLE(text) for every assistant text block + final CALLABLE(text) for every turn-ending assistant reply + user CALLABLE(text) for every human-typed user message + tool CALLABLE({"tool_name", "tool_input"}) for every tool call; --tool narrows + rows CALLABLE(rows) once per session, rows yielding (line_index, row); it yields + (key, text, verdict) or (key, text, verdict, near) per unit it judged + +A truthy verdict is a hit. With no paths, the newest --limit sessions per source +(Claude Code, OMP) are scanned. Files are streamed and never leave this machine. +""" +from __future__ import annotations + +import argparse +import glob +import importlib.util +import json +import os +import random +import re +import shutil +import subprocess +import sys +import tempfile +from collections import Counter + +UNITS = ("assistant", "final", "user", "tool", "rows") +TEXT_UNITS = ("assistant", "final", "user") +SNIFF_LINES = 50 +EXCERPT_WIDTH = 160 +LABEL_WIDTH = 40 +SYSTEM_PREFIXES = ( + "<", + "[Request interrupted", + "This session is being continued", + "Base directory for this skill", + "[IMPORTANT: User invoked", +) +PARAGRAPH_RE = re.compile(r"\n\s*\n") + + +class Detector: + def __init__(self, fn, label: str): + self.fn = fn + self.label = label + + +def parse_spec(spec: str) -> tuple[str, str]: + path, sep, name = spec.rpartition(":") + if not sep or not path or not name: + raise ValueError(f"detector must be PATH:CALLABLE, got {spec!r}") + return path, name + + +def _sibling_names(directory: str) -> set[str]: + names = set() + for entry in os.listdir(directory): + stem, ext = os.path.splitext(entry) + if ext == ".py" or os.path.isdir(os.path.join(directory, entry)): + names.add(stem) + return names + + +def load_callable(path: str, name: str, label: str | None = None) -> Detector: + full = os.path.abspath(path) + if not os.path.isfile(full): + raise ValueError(f"detector file not found: {label or path}") + directory = os.path.dirname(full) + siblings = _sibling_names(directory) + stashed = {n: sys.modules.pop(n) for n in siblings if n in sys.modules} + module_name = f"backtest_detector_{abs(hash((full, label)))}" + spec = importlib.util.spec_from_file_location(module_name, full) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + sys.path.insert(0, directory) + try: + spec.loader.exec_module(module) + finally: + sys.path.remove(directory) + for n in siblings: + sys.modules.pop(n, None) + sys.modules.update(stashed) + fn = getattr(module, name, None) + if not callable(fn): + raise ValueError(f"{label or path} has no callable {name!r}") + return Detector(fn, f"{label or path}:{name}") + + +def _git(cwd: str, *args: str) -> subprocess.CompletedProcess: + return subprocess.run(["git", "-C", cwd, *args], capture_output=True) + + +def materialize_revision(path: str, ref: str) -> tuple[str, str]: + full = os.path.realpath(path) + top = _git(os.path.dirname(full), "rev-parse", "--show-toplevel") + if top.returncode != 0: + raise ValueError(f"{path} is not inside a git repository, so {ref!r} cannot be resolved") + root = os.path.realpath(top.stdout.decode().strip()) + if _git(root, "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}").returncode != 0: + raise ValueError(f"{ref!r} is neither a detector path nor a git revision") + rel = os.path.relpath(full, root) + archive = _git(root, "archive", "--format=tar", ref) + if archive.returncode != 0: + raise ValueError(f"git archive {ref} failed: {archive.stderr.decode().strip()}") + tmp = tempfile.mkdtemp(prefix="backtest-detector-") + subprocess.run(["tar", "-x", "-C", tmp], input=archive.stdout, check=True) + return tmp, os.path.join(tmp, rel) + + +def resolve_baseline(candidate_path: str, candidate_name: str, compare: str, tmpdirs: list) -> Detector: + if os.path.isfile(compare): + return load_callable(compare, candidate_name) + head, sep, tail = compare.rpartition(":") + if sep and os.path.isfile(head): + return load_callable(head, tail) + tmp, path = materialize_revision(candidate_path, compare) + tmpdirs.append(tmp) + return load_callable(path, candidate_name, label=f"{candidate_path}@{compare}") + + +def discover(limit: int) -> list[str]: + home = os.path.expanduser("~") + claude = glob.glob(os.path.join(home, ".claude", "projects", "*", "*.jsonl")) + omp = glob.glob(os.path.join(home, ".omp", "agent", "sessions", "**", "*.jsonl"), recursive=True) + omp += glob.glob(os.path.join(os.environ.get("TMPDIR", "/tmp"), "omp-agent-iso", "*", "sessions", "*", "*.jsonl")) + omp = [p for p in omp if "merge-clones" not in p and "--private-tmp--" not in p] + found = [] + for group in (claude, omp): + group.sort(key=os.path.getmtime, reverse=True) + found.extend(group[:limit]) + return found + + +def expand(paths: list[str]) -> list[str]: + out = [] + for p in paths: + out.extend(sorted(glob.glob(os.path.join(p, "*.jsonl"))) if os.path.isdir(p) else [p]) + return out + + +def iter_rows(path: str): + with open(path, encoding="utf-8", errors="ignore") as handle: + for index, line in enumerate(handle): + try: + row = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(row, dict): + yield index, row + + +def sniff_format(path: str) -> str | None: + try: + with open(path, encoding="utf-8", errors="ignore") as handle: + for i, line in enumerate(handle): + if i >= SNIFF_LINES: + break + try: + kind = json.loads(line).get("type") + except (json.JSONDecodeError, AttributeError): + continue + if kind in ("user", "assistant"): + return "claude" + if kind == "message": + return "omp" + except OSError: + return None + return None + + +def _content_text(content) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join( + b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text" + ) + return "" + + +def shape(row: dict, fmt: str) -> tuple[str | None, str, list]: + if fmt == "omp": + if row.get("type") != "message": + return None, "", [] + msg = row.get("message") or {} + role = msg.get("role") + content = msg.get("content") + if role == "toolResult": + return "result", "", [] + if role not in ("user", "assistant"): + return None, "", [] + tools = [ + (b.get("name"), b.get("arguments") or {}) + for b in (content if isinstance(content, list) else []) + if isinstance(b, dict) and b.get("type") == "toolCall" + ] + return role, _content_text(content), tools + kind = row.get("type") + if kind not in ("user", "assistant"): + return None, "", [] + content = (row.get("message") or {}).get("content") + blocks = content if isinstance(content, list) else [] + if kind == "user" and any(isinstance(b, dict) and b.get("type") == "tool_result" for b in blocks): + return "result", "", [] + tools = [ + (b.get("name"), b.get("input") or {}) + for b in blocks + if isinstance(b, dict) and b.get("type") == "tool_use" + ] + return kind, _content_text(content), tools + + +def _is_human(row: dict, text: str) -> bool: + stripped = text.strip() + if not stripped or row.get("isMeta") or row.get("isSidechain"): + return False + return not stripped.startswith(SYSTEM_PREFIXES) + + +def _tool_text(name, tool_input) -> str: + command = tool_input.get("command") if isinstance(tool_input, dict) else None + if isinstance(command, str): + return command + return f"{name} {json.dumps(tool_input, sort_keys=True, default=str)}" + + +def _once(unit, seen: set[int]): + digest = hash(unit[1]) + if digest not in seen: + seen.add(digest) + yield unit + + +def _final_units(rows, fmt: str): + held = None + seen: set[int] = set() + for index, row in rows: + role, text, _ = shape(row, fmt) + if role is None: + continue + if held is not None and role == "user": + yield from _once(held, seen) + held = (index, text, (text,)) if role == "assistant" and text.strip() else None + if held is not None: + yield from _once(held, seen) + + +def units(rows, fmt: str, unit: str, tools: tuple[str, ...]): + if unit == "final": + yield from _final_units(rows, fmt) + return + for index, row in rows: + role, text, calls = shape(row, fmt) + if unit == "assistant" and role == "assistant" and text.strip(): + yield index, text, (text,) + elif unit == "user" and role == "user" and _is_human(row, text): + yield index, text, (text,) + elif unit == "tool" and role == "assistant": + for offset, (name, tool_input) in enumerate(calls): + if tools and name not in tools: + continue + payload = {"hook_event_name": "PreToolUse", "tool_name": name, "tool_input": tool_input} + yield f"{index}.{offset}", _tool_text(name, tool_input), (payload,) + + +def _paragraph_probe(fn, text: str): + paragraphs = [p for p in PARAGRAPH_RE.split(text) if p.strip()] + if len(paragraphs) < 2: + return None + for para in paragraphs: + verdict = fn(para) + if verdict: + return verdict + return None + + +def judgments(detector: Detector, path: str, fmt: str, unit: str, tools: tuple[str, ...], near: Detector | None): + if unit == "rows": + for item in detector.fn(iter_rows(path)): + if len(item) == 4: + yield item + else: + key, text, verdict = item + yield key, text, verdict, None + return + for key, text, call_args in units(iter_rows(path), fmt, unit, tools): + verdict = detector.fn(*call_args) + near_verdict = None + if not verdict: + if near is not None: + near_verdict = near.fn(*call_args) + elif unit in TEXT_UNITS: + near_verdict = _paragraph_probe(detector.fn, text) + yield key, text, verdict, near_verdict + + +def labels(verdict) -> list[str]: + if isinstance(verdict, dict): + return [f"{k}={v}" for k, v in verdict.items() if isinstance(v, (str, bool))] + if isinstance(verdict, (list, tuple, set, frozenset)): + return [str(v) for v in verdict] + if verdict is True: + return [] + return [str(verdict)] + + +def amounts(verdict) -> dict: + if not isinstance(verdict, dict): + return {} + return {k: v for k, v in verdict.items() if isinstance(v, (int, float)) and not isinstance(v, bool)} + + +def _clip(text: str, width: int) -> str: + return text if len(text) <= width else text[: width - 1] + "…" + + +def tags(verdict) -> list[str]: + return [_clip(tag, LABEL_WIDTH) for tag in labels(verdict)] + + +def excerpt(text: str, verdict=None, width: int = EXCERPT_WIDTH) -> str: + flat = " ".join(str(text).split()) + anchor = next((tag for tag in labels(verdict) if tag and tag.lower() in flat.lower()), "") if verdict else "" + at = flat.lower().find(anchor.lower()) if anchor else -1 + start = max(0, min(at - width // 3, len(flat) - width)) if at > 0 else 0 + clip = flat[start:start + width] + return ("…" if start else "") + clip + ("…" if start + width < len(flat) else "") + + +class Sample: + def __init__(self, size: int, rng: random.Random): + self.size = size + self.rng = rng + self.items: list = [] + self.seen = 0 + + def add(self, item) -> None: + self.seen += 1 + if len(self.items) < self.size: + self.items.append(item) + return + slot = self.rng.randrange(self.seen) + if slot < self.size: + self.items[slot] = item + + +def _kinds(counter: Counter, top: int) -> str: + shown = ",".join(f"{k}:{v}" for k, v in counter.most_common(top)) or "-" + rest = len(counter) - top + return f"{shown},+{rest} more" if rest > 0 else shown + + +def _amounts_text(counter: Counter) -> str: + return "".join(f" {k}={v:g}" for k, v in sorted(counter.items())) + + +def _rate(hits: int, scanned: int) -> str: + return f"{100.0 * hits / scanned:.2f}%" if scanned else "n/a" + + +def _print_sample(title: str, sample: Sample) -> None: + print(f"{title} ({len(sample.items)} of {sample.seen}):") + for session, key, label, text in sample.items: + print(f" {session}:{key} [{label}] {text}") + + +def run_single(detector: Detector, near: Detector | None, files: list[str], args) -> tuple[int, dict]: + rng = random.Random(0) + hit_sample, near_sample = Sample(args.samples, rng), Sample(args.samples, rng) + totals = {"sessions": 0, "unchecked": 0, "errors": 0, "scanned": 0, "hits": 0, "near_misses": 0} + kinds, sums = Counter(), Counter() + report = [] + for path in files: + totals["sessions"] += 1 + name = os.path.basename(path) + fmt = sniff_format(path) + if fmt is None: + totals["unchecked"] += 1 + print(f"SKIP (unrecognized format) {path}") + report.append({"path": path, "status": "unrecognized"}) + continue + scanned = hits = near_count = 0 + s_kinds, s_sums = Counter(), Counter() + s_hits, detail = [], [] + try: + for key, text, verdict, near_verdict in judgments(detector, path, fmt, args.unit, args.tool, near): + scanned += 1 + if verdict: + hits += 1 + hit_tags = tags(verdict) + s_kinds.update(hit_tags) + s_sums.update(amounts(verdict)) + label = ",".join(hit_tags) or "hit" + hit_sample.add((name[:12], key, label, excerpt(text, verdict))) + if args.verbose: + detail.append(f" [{key}] {label}: {excerpt(text, verdict, 90)!r}") + if args.json_out: + s_hits.append({"key": key, "verdict": verdict, "excerpt": excerpt(text, verdict)}) + elif near_verdict: + near_count += 1 + near_label = ",".join(tags(near_verdict)) or "near" + near_sample.add((name[:12], key, near_label, excerpt(text, near_verdict))) + except Exception as exc: + totals["unchecked"] += 1 + totals["errors"] += 1 + print(f"ERROR {name}: {type(exc).__name__}: {exc}") + report.append({"path": path, "status": "error", "error": f"{type(exc).__name__}: {exc}"}) + continue + totals["scanned"] += scanned + totals["hits"] += hits + totals["near_misses"] += near_count + kinds.update(s_kinds) + sums.update(s_sums) + print(f"{fmt:6} {name[:52]:52} hits={hits}/{scanned} kinds={_kinds(s_kinds, 5)}{_amounts_text(s_sums)}") + for line in detail: + print(line) + report.append({"path": path, "status": "ok", "format": fmt, "scanned": scanned, "hits": hits, + "near_misses": near_count, "kinds": dict(s_kinds), "amounts": dict(s_sums), "hit_units": s_hits}) + print() + print(f"detector: {detector.label} unit={args.unit}") + print( + f"sessions={totals['sessions']} unchecked={totals['unchecked']} scanned={totals['scanned']} " + f"hits={totals['hits']} hit_rate={_rate(totals['hits'], totals['scanned'])} " + f"near_misses={totals['near_misses']}" + ) + print(f"kinds: {_kinds(kinds, 12)}{_amounts_text(sums)}") + _print_sample("hits", hit_sample) + _print_sample("near-misses", near_sample) + totals.update({"kinds": dict(kinds), "amounts": dict(sums)}) + return _exit_code(totals), {"detector": detector.label, "unit": args.unit, "totals": totals, "sessions": report} + + +def _collect(detector: Detector, path: str, fmt: str, args) -> tuple[int, dict]: + scanned = 0 + hits = {} + for key, text, verdict, _ in judgments(detector, path, fmt, args.unit, args.tool, None): + scanned += 1 + if verdict: + label = ",".join(tags(verdict)) or "hit" + hits[key] = (label, excerpt(text, verdict)) + return scanned, hits + + +def run_compare(candidate: Detector, baseline: Detector, files: list[str], args) -> tuple[int, dict]: + rng = random.Random(0) + caught_sample, missed_sample = Sample(args.samples, rng), Sample(args.samples, rng) + totals = {"sessions": 0, "unchecked": 0, "errors": 0, "scanned": 0, "baseline_scanned": 0, + "candidate_hits": 0, "baseline_hits": 0, "newly_caught": 0, "newly_missed": 0, "unchanged": 0} + report = [] + for path in files: + totals["sessions"] += 1 + name = os.path.basename(path) + fmt = sniff_format(path) + if fmt is None: + totals["unchecked"] += 1 + print(f"SKIP (unrecognized format) {path}") + report.append({"path": path, "status": "unrecognized"}) + continue + try: + scanned, new_hits = _collect(candidate, path, fmt, args) + base_scanned, old_hits = _collect(baseline, path, fmt, args) + except Exception as exc: + totals["unchecked"] += 1 + totals["errors"] += 1 + print(f"ERROR {name}: {type(exc).__name__}: {exc}") + report.append({"path": path, "status": "error", "error": f"{type(exc).__name__}: {exc}"}) + continue + caught = [k for k in new_hits if k not in old_hits] + missed = [k for k in old_hits if k not in new_hits] + same = len(new_hits) - len(caught) + for key in caught: + caught_sample.add((name[:12], key, *new_hits[key])) + for key in missed: + missed_sample.add((name[:12], key, *old_hits[key])) + totals["scanned"] += scanned + totals["baseline_scanned"] += base_scanned + totals["candidate_hits"] += len(new_hits) + totals["baseline_hits"] += len(old_hits) + totals["newly_caught"] += len(caught) + totals["newly_missed"] += len(missed) + totals["unchanged"] += same + print( + f"{fmt:6} {name[:52]:52} hits={len(new_hits)}/{scanned} was={len(old_hits)}/{base_scanned} " + f"caught={len(caught)} missed={len(missed)}" + ) + if args.verbose: + for key in caught: + print(f" + [{key}] {new_hits[key][0]}: {new_hits[key][1][:90]!r}") + for key in missed: + print(f" - [{key}] {old_hits[key][0]}: {old_hits[key][1][:90]!r}") + report.append({"path": path, "status": "ok", "format": fmt, "scanned": scanned, "baseline_scanned": base_scanned, + "candidate_hits": len(new_hits), "baseline_hits": len(old_hits), + "newly_caught": [{"key": k, "label": new_hits[k][0], "excerpt": new_hits[k][1]} for k in caught], + "newly_missed": [{"key": k, "label": old_hits[k][0], "excerpt": old_hits[k][1]} for k in missed], + "unchanged": same}) + print() + print(f"candidate: {candidate.label} unit={args.unit}") + print(f"baseline: {baseline.label}") + scanned_text = str(totals["scanned"]) + if totals["baseline_scanned"] != totals["scanned"]: + scanned_text += f" (baseline {totals['baseline_scanned']})" + print( + f"sessions={totals['sessions']} unchecked={totals['unchecked']} scanned={scanned_text} " + f"candidate_hits={totals['candidate_hits']} ({_rate(totals['candidate_hits'], totals['scanned'])}) " + f"baseline_hits={totals['baseline_hits']} ({_rate(totals['baseline_hits'], totals['baseline_scanned'])})" + ) + print(f"newly_caught={totals['newly_caught']} newly_missed={totals['newly_missed']} unchanged={totals['unchanged']}") + _print_sample("newly caught", caught_sample) + _print_sample("newly missed", missed_sample) + return _exit_code(totals), {"candidate": candidate.label, "baseline": baseline.label, "unit": args.unit, + "totals": totals, "sessions": report} + + +def _exit_code(totals: dict) -> int: + if totals["errors"] or totals["unchecked"] == totals["sessions"]: + return 1 + return 0 + + +def build_parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("paths", nargs="*", help="transcripts or directories of *.jsonl; overrides discovery") + ap.add_argument("--detector", required=True, help="PATH:CALLABLE") + ap.add_argument("--unit", choices=UNITS, default="assistant") + ap.add_argument("--tool", action="append", default=[], help="tool name to keep for --unit tool (repeatable)") + ap.add_argument("--near", help="PATH:CALLABLE whose hits on a miss count as a near-miss") + ap.add_argument("--compare", help="baseline: a git revision of the same file, or PATH[:CALLABLE]") + ap.add_argument("--limit", type=int, default=5, help="newest sessions per source when discovering (default 5)") + ap.add_argument("--samples", type=int, default=10, help="sample size for each printed list (default 10)") + ap.add_argument("--verbose", action="store_true", help="list every hit under its session line") + ap.add_argument("--json", dest="json_out", help="write the full report as JSON") + return ap + + +def main(argv: list[str] | None = None) -> int: + ap = build_parser() + args = ap.parse_args(argv) + args.tool = tuple(args.tool) + if args.unit == "rows" and args.near: + ap.error("--near applies to per-unit detectors; a rows detector yields its own near flag") + if args.compare and args.near: + ap.error("--near is not used with --compare") + tmpdirs: list[str] = [] + try: + try: + path, name = parse_spec(args.detector) + candidate = load_callable(path, name) + near = load_callable(*parse_spec(args.near)) if args.near else None + baseline = resolve_baseline(path, name, args.compare, tmpdirs) if args.compare else None + except ValueError as exc: + ap.error(str(exc)) + files = expand(args.paths) if args.paths else discover(args.limit) + if not files: + print("no transcripts found", file=sys.stderr) + return 1 + if baseline is None: + code, report = run_single(candidate, near, files, args) + else: + code, report = run_compare(candidate, baseline, files, args) + finally: + for tmp in tmpdirs: + shutil.rmtree(tmp, ignore_errors=True) + if args.json_out: + with open(args.json_out, "w", encoding="utf-8") as handle: + json.dump(report, handle, indent=1, default=str) + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_backtest_detector.py b/tests/test_backtest_detector.py new file mode 100644 index 00000000..a2f6e13d --- /dev/null +++ b/tests/test_backtest_detector.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +"""Tests for scripts/backtest_detector.py. Transcripts here are synthetic and built inline.""" +from __future__ import annotations + +import io +import json +import os +import subprocess +import sys +import tempfile +import threading +import time +import types +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO / "scripts")) +import backtest_detector as bd # noqa: E402 +from git_test_repo import init_repo # noqa: E402 + +STOP_CHECK = str(REPO / "engine" / "hooks" / "diu-stop" / "claude_stop_check.py") + ":find_unverified_claim" + + +def human(text): + return {"type": "user", "message": {"role": "user", "content": text}} + + +def said(text): + return {"type": "assistant", "message": {"role": "assistant", "content": [{"type": "text", "text": text}]}} + + +def calls(tool_id, name, tool_input): + return {"type": "assistant", "message": {"role": "assistant", "content": [ + {"type": "tool_use", "id": tool_id, "name": name, "input": tool_input}]}} + + +def result(tool_id, text): + return {"type": "user", "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": tool_id, "content": text}]}} + + +class Case(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.tmp = self._tmp.name + + def tearDown(self): + self._tmp.cleanup() + + def transcript(self, rows, name="s.jsonl", folder=None): + directory = os.path.join(self.tmp, folder) if folder else self.tmp + os.makedirs(directory, exist_ok=True) + path = os.path.join(directory, name) + with open(path, "w", encoding="utf-8") as handle: + for row in rows: + handle.write((row if isinstance(row, str) else json.dumps(row)) + "\n") + return path + + def module(self, name, source): + path = os.path.join(self.tmp, name) + with open(path, "w", encoding="utf-8") as handle: + handle.write(source) + return path + + def run_main(self, *argv): + out = io.StringIO() + with redirect_stdout(out): + code = bd.main(list(argv)) + return code, out.getvalue() + + +class TestSingleDetector(Case): + def test_hits_unverified_claim_in_assistant_text_and_reports_rate(self): + path = self.transcript([ + human("why is the cache stale"), + said("Confirmed, the cache is stale."), + said("Here is the output:\n\n```\nok\n```"), + ]) + code, out = self.run_main("--detector", STOP_CHECK, path) + self.assertEqual(code, 0) + self.assertIn("hits=1/2 kinds=confirmed:1", out) + self.assertIn("scanned=2 hits=1 hit_rate=50.00%", out) + self.assertIn("[confirmed] Confirmed, the cache is stale.", out) + + def test_clean_transcript_reports_zero_hits(self): + path = self.transcript([human("hi"), said("Here is `ls` output.")]) + code, out = self.run_main("--detector", STOP_CHECK, path) + self.assertEqual(code, 0) + self.assertIn("scanned=1 hits=0 hit_rate=0.00%", out) + + def test_paragraph_that_fires_alone_is_a_near_miss_not_a_hit(self): + path = self.transcript([said("Confirmed, the deploy is `live`.\n\n```\nexit 0\n```")]) + code, out = self.run_main("--detector", STOP_CHECK, path) + self.assertIn("hits=0 hit_rate=0.00% near_misses=1", out) + self.assertIn("[confirmed]", out.split("near-misses")[1]) + + def test_explicit_near_callable_replaces_the_paragraph_probe(self): + det = self.module("det.py", "def hit(text):\n return 'alpha' in text\n\ndef near(text):\n return 'alp' in text\n") + path = self.transcript([said("alpha"), said("alp"), said("zzz")]) + code, out = self.run_main("--detector", det + ":hit", "--near", det + ":near", path) + self.assertIn("scanned=3 hits=1 hit_rate=33.33% near_misses=1", out) + + def test_final_unit_does_not_count_text_followed_by_a_tool_call(self): + det = self.module("det.py", "def hit(text):\n return text\n") + path = self.transcript([ + human("go"), + said("Checking."), + calls("t1", "Bash", {"command": "ls"}), + result("t1", "a"), + said("Done."), + human("thanks"), + said("Done."), + said("Anything else?"), + ]) + code, out = self.run_main("--detector", det + ":hit", "--unit", "final", "--verbose", path) + self.assertIn("hits=2/2", out) + self.assertIn("[4] Done.", out) + self.assertIn("[7] Anything else?", out) + self.assertNotIn("Checking.", out.split("detector:")[0]) + + def test_user_unit_skips_tool_results_and_injected_text(self): + det = self.module("det.py", "def hit(text):\n return text\n") + path = self.transcript([ + human("real ask"), + human("done"), + result("t1", "tool output"), + {"type": "user", "isMeta": True, "message": {"role": "user", "content": "meta"}}, + human("[Request interrupted by user]"), + ]) + code, out = self.run_main("--detector", det + ":hit", "--unit", "user", "--verbose", path) + self.assertIn("hits=1/1", out) + self.assertIn("[0] real ask", out) + + def test_tool_unit_passes_hook_payload_and_filters_by_tool(self): + det = self.module("det.py", "def hit(payload):\n return 'sleep' in payload['tool_input'].get('command', '') and payload['tool_name']\n") + path = self.transcript([ + calls("t1", "Bash", {"command": "sleep 90"}), + calls("t2", "Bash", {"command": "ls"}), + calls("t3", "Monitor", {"command": "sleep 90"}), + ]) + code, out = self.run_main("--detector", det + ":hit", "--unit", "tool", "--tool", "Bash", path) + self.assertIn("hits=1/2 kinds=Bash:1", out) + + def test_rows_unit_takes_verdict_amounts_and_near_flag_from_the_detector(self): + det = self.module("det.py", ( + "def replay(rows):\n" + " for index, row in rows:\n" + " text = row.get('text', '')\n" + " verdict = {'next_try': 'ok', 'saved': 2} if text == 'hit' else None\n" + " yield index, text, verdict, text == 'close'\n" + )) + path = self.transcript([{"type": "user", "text": "hit"}, {"type": "user", "text": "close"}, {"type": "user", "text": "far"}]) + code, out = self.run_main("--detector", det + ":replay", "--unit", "rows", path) + self.assertIn("hits=1/3 kinds=next_try=ok:1 saved=2", out) + self.assertIn("near_misses=1", out) + + def test_json_report_lists_each_hit(self): + path = self.transcript([said("Confirmed, it shipped.")]) + out_path = os.path.join(self.tmp, "report.json") + self.run_main("--detector", STOP_CHECK, "--json", out_path, path) + with open(out_path, encoding="utf-8") as handle: + report = json.load(handle) + self.assertEqual(report["totals"]["hits"], 1) + self.assertEqual(report["sessions"][0]["hit_units"][0]["verdict"], "confirmed") + + +class TestUncheckedIsNotClean(Case): + def test_unrecognized_file_is_skipped_counted_unchecked_and_fails_when_nothing_ran(self): + path = self.transcript(["not json", json.dumps({"type": "other"})]) + code, out = self.run_main("--detector", STOP_CHECK, path) + self.assertEqual(code, 1) + self.assertIn("SKIP (unrecognized format)", out) + self.assertIn("sessions=1 unchecked=1 scanned=0", out) + + def test_detector_exception_is_reported_unchecked_and_fails_the_run(self): + det = self.module("det.py", "def boom(text):\n raise RuntimeError('bad regex')\n") + good = self.transcript([said("hello")], name="a.jsonl") + code, out = self.run_main("--detector", det + ":boom", good) + self.assertEqual(code, 1) + self.assertIn("ERROR a.jsonl: RuntimeError: bad regex", out) + self.assertIn("unchecked=1", out) + + def test_spec_without_callable_is_refused(self): + with self.assertRaises(SystemExit) as caught, redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()): + bd.main(["--detector", "engine/hooks/diu-stop/claude_stop_check.py"]) + self.assertEqual(caught.exception.code, 2) + + def test_missing_callable_is_refused(self): + with self.assertRaises(SystemExit) as caught, patch("sys.stderr", io.StringIO()): + bd.main(["--detector", STOP_CHECK.rsplit(":", 1)[0] + ":no_such_fn", self.tmp]) + self.assertEqual(caught.exception.code, 2) + + +class TestTranscriptSelection(Case): + def test_discovery_takes_newest_limit_sessions(self): + home = os.path.join(self.tmp, "home") + paths = [] + for i, name in enumerate(("old.jsonl", "mid.jsonl", "new.jsonl")): + paths.append(self.transcript([said("x")], name=name, folder=os.path.join("home", ".claude", "projects", "p"))) + os.utime(paths[-1], (1000 + i, 1000 + i)) + det = self.module("det.py", "def hit(text):\n return False\n") + with patch.dict(os.environ, {"HOME": home, "TMPDIR": self.tmp}): + code, out = self.run_main("--detector", det + ":hit", "--limit", "2") + self.assertIn("new.jsonl", out) + self.assertIn("mid.jsonl", out) + self.assertNotIn("old.jsonl", out) + self.assertIn("sessions=2", out) + + def test_explicit_paths_override_discovery_and_directories_expand(self): + folder = os.path.dirname(self.transcript([said("x")], name="a.jsonl", folder="d")) + self.transcript([said("y")], name="b.jsonl", folder="d") + self.transcript([said("z")], name="c.jsonl") + det = self.module("det.py", "def hit(text):\n return False\n") + with patch.object(bd, "discover", side_effect=AssertionError("discovery ran")): + code, out = self.run_main("--detector", det + ":hit", folder) + self.assertIn("sessions=2", out) + self.assertNotIn("c.jsonl", out) + + def test_rows_are_streamed_not_loaded_whole(self): + fifo = os.path.join(self.tmp, "live.jsonl") + os.mkfifo(fifo) + first_seen = threading.Event() + rest_written_after_first = [] + + def writer(): + with open(fifo, "w", encoding="utf-8") as handle: + handle.write(json.dumps(said("one")) + "\n") + handle.flush() + rest_written_after_first.append(first_seen.wait(5)) + handle.write(json.dumps(said("two")) + "\n") + + thread = threading.Thread(target=writer) + thread.start() + rows = bd.iter_rows(fifo) + started = time.monotonic() + index, row = next(rows) + first_seen.set() + self.assertEqual((index, bd.shape(row, "claude")[1]), (0, "one")) + self.assertEqual(next(rows)[0], 1) + thread.join() + self.assertEqual(rest_written_after_first, [True]) + self.assertLess(time.monotonic() - started, 4) + + +class TestCompare(Case): + def test_compare_two_paths_reports_caught_missed_and_unchanged(self): + old = self.module("old.py", "def hit(text):\n return 'alpha' in text or 'beta' in text\n") + new = self.module("new.py", "def hit(text):\n return 'alpha' in text or 'gamma' in text\n") + path = self.transcript([said("alpha"), said("beta"), said("gamma"), said("delta")]) + code, out = self.run_main("--detector", new + ":hit", "--compare", old + ":hit", path) + self.assertEqual(code, 0) + self.assertIn("hits=2/4 was=2/4 caught=1 missed=1", out) + self.assertIn("newly_caught=1 newly_missed=1 unchanged=1", out) + self.assertIn("gamma", out.split("newly caught (")[1].split("newly missed")[0]) + self.assertIn("beta", out.split("newly missed (")[1]) + + def test_compare_against_git_ref_loads_that_revision_and_its_siblings(self): + repo = os.path.join(self.tmp, "repo") + os.makedirs(os.path.join(repo, "hook")) + init_repo(repo) + with open(os.path.join(repo, "hook", "words.py"), "w", encoding="utf-8") as handle: + handle.write("WORDS = ('alpha',)\n") + with open(os.path.join(repo, "hook", "detect.py"), "w", encoding="utf-8") as handle: + handle.write("from words import WORDS\n\ndef hit(text):\n return next((w for w in WORDS if w in text), None)\n") + git = ["git", "-C", repo, "-c", "user.name=t", "-c", "user.email=t@t"] + subprocess.run(git + ["add", "."], check=True, capture_output=True) + subprocess.run(git + ["commit", "-qm", "v1"], check=True, capture_output=True) + with open(os.path.join(repo, "hook", "words.py"), "w", encoding="utf-8") as handle: + handle.write("WORDS = ('alpha', 'gamma')\n") + path = self.transcript([said("alpha"), said("gamma"), said("delta")]) + det = os.path.join(repo, "hook", "detect.py") + ":hit" + code, out = self.run_main("--detector", det, "--compare", "HEAD", path) + self.assertEqual(code, 0, out) + self.assertIn("detect.py@HEAD:hit", out) + self.assertIn("newly_caught=1 newly_missed=0 unchanged=1", out) + self.assertIn("[gamma]", out) + + def test_loading_uses_the_detectors_own_sibling_and_restores_the_callers(self): + folder = os.path.join(self.tmp, "hook") + os.makedirs(folder) + with open(os.path.join(folder, "words.py"), "w", encoding="utf-8") as handle: + handle.write("WORDS = ('alpha',)\n") + path = self.module(os.path.join("hook", "detect.py"), "from words import WORDS\n\ndef hit(text):\n return text in WORDS\n") + callers = types.ModuleType("words") + callers.WORDS = ("zzz",) + with patch.dict(sys.modules, {"words": callers}): + detector = bd.load_callable(path, "hit") + self.assertIs(sys.modules["words"], callers) + self.assertTrue(detector.fn("alpha")) + self.assertFalse(detector.fn("zzz")) + + def test_compare_with_unknown_revision_is_refused(self): + with self.assertRaises(SystemExit) as caught, patch("sys.stderr", io.StringIO()): + bd.main(["--detector", STOP_CHECK, "--compare", "no-such-ref-xyz", self.tmp]) + self.assertEqual(caught.exception.code, 2) + + +if __name__ == "__main__": + unittest.main()