feat(hooks): prompt injection cadence — fire loaders every Nth turn, config-tunable (DPLAN-0200, FPLAN-0249)

Stop re-injecting the global + branch prompts every turn (~3k tokens/turn).
They now fire together every 5th turn; the prior injection persists in context
between fires. Identity + email stay every-turn.

- apps/modules/cadence.py: per-session turn counter (/tmp/aipass-cadence-
  {session_id}.json), should_fire(loader)/reset_counter(), DEFAULTS + deep-merge
  config (api provider.py pattern). 'drone @hooks cadence' introspection.
- global_loader.py + branch_loader.py: cadence guard via importlib (crash-
  isolated); non-fire turn returns empty.
- compact.py: PreCompact resets counter to -1 -> next turn = 0 = all fire
  (rebuild context after compaction). New session = fresh counter = all fire.
- hooks_json/custom_config/cadence_config.json: tunable knob (period/offsets/
  enabled), one file, no code edits. Data lives in the json home, not the code
  dir. Missing file = code DEFAULTS = safe.
- .seedgo/bypass: documented stdlib-json config read (json_handler N/A for a
  dispatch engine).

435 tests pass (26 new), seedgo 100%, pyright 0.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/aipass/hooks/.seedgo/bypass.json

@@ -64,6 +64,10 @@
     {"file": "apps/handlers/lifecycle/auto_process.py", "standard": "unused_function", "reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Wired in UserPromptSubmit.auto_process + PreCompact.auto_process."},
     {"file": "apps/handlers/lifecycle/auto_process.py", "standard": "json_structure", "reason": "Delegates to @memory's auto_process() via importlib — no direct JSON file ops needing json_handler."},
 
+    {"file": "apps/modules/cadence.py", "standard": "dead_code", "reason": "Cadence module — should_fire() called from global_loader.py and branch_loader.py guard lines; reset_counter() called from compact.py PreCompact handler. Not statically discoverable because callers are themselves dynamically dispatched."},
+    {"file": "apps/modules/cadence.py", "standard": "unused_function", "reason": "should_fire() called from global_loader.py + branch_loader.py; reset_counter() called from compact.py. All callers are dynamically dispatched handlers."},
+    {"file": "apps/modules/cadence.py", "standard": "json_structure", "reason": "Uses stdlib json for /tmp state file (turn counter) and hooks_json/custom_config/cadence_config.json config loading — lightweight ephemeral state, not branch data storage."},
+
     {"file": "apps/handlers/notification/announce.py", "standard": "dead_code", "reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.notification.announce.handle' — not statically imported by design. Verified wired in Notification.notification_sound + fires in engine.jsonl."},
     {"file": "apps/handlers/notification/announce.py", "standard": "unused_function", "reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Verified fires in engine.jsonl (Notification.notification_sound)."},
     {"file": "apps/handlers/notification/announce.py", "standard": "json_structure", "reason": "Sound handler — no JSON operations, plays WAV files."},
@@ -91,6 +95,8 @@
 
     {"file": "apps/modules/hookstatus.py", "standard": "json_structure", "reason": "Read-only config viewer — delegates JSON loading to config/loader.py, no direct JSON file ops."},
 
+    {"file": "apps/modules/cadence.py", "standard": "modules", "reason": "Cadence module reads /tmp state file and hooks_json/custom_config/cadence_config.json with stdlib json — lightweight ephemeral state (turn counter) + tunable config. json_handler is for persistent branch data, not /tmp session state or config knobs."},
+
     {"file": "apps/hooks.py", "standard": "unused_function", "reason": "print_introspection() called by drone's discovery system, not internal code."},
 
     {"file": "apps/sound.py", "standard": "unused_function", "reason": "play() called by handler files (stop_sound.py, announce.py) that are dynamically dispatched via importlib — static analysis cannot trace the call chain from hooks.json → engine → handler → sound.play()."},
@@ -203,7 +209,12 @@
     {"file": "tests/test_auto_process.py", "standard": "architecture", "reason": "Test files live in tests/, not in the 3-layer apps structure."},
     {"file": "tests/test_auto_process.py", "standard": "documentation", "reason": "Test methods use descriptive names as documentation per pytest convention."},
     {"file": "tests/test_auto_process.py", "standard": "encapsulation", "reason": "Tests import handlers directly to test implementation details."},
