diff --git a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py index fc488bf24..fccf053a1 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py @@ -55,7 +55,7 @@ import hashlib import json import logging import threading -from collections import OrderedDict, defaultdict +from collections import Counter, OrderedDict, defaultdict, deque from collections.abc import Awaitable, Callable from copy import deepcopy from typing import TYPE_CHECKING, override @@ -199,12 +199,14 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): Default: 20. max_tracked_threads: Maximum number of threads to track before evicting the least recently used. Default: 100. - tool_freq_warn: Number of calls to the same tool *type* (regardless - of arguments) before injecting a frequency warning. Catches - cross-file read loops that hash-based detection misses. - Default: 30. - tool_freq_hard_limit: Number of calls to the same tool type before - forcing a stop. Default: 50. + tool_freq_warn: Maximum number of same-tool-type calls within a + sliding window of ``_tool_freq_window`` before injecting a + frequency warning. Catches cross-file read loops that + hash-based detection misses. Default: 30 (within a window + of 50). + tool_freq_hard_limit: Maximum number of same-tool-type calls within + a sliding window of ``_tool_freq_window`` before forcing a + stop. Default: 50 (within a window of 50). tool_freq_overrides: Per-tool overrides for frequency thresholds, keyed by tool name. Each value is a ``(warn, hard_limit)`` tuple that replaces ``tool_freq_warn`` / ``tool_freq_hard_limit`` for @@ -233,10 +235,39 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): self.tool_freq_warn = tool_freq_warn self.tool_freq_hard_limit = tool_freq_hard_limit self._tool_freq_overrides: dict[str, tuple[int, int]] = tool_freq_overrides or {} + # Layer 2's windowed frequency count can never exceed the deque length, + # so the deque MUST be at least as long as the largest hard limit it is + # compared against — otherwise the hard-stop branch is dead code. Do NOT + # reuse Layer 1's ``window_size`` (which is unrelated and defaults below + # the freq thresholds, e.g. 20 < hard 50); size the frequency window to + # the largest hard limit in play (global + every per-tool override) so a + # tight burst can actually reach it while spread-out calls still decay + # out of the window. Warn thresholds are intentionally excluded: a sane + # config enforces warn <= hard (covered by sizing to hard), and a misconfig + # with warn > hard would hard-stop first anyway, so an unreachable warn + # is harmless and must not inflate the window. + self._tool_freq_window = max( + self.window_size, + self.tool_freq_hard_limit, + *(hard for _, hard in self._tool_freq_overrides.values()), + ) self._lock = threading.Lock() self._history: OrderedDict[str, list[str]] = OrderedDict() self._warned: dict[str, set[str]] = defaultdict(set) - self._tool_freq: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int)) + # Windowed per-tool-type frequency: recent tool names per thread, + # trimmed to ``window_size`` so the count decays instead of growing + # monotonically (replaces the old monotonic ``_tool_freq`` integer). + self._tool_name_history: defaultdict[str, deque[str]] = defaultdict(deque) + # Per-thread Counter mirroring the deque so freq_count is O(1) instead + # of scanning the whole window on every tool call. A single high + # per-tool override (e.g. bash: {hard_limit: 1000}) inflates the window + # globally, so the scan would cost 1000 per call for every tool; Counter + # increments on append and decrements on popleft. + self._tool_name_counter: defaultdict[str, Counter[str]] = defaultdict(Counter) + # Per-thread set of tool names already warned about in Layer 2, so a + # frequency warning is enqueued once rather than on every subsequent + # call. Cleared per name when the windowed count decays back below the + # warn threshold, mirroring the hash-layer ``_warned`` pruning. self._tool_freq_warned: dict[str, set[str]] = defaultdict(set) # Per-thread/run queue of warnings to inject at the next model call. # Populated by ``after_model`` (detection) and drained by @@ -325,7 +356,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): while len(self._history) > self.max_tracked_threads: evicted_id, _ = self._history.popitem(last=False) self._warned.pop(evicted_id, None) - self._tool_freq.pop(evicted_id, None) + self._tool_name_history.pop(evicted_id, None) self._tool_freq_warned.pop(evicted_id, None) for key in list(self._pending_warnings): if key[0] == evicted_id: @@ -450,44 +481,62 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): ) return _WARNING_MSG, False - # --- Layer 2: per-tool-type frequency --- - freq = self._tool_freq[thread_id] + # --- Layer 2: per-tool-type frequency (windowed) --- + tool_name_history = self._tool_name_history[thread_id] + name_counter = self._tool_name_counter[thread_id] for tc in tool_calls: name = tc.get("name", "") if not name: continue - freq[name] += 1 - tc_count = freq[name] + # Windowed counting: append the name and trim to the frequency + # window (>= the largest threshold) so the count can reach the + # warn/hard limits on a tight burst yet still decay for + # spread-out calls. A mirrored Counter gives O(1) freq_count + # even when a per-tool override inflates the window globally. + tool_name_history.append(name) + name_counter[name] += 1 + while len(tool_name_history) > self._tool_freq_window: + old = tool_name_history.popleft() + c = name_counter[old] - 1 + if c <= 0: + del name_counter[old] + else: + name_counter[old] = c + freq_count = name_counter.get(name, 0) if name in self._tool_freq_overrides: eff_warn, eff_hard = self._tool_freq_overrides[name] else: eff_warn, eff_hard = self.tool_freq_warn, self.tool_freq_hard_limit - if tc_count >= eff_hard: + if freq_count >= eff_hard: logger.error( "Tool frequency hard limit reached — forcing stop", extra={ "thread_id": thread_id, "tool_name": name, - "count": tc_count, + "count": freq_count, }, ) - return _TOOL_FREQ_HARD_STOP_MSG.format(tool_name=name, count=tc_count), True + return _TOOL_FREQ_HARD_STOP_MSG.format(tool_name=name, count=freq_count), True - if tc_count >= eff_warn: - warned = self._tool_freq_warned[thread_id] - if name not in warned: - warned.add(name) + if freq_count >= eff_warn: + freq_warned = self._tool_freq_warned[thread_id] + if name not in freq_warned: + freq_warned.add(name) logger.warning( "Tool frequency warning — too many calls to same tool type", extra={ "thread_id": thread_id, "tool_name": name, - "count": tc_count, + "count": freq_count, }, ) - return _TOOL_FREQ_WARNING_MSG.format(tool_name=name, count=tc_count), False + return _TOOL_FREQ_WARNING_MSG.format(tool_name=name, count=freq_count), False + else: + # Windowed count decayed below the warn threshold; allow a + # future burst of this tool to warn again. + self._tool_freq_warned[thread_id].discard(name) return None, False @@ -663,7 +712,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): if thread_id: self._history.pop(thread_id, None) self._warned.pop(thread_id, None) - self._tool_freq.pop(thread_id, None) + self._tool_name_history.pop(thread_id, None) self._tool_freq_warned.pop(thread_id, None) for key in list(self._pending_warnings): if key[0] == thread_id: @@ -671,7 +720,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): else: self._history.clear() self._warned.clear() - self._tool_freq.clear() + self._tool_name_history.clear() self._tool_freq_warned.clear() self._pending_warnings.clear() self._pending_warning_touch_order.clear() diff --git a/backend/tests/test_loop_detection_middleware.py b/backend/tests/test_loop_detection_middleware.py index 6f6890e5f..1caa73c89 100644 --- a/backend/tests/test_loop_detection_middleware.py +++ b/backend/tests/test_loop_detection_middleware.py @@ -581,8 +581,7 @@ class TestLoopDetection: mw._apply(_make_state(tool_calls=call), runtime_new) assert "thread-0" not in mw._history - assert "thread-0" not in mw._tool_freq - assert "thread-0" not in mw._tool_freq_warned + assert "thread-0" not in mw._tool_name_history assert "thread-new" in mw._history assert len(mw._history) == 3 @@ -962,6 +961,51 @@ class TestToolFrequencyDetection: assert "FORCED STOP" in msg.content assert "read_file" in msg.content + def test_windowed_frequency_decay_avoids_hard_stop_when_interleaved(self): + """More than ``window_size`` total calls to one tool type must NOT hard-stop + as long as they are spread out. + + Interleaving read_file with another tool keeps the per-window count under + the hard limit, so the windowed counter decays instead of accumulating + monotonically. The old monotonic counter would hard-stop on the 4th + read_file regardless of spacing. + """ + mw = LoopDetectionMiddleware(tool_freq_warn=100, tool_freq_hard_limit=4, window_size=5) + runtime = _make_runtime() + + # Alternate read_file with bash. In any window of 5 consecutive calls the + # read_file count peaks at 3 (< hard_limit=4), so no hard stop fires even + # though total read_file calls (8) exceeds window_size (5). Distinct args + # keep the Layer-1 hash detector from firing, isolating Layer 2. + read_count = 0 + for i in range(8): + result = mw._apply(_make_state(tool_calls=[self._read_call(f"/file_{i}.py")]), runtime) + assert result is None, f"read call {i} unexpectedly hard-stopped" + read_count += 1 + result = mw._apply(_make_state(tool_calls=[_bash_call(f"cmd_{i}")]), runtime) + assert result is None, f"bash call {i} unexpectedly hard-stopped" + + assert read_count == 8 # more than window_size total read_file calls, no hard stop + + def test_rapid_identical_tool_type_in_one_window_still_hard_stops(self): + """The decay must not weaken the guard: ``window_size``+ rapid calls to the + same tool type within one window still trip the frequency hard-stop.""" + mw = LoopDetectionMiddleware(tool_freq_warn=100, tool_freq_hard_limit=4, window_size=5) + runtime = _make_runtime() + + # Distinct args each call -> the hash-based (Layer 1) detector never fires, + # isolating the per-tool-type frequency layer. + for i in range(3): + assert mw._apply(_make_state(tool_calls=[self._read_call(f"/f_{i}.py")]), runtime) is None + + result = mw._apply(_make_state(tool_calls=[self._read_call("/f_3.py")]), runtime) + assert result is not None + msg = result["messages"][0] + assert isinstance(msg, AIMessage) + assert msg.tool_calls == [] + assert "FORCED STOP" in msg.content + assert "read_file" in msg.content + def test_different_tools_tracked_independently(self): """read_file and bash should have independent frequency counters.""" mw = LoopDetectionMiddleware(tool_freq_warn=3, tool_freq_hard_limit=10) @@ -1008,8 +1052,7 @@ class TestToolFrequencyDetection: # Reset only thread-A mw.reset(thread_id="thread-A") - assert "thread-A" not in mw._tool_freq - assert "thread-A" not in mw._tool_freq_warned + assert "thread-A" not in mw._tool_name_history # thread-B state should still be intact — 3rd call queues a warn. result = mw._apply(_make_state(tool_calls=[self._read_call("/b_2.py")]), runtime_b) @@ -1162,3 +1205,36 @@ class TestFromConfig: queued = mw._pending_warnings.get(_pending_key(), []) assert queued assert "LOOP DETECTED" in queued[0] + + def test_freq_window_sized_to_hard_limit_under_defaults(self): + """Regression for #4072: the Layer-2 frequency window must be >= the + largest threshold, or the warn/hard branches are dead code. With the + shipped defaults (window 20 < warn 30 < hard 50) the freq deque must be + sized to 50, not 20.""" + mw = LoopDetectionMiddleware.from_config(self._config()) + assert mw._tool_freq_window >= mw.tool_freq_hard_limit + assert mw._tool_freq_window >= mw.tool_freq_warn + + def test_freq_window_covers_largest_override_hard_limit(self): + mw = LoopDetectionMiddleware.from_config(self._config(tool_freq_overrides={"bash": {"warn": 60, "hard_limit": 120}})) + assert mw._tool_freq_window >= 120 + + def test_tight_burst_hard_stops_under_default_config(self): + """Under the real default config, one tool type called many times with + *distinct* args (which Layer 1's name+args hash never catches) must still + be hard-stopped by Layer 2. This fails if the freq window is capped at + ``window_size`` (20) below the hard limit (50).""" + mw = LoopDetectionMiddleware.from_config(self._config()) + runtime = _make_runtime() + hard = mw.tool_freq_hard_limit # 50 by default + + # Distinct args every call -> unique hashes -> Layer 1 (hash) never trips. + # Only Layer 2 (per-tool-type frequency) can catch this tight burst. + for i in range(hard - 1): + result = mw._apply(_make_state(tool_calls=[_bash_call(f"cmd_{i}")]), runtime) + assert result is None, f"unexpected hard stop before the limit at call {i}" + + # The call that pushes freq_count to the hard limit fires the stop. + result = mw._apply(_make_state(tool_calls=[_bash_call(f"cmd_{hard}")]), runtime) + assert result is not None + assert mw.consume_stop_reason("test-run") == "loop_capped"