From 2fd9b75d32e97bb35307c167456892485021a963 Mon Sep 17 00:00:00 2001 From: Catalin Lupuleti Date: Sat, 25 Jul 2026 17:14:43 +0100 Subject: [PATCH] fix(hooks): repair a stale SessionEnd hook instead of only installing once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hook command embeds an absolute path to the binary, so renaming the project (claude-recall -> code-recall) left existing installs pointing at a command that no longer exists. The migration for exactly this case was already here — LEGACY_COMMAND_NAMES and _hook_mentions_app — but it sat behind `is_first_run`, which is false for anyone who has ever built an index. The rewrite branch was unreachable, so a broken hook stayed broken and failed silently on every session end. Split the two concerns: refresh an existing hook whenever it differs from the desired config, and only append a new one on first run, so a hook the user deleted is not resurrected. Writes happen only on an actual change, so ordinary runs still leave settings.json alone. Also stop destroying settings.json when it cannot be parsed. The read error was swallowed with `pass` and the resulting empty dict written straight back, replacing every unrelated key with just `hooks`. Bail out instead, and write through a temp file and atomic rename so an interrupted write cannot truncate the file. Co-Authored-By: Claude Opus 5 (1M context) --- src/code_recall/cli.py | 172 +++++++++++++++++++++++------------------ tests/test_cli.py | 159 ++++++++++++++++++++++++++++++++++++- 2 files changed, 256 insertions(+), 75 deletions(-) diff --git a/src/code_recall/cli.py b/src/code_recall/cli.py index 4a9f393..a4eeda7 100644 --- a/src/code_recall/cli.py +++ b/src/code_recall/cli.py @@ -4,6 +4,7 @@ import argparse import json +import os import sys from pathlib import Path @@ -131,71 +132,121 @@ def _first_run_setup(args: argparse.Namespace) -> None: verbose=is_first_run and show_output, ) - # Auto-install hooks on first run - if is_first_run and not HOOKS_MARKER.exists(): - _auto_install_hooks() + # Repair a stale hook on every run, but only add a missing one on first run. + # A hook's command breaks whenever the tool is renamed or reinstalled + # somewhere else, and a broken hook cannot fix itself — it never runs — so + # the repair has to ride along on an ordinary invocation. + from code_recall.config import load_config + + if load_config().get("auto_index_hook", True): + status, _ = _sync_index_hook( + install_if_missing=is_first_run and not HOOKS_MARKER.exists() + ) + if show_output and status == "installed": + print( + " Auto-installed SessionEnd hook for live index updates.\n", + file=sys.stderr, + ) + elif show_output and status == "repaired": + print( + " Repaired the SessionEnd hook — it pointed at a command " + "that no longer exists.\n", + file=sys.stderr, + ) -def _auto_install_hooks() -> None: - """Silently install SessionEnd hooks on first run.""" - from code_recall.config import load_config +def _sync_index_hook( + *, install_if_missing: bool, fallback_to_name: bool = False +) -> tuple[str, str]: + """Point the Claude Code SessionEnd hook at the current binary. - if not load_config().get("auto_index_hook", True): - return + An existing hook is refreshed whenever it differs from the desired config, + so a rename or reinstall self-heals. A missing hook is only added when + ``install_if_missing`` is set, so one the user deleted stays deleted. + Returns ``(status, hook_command)`` where status is one of ``skipped``, + ``unchanged``, ``repaired`` or ``installed``. + """ import shutil settings_path = Path.home() / ".claude" / "settings.json" code_recall_bin = shutil.which(COMMAND_NAME) if not code_recall_bin: - return + if not fallback_to_name: + return "skipped", "" + code_recall_bin = COMMAND_NAME hook_command = f"{code_recall_bin} index --quiet" desired_hook = _index_hook_config(hook_command) - settings = {} + settings: dict = {} if settings_path.exists(): try: with open(settings_path) as f: settings = json.load(f) except (json.JSONDecodeError, OSError): - pass + # Never write back settings we failed to read: doing so would + # replace every unrelated key in the file with just our hook. + return "skipped", hook_command + if not isinstance(settings, dict): + return "skipped", hook_command - hooks = settings.get("hooks", {}) - session_end_hooks = hooks.get("SessionEnd", []) + hooks = settings.get("hooks") or {} + session_end_hooks = hooks.get("SessionEnd") or [] - # Don't install if already present for rule in session_end_hooks: for hook in rule.get("hooks", []): - if _hook_mentions_app(hook.get("command", "")): - hook.update(desired_hook) - try: - settings_path.parent.mkdir(parents=True, exist_ok=True) - with open(settings_path, "w") as f: - json.dump(settings, f, indent=2) - HOOKS_MARKER.parent.mkdir(parents=True, exist_ok=True) - HOOKS_MARKER.touch() - except OSError: - pass - return - - new_hook = { - "hooks": [desired_hook] - } - session_end_hooks.append(new_hook) + if not _hook_mentions_app(hook.get("command", "")): + continue + # Subset check, not equality: leave any extra keys the user added + # in place, and stay a no-op once they are satisfied. + if all(hook.get(key) == value for key, value in desired_hook.items()): + return "unchanged", hook_command + hook.update(desired_hook) + hooks["SessionEnd"] = session_end_hooks + settings["hooks"] = hooks + if not _write_settings(settings_path, settings): + return "skipped", hook_command + _touch_hooks_marker() + return "repaired", hook_command + + if not install_if_missing: + return "unchanged", hook_command + + session_end_hooks.append({"hooks": [desired_hook]}) hooks["SessionEnd"] = session_end_hooks settings["hooks"] = hooks + if not _write_settings(settings_path, settings): + return "skipped", hook_command + _touch_hooks_marker() + return "installed", hook_command + + +def _write_settings(settings_path: Path, settings: dict) -> bool: + """Write settings.json atomically. Returns False if it could not be written. + The rename is atomic, so an interrupted write leaves the user's existing + settings intact rather than a half-truncated file. + """ + tmp_path = settings_path.with_name(settings_path.name + ".tmp") try: settings_path.parent.mkdir(parents=True, exist_ok=True) - with open(settings_path, "w") as f: + with open(tmp_path, "w") as f: json.dump(settings, f, indent=2) + os.replace(tmp_path, settings_path) + return True + except OSError: + try: + tmp_path.unlink(missing_ok=True) + except OSError: + pass + return False + + +def _touch_hooks_marker() -> None: + try: HOOKS_MARKER.parent.mkdir(parents=True, exist_ok=True) HOOKS_MARKER.touch() - print( - " Auto-installed SessionEnd hook for live index updates.\n", - file=sys.stderr, - ) except OSError: pass @@ -599,50 +650,23 @@ def _cmd_install_hooks() -> None: import shutil settings_path = Path.home() / ".claude" / "settings.json" - code_recall_bin = shutil.which(COMMAND_NAME) - if not code_recall_bin: + if not shutil.which(COMMAND_NAME): print(f"Warning: '{COMMAND_NAME}' not found in PATH.") - code_recall_bin = COMMAND_NAME - - hook_command = f"{code_recall_bin} index --quiet" - desired_hook = _index_hook_config(hook_command) - settings = {} - if settings_path.exists(): - try: - with open(settings_path) as f: - settings = json.load(f) - except (json.JSONDecodeError, OSError): - pass - - hooks = settings.get("hooks", {}) - session_end_hooks = hooks.get("SessionEnd", []) - - for rule in session_end_hooks: - for hook in rule.get("hooks", []): - if _hook_mentions_app(hook.get("command", "")): - hook.update(desired_hook) - settings_path.parent.mkdir(parents=True, exist_ok=True) - with open(settings_path, "w") as f: - json.dump(settings, f, indent=2) - print("Hook already installed; refreshed config.") - print(f" Command: {desired_hook['command']}") - return - - new_hook = { - "hooks": [desired_hook] - } - session_end_hooks.append(new_hook) - hooks["SessionEnd"] = session_end_hooks - settings["hooks"] = hooks + status, hook_command = _sync_index_hook( + install_if_missing=True, fallback_to_name=True + ) - settings_path.parent.mkdir(parents=True, exist_ok=True) - with open(settings_path, "w") as f: - json.dump(settings, f, indent=2) + if status == "skipped": + print(f"Could not update {settings_path}.") + print(" Check that it is readable and contains valid JSON.") + return - HOOKS_MARKER.parent.mkdir(parents=True, exist_ok=True) - HOOKS_MARKER.touch() + if status in ("unchanged", "repaired"): + print("Hook already installed; refreshed config.") + print(f" Command: {hook_command}") + return print("Installed Claude Code SessionEnd hook!") print(f" Settings: {settings_path}") diff --git a/tests/test_cli.py b/tests/test_cli.py index c43e3d6..67bfcad 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -14,7 +14,8 @@ import pytest -from code_recall.cli import main +from code_recall import cli +from code_recall.cli import _sync_index_hook, main @pytest.fixture(autouse=True) @@ -354,3 +355,159 @@ def test_config_set(self, capsys, isolate_env): def test_config_set_invalid(self, capsys, isolate_env): with pytest.raises(SystemExit): main(["config", "search_mode", "bogus"]) + + +# =========================================================================== +# SessionEnd hook sync +# =========================================================================== + +class TestSyncIndexHook: + BIN = "/opt/tools/bin/code-recall" + + @pytest.fixture + def home(self, tmp_path, monkeypatch): + """Fake $HOME with a .claude dir, plus a resolvable binary.""" + fake_home = tmp_path / "home" + (fake_home / ".claude").mkdir(parents=True) + monkeypatch.setattr(Path, "home", lambda: fake_home) + monkeypatch.setattr("shutil.which", lambda _: self.BIN) + return fake_home + + @staticmethod + def _settings(home): + return home / ".claude" / "settings.json" + + @staticmethod + def _write(path, data): + path.write_text(json.dumps(data)) + + def _hook_of(self, home): + settings = json.loads(self._settings(home).read_text()) + return settings["hooks"]["SessionEnd"][0]["hooks"][0] + + def _legacy_settings(self, home, command): + self._write( + self._settings(home), + { + "model": "opus", + "hooks": { + "SessionEnd": [ + {"hooks": [{"type": "command", "command": command}]} + ] + }, + }, + ) + + @pytest.mark.parametrize("legacy", ["claude-recall", "claude-code-recall"]) + def test_repairs_legacy_command_when_not_first_run(self, home, legacy): + """Regression: a renamed binary left the hook permanently broken. + + The repair must happen on an ordinary run, because a hook pointing at a + missing command never executes and so can never fix itself. + """ + self._legacy_settings(home, f"/old/venv/bin/{legacy} index --quiet") + + status, _ = _sync_index_hook(install_if_missing=False) + + assert status == "repaired" + assert self._hook_of(home)["command"] == f"{self.BIN} index --quiet" + + def test_repair_preserves_unrelated_settings(self, home): + self._legacy_settings(home, "/old/venv/bin/claude-recall index --quiet") + + _sync_index_hook(install_if_missing=False) + + assert json.loads(self._settings(home).read_text())["model"] == "opus" + + def test_repair_applies_full_desired_config(self, home): + self._legacy_settings(home, "/old/venv/bin/claude-recall index --quiet") + + _sync_index_hook(install_if_missing=False) + + hook = self._hook_of(home) + assert hook["timeout"] == 30 + assert hook["async"] is True + + def test_is_idempotent(self, home): + self._legacy_settings(home, "/old/venv/bin/claude-recall index --quiet") + assert _sync_index_hook(install_if_missing=False)[0] == "repaired" + + assert _sync_index_hook(install_if_missing=False)[0] == "unchanged" + + def test_keeps_extra_user_keys(self, home): + self._write( + self._settings(home), + { + "hooks": { + "SessionEnd": [ + { + "hooks": [ + { + "type": "command", + "command": "/old/bin/claude-recall index", + "statusMessage": "indexing", + } + ] + } + ] + } + }, + ) + + _sync_index_hook(install_if_missing=False) + + assert self._hook_of(home)["statusMessage"] == "indexing" + + def test_does_not_reinstall_a_hook_the_user_removed(self, home): + self._write(self._settings(home), {"model": "opus"}) + + status, _ = _sync_index_hook(install_if_missing=False) + + assert status == "unchanged" + assert "hooks" not in json.loads(self._settings(home).read_text()) + + def test_installs_when_requested(self, home): + self._write(self._settings(home), {"model": "opus"}) + + status, _ = _sync_index_hook(install_if_missing=True) + + assert status == "installed" + assert self._hook_of(home)["command"] == f"{self.BIN} index --quiet" + + def test_unreadable_settings_are_left_untouched(self, home): + """A parse failure must not turn into a rewrite of the whole file.""" + self._settings(home).write_text("{ this is not json") + + status, _ = _sync_index_hook(install_if_missing=True) + + assert status == "skipped" + assert self._settings(home).read_text() == "{ this is not json" + + def test_skips_when_binary_missing(self, home, monkeypatch): + monkeypatch.setattr("shutil.which", lambda _: None) + self._write(self._settings(home), {"model": "opus"}) + + status, _ = _sync_index_hook(install_if_missing=True) + + assert status == "skipped" + assert "hooks" not in json.loads(self._settings(home).read_text()) + + @patch("code_recall.cli.search", return_value=[]) + @patch("code_recall.cli.ensure_index") + def test_ordinary_run_repairs_stale_hook( + self, mock_index, mock_search, home, isolate_env + ): + """The established-install path must actually reach the repair. + + This is the real regression: hook sync used to be gated behind + "the index does not exist yet", so anyone past their first run kept a + broken hook forever. Exercised through main() because the bug was in + the gate, not in the sync itself. + """ + isolate_env["db_path"].touch() # an existing install, not a first run + cli.HOOKS_MARKER.touch() # and one that already installed hooks once + self._legacy_settings(home, "/old/venv/bin/claude-recall index --quiet") + + main(["some", "query", "--no-tui"]) + + assert self._hook_of(home)["command"] == f"{self.BIN} index --quiet"