diff --git a/src/primer/hook/installer.py b/src/primer/hook/installer.py index 96b51f5..914cf6b 100644 --- a/src/primer/hook/installer.py +++ b/src/primer/hook/installer.py @@ -94,19 +94,65 @@ def _save_settings(settings: dict, path: Path) -> None: # --------------------------------------------------------------------------- # Claude Code format helpers +# +# Claude Code uses a matcher-wrapped hook structure: +# { +# "EVENT_NAME": [ +# { +# "matcher": "", +# "hooks": [ +# {"type": "command", "command": "...", "timeout": 30} +# ] +# } +# ] +# } +# +# Timeouts are specified in SECONDS (not milliseconds). +# +# Detection also handles a legacy flat format that earlier versions of this +# installer wrote (`{"command": ..., "timeout": ...}` directly at the event +# level) so reinstall / uninstall can migrate existing users. # --------------------------------------------------------------------------- +_CLAUDE_HOOK_TIMEOUT_SECONDS = 30 -def _find_hook_claude(session_end_hooks: list, command: str) -> bool: - """Check if the Primer hook is in a Claude-format hook list.""" - for hook in session_end_hooks: - if isinstance(hook, dict) and hook.get("command") == command: + +def _find_hook_claude(event_hooks: list, command: str) -> bool: + """Check if the Primer hook is present in a Claude event hook list. + + Detects both the current matcher-wrapped format and the legacy flat format + written by earlier versions of this installer. + """ + for entry in event_hooks: + # Legacy flat format: {"command": cmd, ...} + if isinstance(entry, dict) and entry.get("command") == command: return True - if isinstance(hook, str) and hook == command: + if isinstance(entry, str) and entry == command: return True + # Current matcher-wrapped format: {"matcher": ..., "hooks": [{"command": cmd}]} + if isinstance(entry, dict): + inner_hooks = entry.get("hooks", []) + if isinstance(inner_hooks, list): + for inner in inner_hooks: + if isinstance(inner, dict) and inner.get("command") == command: + return True return False +def _new_claude_hook_entry(command: str) -> dict: + """Build a matcher-wrapped hook entry in Claude Code's expected schema.""" + return { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": command, + "timeout": _CLAUDE_HOOK_TIMEOUT_SECONDS, + } + ], + } + + def _install_claude(settings: dict, config: AgentHookConfig) -> bool: """Add the Primer hook to Claude Code settings. Returns True if already present. @@ -119,40 +165,63 @@ def _install_claude(settings: dict, config: AgentHookConfig) -> bool: already_end = False already_compact = False - # SessionEnd session_end_hooks = hooks.setdefault("SessionEnd", []) if _find_hook_claude(session_end_hooks, config.hook_command): already_end = True else: - session_end_hooks.append({"command": config.hook_command, "timeout": 10000}) + session_end_hooks.append(_new_claude_hook_entry(config.hook_command)) - # PreCompact — same command, shorter timeout (non-blocking, best-effort) pre_compact_hooks = hooks.setdefault("PreCompact", []) if _find_hook_claude(pre_compact_hooks, config.hook_command): already_compact = True else: - pre_compact_hooks.append({"command": config.hook_command, "timeout": 10000}) + pre_compact_hooks.append(_new_claude_hook_entry(config.hook_command)) return already_end and already_compact def _uninstall_claude(settings: dict, config: AgentHookConfig) -> bool: - """Remove the Primer hook from Claude Code settings. Returns True if found.""" + """Remove the Primer hook from Claude Code settings. Returns True if found. + + Handles both the current matcher-wrapped format and the legacy flat format. + Third-party entries are left untouched — including matcher entries whose + `hooks` array is already empty, which would otherwise be collapsed to None. + """ hooks = settings.get("hooks", {}) + command = config.hook_command found = False for event in ("SessionEnd", "PreCompact"): event_hooks = hooks.get(event, []) - if _find_hook_claude(event_hooks, config.hook_command): - found = True - hooks[event] = [ - h - for h in event_hooks - if not ( - (isinstance(h, dict) and h.get("command") == config.hook_command) - or (isinstance(h, str) and h == config.hook_command) - ) - ] + if not _find_hook_claude(event_hooks, command): + continue + found = True + new_list: list = [] + for entry in event_hooks: + # Legacy flat: drop entirely if it matches + if isinstance(entry, str) and entry == command: + continue + if isinstance(entry, dict) and entry.get("command") == command: + continue + # Matcher-wrapped: strip our command only if this entry contains + # it. Third-party entries pass through unchanged — including ones + # that happen to have an empty `hooks` array already. + if isinstance(entry, dict) and isinstance(entry.get("hooks"), list): + inner = entry["hooks"] + if any(isinstance(h, dict) and h.get("command") == command for h in inner): + remaining = [ + h + for h in inner + if not (isinstance(h, dict) and h.get("command") == command) + ] + if remaining: + cleaned = dict(entry) + cleaned["hooks"] = remaining + new_list.append(cleaned) + # else: entry held only our command → drop it + continue + new_list.append(entry) + hooks[event] = new_list return found diff --git a/tests/test_hook_installer.py b/tests/test_hook_installer.py index 1adcae5..b180b7a 100644 --- a/tests/test_hook_installer.py +++ b/tests/test_hook_installer.py @@ -33,14 +33,25 @@ def test_install_creates_hook(tmp_path): assert "installed" in msg.lower() data = json.loads(settings_path.read_text()) + + # SessionEnd: one matcher-wrapped entry whose inner hook runs our command hooks = data["hooks"]["SessionEnd"] assert len(hooks) == 1 - assert hooks[0]["command"] == HOOK_COMMAND + assert hooks[0]["matcher"] == "" + inner = hooks[0]["hooks"] + assert len(inner) == 1 + assert inner[0]["type"] == "command" + assert inner[0]["command"] == HOOK_COMMAND + # Timeout is in seconds per Claude Code schema + assert inner[0]["timeout"] == 30 - # PreCompact should also be registered + # PreCompact mirrors SessionEnd pre_compact = data["hooks"]["PreCompact"] assert len(pre_compact) == 1 - assert pre_compact[0]["command"] == HOOK_COMMAND + assert pre_compact[0]["matcher"] == "" + assert pre_compact[0]["hooks"][0]["type"] == "command" + assert pre_compact[0]["hooks"][0]["command"] == HOOK_COMMAND + assert pre_compact[0]["hooks"][0]["timeout"] == 30 def test_install_idempotent(tmp_path): @@ -112,6 +123,122 @@ def test_status_missing_file(tmp_path): assert not installed +def test_status_detects_legacy_flat_format(tmp_path): + """Earlier versions of the installer wrote a flat `{command, timeout}` entry + directly at the event level. Status must still detect those installs.""" + settings_path = tmp_path / "settings.json" + settings_path.write_text( + json.dumps( + { + "hooks": { + "SessionEnd": [{"command": HOOK_COMMAND, "timeout": 10000}], + "PreCompact": [{"command": HOOK_COMMAND, "timeout": 10000}], + } + } + ) + ) + installed, _msg = status(path=settings_path) + assert installed + + +def test_uninstall_removes_legacy_flat_format(tmp_path): + """Uninstall should clean up hooks written by earlier installer versions.""" + settings_path = tmp_path / "settings.json" + settings_path.write_text( + json.dumps( + { + "hooks": { + "SessionEnd": [{"command": HOOK_COMMAND, "timeout": 10000}], + "PreCompact": [{"command": HOOK_COMMAND, "timeout": 10000}], + } + } + ) + ) + ok, msg = uninstall(path=settings_path) + assert ok + assert "removed" in msg.lower() + + data = json.loads(settings_path.read_text()) + assert data["hooks"]["SessionEnd"] == [] + assert data["hooks"]["PreCompact"] == [] + + +def test_install_migrates_legacy_flat_format(tmp_path): + """If a legacy flat hook is present, install is idempotent (treats it as + already installed) — callers can uninstall + reinstall to migrate.""" + settings_path = tmp_path / "settings.json" + settings_path.write_text( + json.dumps( + { + "hooks": { + "SessionEnd": [{"command": HOOK_COMMAND, "timeout": 10000}], + "PreCompact": [{"command": HOOK_COMMAND, "timeout": 10000}], + } + } + ) + ) + ok, msg = install(path=settings_path) + assert ok + assert "already" in msg.lower() + + +def test_uninstall_preserves_third_party_entry_with_empty_hooks(tmp_path): + """A third-party matcher-wrapped entry whose `hooks` array is empty must + survive uninstall untouched. Only entries that actually contained the + primer command may be stripped or dropped.""" + settings_path = tmp_path / "settings.json" + settings_path.write_text( + json.dumps( + { + "hooks": { + "SessionEnd": [ + {"matcher": "Bash", "hooks": []}, # third-party, already empty + ] + } + } + ) + ) + # Install our hook alongside the third-party one + install(path=settings_path) + data = json.loads(settings_path.read_text()) + assert len(data["hooks"]["SessionEnd"]) == 2 # third-party + primer + + # Uninstall should remove the primer entry but preserve the third-party one + uninstall(path=settings_path) + data = json.loads(settings_path.read_text()) + assert {"matcher": "Bash", "hooks": []} in data["hooks"]["SessionEnd"] + + +def test_uninstall_preserves_third_party_sibling_hooks(tmp_path): + """When our command sits in a matcher entry alongside a third-party hook, + stripping should leave the third-party hook in place.""" + settings_path = tmp_path / "settings.json" + settings_path.write_text( + json.dumps( + { + "hooks": { + "SessionEnd": [ + { + "matcher": "", + "hooks": [ + {"type": "command", "command": HOOK_COMMAND, "timeout": 30}, + {"type": "command", "command": "echo bye"}, + ], + } + ] + } + } + ) + ) + uninstall(path=settings_path) + data = json.loads(settings_path.read_text()) + entries = data["hooks"]["SessionEnd"] + assert len(entries) == 1 + assert entries[0]["matcher"] == "" + assert len(entries[0]["hooks"]) == 1 + assert entries[0]["hooks"][0]["command"] == "echo bye" + + # --------------------------------------------------------------------------- # Gemini CLI tests # ---------------------------------------------------------------------------