fix(hooks): cadence sound-migration tail — action-gated sound via return-key (FPLAN-0249)

Notification handlers (announce, email, stop_sound, tool_sound) return a
'sound' key the engine plays on action instead of calling speak() on every
invocation — quieter and honest (skipped loaders stay silent). Slim
cadence_investigation.md. Tests updated to assert the return-key form. 472/472
hooks green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/aipass/hooks/apps/handlers/notification/announce.py

@@ -13,14 +13,12 @@
 import os
 from pathlib import Path
 
-from aipass.hooks.apps.sound import speak
-
 AIPASS_HOME = Path(os.environ.get("AIPASS_HOME", ""))
 SOUNDS_DIR = AIPASS_HOME / ".claude" / "sounds"
 SOUND_FILE = SOUNDS_DIR / "mixkit-clear-announce-tones-2861.wav"
 
 
-def handle(hook_data: dict) -> dict:
+def handle(hook_data: dict) -> dict:  # noqa: ARG001
     """Play notification tone and speak hook name for identification.
 
     Args:
@@ -29,5 +27,4 @@ def handle(hook_data: dict) -> dict:
     Returns:
         Result dict with stdout (empty) and exit_code.
     """
-    speak("notification sound")
+    return {"stdout": "", "exit_code": 0, "sound": "notification sound"}
-    return {"stdout": "", "exit_code": 0}

src/aipass/hooks/apps/handlers/notification/email.py

@@ -13,7 +13,6 @@
 import json
 from pathlib import Path
 
-from aipass.hooks.apps.sound import speak
 from aipass.prax.apps.modules.logger import system_logger as logger
 
 
@@ -97,7 +96,13 @@ def handle(hook_data: dict) -> dict:
         return {"stdout": "", "exit_code": 0}
 
     plural = "s" if new_count != 1 else ""
-    speak(f"email notification: {new_count} new email{plural}")
+    msg = (
-    msg = f"You have {new_count} new email{plural} - check with: drone @ai_mail inbox | then: drone @ai_mail view <id> | close with: drone @ai_mail close <id>"
+        f"You have {new_count} new email{plural} - check with: drone @ai_mail inbox"
+        " | then: drone @ai_mail view <id> | close with: drone @ai_mail close <id>"
+    )
     logger.info("[HOOKS] email: %d new email%s", new_count, plural)
-    return {"stdout": msg, "exit_code": 0}
+    return {
+        "stdout": msg,
+        "exit_code": 0,
+        "sound": f"email notification: {new_count} new email{plural}",
+    }

src/aipass/hooks/apps/handlers/notification/stop_sound.py

@@ -13,8 +13,6 @@
 import os
 from pathlib import Path
 
-from aipass.hooks.apps.sound import speak
-
 AIPASS_HOME = Path(os.environ.get("AIPASS_HOME", ""))
 SOUNDS_DIR = AIPASS_HOME / ".claude" / "sounds"
 SOUND_FILE = SOUNDS_DIR / "mixkit-achievement-bell-600.wav"
@@ -32,5 +30,4 @@ def handle(hook_data: dict) -> dict:
     if hook_data.get("stop_hook_active", False):
         return {"stdout": "", "exit_code": 0}
 
-    speak("stop sound")
+    return {"stdout": "", "exit_code": 0, "sound": "stop sound"}
-    return {"stdout": "", "exit_code": 0}

src/aipass/hooks/apps/handlers/notification/tool_sound.py

@@ -10,8 +10,6 @@
 
 """Announces hook name via Piper TTS when the AI uses tools (PreToolUse event)."""
 
