diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d732f1f..073fdbdd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,30 @@ PyPI version — not the changelog header. ## [2026-07-20] +**fix(hooks)** — persistent_alert dedup + loud trust-break banner (5-agent +trace round follow-ups, DPLAN-0253 tail): + +- persistent_alert's once-per-session sound dedup lived in a module-global set, + but every bridge call is a fresh process — TTS would have announced on every + prompt while any alert was active. Replaced with session+alert-keyed tempdir + guard files (context_gauge idiom); banner capped at 10 alerts with an + "...and N more" note. +- Trust-registry breaks are now LOUD: any `.aipass/hooks.json` change breaks + the enrolled hash and silently disabled the entire hook layer (bit us live + for 2+ hours — tier prompts, security gates, everything dark, one log-file + WARNING as the only signal). New `is_hash_mismatch()` distinguishes a + genuine break from never-enrolled; `trust_break_banner()` does a + config-independent walk+hash check; the engine emits a full-width banner + once per prompt via the presence_gate bridge call. No auto-heal — + re-enrollment stays a deliberate human checkpoint. Live-fired: hash broken → + banner; restored → healthy. 16 new tests, suite 1206 green, seedgo 100%. +- Go-live day for the whole handler roster: 11/12 manifest entries wired into + provider settings by devpulse with Patrick accepting (user_message_relay + held: synchronous Telegram call + full prompt text off-machine — needs a + background send and an explicit call first). @hooks branch prompt corrected + and hardened: two-wires checklist + mandatory provider-wire flag in every + build reply. + **feat(hooks)** — auto-compact prep: context gauge + mechanical snapshot (DPLAN-0253, built by @hooks, two rounds): diff --git a/src/aipass/hooks/.aipass/aipass_local_prompt.md b/src/aipass/hooks/.aipass/aipass_local_prompt.md index ba1d8ec4..c28cd235 100644 --- a/src/aipass/hooks/.aipass/aipass_local_prompt.md +++ b/src/aipass/hooks/.aipass/aipass_local_prompt.md @@ -9,7 +9,7 @@ HOOKS -- hook infrastructure owner. Single engine dispatches all hooks across pl ## What I Do - Own the hook engine -- receives events from platform bridges, routes to handlers, logs everything -- Maintain 14 native handlers across 4 categories (prompt, security, lifecycle, notification) +- Maintain 26 native handlers across 4 categories (prompt, security, lifecycle, notification) - Bridge platforms -- thin normalization layer per provider (Claude today, Codex planned) - Per-project config -- `.aipass/hooks.json` controls what fires per project - Log everything -- prax integration + JSONL diagnostics for every hook execution @@ -40,41 +40,54 @@ apps/ handlers/ bridges/ claude.py # Claude Code bridge (provider settings entry point) - prompt/ # Prompt injection hooks + codex.py # Codex bridge (planned) + prompt/ # Prompt injection hooks (UserPromptSubmit) branch_loader.py # Injects aipass_local_prompt.md tier0_kernel.py # Injects tier0 kernel prompt (every turn) navmap.py # Injects tier1 navmap prompt (periodic) identity.py # Injects passport identity block + compass_recall.py # Governance recall injection + feedback_pulse.py # 10-turn cadence feedback nudge (disabled default) + context_gauge.py # Live transcript-fill nudge toward /prep + persistent_alert.py # Advisory banners for .aipass/alerts.json security/ # Enforcement hooks + presence_gate.py # Session presence gate (UserPromptSubmit + Stop release) edit_gate.py # Blocks edits while type errors exist git_gate.py # Enforces git access tiers + rm_gate.py # Guards destructive rm commands + registry_gate.py # Guards registry-modifying commands subagent_gate.py # Blocks sub-agent stop until clean lifecycle/ # Session management hooks auto_fix.py # Post-edit diagnostics (ruff, pyright, py_compile) auto_watchdog.py # Watchdog arming after dispatch + auto_process.py # Scheduled inbox/task processing compact.py # Pre-compact memory archival rollover.py # Pre-compact memory rollover + pre_compact_prep.py # Pre-compact snapshot stamp (context/dispatch/plans) + session_start.py # SessionStart cadence reset notification/ # Alert hooks announce.py # Inbox banner on prompt email.py # Email notification stop_sound.py # Sound on session stop tool_sound.py # Sound on tool use + telegram_response.py # Telegram reply delivery on Stop config/ - loader.py # hooks.json discovery + validation - diagnostics.py # Diagnostics config + loader.py # hooks.json discovery + validation, config-independent trust checks + trust_registry.py # Trusted-project registry (enroll/revoke/hash checks) + diagnostics.py # JSONL diagnostics config logs/ engine.jsonl # JSONL diagnostics (every hook execution) -tests/ # 15 test files, 244 tests +tests/ # 42 test files, 1206 tests ``` ## Handler Categories | Category | Count | Handlers | |----------|-------|----------| -| prompt | 4 | branch_loader, tier0_kernel, navmap, identity | -| security | 3 | edit_gate, git_gate, subagent_gate | -| lifecycle | 4 | auto_fix, auto_watchdog, compact, rollover | -| notification | 4 | announce, email, stop_sound, tool_sound | +| prompt | 8 | branch_loader, tier0_kernel, navmap, identity, compass_recall, feedback_pulse, context_gauge, persistent_alert | +| security | 6 | presence_gate, edit_gate, git_gate, rm_gate, registry_gate, subagent_gate | +| lifecycle | 7 | auto_fix, auto_watchdog, auto_process, compact, rollover, pre_compact_prep, session_start | +| notification | 5 | announce, email, stop_sound, tool_sound, telegram_response | ## How It Works @@ -90,6 +103,8 @@ tests/ # 15 test files, 244 tests hooks.json alone is not live: UserPromptSubmit + PreCompact are invoked per-handler (`claude.py Event:name`) -- handlers on those events ALSO need a command entry in `.claude/provider_manifest.json` (PreCompact: manual + auto pair). Verify with firing evidence in engine.jsonl, not just the suite. +EVERY reply that adds/renames/moves a handler MUST state either "provider settings update needed: " or "no provider wire needed" -- never silent. Devpulse + Patrick apply live-settings changes; flag it every time, even if the manifest is already updated. + ## Integration - **Depends on:** @prax for logging (system_logger for prax monitor visibility) diff --git a/src/aipass/hooks/README.md b/src/aipass/hooks/README.md index 538d2d52..981bb4d6 100644 --- a/src/aipass/hooks/README.md +++ b/src/aipass/hooks/README.md @@ -114,7 +114,7 @@ src/aipass/hooks/ │ └── diagnostics.py # JSONL logging for hook execution ├── logs/ │ └── engine.jsonl # JSONL diagnostics (every hook execution) -└── tests/ # 1071 tests across 29 test files +└── tests/ # 1206 tests across 42 test files ``` ## How It Works diff --git a/src/aipass/hooks/apps/handlers/config/loader.py b/src/aipass/hooks/apps/handlers/config/loader.py index 19118990..62152f83 100644 --- a/src/aipass/hooks/apps/handlers/config/loader.py +++ b/src/aipass/hooks/apps/handlers/config/loader.py @@ -56,3 +56,37 @@ def find_project_config() -> dict | None: return None search = search.parent return None + + +def trust_break_banner() -> str | None: + """Loud, config-independent check for a stale trust enrollment. + + find_project_config() goes silent on a hash mismatch (logs one WARNING, + returns None) — the fallback config downstream then has no event_type + keys at all, so every hook including this one's own dispatch path goes + dark. This check does its own walk-up and hash comparison, never touches + hooks.json content, and does not depend on the project being trusted — + so it still works precisely when trust is broken. Returns None when + nothing is broken, or when the project was simply never enrolled (normal + first-run state, not a break). + """ + from aipass.hooks.apps.handlers.config.trust_registry import is_hash_mismatch + + search = Path.cwd() + home = Path.home() + while search != home and search.parent != search: + config_file = search / ".aipass" / "hooks.json" + if config_file.exists(): + if is_hash_mismatch(str(search)): + return ( + "# TRUST BREAK — ALL AIPASS HOOKS DISABLED\n\n" + f"{search}/.aipass/hooks.json no longer matches its enrolled hash. " + "Every hook for this project (including this warning's own delivery " + "path) is silently skipped until a human re-enrolls.\n\n" + "Fix: drone @hooks trust enroll (or: aipass init update)\n" + "This does not auto-heal — re-enrollment is a deliberate human " + "checkpoint, not a bug." + ) + return None + search = search.parent + return None diff --git a/src/aipass/hooks/apps/handlers/config/trust_registry.py b/src/aipass/hooks/apps/handlers/config/trust_registry.py index 6258d086..4e919c83 100644 --- a/src/aipass/hooks/apps/handlers/config/trust_registry.py +++ b/src/aipass/hooks/apps/handlers/config/trust_registry.py @@ -100,6 +100,25 @@ def is_trusted(project_dir: str) -> bool: return current_hash == entry.get("config_hash", "") +def is_hash_mismatch(project_dir: str) -> bool: + """True only when a project WAS enrolled but its hooks.json hash has since changed. + + Distinct from is_trusted()==False for a never-enrolled project (normal + first-run state, not a break). This flags a genuine trust break — an + existing enrollment gone stale — so callers can warn loudly instead of + treating it the same as "not set up yet". + """ + project_path = str(Path(project_dir).resolve()) + registry = read_registry() + entry = registry["projects"].get(project_path) + if entry is None: + return False + config_path = Path(project_dir).resolve() / ".aipass" / "hooks.json" + if not config_path.exists(): + return False + return _hash_file(config_path) != entry.get("config_hash", "") + + def bootstrap() -> bool: """Bootstrap the registry with ONLY the AIPass install. Returns True on success. diff --git a/src/aipass/hooks/apps/handlers/prompt/persistent_alert.py b/src/aipass/hooks/apps/handlers/prompt/persistent_alert.py index 98f2f81c..f5a4559c 100644 --- a/src/aipass/hooks/apps/handlers/prompt/persistent_alert.py +++ b/src/aipass/hooks/apps/handlers/prompt/persistent_alert.py @@ -11,13 +11,16 @@ """Injects advisory banners for active alerts from .aipass/alerts.json.""" import json +import os +import tempfile from datetime import datetime, timezone from pathlib import Path from aipass.hooks.apps.handlers.json import json_handler from aipass.prax.apps.modules.logger import system_logger as logger -_announced: set[str] = set() +_GUARD_DIR = Path(tempfile.gettempdir()) +_MAX_ALERTS_SHOWN = 10 def _find_aipass_dir() -> Path | None: @@ -77,10 +80,33 @@ def _load_and_clean(alerts_path: Path) -> list[dict]: return active +def _guard_path(session_id: str, alert_id: str) -> Path | None: + if not session_id: + return None + return _GUARD_DIR / f"aipass-persistent-alert-{session_id}-{alert_id}" + + +def _already_announced(session_id: str, alert_id: str) -> bool: + path = _guard_path(session_id, alert_id) + return path is not None and path.exists() + + +def _mark_announced(session_id: str, alert_id: str) -> None: + path = _guard_path(session_id, alert_id) + if path is not None: + try: + path.touch() + except OSError as exc: + logger.info("[HOOKS] persistent_alert: guard write failed: %s", exc) + + def _format_banner(alerts: list[dict]) -> str: - """Format alert banners for prompt injection.""" + """Format alert banners for prompt injection, capped at _MAX_ALERTS_SHOWN.""" + shown = alerts[:_MAX_ALERTS_SHOWN] + hidden = len(alerts) - len(shown) + lines = [] - for alert in alerts: + for alert in shown: severity = alert.get("severity", "warning").upper() title = alert.get("title", "Untitled alert") body = alert.get("body", "") @@ -89,6 +115,8 @@ def _format_banner(alerts: list[dict]) -> str: lines.append(f"[{severity}] {title} (from @{source}, id: {alert_id})") if body: lines.append(f" {body}") + if hidden > 0: + lines.append(f"...and {hidden} more (dismiss some to see the rest)") header = "# Active Alerts" dismiss_hint = "Dismiss with: drone @hooks dismiss " return "\n".join([header, ""] + lines + ["", dismiss_hint]) @@ -117,10 +145,12 @@ def handle(hook_data: dict) -> dict: banner = _format_banner(alerts) - new_ids = [a["id"] for a in alerts if a.get("id") and a["id"] not in _announced] + session_id = hook_data.get("session_id", "") or os.environ.get("CLAUDE_CODE_SESSION_ID", "") + new_ids = [a["id"] for a in alerts if a.get("id") and not _already_announced(session_id, a["id"])] sound = "" if new_ids: - _announced.update(new_ids) + for alert_id in new_ids: + _mark_announced(session_id, alert_id) count = len(alerts) plural = "s" if count != 1 else "" sound = f"alert: {count} active alert{plural}" diff --git a/src/aipass/hooks/apps/modules/engine.py b/src/aipass/hooks/apps/modules/engine.py index b2583f40..af1133c3 100644 --- a/src/aipass/hooks/apps/modules/engine.py +++ b/src/aipass/hooks/apps/modules/engine.py @@ -167,6 +167,16 @@ def dispatch(event_type: str, stdin_data: str, config: dict) -> tuple[str, int]: return "", 0 event_hooks = config.get(event_type, {}) + + if event_type == "UserPromptSubmit" and "presence_gate" in event_hooks: + from aipass.hooks.apps.handlers.config.loader import trust_break_banner + + banner = trust_break_banner() + if banner: + logger.error("[HOOKS] trust break detected — emitting loud banner") + _log({"ts": time.time(), "event": event_type, "action": "trust_break_banner"}) + return banner, 0 + if not event_hooks: _log({"ts": time.time(), "event": event_type, "action": "no_hooks_configured"}) return "", 0 diff --git a/src/aipass/hooks/tests/test_engine.py b/src/aipass/hooks/tests/test_engine.py index 958963fb..d30202a6 100644 --- a/src/aipass/hooks/tests/test_engine.py +++ b/src/aipass/hooks/tests/test_engine.py @@ -272,6 +272,54 @@ class TestDispatch: assert "ok" in result[0] assert result[1] == 0 + def test_trust_break_banner_short_circuits_presence_gate_dispatch(self, mock_logger): + """A hash-mismatch banner must reach the user even though presence_gate's own hook_def is empty.""" + config = {"hooks_enabled": True, "UserPromptSubmit": {"presence_gate": {}}} + with ( + patch("aipass.hooks.apps.modules.engine._log"), + patch( + "aipass.hooks.apps.handlers.config.loader.trust_break_banner", + return_value="# TRUST BREAK — ALL AIPASS HOOKS DISABLED", + ), + patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run, + ): + result = dispatch("UserPromptSubmit", "{}", config) + mock_run.assert_not_called() + assert result == ("# TRUST BREAK — ALL AIPASS HOOKS DISABLED", 0) + + def test_no_trust_break_falls_through_to_normal_dispatch(self, mock_logger): + config = {"hooks_enabled": True, "UserPromptSubmit": {"presence_gate": {}}} + with ( + patch("aipass.hooks.apps.modules.engine._log"), + patch("aipass.hooks.apps.handlers.config.loader.trust_break_banner", return_value=None), + ): + result = dispatch("UserPromptSubmit", "{}", config) + assert result == ("", 0) + + def test_trust_break_check_skipped_when_presence_gate_not_dispatched(self, mock_logger): + """The check is scoped to the presence_gate-filtered bridge call, not every UserPromptSubmit dispatch.""" + config = { + "hooks_enabled": True, + "UserPromptSubmit": {"other_hook": {"enabled": True, "command": "echo hi", "matcher": ""}}, + } + with ( + patch("aipass.hooks.apps.modules.engine._log"), + patch("aipass.hooks.apps.handlers.config.loader.trust_break_banner") as mock_banner, + patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run, + ): + mock_run.return_value = {"exit_code": 0, "stdout": "hi", "stderr": "", "elapsed_ms": 5} + dispatch("UserPromptSubmit", "{}", config) + mock_banner.assert_not_called() + + def test_trust_break_check_skipped_for_non_prompt_events(self, mock_logger): + config = {"hooks_enabled": True, "PreToolUse": {"presence_gate": {}}} + with ( + patch("aipass.hooks.apps.modules.engine._log"), + patch("aipass.hooks.apps.handlers.config.loader.trust_break_banner") as mock_banner, + ): + dispatch("PreToolUse", "{}", config) + mock_banner.assert_not_called() + class TestFindProjectConfig: """Tests for find_project_config() CWD walk.""" diff --git a/src/aipass/hooks/tests/test_persistent_alert.py b/src/aipass/hooks/tests/test_persistent_alert.py index 782846f7..42db51dc 100644 --- a/src/aipass/hooks/tests/test_persistent_alert.py +++ b/src/aipass/hooks/tests/test_persistent_alert.py @@ -257,19 +257,20 @@ class TestExpiredAlertCleanup: class TestAlertSound: - """Sound fires on first injection only.""" + """Sound fires once per alert per session (session-keyed tempdir guard).""" def test_sound_on_first_injection(self, tmp_path): from aipass.hooks.apps.handlers.prompt import persistent_alert - persistent_alert._announced.clear() - aipass_dir = tmp_path / ".aipass" aipass_dir.mkdir() _write_alerts(aipass_dir, [_make_alert(alert_id="snd-001")]) - with patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir): - result = persistent_alert.handle({}) + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + result = persistent_alert.handle({"session_id": "s-snd-001"}) assert "sound" in result assert "1 active alert" in result["sound"] @@ -277,29 +278,59 @@ class TestAlertSound: def test_no_sound_on_repeat_injection(self, tmp_path): from aipass.hooks.apps.handlers.prompt import persistent_alert - persistent_alert._announced.clear() - aipass_dir = tmp_path / ".aipass" aipass_dir.mkdir() _write_alerts(aipass_dir, [_make_alert(alert_id="snd-002")]) - with patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir): - persistent_alert.handle({}) - result = persistent_alert.handle({}) + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + persistent_alert.handle({"session_id": "s-snd-002"}) + result = persistent_alert.handle({"session_id": "s-snd-002"}) assert "sound" not in result - def test_sound_on_new_alert_added(self, tmp_path): + def test_fresh_process_same_session_still_dedupes(self, tmp_path): + """Regression: module-global set used to reset every process (real bridge calls + are fresh processes each time) so sound fired on every prompt. Guard file persists + across separate handle() calls even after re-importing the module fresh.""" + import importlib + from aipass.hooks.apps.handlers.prompt import persistent_alert - persistent_alert._announced.clear() + aipass_dir = tmp_path / ".aipass" + aipass_dir.mkdir() + _write_alerts(aipass_dir, [_make_alert(alert_id="snd-fresh")]) + + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + persistent_alert.handle({"session_id": "s-fresh"}) + + fresh_module = importlib.reload(persistent_alert) + with ( + patch.object(fresh_module, "_find_aipass_dir", return_value=aipass_dir), + patch.object(fresh_module, "_GUARD_DIR", tmp_path), + ): + result = fresh_module.handle({"session_id": "s-fresh"}) + + assert "sound" not in result + importlib.reload(persistent_alert) + + def test_sound_on_new_alert_added(self, tmp_path): + from aipass.hooks.apps.handlers.prompt import persistent_alert aipass_dir = tmp_path / ".aipass" aipass_dir.mkdir() _write_alerts(aipass_dir, [_make_alert(alert_id="snd-003")]) - with patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir): - persistent_alert.handle({}) + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + persistent_alert.handle({"session_id": "s-snd-003"}) _write_alerts( aipass_dir, @@ -309,27 +340,86 @@ class TestAlertSound: ], ) - with patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir): - result = persistent_alert.handle({}) + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + result = persistent_alert.handle({"session_id": "s-snd-003"}) assert "sound" in result assert "2 active alerts" in result["sound"] - def test_no_sound_when_no_alerts(self, tmp_path): + def test_different_sessions_both_hear_sound(self, tmp_path): from aipass.hooks.apps.handlers.prompt import persistent_alert - persistent_alert._announced.clear() + aipass_dir = tmp_path / ".aipass" + aipass_dir.mkdir() + _write_alerts(aipass_dir, [_make_alert(alert_id="snd-indep")]) + + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + first = persistent_alert.handle({"session_id": "s-a"}) + second = persistent_alert.handle({"session_id": "s-b"}) + + assert "sound" in first + assert "sound" in second + + def test_no_sound_when_no_alerts(self, tmp_path): + from aipass.hooks.apps.handlers.prompt import persistent_alert aipass_dir = tmp_path / ".aipass" aipass_dir.mkdir() _write_alerts(aipass_dir, []) - with patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir): - result = persistent_alert.handle({}) + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + result = persistent_alert.handle({"session_id": "s-none"}) assert "sound" not in result +class TestAlertBannerCap: + """Banner truncates at _MAX_ALERTS_SHOWN with a hidden-count note.""" + + def test_cap_truncates_and_notes_hidden_count(self, tmp_path): + from aipass.hooks.apps.handlers.prompt import persistent_alert + + aipass_dir = tmp_path / ".aipass" + aipass_dir.mkdir() + alerts = [_make_alert(alert_id=f"cap-{i}", title=f"Alert {i}") for i in range(13)] + _write_alerts(aipass_dir, alerts) + + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + result = persistent_alert.handle({"session_id": "s-cap"}) + + assert "Alert 0" in result["stdout"] + assert "Alert 9" in result["stdout"] + assert "Alert 10" not in result["stdout"] + assert "...and 3 more" in result["stdout"] + + def test_under_cap_no_hidden_note(self, tmp_path): + from aipass.hooks.apps.handlers.prompt import persistent_alert + + aipass_dir = tmp_path / ".aipass" + aipass_dir.mkdir() + _write_alerts(aipass_dir, [_make_alert(alert_id="under-cap")]) + + with ( + patch.object(persistent_alert, "_find_aipass_dir", return_value=aipass_dir), + patch.object(persistent_alert, "_GUARD_DIR", tmp_path), + ): + result = persistent_alert.handle({"session_id": "s-under-cap"}) + + assert "more (dismiss some" not in result["stdout"] + + class TestAlertDismiss: """drone @hooks dismiss behavior.""" diff --git a/src/aipass/hooks/tests/test_trust_registry.py b/src/aipass/hooks/tests/test_trust_registry.py index 5487fd0d..ba46bb76 100644 --- a/src/aipass/hooks/tests/test_trust_registry.py +++ b/src/aipass/hooks/tests/test_trust_registry.py @@ -17,11 +17,12 @@ from aipass.hooks.apps.handlers.config.trust_registry import ( _hash_file, bootstrap, enroll, + is_hash_mismatch, is_trusted, read_registry, revoke, ) -from aipass.hooks.apps.handlers.config.loader import find_project_config +from aipass.hooks.apps.handlers.config.loader import find_project_config, trust_break_banner class TestRegistryHelpers: @@ -151,6 +152,110 @@ class TestIsTrusted: assert is_trusted(str(project)) is False +class TestIsHashMismatch: + """Unit tests for is_hash_mismatch() — distinguishes a break from never-enrolled.""" + + def test_never_enrolled_is_not_a_mismatch(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + assert is_hash_mismatch("/not/registered") is False + + def test_enrolled_matching_hash_is_not_a_mismatch(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + assert is_hash_mismatch(str(project)) is False + + def test_enrolled_changed_hash_is_a_mismatch(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + hooks_file = project / ".aipass" / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + hooks_file.write_text('{"hooks_enabled": true, "tampered": true}') + assert is_hash_mismatch(str(project)) is True + + def test_enrolled_but_config_deleted_is_not_a_mismatch(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + hooks_file = project / ".aipass" / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + hooks_file.unlink() + assert is_hash_mismatch(str(project)) is False + + +class TestTrustBreakBanner: + """Tests for loader.trust_break_banner() — the loud, config-independent signal.""" + + def test_no_hooks_json_anywhere_is_none(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + empty_dir = temp_test_dir / "no_project" + empty_dir.mkdir() + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=empty_dir), + patch("aipass.hooks.apps.handlers.config.loader.Path.home", return_value=temp_test_dir), + ): + assert trust_break_banner() is None + + def test_trusted_project_is_none(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "trusted_project" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=project), + ): + assert trust_break_banner() is None + + def test_never_enrolled_is_none_not_a_break(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + reg_path.write_text('{"version": 1, "projects": {}}') + project = temp_test_dir / "fresh_project" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=project), + ): + assert trust_break_banner() is None + + def test_hash_mismatch_returns_loud_banner(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "tampered_project" + project.mkdir() + (project / ".aipass").mkdir() + hooks_file = project / ".aipass" / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + hooks_file.write_text('{"hooks_enabled": true, "tampered": true}') + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=project), + ): + banner = trust_break_banner() + assert banner is not None + assert "TRUST BREAK" in banner + assert "trust enroll" in banner + + class TestBootstrap: """Tests for bootstrap() — enrolls ONLY AIPASS_HOME."""