feat(hooks): add Claude PreCompact snapshot hook

Gradata/src/gradata/hooks/adapters/_base.py

@@ -136,6 +136,13 @@ def hook_command(brain_dir: Path) -> str:
     )
 
 
+def pre_compact_hook_command(brain_dir: Path) -> str:
+    return (
+        f"BRAIN_DIR={shlex.quote(str(brain_dir))} "
+        f"{shlex.quote(sys.executable)} -m gradata.hooks.pre_compact"
+    )
+
+
 def mcp_command(brain_dir: Path) -> list[str]:
     return [sys.executable, "-m", "gradata.mcp_server", "--brain-dir", str(brain_dir)]
 

Gradata/src/gradata/hooks/adapters/claude_code.py

@@ -12,17 +12,53 @@
     failure,
     hook_command,
     hook_signature,
+    pre_compact_hook_command,
     read_json,
     write_json,
 )
 
 AGENT = "claude-code"
+PRE_COMPACT_ID_SUFFIX = ":precompact"
 
 # Claude Code's canonical tool names (capitalised, no prefix).
 EDIT_TOOLS: frozenset[str] = frozenset({"Edit", "MultiEdit"}) | (EDIT_TOOL_ALIASES & {"Edit"})
 WRITE_TOOLS: frozenset[str] = frozenset({"Write"}) | (WRITE_TOOL_ALIASES & {"Write"})
 
 
