diff --git a/src/aipass/hooks/.seedgo/bypass.json b/src/aipass/hooks/.seedgo/bypass.json index e0e72134..d08e1023 100644 --- a/src/aipass/hooks/.seedgo/bypass.json +++ b/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.", diff --git a/src/aipass/hooks/README.md b/src/aipass/hooks/README.md index 6cc926c6..17cfac04 100644 --- a/src/aipass/hooks/README.md +++ b/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 diff --git a/src/aipass/hooks/apps/handlers/lifecycle/compact.py b/src/aipass/hooks/apps/handlers/lifecycle/compact.py index 1f4111e5..2ac7e88a 100644 --- a/src/aipass/hooks/apps/handlers/lifecycle/compact.py +++ b/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) diff --git a/src/aipass/hooks/apps/handlers/prompt/branch_loader.py b/src/aipass/hooks/apps/handlers/prompt/branch_loader.py index 4d2a63c2..bbde785e 100644 --- a/src/aipass/hooks/apps/handlers/prompt/branch_loader.py +++ b/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) diff --git a/src/aipass/hooks/apps/handlers/prompt/global_loader.py b/src/aipass/hooks/apps/handlers/prompt/global_loader.py index e9d74eab..ab7dafc4 100644 --- a/src/aipass/hooks/apps/handlers/prompt/global_loader.py +++ b/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()) diff --git a/src/aipass/hooks/apps/modules/cadence.py b/src/aipass/hooks/apps/modules/cadence.py new file mode 100644 index 00000000..2d4342ec --- /dev/null +++ b/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 diff --git a/src/aipass/hooks/docs/cadence_investigation.md b/src/aipass/hooks/docs/cadence_investigation.md new file mode 100644 index 00000000..50a43c57 --- /dev/null +++ b/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* diff --git a/src/aipass/hooks/tests/test_branch_loader.py b/src/aipass/hooks/tests/test_branch_loader.py index cc219d64..e0b1c376 100644 --- a/src/aipass/hooks/tests/test_branch_loader.py +++ b/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"] diff --git a/src/aipass/hooks/tests/test_cadence.py b/src/aipass/hooks/tests/test_cadence.py new file mode 100644 index 00000000..bb000231 --- /dev/null +++ b/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 diff --git a/src/aipass/hooks/tests/test_global_loader.py b/src/aipass/hooks/tests/test_global_loader.py index b20b86d8..3b93c26b 100644 --- a/src/aipass/hooks/tests/test_global_loader.py +++ b/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"]