Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions Gradata/src/gradata/_doctor.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,33 @@ def _check_disk_space(brain_path):
return {"name": "disk_space", "status": "error", "detail": str(e)}


def _check_missing_hook_brain_dirs(auto_remove: bool = True):
"""Warn on, and optionally remove, Claude hooks pointing at deleted brains."""
try:
from gradata.hooks.stale_hook_check import (
_missing_hook_brain_dirs,
_remove_missing_hook_brain_dirs,
)

missing = _remove_missing_hook_brain_dirs() if auto_remove else _missing_hook_brain_dirs()
except Exception as e:
return {"name": "hook_brain_dirs", "status": "error", "detail": str(e)}
if not missing:
return {
"name": "hook_brain_dirs",
"status": "ok",
"detail": "no missing hook BRAIN_DIR targets",
}
listed = ", ".join(str(path) for path in missing[:5])
suffix = "" if len(missing) <= 5 else f" (+{len(missing) - 5} more)"
action = "removed" if auto_remove else "found"
return {
"name": "hook_brain_dirs",
"status": "warn",
"detail": f"{action} {len(missing)} Gradata hook(s) pointing at missing BRAIN_DIR: {listed}{suffix}",
}


def _gradata_config_path() -> Path:
env = os.environ.get("GRADATA_CONFIG")
if env:
Expand Down Expand Up @@ -495,6 +522,7 @@ def diagnose(
_check_manifest(brain_path),
_check_vectorstore(brain_path),
_check_disk_space(brain_path),
_check_missing_hook_brain_dirs(),
]
if include_cloud:
checks.extend(_cloud_checks())
Expand Down
54 changes: 53 additions & 1 deletion Gradata/src/gradata/hooks/adapters/claude_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,11 +64,12 @@ def install(brain_dir: Path, agent_config_path: Path) -> InstallResult:
sig = hook_signature(AGENT, brain_dir)
data = read_json(agent_config_path)
hooks = data.setdefault("hooks", {})
if any(sig in str(item) for groups in hooks.values() for item in (groups if isinstance(groups, list) else [])):
if _has_exact_installed_hook_set(hooks, sig):
return InstallResult(
AGENT, agent_config_path, "already_present", "hook already present"
)
for event, matcher, module in HOOKS:
_remove_existing_module_hook(hooks, event, module)
group = {
"matcher": "*",
"hooks": [
Expand All @@ -90,6 +91,57 @@ def install(brain_dir: Path, agent_config_path: Path) -> InstallResult:
return failure(AGENT, agent_config_path, exc)


def _is_gradata_module_hook(item: object, event: str, module: str) -> bool:
text = str(item)
return f"gradata.hooks.{module}" in text or (
f"gradata:{AGENT}:" in text and f":{event}:{module}" in text
)


def _has_exact_installed_hook_set(hooks: dict, sig: str) -> bool:
"""Return true only when every Gradata module hook is the desired one."""
for event, _matcher, module in HOOKS:
entries = hooks.get(event)
if not isinstance(entries, list):
return False
module_entries = [
entry for entry in entries if _is_gradata_module_hook(entry, event, module)
]
if len(module_entries) != 1:
return False
if f"{sig}:{event}:{module}" not in str(module_entries[0]):
return False
Comment on lines +101 to +113

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

_has_exact_installed_hook_set() can miss duplicates inside a single group.

Line 107 builds module_entries from top-level event entries. If one group contains two gradata.hooks.<module> commands, it still counts as one entry and Line 110 passes, so install() returns already_present and skips dedupe.

Suggested fix
 def _has_exact_installed_hook_set(hooks: dict, sig: str) -> bool:
     """Return true only when every Gradata module hook is the desired one."""
     for event, _matcher, module in HOOKS:
         entries = hooks.get(event)
         if not isinstance(entries, list):
             return False
-        module_entries = [
-            entry for entry in entries if _is_gradata_module_hook(entry, event, module)
-        ]
-        if len(module_entries) != 1:
+        module_hooks: list[object] = []
+        for entry in entries:
+            if isinstance(entry, dict) and isinstance(entry.get("hooks"), list):
+                module_hooks.extend(
+                    hook
+                    for hook in entry["hooks"]
+                    if _is_gradata_module_hook(hook, event, module)
+                )
+            elif _is_gradata_module_hook(entry, event, module):
+                module_hooks.append(entry)
+        if len(module_hooks) != 1:
             return False
-        if f"{sig}:{event}:{module}" not in str(module_entries[0]):
+        if f"{sig}:{event}:{module}" not in str(module_hooks[0]):
             return False
     return True
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@Gradata/src/gradata/hooks/adapters/claude_code.py` around lines 101 - 113,
The function _has_exact_installed_hook_set currently can miss duplicate
installed hooks inside a single event group; update the loop so you explicitly
detect duplicates by ensuring module_entries contains exactly one matching entry
(len(module_entries) == 1) and that the single entry exactly equals the expected
signature (compare str(module_entries[0]) == f"{sig}:{event}:{module}" rather
than using substring containment); if len(module_entries) != 1 or the equality
check fails, return False. Use the existing HOOKS and _is_gradata_module_hook
helpers to build module_entries but rely on explicit count+exact-string-check to
catch duplicates.

return True


def _remove_existing_module_hook(hooks: dict, event: str, module: str) -> None:
entries = hooks.get(event)
if not isinstance(entries, list):
return
kept_groups: list = []
for group in entries:
if not isinstance(group, dict):
if not _is_gradata_module_hook(group, event, module):
kept_groups.append(group)
continue
group_hooks = group.get("hooks")
if not isinstance(group_hooks, list):
if not _is_gradata_module_hook(group, event, module):
kept_groups.append(group)
continue
kept_hooks = [
hook for hook in group_hooks if not _is_gradata_module_hook(hook, event, module)
]
if kept_hooks:
new_group = dict(group)
new_group["hooks"] = kept_hooks
kept_groups.append(new_group)
if kept_groups:
hooks[event] = kept_groups
else:
hooks.pop(event, None)


def uninstall(brain_dir: Path, agent_config_path: Path) -> InstallResult:
"""Reverse ``install()`` across all hook events.

Expand Down
80 changes: 73 additions & 7 deletions Gradata/src/gradata/hooks/stale_hook_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,16 +163,15 @@ def _extract_brain_dir_from_command(command: str) -> Path | None:

def _missing_hook_brain_dirs(settings_path: Path | None = None) -> list[Path]:
"""Find Gradata hook BRAIN_DIR targets in Claude settings that no longer exist."""
if settings_path is None and (env_str("GRADATA_HOOK_ROOT") or env_str("GRADATA_HOOK_ROOT_POST")):
if settings_path is None and (
env_str("GRADATA_HOOK_ROOT") or env_str("GRADATA_HOOK_ROOT_POST")
):
# Test/dev hook roots are intentionally isolated; do not let a developer's
# real ~/.claude/settings.json leak warnings into those runs.
return []
path = settings_path or _settings_path()
if not path.is_file():
return []
try:
settings = json.loads(path.read_text(encoding="utf-8"))
except Exception:
settings = _read_settings(path)
if not settings:
return []
hooks = settings.get("hooks", {})
if not isinstance(hooks, dict):
Expand All @@ -198,6 +197,71 @@ def _missing_hook_brain_dirs(settings_path: Path | None = None) -> list[Path]:
return missing


def _read_settings(path: Path) -> dict | None:
if not path.is_file():
return None
try:
settings = json.loads(path.read_text(encoding="utf-8"))
except Exception:
return None
return settings if isinstance(settings, dict) else None


def _remove_missing_hook_brain_dirs(settings_path: Path | None = None) -> list[Path]:
"""Remove Gradata hook commands whose BRAIN_DIR targets no longer exist."""
path = settings_path or _settings_path()
settings = _read_settings(path)
if not settings:
return []
hooks = settings.get("hooks")
Comment on lines +210 to +216

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Preserve hook-root isolation in the removal path.

Line 210 starts a destructive cleanup path, but unlike Line 166 it does not short-circuit when GRADATA_HOOK_ROOT/GRADATA_HOOK_ROOT_POST are set. Since doctor calls this remover by default, isolated test/dev runs can still mutate a real Claude settings file.

Suggested fix
 def _remove_missing_hook_brain_dirs(settings_path: Path | None = None) -> list[Path]:
     """Remove Gradata hook commands whose BRAIN_DIR targets no longer exist."""
+    if settings_path is None and (
+        env_str("GRADATA_HOOK_ROOT") or env_str("GRADATA_HOOK_ROOT_POST")
+    ):
+        return []
     path = settings_path or _settings_path()
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@Gradata/src/gradata/hooks/stale_hook_check.py` around lines 210 - 216, The
_remove_missing_hook_brain_dirs function can mutate real settings when
GRADATA_HOOK_ROOT or GRADATA_HOOK_ROOT_POST are set; add the same isolation
guard used earlier (see the check around line 166) to short-circuit and return
an empty list if either environment overrides are present, before reading or
modifying settings. Specifically, in _remove_missing_hook_brain_dirs (and using
the same helpers _settings_path and _read_settings), check for
GRADATA_HOOK_ROOT/GRADATA_HOOK_ROOT_POST and immediately return [] to preserve
hook-root isolation for test/dev runs.

if not isinstance(hooks, dict):
return []
removed: list[Path] = []
seen: set[str] = set()
for event in list(hooks):
groups = hooks.get(event)
if not isinstance(groups, list):
continue
kept_groups: list = []
for group in groups:
if not isinstance(group, dict):
kept_groups.append(group)
continue
group_hooks = group.get("hooks")
if not isinstance(group_hooks, list):
kept_groups.append(group)
continue
kept_hook_entries: list = []
for hook in group_hooks:
if not isinstance(hook, dict):
kept_hook_entries.append(hook)
continue
brain_dir = _extract_brain_dir_from_command(str(hook.get("command", "")))
if brain_dir is None or brain_dir.exists():
kept_hook_entries.append(hook)
continue
key = brain_dir.as_posix()
if key not in seen:
removed.append(brain_dir)
seen.add(key)
if kept_hook_entries:
new_group = dict(group)
new_group["hooks"] = kept_hook_entries
kept_groups.append(new_group)
if kept_groups:
hooks[event] = kept_groups
else:
hooks.pop(event, None)
if removed:
if hooks:
settings["hooks"] = hooks
else:
settings.pop("hooks", None)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(settings, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return removed
Comment on lines +260 to +262

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Use the atomic JSON write helper for settings persistence.

Line 261 writes settings.json directly; a mid-write interruption can leave a truncated file.

As per coding guidelines, “Use atomic-write helper when writing JSON files to prevent corruption from mid-write crashes.”

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@Gradata/src/gradata/hooks/stale_hook_check.py` around lines 260 - 262, The
code currently calls path.parent.mkdir(...) then
path.write_text(json.dumps(settings,...)) in stale_hook_check.py which can
produce a truncated settings.json if interrupted; replace the direct write with
the project's atomic JSON write helper (e.g., use the atomic-write helper
function used elsewhere) to write the serialized settings atomically: ensure the
directory is created as before, then call the helper (passing the target Path
and the settings dict/serialized JSON) instead of path.write_text; keep the same
formatting (indent and sort_keys) and return removed as before.

Source: Coding guidelines



def _hook_dirs() -> list[Path]:
pre = os.environ.get("GRADATA_HOOK_ROOT") or ".claude/hooks/pre-tool/generated"
post = os.environ.get("GRADATA_HOOK_ROOT_POST") or ".claude/hooks/post-tool/generated"
Expand Down Expand Up @@ -280,7 +344,9 @@ def main() -> int:
for path in missing_brain_dirs:
print(f" - missing BRAIN_DIR: {path}")
target_brain = env_str("GRADATA_BRAIN") or env_str("BRAIN_DIR") or "~/.gradata/brain"
print(f" fix: gradata install --agent claude-code --brain {shlex.quote(target_brain)}")
print(
f" fix: gradata install --agent claude-code --brain {shlex.quote(target_brain)}"
)
print()

return 0
Expand Down
59 changes: 59 additions & 0 deletions Gradata/tests/test_doctor_hooks.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
from __future__ import annotations

import json
from pathlib import Path

from gradata import _doctor


def test_doctor_warns_and_removes_missing_gradata_hook_brain_dirs(
tmp_path: Path, monkeypatch
) -> None:
home = tmp_path / "home"
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
missing_brain = tmp_path / "pytest-55" / "test_hook_adapter_install_is_i0" / "brain"
live_brain = tmp_path / "live-brain"
live_brain.mkdir()
settings_path = home / ".claude" / "settings.json"
settings_path.parent.mkdir(parents=True)
settings_path.write_text(
json.dumps(
{
"hooks": {
"PreToolUse": [
{
"matcher": "*",
"hooks": [
{
"type": "command",
"command": f"BRAIN_DIR={missing_brain} python -m gradata.hooks.inject_brain_rules",
}
],
},
{
"matcher": "*",
"hooks": [
{
"type": "command",
"command": f"BRAIN_DIR={live_brain} python -m gradata.hooks.inject_brain_rules",
}
],
},
]
}
}
),
encoding="utf-8",
)

check = _doctor._check_missing_hook_brain_dirs(auto_remove=True)

assert check["status"] == "warn"
assert str(missing_brain) in check["detail"]
settings = json.loads(settings_path.read_text(encoding="utf-8"))
commands = [
hook["command"] for group in settings["hooks"]["PreToolUse"] for hook in group["hooks"]
]
assert all(str(missing_brain) not in command for command in commands)
assert any(str(live_brain) in command for command in commands)
83 changes: 83 additions & 0 deletions Gradata/tests/test_hook_adapters.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import json
import os
import tomllib
from pathlib import Path
Expand Down Expand Up @@ -129,3 +130,85 @@ def test_agent_brain_resolution_absolutizes_relative_cli_path(
assert _resolve_agent_brain_root(Namespace(brain="relative-brain", brain_dir=None)) == (
tmp_path / "relative-brain"
)


def test_claude_code_install_replaces_stale_same_event_module_hook(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Installing Claude hooks must not accumulate temp-brain duplicates."""
home = tmp_path / "home"
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
stale_brain = tmp_path / "missing-brain"
config_path = adapter_config_path("claude-code")
config_path.parent.mkdir(parents=True)
config_path.write_text(
json.dumps(
{
"hooks": {
"PreToolUse": [
{
"matcher": "*",
"hooks": [
{
"type": "command",
"command": f"BRAIN_DIR={stale_brain} python -m gradata.hooks.inject_brain_rules",
}
],
}
]
}
}
),
encoding="utf-8",
)
brain_dir = tmp_path / "brain"
brain_dir.mkdir()

result = get_adapter("claude-code").install(brain_dir, config_path)

assert result.action == "added"
settings = json.loads(config_path.read_text(encoding="utf-8"))
pre_tool_use = settings["hooks"]["PreToolUse"]
commands = [hook["command"] for group in pre_tool_use for hook in group["hooks"]]
inject_commands = [cmd for cmd in commands if "gradata.hooks.inject_brain_rules" in cmd]
assert len(inject_commands) == 1
assert str(stale_brain) not in inject_commands[0]


def test_claude_code_reinstall_removes_stale_duplicate_when_current_hook_exists(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
brain_dir = tmp_path / "brain"
brain_dir.mkdir()
config_path = adapter_config_path("claude-code")
adapter = get_adapter("claude-code")
assert adapter.install(brain_dir, config_path).action == "added"
assert adapter.install(brain_dir, config_path).action == "already_present"

stale_brain = tmp_path / "missing-brain"
settings = json.loads(config_path.read_text(encoding="utf-8"))
settings["hooks"]["PreToolUse"].append(
{
"matcher": "*",
"hooks": [
{
"type": "command",
"command": f"BRAIN_DIR={stale_brain} python -m gradata.hooks.inject_brain_rules",
}
],
}
)
config_path.write_text(json.dumps(settings), encoding="utf-8")

assert adapter.install(brain_dir, config_path).action == "added"
settings = json.loads(config_path.read_text(encoding="utf-8"))
commands = [
hook["command"] for group in settings["hooks"]["PreToolUse"] for hook in group["hooks"]
]
inject_commands = [cmd for cmd in commands if "gradata.hooks.inject_brain_rules" in cmd]
assert len(inject_commands) == 1
assert str(stale_brain) not in inject_commands[0]
Loading