+def _pre_compact_signature(brain_dir: Path) -> str:
+    return f"{hook_signature(AGENT, brain_dir)}{PRE_COMPACT_ID_SUFFIX}"
+
+
+def _hook_matches_signature(hook: object, signature: str) -> bool:
+    if isinstance(hook, dict):
+        return hook.get("id") == signature
+    return signature in str(hook)
+
+
+def _prune_matching_hooks(entries: list, signature: str) -> tuple[list, int]:
+    kept_entries: list = []
+    removed = 0
+    for entry in entries:
+        if not isinstance(entry, dict) or not isinstance(entry.get("hooks"), list):
+            if signature in str(entry):
+                removed += 1
+                continue
+            kept_entries.append(entry)
+            continue
+
+        kept_hooks = []
+        for hook in entry["hooks"]:
+            if _hook_matches_signature(hook, signature):
+                removed += 1
+                continue
+            kept_hooks.append(hook)
+        if kept_hooks:
+            pruned = dict(entry)
+            pruned["hooks"] = kept_hooks
+            kept_entries.append(pruned)
+    return kept_entries, removed
+
+
 def detect(payload: dict) -> bool:
     """Claude Code stdin signature: capitalised tool name + args at ``input``.
 
@@ -55,33 +91,55 @@ def extract_correction(
 def install(brain_dir: Path, agent_config_path: Path) -> InstallResult:
     try:
         sig = hook_signature(AGENT, brain_dir)
+        pre_compact_sig = _pre_compact_signature(brain_dir)
         data = read_json(agent_config_path)
         hooks = data.setdefault("hooks", {})
+
+        added = 0
         pre_tool = hooks.setdefault("PreToolUse", [])
-        if any(sig in str(item) for item in pre_tool):
+        if not any(sig in str(item) for item in pre_tool):
+            pre_tool.append(
+                {
+                    "matcher": "*",
+                    "hooks": [
+                        {
+                            "type": "command",
+                            "command": hook_command(brain_dir),
+                            "id": sig,
+                        }
+                    ],
+                }
+            )
+            added += 1
+
+        pre_compact = hooks.setdefault("PreCompact", [])
+        if not any(pre_compact_sig in str(item) for item in pre_compact):
+            pre_compact.append(
+                {
+                    "matcher": "manual|auto",
+                    "hooks": [
+                        {
+                            "type": "command",
+                            "command": pre_compact_hook_command(brain_dir),
+                            "id": pre_compact_sig,
+                        }
+                    ],
+                }
+            )
+            added += 1
+
+        if added == 0:
             return InstallResult(
                 AGENT, agent_config_path, "already_present", "hook already present"
             )
-        pre_tool.append(
-            {
-                "matcher": "*",
-                "hooks": [
-                    {
-                        "type": "command",
-                        "command": hook_command(brain_dir),
-                        "id": sig,
-                    }
-                ],
-            }
-        )
         write_json(agent_config_path, data)
-        return InstallResult(AGENT, agent_config_path, "added", "installed PreToolUse hook")
+        return InstallResult(AGENT, agent_config_path, "added", "installed PreToolUse and PreCompact hooks")
     except Exception as exc:
         return failure(AGENT, agent_config_path, exc)
 
 
 def uninstall(brain_dir: Path, agent_config_path: Path) -> InstallResult:
-    """Reverse ``install()``: drop the signature-matching PreToolUse entry.
+    """Reverse ``install()``: drop signature-matching hook entries.
 
     Idempotent — calling on an already-clean config returns ``already_present``
     (semantically: 'already in the desired absent state'). Empty containers
@@ -94,31 +152,33 @@ def uninstall(brain_dir: Path, agent_config_path: Path) -> InstallResult:
                 AGENT, agent_config_path, "already_present", "config file does not exist"
             )
         sig = hook_signature(AGENT, brain_dir)
+        pre_compact_sig = _pre_compact_signature(brain_dir)
         data = read_json(agent_config_path)
         hooks = data.get("hooks")
         if not isinstance(hooks, dict):
             return InstallResult(AGENT, agent_config_path, "already_present", "no hooks block")
+        removed = 0
+
         pre_tool = hooks.get("PreToolUse")
-        if not isinstance(pre_tool, list):
-            return InstallResult(AGENT, agent_config_path, "already_present", "no PreToolUse")
+        if isinstance(pre_tool, list):
+            kept, count = _prune_matching_hooks(pre_tool, sig)
+            removed += count
+            if kept:
+                hooks["PreToolUse"] = kept
+            else:
+                hooks.pop("PreToolUse", None)
+
+        pre_compact = hooks.get("PreCompact")
+        if isinstance(pre_compact, list):
+            kept, count = _prune_matching_hooks(pre_compact, pre_compact_sig)
+            removed += count
+            if kept:
+                hooks["PreCompact"] = kept
+            else:
+                hooks.pop("PreCompact", None)
 
-        removed = 0
-        kept: list = []
-        for entry in pre_tool:
-            entry_str = str(entry)
-            if sig in entry_str:
-                # Either the entry's `hooks[].id` carries our sig, or the
-                # whole entry was ours. Drop it.
-                removed += 1
-                continue
-            kept.append(entry)
         if removed == 0:
             return InstallResult(AGENT, agent_config_path, "already_present", "hook not present")
-
-        if kept:
-            hooks["PreToolUse"] = kept
-        else:
-            hooks.pop("PreToolUse", None)
         if not hooks:
             data.pop("hooks", None)
         write_json(agent_config_path, data)

Gradata/src/gradata/hooks/pre_compact.py

@@ -1,14 +1,16 @@
-"""PreCompact hook: save brain state snapshot before context compaction."""
+"""Claude Code PreCompact hook: snapshot brain context before compaction."""
 
 from __future__ import annotations
 
 import hashlib
 import json
-import os
-import tempfile
+import logging
+import re
 from datetime import UTC, datetime
 from pathlib import Path
+from typing import Any
 
+from gradata._atomic import atomic_write_text
 from gradata.hooks._base import resolve_brain_dir, run_hook
 from gradata.hooks._profiles import Profile
 
@@ -19,50 +21,94 @@
     "timeout": 5000,
 }
 
+SNAPSHOT_DIR_NAME = ".precompact-snapshots"
+MAX_FILE_CHARS = 200_000
+logger = logging.getLogger(__name__)
+_CONTEXT_FILES = (
+    "lessons.md",
+    "rules.md",
+    "meta-rules.json",
+    "brain.manifest.json",
+    "handoff.md",
+    "handoff.json",
+)
+_SAFE_SESSION_ID = re.compile(r"[^A-Za-z0-9_.-]+")
+
+
+def _session_id(payload: dict[str, Any]) -> str:
+    raw: Any = payload.get("session_id") or payload.get("sessionId")
+    session = payload.get("session")
+    if not raw and isinstance(session, dict):
+        raw = session.get("id")
+    if not isinstance(raw, str) or not raw.strip():
+        raw = hashlib.sha256(json.dumps(payload, sort_keys=True, default=str).encode()).hexdigest()[:16]
+    safe = _SAFE_SESSION_ID.sub("-", raw.strip()).strip(".-")
+    return safe or "unknown-session"
+
+
+def _read_context_file(path: Path) -> dict[str, Any] | None:
+    if not path.is_file():
+        return None
+    try:
+        with path.open("r", encoding="utf-8", errors="replace") as handle:
+            text = handle.read(MAX_FILE_CHARS + 1)
+    except OSError:
+        logger.warning("failed to read PreCompact context file %s", path, exc_info=True)
+        return None
+    truncated = len(text) > MAX_FILE_CHARS
+    if truncated:
+        text = text[:MAX_FILE_CHARS]
+    return {
+        "path": path.name,
+        "chars": path.stat().st_size,
+        "truncated": truncated,
+        "content": text,
+    }
+
+
+def _brain_context(brain_dir: Path) -> dict[str, Any]:
+    files: dict[str, dict[str, Any]] = {}
+    context: dict[str, Any] = {"files": files}
+    for rel in _CONTEXT_FILES:
+        item = _read_context_file(brain_dir / rel)
+        if item is not None:
+            files[rel] = item
+            context[rel.replace(".", "_").replace("-", "_")] = item["content"]
+    return context
+
+
+def write_snapshot(payload: dict[str, Any], brain_dir: Path) -> Path:
+    """Write a PreCompact snapshot and return its path."""
+    session_id = _session_id(payload)
+    snapshot_dir = brain_dir / SNAPSHOT_DIR_NAME
+    snapshot_dir.mkdir(parents=True, exist_ok=True)
+    snapshot_path = snapshot_dir / f"{session_id}.json"
+
+    snapshot = {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "hook_event_name": "PreCompact",
+        "session_id": session_id,
+        "compact_type": payload.get("type") or payload.get("compact_type") or payload.get("trigger"),
+        "trigger": payload.get("trigger") or payload.get("type") or payload.get("compact_type"),
+        "brain_dir": str(brain_dir),
+        "payload": payload,
+        "context": _brain_context(brain_dir),
+    }
+    atomic_write_text(snapshot_path, json.dumps(snapshot, indent=2, sort_keys=True) + "\n")
+    return snapshot_path
+
 
 def main(data: dict) -> dict | None:
+    brain_dir_str = resolve_brain_dir()
+    if not brain_dir_str:
+        return None
     try:
-        brain_dir_str = resolve_brain_dir()
-        if not brain_dir_str:
-            return None
-        brain_dir = Path(brain_dir_str)
-
-        compact_type = data.get("type", "unknown") if data else "unknown"
-
-        snapshot = {
-            "timestamp": datetime.now(UTC).isoformat(),
-            "compact_type": compact_type,
-            "brain_dir": str(brain_dir),
-        }
-
-        # Include lesson count if available
-        lessons_path = brain_dir / "lessons.md"
-        if lessons_path.is_file():
-            text = lessons_path.read_text(encoding="utf-8")
-            snapshot["lesson_count"] = len(
-                [
-                    line
-                    for line in text.splitlines()
-                    if (stripped := line.strip()) and not stripped.startswith("#")
-                ]
-            )
-
-        if hasattr(os, "getuid"):
-            uid = os.getuid()
-        else:
-            try:
-                uid = os.getlogin()
-            except OSError:
-                uid = f"pid{os.getpid()}"
-        user_tmp = Path(tempfile.gettempdir()) / f"gradata-{uid}"
-        user_tmp.mkdir(parents=True, exist_ok=True)
-        dir_hash = hashlib.md5(str(brain_dir).encode()).hexdigest()[:8]
-        snapshot_path = user_tmp / f"compact-snapshot-{dir_hash}.json"
-        snapshot_path.write_text(json.dumps(snapshot, indent=2), encoding="utf-8")
-
-        return {"result": "State saved before compaction"}
+        path = write_snapshot(data or {}, Path(brain_dir_str))
     except Exception:
+        logger.warning("failed to write PreCompact snapshot", exc_info=True)
         return None
+    return {"result": f"PreCompact snapshot saved to {path}"}
 
 
 if __name__ == "__main__":