Further extended hook heuristics

This commit is contained in:
Michael Panchenko committed 2026-04-10 13:28:41 +02:00
1 parent b21357e430
commit c9af41afd4
2 files changed
+269 -24

No files matched your search

+146 -22
View File
@@ -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:
+123 -2
View File
@@ -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.