diff --git a/src/serena/hooks.py b/src/serena/hooks.py index 8988d8a8..4a04e9f4 100644 --- a/src/serena/hooks.py +++ b/src/serena/hooks.py @@ -87,9 +87,27 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): 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. + use symbolic alternatives. + + The counter for a given tool type is reset whenever + + * a Serena tool is invoked (both counters are reset), + * a deny is emitted (the acting counter is reset so the next retry starts fresh), + * or the configured reset period elapses *between two consecutive calls of that + same tool type* — i.e. the period gates the gap between successive calls, not + an absolute sliding window. Three grep calls at t=0, t=9, t=18 therefore count + as a burst of three, even though the total span (18s) exceeds the 10s grep + period; only an individual pair that is more than 10s apart resets the counter. + + Non-tracked tools (Edit, Write, Bash, etc.) are deliberately neutral: they neither + increment nor reset counters, so they also do not mask bursts by pushing the last + timestamp forward. + + Deny emissions are additionally rate-limited to at most one per + :attr:`ToolUseCounter._MIN_DENY_INTERVAL_SECONDS` (one minute): once a deny has + been emitted, subsequent bursts detected within that window still reset the + counters but do not produce further nudges, so the agent is not swamped with + reminders during a sustained non-symbolic-tool burst. """ @dataclass @@ -97,14 +115,28 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): _FILE_NAME = "tool_use_counter.pkl" _GREP_USES_THRESHOLD = 3 _READ_FILE_USES_THRESHOLD = 3 + # threshold for the combined "non-symbolic" counter that catches mixed sequences of grep+read + _NON_SYMBOLIC_USES_THRESHOLD = 4 - _GREP_RESET_PERIOD_SECONDS = 10 - _READ_FILE_RESET_PERIOD_SECONDS = 20 + # The following periods are set to essentially infinity since we neglect the per-tool reset periods for them + _READ_FILE_RESET_PERIOD_SECONDS = 1000 + _GREP_RESET_PERIOD_SECONDS = 1000 + # reset period for the combined counter + _NON_SYMBOLIC_RESET_PERIOD_SECONDS = 2000 + + # minimum seconds between two emitted denies; rate-limits reminder nudges so + # a single sustained burst triggers at most one deny per minute + _MIN_DENY_INTERVAL_SECONDS = 120 n_recent_read_file_uses: int = 0 n_recent_grep_uses: int = 0 + n_recent_non_symbolic_uses: int = 0 last_grep_use_timestamp: datetime | None = None last_read_file_use_timestamp: datetime | None = None + last_non_symbolic_use_timestamp: datetime | None = None + # timestamp of the most recently emitted deny; deliberately not cleared by + # :meth:`reset` so the rate limit survives counter resets (e.g. Serena tool use) + last_deny_timestamp: datetime | None = None def too_many_recent_reads(self) -> bool: return self.n_recent_read_file_uses >= self._READ_FILE_USES_THRESHOLD @@ -112,6 +144,21 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): def too_many_recent_greps(self) -> bool: return self.n_recent_grep_uses >= self._GREP_USES_THRESHOLD + def too_many_recent_non_symbolic(self) -> bool: + return self.n_recent_non_symbolic_uses >= self._NON_SYMBOLIC_USES_THRESHOLD + + def can_emit_deny(self, now: datetime) -> bool: + """:return: whether enough time has elapsed since the last emitted deny + to allow another one. A value of ``True`` is also returned when no + deny has been emitted yet in this session. The rate limit prevents + the agent from being nudged more than once per + :attr:`_MIN_DENY_INTERVAL_SECONDS`, even if further bursts are + detected in the meantime. + """ + if self.last_deny_timestamp is None: + return True + return (now - self.last_deny_timestamp).total_seconds() >= self._MIN_DENY_INTERVAL_SECONDS + @classmethod def _get_persistence_path(cls, hook: Hook) -> Path: return Path(hook.session_persistence_dir) / cls._FILE_NAME @@ -140,9 +187,11 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): return now = hook.triggered_at_timestamp + is_grep = hook.is_grep_tool() + is_read = hook.is_read_file_tool() # update grep counter - if hook.is_grep_tool(): + if is_grep: grep_period = self._GREP_RESET_PERIOD_SECONDS if self.last_grep_use_timestamp is not None and (now - self.last_grep_use_timestamp).total_seconds() <= grep_period: self.n_recent_grep_uses += 1 @@ -151,7 +200,7 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): self.last_grep_use_timestamp = now # update read file counter - if hook.is_read_file_tool(): + if is_read: read_period = self._READ_FILE_RESET_PERIOD_SECONDS if ( self.last_read_file_use_timestamp is not None @@ -162,11 +211,34 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): self.n_recent_read_file_uses = 1 self.last_read_file_use_timestamp = now + # update combined non-symbolic counter — catches mixed bursts (e.g. + # alternating grep/read) that neither per-tool counter would trip + if is_grep or is_read: + combined_period = self._NON_SYMBOLIC_RESET_PERIOD_SECONDS + if ( + self.last_non_symbolic_use_timestamp is not None + and (now - self.last_non_symbolic_use_timestamp).total_seconds() <= combined_period + ): + self.n_recent_non_symbolic_uses += 1 + else: + self.n_recent_non_symbolic_uses = 1 + self.last_non_symbolic_use_timestamp = now + def reset(self) -> None: self.n_recent_read_file_uses = 0 self.n_recent_grep_uses = 0 + self.n_recent_non_symbolic_uses = 0 self.last_grep_use_timestamp = None self.last_read_file_use_timestamp = None + self.last_non_symbolic_use_timestamp = None + + #: substrings that, combined with ``"file"`` in the tool name, identify a + #: read-file tool for non-Claude-Code clients (whose tool names vary across + #: editors/agents). Lowercase because :attr:`_tool_name` is lowercased on + #: ingest. Conservative on purpose: only verbs that strongly imply *reading* + #: a file, never *modifying* one — so ``view_file``/``open_file``/``show_file`` + #: are caught alongside ``read_file``, while ``write_file``/``edit_file`` are not. + _READ_FILE_VERB_SUBSTRINGS: tuple[str, ...] = ("read", "view", "open", "show") def __init__(self, client: HookClient): super().__init__(client) @@ -180,28 +252,80 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook): 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 + name = self._tool_name + if "file" not in name: + return False + return any(verb in name for verb in self._READ_FILE_VERB_SUBSTRINGS) 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="Too many consecutive grep calls without using symbolic tools.", - additional_context="You were using many grep calls recently. Consider using Serena's symbolic search tools instead. You can continue using grep now if needed, the counter was reset.", - ) - if self._tool_call_counter.too_many_recent_reads(): - output_data = self.OutputData( - permission_decision="deny", - permission_decision_reason="Too many consecutive read calls without using symbolic tools.", - additional_context="You were using many read file calls recently. Consider using Serena's symbolic overview and search instead. You can continue using read now if needed, the counter was reset", - ) + + # pick the deny that matches the current call first, so the emitted message lines up + # with the tool the agent just invoked; fall back to the other per-tool counter only + # if the current call did not itself trip a threshold (e.g. stale state loaded from + # pickle). The combined non-symbolic deny is checked last — it only fires when neither + # per-tool counter tripped, which is exactly the mixed-burst case (alternating grep/read). + too_many_greps = self._tool_call_counter.too_many_recent_greps() + too_many_reads = self._tool_call_counter.too_many_recent_reads() + too_many_non_symbolic = self._tool_call_counter.too_many_recent_non_symbolic() + + output_data: PreToolUseHook.OutputData | None = None + if self.is_grep_tool() and too_many_greps: + output_data = self._build_grep_deny() + elif self.is_read_file_tool() and too_many_reads: + output_data = self._build_read_deny() + elif too_many_greps: + output_data = self._build_grep_deny() + elif too_many_reads: + output_data = self._build_read_deny() + elif too_many_non_symbolic: + output_data = self._build_non_symbolic_deny() + if output_data is not None: + # reset first so burst detection starts fresh regardless of whether the + # deny is actually emitted; the rate limit only gates *emission*, not + # detection self._tool_call_counter.reset() - click.echo(output_data.to_json_string()) + if self._tool_call_counter.can_emit_deny(self.triggered_at_timestamp): + self._tool_call_counter.last_deny_timestamp = self.triggered_at_timestamp + click.echo(output_data.to_json_string()) self._tool_call_counter.save(self) + def _build_grep_deny(self) -> "PreToolUseHook.OutputData": + return self.OutputData( + permission_decision="deny", + permission_decision_reason="Too many consecutive grep calls without 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 " + "mcp tools instead for more code-centric search. You can continue using grep now if needed, the counter was reset." + ), + ) + + def _build_read_deny(self) -> "PreToolUseHook.OutputData": + return self.OutputData( + permission_decision="deny", + permission_decision_reason="Too many consecutive read calls without 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 " + "mcp tools instead for more targeted reads. You can continue using read now if needed, the counter was reset." + ), + ) + + def _build_non_symbolic_deny(self) -> "PreToolUseHook.OutputData": + return self.OutputData( + permission_decision="deny", + permission_decision_reason="Too many consecutive non-symbolic tool calls (mixed grep and read). " + "You can continue using these tools now if needed, the counter was reset.", + additional_context=( + "You were alternating between grep and read file calls recently without using " + "Serena's symbolic mcp tools. Consider using symbolic search and targeted symbol " + "reads instead for more code-centric exploration. You can continue using these tools " + "now if needed, the counter was reset." + ), + ) + class SessionStartActivateProjectHook(Hook): def execute(self) -> None: diff --git a/test/serena/test_hooks.py b/test/serena/test_hooks.py index 3023261f..a98714fd 100644 --- a/test/serena/test_hooks.py +++ b/test/serena/test_hooks.py @@ -85,8 +85,22 @@ class TestPreToolUseRemindAboutSerenaHook: assert hook.is_read_file_tool() == expected, f"is_read_file_tool() wrong for {name} (claude-code)" def test_read_file_tool_detection_non_claude_code(self, tmp_path: Path): - """Non-Claude-Code clients fall back to substring matching requiring both ``read`` and ``file``.""" - for name, expected in [("read_file", True), ("readFile", True), ("grep_search", False), ("file_writer", False)]: + """Non-Claude-Code clients accept any read-style verb (``read``/``view``/``open``/``show``) combined with ``file``.""" + cases = [ + # canonical names + ("read_file", True), + ("readFile", True), + # alternative read verbs used by other agents/editors + ("view_file", True), + ("open_file", True), + ("show_file", True), + # negatives: no "file", or "file" without a read verb, or modifying verbs + ("grep_search", False), + ("file_writer", False), + ("write_file", False), + ("edit_file", False), + ] + for name, expected in cases: with patch("sys.stdin", _make_stdin(_base_input(tool_name=name))), patch("serena.hooks.serena_home_dir", str(tmp_path)): hook = PreToolUseRemindAboutSerenaHook(HookClient.VSCODE) assert hook.is_read_file_tool() == expected, f"is_read_file_tool() wrong for {name} (vscode)" @@ -166,6 +180,79 @@ class TestPreToolUseRemindAboutSerenaHook: assert capsys.readouterr().out == "" + def test_rate_limit_suppresses_second_deny_within_interval(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """A second burst detected within the minimum deny interval must not emit a deny.""" + # first burst: should emit a deny + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + first_output = capsys.readouterr().out.strip() + assert first_output, "first burst should have emitted a deny" + + # second burst immediately after: within the 1-minute window, must be suppressed + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + assert capsys.readouterr().out == "" + + def test_rate_limit_allows_deny_after_interval(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Once the minimum deny interval has elapsed, a fresh burst emits a deny again.""" + # first burst: emits a deny + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + capsys.readouterr() + + # backdate the persisted last_deny_timestamp so the rate-limit window has expired + stub_for_path = object.__new__(PreToolUseRemindAboutSerenaHook) + stub_for_path.session_persistence_dir = str(tmp_path / "hook_data" / _base_input()["session_id"]) + counter = ToolUseCounter.load(stub_for_path) + assert counter.last_deny_timestamp is not None + counter.last_deny_timestamp -= timedelta(seconds=ToolUseCounter._MIN_DENY_INTERVAL_SECONDS + 1) + counter.save(stub_for_path) + + # second burst should now emit a deny again + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(_base_input("grep"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + second_output = capsys.readouterr().out.strip() + assert second_output, "after the rate-limit window elapsed, a new burst should emit a deny" + result = json.loads(second_output) + assert result["hookSpecificOutput"]["permissionDecision"] == "deny" + + def test_non_symbolic_deny_emitted_when_combined_threshold_tripped(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Pre-populated state trips only the combined non-symbolic counter (not per-tool ones) + so execute() must fall through to the _build_non_symbolic_deny branch. + """ + # pre-populate the pickle so only the combined counter is over threshold + session_dir = tmp_path / "hook_data" / _base_input()["session_id"] + session_dir.mkdir(parents=True) + counter = ToolUseCounter( + n_recent_grep_uses=ToolUseCounter._GREP_USES_THRESHOLD - 1, + n_recent_read_file_uses=ToolUseCounter._READ_FILE_USES_THRESHOLD - 1, + n_recent_non_symbolic_uses=ToolUseCounter._NON_SYMBOLIC_USES_THRESHOLD, + last_grep_use_timestamp=datetime.now(), + last_read_file_use_timestamp=datetime.now(), + last_non_symbolic_use_timestamp=datetime.now(), + ) + stub_for_path = object.__new__(PreToolUseRemindAboutSerenaHook) + stub_for_path.session_persistence_dir = str(session_dir) + counter.save(stub_for_path) + + # invoke execute with a neutral (non-grep, non-read, non-serena) tool so that + # update() leaves the counters untouched and the fall-through branch is taken + with patch("sys.stdin", _make_stdin(_base_input("Edit"))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSerenaHook(HookClient.CLAUDE_CODE).execute() + + output = capsys.readouterr().out.strip() + assert output, "expected a non-symbolic deny to be emitted" + result = json.loads(output) + hook_output = result["hookSpecificOutput"] + assert hook_output["permissionDecision"] == "deny" + assert "symbolic" in hook_output["additionalContext"].lower() + class TestToolUseCounter: """Tests for the time-windowed tool-use counter logic.""" @@ -261,6 +348,40 @@ class TestToolUseCounter: loaded = ToolUseCounter.load(hook_stub) # type: ignore[arg-type] assert loaded == ToolUseCounter() + def test_can_emit_deny_respects_min_interval(self): + """:meth:`can_emit_deny` returns False within the minimum interval, True outside it.""" + counter = ToolUseCounter() + base = datetime.now() + + # no prior deny → always allowed + assert counter.can_emit_deny(base) + + # within the interval → rate-limited + counter.last_deny_timestamp = base + interval = ToolUseCounter._MIN_DENY_INTERVAL_SECONDS + assert not counter.can_emit_deny(base + timedelta(seconds=interval - 1)) + assert not counter.can_emit_deny(base) + + # at/after the interval → allowed again + assert counter.can_emit_deny(base + timedelta(seconds=interval)) + assert counter.can_emit_deny(base + timedelta(seconds=interval + 1)) + + def test_reset_preserves_last_deny_timestamp(self): + """``reset`` clears burst counters but must keep ``last_deny_timestamp`` intact.""" + counter = ToolUseCounter() + base = datetime.now() + counter.last_deny_timestamp = base + counter.n_recent_grep_uses = 5 + counter.n_recent_read_file_uses = 4 + counter.n_recent_non_symbolic_uses = 7 + + counter.reset() + + assert counter.n_recent_grep_uses == 0 + assert counter.n_recent_read_file_uses == 0 + assert counter.n_recent_non_symbolic_uses == 0 + assert counter.last_deny_timestamp == base + @staticmethod def _make_hook_stub(tool_name: str, timestamp: datetime) -> PreToolUseRemindAboutSerenaHook: """Create a minimal stub that satisfies ToolUseCounter.update without reading stdin.