Fixed global hook grace period

This commit is contained in:
Michael Panchenko committed 2026-04-10 15:33:35 +02:00
1 parent c7b4fb49cb
commit 6197f6d8e6
2 files changed
+53 -33

No files matched your search

+29 -20
View File
@@ -103,11 +103,13 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook):
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.
The hook is additionally gated by :attr:`ToolUseCounter._MIN_DENY_INTERVAL_SECONDS`
(two minutes by default): once a deny has been emitted, *every* subsequent
invocation of this hook is a no-op until the window has elapsed — neither the
counters are updated nor any further deny is produced. This prevents the agent
from being nudged more than once per window during a sustained non-symbolic-tool
burst, and also avoids surprising the user with reminders that were already
counted up under stale state.
"""
@dataclass
@@ -124,8 +126,9 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook):
# 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
# minimum seconds between two engagements of the hook after a deny; the entire
# hook (counter updates included) is a no-op while this window is active, so a
# single sustained burst triggers at most one nudge per window
_MIN_DENY_INTERVAL_SECONDS = 120
n_recent_read_file_uses: int = 0
@@ -147,13 +150,13 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook):
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.
def is_hook_active(self, now: datetime) -> bool:
""":return: whether the hook should engage at all at ``now``. Returns
``False`` while we are still within :attr:`_MIN_DENY_INTERVAL_SECONDS`
of the most recent emitted deny — in that case the entire hook is
short-circuited (no counter updates, no deny). Returns ``True`` when
no deny has been emitted yet in this session, or when the window has
elapsed.
"""
if self.last_deny_timestamp is None:
return True
@@ -258,6 +261,13 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook):
return any(verb in name for verb in self._READ_FILE_VERB_SUBSTRINGS)
def execute(self) -> None:
# gate the entire hook on the rate-limit window: while we are within
# _MIN_DENY_INTERVAL_SECONDS of the last emitted deny, no counter
# updates and no deny detection happen at all. The pickle is left
# untouched so state survives the gating window unchanged.
if not self._tool_call_counter.is_hook_active(self.triggered_at_timestamp):
return
self._tool_call_counter.update(self)
# pick the deny that matches the current call first, so the emitted message lines up
@@ -282,13 +292,12 @@ class PreToolUseRemindAboutSerenaHook(PreToolUseHook):
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
# reset burst counters so the next interval starts fresh, then record
# the deny timestamp and emit. The is_hook_active() guard at the top
# ensures we never reach this branch within the rate-limit window.
self._tool_call_counter.reset()
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.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":
+24 -13
View File
@@ -180,8 +180,10 @@ 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."""
def test_rate_limit_gates_entire_hook_within_interval(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]):
"""While within the rate-limit window, the entire hook must be a no-op:
no deny is emitted, AND the persisted counters must remain untouched.
"""
# 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)):
@@ -189,13 +191,22 @@ class TestPreToolUseRemindAboutSerenaHook:
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
# snapshot the persisted counter immediately after the deny was emitted
stub_for_path = object.__new__(PreToolUseRemindAboutSerenaHook)
stub_for_path.session_persistence_dir = str(tmp_path / "hook_data" / _base_input()["session_id"])
counter_before = ToolUseCounter.load(stub_for_path)
# second burst immediately after: within the rate-limit window, the entire
# hook must short-circuit — no deny output and no counter mutation
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 == ""
counter_after = ToolUseCounter.load(stub_for_path)
assert counter_after == counter_before, "gated hook must not mutate the persisted counter"
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
@@ -348,23 +359,23 @@ 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."""
def test_is_hook_active_respects_min_interval(self):
""":meth:`is_hook_active` 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)
# no prior deny → hook is always active
assert counter.is_hook_active(base)
# within the interval → rate-limited
# within the interval → hook gated
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)
assert not counter.is_hook_active(base + timedelta(seconds=interval - 1))
assert not counter.is_hook_active(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))
# at/after the interval → hook active again
assert counter.is_hook_active(base + timedelta(seconds=interval))
assert counter.is_hook_active(base + timedelta(seconds=interval + 1))
def test_reset_preserves_last_deny_timestamp(self):
"""``reset`` clears burst counters but must keep ``last_deny_timestamp`` intact."""