Backtest: one replay tool for every detector

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/<project-dir> [...]
 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/<project-dir> [...]
+python3 scripts/backtest_detector.py --detector engine/hooks/repeat-error-stop/detect.py:replay_blocks --unit rows ~/.claude/projects/<project-dir> [...]
+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 <git-ref>` lists the blocks a change adds or
+removes.
+
 286 sessions, 38.6k tool results, Aug 2–Sep 1 2026 (Invoker + catstack +
 two other repos):
 

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)

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"}]}]}}

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 <git-ref>` to see what a change newly blocks or lets
+through.