[Hook architecture](17) Put the diu-stop hook onto the shared hook code

engine/hooks/diu-stop/claude_prompt_reminder.py

@@ -16,11 +16,19 @@
 reminder repeated every turn is exactly the kind of thing this skill tells
 the model to cut.
 """
-import json
+import os
 import sys
 
 from diu_limit import rule_text
 
+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
+
+RULE_REMINDER = "diu-stop.prompt-reminder"
+
 REMINDER = (
     "diu reminder: lead with the outcome, no preamble or closing "
     f"pleasantries, ELI5 {rule_text()}, unless this turn needs technical "
@@ -29,17 +37,13 @@
 )
 
 
+def detect(event):
+    subject = event.get("session_id") or ""
+    return [Finding(rule_id=RULE_REMINDER, subject=subject, message=REMINDER, evidence=REMINDER)]
+
+
 def main():
-    try:
-        json.load(sys.stdin)
-    except json.JSONDecodeError:
-        return
-    print(json.dumps({
-        "hookSpecificOutput": {
-            "hookEventName": "UserPromptSubmit",
-            "additionalContext": REMINDER,
-        }
-    }))
+    run_hook("diu-stop", "claude", detect, "UserPromptSubmit")
 
 
 if __name__ == "__main__":

engine/hooks/diu-stop/claude_stop_check.py

@@ -45,7 +45,6 @@
 Every block names every flagged sentence, so one rewrite that fixes them
 all gets through.
 """
-import json
 import os
 import re
 import sys
