period= offset= session=<8char>
+```
+
+Use `.info` (SystemLogger has no `.debug`). ALSO gate/dedup the "counter reset" log — it spammed ~8x per cluster; confirm PreCompact reset fires EXACTLY once and logs once.
+
+## FIX 3 — ACTION-GATED SOUND (the false signal Patrick HEARD)
+Right now `speak("global prompt")` / `speak("branch prompt")` is the FIRST line of each loader, BEFORE the `should_fire` check — so piper announces every turn even when the loader SKIPS injection. The voice lies. Patrick's rule: **if global/branch SKIP, they must be SILENT — sound ONLY on actual injection.**
+
+Build the **system-wide** version (Patrick wants it right for ALL hooks): handlers return an explicit `sound` key in their result dict, e.g. `{"stdout": content, "sound": "global prompt", "exit_code": 0}`; the engine plays it at `engine.py:208` inside the `if result["stdout"]:` block (or whenever the `sound` key is present) — ONE integration point, every hook auto action-gated + self-identifying. Remove the scattered leading `speak()` calls from the loaders. Preserve the gates/notifications that legitimately emit empty stdout (let them set the `sound` key explicitly). `is_muted()` still short-circuits.
+
+Sound architecture for reference: `hooks/apps/sound.py` `speak()`/`play()` → piper → aplay; mute flag `/tmp/aipass-hooks-muted`.
+
+## TEST PLAN (this is what 435 green MISSED — required)
+- **Model separate-process execution:** simulate N independent processes each calling `_load_and_increment` for the same turn (no shared module cache) and assert the counter advances EXACTLY ONCE. REWRITE `test_cadence.py:113` `test_counter_increments_once_per_process` (it encodes the invalid single-process assumption).
+- Assert the leapfrog is gone: two loaders in the same turn see the SAME turn number — both fire on offset-0 turns, both skip otherwise.
+- Assert reset → next turn = 0 = all fire (the turn<0 special-case survives the debounce).
+- Assert SKIP = silent (no `sound` key) AND logs `action=skipped` (not fired).
+- Assert flock prevents double-increment under simulated simultaneity.
+
+## ACCEPTANCE
+Multi-process simulation tests green + seedgo 100% + pyright 0. But do **NOT** claim "works" from unit tests alone — that is exactly what failed. devpulse will LIVE-VERIFY next session (prax monitor shows correct fire/skip, sound only on inject, counter advances once/turn). Report what you built + test results. NO git commits (devpulse commits). Reply via dispatch if blocked.
+
+Track in your FPLAN (reopen FPLAN-0249). This is the careful re-do — get it right, verify against the REAL execution model.
diff --git a/src/aipass/hooks/tests/test_announce.py b/src/aipass/hooks/tests/test_announce.py
index 3e921d86..66d8f640 100644
--- a/src/aipass/hooks/tests/test_announce.py
+++ b/src/aipass/hooks/tests/test_announce.py
@@ -1,16 +1,14 @@
# =================== AIPass ====================
# Name: test_announce.py
-# Version: 1.2.0
+# Version: 1.3.0
# Description: Tests for announce notification handler
# Branch: hooks
# Created: 2026-05-20
-# Modified: 2026-05-22
+# Modified: 2026-06-09
# =============================================
"""Tests for handlers/notification/announce.py."""
-from unittest.mock import patch
-
class TestAnnounceHandler:
"""Core handler behavior tests."""
@@ -18,17 +16,15 @@ class TestAnnounceHandler:
def test_handle_returns_result_dict(self):
from aipass.hooks.apps.handlers.notification.announce import handle
- with patch("aipass.hooks.apps.handlers.notification.announce.speak"):
- result = handle({})
+ result = handle({})
assert isinstance(result, dict)
assert result["stdout"] == ""
assert result["exit_code"] == 0
- def test_handle_speaks_notification_sound(self):
+ def test_handle_sets_sound_key(self):
from aipass.hooks.apps.handlers.notification.announce import handle
- with patch("aipass.hooks.apps.handlers.notification.announce.speak") as mock_speak:
- handle({})
+ result = handle({})
- mock_speak.assert_called_once_with("notification sound")
+ assert result["sound"] == "notification sound"
diff --git a/src/aipass/hooks/tests/test_auto_fix.py b/src/aipass/hooks/tests/test_auto_fix.py
index ed9ec706..d246c493 100644
--- a/src/aipass/hooks/tests/test_auto_fix.py
+++ b/src/aipass/hooks/tests/test_auto_fix.py
@@ -7,9 +7,15 @@
# Modified: 2026-05-22
# =============================================
-"""Tests for handlers/lifecycle/auto_fix.py."""
+"""Tests for handlers/lifecycle/auto_fix.py.
+
+NOTE: sound is action-gated via the result "sound" key — it is set to
+"auto fix diagnostics" only on the error-surfacing path; clean and skip
+paths stay silent (no "sound" key).
+"""
import json
+import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -61,32 +67,31 @@ class TestAutoFixSkips:
def test_skip_unknown_extension(self):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
- with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak"):
- result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/file.xyz"}})
+ result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/file.xyz"}})
assert result["stdout"] == ""
assert result["exit_code"] == 0
+ assert "sound" not in result
class TestAutofixPython:
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
- def test_python_no_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo, mock_speak):
+ def test_python_no_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/clean.py"}})
assert result["exit_code"] == 0
parsed = json.loads(result["stdout"])
assert parsed["systemMessage"] == "[diagnostics] ok"
+ assert "sound" not in result
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks")
- def test_python_syntax_error(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo, mock_speak):
+ def test_python_syntax_error(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
mock_py.return_value = ["SYNTAX: invalid syntax at line 5"]
@@ -96,25 +101,25 @@ class TestAutofixPython:
assert "additionalContext" in parsed.get("hookSpecificOutput", {})
assert "SYNTAX" in parsed["hookSpecificOutput"]["additionalContext"]
assert "1 error(s)" in parsed["systemMessage"]
+ assert result.get("sound") == "auto fix diagnostics"
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks")
- def test_python_ruff_lint_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo, mock_speak):
+ def test_python_ruff_lint_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
mock_py.return_value = ["LINT: bad.py:10:1: F401 unused import"]
result = handle({"tool_name": "Write", "tool_input": {"file_path": "/tmp/bad.py"}})
parsed = json.loads(result["stdout"])
assert "LINT" in parsed["hookSpecificOutput"]["additionalContext"]
+ assert result.get("sound") == "auto fix diagnostics"
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
- def test_python_pyright_errors(self, mock_py, mock_ruff_s, mock_seedgo, mock_speak):
+ def test_python_pyright_errors(self, mock_py, mock_ruff_s, mock_seedgo):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
with patch(
@@ -125,12 +130,12 @@ class TestAutofixPython:
parsed = json.loads(result["stdout"])
assert "TYPE: L42" in parsed["hookSpecificOutput"]["additionalContext"]
+ assert result.get("sound") == "auto fix diagnostics"
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
- def test_seedgo_violations_surfaced(self, mock_py, mock_ruff_s, mock_pyright, mock_speak):
+ def test_seedgo_violations_surfaced(self, mock_py, mock_ruff_s, mock_pyright):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
with patch(
@@ -141,15 +146,15 @@ class TestAutofixPython:
parsed = json.loads(result["stdout"])
assert "SEEDGO: missing file header" in parsed["hookSpecificOutput"]["additionalContext"]
+ assert result.get("sound") == "auto fix diagnostics"
class TestAutoFixStateFile:
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
- def test_state_file_written_on_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo, mock_speak):
+ def test_state_file_written_on_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
mock_ruff_s.return_value = [{"line": 5, "message": "F401: unused import"}]
@@ -160,8 +165,9 @@ class TestAutoFixStateFile:
try:
with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.STATE_FILE", state_path):
- handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/errors.py"}})
+ result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/errors.py"}})
+ assert result.get("sound") == "auto fix diagnostics"
assert state_path.exists()
state = json.loads(state_path.read_text(encoding="utf-8"))
assert len(state["errors"]) == 2
@@ -171,12 +177,11 @@ class TestAutoFixStateFile:
if state_path.exists():
state_path.unlink()
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
- def test_state_file_cleared_on_no_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo, mock_speak):
+ def test_state_file_cleared_on_no_errors(self, mock_py, mock_ruff_s, mock_pyright, mock_seedgo):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
with tempfile.NamedTemporaryFile(suffix=".json", delete=False, mode="w") as tf:
@@ -185,8 +190,9 @@ class TestAutoFixStateFile:
try:
with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.STATE_FILE", state_path):
- handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/clean.py"}})
+ result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/clean.py"}})
+ assert "sound" not in result
assert not state_path.exists()
finally:
if state_path.exists():
@@ -194,8 +200,7 @@ class TestAutoFixStateFile:
class TestAutoFixJson:
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
- def test_json_valid(self, mock_speak, tmp_path):
+ def test_json_valid(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
json_file = tmp_path / "good.json"
@@ -204,9 +209,9 @@ class TestAutoFixJson:
result = handle({"tool_name": "Edit", "tool_input": {"file_path": str(json_file)}})
parsed = json.loads(result["stdout"])
assert parsed["systemMessage"] == "[diagnostics] ok"
+ assert "sound" not in result
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
- def test_json_invalid_syntax(self, mock_speak, tmp_path):
+ def test_json_invalid_syntax(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
json_file = tmp_path / "bad.json"
@@ -215,9 +220,9 @@ class TestAutoFixJson:
result = handle({"tool_name": "Write", "tool_input": {"file_path": str(json_file)}})
parsed = json.loads(result["stdout"])
assert "JSON SYNTAX" in parsed["hookSpecificOutput"]["additionalContext"]
+ assert result.get("sound") == "auto fix diagnostics"
- @patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
- def test_json_corruption_detected(self, mock_speak, tmp_path):
+ def test_json_corruption_detected(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
json_file = tmp_path / "corrupt.json"
@@ -226,6 +231,7 @@ class TestAutoFixJson:
result = handle({"tool_name": "Edit", "tool_input": {"file_path": str(json_file)}})
parsed = json.loads(result["stdout"])
assert "EMOJI CORRUPTION" in parsed["hookSpecificOutput"]["additionalContext"]
+ assert result.get("sound") == "auto fix diagnostics"
class TestAutoFixSubprocessChecks:
@@ -254,6 +260,8 @@ class TestAutoFixSubprocessChecks:
errors = _check_ruff_lint("/tmp/bad.py")
assert len(errors) == 1
assert "LINT" in errors[0]
+ # bare "ruff" relies on PATH the hook env doesn't have — must go through the venv interpreter
+ assert mock_run.call_args[0][0][:3] == [sys.executable, "-m", "ruff"]
@patch("subprocess.run")
def test_check_ruff_format_drift(self, mock_run):
@@ -263,6 +271,7 @@ class TestAutoFixSubprocessChecks:
errors = _check_ruff_format("/tmp/unformatted.py")
assert len(errors) == 1
assert "FORMAT" in errors[0]
+ assert mock_run.call_args[0][0][:3] == [sys.executable, "-m", "ruff"]
@patch("subprocess.run")
def test_run_ruff_lint_structured_returns_dicts(self, mock_run):
@@ -280,6 +289,7 @@ class TestAutoFixSubprocessChecks:
assert len(errors) == 1
assert errors[0]["line"] == 5
assert "F401" in errors[0]["message"]
+ assert mock_run.call_args[0][0][:3] == [sys.executable, "-m", "ruff"]
@patch("subprocess.run")
def test_run_ruff_lint_structured_skips_claude_hooks(self, mock_run):
diff --git a/src/aipass/hooks/tests/test_branch_loader.py b/src/aipass/hooks/tests/test_branch_loader.py
index cc219d64..41c0e280 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:
@@ -24,12 +31,13 @@ class TestBranchLoaderHandler:
prompt = aipass_dir / "aipass_local_prompt.md"
prompt.write_text("# Test Branch\nSome instructions", encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ 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"]
assert "Some instructions" in result["stdout"]
+ assert result["sound"] == "branch prompt"
def test_loads_private_integrations(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
@@ -41,10 +49,11 @@ class TestBranchLoaderHandler:
private = integration / "private_prompt.md"
private.write_text("# Private Integration\nSecret stuff", encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({"cwd": str(tmp_path)})
assert "Private Integration" in result["stdout"]
+ assert result["sound"] == "branch prompt"
def test_loads_both_prompt_and_integrations(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
@@ -58,7 +67,7 @@ class TestBranchLoaderHandler:
integration.mkdir(parents=True)
(integration / "private_prompt.md").write_text("Compass prompt", encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({"cwd": str(tmp_path)})
assert "Branch prompt" in result["stdout"]
@@ -67,10 +76,11 @@ class TestBranchLoaderHandler:
def test_returns_empty_when_no_branch_root(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({"cwd": str(tmp_path)})
assert result["stdout"] == ""
+ assert "sound" not in result
def test_stops_at_repo_root(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
@@ -79,10 +89,11 @@ class TestBranchLoaderHandler:
nested = tmp_path / "some" / "deep" / "path"
nested.mkdir(parents=True)
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({"cwd": str(nested)})
assert result["stdout"] == ""
+ assert "sound" not in result
def test_walks_up_to_find_branch(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
@@ -95,7 +106,7 @@ class TestBranchLoaderHandler:
nested = tmp_path / "apps" / "handlers" / "security"
nested.mkdir(parents=True)
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({"cwd": str(nested)})
assert "Found it" in result["stdout"]
@@ -103,12 +114,15 @@ class TestBranchLoaderHandler:
def test_empty_hook_data(self):
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")):
+ # Path.cwd patch must be OUTSIDE the importlib patch — mock.patch uses
+ # importlib.import_module to resolve "pathlib", which the inner mock hijacks.
+ with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_no_prompt_file_but_has_branch_root(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
@@ -116,10 +130,11 @@ class TestBranchLoaderHandler:
trinity = tmp_path / ".trinity"
trinity.mkdir()
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({"cwd": str(tmp_path)})
assert result["stdout"] == ""
+ assert "sound" not in result
def test_includes_source_path_in_output(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
@@ -130,7 +145,7 @@ class TestBranchLoaderHandler:
aipass_dir.mkdir()
(aipass_dir / "aipass_local_prompt.md").write_text("content", encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
+ 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..016214ef
--- /dev/null
+++ b/src/aipass/hooks/tests/test_cadence.py
@@ -0,0 +1,491 @@
+# =================== 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.
+
+Cadence runs MULTI-PROCESS in production: each UserPromptSubmit hook is a
+separate OS process. Tests model that by resetting the module _turn cache
+between calls (= new process) and aging the state file past the mtime
+debounce window (= a real prior turn, not a sibling in the same turn).
+"""
+
+import json
+import importlib
+import os
+import time
+from unittest.mock import patch
+
+MODULE = "aipass.hooks.apps.modules.cadence"
+
+
+def _reset_module_globals():
+ """Reset module-level caches between tests (also = simulate a new process)."""
+ import aipass.hooks.apps.modules.cadence as mod
+
+ mod._turn = None
+ mod._config = None
+
+
+def _write_state(tmp_path, turn, token=-1, session="test-session", aged=True):
+ """Write a cadence state file. aged=True backdates mtime past the debounce
+ window so it reads as a PREVIOUS turn; aged=False = sibling in same turn."""
+ state_file = tmp_path / f"aipass-cadence-{session}.json"
+ state_file.write_text(json.dumps({"turn": turn, "token": token}))
+ if aged:
+ old = time.time() - 10
+ os.utime(state_file, (old, old))
+ return state_file
+
+
+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
+
+ _write_state(tmp_path, 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
+
+ _write_state(tmp_path, 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_across_sibling_processes(self, tmp_path):
+ """Each loader is a SEPARATE OS process. The counter must advance
+ exactly once per real turn no matter how many siblings call it."""
+ from aipass.hooks.apps.modules.cadence import should_fire
+
+ state_file = _write_state(tmp_path, 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")
+ for _ in range(4): # 4 more siblings, each a fresh process
+ _reset_module_globals()
+ should_fire("branch")
+ data = json.loads(state_file.read_text())
+ assert data["turn"] == 4
+
+ def test_sibling_processes_agree_on_turn_no_leapfrog(self, tmp_path):
+ """The S210 live bug: global saw turn N, branch saw N+1 — they
+ leapfrogged and never both fired. Both siblings must see the SAME
+ turn and make the SAME decision."""
+ from aipass.hooks.apps.modules.cadence import should_fire
+
+ _write_state(tmp_path, turn=4) # next real turn = 5 = fire (5 % 5 == 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 True
+ _reset_module_globals() # branch runs as a separate process
+ assert should_fire("branch") is True
+
+ def test_token_backstop_blocks_double_increment(self, tmp_path):
+ """Even past the debounce window, an unchanged transcript token means
+ no new turn happened — the counter must not advance."""
+ from aipass.hooks.apps.modules.cadence import should_fire
+
+ transcript = tmp_path / "transcript.jsonl"
+ transcript.write_text("x" * 100)
+ state_file = _write_state(tmp_path, turn=3, token=100)
+
+ 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", {"transcript_path": str(transcript)})
+ assert json.loads(state_file.read_text())["turn"] == 3
+
+ def test_reset_special_case_survives_debounce(self, tmp_path):
+ """turn < 0 (post-compact reset) must ALWAYS increment to 0, even when
+ the reset just happened (fresh mtime would normally debounce)."""
+ from aipass.hooks.apps.modules.cadence import should_fire
+
+ state_file = _write_state(tmp_path, turn=-1, aged=False)
+
+ 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_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}}})
+ )
+
+ _write_state(tmp_path, 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
+
+ _write_state(tmp_path, 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 = _write_state(tmp_path, turn=4, session="session-a")
+
+ 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):
+ """Skip = empty stdout AND no sound key — a skipped loader is SILENT."""
+ from aipass.hooks.apps.handlers.prompt.global_loader import handle
+
+ _write_state(tmp_path, turn=0) # next turn = 1 = skip
+
+ 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"),
+ ):
+ result = handle({})
+
+ assert result["stdout"] == ""
+ assert result["exit_code"] == 0
+ assert "sound" not in result
+
+ def test_branch_loader_skips_on_non_fire_turn(self, tmp_path):
+ """Skip = empty stdout AND no sound key — a skipped loader is SILENT."""
+ from aipass.hooks.apps.handlers.prompt.branch_loader import handle
+
+ _write_state(tmp_path, turn=0) # next turn = 1 = skip
+
+ 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"),
+ ):
+ result = handle({})
+
+ assert result["stdout"] == ""
+ assert result["exit_code"] == 0
+ assert "sound" not in result
diff --git a/src/aipass/hooks/tests/test_compact.py b/src/aipass/hooks/tests/test_compact.py
index 9e704546..c33663ee 100644
--- a/src/aipass/hooks/tests/test_compact.py
+++ b/src/aipass/hooks/tests/test_compact.py
@@ -29,25 +29,25 @@ class TestCompactHandler:
),
encoding="utf-8",
)
- with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value="Git branch: dev"):
- result = handle({"cwd": str(tmp_path)})
+ with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value="Git branch: dev"):
+ result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
assert "POST-COMPACT RECOVERY" in result["stdout"]
assert "Git branch: dev" in result["stdout"]
assert "did stuff" in result["stdout"]
assert "STATUS.local.md" not in result["stdout"]
+ assert result["sound"] == "pre compact"
def test_returns_recovery_when_no_branch_dir(self):
from aipass.hooks.apps.handlers.lifecycle.compact import handle
- with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
- result = handle({"cwd": "/tmp/nonexistent"})
+ with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
+ result = handle({"cwd": "/tmp/nonexistent"})
assert result["exit_code"] == 0
assert "POST-COMPACT RECOVERY" in result["stdout"]
+ assert result["sound"] == "pre compact"
def test_dispatched_agent_gets_save_warning(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.compact import handle
@@ -55,10 +55,9 @@ class TestCompactHandler:
trinity = tmp_path / ".trinity"
trinity.mkdir()
- with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
- with patch.dict("os.environ", {"AIPASS_SESSION_TYPE": "dispatched"}):
- result = handle({"cwd": str(tmp_path)})
+ with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
+ with patch.dict("os.environ", {"AIPASS_SESSION_TYPE": "dispatched"}):
+ result = handle({"cwd": str(tmp_path)})
assert "SAVE STATE NOW" in result["stdout"]
assert "STATUS.local.md" not in result["stdout"]
@@ -69,10 +68,9 @@ class TestCompactHandler:
trinity = tmp_path / ".trinity"
trinity.mkdir()
- with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
- with patch.dict("os.environ", {"AIPASS_SESSION_TYPE": ""}):
- result = handle({"cwd": str(tmp_path)})
+ with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
+ with patch.dict("os.environ", {"AIPASS_SESSION_TYPE": ""}):
+ result = handle({"cwd": str(tmp_path)})
assert "Recovery Protocol" in result["stdout"]
assert "STATUS.local.md" not in result["stdout"]
@@ -80,9 +78,8 @@ class TestCompactHandler:
def test_empty_hook_data(self):
from aipass.hooks.apps.handlers.lifecycle.compact import handle
- with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
- with patch("pathlib.Path.cwd", return_value=MagicMock(parts=("/", "tmp"))):
- result = handle({})
+ with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
+ with patch("pathlib.Path.cwd", return_value=MagicMock(parts=("/", "tmp"))):
+ result = handle({})
assert result["exit_code"] == 0
diff --git a/src/aipass/hooks/tests/test_email.py b/src/aipass/hooks/tests/test_email.py
index c2fb2804..fa65b1fc 100644
--- a/src/aipass/hooks/tests/test_email.py
+++ b/src/aipass/hooks/tests/test_email.py
@@ -1,10 +1,10 @@
# =================== AIPass ====================
# Name: test_email.py
-# Version: 1.2.0
+# Version: 1.3.0
# Description: Tests for email notification handler
# Branch: hooks
# Created: 2026-05-21
-# Modified: 2026-05-22
+# Modified: 2026-06-09
# =============================================
"""Tests for handlers/notification/email.py."""
@@ -41,12 +41,9 @@ class TestEmailHandler:
encoding="utf-8",
)
- with (
- 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({})
diff --git a/src/aipass/hooks/tests/test_global_loader.py b/src/aipass/hooks/tests/test_global_loader.py
index b20b86d8..30b3c01c 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:
@@ -22,24 +29,26 @@ class TestGlobalLoaderHandler:
prompt.write_text("# AIPass Global\nContext here", encoding="utf-8")
monkeypatch.chdir(tmp_path)
- with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
- with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
+ with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert "AIPass Global" in result["stdout"]
assert "Context here" in result["stdout"]
+ assert result["sound"] == "global prompt"
def test_returns_empty_when_file_missing(self, tmp_path, monkeypatch):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
monkeypatch.chdir(tmp_path)
- with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
- with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
+ with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_empty_hook_data(self, tmp_path, monkeypatch):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
@@ -49,8 +58,8 @@ class TestGlobalLoaderHandler:
(aipass_dir / "aipass_global_prompt.md").write_text("content", encoding="utf-8")
monkeypatch.chdir(tmp_path)
- with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
- with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
+ with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
@@ -67,9 +76,10 @@ class TestGlobalLoaderHandler:
(aipass_dir / "aipass_global_prompt.md").write_text("# Project Prompt", encoding="utf-8")
monkeypatch.chdir(project)
- with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
- with patch.dict("os.environ", {"AIPASS_HOME": "/some/other/path"}):
+ with patch.dict("os.environ", {"AIPASS_HOME": "/some/other/path"}):
+ with patch("importlib.import_module", return_value=_mock_cadence_fires()):
result = handle({})
assert result["exit_code"] == 0
assert "Project Prompt" in result["stdout"]
+ assert result["sound"] == "global prompt"
diff --git a/src/aipass/hooks/tests/test_identity.py b/src/aipass/hooks/tests/test_identity.py
index 58aab74a..3a1fb062 100644
--- a/src/aipass/hooks/tests/test_identity.py
+++ b/src/aipass/hooks/tests/test_identity.py
@@ -40,22 +40,22 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text(json.dumps(SAMPLE_PASSPORT), encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- result = handle({"cwd": str(tmp_path)})
+ result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
assert "devpulse Identity" in result["stdout"]
assert "orchestration_hub" in result["stdout"]
assert "Pragmatic" in result["stdout"]
+ assert result["sound"] == "identity"
def test_returns_empty_when_no_passport(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.identity import handle
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- result = handle({"cwd": str(tmp_path)})
+ result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_walks_up_to_find_passport(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.identity import handle
@@ -67,8 +67,7 @@ class TestIdentityHandler:
nested = tmp_path / "apps" / "handlers"
nested.mkdir(parents=True)
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- result = handle({"cwd": str(nested)})
+ result = handle({"cwd": str(nested)})
assert "devpulse Identity" in result["stdout"]
@@ -80,8 +79,7 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text(json.dumps(SAMPLE_PASSPORT), encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- result = handle({"cwd": str(tmp_path)})
+ result = handle({"cwd": str(tmp_path)})
out = result["stdout"]
assert "Path: src/aipass/devpulse" in out
@@ -100,21 +98,21 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text(json.dumps({"branch_info": {"branch_name": "test"}, "identity": {}}), encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- result = handle({"cwd": str(tmp_path)})
+ result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
assert "test Identity" in result["stdout"]
+ assert result["sound"] == "identity"
def test_empty_hook_data(self):
from aipass.hooks.apps.handlers.prompt.identity import handle
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
- result = handle({})
+ with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
+ result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_corrupt_passport_json(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.identity import handle
@@ -124,8 +122,8 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text("{broken json", encoding="utf-8")
- with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
- result = handle({"cwd": str(tmp_path)})
+ result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
diff --git a/src/aipass/hooks/tests/test_rollover.py b/src/aipass/hooks/tests/test_rollover.py
index 88aef64b..180b08d4 100644
--- a/src/aipass/hooks/tests/test_rollover.py
+++ b/src/aipass/hooks/tests/test_rollover.py
@@ -17,39 +17,37 @@ class TestRolloverHandler:
def test_no_repo_root_returns_empty(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=None):
- result = handle({})
+ with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=None):
+ result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_no_overdue_returns_empty(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue", return_value=[]):
- result = handle({})
+ with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
+ with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue", return_value=[]):
+ result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_overdue_triggers_rollover(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover.speak"):
- with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
- with patch(
- "aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue",
- return_value=[("devpulse", "local", "21/20 sessions")],
- ):
- with patch(
- "aipass.hooks.apps.handlers.lifecycle.rollover._run_rollover", return_value=(True, "ok")
- ):
- result = handle({})
+ with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
+ with patch(
+ "aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue",
+ return_value=[("devpulse", "local", "21/20 sessions")],
+ ):
+ with patch("aipass.hooks.apps.handlers.lifecycle.rollover._run_rollover", return_value=(True, "ok")):
+ result = handle({})
assert result["exit_code"] == 0
+ assert result["sound"] == "pre compact rollover"
def test_check_file_v2_sessions_overdue(self, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.rollover import _check_file
diff --git a/src/aipass/hooks/tests/test_sandbox.py b/src/aipass/hooks/tests/test_sandbox.py
new file mode 100644
index 00000000..83c94009
--- /dev/null
+++ b/src/aipass/hooks/tests/test_sandbox.py
@@ -0,0 +1,487 @@
+# =================== AIPass ====================
+# Name: test_sandbox.py
+# Version: 1.0.0
+# Description: Tests for sandbox wrapper module
+# Branch: hooks
+# Created: 2026-06-09
+# Modified: 2026-06-09
+# =============================================
+
+"""Tests for apps/modules/sandbox.py."""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="sandbox is Linux-only: bwrap mount namespaces")
+
+
+class TestBuildSrtConfig:
+ """Config generation from policy dict."""
+
+ def test_minimal_policy(self):
+ from aipass.hooks.apps.modules.sandbox import build_srt_config
+
+ with patch("aipass.hooks.apps.modules.sandbox._find_rg", return_value="/usr/bin/rg"):
+ config = build_srt_config({"allow_write": ["/tmp"]})
+
+ assert config["network"] == {"allowAllUnixSockets": True}
+ assert config["filesystem"]["allowWrite"] == ["/tmp"]
+ assert config["filesystem"]["denyRead"] == []
+ assert config["filesystem"]["denyWrite"] == []
+ assert config["ripgrep"]["command"] == "/usr/bin/rg"
+
+ def test_full_policy(self):
+ from aipass.hooks.apps.modules.sandbox import build_srt_config
+
+ policy = {
+ "allow_write": ["/tmp", "/home/user/branch"],
+ "deny_write": ["/home/user/branch/.git"],
+ "deny_read": ["/etc/shadow"],
+ }
+ with patch("aipass.hooks.apps.modules.sandbox._find_rg", return_value="/usr/bin/rg"):
+ config = build_srt_config(policy)
+
+ assert config["filesystem"]["allowWrite"] == ["/tmp", "/home/user/branch"]
+ assert config["filesystem"]["denyWrite"] == ["/home/user/branch/.git"]
+ assert config["filesystem"]["denyRead"] == ["/etc/shadow"]
+
+ def test_paths_stringified(self):
+ from aipass.hooks.apps.modules.sandbox import build_srt_config
+
+ policy = {"allow_write": [Path("/tmp"), Path("/home/x")]}
+ with patch("aipass.hooks.apps.modules.sandbox._find_rg", return_value="/usr/bin/rg"):
+ config = build_srt_config(policy)
+
+ assert all(isinstance(p, str) for p in config["filesystem"]["allowWrite"])
+
+ def test_missing_allow_write_raises(self):
+ from aipass.hooks.apps.modules.sandbox import build_srt_config
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._find_rg", return_value="/usr/bin/rg"),
+ pytest.raises(KeyError),
+ ):
+ build_srt_config({})
+
+
+class TestFindNode:
+ """Node.js binary discovery."""
+
+ def test_finds_node_on_path(self):
+ from aipass.hooks.apps.modules.sandbox import _find_node
+
+ with patch("aipass.hooks.apps.modules.sandbox.shutil.which", return_value="/usr/bin/node"):
+ assert _find_node() == "/usr/bin/node"
+
+ def test_raises_when_not_found(self):
+ from aipass.hooks.apps.modules.sandbox import _find_node
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox.shutil.which", return_value=None),
+ pytest.raises(FileNotFoundError, match="node not found"),
+ ):
+ _find_node()
+
+
+class TestFindRg:
+ """Ripgrep binary discovery."""
+
+ def test_finds_rg_on_path(self):
+ from aipass.hooks.apps.modules.sandbox import _find_rg
+
+ with patch("aipass.hooks.apps.modules.sandbox.shutil.which", return_value="/usr/bin/rg"):
+ assert _find_rg() == "/usr/bin/rg"
+
+ def test_falls_back_to_local_bin(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import _find_rg
+
+ fake_rg = tmp_path / ".local" / "bin" / "rg"
+ fake_rg.parent.mkdir(parents=True)
+ fake_rg.touch()
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox.shutil.which", return_value=None),
+ patch("aipass.hooks.apps.modules.sandbox.Path.home", return_value=tmp_path),
+ ):
+ assert _find_rg() == str(fake_rg)
+
+ def test_raises_when_not_found(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import _find_rg
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox.shutil.which", return_value=None),
+ patch("aipass.hooks.apps.modules.sandbox.Path.home", return_value=tmp_path),
+ pytest.raises(FileNotFoundError, match="ripgrep"),
+ ):
+ _find_rg()
+
+
+class TestResolveBwrapCommand:
+ """Bwrap command resolution via Node helper."""
+
+ def test_returns_bwrap_string(self):
+ from aipass.hooks.apps.modules.sandbox import resolve_bwrap_command
+
+ fake_result = MagicMock()
+ fake_result.returncode = 0
+ fake_result.stdout = "bwrap --ro-bind / / -- /bin/bash -c 'echo hello'"
+ fake_result.stderr = ""
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._find_node", return_value="/usr/bin/node"),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.run", return_value=fake_result),
+ ):
+ cmd = resolve_bwrap_command("echo hello", {"network": {}})
+
+ assert "bwrap" in cmd
+
+ def test_raises_on_nonzero_exit(self):
+ from aipass.hooks.apps.modules.sandbox import resolve_bwrap_command
+
+ fake_result = MagicMock()
+ fake_result.returncode = 1
+ fake_result.stdout = ""
+ fake_result.stderr = "some error"
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._find_node", return_value="/usr/bin/node"),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.run", return_value=fake_result),
+ pytest.raises(RuntimeError, match="srt resolve failed"),
+ ):
+ resolve_bwrap_command("echo hello", {"network": {}})
+
+ def test_raises_on_empty_output(self):
+ from aipass.hooks.apps.modules.sandbox import resolve_bwrap_command
+
+ fake_result = MagicMock()
+ fake_result.returncode = 0
+ fake_result.stdout = ""
+ fake_result.stderr = ""
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._find_node", return_value="/usr/bin/node"),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.run", return_value=fake_result),
+ pytest.raises(RuntimeError, match="empty command"),
+ ):
+ resolve_bwrap_command("echo hello", {"network": {}})
+
+ def test_resolver_cwd_is_not_branch_dir(self):
+ """srt resolves DANGEROUS_FILES relative to CWD. Using /var/tmp (or
+ fallback) prevents mount-point pollution in the branch directory."""
+ from aipass.hooks.apps.modules.sandbox import resolve_bwrap_command
+
+ fake_result = MagicMock()
+ fake_result.returncode = 0
+ fake_result.stdout = "bwrap --test"
+ fake_result.stderr = ""
+
+ captured_kwargs = {}
+
+ def capture_run(args, **kwargs):
+ captured_kwargs.update(kwargs)
+ return fake_result
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._find_node", return_value="/usr/bin/node"),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.run", side_effect=capture_run),
+ ):
+ resolve_bwrap_command("echo hello", {"network": {}})
+
+ cwd = captured_kwargs.get("cwd", "")
+ assert cwd and not cwd.startswith(str(Path.cwd()))
+
+ def test_cleans_up_temp_file(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import resolve_bwrap_command
+
+ fake_result = MagicMock()
+ fake_result.returncode = 0
+ fake_result.stdout = "bwrap --test"
+ fake_result.stderr = ""
+
+ created_files = []
+
+ def capture_run(args, **kwargs):
+ config_path = args[2]
+ created_files.append(config_path)
+ return fake_result
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._find_node", return_value="/usr/bin/node"),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.run", side_effect=capture_run),
+ ):
+ resolve_bwrap_command("echo hello", {"network": {}})
+
+ assert len(created_files) == 1
+ assert not Path(created_files[0]).exists()
+
+
+class TestSandboxLaunch:
+ """Full launch flow (mocked resolver)."""
+
+ def test_returns_popen(self):
+ from aipass.hooks.apps.modules.sandbox import sandbox_launch
+
+ fake_popen = MagicMock()
+
+ with (
+ patch(
+ "aipass.hooks.apps.modules.sandbox.resolve_bwrap_command",
+ return_value="bwrap --test -- /bin/bash -c 'echo hi'",
+ ),
+ patch(
+ "aipass.hooks.apps.modules.sandbox.build_srt_config",
+ return_value={"network": {}},
+ ),
+ patch(
+ "aipass.hooks.apps.modules.sandbox.subprocess.Popen",
+ return_value=fake_popen,
+ ) as mock_popen,
+ ):
+ result = sandbox_launch("echo hi", policy={"allow_write": ["/tmp"]})
+
+ assert result is fake_popen
+ call_args = mock_popen.call_args
+ assert call_args[0][0] == ["/bin/bash", "-c", "bwrap --test -- /bin/bash -c 'echo hi'"]
+
+ def test_passes_cwd(self):
+ from aipass.hooks.apps.modules.sandbox import sandbox_launch
+
+ with (
+ patch(
+ "aipass.hooks.apps.modules.sandbox.resolve_bwrap_command",
+ return_value="bwrap --test",
+ ),
+ patch(
+ "aipass.hooks.apps.modules.sandbox.build_srt_config",
+ return_value={"network": {}},
+ ),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.Popen") as mock_popen,
+ ):
+ sandbox_launch("echo hi", cwd="/tmp/test", policy={"allow_write": ["/tmp"]})
+
+ assert mock_popen.call_args[1]["cwd"] == "/tmp/test"
+
+ def test_passes_custom_env(self):
+ from aipass.hooks.apps.modules.sandbox import sandbox_launch
+
+ custom_env = {"PATH": "/usr/bin", "HOME": "/tmp"}
+
+ with (
+ patch(
+ "aipass.hooks.apps.modules.sandbox.resolve_bwrap_command",
+ return_value="bwrap --test",
+ ),
+ patch(
+ "aipass.hooks.apps.modules.sandbox.build_srt_config",
+ return_value={"network": {}},
+ ),
+ patch("aipass.hooks.apps.modules.sandbox.subprocess.Popen") as mock_popen,
+ ):
+ sandbox_launch("echo hi", policy={"allow_write": ["/tmp"]}, env=custom_env)
+
+ assert mock_popen.call_args[1]["env"] is custom_env
+
+
+class TestSrtResolveCwd:
+ """CWD selection for srt resolver — prevents mask-placeholder pollution."""
+
+ def test_returns_var_tmp_when_available(self):
+ from aipass.hooks.apps.modules.sandbox import _srt_resolve_cwd
+
+ mock_var = MagicMock()
+ mock_var.is_dir.return_value = True
+ mock_var.__str__ = MagicMock(return_value="/var/tmp")
+ with patch("aipass.hooks.apps.modules.sandbox._VAR_TMP", mock_var):
+ assert _srt_resolve_cwd() == "/var/tmp"
+
+ def test_falls_back_to_tempdir(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import _srt_resolve_cwd
+
+ with (
+ patch("aipass.hooks.apps.modules.sandbox._VAR_TMP") as mock_var,
+ patch("aipass.hooks.apps.modules.sandbox.tempfile.gettempdir", return_value=str(tmp_path)),
+ ):
+ mock_var.is_dir.return_value = False
+ assert _srt_resolve_cwd() == str(tmp_path)
+
+
+class TestBuildPolicy:
+ """Policy generation from branch path."""
+
+ def _make_branch(self, tmp_path, name, citizen_class="builder", is_devpulse=False):
+ """Create a minimal branch structure for testing."""
+ import json
+
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ (repo / ".git").mkdir()
+ src_aipass = repo / "src" / "aipass"
+ src_aipass.mkdir(parents=True)
+
+ branch = src_aipass / name
+ branch.mkdir()
+ trinity = branch / ".trinity"
+ trinity.mkdir()
+ passport = {
+ "branch_info": {"branch_name": "devpulse" if is_devpulse else name},
+ "identity": {"citizen_class": citizen_class},
+ }
+ (trinity / "passport.json").write_text(json.dumps(passport), encoding="utf-8")
+
+ for shared in ["system_logs", ".ai_central"]:
+ (repo / shared).mkdir()
+ (src_aipass / "memory" / "memory_pool").mkdir(parents=True)
+ (repo / "AIPASS_REGISTRY.json").touch()
+ (src_aipass / "flow" / "flow_json").mkdir(parents=True)
+
+ return branch
+
+ def _make_sibling(self, branch_path, name, with_mail=True, with_dashboard=True):
+ """Create a sibling branch with mail/dashboard."""
+ src_aipass = branch_path.parent
+ sibling = src_aipass / name
+ sibling.mkdir()
+ if with_mail:
+ (sibling / ".ai_mail.local").mkdir()
+ if with_dashboard:
+ (sibling / "DASHBOARD.local.json").touch()
+ return sibling
+
+ def test_builder_includes_own_tree(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ policy = build_policy(branch)
+ assert str(branch) in policy["allow_write"]
+
+ def test_builder_includes_tmp(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ policy = build_policy(branch)
+ assert "/tmp" in policy["allow_write"]
+
+ def test_builder_includes_shared_channels(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ repo = tmp_path / "repo"
+ policy = build_policy(branch)
+ assert str(repo / "system_logs") in policy["allow_write"]
+ assert str(repo / ".ai_central") in policy["allow_write"]
+ assert str(repo / "AIPASS_REGISTRY.json") in policy["allow_write"]
+
+ def test_builder_excludes_git(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ repo = tmp_path / "repo"
+ policy = build_policy(branch)
+ assert str(repo / ".git") not in policy["allow_write"]
+
+ def test_devpulse_includes_git(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "devpulse", is_devpulse=True)
+ repo = tmp_path / "repo"
+ policy = build_policy(branch)
+ assert str(repo / ".git") in policy["allow_write"]
+
+ def test_sibling_mail_writable(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ sibling = self._make_sibling(branch, "hooks")
+ policy = build_policy(branch)
+ assert str(sibling / ".ai_mail.local") in policy["allow_write"]
+
+ def test_sibling_dashboard_writable(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ sibling = self._make_sibling(branch, "hooks")
+ policy = build_policy(branch)
+ assert str(sibling / "DASHBOARD.local.json") in policy["allow_write"]
+
+ def test_sibling_source_not_writable(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ sibling = self._make_sibling(branch, "hooks")
+ policy = build_policy(branch)
+ assert str(sibling) not in policy["allow_write"]
+
+ def test_policy_shape(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ policy = build_policy(branch)
+ assert "allow_write" in policy
+ assert "deny_write" in policy
+ assert "deny_read" in policy
+ secret = str(tmp_path / "repo" / ".ai_central" / "broker_secret")
+ assert policy["deny_write"] == [secret]
+ assert policy["deny_read"] == [secret]
+
+ def test_broker_secret_masked_for_all_roles(self, tmp_path):
+ """The broker secret sits inside writable .ai_central — it must be
+ deny_read AND deny_write for every role, or a sandboxed agent could
+ read it and forge a devpulse identity to the broker."""
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ for name in ("seedgo", "devpulse"):
+ base = tmp_path / f"case_{name}"
+ base.mkdir()
+ branch = self._make_branch(base, name)
+ repo_root = base / "repo"
+ policy = build_policy(branch)
+ secret = str(repo_root / ".ai_central" / "broker_secret")
+ assert secret in policy["deny_read"]
+ assert secret in policy["deny_write"]
+ assert str(repo_root / ".ai_central") in policy["allow_write"]
+
+ def test_claude_project_dir_included(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ branch = self._make_branch(tmp_path, "seedgo")
+ encoded = str(branch.resolve()).replace("/", "-")
+ claude_proj = tmp_path / ".claude" / "projects" / encoded
+ claude_proj.mkdir(parents=True)
+
+ with patch("aipass.hooks.apps.modules.sandbox.Path.home", return_value=tmp_path):
+ policy = build_policy(branch)
+
+ assert str(claude_proj) in policy["allow_write"]
+
+ def test_no_repo_root_raises(self, tmp_path):
+ from aipass.hooks.apps.modules.sandbox import build_policy
+
+ bare = tmp_path / "no_repo" / "branch"
+ bare.mkdir(parents=True)
+ with pytest.raises(FileNotFoundError, match="No .git found"):
+ build_policy(bare)
+
+
+class TestHandleCommand:
+ """Drone routing for sandbox module."""
+
+ def test_sandbox_no_args_calls_introspection(self):
+ from aipass.hooks.apps.modules.sandbox import handle_command
+
+ result = handle_command("sandbox", [])
+ assert result is True
+
+ def test_sandbox_help(self):
+ from aipass.hooks.apps.modules.sandbox import handle_command
+
+ result = handle_command("sandbox", ["--help"])
+ assert result is True
+
+ def test_unknown_command_returns_false(self):
+ from aipass.hooks.apps.modules.sandbox import handle_command
+
+ result = handle_command("other", [])
+ assert result is False
diff --git a/src/aipass/hooks/tests/test_stop_sound.py b/src/aipass/hooks/tests/test_stop_sound.py
index a0a1d6b0..f554ea45 100644
--- a/src/aipass/hooks/tests/test_stop_sound.py
+++ b/src/aipass/hooks/tests/test_stop_sound.py
@@ -1,16 +1,14 @@
# =================== AIPass ====================
# Name: test_stop_sound.py
-# Version: 1.2.0
+# Version: 1.3.0
# Description: Tests for stop_sound notification handler
# Branch: hooks
# Created: 2026-05-20
-# Modified: 2026-05-22
+# Modified: 2026-06-09
# =============================================
"""Tests for handlers/notification/stop_sound.py."""
-from unittest.mock import patch
-
class TestStopSoundHandler:
"""Core handler behavior tests."""
@@ -18,26 +16,23 @@ class TestStopSoundHandler:
def test_handle_returns_result_dict(self):
from aipass.hooks.apps.handlers.notification.stop_sound import handle
- with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak"):
- result = handle({})
+ result = handle({})
assert isinstance(result, dict)
assert result["stdout"] == ""
assert result["exit_code"] == 0
- def test_handle_speaks_stop_sound(self):
+ def test_handle_sets_sound_key(self):
from aipass.hooks.apps.handlers.notification.stop_sound import handle
- with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak") as mock_speak:
- 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
diff --git a/src/aipass/hooks/tests/test_subagent_gate.py b/src/aipass/hooks/tests/test_subagent_gate.py
index 82563d8e..1f4e098d 100644
--- a/src/aipass/hooks/tests/test_subagent_gate.py
+++ b/src/aipass/hooks/tests/test_subagent_gate.py
@@ -18,24 +18,23 @@ from aipass.hooks.apps.handlers.security.subagent_gate import handle
class TestSubagentGateHandler:
def test_no_repo_root_allows(self):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root", return_value=None):
- with patch("aipass.hooks.apps.handlers.security.subagent_gate.speak"):
- result = handle({"cwd": "/tmp/nowhere"})
+ result = handle({"cwd": "/tmp/nowhere"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
def test_no_modified_files_allows(self):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root", return_value=None):
- with patch("aipass.hooks.apps.handlers.security.subagent_gate.speak"):
- result = handle({"cwd": "/tmp/somewhere"})
+ result = handle({"cwd": "/tmp/somewhere"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
@patch("aipass.hooks.apps.handlers.security.subagent_gate._check_hook_readme_accountability", return_value=None)
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_modified_files_no_violations_allows(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
+ def test_modified_files_no_violations_allows(self, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
mock_root.return_value = Path("/fake/repo")
@@ -43,13 +42,13 @@ class TestSubagentGateHandler:
result = handle({"cwd": "/fake/repo/src/aipass/hooks"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
@patch("aipass.hooks.apps.handlers.security.subagent_gate._check_hook_readme_accountability", return_value=None)
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_violations_blocks(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
+ def test_violations_blocks(self, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
mock_root.return_value = Path("/fake/repo")
@@ -62,10 +61,10 @@ class TestSubagentGateHandler:
assert "Missing docstring" in parsed["reason"]
assert "No tests" in parsed["reason"]
assert "bad.py" in parsed["reason"]
+ assert result["sound"] == "subagent gate"
@patch("subprocess.run")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_skip_claude_hooks_from_modified_files(self, mock_speak, mock_run, tmp_path):
+ def test_skip_claude_hooks_from_modified_files(self, mock_run, tmp_path):
src = tmp_path / "src" / "aipass" / "hooks"
src.mkdir(parents=True)
@@ -91,8 +90,7 @@ class TestSubagentGateHandler:
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_readme_accountability_advisory(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
+ def test_readme_accountability_advisory(self, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
mock_root.return_value = Path("/fake/repo")
@@ -106,18 +104,18 @@ class TestSubagentGateHandler:
parsed = json.loads(result["stdout"])
assert parsed["decision"] == "allow"
assert "README" in parsed["reason"]
+ assert "sound" not in result
def test_empty_hook_data_allows(self):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root", return_value=None):
- with patch("aipass.hooks.apps.handlers.security.subagent_gate.speak"):
- result = handle({})
+ result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_exception_in_get_modified_allows(self, mock_speak, mock_root, mock_modified):
+ def test_exception_in_get_modified_allows(self, mock_root, mock_modified):
from pathlib import Path
mock_root.return_value = Path("/fake/repo")
@@ -125,6 +123,7 @@ class TestSubagentGateHandler:
result = handle({"cwd": "/fake/repo/src/aipass/hooks"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
class TestSubagentGateExternalProject:
@@ -161,8 +160,7 @@ class TestSubagentGateExternalProject:
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_violations_block_external_project(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
+ def test_violations_block_external_project(self, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
mock_root.return_value = Path("/fake/vera")
@@ -173,13 +171,13 @@ class TestSubagentGateExternalProject:
parsed = json.loads(result["stdout"])
assert parsed["decision"] == "block"
assert "Missing docstring" in parsed["reason"]
+ assert result["sound"] == "subagent gate"
@patch("aipass.hooks.apps.handlers.security.subagent_gate._check_hook_readme_accountability", return_value=None)
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
- @patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
- def test_clean_files_allow_external_project(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
+ def test_clean_files_allow_external_project(self, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
mock_root.return_value = Path("/fake/vera")
@@ -187,3 +185,4 @@ class TestSubagentGateExternalProject:
result = handle({"cwd": "/fake/vera/src/vera_studio/quality"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
+ assert "sound" not in result
diff --git a/src/aipass/hooks/tests/test_tool_sound.py b/src/aipass/hooks/tests/test_tool_sound.py
index 0e9cb1a7..107d70f7 100644
--- a/src/aipass/hooks/tests/test_tool_sound.py
+++ b/src/aipass/hooks/tests/test_tool_sound.py
@@ -1,16 +1,14 @@
# =================== AIPass ====================
# Name: test_tool_sound.py
-# Version: 1.2.0
+# Version: 1.3.0
# Description: Tests for tool_sound notification handler
# Branch: hooks
# Created: 2026-05-19
-# Modified: 2026-05-22
+# Modified: 2026-06-09
# =============================================
"""Tests for handlers/notification/tool_sound.py."""
-from unittest.mock import patch
-
class TestToolSoundHandler:
"""Core handler behavior tests."""
@@ -18,8 +16,7 @@ class TestToolSoundHandler:
def test_handle_returns_result_dict(self):
from aipass.hooks.apps.handlers.notification.tool_sound import handle
- with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak"):
- result = handle({"tool_name": "Bash"})
+ result = handle({"tool_name": "Bash"})
assert isinstance(result, dict)
assert "stdout" in result
@@ -27,26 +24,23 @@ class TestToolSoundHandler:
assert result["stdout"] == ""
assert result["exit_code"] == 0
- def test_speaks_tool_name(self):
+ def test_sound_key_includes_tool_name(self):
from aipass.hooks.apps.handlers.notification.tool_sound import handle
- with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
- 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", "") == ""
diff --git a/src/aipass/memory/apps/handlers/monitor/memory_watcher.py b/src/aipass/memory/apps/handlers/monitor/memory_watcher.py
index 18905626..2ee993cf 100644
--- a/src/aipass/memory/apps/handlers/monitor/memory_watcher.py
+++ b/src/aipass/memory/apps/handlers/monitor/memory_watcher.py
@@ -43,11 +43,11 @@ except ImportError:
FileSystemEventHandler = object # type: ignore[assignment,misc]
logger.info("Optional dependency 'watchdog' not available")
-# Handler imports (relative within package)
-from aipass.memory.apps.handlers.tracking.line_counter import update_line_count
-from aipass.memory.apps.handlers.monitor.detector import check_single_file
-from aipass.prax.apps.modules.logger import get_system_logger
-from aipass.memory.apps.handlers.json import json_handler
+# Handler imports (relative within package — after conditional watchdog block)
+from aipass.memory.apps.handlers.tracking.line_counter import update_line_count # noqa: E402
+from aipass.memory.apps.handlers.monitor.detector import check_single_file # noqa: E402
+from aipass.prax.apps.modules.logger import get_system_logger # noqa: E402
+from aipass.memory.apps.handlers.json import json_handler # noqa: E402
logger = get_system_logger()
@@ -200,8 +200,6 @@ def check_and_rollover() -> Dict[str, Any]:
return results
# Check each branch for memory files over limit
- # Also sync current_lines metadata to keep it accurate
- lines_synced = 0
for branch_path in branch_paths:
branch = Path(branch_path)
# Find memory files in .trinity/ subdirectory
@@ -213,33 +211,23 @@ def check_and_rollover() -> Dict[str, Any]:
results["files_checked"] += 1
try:
- line_count = len(memory_file.read_text(encoding="utf-8").splitlines())
+ # Auto-heal: reconcile file against template (strips orphan keys)
+ from aipass.memory.apps.handlers.schema.normalize import normalize_memory_file
- # Sync current_lines metadata if stale
- try:
- import json as _json
-
- _data = _json.loads(memory_file.read_text(encoding="utf-8"))
- meta_lines = _data.get("document_metadata", {}).get("status", {}).get("current_lines")
- if meta_lines != line_count:
- sync_result = update_line_count(memory_file)
- if sync_result.get("success"):
- lines_synced += 1
- except Exception as e:
- logger.warning(f"[memory_watcher] Non-critical metadata sync failed for {memory_file}: {e}")
+ normalize_memory_file(memory_file)
# Use detector for trigger decision (handles both v1 line-based and v2 entry-count)
from aipass.memory.apps.handlers.monitor.detector import _should_rollover
- triggered, _, _, _, _ = _should_rollover(memory_file)
+ triggered, current_lines, _, _, _ = _should_rollover(memory_file)
if triggered:
results["files_over_limit"].append(
- {"file": str(memory_file), "lines": line_count, "threshold": 0}
+ {"file": str(memory_file), "lines": current_lines, "threshold": 0}
)
except Exception as e:
logger.warning(f"[memory_watcher] Failed to read memory file {memory_file}: {e}")
- results["lines_synced"] = lines_synced
+ results["lines_synced"] = 0
# Trigger rollover if any files are over limit
if results["files_over_limit"]:
@@ -537,19 +525,23 @@ class MemoryFileWatcher(FileSystemEventHandler): # type: ignore[misc]
logger.info(f"[memory_watcher] Detected modification: {file_path.name}")
- # Step 1: Update line count metadata
+ # Step 1: Auto-heal schema drift (strips orphan keys)
+ from aipass.memory.apps.handlers.schema.normalize import normalize_memory_file
+
+ norm_result = normalize_memory_file(file_path)
+ if norm_result.get("changes"):
+ self._recent_modifications.add(file_key)
+
+ # Step 2: Update health check metadata
update_result = update_line_count(file_path)
if not update_result["success"]:
logger.error(
- f"[memory_watcher] Failed to update line count for {file_path.name}: {update_result.get('error')}"
+ f"[memory_watcher] Failed to update metadata for {file_path.name}: {update_result.get('error')}"
)
return
- current_lines = update_result.get("lines", 0)
- logger.info(f"[memory_watcher] Updated {file_path.name}: {current_lines} lines")
-
- # Step 2: Check if rollover needed
+ # Step 3: Check if rollover needed
check_result = check_single_file(file_path)
if not check_result["success"]:
diff --git a/src/aipass/memory/apps/handlers/schema/normalize.py b/src/aipass/memory/apps/handlers/schema/normalize.py
index 897c0d9d..bd39f6dc 100644
--- a/src/aipass/memory/apps/handlers/schema/normalize.py
+++ b/src/aipass/memory/apps/handlers/schema/normalize.py
@@ -1,24 +1,17 @@
# =================== AIPass ====================
# Name: normalize.py
# Description: Memory File Schema Normalizer
-# Version: 0.2.0
+# Version: 0.3.0
# Created: 2026-01-22
-# Modified: 2026-03-06
+# Modified: 2026-06-08
# =============================================
"""
Memory File Schema Normalizer
-Fixes inconsistent schema in memory JSON files:
-1. Moves root-level 'limits' into document_metadata.limits
-2. Removes redundant root-level 'status'
-3. Removes auto_compress_at (redundant with max_lines)
-4. Ensures document_metadata.status has current_lines
-
-Supports two schema versions:
- v1 (schema_version <2.0.0): { "limits": { "max_lines": N } }
- v2 (schema_version >=2.0.0): { "limits": { "max_sessions": N, "max_key_learnings": N,
- "session_summary_max_chars": N, "learning_value_max_chars": N } }
+Reconciles memory files against their canonical template schema.
+Strips any key not present in the template at every level (root,
+document_metadata, limits, status). Template = the whole truth.
"""
import json
@@ -31,6 +24,36 @@ from aipass.memory.apps.handlers.json import json_handler
logger = get_system_logger()
+_MEMORY_ROOT = Path(__file__).parents[3] # normalize.py -> schema/ -> handlers/ -> apps/ -> memory/
+
+
+def _load_template(file_path: Path) -> Dict[str, Any] | None:
+ """Load the matching template for a memory file (local or observations)."""
+ templates_dir = _MEMORY_ROOT / "templates"
+ name = file_path.name.lower()
+
+ if "local" in name:
+ tmpl_path = templates_dir / "LOCAL.template.json"
+ elif "observation" in name:
+ tmpl_path = templates_dir / "OBSERVATIONS.template.json"
+ else:
+ return None
+
+ try:
+ with open(tmpl_path, "r", encoding="utf-8") as f:
+ return json.load(f)
+ except Exception as e:
+ logger.warning(f"[normalize] Failed to load template {tmpl_path}: {e}")
+ return None
+
+
+def _strip_orphan_keys(data: Dict, allowed: set, level_name: str, changes: list) -> None:
+ """Remove keys from data that aren't in the allowed set."""
+ orphans = set(data.keys()) - allowed
+ for key in orphans:
+ del data[key]
+ changes.append(f"Stripped orphan '{key}' from {level_name}")
+
def _find_repo_root() -> Path:
"""Walk up from this file to find repo root (contains AIPASS_REGISTRY.json)."""
@@ -43,7 +66,7 @@ def _find_repo_root() -> Path:
def normalize_memory_file(file_path: Path, dry_run: bool = False) -> Dict[str, Any]:
"""
- Normalize schema for a single memory file.
+ Normalize a memory file against its canonical template.
Args:
file_path: Path to memory JSON file
@@ -71,59 +94,53 @@ def normalize_memory_file(file_path: Path, dry_run: bool = False) -> Dict[str, A
metadata = data["document_metadata"]
- # 1. Move root 'limits' into document_metadata.limits
+ # Legacy fix: move root 'limits' into document_metadata.limits
if "limits" in data and "limits" not in metadata:
metadata["limits"] = data.pop("limits")
changes.append("Moved root 'limits' into document_metadata")
elif "limits" in data and "limits" in metadata:
- # Both exist - merge, preferring document_metadata values
root_limits = data.pop("limits")
for key, val in root_limits.items():
if key not in metadata["limits"]:
metadata["limits"][key] = val
changes.append("Merged root 'limits' into document_metadata.limits")
- # 2. Remove root 'status' (redundant)
+ # Legacy fix: move root 'status' into document_metadata.status
if "status" in data:
- root_status = data.pop("status")
- # If document_metadata.status doesn't have current_lines, copy it
+ data.pop("status")
if "status" not in metadata:
metadata["status"] = {}
- if "current_lines" not in metadata["status"] and "current_lines" in root_status:
- metadata["status"]["current_lines"] = root_status["current_lines"]
changes.append("Removed redundant root 'status'")
- # 3. Remove auto_compress_at from document_metadata.status (redundant with max_lines)
- if "status" in metadata and "auto_compress_at" in metadata["status"]:
- del metadata["status"]["auto_compress_at"]
- changes.append("Removed redundant 'auto_compress_at'")
-
- # 4. Remove unused limits fields (max_word_count, max_token_count - no code uses these)
- # Preserve v2 fields: max_sessions, max_key_learnings, session_summary_max_chars, learning_value_max_chars,
- # max_observations, max_lines, note
- if "limits" in metadata:
- for unused_field in ["max_word_count", "max_token_count"]:
- if unused_field in metadata["limits"]:
- del metadata["limits"][unused_field]
- changes.append(f"Removed unused '{unused_field}'")
-
- # 4. Ensure status has required fields
+ # Ensure status has required fields
if "status" not in metadata:
metadata["status"] = {}
- if "current_lines" not in metadata["status"]:
- # Count actual lines
- try:
- with open(file_path, "r", encoding="utf-8") as f:
- metadata["status"]["current_lines"] = len(f.readlines())
- changes.append("Added current_lines count")
- except Exception as e:
- logger.warning(f"[normalize] Failed to count lines in {file_path}: {e}")
-
if "last_health_check" not in metadata["status"]:
metadata["status"]["last_health_check"] = datetime.now().strftime("%Y-%m-%d")
changes.append("Added last_health_check")
+ # Template-conformance: strip orphan keys at every level
+ template = _load_template(file_path)
+ if template is not None:
+ tmpl_meta = template.get("document_metadata", {})
+
+ # Root level
+ _strip_orphan_keys(data, set(template.keys()), "root", changes)
+
+ # document_metadata level
+ _strip_orphan_keys(metadata, set(tmpl_meta.keys()), "document_metadata", changes)
+
+ # limits level
+ tmpl_limits = tmpl_meta.get("limits", {})
+ if "limits" in metadata:
+ _strip_orphan_keys(metadata["limits"], set(tmpl_limits.keys()), "limits", changes)
+
+ # status level
+ tmpl_status = tmpl_meta.get("status", {})
+ if "status" in metadata:
+ _strip_orphan_keys(metadata["status"], set(tmpl_status.keys()), "status", changes)
+
# Write if changes made and not dry run
if changes and not dry_run:
try:
diff --git a/src/aipass/memory/apps/handlers/tracking/line_counter.py b/src/aipass/memory/apps/handlers/tracking/line_counter.py
index 1ae44339..a6464126 100644
--- a/src/aipass/memory/apps/handlers/tracking/line_counter.py
+++ b/src/aipass/memory/apps/handlers/tracking/line_counter.py
@@ -62,24 +62,20 @@ def _count_physical_lines(file_path: Path) -> int:
def update_line_count(file_path: Path) -> Dict[str, Any]:
"""
- Update current_lines in document_metadata.status
-
- Reads file, counts lines, updates metadata field using safe json_handler.
+ Update health check metadata after file modification.
Args:
file_path: Path to memory JSON file
Returns:
- Dict with success status and updated line count
+ Dict with success status
"""
if not file_path.exists():
return {"success": False, "error": f"File not found: {file_path}"}
- # Count lines
line_count = _count_physical_lines(file_path)
- # Update metadata using safe handler (atomic write)
- result = update_metadata(file_path, current_lines=line_count, last_health_check=datetime.now().strftime("%Y-%m-%d"))
+ result = update_metadata(file_path, last_health_check=datetime.now().strftime("%Y-%m-%d"))
if not result["success"]:
return {"success": False, "error": f"Failed to update metadata: {result['error']}"}
diff --git a/src/aipass/memory/templates/LOCAL.template.json b/src/aipass/memory/templates/LOCAL.template.json
index c050ba85..3ffb46d2 100644
--- a/src/aipass/memory/templates/LOCAL.template.json
+++ b/src/aipass/memory/templates/LOCAL.template.json
@@ -23,8 +23,7 @@
},
"status": {
"health": "healthy",
- "last_health_check": "{{DATE}}",
- "current_lines": 0
+ "last_health_check": "{{DATE}}"
}
},
"key_learnings": {},
diff --git a/src/aipass/memory/templates/OBSERVATIONS.template.json b/src/aipass/memory/templates/OBSERVATIONS.template.json
index 15d36d99..31a7afb1 100644
--- a/src/aipass/memory/templates/OBSERVATIONS.template.json
+++ b/src/aipass/memory/templates/OBSERVATIONS.template.json
@@ -13,12 +13,11 @@
"{{BRANCHNAME}}"
],
"limits": {
- "max_lines": 600,
- "note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_lines exceeded."
+ "max_observations": 25,
+ "note": "DO NOT trim, prune, or delete entries. Auto-rollover to @memory when max_observations exceeded."
},
"status": {
"health": "healthy",
- "current_lines": 0,
"last_health_check": "{{DATE}}"
}
},
diff --git a/src/aipass/memory/tests/test_handlers.py b/src/aipass/memory/tests/test_handlers.py
index 7f8cd6da..ba7e0750 100644
--- a/src/aipass/memory/tests/test_handlers.py
+++ b/src/aipass/memory/tests/test_handlers.py
@@ -350,7 +350,7 @@ class TestCountPhysicalLines:
def test_counts_lines_correctly(self, monkeypatch, tmp_path):
lc, _ = _import_line_counter(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
f.write_text("line1\nline2\nline3\n", encoding="utf-8")
assert lc._count_physical_lines(f) == 3
@@ -376,7 +376,7 @@ class TestUpdateLineCount:
def test_updates_line_count_successfully(self, monkeypatch, tmp_path):
lc, mocks = _import_line_counter(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
f.write_text('{\n "a": 1\n}\n', encoding="utf-8")
result = lc.update_line_count(f)
@@ -387,7 +387,7 @@ class TestUpdateLineCount:
def test_reports_failure_when_metadata_update_fails(self, monkeypatch, tmp_path):
lc, mocks = _import_line_counter(monkeypatch)
mocks["memory_files"].update_metadata.return_value = {"success": False, "error": "write error"}
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
f.write_text("{}\n", encoding="utf-8")
result = lc.update_line_count(f)
@@ -413,12 +413,12 @@ class TestNormalizeMemoryFile:
def test_moves_root_limits_into_metadata(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
self._write_json(
f,
{
- "document_metadata": {"status": {"current_lines": 10}},
- "limits": {"max_lines": 600},
+ "document_metadata": {"status": {}},
+ "limits": {"max_sessions": 20},
"sessions": [],
},
)
@@ -427,19 +427,19 @@ class TestNormalizeMemoryFile:
data = json.loads(f.read_text(encoding="utf-8"))
assert "limits" not in {k for k in data if k != "document_metadata"}
- assert data["document_metadata"]["limits"]["max_lines"] == 600
+ assert data["document_metadata"]["limits"]["max_sessions"] == 20
def test_merges_root_limits_preserving_metadata_values(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
self._write_json(
f,
{
"document_metadata": {
- "limits": {"max_lines": 500},
- "status": {"current_lines": 10},
+ "limits": {"max_sessions": 20},
+ "status": {},
},
- "limits": {"max_lines": 600, "extra_field": 42},
+ "limits": {"max_sessions": 30, "max_key_learnings": 25},
"sessions": [],
},
)
@@ -447,19 +447,19 @@ class TestNormalizeMemoryFile:
assert result["success"] is True
data = json.loads(f.read_text(encoding="utf-8"))
- # metadata value (500) wins over root value (600)
- assert data["document_metadata"]["limits"]["max_lines"] == 500
- # extra_field from root gets merged in
- assert data["document_metadata"]["limits"]["extra_field"] == 42
+ # metadata value (20) wins over root value (30)
+ assert data["document_metadata"]["limits"]["max_sessions"] == 20
+ # valid key from root gets merged in
+ assert data["document_metadata"]["limits"]["max_key_learnings"] == 25
def test_removes_root_status(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
self._write_json(
f,
{
- "document_metadata": {"status": {"current_lines": 10}},
- "status": {"health": "ok", "current_lines": 5},
+ "document_metadata": {"status": {"last_health_check": "2026-01-01"}},
+ "status": {"health": "ok"},
"sessions": [],
},
)
@@ -472,12 +472,12 @@ class TestNormalizeMemoryFile:
def test_removes_auto_compress_at(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
self._write_json(
f,
{
"document_metadata": {
- "status": {"current_lines": 10, "auto_compress_at": 500},
+ "status": {"auto_compress_at": 500, "last_health_check": "2026-01-01"},
},
"sessions": [],
},
@@ -490,7 +490,7 @@ class TestNormalizeMemoryFile:
def test_dry_run_does_not_write(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
original = {
"document_metadata": {},
"limits": {"max_lines": 600},
@@ -508,13 +508,13 @@ class TestNormalizeMemoryFile:
def test_no_changes_when_already_normalized(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
self._write_json(
f,
{
"document_metadata": {
"limits": {"max_sessions": 20},
- "status": {"current_lines": 10, "last_health_check": "2026-03-31"},
+ "status": {"last_health_check": "2026-03-31"},
},
"sessions": [],
},
@@ -525,13 +525,13 @@ class TestNormalizeMemoryFile:
def test_removes_unused_limit_fields(self, monkeypatch, tmp_path):
norm, _ = _import_normalize(monkeypatch)
- f = tmp_path / "test.json"
+ f = tmp_path / "test.local.json"
self._write_json(
f,
{
"document_metadata": {
- "limits": {"max_lines": 600, "max_word_count": 9999, "max_token_count": 5000},
- "status": {"current_lines": 10, "last_health_check": "2026-03-31"},
+ "limits": {"max_sessions": 20, "max_lines": 600, "max_word_count": 9999, "max_token_count": 5000},
+ "status": {"last_health_check": "2026-03-31"},
},
"sessions": [],
},
@@ -542,7 +542,8 @@ class TestNormalizeMemoryFile:
data = json.loads(f.read_text(encoding="utf-8"))
assert "max_word_count" not in data["document_metadata"]["limits"]
assert "max_token_count" not in data["document_metadata"]["limits"]
- assert data["document_metadata"]["limits"]["max_lines"] == 600
+ assert "max_lines" not in data["document_metadata"]["limits"]
+ assert data["document_metadata"]["limits"]["max_sessions"] == 20
class TestTodosOperational:
diff --git a/src/aipass/memory/tests/test_watcher.py b/src/aipass/memory/tests/test_watcher.py
index 1fe7bd4b..35096477 100644
--- a/src/aipass/memory/tests/test_watcher.py
+++ b/src/aipass/memory/tests/test_watcher.py
@@ -71,6 +71,21 @@ def _prepare_watcher_mocks(monkeypatch):
monkeypatch.setitem(sys.modules, "watchdog.observers", mock_watchdog_observers)
monkeypatch.setitem(sys.modules, "watchdog.events", mock_watchdog_events)
+ # Mock normalize (lazy import inside on_modified and check_and_rollover)
+ mock_normalize_memory_file = MagicMock(return_value={"success": True, "changes": []})
+ mock_normalize = MagicMock()
+ mock_normalize.normalize_memory_file = mock_normalize_memory_file
+ monkeypatch.setitem(
+ sys.modules,
+ "aipass.memory.apps.handlers.schema",
+ MagicMock(),
+ )
+ monkeypatch.setitem(
+ sys.modules,
+ "aipass.memory.apps.handlers.schema.normalize",
+ mock_normalize,
+ )
+
# Mock rollover orchestrator (lazy import inside on_modified)
mock_execute_rollover = MagicMock(return_value={"success": True})
mock_orchestrator = MagicMock()
@@ -92,6 +107,7 @@ def _prepare_watcher_mocks(monkeypatch):
"observer_instance": mock_observer_instance,
"observer_cls": mock_observer_cls,
"execute_rollover": mock_execute_rollover,
+ "normalize_memory_file": mock_normalize_memory_file,
}
@@ -495,3 +511,61 @@ class TestMemoryFileWatcherOnModified:
watcher.on_modified(event)
mocks["execute_rollover"].assert_called_once()
+
+ def test_normalize_called_on_modification(self, monkeypatch):
+ """on_modified calls normalize_memory_file before update_line_count."""
+ mod, mocks = _import_watcher(monkeypatch)
+
+ watcher = mod.MemoryFileWatcher()
+ event = MagicMock()
+ event.is_directory = False
+ event.src_path = "/some/branch/.trinity/local.json"
+
+ watcher.on_modified(event)
+
+ mocks["normalize_memory_file"].assert_called_once()
+ mocks["update_line_count"].assert_called_once()
+
+ def test_normalize_changes_guard_write_loop(self, monkeypatch):
+ """When normalize makes changes, file_key is added to _recent_modifications to prevent write-loop."""
+ mod, mocks = _import_watcher(monkeypatch)
+
+ mocks["normalize_memory_file"].return_value = {
+ "success": True,
+ "changes": ["Stripped orphan 'stale_key' from root"],
+ }
+
+ watcher = mod.MemoryFileWatcher()
+ event = MagicMock()
+ event.is_directory = False
+ event.src_path = "/some/branch/.trinity/local.json"
+
+ watcher.on_modified(event)
+
+ from pathlib import Path as _P
+
+ file_key = str(_P("/some/branch/.trinity/local.json"))
+ assert file_key in watcher._recent_modifications
+
+ def test_normalize_no_changes_no_guard(self, monkeypatch):
+ """When normalize makes no changes, _recent_modifications is not populated by normalize step."""
+ mod, mocks = _import_watcher(monkeypatch)
+
+ mocks["normalize_memory_file"].return_value = {
+ "success": True,
+ "changes": [],
+ }
+
+ watcher = mod.MemoryFileWatcher()
+ event = MagicMock()
+ event.is_directory = False
+ event.src_path = "/some/branch/.trinity/local.json"
+
+ watcher.on_modified(event)
+
+ from pathlib import Path as _P
+
+ file_key = str(_P("/some/branch/.trinity/local.json"))
+ # file_key should NOT be in _recent_modifications from normalize (may be from rollover)
+ mocks["check_single_file"].return_value = {"success": True, "should_rollover": False}
+ assert file_key not in watcher._recent_modifications
diff --git a/src/aipass/prax/apps/handlers/monitoring/__init__.py b/src/aipass/prax/apps/handlers/monitoring/__init__.py
index f22c7aa2..6a27bc52 100644
--- a/src/aipass/prax/apps/handlers/monitoring/__init__.py
+++ b/src/aipass/prax/apps/handlers/monitoring/__init__.py
@@ -6,7 +6,7 @@ Handlers for file watching, log monitoring, branch detection, and filtering.
"""
# Export main handler interfaces
-from .unified_stream import print_event, print_command_separator
+from .unified_stream import print_event, print_command_separator, print_hook_event
from .branch_detector import detect_branch_from_path
from .interactive_filter import (
parse_command,
@@ -34,6 +34,7 @@ from .log_watcher import start_log_watcher, stop_log_watcher, is_log_watcher_act
__all__ = [
"print_event",
"print_command_separator",
+ "print_hook_event",
"detect_branch_from_path",
"parse_command",
"get_help_text",
diff --git a/src/aipass/prax/apps/handlers/monitoring/log_watcher.py b/src/aipass/prax/apps/handlers/monitoring/log_watcher.py
index 8d88b500..a67f4026 100644
--- a/src/aipass/prax/apps/handlers/monitoring/log_watcher.py
+++ b/src/aipass/prax/apps/handlers/monitoring/log_watcher.py
@@ -28,8 +28,8 @@ from typing import Optional, Dict, Any
import re
from aipass.prax.apps.modules.logger import get_direct_logger
-from watchdog.observers import Observer as WatchdogObserver
-from watchdog.events import FileSystemEventHandler
+from watchdog.observers import Observer as WatchdogObserver # type: ignore
+from watchdog.events import FileSystemEventHandler # type: ignore
# Import from prax config
from aipass.prax.apps.handlers.config.load import get_system_logs_dir
@@ -102,9 +102,13 @@ class LogFileWatcher(FileSystemEventHandler):
self.last_command_per_branch: Dict[str, str] = {}
def _process_log_line(self, branch: str, line: str, file_path: str) -> None:
- """Process a single log line: detect commands or emit as log event."""
+ """Process a single log line: detect hooks, commands, or emit as log event."""
if not line.strip():
return
+ hook_info = self._extract_hook_info(line)
+ if hook_info:
+ self._emit_hook_event(branch, hook_info)
+ return
command_info = self._extract_command_info(line)
if command_info:
self._emit_command_separator(branch, command_info)
@@ -159,6 +163,63 @@ class LogFileWatcher(FileSystemEventHandler):
except Exception as e:
logger.info(f"Error reading log file {file_path}: {e}")
+ _HOOK_PATTERN = re.compile(r"\[HOOKS\]\s+(\w+)\s+(\w+)")
+
+ def _extract_hook_info(self, log_line: str) -> Optional[Dict[str, str]]:
+ """Extract hook event info from structured [HOOKS] log lines.
+
+ The action is the bare second word (matches what hooks emits), e.g.:
+ [HOOKS] cadence fired loader=global turn=35 period=5 offset=0 session=...
+ [HOOKS] cadence skipped loader=branch turn=37 period=5 offset=0 session=...
+ group(1)=name (cadence), group(2)=action (fired/skipped). Remaining
+ key=value fields are parsed by the finditer loop below.
+ """
+ match = self._HOOK_PATTERN.search(log_line)
+ if not match:
+ return None
+ name = match.group(1)
+ action = match.group(2)
+ details: Dict[str, str] = {"name": name, "action": action}
+ for kv_match in re.finditer(r"(\w+)=(\S+)", log_line):
+ details[kv_match.group(1)] = kv_match.group(2)
+ return details
+
+ def _emit_hook_event(self, branch: str, hook_info: Dict[str, str]) -> None:
+ """Emit a hook event to the monitoring queue."""
+ action = hook_info.get("action", "unknown")
+ name = hook_info.get("name", "hook")
+ loader = hook_info.get("loader", "")
+ turn = hook_info.get("turn", "")
+ period = hook_info.get("period", "")
+ offset = hook_info.get("offset", "")
+ session = hook_info.get("session", "")
+
+ parts = [f"{name}:{action}"]
+ if loader:
+ parts.append(f"loader={loader}")
+ if turn:
+ parts.append(f"t={turn}")
+ if period:
+ parts.append(f"p={period}")
+ if offset and offset != "0":
+ parts.append(f"off={offset}")
+ if session:
+ parts.append(f"s={session[:8]}")
+ message = " ".join(parts)
+
+ level = "success" if action == "fired" else "info"
+
+ hook_event = MonitoringEvent(
+ priority=2,
+ event_type="hook",
+ branch=branch,
+ action=action,
+ message=message,
+ level=level,
+ timestamp=datetime.now(),
+ )
+ self.event_queue.enqueue(hook_event)
+
def _should_display_log(self, _log_line: str) -> bool:
"""Check if log line should be displayed. No filtering — show everything."""
return True
@@ -483,7 +544,7 @@ def start_log_watcher(event_queue: MonitoringQueue, use_polling: bool = False) -
# Create observer — polling fallback when inotify unavailable
if use_polling:
- from watchdog.observers.polling import PollingObserver
+ from watchdog.observers.polling import PollingObserver # type: ignore
observer = PollingObserver(timeout=1)
logger.info("Log watcher using polling observer (1s interval)")
diff --git a/src/aipass/prax/apps/handlers/monitoring/unified_stream.py b/src/aipass/prax/apps/handlers/monitoring/unified_stream.py
index e79028f7..411919d5 100644
--- a/src/aipass/prax/apps/handlers/monitoring/unified_stream.py
+++ b/src/aipass/prax/apps/handlers/monitoring/unified_stream.py
@@ -185,6 +185,28 @@ def print_command_separator(branch: str, command: str, caller: Optional[str] = N
console.print(f"[bold {branch_color}]{'─' * 60}[/bold {branch_color}]")
+def print_hook_event(branch: str, message: str, action: str = "unknown"):
+ """Print a hook event with distinct fired/skipped styling.
+
+ Args:
+ branch: Branch the hook event originated from
+ message: Hook event summary (e.g. "cadence:fired loader=global turn=35")
+ action: "fired" or "skipped" (controls color)
+ """
+ with _print_lock:
+ timestamp = datetime.now().strftime("%H:%M:%S")
+ if action == "fired":
+ style = "bold green"
+ symbol = "⚡"
+ elif action == "skipped":
+ style = "dim"
+ symbol = "·"
+ else:
+ style = "white"
+ symbol = "?"
+ console.print(f"[dim]{timestamp}[/dim] [{style}]{symbol} HOOK {message}[/{style}]")
+
+
def print_status(watched_branches: List[str], verbosity: int, filters: Optional[Dict] = None):
"""
Display current monitoring status
diff --git a/src/aipass/prax/apps/modules/monitor.py b/src/aipass/prax/apps/modules/monitor.py
index 2a5eafd0..4b04251d 100755
--- a/src/aipass/prax/apps/modules/monitor.py
+++ b/src/aipass/prax/apps/modules/monitor.py
@@ -38,6 +38,7 @@ from aipass.prax.apps.handlers.json import json_handler
from aipass.prax.apps.handlers.monitoring import (
print_event, # unified_stream.py
print_command_separator, # unified_stream.py - command headers
+ print_hook_event, # unified_stream.py - hook fire/skip display
MonitoringQueue, # event_queue.py
ModuleTracker, # module_tracker.py
)
@@ -338,6 +339,8 @@ def _render_event(event) -> None:
if len(parts) == 2 and parts[1]:
target = parts[1]
print_command_separator(event.branch, event.message, caller, target)
+ elif event.event_type == "hook":
+ print_hook_event(event.branch, event.message, event.action)
else:
print_event(event.event_type, event.branch, event.message, event.level, pid=branch_pid)
diff --git a/src/aipass/prax/tests/test_log_watcher.py b/src/aipass/prax/tests/test_log_watcher.py
index efab823c..0a58d3c8 100644
--- a/src/aipass/prax/tests/test_log_watcher.py
+++ b/src/aipass/prax/tests/test_log_watcher.py
@@ -1063,3 +1063,143 @@ class TestInitializePositionsAdditional:
watcher.initialize_positions()
assert str(log_file) not in watcher.log_positions
+
+
+class TestExtractHookInfo:
+ """Test _extract_hook_info for structured [HOOKS] log lines."""
+
+ def test_fired_line_extracted(self):
+ """Should extract name, action, and key-value details from a fired line."""
+ mod = _import_log_watcher()
+ watcher, _ = _make_watcher(mod)
+
+ # Real format hooks emits: action is the bare second word, no action= field.
+ line = "[HOOKS] cadence fired loader=global turn=35 period=5 offset=0 session=abc12345"
+ result = watcher._extract_hook_info(line)
+ assert result is not None
+ assert result["name"] == "cadence"
+ assert result["action"] == "fired"
+ assert result["loader"] == "global"
+ assert result["turn"] == "35"
+
+ def test_skipped_line_extracted(self):
+ """Should extract skipped hook events."""
+ mod = _import_log_watcher()
+ watcher, _ = _make_watcher(mod)
+
+ line = "[HOOKS] cadence skipped loader=branch turn=37 period=5 offset=0 session=abc12345"
+ result = watcher._extract_hook_info(line)
+ assert result is not None
+ assert result["name"] == "cadence"
+ assert result["action"] == "skipped"
+ assert result["loader"] == "branch"
+
+ def test_non_hook_line_returns_none(self):
+ """Non-hook log lines should return None."""
+ mod = _import_log_watcher()
+ watcher, _ = _make_watcher(mod)
+
+ result = watcher._extract_hook_info("[FLOW] Creating plan FPLAN-0099")
+ assert result is None
+
+ def test_hook_info_line_returns_none(self):
+ """A [HOOKS] info/error line (colon after the name) is not a fire/skip event → None."""
+ mod = _import_log_watcher()
+ watcher, _ = _make_watcher(mod)
+
+ # These are real cadence info lines; the colon stops the action capture.
+ assert watcher._extract_hook_info("[HOOKS] cadence: config load failed, using defaults") is None
+ assert watcher._extract_hook_info("[HOOKS] cadence: counter reset for post-compact re-injection") is None
+
+
+class TestEmitHookEvent:
+ """Test _emit_hook_event queues properly."""
+
+ def test_fired_event_queued_with_correct_kwargs(self):
+ """Fired hook events should pass event_type=hook, level=success to MonitoringEvent."""
+ mod = _import_log_watcher()
+ watcher, mock_queue = _make_watcher(mod)
+
+ mock_event_cls = MagicMock()
+ with patch.object(mod, "MonitoringEvent", mock_event_cls):
+ hook_info = {
+ "name": "cadence",
+ "action": "fired",
+ "loader": "global",
+ "turn": "35",
+ "period": "5",
+ "offset": "0",
+ "session": "abc12345",
+ }
+ watcher._emit_hook_event("HOOKS", hook_info)
+
+ mock_event_cls.assert_called_once()
+ kwargs = mock_event_cls.call_args[1]
+ assert kwargs["event_type"] == "hook"
+ assert kwargs["action"] == "fired"
+ assert kwargs["level"] == "success"
+ assert "cadence:fired" in kwargs["message"]
+ assert "loader=global" in kwargs["message"]
+ assert "t=35" in kwargs["message"]
+ assert "p=5" in kwargs["message"]
+ assert "s=abc12345" in kwargs["message"]
+
+ def test_skipped_event_queued_with_info_level(self):
+ """Skipped hook events should pass level=info to MonitoringEvent."""
+ mod = _import_log_watcher()
+ watcher, mock_queue = _make_watcher(mod)
+
+ mock_event_cls = MagicMock()
+ with patch.object(mod, "MonitoringEvent", mock_event_cls):
+ hook_info = {"name": "cadence", "action": "skipped", "loader": "branch", "turn": "37"}
+ watcher._emit_hook_event("HOOKS", hook_info)
+
+ kwargs = mock_event_cls.call_args[1]
+ assert kwargs["action"] == "skipped"
+ assert kwargs["level"] == "info"
+
+ def test_process_log_line_routes_hook_to_emit(self):
+ """Hook lines in _process_log_line should route to _emit_hook_event, not _emit_log_event."""
+ mod = _import_log_watcher()
+ watcher, mock_queue = _make_watcher(mod)
+
+ with (
+ patch.object(watcher, "_emit_hook_event") as mock_hook,
+ patch.object(watcher, "_emit_command_separator") as mock_cmd,
+ patch.object(watcher, "_emit_log_event") as mock_log,
+ ):
+ watcher._process_log_line(
+ "HOOKS",
+ "[HOOKS] cadence fired loader=global turn=35 period=5 offset=0 session=abc",
+ "/fake/file.log",
+ )
+
+ mock_hook.assert_called_once()
+ mock_cmd.assert_not_called()
+ mock_log.assert_not_called()
+
+ def test_process_real_pipe_delimited_hook_line(self):
+ """Real log lines are pipe-delimited — hook detection must match through the prefix."""
+ mod = _import_log_watcher()
+ watcher, mock_queue = _make_watcher(mod)
+
+ real_line = (
+ "2026-06-09 19:56:04 | captured_cadence | INFO | "
+ "[HOOKS] cadence skipped loader=branch turn=18 period=5 offset=0 session=c98a722b"
+ )
+
+ with (
+ patch.object(watcher, "_emit_hook_event") as mock_hook,
+ patch.object(watcher, "_emit_command_separator") as mock_cmd,
+ patch.object(watcher, "_emit_log_event") as mock_log,
+ ):
+ watcher._process_log_line("HOOKS", real_line, "/fake/hooks_cadence.log")
+
+ mock_hook.assert_called_once()
+ hook_info = mock_hook.call_args[0][1]
+ assert hook_info["name"] == "cadence"
+ assert hook_info["action"] == "skipped"
+ assert hook_info["loader"] == "branch"
+ assert hook_info["turn"] == "18"
+ mock_cmd.assert_not_called()
+ mock_log.assert_not_called()
diff --git a/src/aipass/seedgo/.seedgo/bypass.json b/src/aipass/seedgo/.seedgo/bypass.json
index f44de80d..0acabab0 100644
--- a/src/aipass/seedgo/.seedgo/bypass.json
+++ b/src/aipass/seedgo/.seedgo/bypass.json
@@ -261,6 +261,16 @@
{
"file": "tests/test_coverage_arch_checklist.py",
"reason": "Test file: outside 3-layer structure by convention, imports handlers directly for unit testing, and test functions omit docstrings by pytest convention."
+ },
+ {
+ "file": "apps/handlers/aipass_standards/skip_dirs.py",
+ "standard": "json_structure",
+ "reason": "Pure constants module — defines SOURCE_SKIP_DIRS frozenset only, no operations to log."
+ },
+ {
+ "file": "apps/handlers/aipass_proof/",
+ "standard": "handlers",
+ "reason": "Proof handlers import SOURCE_SKIP_DIRS from aipass_standards/skip_dirs.py — shared constant, same-branch cross-handler import, intentional (FPLAN-0261)."
}
],
"notes": {
diff --git a/src/aipass/seedgo/apps/handlers/aipass_proof/content_naming.py b/src/aipass/seedgo/apps/handlers/aipass_proof/content_naming.py
index b81ae026..f2b7e1f0 100644
--- a/src/aipass/seedgo/apps/handlers/aipass_proof/content_naming.py
+++ b/src/aipass/seedgo/apps/handlers/aipass_proof/content_naming.py
@@ -26,9 +26,10 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
+from aipass.seedgo.apps.handlers.aipass_standards.skip_dirs import SOURCE_SKIP_DIRS
# Directories to skip during scanning
-_SKIP_DIRS = {".archive", ".sorting_unprocessed", "__pycache__"}
+_SKIP_DIRS = SOURCE_SKIP_DIRS
def _parse_public_functions(file_path: Path) -> list[str]:
diff --git a/src/aipass/seedgo/apps/handlers/aipass_proof/interface.py b/src/aipass/seedgo/apps/handlers/aipass_proof/interface.py
index e06b075d..096028ff 100644
--- a/src/aipass/seedgo/apps/handlers/aipass_proof/interface.py
+++ b/src/aipass/seedgo/apps/handlers/aipass_proof/interface.py
@@ -28,11 +28,12 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
+from aipass.seedgo.apps.handlers.aipass_standards.skip_dirs import SOURCE_SKIP_DIRS
VALID_SCOPES = {"all_files", "entry_point", "branch_level"}
# Directories / files to skip inside the pack
-_SKIP_DIRS = frozenset({".archive", ".sorting_unprocessed"})
+_SKIP_DIRS = SOURCE_SKIP_DIRS
# -- AST helpers ---------------------------------------------------------------
diff --git a/src/aipass/seedgo/apps/handlers/aipass_proof/triplet.py b/src/aipass/seedgo/apps/handlers/aipass_proof/triplet.py
index f8f870e2..671eb64a 100644
--- a/src/aipass/seedgo/apps/handlers/aipass_proof/triplet.py
+++ b/src/aipass/seedgo/apps/handlers/aipass_proof/triplet.py
@@ -23,9 +23,10 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
+from aipass.seedgo.apps.handlers.aipass_standards.skip_dirs import SOURCE_SKIP_DIRS
# Directories to skip entirely
-_SKIP_DIRS = {".archive", ".sorting_unprocessed", "__pycache__"}
+_SKIP_DIRS = SOURCE_SKIP_DIRS
def _top_level_files(pack_dir: Path) -> list[Path]:
diff --git a/src/aipass/seedgo/apps/handlers/aipass_standards/dead_code_check.py b/src/aipass/seedgo/apps/handlers/aipass_standards/dead_code_check.py
index 0e521676..f8b95016 100644
--- a/src/aipass/seedgo/apps/handlers/aipass_standards/dead_code_check.py
+++ b/src/aipass/seedgo/apps/handlers/aipass_standards/dead_code_check.py
@@ -24,34 +24,12 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
+from aipass.seedgo.apps.handlers.aipass_standards.skip_dirs import SOURCE_SKIP_DIRS
AUDIT_SCOPE = "branch_level"
# Directories to skip when collecting source files
-_SKIP_DIRS = {
- "__pycache__",
- ".archive",
- ".mypy_cache",
- ".ruff_cache",
- ".pytest_cache",
- "json_templates",
- "logs",
- "tools",
- ".venv",
- "venv",
- "node_modules",
- ".git",
- "site-packages",
- ".trinity",
- ".aipass",
- ".ai_mail.local",
- ".spawn",
- "backups",
- "reports",
- "docs",
- "tests",
- ".sorting_unprocessed",
-}
+_SKIP_DIRS = SOURCE_SKIP_DIRS | {"tests", "json_templates"}
# =============================================
diff --git a/src/aipass/seedgo/apps/handlers/aipass_standards/skip_dirs.py b/src/aipass/seedgo/apps/handlers/aipass_standards/skip_dirs.py
new file mode 100644
index 00000000..3d5b89f7
--- /dev/null
+++ b/src/aipass/seedgo/apps/handlers/aipass_standards/skip_dirs.py
@@ -0,0 +1,50 @@
+# =================== AIPass ====================
+# Name: skip_dirs.py
+# Description: Shared source-skip directories for branch-tree-scanning checkers
+# Version: 1.0.0
+# Created: 2026-06-10
+# Modified: 2026-06-10
+# =============================================
+
+"""Shared skip-directory set for checkers that rglob branch trees.
+
+Single source of truth so per-checker lists cannot drift.
+Output dirs (artifacts, dropbox, docs.local, system_logs) are runtime
+products, not committed source — scanning them causes local-vs-CI
+audit divergence (FPLAN-0261).
+"""
+
+SOURCE_SKIP_DIRS: frozenset[str] = frozenset(
+ {
+ # Build / cache
+ "__pycache__",
+ ".mypy_cache",
+ ".ruff_cache",
+ ".pytest_cache",
+ # Environment
+ ".venv",
+ "venv",
+ "node_modules",
+ ".git",
+ "site-packages",
+ # Archive / staging
+ ".archive",
+ ".sorting_unprocessed",
+ # Project infrastructure (not auditable source)
+ ".trinity",
+ ".aipass",
+ ".ai_mail.local",
+ ".spawn",
+ # Output / runtime (gitignored — causes local-vs-CI divergence)
+ "artifacts",
+ "dropbox",
+ "docs.local",
+ "system_logs",
+ # Data / non-source
+ "logs",
+ "tools",
+ "backups",
+ "reports",
+ "docs",
+ }
+)
diff --git a/src/aipass/seedgo/apps/handlers/aipass_standards/test_quality_check.py b/src/aipass/seedgo/apps/handlers/aipass_standards/test_quality_check.py
index 2771ba42..1c3026cc 100644
--- a/src/aipass/seedgo/apps/handlers/aipass_standards/test_quality_check.py
+++ b/src/aipass/seedgo/apps/handlers/aipass_standards/test_quality_check.py
@@ -29,32 +29,12 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
+from aipass.seedgo.apps.handlers.aipass_standards.skip_dirs import SOURCE_SKIP_DIRS
AUDIT_SCOPE = "branch_level"
# -- Directories to skip when scanning for module coverage --------------------
-SKIP_DIRS: set[str] = {
- "__pycache__",
- ".archive",
- ".mypy_cache",
- ".ruff_cache",
- ".pytest_cache",
- ".venv",
- "venv",
- "node_modules",
- ".git",
- "site-packages",
- "logs",
- "tools",
- ".trinity",
- ".aipass",
- ".ai_mail.local",
- ".spawn",
- "backups",
- "reports",
- "docs",
- ".sorting_unprocessed",
-}
+SKIP_DIRS: set[str] = set(SOURCE_SKIP_DIRS)
# -- Regex patterns for module coverage (from test_coverage_check.py) ---------
RE_TEST_FUNC = re.compile(r"^\s*(?:async\s+)?def\s+(test_\w+)", re.MULTILINE)
diff --git a/src/aipass/seedgo/apps/handlers/aipass_standards/unused_function_check.py b/src/aipass/seedgo/apps/handlers/aipass_standards/unused_function_check.py
index 6e13b8a1..7752a24a 100644
--- a/src/aipass/seedgo/apps/handlers/aipass_standards/unused_function_check.py
+++ b/src/aipass/seedgo/apps/handlers/aipass_standards/unused_function_check.py
@@ -32,34 +32,12 @@ from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
+from aipass.seedgo.apps.handlers.aipass_standards.skip_dirs import SOURCE_SKIP_DIRS
AUDIT_SCOPE = "branch_level"
# -- Directories to skip when collecting source files -------------------------
-SKIP_DIRS = {
- "__pycache__",
- ".archive",
- "logs",
- "tests",
- "json_templates",
- "tools",
- ".trinity",
- ".aipass",
- ".ai_mail.local",
- ".venv",
- "venv",
- "node_modules",
- ".git",
- "site-packages",
- ".mypy_cache",
- ".ruff_cache",
- ".pytest_cache",
- ".spawn",
- "backups",
- "reports",
- "docs",
- ".sorting_unprocessed",
-}
+SKIP_DIRS = SOURCE_SKIP_DIRS | {"tests", "json_templates"}
# -- Function names excluded from analysis ------------------------------------
EXCLUDED_NAMES = {
diff --git a/src/aipass/seedgo/apps/handlers/diagnostics/diagnostics_check.py b/src/aipass/seedgo/apps/handlers/diagnostics/diagnostics_check.py
index 4b98c7e0..caaf75ba 100644
--- a/src/aipass/seedgo/apps/handlers/diagnostics/diagnostics_check.py
+++ b/src/aipass/seedgo/apps/handlers/diagnostics/diagnostics_check.py
@@ -25,13 +25,12 @@ from typing import Dict, List, Optional
from rich.console import Console
-console = Console()
-
-# Import ignore patterns from bypass handler
from aipass.prax import logger
from aipass.seedgo.apps.handlers.bypass.ignore_handler import get_audit_ignore_patterns
from aipass.seedgo.apps.handlers.json import json_handler
+console = Console()
+
AUDIT_SCOPE = "branch_level"
# Diagnostics handler directory (where this file lives)
@@ -86,19 +85,29 @@ def check_file(file_path: str) -> Dict:
try:
# Run pyright with JSON output
result = subprocess.run(
- ["python3", "-m", "pyright", "--outputjson", str(path)], capture_output=True, text=True, timeout=30
+ [sys.executable, "-m", "pyright", "--pythonpath", sys.executable, "--outputjson", str(path)],
+ capture_output=True,
+ text=True,
+ timeout=30,
)
# Parse JSON output
try:
output = json.loads(result.stdout)
except json.JSONDecodeError:
- logger.info("Failed to parse pyright JSON output for %s", file_path)
+ logger.error("Failed to parse pyright JSON output for %s", file_path)
return {
"file": str(file_path),
- "errors": 0,
+ "errors": 1,
"warnings": 0,
- "diagnostics": [],
+ "diagnostics": [
+ {
+ "line": 0,
+ "severity": "error",
+ "message": "Pyright failed to run — cannot parse output",
+ "rule": "",
+ }
+ ],
"error": f"Failed to parse pyright output: {result.stderr or result.stdout}",
}
@@ -125,11 +134,23 @@ def check_file(file_path: str) -> Dict:
return {"file": str(file_path), "errors": errors, "warnings": warnings, "diagnostics": diagnostics}
except subprocess.TimeoutExpired:
- logger.info("Pyright timed out for %s", file_path)
- return {"file": str(file_path), "errors": 0, "warnings": 0, "diagnostics": [], "error": "Pyright timed out"}
+ logger.error("Pyright timed out for %s", file_path)
+ return {
+ "file": str(file_path),
+ "errors": 1,
+ "warnings": 0,
+ "diagnostics": [{"line": 0, "severity": "error", "message": "Pyright timed out", "rule": ""}],
+ "error": "Pyright timed out",
+ }
except Exception as e:
- logger.info("Pyright check failed for %s: %s", file_path, e)
- return {"file": str(file_path), "errors": 0, "warnings": 0, "diagnostics": [], "error": str(e)}
+ logger.error("Pyright check failed for %s: %s", file_path, e)
+ return {
+ "file": str(file_path),
+ "errors": 1,
+ "warnings": 0,
+ "diagnostics": [{"line": 0, "severity": "error", "message": f"Pyright failed: {e}", "rule": ""}],
+ "error": str(e),
+ }
def check_directory(directory: str, pattern: str = "**/*.py") -> Dict:
@@ -164,7 +185,7 @@ def check_directory(directory: str, pattern: str = "**/*.py") -> Dict:
# Run pyright on entire directory for efficiency
try:
result = subprocess.run(
- ["python3", "-m", "pyright", "--outputjson", str(path)],
+ [sys.executable, "-m", "pyright", "--pythonpath", sys.executable, "--outputjson", str(path)],
capture_output=True,
text=True,
timeout=300, # 5 minutes for full directory
@@ -173,13 +194,27 @@ def check_directory(directory: str, pattern: str = "**/*.py") -> Dict:
try:
output = json.loads(result.stdout)
except json.JSONDecodeError:
- logger.info("Failed to parse pyright JSON output for directory %s", directory)
+ logger.error("Failed to parse pyright JSON output for directory %s", directory)
return {
"total_files": 0,
- "files_with_errors": 0,
- "total_errors": 0,
+ "files_with_errors": 1,
+ "total_errors": 1,
"total_warnings": 0,
- "results": [],
+ "results": [
+ {
+ "file": str(directory),
+ "errors": 1,
+ "warnings": 0,
+ "diagnostics": [
+ {
+ "line": 0,
+ "severity": "error",
+ "message": "Pyright failed to run — cannot parse output",
+ "rule": "",
+ }
+ ],
+ }
+ ],
"error": "Failed to parse pyright output",
}
@@ -227,23 +262,37 @@ def check_directory(directory: str, pattern: str = "**/*.py") -> Dict:
}
except subprocess.TimeoutExpired:
- logger.info("Pyright timed out for directory %s", directory)
+ logger.error("Pyright timed out for directory %s", directory)
return {
"total_files": 0,
- "files_with_errors": 0,
- "total_errors": 0,
+ "files_with_errors": 1,
+ "total_errors": 1,
"total_warnings": 0,
- "results": [],
+ "results": [
+ {
+ "file": str(directory),
+ "errors": 1,
+ "warnings": 0,
+ "diagnostics": [{"line": 0, "severity": "error", "message": "Pyright timed out", "rule": ""}],
+ }
+ ],
"error": "Pyright timed out (directory too large?)",
}
except Exception as e:
- logger.info("Pyright directory check failed for %s: %s", directory, e)
+ logger.error("Pyright directory check failed for %s: %s", directory, e)
return {
"total_files": 0,
- "files_with_errors": 0,
- "total_errors": 0,
+ "files_with_errors": 1,
+ "total_errors": 1,
"total_warnings": 0,
- "results": [],
+ "results": [
+ {
+ "file": str(directory),
+ "errors": 1,
+ "warnings": 0,
+ "diagnostics": [{"line": 0, "severity": "error", "message": f"Pyright failed: {e}", "rule": ""}],
+ }
+ ],
"error": str(e),
}
diff --git a/src/aipass/spawn/.seedgo/bypass.json b/src/aipass/spawn/.seedgo/bypass.json
index c4bbc0a8..8858505f 100644
--- a/src/aipass/spawn/.seedgo/bypass.json
+++ b/src/aipass/spawn/.seedgo/bypass.json
@@ -103,7 +103,7 @@
{
"file": "apps/handlers/json/json_handler.py",
"standard": "naming",
- "reason": "Module-level names are function re-exports from aipass.common.json_handler.JsonHandler, not constants. Lowercase is correct for callable bindings."
+ "reason": "Module-level names are function re-exports from aipass.aipass.shared.json_handler.JsonHandler, not constants. Lowercase is correct for callable bindings."
},
{
"file": "apps/handlers/delete_ops.py",
diff --git a/src/aipass/spawn/apps/handlers/json/json_handler.py b/src/aipass/spawn/apps/handlers/json/json_handler.py
index 28361e76..01580975 100644
--- a/src/aipass/spawn/apps/handlers/json/json_handler.py
+++ b/src/aipass/spawn/apps/handlers/json/json_handler.py
@@ -1,12 +1,12 @@
# =================== AIPass ====================
# Name: json_handler.py
-# Description: Spawn JSON handler — configured instance of aipass.common
+# Description: Spawn JSON handler — configured instance of aipass.aipass.shared
# Version: 3.0.0
# Created: 2026-03-07
-# Modified: 2026-06-06
+# Modified: 2026-06-10
# =============================================
-"""Spawn JSON handler — thin shim over aipass.common.json_handler.
+"""Spawn JSON handler — thin shim over aipass.aipass.shared.json_handler.
Creates a JsonHandler instance configured with spawn's json_dir.
All functions are re-exported for backward-compatible imports.
@@ -14,7 +14,7 @@ All functions are re-exported for backward-compatible imports.
from pathlib import Path
-from aipass.common.json_handler import JsonHandler
+from aipass.aipass.shared.json_handler import JsonHandler
_SPAWN_ROOT = Path(__file__).resolve().parents[3]
_JSON_DIR = _SPAWN_ROOT / "spawn_json"
diff --git a/src/aipass/spawn/apps/handlers/json_ops.py b/src/aipass/spawn/apps/handlers/json_ops.py
index 7fb0b9a3..6637004a 100644
--- a/src/aipass/spawn/apps/handlers/json_ops.py
+++ b/src/aipass/spawn/apps/handlers/json_ops.py
@@ -1,13 +1,13 @@
# =================== AIPass ====================
# Name: json_ops.py
-# Description: JSON operations — re-exports from aipass.common
+# Description: JSON operations — re-exports from aipass.aipass.shared
# Version: 2.0.0
# Created: 2026-03-07
-# Modified: 2026-06-06
+# Modified: 2026-06-10
# =============================================
-"""JSON operations — thin re-export from aipass.common.json_ops."""
+"""JSON operations — thin re-export from aipass.aipass.shared.json_ops."""
-from aipass.common.json_ops import backup_json, deep_merge
+from aipass.aipass.shared.json_ops import backup_json, deep_merge
__all__ = ["deep_merge", "backup_json"]
diff --git a/src/aipass/spawn/apps/handlers/registry.py b/src/aipass/spawn/apps/handlers/registry.py
index 41a78fe5..21bfc368 100644
--- a/src/aipass/spawn/apps/handlers/registry.py
+++ b/src/aipass/spawn/apps/handlers/registry.py
@@ -3,7 +3,7 @@
# Description: *_REGISTRY.json discovery and CRUD operations
# Version: 1.0.0
# Created: 2026-03-05
-# Modified: 2026-03-13
+# Modified: 2026-06-10
# =============================================
"""*_REGISTRY.json discovery and CRUD operations."""
@@ -34,7 +34,7 @@ def branches_as_list(branches):
def find_registry(start_path=None):
- """Find *_REGISTRY.json — delegates to aipass.common.registry_discovery.
+ """Find *_REGISTRY.json — delegates to aipass.aipass.shared.registry_discovery.
Passes spawn's package root as the fallback for editable installs.
@@ -44,7 +44,7 @@ def find_registry(start_path=None):
Returns:
Path to *_REGISTRY.json.
"""
- from aipass.common.registry_discovery import find_registry as _common_find
+ from aipass.aipass.shared.registry_discovery import find_registry as _common_find
return _common_find(start_path=start_path, package_root=Path(__file__).resolve().parent)
diff --git a/src/aipass/spawn/templates/builder/.spawn/.template_registry.json b/src/aipass/spawn/templates/builder/.spawn/.template_registry.json
index 1c54b2d8..b051554a 100644
--- a/src/aipass/spawn/templates/builder/.spawn/.template_registry.json
+++ b/src/aipass/spawn/templates/builder/.spawn/.template_registry.json
@@ -1,7 +1,7 @@
{
"metadata": {
"version": "1.0.0",
- "last_updated": "2026-06-07",
+ "last_updated": "2026-06-10",
"description": "Template file tracking registry for ID-based updates"
},
"files": {
@@ -149,7 +149,7 @@
"content_hash": "a4cf0a8e3b4f",
"has_branch_placeholder": false
},
- "f018": {
+ "f015": {
"path": "apps/modules/__init__.py",
"name": "__init__.py",
"content_hash": "e3b0c44298fc",
@@ -263,7 +263,7 @@
"content_hash": "28e9ae373563",
"has_branch_placeholder": false
},
- "f015": {
+ "f018": {
"path": "apps/plugins/__init__.py",
"name": "__init__.py",
"content_hash": "e3b0c44298fc",
diff --git a/src/aipass/spawn/tests/test_json_handler.py b/src/aipass/spawn/tests/test_json_handler.py
index 1dd50db1..5bc3ac6b 100644
--- a/src/aipass/spawn/tests/test_json_handler.py
+++ b/src/aipass/spawn/tests/test_json_handler.py
@@ -186,7 +186,7 @@ def isolate_json_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
monkeypatch.setattr(_mod, _JSON_DIR_ATTR, str(tmp_path))
else:
monkeypatch.setattr(_mod, _JSON_DIR_ATTR, tmp_path)
- # Patch the JsonHandler instance if one exists (aipass.common migration)
+ # Patch the JsonHandler instance if one exists (aipass.aipass.shared migration)
if hasattr(_mod, "_handler") and hasattr(_mod._handler, "_json_dir"):
monkeypatch.setattr(_mod._handler, "_json_dir", tmp_path)
return tmp_path