diff --git a/Gradata/src/gradata/hooks/adapters/_base.py b/Gradata/src/gradata/hooks/adapters/_base.py index 7ea67bc5..b9945317 100644 --- a/Gradata/src/gradata/hooks/adapters/_base.py +++ b/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)] diff --git a/Gradata/src/gradata/hooks/adapters/claude_code.py b/Gradata/src/gradata/hooks/adapters/claude_code.py index 04e005a2..122428ef 100644 --- a/Gradata/src/gradata/hooks/adapters/claude_code.py +++ b/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) diff --git a/Gradata/src/gradata/hooks/pre_compact.py b/Gradata/src/gradata/hooks/pre_compact.py index 237f2c41..f2b5bbb6 100644 --- a/Gradata/src/gradata/hooks/pre_compact.py +++ b/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__": diff --git a/Gradata/tests/test_hook_adapters.py b/Gradata/tests/test_hook_adapters.py index bccdb3bd..278fde81 100644 --- a/Gradata/tests/test_hook_adapters.py +++ b/Gradata/tests/test_hook_adapters.py @@ -1,6 +1,8 @@ from __future__ import annotations +import json import os +import shlex import tomllib from pathlib import Path @@ -63,3 +65,60 @@ def test_adapter_install_does_not_touch_real_user_config(tmp_path: Path) -> None assert result.action == "added" after = real_config.read_text(encoding="utf-8") if real_config.exists() else None assert after == before + + +def test_claude_code_install_writes_pre_compact_entry(tmp_path: Path) -> None: + brain_dir = tmp_path / "brain" + brain_dir.mkdir() + config_path = tmp_path / ".claude" / "settings.json" + + result = get_adapter("claude-code").install(brain_dir, config_path) + + assert result.action == "added" + settings = json.loads(config_path.read_text(encoding="utf-8")) + hooks = settings["hooks"] + assert "PreToolUse" in hooks + pre_compact = hooks["PreCompact"] + assert any( + entry.get("matcher") == "manual|auto" + and any( + hook.get("type") == "command" + and "-m gradata.hooks.pre_compact" in hook.get("command", "") + and f"BRAIN_DIR={shlex.quote(str(brain_dir))}" in hook.get("command", "") + and hook.get("id", "").startswith("gradata:claude-code:") + and hook.get("id", "").endswith(":precompact") + for hook in entry.get("hooks", []) + ) + for entry in pre_compact + ) + + +def test_claude_code_uninstall_prunes_only_gradata_hooks_from_mixed_entries(tmp_path: Path) -> None: + brain_dir = tmp_path / "brain" + brain_dir.mkdir() + config_path = tmp_path / ".claude" / "settings.json" + adapter = get_adapter("claude-code") + adapter.install(brain_dir, config_path) + + settings = json.loads(config_path.read_text(encoding="utf-8")) + settings["hooks"]["PreToolUse"][0]["hooks"].append( + {"type": "command", "command": "echo user pretool", "id": "user:pretool"} + ) + settings["hooks"]["PreCompact"][0]["hooks"].append( + {"type": "command", "command": "echo user precompact", "id": "user:precompact"} + ) + config_path.write_text(json.dumps(settings), encoding="utf-8") + + result = adapter.uninstall(brain_dir, config_path) + + assert result.action == "removed" + after = json.loads(config_path.read_text(encoding="utf-8")) + assert after["hooks"]["PreToolUse"] == [ + {"matcher": "*", "hooks": [{"type": "command", "command": "echo user pretool", "id": "user:pretool"}]} + ] + assert after["hooks"]["PreCompact"] == [ + { + "matcher": "manual|auto", + "hooks": [{"type": "command", "command": "echo user precompact", "id": "user:precompact"}], + } + ] diff --git a/Gradata/tests/test_hooks_intelligence.py b/Gradata/tests/test_hooks_intelligence.py index d5f9d32b..725fa9aa 100644 --- a/Gradata/tests/test_hooks_intelligence.py +++ b/Gradata/tests/test_hooks_intelligence.py @@ -279,32 +279,24 @@ def test_config_validate_no_settings(): def test_pre_compact_saves_snapshot(tmp_path): - import hashlib - lessons = tmp_path / "lessons.md" lessons.write_text("[2026-04-01] [RULE:0.92] PROCESS: Plan first\n# header\n") - 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}" - dir_hash = hashlib.md5(str(tmp_path).encode()).hexdigest()[:8] - snapshot_path = user_tmp / f"compact-snapshot-{dir_hash}.json" + payload = {"type": "auto", "session_id": "legacy-session"} + snapshot_path = tmp_path / ".precompact-snapshots" / "legacy-session.json" snapshot_path.unlink(missing_ok=True) with patch.dict(os.environ, {"GRADATA_BRAIN_DIR": str(tmp_path)}): - result = compact_main({"type": "auto"}) + result = compact_main(payload) assert result is not None - assert "State saved" in result["result"] + assert "PreCompact snapshot saved" in result["result"] assert snapshot_path.exists() data = json.loads(snapshot_path.read_text()) assert data["compact_type"] == "auto" - assert data["lesson_count"] >= 1 + assert data["hook_event_name"] == "PreCompact" + assert data["payload"] == payload + assert data["context"]["files"]["lessons.md"]["content"].startswith("[2026-04-01]") snapshot_path.unlink(missing_ok=True) diff --git a/Gradata/tests/test_pre_compact_hook.py b/Gradata/tests/test_pre_compact_hook.py new file mode 100644 index 00000000..80594d49 --- /dev/null +++ b/Gradata/tests/test_pre_compact_hook.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + + +def test_pre_compact_hook_writes_snapshot_from_stdin(tmp_path: Path) -> None: + brain_dir = tmp_path / "brain" + brain_dir.mkdir() + (brain_dir / "lessons.md").write_text("# Lessons\n\n- Prefer focused tests.\n", encoding="utf-8") + + payload = { + "hook_event_name": "PreCompact", + "session_id": "session/abc", + "trigger": "manual", + "transcript_path": "/tmp/claude-transcript.jsonl", + } + env = os.environ.copy() + env["BRAIN_DIR"] = str(brain_dir) + env["GRADATA_TELEMETRY"] = "off" + env["PYTHONPATH"] = str(Path.cwd() / "src") + + result = subprocess.run( + [sys.executable, "-m", "gradata.hooks.pre_compact"], + input=json.dumps(payload), + text=True, + capture_output=True, + check=False, + cwd=Path.cwd(), + env=env, + ) + + assert result.returncode == 0, result.stderr + snapshot_path = brain_dir / ".precompact-snapshots" / "session-abc.json" + assert snapshot_path.is_file() + snapshot = json.loads(snapshot_path.read_text(encoding="utf-8")) + assert snapshot["hook_event_name"] == "PreCompact" + assert snapshot["session_id"] == "session-abc" + assert snapshot["payload"] == payload + assert snapshot["context"]["files"]["lessons.md"]["content"].startswith("# Lessons") + assert "PreCompact snapshot" in result.stdout