-from aipass.hooks.apps.sound import speak
-
 
 def handle(hook_data: dict) -> dict:
     """Announce hook name for matching tool use events.
@@ -26,5 +24,4 @@ def handle(hook_data: dict) -> dict:
     if not tool_name:
         return {"stdout": "", "exit_code": 0}
 
-    speak(f"tool sound: {tool_name}")
+    return {"stdout": "", "exit_code": 0, "sound": f"tool sound: {tool_name}"}
-    return {"stdout": "", "exit_code": 0}

src/aipass/hooks/docs/cadence_investigation.md

@@ -1,109 +1,54 @@
 # Cadence Investigation — DPLAN-0200
 
-Per-turn injection cadence mechanism for prompt loaders (global_loader, branch_loader, identity).
+Per-turn injection cadence mechanism for prompt loaders (global_loader, branch_loader).
-Investigation only — no build. Findings for @devpulse.
+Investigation + build. Updated post-REDO to reflect the real execution model.
 
 ---
 
 ## 1. Per-session turn counter — session keying
 
-YES, fully reliable. `CLAUDE_CODE_SESSION_ID` is available as an env var to every hook invocation (confirmed live: UUID format, stable across turns, unique per session). This is the same mechanism `auto_process.py` already uses for its once-per-session guard (`auto_process.py:24`, `/tmp` sentinel keyed by session_id).
+`CLAUDE_CODE_SESSION_ID` is available as an env var to every hook invocation (UUID format, stable across turns, unique per session). Counter file keyed by session_id at `/tmp/aipass-cadence-{session_id}.json`.
 
-`hook_data` (the parsed stdin JSON) does NOT contain session_id — it comes from the env var only. For UserPromptSubmit, hook_data contains `{"user_prompt": "..."}` and sometimes other fields, but session keying must use `os.environ`.
+UserPromptSubmit stdin fields: `session_id`, `transcript_path`, `cwd`, `hook_event_name`, `prompt`. The field is `prompt` (not `user_prompt`); `session_id` IS present in hook_data.
 
-A new session gets a new UUID — fresh counter automatically. No inheritance risk.
+## 2. Execution model — SEPARATE PROCESSES (corrected)
 
-## 2. Counter state location
+**Each hook runs as a separate OS process.** `settings.json` registers distinct commands per handler: `claude.py UserPromptSubmit:global_prompt`, `:branch_prompt`, `:identity_injector`, `:email_notification`, `:auto_process` — 5 separate Python subprocesses spawned near-simultaneously by Claude Code.
 
-Engine has NO per-session state infrastructure today. `auto_process.py`'s `/tmp` sentinel is the closest precedent — existence-only, no data payload.
+Module-level caches do NOT persist across these processes. The original investigation (pre-REDO) incorrectly assumed sequential single-process dispatch. Live observation proved the counter double-incremented (33 → 35 → 37 across single turns).
 
-**Recommended:** `/tmp/aipass-cadence-{session_id}.json` — tiny JSON file (`{"turn": N}`), one per session, naturally cleaned on reboot. Same `/tmp` pattern as auto_process but with a data payload instead of touch-only.
+## 3. Multi-process dedup mechanism
 
-This is net-new state. The engine doesn't need to know about it — a shared cadence module handles it.
+The counter must advance exactly once per real user turn regardless of sibling process count.
 
-## 3. Mechanism sketch — feasible, confirmed
+Three-layer dedup in `_load_and_increment()`:
 
-Shared module: `apps/handlers/prompt/cadence.py`
+1. **fcntl.flock** — exclusive lock around read-modify-write of the state file. Prevents simultaneous siblings from both reading stale state.
+2. **mtime debounce** (~2s) — if the state file was modified < 2 seconds ago, treat as the same turn. The first sibling increments; the rest see fresh mtime and reuse the current value.
+3. **Per-turn token** — `transcript_path` file size (monotonic, identical across siblings). Only increment if BOTH the debounce window elapsed AND the token changed. Kills pathologically fast turns and identical-prompt collisions.
 
-Key insight: the bridge spawns ONE Python process per UserPromptSubmit dispatch, and the engine runs all handlers SEQUENTIALLY within that process (`engine.py:114` loop). So a module-level cache ensures the counter increments exactly ONCE per turn, even though 3 handlers call into it.
+Special case: `turn < 0` (post-compact reset) always increments — debounce must not swallow the turn-0 all-fire guarantee.
 
-```python
+Module: `apps/modules/cadence.py` (shared utility, accessed via `importlib.import_module` from handlers).
-# cadence.py sketch
-_turn = None  # process-level cache, reset each dispatch = each turn
 
-def _load_and_increment():
+## 4. Action-gated sound
-    global _turn
-    if _turn is not None:
-        return _turn  # already incremented this dispatch
-    path = _state_path()  # /tmp/aipass-cadence-{session_id}.json
-    if path is None:
-        _turn = 0
-        return 0
-    count = 0
-    if path.exists():
-        data = json.loads(path.read_text())
-        count = data.get("turn", 0) + 1
-    path.write_text(json.dumps({"turn": count}))
-    _turn = count
-    return count
 
-def should_fire(offset, period=5):
+Handlers return a `"sound"` key in their result dict. The engine plays it at the output collection point (`engine.py`). Removed all scattered leading `speak()` calls — sound is now tied to handler action, not invocation. A skipped loader stays silent.
-    turn = _load_and_increment()
-    if turn == 0:
-        return True  # first turn ALWAYS fires
-    return (turn % period) == offset
-```
 
-Each loader adds ONE guard line:
+## 5. Edge cases
 
-```python
+**FIRST TURN:** `should_fire` returns True when `turn==0`. Agent always gets full context on session start.
-from aipass.hooks.apps.handlers.prompt.cadence import should_fire
 
-def handle(hook_data):
+**CONCURRENT SESSIONS:** Counter file keyed by session_id — no cross-session conflict.
-    if not should_fire(offset=0):  # different offset per loader
-        return {"stdout": "", "exit_code": 0}
-    # ... existing logic unchanged
-```
 
-**Stagger example (period=5):**
+**COMPACTION:** PreCompact handler (`compact.py`) resets counter to -1 via `cadence.reset_counter()`. Next turn reads -1+1=0, all loaders fire.
 
-| Loader | Offset | Fires on turns |
+**FILE I/O COST:** One flock + read + conditional write of ~30 bytes per turn. Negligible vs ~3,750 tokens saved.
-|--------|--------|----------------|
-| global_loader | 0 | 0, 5, 10, 15... |
-| branch_loader | 2 | 0, 2, 7, 12... |
-| identity | 4 | 0, 4, 9, 14... |
-
-Turn 0 = ALL fire (first turn guarantee). After that, max 1 loader per turn, each refreshed every 5 turns, staggered so they never collide.
-
-## 4. Edge cases
-
-**FIRST TURN:** Handled — `should_fire` returns True unconditionally when `turn==0`. Agent always gets full context on session start.
-
-**CONCURRENT SESSIONS:** Safe — counter file is keyed by session_id. Two sessions in different branches use different files, no conflict.
-
-**COMPACTION — CRITICAL INTERACTION:** When Claude Code compacts, injected prompts from prior turns get summarized or dropped from context. If a loader's next fire is 3-4 turns away post-compaction, the agent operates without that prompt content until it re-fires.
-
-**Fix:** Add a PreCompact handler (or extend the existing `compact.py`) that RESETS the cadence counter to -1. On the next UserPromptSubmit after compaction, `_load_and_increment` reads `-1+1=0`, and turn 0 = all loaders fire. Cost: one extra full-injection turn after each compaction, which is exactly right — the agent needs the prompts re-injected after losing context.
-
-**FILE I/O COST:** Negligible. One stat + read + write of ~15 bytes per turn. Way cheaper than the ~3,750 tokens saved.
-
-## 5. Robustness assessment — SOLID
-
-**Strengths:**
-- Pattern is simple and deterministic. No async, no races, no distributed state.
-- Sequential dispatch (`engine.py`) guarantees no concurrent access to the counter file within a single turn.
-- `/tmp` cleanup on reboot = no accumulation. Session files are tiny and ephemeral.
-- Module-level cache (`_turn`) prevents double-increment even if called from multiple handlers.
-- Testable in isolation — mock `os.environ` + `/tmp` path, assert `should_fire` returns correctly.
-
-**One fragility to flag:** if Claude Code ever changes to dispatch UserPromptSubmit handlers in PARALLEL (separate processes), the module-level cache breaks and you'd get 3 increments per turn. Current architecture is sequential — but worth a comment noting the assumption. Mitigation: use file locking (`fcntl.flock`) if parallelism ever arrives, but don't build it now.
-
-No other fragility concerns. The mechanism is as robust as `auto_process.py`'s session guard, which has been running reliably since S10.
 
 ---
 
 ## Summary
 
-Fully feasible. Shared `cadence.py` module, `/tmp` state file keyed by session_id, modulo+offset check per handler, turn-0 guarantee, PreCompact counter reset for compaction safety. Ready for implementation when DPLAN-0200 greenlights it.
+Multi-process safe cadence via fcntl.flock + mtime debounce + transcript-size token. Shared module in `apps/modules/`, handlers access via importlib. Action-gated sound system-wide. 438 tests, seedgo 100%.
 
-*Investigation by @hooks, 2026-06-08*
+*Investigation by @hooks, 2026-06-08. Updated post-REDO 2026-06-09.*

src/aipass/hooks/tests/test_announce.py

@@ -1,16 +1,14 @@
 # =================== AIPass ====================
 # Name: test_announce.py
-# Version: 1.2.0
+# Version: 1.3.0
 # Description: Tests for announce notification handler
 # Branch: hooks
 # Created: 2026-05-20
-# Modified: 2026-05-22
+# Modified: 2026-06-09
 # =============================================
 
 """Tests for handlers/notification/announce.py."""
 
-from unittest.mock import patch
-
 
 class TestAnnounceHandler:
     """Core handler behavior tests."""
@@ -18,17 +16,15 @@ class TestAnnounceHandler:
     def test_handle_returns_result_dict(self):
         from aipass.hooks.apps.handlers.notification.announce import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.announce.speak"):
+        result = handle({})
-            result = handle({})
 
         assert isinstance(result, dict)
         assert result["stdout"] == ""
         assert result["exit_code"] == 0
 
-    def test_handle_speaks_notification_sound(self):
+    def test_handle_sets_sound_key(self):
         from aipass.hooks.apps.handlers.notification.announce import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.announce.speak") as mock_speak:
+        result = handle({})
-            handle({})
 
-        mock_speak.assert_called_once_with("notification sound")
+        assert result["sound"] == "notification sound"

src/aipass/hooks/tests/test_email.py

@@ -1,10 +1,10 @@
 # =================== AIPass ====================
 # Name: test_email.py
-# Version: 1.2.0
+# Version: 1.3.0
 # Description: Tests for email notification handler
 # Branch: hooks
 # Created: 2026-05-21
-# Modified: 2026-05-22
+# Modified: 2026-06-09
 # =============================================
 
 """Tests for handlers/notification/email.py."""
@@ -41,12 +41,9 @@ class TestEmailHandler:
             encoding="utf-8",
         )
 
-        with (
+        with patch(
-            patch(
+            "aipass.hooks.apps.handlers.notification.email._find_branch_root",
-                "aipass.hooks.apps.handlers.notification.email._find_branch_root",
+            return_value=tmp_path,
-                return_value=tmp_path,
-            ),
-            patch("aipass.hooks.apps.handlers.notification.email.speak"),
         ):
             result = handle({})
 
@@ -54,7 +51,7 @@ class TestEmailHandler:
         assert "drone @ai_mail inbox" in result["stdout"]
         assert result["exit_code"] == 0
 
-    def test_handle_speaks_when_new_emails(self, tmp_path):
+    def test_handle_sets_sound_when_new_emails(self, tmp_path):
         from aipass.hooks.apps.handlers.notification.email import handle
 
         inbox_dir = tmp_path / ".ai_mail.local"
@@ -65,18 +62,15 @@ class TestEmailHandler:
             encoding="utf-8",
         )
 
-        with (
+        with patch(
-            patch(
+            "aipass.hooks.apps.handlers.notification.email._find_branch_root",
-                "aipass.hooks.apps.handlers.notification.email._find_branch_root",
+            return_value=tmp_path,
-                return_value=tmp_path,
-            ),
-            patch("aipass.hooks.apps.handlers.notification.email.speak") as mock_speak,
         ):
-            handle({})
+            result = handle({})
 
-        mock_speak.assert_called_once_with("email notification: 1 new email")
+        assert result["sound"] == "email notification: 1 new email"
 
-    def test_handle_does_not_speak_when_no_emails(self, tmp_path):
+    def test_handle_no_sound_when_no_emails(self, tmp_path):
         from aipass.hooks.apps.handlers.notification.email import handle
 
         inbox_dir = tmp_path / ".ai_mail.local"
@@ -87,16 +81,13 @@ class TestEmailHandler:
             encoding="utf-8",
         )
 
-        with (
+        with patch(
-            patch(
+            "aipass.hooks.apps.handlers.notification.email._find_branch_root",
-                "aipass.hooks.apps.handlers.notification.email._find_branch_root",
+            return_value=tmp_path,
-                return_value=tmp_path,
-            ),
-            patch("aipass.hooks.apps.handlers.notification.email.speak") as mock_speak,
         ):
-            handle({})
+            result = handle({})
 
-        mock_speak.assert_not_called()
+        assert result.get("sound", "") == ""
 
     def test_handle_returns_empty_when_no_new_emails(self, tmp_path):
         from aipass.hooks.apps.handlers.notification.email import handle
@@ -109,12 +100,9 @@ class TestEmailHandler:
             encoding="utf-8",
         )
 
-        with (
+        with patch(
-            patch(
+            "aipass.hooks.apps.handlers.notification.email._find_branch_root",
-                "aipass.hooks.apps.handlers.notification.email._find_branch_root",
+            return_value=tmp_path,
-                return_value=tmp_path,
-            ),
-            patch("aipass.hooks.apps.handlers.notification.email.speak"),
         ):
             result = handle({})
 
@@ -140,12 +128,9 @@ class TestEmailHandler:
             encoding="utf-8",
         )
 
-        with (
+        with patch(
-            patch(
+            "aipass.hooks.apps.handlers.notification.email._find_branch_root",
-                "aipass.hooks.apps.handlers.notification.email._find_branch_root",
+            return_value=tmp_path,
-                return_value=tmp_path,
-            ),
-            patch("aipass.hooks.apps.handlers.notification.email.speak"),
         ):
             result = handle({})
 

src/aipass/hooks/tests/test_stop_sound.py

@@ -1,16 +1,14 @@
 # =================== AIPass ====================
 # Name: test_stop_sound.py
-# Version: 1.2.0
+# Version: 1.3.0
 # Description: Tests for stop_sound notification handler
 # Branch: hooks
 # Created: 2026-05-20
-# Modified: 2026-05-22
+# Modified: 2026-06-09
 # =============================================
 
 """Tests for handlers/notification/stop_sound.py."""
 
-from unittest.mock import patch
-
 
 class TestStopSoundHandler:
     """Core handler behavior tests."""
@@ -18,26 +16,23 @@ class TestStopSoundHandler:
     def test_handle_returns_result_dict(self):
         from aipass.hooks.apps.handlers.notification.stop_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak"):
+        result = handle({})
-            result = handle({})
 
         assert isinstance(result, dict)
         assert result["stdout"] == ""
         assert result["exit_code"] == 0
 
-    def test_handle_speaks_stop_sound(self):
+    def test_handle_sets_sound_key(self):
         from aipass.hooks.apps.handlers.notification.stop_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak") as mock_speak:
+        result = handle({})
-            handle({})
 
-        mock_speak.assert_called_once_with("stop sound")
+        assert result["sound"] == "stop sound"
 
-    def test_handle_skips_when_stop_hook_active(self):
+    def test_handle_no_sound_when_stop_hook_active(self):
         from aipass.hooks.apps.handlers.notification.stop_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak") as mock_speak:
+        result = handle({"stop_hook_active": True})
-            result = handle({"stop_hook_active": True})
 
-        mock_speak.assert_not_called()
+        assert result.get("sound", "") == ""
         assert result["exit_code"] == 0

src/aipass/hooks/tests/test_tool_sound.py

@@ -1,16 +1,14 @@
 # =================== AIPass ====================
 # Name: test_tool_sound.py
-# Version: 1.2.0
+# Version: 1.3.0
 # Description: Tests for tool_sound notification handler
 # Branch: hooks
 # Created: 2026-05-19
-# Modified: 2026-05-22
+# Modified: 2026-06-09
 # =============================================
 
 """Tests for handlers/notification/tool_sound.py."""
 
-from unittest.mock import patch
-
 
 class TestToolSoundHandler:
     """Core handler behavior tests."""
@@ -18,8 +16,7 @@ class TestToolSoundHandler:
     def test_handle_returns_result_dict(self):
         from aipass.hooks.apps.handlers.notification.tool_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak"):
+        result = handle({"tool_name": "Bash"})
-            result = handle({"tool_name": "Bash"})
 
         assert isinstance(result, dict)
         assert "stdout" in result
@@ -27,26 +24,23 @@ class TestToolSoundHandler:
         assert result["stdout"] == ""
         assert result["exit_code"] == 0
 
-    def test_speaks_tool_name(self):
+    def test_sound_key_includes_tool_name(self):
         from aipass.hooks.apps.handlers.notification.tool_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
+        result = handle({"tool_name": "Edit"})
-            handle({"tool_name": "Edit"})
 
-        mock_speak.assert_called_once_with("tool sound: Edit")
+        assert result["sound"] == "tool sound: Edit"
 
-    def test_no_speak_when_no_tool_name(self):
+    def test_no_sound_when_no_tool_name(self):
         from aipass.hooks.apps.handlers.notification.tool_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
+        result = handle({})
-            handle({})
 
-        mock_speak.assert_not_called()
+        assert result.get("sound", "") == ""
 
-    def test_no_speak_when_empty_tool_name(self):
+    def test_no_sound_when_empty_tool_name(self):
         from aipass.hooks.apps.handlers.notification.tool_sound import handle
 
-        with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
+        result = handle({"tool_name": ""})
-            handle({"tool_name": ""})
 
-        mock_speak.assert_not_called()
+        assert result.get("sound", "") == ""