[Hook architecture](20) Put the frustration-watchdog hook onto the shared hook code

engine/hooks/frustration-watchdog/claude_stop_check.py

@@ -34,8 +34,15 @@
 import os
 import re
 import sys
+import hashlib
 from datetime import datetime
 
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from finding import Finding  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
 IMPATIENCE_PATTERNS = [
     ("profanity", re.compile(r"\b(fuck\w*|wtf|shit\w*|goddamn|dammit|damn it|stupid)\b", re.I)),
     ("told-you", re.compile(r"\bi (already |just )?told you\b|\bi asked you not\b|\bi already said\b", re.I)),
@@ -83,6 +90,8 @@
 """How Claude Code records a tool call a PreToolUse hook refused: an is_error
 tool_result whose text opens "PreToolUse:Bash hook error: [<command>]: ..."."""
 
+RULE_PREFIX = "frustration-watchdog"
+
 
 def _is_allcaps(text):
     letters = [c for c in text if c.isalpha()]
@@ -203,27 +212,23 @@ def ends_the_wait(message):
     return bool(NEXT_STEP_RE.search(message) or ETA_RE.search(message))
 
 
-def main():
-    try:
-        data = json.load(sys.stdin)
-    except json.JSONDecodeError:
-        return
+def detect(data):
     if data.get("stop_hook_active") or data.get("agent_id"):
-        return
+        return []
     message = data.get("last_assistant_message") or ""
     transcript_path = data.get("transcript_path") or ""
     if not message or not transcript_path or not os.path.isfile(transcript_path):
-        return
+        return []
     try:
         msgs = human_user_messages(transcript_path)
         kinds = impatience_kinds(msgs)
     except Exception as exc:
         print(f"catstack-hook-error frustration-watchdog: {type(exc).__name__}: {exc}", file=sys.stderr)
-        return  # fail open: a broken watchdog must never brick a session
+        return []
     if not kinds:
-        return
+        return []
     if ends_the_wait(message):
-        return
+        return []
     try:
         refused = turn_has_hook_refusal(transcript_path)
         unchecked = None
@@ -236,26 +241,48 @@ def main():
         "and this reply hands them nothing visible. "
     )
     if refused:
-        sys.stderr.write(
-            head + "A hook refused a tool call this turn, so you are the one blocked: "
+        feedback = (
+            head
+            + "A hook refused a tool call this turn, so you are the one blocked: "
             "do not hand the user steps to work around it. End the wait: ask them a "
             "direct question, or state an explicit no-action window "
-            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules.\n"
+            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules."
         )
     else:
-        sys.stderr.write(
-            head + "End the wait: give exactly one "
+        feedback = (
+            head
+            + "End the wait: give exactly one "
             "concrete action for the user (\"click X\", \"run Y\", \"say Z\"), ask them a "
             "direct question, or state an explicit no-action window "
-            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules.\n"
+            "(\"nothing needed from you for ~2 min\"). Per CLAUDE.md live-demo rules."
         )
     if unchecked:
-        sys.stderr.write(
-            f"(frustration-watchdog could not read this turn's tool results ({unchecked}), "
+        feedback = (
+            f"catstack-hook-error frustration-watchdog: {unchecked}\n"
+            + feedback
+            + "\n"
+            + f"(frustration-watchdog could not read this turn's tool results ({unchecked}), "
             "so it could not tell whether a hook refused a tool call; the wording above "
-            "is the default.)\n"
+            "is the default.)"
         )
-    sys.exit(2)
+    primary_kind = sorted(set(kinds))[0]
+    return [
+        Finding(
+            rule_id=f"{RULE_PREFIX}.{primary_kind}",
+            subject="reply:" + hashlib.sha256(message.encode("utf-8")).hexdigest(),
+            message=feedback,
+            evidence=", ".join(sorted(set(kinds))),
+        )
+    ]
+
+
+def main():
+    try:
+        run_hook("frustration-watchdog", "claude", detect, "Stop")
+    except SystemExit as exc:
+        if exc.code == 0:
+            return
+        raise
 
 
 if __name__ == "__main__":

engine/hooks/frustration-watchdog/codex_stop_check.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Codex Stop entrypoint for frustration-watchdog."""
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from claude_stop_check import detect  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+
+def main() -> None:
+    run_hook("frustration-watchdog", "codex", detect, "Stop")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/frustration-watchdog/cursor_stop_check.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Cursor stop entrypoint for frustration-watchdog."""
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from claude_stop_check import detect  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+
+def main() -> None:
+    run_hook("frustration-watchdog", "cursor", detect, "stop")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/frustration-watchdog/tests/test_hooks_sdk_mode.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from test_hooks import NARRATION, WAITING, human, transcript_lines, tool_turn
+
+HOOK_DIR = Path(__file__).resolve().parents[1]
+HOOK = HOOK_DIR / "claude_stop_check.py"
+
+
+def payload(transcript_path: str) -> dict[str, object]:
+    return {
+        "hook_event_name": "Stop",
+        "session_id": "frustration-watchdog-sdk-mode",
+        "transcript_path": transcript_path,
+        "last_assistant_message": NARRATION,
+        "stop_hook_active": False,
+    }
+
+
+def run_hook(data: dict[str, object], env: dict[str, str]) -> subprocess.CompletedProcess[str]:
+    merged_env = os.environ.copy()
+    merged_env.pop("CATSTACK_HOOK_MODE_FRUSTRATION_WATCHDOG", None)
+    merged_env.update(env)
+    return subprocess.run(
+        [sys.executable, str(HOOK)],
+        input=json.dumps(data),
+        capture_output=True,
+        text=True,
+        timeout=10,
+        env=merged_env,
+    )
+
+
+class SdkModeTest(unittest.TestCase):
+    def test_mode_override_warn_turns_stop_into_warning(self) -> None:
+        transcript = transcript_lines([human(WAITING)])
+        try:
+            result = run_hook(
+                payload(transcript),
+                {"CATSTACK_HOOK_MODE_FRUSTRATION_WATCHDOG": "warn"},
+            )
+        finally:
+            os.unlink(transcript)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual("", result.stderr)
+        output = json.loads(result.stdout)
+        self.assertIn(
+            "The user's last message was impatience-shaped (waiting)",
+            output["hookSpecificOutput"]["additionalContext"],
+        )
+
+    def test_each_finding_writes_one_event_row_with_rule_id(self) -> None:
+        with tempfile.TemporaryDirectory() as metrics_dir:
+            transcript = transcript_lines([human(WAITING)] + tool_turn("Created PR #12", is_error=False))
+            try:
+                result = run_hook(
+                    payload(transcript),
+                    {
+                        "CATSTACK_HOOK_METRICS_DIR": metrics_dir,
+                        "CATSTACK_HOOK_MODE_FRUSTRATION_WATCHDOG": "warn",
+                    },
+                )
+            finally:
+                os.unlink(transcript)
+            rows = self._event_rows(metrics_dir)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(1, len(rows))
+        self.assertEqual("frustration-watchdog", rows[0]["hook"])
+        self.assertEqual("frustration-watchdog.waiting", rows[0]["rule_id"])
+        self.assertEqual("warn", rows[0]["mode"])
+        self.assertEqual("override", rows[0]["mode_source"])
+        self.assertEqual("warned", rows[0]["action"])
+
+    def _event_rows(self, metrics_dir: str) -> list[dict[str, object]]:
+        files = list(Path(metrics_dir).glob("events-*.jsonl"))
+        self.assertEqual(1, len(files))
+        return [json.loads(line) for line in files[0].read_text(encoding="utf-8").splitlines()]
+
+
+if __name__ == "__main__":
+    unittest.main()