-    {"file": "tests/test_auto_process.py", "standard": "meta", "reason": "Test files do not need Version/Modified metadata headers."}
+    {"file": "tests/test_auto_process.py", "standard": "meta", "reason": "Test files do not need Version/Modified metadata headers."},
+
+    {"file": "tests/test_cadence.py", "standard": "architecture", "reason": "Test files live in tests/, not in the 3-layer apps structure."},
+    {"file": "tests/test_cadence.py", "standard": "documentation", "reason": "Test methods use descriptive names as documentation per pytest convention."},
+    {"file": "tests/test_cadence.py", "standard": "encapsulation", "reason": "Tests import modules and handlers directly to test implementation details."},
+    {"file": "tests/test_cadence.py", "standard": "meta", "reason": "Test files do not need Version/Modified metadata headers."}
   ],
   "notes": {
     "removed_2026-05-19": "Stripped 4 illegitimate bypasses — hooks.py/cli, hooks.py/cli_flags, engine.py/modules, engine.py/introspection. Code fixed to meet standards instead.",

src/aipass/hooks/README.md

@@ -25,6 +25,7 @@ Every hook event flows through one engine. Platform bridges normalize the event
 | `drone @hooks hooksound` | Show current sound mute status |
 | `drone @hooks hooksound off` | Mute all hook sounds |
 | `drone @hooks hooksound on` | Unmute all hook sounds |
+| `drone @hooks cadence` | Show prompt injection cadence config and state |
 | `drone @hooks --help` | Full help reference |
 | `drone @hooks --version` | Version info |
 
@@ -47,6 +48,7 @@ src/aipass/hooks/
 │   ├── hooks.py                 # Entry point (drone @hooks)
 │   ├── sound.py                 # Shared sound utilities (speak, play, mute)
 │   ├── modules/
+│   │   ├── cadence.py           # Prompt injection cadence (every-Nth-turn gating)
 │   │   ├── engine.py            # Core dispatch — routes events to handlers
 │   │   ├── hooksound.py         # Sound control (drone @hooks hooksound on/off)
 │   │   └── hookstatus.py        # Config viewer (drone @hooks status)
@@ -77,7 +79,7 @@ src/aipass/hooks/
 │       └── diagnostics.py       # JSONL logging for hook execution
 ├── logs/
 │   └── engine.jsonl             # JSONL diagnostics (every hook execution)
-└── tests/                       # 385 tests across 20 test files
+└── tests/                       # 435 tests across 21 test files
 ```
 
 ## How It Works

src/aipass/hooks/apps/handlers/lifecycle/compact.py

@@ -84,6 +84,14 @@ def handle(hook_data: dict) -> dict:
     """Inject live branch state for post-compact recovery."""
     speak("pre compact")
 
+    try:
+        import importlib
+
+        cadence = importlib.import_module("aipass.hooks.apps.modules.cadence")
+        cadence.reset_counter()
+    except Exception as exc:
+        logger.info("[HOOKS] compact: cadence reset failed: %s", exc)
+
     try:
         cwd = hook_data.get("cwd", "") or str(Path.cwd())
         branch_dir = _find_branch_dir(cwd)

src/aipass/hooks/apps/handlers/prompt/branch_loader.py

@@ -32,6 +32,15 @@ def handle(hook_data: dict) -> dict:
     """Load branch prompt and private integration prompts."""
     speak("branch prompt")
 
+    try:
+        import importlib
+
+        cadence = importlib.import_module("aipass.hooks.apps.modules.cadence")
+        if not cadence.should_fire("branch"):
+            return {"stdout": "", "exit_code": 0}
+    except Exception as exc:
+        logger.info("[HOOKS] branch_loader: cadence check failed, firing anyway: %s", exc)
+
     try:
         cwd = hook_data.get("cwd", "") or str(Path.cwd())
         branch_root = _find_branch_root(cwd)

src/aipass/hooks/apps/handlers/prompt/global_loader.py

@@ -33,6 +33,15 @@ def handle(hook_data: dict) -> dict:
     """Load global prompt — project-local if outside AIPass, AIPass-internal if inside."""
     speak("global prompt")
 
+    try:
+        import importlib
+
+        cadence = importlib.import_module("aipass.hooks.apps.modules.cadence")
+        if not cadence.should_fire("global"):
+            return {"stdout": "", "exit_code": 0}
+    except Exception as exc:
+        logger.info("[HOOKS] global_loader: cadence check failed, firing anyway: %s", exc)
+
     try:
         aipass_home = os.environ.get("AIPASS_HOME", "")
         cwd = str(Path.cwd())

src/aipass/hooks/apps/modules/cadence.py

@@ -0,0 +1,182 @@
+# =================== AIPass ====================
+# Name: cadence.py
+# Version: 1.0.0
+# Description: Per-session turn counter for prompt injection cadence (DPLAN-0200)
+# Branch: hooks
+# Layer: apps/modules
+# Created: 2026-06-08
+# Modified: 2026-06-08
+# =============================================
+
+"""Turn counter for prompt injection cadence — fires loaders every Nth turn."""
+
+import json
+import os
+from pathlib import Path
+
+from aipass.cli.apps.modules import err_console
+from aipass.prax.apps.modules.logger import system_logger as logger
+
+CONSOLE = err_console
+
+_GUARD_DIR = Path("/tmp")
+_BRANCH_ROOT = Path(__file__).resolve().parent.parent.parent
+_CONFIG_PATH = _BRANCH_ROOT / "hooks_json" / "custom_config" / "cadence_config.json"
+
+HELP_COMMANDS = [
+    ("cadence", "Show prompt injection cadence config and state"),
+]
+
+DEFAULTS = {
+    "enabled": True,
+    "period": 5,
+    "loaders": {
+        "global": {"offset": 0},
+        "branch": {"offset": 0},
+    },
+}
+
+_turn: int | None = None
+_config: dict | None = None
+
+
+def _deep_merge(base: dict, updates: dict) -> dict:
+    """Deep merge updates into base (modifies base in-place)."""
+    for key, value in updates.items():
+        if isinstance(value, dict) and key in base and isinstance(base[key], dict):
+            _deep_merge(base[key], value)
+        else:
+            base[key] = value
+    return base
+
+
+def _load_config() -> dict:
+    global _config
+    if _config is not None:
+        return _config
+
+    import copy
+
+    result = copy.deepcopy(DEFAULTS)
+
+    if _CONFIG_PATH.is_file():
+        try:
+            overrides = json.loads(_CONFIG_PATH.read_text(encoding="utf-8"))
+            _deep_merge(result, overrides)
+        except (json.JSONDecodeError, OSError) as exc:
+            logger.info("[HOOKS] cadence: config load failed, using defaults: %s", exc)
+
+    _config = result
+    return result
+
+
+def _state_path() -> Path | None:
+    session_id = os.environ.get("CLAUDE_CODE_SESSION_ID", "")
+    if not session_id:
+        return None
+    return _GUARD_DIR / f"aipass-cadence-{session_id}.json"
+
+
+def _load_and_increment() -> int:
+    """Load turn counter from /tmp, increment, write back. Cached per-process."""
+    global _turn
+    if _turn is not None:
+        return _turn
+
+    path = _state_path()
+    if path is None:
+        _turn = 0
+        return 0
+
+    count = 0
+    if path.exists():
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            count = data.get("turn", 0) + 1
+        except (json.JSONDecodeError, OSError) as exc:
+            logger.info("[HOOKS] cadence: state read failed, resetting: %s", exc)
+            count = 0
+
+    try:
+        path.write_text(json.dumps({"turn": count}), encoding="utf-8")
+    except OSError as exc:
+        logger.info("[HOOKS] cadence: state write failed: %s", exc)
+
+    _turn = count
+    return count
+
+
+def should_fire(loader_name: str) -> bool:
+    """Check if a loader should fire this turn. Always True on turn 0 or if cadence disabled."""
+    config = _load_config()
+
+    if not config.get("enabled", True):
+        return True
+
+    period = config.get("period", 5)
+    if period <= 0:
+        return True
+
+    loader_config = config.get("loaders", {}).get(loader_name, {})
+    offset = loader_config.get("offset", 0)
+
+    turn = _load_and_increment()
+
+    if turn == 0:
+        return True
+
+    return (turn % period) == offset
+
+
+def reset_counter() -> None:
+    """Reset counter to -1 so next turn reads 0 (all loaders fire). Called from PreCompact."""
+    path = _state_path()
+    if path is None:
+        return
+    try:
+        path.write_text(json.dumps({"turn": -1}), encoding="utf-8")
+        logger.info("[HOOKS] cadence: counter reset for post-compact re-injection")
+    except OSError as exc:
+        logger.info("[HOOKS] cadence: reset write failed: %s", exc)
+
+
+# =============================================================================
+# MODULE INTERFACE (drone @hooks routing)
+# =============================================================================
+
+
+def print_introspection() -> None:
+    """Print cadence config and current state."""
+    config = _load_config()
+    CONSOLE.print("[bold cyan]cadence[/bold cyan] Module")
+    CONSOLE.print(f"  Enabled: {config.get('enabled', True)}")
+    CONSOLE.print(f"  Period: {config.get('period', 5)} turns")
+    loaders = config.get("loaders", {})
+    for name, lcfg in loaders.items():
+        CONSOLE.print(f"  Loader '{name}': offset={lcfg.get('offset', 0)}")
+    path = _state_path()
+    if path and path.exists():
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            CONSOLE.print(f"  Current turn: {data.get('turn', '?')}")
+        except (json.JSONDecodeError, OSError) as exc:
+            logger.info("[HOOKS] cadence: state read for introspection failed: %s", exc)
+            CONSOLE.print("  Current turn: (unreadable)")
+    else:
+        CONSOLE.print("  Current turn: (no state file)")
+    CONSOLE.print(f"  Config file: {_CONFIG_PATH}")
+
+
+def handle_command(command: str, args: list) -> bool:
+    """Route cadence commands from drone @hooks."""
+    if command in ("--help", "-h", "help"):
+        CONSOLE.print("[bold cyan]cadence[/bold cyan] — Prompt injection cadence control")
+        CONSOLE.print()
+        CONSOLE.print("  drone @hooks cadence    Show cadence config and current turn state")
+        return True
+
+    if command == "cadence":
+        if not args:
+            print_introspection()
+            return True
+    return False

src/aipass/hooks/docs/cadence_investigation.md

@@ -0,0 +1,109 @@
+# Cadence Investigation — DPLAN-0200
+
+Per-turn injection cadence mechanism for prompt loaders (global_loader, branch_loader, identity).
+Investigation only — no build. Findings for @devpulse.
+
+---
+
+## 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).
+
+`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`.
+
+A new session gets a new UUID — fresh counter automatically. No inheritance risk.
+
+## 2. Counter state location
+
+Engine has NO per-session state infrastructure today. `auto_process.py`'s `/tmp` sentinel is the closest precedent — existence-only, no data payload.
+
+**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.
+
+This is net-new state. The engine doesn't need to know about it — a shared cadence module handles it.
+
+## 3. Mechanism sketch — feasible, confirmed
+
+Shared module: `apps/handlers/prompt/cadence.py`
+
+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.
+
+```python
+# cadence.py sketch
+_turn = None  # process-level cache, reset each dispatch = each turn
+
+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
+
+def should_fire(offset, period=5):
+    turn = _load_and_increment()
+    if turn == 0:
+        return True  # first turn ALWAYS fires
+    return (turn % period) == offset
+```
+
+Each loader adds ONE guard line:
+
+```python
+from aipass.hooks.apps.handlers.prompt.cadence import should_fire
+
+def handle(hook_data):
+    if not should_fire(offset=0):  # different offset per loader
+        return {"stdout": "", "exit_code": 0}
+    # ... existing logic unchanged
+```
+
+**Stagger example (period=5):**
+
+| 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.
+
+---
+
+## 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.
+
+*Investigation by @hooks, 2026-06-08*

src/aipass/hooks/tests/test_branch_loader.py

@@ -10,7 +10,14 @@
 """Tests for handlers/prompt/branch_loader.py."""
 
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
+
+
+def _mock_cadence_fires():
+    """Return a mock cadence module where should_fire always returns True."""
+    mock = MagicMock()
+    mock.should_fire.return_value = True
+    return mock
 
 
 class TestBranchLoaderHandler:
@@ -25,7 +32,8 @@ class TestBranchLoaderHandler:
         prompt.write_text("# Test Branch\nSome instructions", encoding="utf-8")
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(tmp_path)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(tmp_path)})
 
         assert result["exit_code"] == 0
         assert "Branch Context:" in result["stdout"]
@@ -42,7 +50,8 @@ class TestBranchLoaderHandler:
         private.write_text("# Private Integration\nSecret stuff", encoding="utf-8")
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(tmp_path)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(tmp_path)})
 
         assert "Private Integration" in result["stdout"]
 
@@ -59,7 +68,8 @@ class TestBranchLoaderHandler:
         (integration / "private_prompt.md").write_text("Compass prompt", encoding="utf-8")
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(tmp_path)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(tmp_path)})
 
         assert "Branch prompt" in result["stdout"]
         assert "Compass prompt" in result["stdout"]
@@ -68,7 +78,8 @@ class TestBranchLoaderHandler:
         from aipass.hooks.apps.handlers.prompt.branch_loader import handle
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(tmp_path)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(tmp_path)})
 
         assert result["stdout"] == ""
 
@@ -80,7 +91,8 @@ class TestBranchLoaderHandler:
         nested.mkdir(parents=True)
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(nested)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(nested)})
 
         assert result["stdout"] == ""
 
@@ -96,7 +108,8 @@ class TestBranchLoaderHandler:
         nested.mkdir(parents=True)
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(nested)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(nested)})
 
         assert "Found it" in result["stdout"]
 
@@ -104,8 +117,9 @@ class TestBranchLoaderHandler:
         from aipass.hooks.apps.handlers.prompt.branch_loader import handle
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
-                result = handle({})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
+                    result = handle({})
 
         assert result["exit_code"] == 0
         assert result["stdout"] == ""
@@ -117,7 +131,8 @@ class TestBranchLoaderHandler:
         trinity.mkdir()
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(tmp_path)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(tmp_path)})
 
         assert result["stdout"] == ""
 
@@ -131,6 +146,7 @@ class TestBranchLoaderHandler:
         (aipass_dir / "aipass_local_prompt.md").write_text("content", encoding="utf-8")
 
         with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
-            result = handle({"cwd": str(tmp_path)})
+            with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                result = handle({"cwd": str(tmp_path)})
 
         assert "Source:" in result["stdout"]

src/aipass/hooks/tests/test_cadence.py

@@ -0,0 +1,425 @@
+# =================== AIPass ====================
+# Name: test_cadence.py
+# Version: 1.0.0
+# Description: Tests for cadence module (DPLAN-0200)
+# Branch: hooks
+# Created: 2026-06-08
+# Modified: 2026-06-08
+# =============================================
+
+"""Tests for apps/modules/cadence.py."""
+
+import json
+import importlib
+from unittest.mock import patch
+
+MODULE = "aipass.hooks.apps.modules.cadence"
+
+
+def _reset_module_globals():
+    """Reset module-level caches between tests."""
+    import aipass.hooks.apps.modules.cadence as mod
+
+    mod._turn = None
+    mod._config = None
+
+
+class TestShouldFire:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_turn_0_always_fires(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            assert should_fire("global") is True
+            assert json.loads(state_file.read_text())["turn"] == 0
+
+    def test_turn_0_fires_all_loaders(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            assert should_fire("global") is True
+            assert should_fire("branch") is True
+
+    def test_non_fire_turn_returns_false(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 0}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            assert should_fire("global") is False
+
+    def test_fire_turn_returns_true(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 3}))
+
+        config = tmp_path / "cadence.json"
+        config.write_text(json.dumps({"enabled": True, "period": 5, "loaders": {"global": {"offset": 4}}}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", config),
+        ):
+            assert should_fire("global") is True
+
+    def test_cadence_disabled_always_fires(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 1}))
+
+        config = tmp_path / "cadence.json"
+        config.write_text(json.dumps({"enabled": False}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", config),
+        ):
+            assert should_fire("global") is True
+
+    def test_no_session_id_fires(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {}, clear=False),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            env = dict(__import__("os").environ)
+            env.pop("CLAUDE_CODE_SESSION_ID", None)
+            with patch.dict("os.environ", env, clear=True):
+                assert should_fire("global") is True
+
+    def test_counter_increments_once_per_process(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 3}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            should_fire("global")
+            should_fire("branch")
+            data = json.loads(state_file.read_text())
+            assert data["turn"] == 4
+
+    def test_period_zero_always_fires(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 2}))
+
+        config = tmp_path / "cadence.json"
+        config.write_text(json.dumps({"enabled": True, "period": 0}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", config),
+        ):
+            assert should_fire("global") is True
+
+    def test_stagger_offsets(self, tmp_path):
+        config = tmp_path / "cadence.json"
+        config.write_text(
+            json.dumps({"enabled": True, "period": 5, "loaders": {"global": {"offset": 0}, "branch": {"offset": 2}}})
+        )
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 4}))
+
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", config),
+        ):
+            assert should_fire("global") is True
+            assert should_fire("branch") is False
+
+    def test_unknown_loader_uses_offset_zero(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 4}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            assert should_fire("unknown_loader") is True
+
+
+class TestResetCounter:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_reset_writes_minus_one(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import reset_counter
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 7}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+        ):
+            reset_counter()
+
+        data = json.loads(state_file.read_text())
+        assert data["turn"] == -1
+
+    def test_reset_then_next_turn_is_zero(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import reset_counter, should_fire
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 7}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            reset_counter()
+
+        _reset_module_globals()
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            assert should_fire("global") is True
+            data = json.loads(state_file.read_text())
+            assert data["turn"] == 0
+
+    def test_reset_no_session_id_is_noop(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import reset_counter
+
+        with patch(f"{MODULE}._GUARD_DIR", tmp_path):
+            env = dict(__import__("os").environ)
+            env.pop("CLAUDE_CODE_SESSION_ID", None)
+            with patch.dict("os.environ", env, clear=True):
+                reset_counter()
+
+        assert not list(tmp_path.glob("aipass-cadence-*"))
+
+    def test_reset_creates_file_if_missing(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import reset_counter
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+        ):
+            reset_counter()
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        assert state_file.exists()
+        assert json.loads(state_file.read_text())["turn"] == -1
+
+
+class TestConfig:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_defaults_used_when_no_config_file(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import _load_config
+
+        with patch(f"{MODULE}._CONFIG_PATH", tmp_path / "nonexistent.json"):
+            config = _load_config()
+
+        assert config["enabled"] is True
+        assert config["period"] == 5
+        assert config["loaders"]["global"]["offset"] == 0
+        assert config["loaders"]["branch"]["offset"] == 0
+
+    def test_config_deep_merges_over_defaults(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import _load_config
+
+        config_file = tmp_path / "cadence.json"
+        config_file.write_text(json.dumps({"period": 10, "loaders": {"global": {"offset": 3}}}))
+
+        with patch(f"{MODULE}._CONFIG_PATH", config_file):
+            config = _load_config()
+
+        assert config["period"] == 10
+        assert config["loaders"]["global"]["offset"] == 3
+        assert config["loaders"]["branch"]["offset"] == 0
+        assert config["enabled"] is True
+
+    def test_bad_config_falls_back_to_defaults(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import _load_config
+
+        config_file = tmp_path / "cadence.json"
+        config_file.write_text("not valid json{{{")
+
+        with patch(f"{MODULE}._CONFIG_PATH", config_file):
+            config = _load_config()
+
+        assert config["period"] == 5
+
+
+class TestDeepMerge:
+    def test_nested_merge(self):
+        from aipass.hooks.apps.modules.cadence import _deep_merge
+
+        base = {"a": 1, "b": {"c": 2, "d": 3}}
+        updates = {"b": {"c": 99}, "e": 4}
+        result = _deep_merge(base, updates)
+
+        assert result["a"] == 1
+        assert result["b"]["c"] == 99
+        assert result["b"]["d"] == 3
+        assert result["e"] == 4
+
+    def test_overwrites_non_dict(self):
+        from aipass.hooks.apps.modules.cadence import _deep_merge
+
+        base = {"a": [1, 2]}
+        result = _deep_merge(base, {"a": [3]})
+        assert result["a"] == [3]
+
+
+class TestPerSessionIsolation:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_different_sessions_use_different_files(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import should_fire
+
+        state_a = tmp_path / "aipass-cadence-session-a.json"
+        state_a.write_text(json.dumps({"turn": 4}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "session-a"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            should_fire("global")
+            data_a = json.loads(state_a.read_text())
+            assert data_a["turn"] == 5
+
+        _reset_module_globals()
+
+        state_b = tmp_path / "aipass-cadence-session-b.json"
+        assert not state_b.exists()
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "session-b"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            assert should_fire("global") is True
+            data_b = json.loads(state_b.read_text())
+            assert data_b["turn"] == 0
+
+
+class TestModuleInterface:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_handle_command_cadence_returns_true(self):
+        from aipass.hooks.apps.modules.cadence import handle_command
+
+        with patch(f"{MODULE}.print_introspection"):
+            assert handle_command("cadence", []) is True
+
+    def test_handle_command_unknown_returns_false(self):
+        from aipass.hooks.apps.modules.cadence import handle_command
+
+        assert handle_command("other", []) is False
+
+    def test_print_introspection_runs(self, tmp_path):
+        from aipass.hooks.apps.modules.cadence import print_introspection
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+        ):
+            print_introspection()
+
+
+class TestCompactIntegration:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_compact_handler_resets_cadence(self, tmp_path):
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 7}))
+
+        import aipass.hooks.apps.modules.cadence as cadence_mod
+
+        with (
+            patch.object(cadence_mod, "_GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+        ):
+            mock_cadence = importlib.import_module("aipass.hooks.apps.modules.cadence")
+            mock_cadence.reset_counter()
+
+        data = json.loads(state_file.read_text())
+        assert data["turn"] == -1
+
+
+class TestLoaderCadenceGuard:
+    def setup_method(self):
+        _reset_module_globals()
+
+    def test_global_loader_skips_on_non_fire_turn(self, tmp_path):
+        from aipass.hooks.apps.handlers.prompt.global_loader import handle
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 0}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+            patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"),
+        ):
+            result = handle({})
+
+        assert result["stdout"] == ""
+        assert result["exit_code"] == 0
+
+    def test_branch_loader_skips_on_non_fire_turn(self, tmp_path):
+        from aipass.hooks.apps.handlers.prompt.branch_loader import handle
+
+        state_file = tmp_path / "aipass-cadence-test-session.json"
+        state_file.write_text(json.dumps({"turn": 0}))
+
+        with (
+            patch(f"{MODULE}._GUARD_DIR", tmp_path),
+            patch.dict("os.environ", {"CLAUDE_CODE_SESSION_ID": "test-session"}),
+            patch(f"{MODULE}._CONFIG_PATH", tmp_path / "cadence.json"),
+            patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"),
+        ):
+            result = handle({})
+
+        assert result["stdout"] == ""
+        assert result["exit_code"] == 0

src/aipass/hooks/tests/test_global_loader.py

@@ -9,7 +9,14 @@
 
 """Tests for handlers/prompt/global_loader.py."""
 
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
+
+
+def _mock_cadence_fires():
+    """Return a mock cadence module where should_fire always returns True."""
+    mock = MagicMock()
+    mock.should_fire.return_value = True
+    return mock
 
 
 class TestGlobalLoaderHandler:
@@ -24,7 +31,8 @@ class TestGlobalLoaderHandler:
 
         with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
             with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
-                result = handle({})
+                with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                    result = handle({})
 
         assert result["exit_code"] == 0
         assert "AIPass Global" in result["stdout"]
@@ -36,7 +44,8 @@ class TestGlobalLoaderHandler:
         monkeypatch.chdir(tmp_path)
         with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
             with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
-                result = handle({})
+                with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                    result = handle({})
 
         assert result["exit_code"] == 0
         assert result["stdout"] == ""
@@ -51,7 +60,8 @@ class TestGlobalLoaderHandler:
 
         with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
             with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
-                result = handle({})
+                with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                    result = handle({})
 
         assert result["exit_code"] == 0
         assert result["stdout"] == "content"
@@ -69,7 +79,8 @@ class TestGlobalLoaderHandler:
 
         with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
             with patch.dict("os.environ", {"AIPASS_HOME": "/some/other/path"}):
-                result = handle({})
+                with patch("importlib.import_module", return_value=_mock_cadence_fires()):
+                    result = handle({})
 
         assert result["exit_code"] == 0
         assert "Project Prompt" in result["stdout"]