@@ -55,8 +54,18 @@
 
 sys.path.insert(0, os.path.join(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_markers"))
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
 
 import markers  # noqa: E402
+from finding import Finding  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+RULE_WORD_LIMIT = "diu-stop.word-limit"
+RULE_UNVERIFIED_CLAIM = "diu-stop.unverified-claim"
+RULE_MALFORMED_MARKER_TAG = "diu-stop.malformed-marker-tag"
+RULE_LEGACY_MARKER = "diu-stop.legacy-marker"
+RULE_PLAIN_WORDS = "diu-stop.plain-words"
 
 # Phrases banned outright (from this user's global CLAUDE.md evidence
 # rules) -- rarely legitimate even mid-sentence, so no opener restriction.
@@ -219,31 +228,28 @@ def find_unverified_claim(message):
     return claims[0][0] if claims else None
 
 
-def main():
-    try:
-        data = json.load(sys.stdin)
-    except json.JSONDecodeError:
-        return
-
-    if data.get("agent_id"):
-        return
-    retry = bool(data.get("stop_hook_active"))
+def detect(event):
+    if event.get("agent_id"):
+        return []
+    retry = bool(event.get("stop_hook_active"))
 
-    message = data.get("last_assistant_message") or ""
+    message = event.get("last_assistant_message") or ""
 
-    plain_words_note = try_check_reply(data)
+    plain_words_note = try_check_reply(event)
 
     word_count = counted_words(message)
     over_limit = word_count > WORD_LIMIT and not retry
     claims = find_unverified_claims(message)
     marker_problems = find_marker_problems(message)
 
-    if not over_limit and not claims and not marker_problems and not plain_words_note:
-        return
-
-    parts = []
+    findings = []
     if plain_words_note:
-        parts.append(plain_words_note)
+        findings.append(Finding(
+            rule_id=RULE_PLAIN_WORDS,
+            subject=message,
+            message=plain_words_note,
+            evidence=plain_words_note,
+        ))
     if claims:
         lines = [
             "This message makes an unverified-shaped claim with no adjacent "
@@ -259,18 +265,30 @@ def main():
             "of what was actually run/checked in its paragraph, or -- only if "
             "the check cannot run -- tag the claim there and say why."
         )
-        parts.append("\n".join(lines))
-    parts.extend(marker_problems)
+        claim_message = "\n".join(lines)
+        findings.append(Finding(
+            rule_id=RULE_UNVERIFIED_CLAIM,
+            subject=claim_message,
+            message=claim_message,
+            evidence=claim_message,
+        ))
+    for problem in marker_problems:
+        rule_id = RULE_MALFORMED_MARKER_TAG if problem == markers.MALFORMED_TAG_MESSAGE else RULE_LEGACY_MARKER
+        findings.append(Finding(rule_id=rule_id, subject=message, message=problem, evidence=problem))
     if over_limit:
-        parts.append(
+        over_message = (
             f"Apply diu: {word_count} words, over the {WORD_LIMIT}-word "
             f"guideline. Cut at least {word_count - WORD_LIMIT} words by "
             "dropping a whole section or list, not by trimming words. "
             "Unless this turn genuinely asked for full technical detail "
             "or a specific long format."
         )
-    sys.stderr.write("\n".join(parts) + "\n")
-    sys.exit(2)
+        findings.append(Finding(rule_id=RULE_WORD_LIMIT, subject=message, message=over_message, evidence=over_message))
+    return findings
+
+
+def main():
+    run_hook("diu-stop", "claude", detect, "Stop")
 
 
 if __name__ == "__main__":

engine/hooks/diu-stop/codex_notify.py

@@ -17,10 +17,34 @@
 becomes `old-notify-binary some-arg <json-payload>` when this fires.
 """
 import json
+import os
 import subprocess
 import sys
 
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from events import write_events  # noqa: E402
+from finding import Finding  # noqa: E402
+from modes import effective_mode  # noqa: E402
+
 WORD_LIMIT = 150
+RULE_WORD_LIMIT = "diu-stop.word-limit"
+
+
+def detect(payload):
+    if payload.get("type") != "agent-turn-complete":
+        return []
+    message = payload.get("last-assistant-message") or ""
+    word_count = len(message.split())
+    if word_count <= WORD_LIMIT:
+        return []
+    message_text = (
+        f"diu-stop: last response was {word_count} words (over the "
+        f"{WORD_LIMIT}-word diu guideline). Codex can't be forced to redo "
+        "it -- check by hand whether it should have been ELI5."
+    )
+    return [Finding(rule_id=RULE_WORD_LIMIT, subject=message, message=message_text, evidence=message_text)]
 
 
 def main():
@@ -40,18 +64,18 @@ def main():
     except json.JSONDecodeError:
         return
 
-    if payload.get("type") != "agent-turn-complete":
+    findings = detect(payload)
+    if not findings:
         return
 
-    message = payload.get("last-assistant-message") or ""
-    word_count = len(message.split())
-    if word_count > WORD_LIMIT:
-        print(
-            f"diu-stop: last response was {word_count} words (over the "
-            f"{WORD_LIMIT}-word diu guideline). Codex can't be forced to redo "
-            "it -- check by hand whether it should have been ELI5.",
-            file=sys.stderr,
-        )
+    mode, mode_source = effective_mode("diu-stop", payload)
+    if mode == "off":
+        return
+
+    for finding in findings:
+        print(finding.message, file=sys.stderr)
+
+    write_events("diu-stop", "codex", payload, findings, mode, mode_source, 0)
 
 
 if __name__ == "__main__":

engine/hooks/diu-stop/tests/test_hooks.py

@@ -56,7 +56,10 @@ def run_prompt_reminder(stdin_obj):
     buf = io.StringIO()
     with patch.object(sys, "stdin", io.StringIO(json.dumps(stdin_obj))):
         with redirect_stdout(buf):
-            claude_prompt_reminder.main()
+            try:
+                claude_prompt_reminder.main()
+            except SystemExit:
+                pass
     return buf.getvalue()
 
 
@@ -140,7 +143,10 @@ def test_malformed_stdin_json_does_not_crash(self):
         buf = io.StringIO()
         with patch.object(sys, "stdin", io.StringIO("not json")):
             with redirect_stdout(buf):
-                claude_stop_check.main()  # must not raise
+                try:
+                    claude_stop_check.main()
+                except SystemExit:
+                    pass
         self.assertEqual(buf.getvalue(), "")
 
 
@@ -176,7 +182,10 @@ def test_malformed_stdin_json_does_not_crash(self):
         buf = io.StringIO()
         with patch.object(sys, "stdin", io.StringIO("not json")):
             with redirect_stdout(buf):
-                claude_prompt_reminder.main()  # must not raise
+                try:
+                    claude_prompt_reminder.main()
+                except SystemExit:
+                    pass
         self.assertEqual(buf.getvalue(), "")
 
 

engine/hooks/diu-stop/tests/test_hooks_sdk_mode.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+HOOK_DIR = os.path.dirname(HERE)
+ENTRYPOINT = os.path.join(HOOK_DIR, "claude_stop_check.py")
+
+LONG_MESSAGE = " ".join(["word"] * 200)
+
+
+def run_entrypoint(payload: dict[str, object], env: dict[str, str]) -> subprocess.CompletedProcess[str]:
+    merged_env = os.environ.copy()
+    merged_env.update(env)
+    return subprocess.run(
+        [sys.executable, ENTRYPOINT],
+        input=json.dumps(payload),
+        capture_output=True,
+        text=True,
+        env=merged_env,
+    )
+
+
+class SdkModeTest(unittest.TestCase):
+    def test_mode_override_warn_changes_block_to_warning(self) -> None:
+        result = run_entrypoint(
+            {"last_assistant_message": LONG_MESSAGE},
+            {"CATSTACK_HOOK_MODE_DIU_STOP": "warn"},
+        )
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual("", result.stderr)
+        rendered = json.loads(result.stdout)
+        self.assertIn(
+            "Apply diu",
+            rendered["hookSpecificOutput"]["additionalContext"],
+        )
+
+    def test_mode_stop_still_blocks_by_default(self) -> None:
+        result = run_entrypoint({"last_assistant_message": LONG_MESSAGE}, {})
+
+        self.assertEqual(2, result.returncode)
+        self.assertIn("Apply diu", result.stderr)
+
+    def test_each_finding_writes_one_event_row_with_rule_id(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            result = run_entrypoint(
+                {
+                    "session_id": "diu-stop-sdk-mode",
+                    "last_assistant_message": LONG_MESSAGE,
+                },
+                {
+                    "CATSTACK_HOOK_METRICS_DIR": tmp,
+                    "CATSTACK_HOOK_MODE_DIU_STOP": "warn",
+                },
+            )
+            rows = [
+                json.loads(line)
+                for file in Path(tmp).glob("events-*.jsonl")
+                for line in file.read_text(encoding="utf-8").splitlines()
+            ]
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(1, len(rows))
+        self.assertEqual("diu-stop", rows[0]["hook"])
+        self.assertEqual("diu-stop.word-limit", rows[0]["rule_id"])
+        self.assertEqual("warn", rows[0]["mode"])
+        self.assertEqual("override", rows[0]["mode_source"])
+        self.assertEqual("warned", rows[0]["action"])
+
+
+if __name__ == "__main__":
+    unittest.main()