slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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>
This commit is contained in:
AIOSAI
2026-06-10 13:09:32 -07:00
co-authored by Claude Opus 4.8
parent 00edd8b3a0
commit b26bd7c853
9 changed files with 87 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/hooks/apps/handlers/notification/announce.py
+2 -5
View File
@@ -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}
return {"stdout": "", "exit_code": 0, "sound": "notification sound"}
src/aipass/hooks/apps/handlers/notification/email.py
+9 -4
View File
@@ -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 = 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>"
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>"
)
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
+1 -4
View File
@@ -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}
return {"stdout": "", "exit_code": 0, "sound": "stop sound"}
src/aipass/hooks/apps/handlers/notification/tool_sound.py
+1 -4
View File
@@ -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}
return {"stdout": "", "exit_code": 0, "sound": f"tool sound: {tool_name}"}
src/aipass/hooks/docs/cadence_investigation.md
+24 -79
View File
@@ -1,109 +1,54 @@
# Cadence Investigation — DPLAN-0200
Per-turn injection cadence mechanism for prompt loaders (global_loader, branch_loader, identity).
Investigation only — no build. Findings for @devpulse.
Per-turn injection cadence mechanism for prompt loaders (global_loader, branch_loader).
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
# cadence.py sketch
_turn = None # process-level cache, reset each dispatch = each turn
Module: `apps/modules/cadence.py` (shared utility, accessed via `importlib.import_module` from handlers).
def _load_and_increment():
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
## 4. Action-gated sound
def should_fire(offset, period=5):
turn = _load_and_increment()
if turn == 0:
return True # first turn ALWAYS fires
return (turn % period) == offset
```
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.
Each loader adds ONE guard line:
## 5. Edge cases
```python
from aipass.hooks.apps.handlers.prompt.cadence import should_fire
**FIRST TURN:** `should_fire` returns True when `turn==0`. Agent always gets full context on session start.
def handle(hook_data):
if not should_fire(offset=0): # different offset per loader
return {"stdout": "", "exit_code": 0}
# ... existing logic unchanged
```
**CONCURRENT SESSIONS:** Counter file keyed by session_id — no cross-session conflict.
**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 |
|--------|--------|----------------|
| 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.
**FILE I/O COST:** One flock + read + conditional write of ~30 bytes per turn. Negligible vs ~3,750 tokens saved.
---
## 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
+6 -10
View File
@@ -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:
handle({})
result = handle({})
mock_speak.assert_called_once_with("notification sound")
assert result["sound"] == "notification sound"
src/aipass/hooks/tests/test_email.py
+23 -38
View File
@@ -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 (
patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email.speak"),
with patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
):
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 (
patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email.speak") as mock_speak,
with patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
):
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 (
patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email.speak") as mock_speak,
with patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
):
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 (
patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email.speak"),
with patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
):
result = handle({})
@@ -140,12 +128,9 @@ class TestEmailHandler:
encoding="utf-8",
)
with (
patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email.speak"),
with patch(
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
):
result = handle({})
src/aipass/hooks/tests/test_stop_sound.py
+9 -14
View File
@@ -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:
handle({})
result = 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
+12 -18
View File
@@ -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:
handle({"tool_name": "Edit"})
result = 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:
handle({})
result = 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:
handle({"tool_name": ""})
result = handle({"tool_name": ""})
mock_speak.assert_not_called()
assert result.get("sound", "") == ""