diff --git a/engine/hooks/frustration-watchdog/claude_stop_check.py b/engine/hooks/frustration-watchdog/claude_stop_check.py index 89a01493f..8b9e6f513 100755 --- a/engine/hooks/frustration-watchdog/claude_stop_check.py +++ b/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: []: ...".""" +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__": diff --git a/engine/hooks/frustration-watchdog/codex_stop_check.py b/engine/hooks/frustration-watchdog/codex_stop_check.py new file mode 100644 index 000000000..7d941052c --- /dev/null +++ b/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() diff --git a/engine/hooks/frustration-watchdog/cursor_stop_check.py b/engine/hooks/frustration-watchdog/cursor_stop_check.py new file mode 100644 index 000000000..64986bfe1 --- /dev/null +++ b/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() diff --git a/engine/hooks/frustration-watchdog/tests/test_hooks_sdk_mode.py b/engine/hooks/frustration-watchdog/tests/test_hooks_sdk_mode.py new file mode 100644 index 000000000..f26b8e619 --- /dev/null +++ b/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()