From ace531b17aca17e01f5c992f67b0429c8308501e Mon Sep 17 00:00:00 2001 From: Michael Panchenko Date: Thu, 9 Apr 2026 21:05:30 +0200 Subject: [PATCH] Added serena-hooks cli --- pyproject.toml | 1 + src/serena/cli.py | 29 +-- src/serena/hooks.py | 251 ++++++++++++++++++++++++++ src/serena/util/cli_util.py | 24 +++ test/serena/test_hooks.py | 344 ++++++++++++++++++++++++++++++++++++ 5 files changed, 627 insertions(+), 22 deletions(-) create mode 100644 src/serena/hooks.py create mode 100644 test/serena/test_hooks.py diff --git a/pyproject.toml b/pyproject.toml index 0cc71524..a69e2ed6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -61,6 +61,7 @@ explicit = true [project.scripts] serena = "serena.cli:top_level" +serena-hooks = "serena.hooks:hook_commands" [project.license] text = "MIT" diff --git a/src/serena/cli.py b/src/serena/cli.py index 6e297cac..84f2fcc9 100644 --- a/src/serena/cli.py +++ b/src/serena/cli.py @@ -39,6 +39,7 @@ from serena.constants import ( from serena.mcp import SerenaMCPFactory from serena.project import Project from serena.tools import FindReferencingSymbolsTool, FindSymbolTool, GetSymbolsOverviewTool, SearchForPatternTool, ToolRegistry +from serena.util.cli_util import AutoRegisteringGroup from serena.util.dataclass import get_dataclass_default from serena.util.logging import MemoryLogHandler from solidlsp.ls_config import Language @@ -127,32 +128,16 @@ class ProjectType(click.ParamType): PROJECT_TYPE = ProjectType() -class AutoRegisteringGroup(click.Group): - """ - A click.Group subclass that automatically registers any click.Command - attributes defined on the class into the group. - - After initialization, it inspects its own class for attributes that are - instances of click.Command (typically created via @click.command) and - calls self.add_command(cmd) on each. This lets you define your commands - as static methods on the subclass for IDE-friendly organization without - manual registration. - """ - - def __init__(self, name: str, help: str): - super().__init__(name=name, help=help) - # Scan class attributes for click.Command instances and register them. - for attr in dir(self.__class__): - cmd = getattr(self.__class__, attr) - if isinstance(cmd, click.Command): - self.add_command(cmd) - - class TopLevelCommands(AutoRegisteringGroup): """Root CLI group containing the core Serena commands.""" def __init__(self) -> None: - super().__init__(name="serena", help="Serena CLI commands. You can run ` --help` for more info on each command.") + super().__init__( + name="serena", + help="Main serena CLI commands. " + "Note that you also have access to `serena-hooks` CLI commands which are kept under " + "that separate entrypoint for performance reasons, see `serena-hooks --help`. You can run ` --help` for more info on each command.", + ) @staticmethod @click.command( diff --git a/src/serena/hooks.py b/src/serena/hooks.py new file mode 100644 index 00000000..264e6fe5 --- /dev/null +++ b/src/serena/hooks.py @@ -0,0 +1,251 @@ +import json +import os +import pickle +import shutil +import sys +from abc import ABC, abstractmethod +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from pathlib import Path +from typing import Literal, Self + +import click + +from serena.util.cli_util import AutoRegisteringGroup + +# copied from serena_config.py, we don't want to import anything here to keep the hook commands fast +serena_home_dir = os.getenv("SERENA_HOME", "").strip() or str(Path.home() / ".serena") + + +class HookClient(Enum): + """The client application that triggered the hook.""" + + CLAUDE_CODE = "claude-code" + VSCODE = "vscode" + CODEX = "codex" + + +class Hook(ABC): + def __init__(self, client: HookClient): + raw = sys.stdin.read() + input_data = json.loads(raw) + self._input_data = input_data + self._client = client + + session_id = input_data.get("session_id") or input_data.get("sessionId") + if not session_id: + raise ValueError("Session ID is required in the hook input data") + self._session_id = str(session_id) + self.session_persistence_dir = os.path.join(serena_home_dir, "hook_data", self._session_id) + # tool input has a timestamp but using now is enough + self.triggered_at_timestamp = datetime.now() + + @abstractmethod + def execute(self) -> None: + pass + + +class PreToolUseRemindAboutSerenaHook(Hook): + """Pre-tool-use hook that nudges the agent toward Serena's symbolic tools. + + Tracks consecutive uses of grep and read-file tools via a persisted + :class:`ToolUseCounter`. When the number of recent calls reaches the + configured threshold, a deny response is emitted with a reminder to + use symbolic alternatives. The counter resets whenever a Serena tool + is invoked or a deny is emitted, so the agent is only reminded after + sustained non-symbolic tool use without intermittent Serena calls. + """ + + @dataclass + class OutputData: + permission_decision: Literal["deny", "allow"] + permission_decision_reason: str + additional_context: str + + def to_json_string(self) -> str: + hook_output = { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": self.permission_decision, + "permissionDecisionReason": self.permission_decision_reason, + "additionalContext": self.additional_context, + } + } + return json.dumps(hook_output) + + @dataclass + class ToolUseCounter: + _FILE_NAME = "tool_use_counter.pkl" + _GREP_USES_THRESHOLD = 3 + _READ_FILE_USES_THRESHOLD = 3 + + _PERIOD_TO_RESET_COUNTERS_SECONDS = 10 + + n_recent_read_file_uses: int = 0 + n_recent_grep_uses: int = 0 + last_grep_use_timestamp: datetime | None = None + last_read_file_use_timestamp: datetime | None = None + + def too_many_recent_reads(self) -> bool: + return self.n_recent_read_file_uses >= self._READ_FILE_USES_THRESHOLD + + def too_many_recent_greps(self) -> bool: + return self.n_recent_grep_uses >= self._GREP_USES_THRESHOLD + + @classmethod + def _get_persistence_path(cls, hook: Hook) -> Path: + return Path(hook.session_persistence_dir) / cls._FILE_NAME + + @classmethod + def load(cls, hook: Hook) -> Self: + path = cls._get_persistence_path(hook) + try: + with open(path, "rb") as f: + return pickle.load(f) + except Exception: + return cls() + + def save(self, hook: Hook) -> None: + path = self._get_persistence_path(hook) + try: + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + pickle.dump(self, f) + except Exception: + pass + + def update(self, hook: "PreToolUseRemindAboutSerenaHook") -> None: + if hook.is_serena_tool(): + self.reset() + return + + now = hook.triggered_at_timestamp + period = self._PERIOD_TO_RESET_COUNTERS_SECONDS + + # update grep counter + if hook.is_grep_tool(): + if self.last_grep_use_timestamp is not None and (now - self.last_grep_use_timestamp).total_seconds() <= period: + self.n_recent_grep_uses += 1 + else: + self.n_recent_grep_uses = 1 + self.last_grep_use_timestamp = now + + # update read file counter + if hook.is_read_file_tool(): + if self.last_read_file_use_timestamp is not None and (now - self.last_read_file_use_timestamp).total_seconds() <= period: + self.n_recent_read_file_uses += 1 + else: + self.n_recent_read_file_uses = 1 + self.last_read_file_use_timestamp = now + + def reset(self) -> None: + self.n_recent_read_file_uses = 0 + self.n_recent_grep_uses = 0 + self.last_grep_use_timestamp = None + self.last_read_file_use_timestamp = None + + def __init__(self, client: HookClient): + super().__init__(client) + _tool_name = self._input_data.get("tool_name") or self._input_data.get("toolName", "") or "" + _tool_name = str(_tool_name).lower().strip() + if not _tool_name: + raise ValueError("Tool name is required in the hook input data") + self._tool_name = _tool_name + self._tool_input: dict | None = self._input_data.get("tool_input") or self._input_data.get("toolInput") + self._tool_call_counter = self.ToolUseCounter.load(self) + + def is_grep_tool(self) -> bool: + if self._client == HookClient.CLAUDE_CODE: + return self._tool_name == "grep" + return "grep" in self._tool_name + + def is_read_file_tool(self) -> bool: + if self._client == HookClient.CLAUDE_CODE: + return self._tool_name == "read" + return "read" in self._tool_name and "file" in self._tool_name + + def is_serena_tool(self) -> bool: + return "serena" in self._tool_name + + def execute(self) -> None: + self._tool_call_counter.update(self) + output_data = None + if self._tool_call_counter.too_many_recent_greps(): + output_data = self.OutputData( + permission_decision="deny", + permission_decision_reason="Not using symbolic tools. You can continue using grep now if needed, the counter was reset.", + additional_context="You were using many grep calls recently. Consider using Serena's symbolic search tools instead.", + ) + if self._tool_call_counter.too_many_recent_reads(): + output_data = self.OutputData( + permission_decision="deny", + permission_decision_reason="Not using symbolic tools. You can continue using read now if needed, the counter was reset.", + additional_context="You were using many read file calls recently. Consider using Serena's symbolic overview and search instead.", + ) + if output_data is not None: + self._tool_call_counter.reset() + click.echo(output_data.to_json_string()) + self._tool_call_counter.save(self) + + +class SessionStartActivateProjectHook(Hook): + def execute(self) -> None: + message = ( + "**IMPORTANT**: Activate the current working directory as project using Serena's tools. " + "If you haven't read Serena Instructions Manual yet, do so. " + "Follow this instruction before doing anything else." + ) + result = { + "hookSpecificOutput": { + "hookEventName": "SessionStart", + "additionalContext": message, + } + } + click.echo(json.dumps(result)) + + +class SessionEndCleanupHook(Hook): + def execute(self) -> None: + shutil.rmtree(self.session_persistence_dir, ignore_errors=True) + + +_client_option = click.option( + "--client", + type=click.Choice([e.value for e in HookClient], case_sensitive=False), + default=HookClient.CLAUDE_CODE.value, + show_default=True, + help="The client application that triggered the hook.", +) + + +class HookCommands(AutoRegisteringGroup): + def __init__(self) -> None: + super().__init__(name="serena-hook", help="Commands that send reminders to agents when appropriate, to be used in hooks.") + + @staticmethod + @click.command( + "activate", + help="Set this as hook at session startup to prompt the agent to activate the project at the start of the session and read Serena's instructions", + ) + @_client_option + def activate(client: str) -> None: + SessionStartActivateProjectHook(HookClient(client)).execute() + + @staticmethod + @click.command("cleanup", help="Set this as hook at session end all hook data for the current session") + @_client_option + def cleanup(client: str) -> None: + SessionEndCleanupHook(HookClient(client)).execute() + + @staticmethod + @click.command( + "remind", + help="Set this as hook at PreToolUse to remind the agent to use Serena's tools instead of overrelying on read_file and grep", + ) + @_client_option + def remind(client: str) -> None: + PreToolUseRemindAboutSerenaHook(HookClient(client)).execute() + + +hook_commands = HookCommands() diff --git a/src/serena/util/cli_util.py b/src/serena/util/cli_util.py index 66f725bb..807ae7c0 100644 --- a/src/serena/util/cli_util.py +++ b/src/serena/util/cli_util.py @@ -1,3 +1,6 @@ +import click + + def ask_yes_no(question: str, default: bool | None = None) -> bool: default_prompt = "Y/n" if default else "y/N" @@ -10,3 +13,24 @@ def ask_yes_no(question: str, default: bool | None = None) -> bool: if answer in ("n", "no"): return False print("Please answer yes/y or no/n.") + + +class AutoRegisteringGroup(click.Group): + """ + A click.Group subclass that automatically registers any click.Command + attributes defined on the class into the group. + + After initialization, it inspects its own class for attributes that are + instances of click.Command (typically created via @click.command) and + calls self.add_command(cmd) on each. This lets you define your commands + as static methods on the subclass for IDE-friendly organization without + manual registration. + """ + + def __init__(self, name: str, help: str): + super().__init__(name=name, help=help) + # Scan class attributes for click.Command instances and register them. + for attr in dir(self.__class__): + cmd = getattr(self.__class__, attr) + if isinstance(cmd, click.Command): + self.add_command(cmd) diff --git a/test/serena/test_hooks.py b/test/serena/test_hooks.py new file mode 100644 index 00000000..e8291bc7 --- /dev/null +++ b/test/serena/test_hooks.py @@ -0,0 +1,344 @@ +import json +import pickle +from datetime import datetime, timedelta +from io import StringIO +from pathlib import Path +from unittest.mock import patch + +import pytest +from click.testing import CliRunner + +from serena.hooks import ( + HookClient, + PreToolUseRemindAboutSerenaHook, + SessionEndCleanupHook, + SessionStartActivateProjectHook, + hook_commands, +) + +ToolUseCounter = PreToolUseRemindAboutSerenaHook.ToolUseCounter + + +def _make_stdin(data: dict) -> StringIO: + return StringIO(json.dumps(data)) + + +def _base_input(tool_name: str = "grep_search", session_id: str = "test-session-123") -> dict: + return { + "session_id": session_id, + "tool_name": tool_name, + "tool_input": {"query": "foo"}, + } + + +class TestHookClientDetection: + """Tests for the --client option propagation.""" + + def test_claude_code_client(self, tmp_path: Path): + stdin_data = _base_input() + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE) + assert hook._client == HookClient.CLAUDE_CODE + + def test_vscode_client(self, tmp_path: Path): + stdin_data = _base_input() + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSerenaHook(HookClient.VSCODE) + assert hook._client == HookClient.VSCODE + + +class TestPreToolUseRemindAboutSerenaHook: + """Tests for the PreToolUse hook that nudges the agent toward symbolic tools.""" + + def test_missing_tool_name_raises(self, tmp_path: Path): + stdin_data = {"session_id": "s1"} + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + with pytest.raises(ValueError, match="Tool name is required"): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE) + + def test_missing_session_id_raises(self, tmp_path: Path): + stdin_data = {"tool_name": "grep"} + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + with pytest.raises(ValueError, match="Session ID is required"): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE) + + def test_grep_tool_detection(self, tmp_path: Path): + for name, expected in [("grep_search", True), ("mcp_grep", True), ("read_file", False), ("serena_find", False)]: + with patch("sys.stdin", _make_stdin(_base_input(tool_name=name))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE) + assert hook.is_grep_tool() == expected, f"is_grep_tool() wrong for {name}" + + def test_read_file_tool_detection(self, tmp_path: Path): + for name, expected in [("read_file", True), ("readFile", True), ("grep_search", False), ("file_writer", False)]: + with patch("sys.stdin", _make_stdin(_base_input(tool_name=name))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE) + assert hook.is_read_file_tool() == expected, f"is_read_file_tool() wrong for {name}" + + def test_serena_tool_detection(self, tmp_path: Path): + for name, expected in [("mcp_serena_find_symbol", True), ("serena_overview", True), ("grep_search", False)]: + with patch("sys.stdin", _make_stdin(_base_input(tool_name=name))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE) + assert hook.is_serena_tool() == expected, f"is_serena_tool() wrong for {name}" + + def test_no_output_below_threshold(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Below the threshold, the hook should produce no output (tool is allowed).""" + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD - 1): + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + assert capsys.readouterr().out == "" + + def test_deny_output_after_threshold_greps_claude_code(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """After reaching the grep threshold, the hook should output a deny.""" + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + output = capsys.readouterr().out.strip() + result = json.loads(output) + hook_output = result["hookSpecificOutput"] + assert hook_output["permissionDecision"] == "deny" + assert "grep" in hook_output["additionalContext"].lower() + + def test_deny_output_after_threshold_greps_vscode(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """After reaching the grep threshold, the hook should output a deny for VS Code.""" + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.VSCODE).execute() + + output = capsys.readouterr().out.strip() + result = json.loads(output) + hook_output = result["hookSpecificOutput"] + assert hook_output["permissionDecision"] == "deny" + assert "grep" in hook_output["additionalContext"].lower() + + def test_deny_output_after_threshold_reads(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """After reaching the read file threshold, the hook should output a deny.""" + for _ in range(ToolUseCounter._READ_FILE_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("read_file"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + output = capsys.readouterr().out.strip() + result = json.loads(output) + hook_output = result["hookSpecificOutput"] + assert hook_output["permissionDecision"] == "deny" + assert "read file" in hook_output["additionalContext"].lower() + + def test_serena_tool_resets_counters(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Using a Serena tool should reset counters, so the threshold is not reached.""" + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD - 1): + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + with patch("sys.stdin", _make_stdin(_base_input("mcp_serena_find_symbol"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + assert capsys.readouterr().out == "" + + def test_counter_resets_after_deny(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """After a deny is emitted, the counter is reset so the next burst starts fresh.""" + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + capsys.readouterr() + + with patch("sys.stdin", _make_stdin(_base_input("grep_search"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + assert capsys.readouterr().out == "" + + +class TestToolUseCounter: + """Tests for the time-windowed tool-use counter logic.""" + + def test_update_increments_grep_within_period(self): + counter = ToolUseCounter() + now = datetime.now() + counter.last_grep_use_timestamp = now - timedelta(seconds=1) + counter.n_recent_grep_uses = 1 + + hook = self._make_hook_stub("grep_search", now) + counter.update(hook) + + assert counter.n_recent_grep_uses == 2 + assert counter.last_grep_use_timestamp == now + + def test_update_resets_grep_outside_period(self): + counter = ToolUseCounter() + now = datetime.now() + counter.last_grep_use_timestamp = now - timedelta(seconds=ToolUseCounter._PERIOD_TO_RESET_COUNTERS_SECONDS + 1) + counter.n_recent_grep_uses = 2 + + hook = self._make_hook_stub("grep_search", now) + counter.update(hook) + + assert counter.n_recent_grep_uses == 1 + assert counter.last_grep_use_timestamp == now + + def test_update_increments_read_file_within_period(self): + counter = ToolUseCounter() + now = datetime.now() + counter.last_read_file_use_timestamp = now - timedelta(seconds=1) + counter.n_recent_read_file_uses = 1 + + hook = self._make_hook_stub("read_file", now) + counter.update(hook) + + assert counter.n_recent_read_file_uses == 2 + + def test_update_resets_read_file_outside_period(self): + counter = ToolUseCounter() + now = datetime.now() + counter.last_read_file_use_timestamp = now - timedelta(seconds=ToolUseCounter._PERIOD_TO_RESET_COUNTERS_SECONDS + 1) + counter.n_recent_read_file_uses = 2 + + hook = self._make_hook_stub("read_file", now) + counter.update(hook) + + assert counter.n_recent_read_file_uses == 1 + + def test_serena_tool_resets_all_counters(self): + counter = ToolUseCounter( + n_recent_grep_uses=2, + n_recent_read_file_uses=2, + last_grep_use_timestamp=datetime.now(), + last_read_file_use_timestamp=datetime.now(), + ) + hook = self._make_hook_stub("mcp_serena_overview", datetime.now()) + counter.update(hook) + + assert counter.n_recent_grep_uses == 0 + assert counter.n_recent_read_file_uses == 0 + assert counter.last_grep_use_timestamp is None + assert counter.last_read_file_use_timestamp is None + + def test_non_matching_tool_leaves_counters_unchanged(self): + counter = ToolUseCounter(n_recent_grep_uses=1, n_recent_read_file_uses=1) + hook = self._make_hook_stub("write_file", datetime.now()) + counter.update(hook) + + assert counter.n_recent_grep_uses == 1 + assert counter.n_recent_read_file_uses == 1 + + def test_persistence_round_trip(self, tmp_path: Path): + counter = ToolUseCounter(n_recent_grep_uses=2, n_recent_read_file_uses=1) + + hook_stub = type("HookStub", (), {"session_persistence_dir": str(tmp_path)})() + counter.save(hook_stub) # type: ignore[arg-type] + loaded = ToolUseCounter.load(hook_stub) # type: ignore[arg-type] + + assert loaded.n_recent_grep_uses == 2 + assert loaded.n_recent_read_file_uses == 1 + + def test_load_returns_fresh_counter_on_missing_file(self, tmp_path: Path): + hook_stub = type("HookStub", (), {"session_persistence_dir": str(tmp_path / "nonexistent")})() + loaded = ToolUseCounter.load(hook_stub) # type: ignore[arg-type] + assert loaded == ToolUseCounter() + + def test_load_returns_fresh_counter_on_corrupt_file(self, tmp_path: Path): + hook_stub = type("HookStub", (), {"session_persistence_dir": str(tmp_path)})() + path = tmp_path / ToolUseCounter._FILE_NAME + path.write_bytes(b"not a pickle") + loaded = ToolUseCounter.load(hook_stub) # type: ignore[arg-type] + assert loaded == ToolUseCounter() + + @staticmethod + def _make_hook_stub(tool_name: str, timestamp: datetime) -> PreToolUseRemindAboutSerenaHook: + """Create a minimal stub that satisfies ToolUseCounter.update without reading stdin.""" + stub = object.__new__(PreToolUseRemindAboutSerenaHook) + stub._tool_name = tool_name.lower() + stub.triggered_at_timestamp = timestamp + return stub + + +class TestSessionStartActivateProjectHook: + def test_outputs_activation_message(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + stdin_data = {"session_id": "s1"} + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + SessionStartActivateProjectHook(HookClient.VSCODE).execute() + + output = capsys.readouterr().out.strip() + result = json.loads(output) + context = result["hookSpecificOutput"]["additionalContext"] + assert "Activate" in context + assert "Serena Instructions Manual" in context + + +class TestSessionEndCleanupHook: + def test_removes_session_dir(self, tmp_path: Path): + session_dir = tmp_path / "hook_data" / "cleanup-session" + session_dir.mkdir(parents=True) + # place a file inside to verify recursive removal + (session_dir / "tool_use_counter.pkl").write_bytes(pickle.dumps(ToolUseCounter())) + + stdin_data = {"session_id": "cleanup-session"} + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + SessionEndCleanupHook(HookClient.CLAUDE_CODE).execute() + + assert not session_dir.exists() + + def test_cleanup_is_idempotent(self, tmp_path: Path): + """Cleaning up a non-existent session directory should not raise.""" + stdin_data = {"session_id": "nonexistent-session"} + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + SessionEndCleanupHook(HookClient.CLAUDE_CODE).execute() + + +class TestHookCli: + """Tests for the Click CLI entry point (serena-hooks).""" + + def test_activate_command(self, tmp_path: Path): + stdin_json = json.dumps({"session_id": "cli-test"}) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["activate", "--client", "vscode"], input=stdin_json) + assert result.exit_code == 0 + output = json.loads(result.output) + assert "Activate" in output["hookSpecificOutput"]["additionalContext"] + + def test_cleanup_command(self, tmp_path: Path): + session_dir = tmp_path / "hook_data" / "cli-cleanup" + session_dir.mkdir(parents=True) + (session_dir / "somefile").write_text("data") + + stdin_json = json.dumps({"session_id": "cli-cleanup"}) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["cleanup", "--client", "claude-code"], input=stdin_json) + assert result.exit_code == 0 + assert not session_dir.exists() + + def test_remind_command(self, tmp_path: Path): + """Invoke the remind command enough times to trigger a deny.""" + runner = CliRunner() + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + stdin_json = json.dumps({"session_id": "cli-remind", "tool_name": "grep_search", "tool_input": {}}) + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["remind", "--client", "claude-code"], input=stdin_json) + assert result.exit_code == 0 + + output = json.loads(result.output) + assert output["hookSpecificOutput"]["permissionDecision"] == "deny" + + def test_client_default_is_claude_code(self, tmp_path: Path): + """When --client is omitted, it defaults to claude-code.""" + stdin_json = json.dumps({"session_id": "cli-default"}) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["activate"], input=stdin_json) + assert result.exit_code == 0 + + def test_invalid_client_rejected(self, tmp_path: Path): + stdin_json = json.dumps({"session_id": "s1"}) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["activate", "--client", "invalid"], input=stdin_json) + assert result.exit_code != 0 + + def test_invalid_stdin_exits_nonzero(self, tmp_path: Path): + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["activate", "--client", "claude-code"], input="not json") + assert result.exit_code != 0