mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-13 08:15:59 +00:00
Added serena-hooks cli
This commit is contained in:
1 parent
aafad5c2b0
commit
ace531b17a
5 files changed
+627
-22
No files matched your search
@@ -61,6 +61,7 @@ explicit = true
|
||||
|
||||
[project.scripts]
|
||||
serena = "serena.cli:top_level"
|
||||
serena-hooks = "serena.hooks:hook_commands"
|
||||
|
||||
[project.license]
|
||||
text = "MIT"
|
||||
|
||||
+7
-22
@@ -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 `<command> --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 `<command> --help` for more info on each command.",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
@click.command(
|
||||
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
@@ -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
|
||||
Reference in new issue
Block a user