fix(hooks): persistent_alert cross-process dedup + loud trust-break banner (DPLAN-0253 trace round). Dedup: module-global _announced set replaced with session+alert-keyed tempdir guard files - fresh bridge process per prompt meant TTS would announce every turn while an alert existed; banner capped at 10 with overflow note. Trust: is_hash_mismatch distinguishes real break from never-enrolled, trust_break_banner does config-independent walk+hash, engine emits loud banner once per prompt via presence_gate call - a hooks.json edit had silently darkened the ENTIRE hook layer for 2+ hours (found by 5-agent trace round). No auto-heal, re-enroll stays human. Live-fired both ways by devpulse: hash broken = banner, restored = healthy. 16 new tests, 1206 green, seedgo 100 pct, both re-run by devpulse. Also: @hooks prompt corrected (taught one-entry-per-event model) + two-wires checklist + mandatory provider-wire flag; README tree/counts refreshed. Built by @hooks
This commit is contained in:
@@ -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):
|
||||
|
||||
|
||||
@@ -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: <exact entries>" 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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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 <alert-id>"
|
||||
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}"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user