Skip to content
Merged
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
109 changes: 89 additions & 20 deletions src/primer/hook/installer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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
Comment thread
cursor[bot] marked this conversation as resolved.
# 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

Expand Down
133 changes: 130 additions & 3 deletions tests/test_hook_installer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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
# ---------------------------------------------------------------------------
Expand Down
Loading