From e7c059d8d4c2683533035312d62cdc1e12c3bfba Mon Sep 17 00:00:00 2001 From: PeaceMaker-best Date: Mon, 7 Sep 2026 18:31:11 +0800 Subject: [PATCH] fix(agents): prioritize loop hard stops across tool batches (#5245) * fix(agents): prioritize loop hard stops across tool batches Scan an admitted multi-tool response completely before selecting a soft warning, so any configured hard limit can reject the whole batch. Preserve warning priority and sliding-window accounting. Add counter-level plus sync and async compiled-agent regressions proving rejected tools are not executed. Fixes bytedance/deer-flow#5243. AI-assisted implementation and tests. * fix(agents): rearm loop warnings after cross-tool eviction When another tool evicts an older tool below its frequency warning threshold, clear the older suppression mark so a later burst can warn again. Add the cross-tool sliding-window regression from the final boundary review. AI-assisted implementation and tests. * test(agents): cover override-aware loop warning rearm Cache the default frequency thresholds for sliding-window eviction and verify that an evicted tool uses its configured override when warning eligibility is rearmed. Document that simultaneous frequency warnings preserve legacy first-crossing selection while hard stops remain batch-severity-first. Addresses review on #5245. AI-assisted implementation and tests. --------- Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: Willem Jiang --- backend/README.md | 6 + .../deerflow/agents/middlewares/AGENTS.md | 11 + .../middlewares/loop_detection_middleware.py | 89 +++++---- .../tests/test_loop_detection_middleware.py | 188 ++++++++++++++++++ 4 files changed, 257 insertions(+), 37 deletions(-) diff --git a/backend/README.md b/backend/README.md index a3a74c276..88f87f7f5 100644 --- a/backend/README.md +++ b/backend/README.md @@ -66,6 +66,12 @@ Middlewares execute in strict order, each handling a specific concern: | 8 | **ViewImageMiddleware** | Injects image data for vision-capable models (conditional) | | 9 | **ClarificationMiddleware** | Intercepts clarification requests and interrupts execution (must be last) | +When `loop_detection.enabled` is set, loop detection checks both repeated +tool-call sets and per-tool frequency. Warnings do not skip the rest of a +tool-call batch: any hard limit reached takes precedence and stops the entire +batch before tool execution. Warning-only batches remain fully counted and +receive a transient hint on the next model request. + ### Sandbox System Per-thread isolated execution with virtual path translation: diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index d74c1e38d..7eccbdc89 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -95,6 +95,17 @@ Before changing a later authorization phase, read the [authorization RFC](../../ 27. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives. The subagent builder places its date-only context middleware immediately before this coalescer, so the built-in subagent prompt and hidden date reminder still reach providers as one leading system block 28. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess ordinary `task` tool calls to enforce both the per-response concurrency limit (`max_concurrent_subagents`, resolved against startup `subagent_runtime.max_running` and the 1-64 safety range before construction) and the per-run total delegation cap (`max_total_subagents` runtime override or `subagents.max_total_per_run`, default 6, clamped to 1-50). The total cap counts current-run entries in the durable delegation ledger (entries are tagged with `run_id` when captured), so repeated planning checkpoints in one run cannot keep launching legal-sized batches indefinitely, while later user turns in the same thread get a fresh run budget. Explicit durable `batch_task` calls are a separate mode with persisted total/live/running limits and are not rewritten into ordinary ledger entries. If the ordinary cap is exhausted, the middleware strips remaining `task` calls, forces `finish_reason="stop"`, and appends a visible limit note so the run can synthesize existing results instead of ending with an empty tool-call response. 29. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware`; persists warned-state transitions (first per call hash or per tool-frequency burst) and hard stops as `middleware:loop_detection`, attributed with `is_subagent` and the optional `agent_id`, without tool arguments, message content, tool results, or argument-derived hashes. Ordinary task subagents get dedicated recorder keys through a parent-loop proxy; never pass `RunJournal` into their isolated loop. Durable batch subagents have no parent run journal and do not persist these transitions + Loop decisions are severity-first across both detection layers: a warning + candidate never short-circuits frequency accounting for the remaining calls + in an admitted batch. A hard limit can stop scanning immediately because it + rejects the entire batch. Only the selected warning is marked and logged; + hash warnings still precede frequency warnings when neither layer stops the + run. Among simultaneous frequency-warning candidates, the first crossing in + model tool-call order remains selected for compatibility; later calls are + still counted and can warn in a later batch. A frequency warning whose burst + decays within the batch must not leave a stale suppression mark. + `tests/test_loop_detection_middleware.py` covers mixed-tool batches, window + decay, overrides, and sync/async compiled-graph execution. 30. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits 31. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before config-declared extensions and the terminal-response/safety/clarification tail 32. **Configured extension middlewares** - *(optional, if `extensions.middlewares` is set in `config.yaml` or `extensions_config.json`)* Zero-argument `AgentMiddleware` classes loaded from `module.path:ClassName` entries via `deerflow.reflection.resolve_class`. Missing packages, invalid classes, and broken modules fail loudly at agent creation. These run after built-ins/programmatic custom middleware and after the lead/subagent loop/token guards, but before the terminal-response/safety/clarification tail; subagents receive the same configured extension middleware class list before their safety tail. Treat these files as trusted operator config because middleware paths instantiate arbitrary code. Gateway skill/MCP toggle endpoints preserve this field through `to_file_dict()` but must not add a write path for `extensions.middlewares` without an explicit trust-boundary review. Lead-only vs subagent-only middleware lists and per-context constructor parameters are not expressible in this MVP. 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 165f9d910..061af7f2b 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py @@ -253,6 +253,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): self.max_tracked_threads = max_tracked_threads self.tool_freq_warn = tool_freq_warn self.tool_freq_hard_limit = tool_freq_hard_limit + self._default_tool_freq_thresholds = (tool_freq_warn, 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 @@ -508,27 +509,21 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): threshold=self.hard_limit, ) - if count >= self.warn_threshold: - warned = self._warned[thread_id] - if call_hash not in warned: - warned.add(call_hash) - logger.warning( - "Repetitive tool calls detected — injecting warning", - extra={ - "thread_id": thread_id, - "call_hash": call_hash, - "count": count, - "tools": tool_names, - }, - ) - return _LoopDecision( - message=_WARNING_MSG, - action="warn", - detection_layer="identical_call_set", - tool_names=tuple(tool_names), - count=count, - threshold=self.warn_threshold, - ) + # Warnings admit the whole batch, so they must not skip frequency + # accounting or hide a later hard limit. Keep one candidate (hash + # warnings retain priority over frequency warnings) until every + # admitted call has been checked. Only the selected warning is + # marked/logged; a hard stop may supersede it below. + warning: _LoopDecision | None = None + if count >= self.warn_threshold and call_hash not in self._warned.get(thread_id, set()): + warning = _LoopDecision( + message=_WARNING_MSG, + action="warn", + detection_layer="identical_call_set", + tool_names=tuple(tool_names), + count=count, + threshold=self.warn_threshold, + ) # --- Layer 2: per-tool-type frequency (windowed) --- tool_name_history = self._tool_name_history[thread_id] @@ -551,12 +546,15 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): del name_counter[old] else: name_counter[old] = c + old_warn = self._tool_freq_overrides.get(old, self._default_tool_freq_thresholds)[0] + if c < old_warn: + # Any tool can evict an older name from the shared + # window. Rearm that name as soon as its burst decays, + # even when the current call belongs to another tool. + self._tool_freq_warned[thread_id].discard(old) 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 + eff_warn, eff_hard = self._tool_freq_overrides.get(name, self._default_tool_freq_thresholds) if freq_count >= eff_hard: logger.error( @@ -578,17 +576,8 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): 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": freq_count, - }, - ) - return _LoopDecision( + if warning is None and name not in freq_warned: + warning = _LoopDecision( message=_TOOL_FREQ_WARNING_MSG.format(tool_name=name, count=freq_count), action="warn", detection_layer="tool_frequency", @@ -601,7 +590,33 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): # future burst of this tool to warn again. self._tool_freq_warned[thread_id].discard(name) - return None + if warning is not None: + if warning.detection_layer == "identical_call_set": + self._warned[thread_id].add(call_hash) + logger.warning( + "Repetitive tool calls detected — injecting warning", + extra={ + "thread_id": thread_id, + "call_hash": call_hash, + "count": warning.count, + "tools": list(warning.tool_names), + }, + ) + else: + warned_name = warning.tool_names[0] + # Later calls in this batch may already have decayed this + # burst. Do not suppress the next burst with a stale mark. + if name_counter.get(warned_name, 0) >= warning.threshold: + self._tool_freq_warned[thread_id].add(warned_name) + logger.warning( + "Tool frequency warning — too many calls to same tool type", + extra={ + "thread_id": thread_id, + "tool_name": warned_name, + "count": warning.count, + }, + ) + return warning @staticmethod def _append_text(content: str | list | None, text: str) -> str | list: diff --git a/backend/tests/test_loop_detection_middleware.py b/backend/tests/test_loop_detection_middleware.py index 0c75bc885..de893fb7a 100644 --- a/backend/tests/test_loop_detection_middleware.py +++ b/backend/tests/test_loop_detection_middleware.py @@ -1418,6 +1418,194 @@ class TestToolFrequencyDetection: assert _HARD_STOP_MSG in msg.content +class TestToolCallBatchDecisions: + """A soft warning must not skip accounting or mask a hard stop in a batch.""" + + @staticmethod + def _call(name, value): + if name == "read_file": + args = {"path": f"/{value}.py"} + elif name == "bash": + args = {"command": f"echo {value}"} + else: + args = {"value": value} + return {"name": name, "id": f"call_{name}_{value}", "args": args} + + def test_single_batch_crossing_both_thresholds_hard_stops(self): + mw = LoopDetectionMiddleware(tool_freq_warn=2, tool_freq_hard_limit=3) + runtime = _make_runtime() + journal = MagicMock() + runtime.context["__run_journal"] = journal + calls = [self._call("read_file", i) for i in range(3)] + + result = mw.after_model(_make_state(tool_calls=calls), runtime) + + assert result is not None + assert result["messages"][0].tool_calls == [] + assert mw.consume_stop_reason("test-run") == "loop_capped" + assert not mw._pending_warnings + assert not mw._tool_freq_warned.get("test-thread") + journal.record_middleware.assert_called_once() + recorded = journal.record_middleware.call_args.kwargs + assert recorded["action"] == "hard_stop" + assert recorded["changes"]["detection_layer"] == "tool_frequency" + assert recorded["changes"]["count"] == 3 + + @pytest.mark.parametrize("hard_stop_first", [False, True]) + def test_one_tools_warning_cannot_mask_another_tools_hard_stop(self, hard_stop_first): + mw = LoopDetectionMiddleware(tool_freq_warn=2, tool_freq_hard_limit=3) + runtime = _make_runtime() + mw._apply(_make_state(tool_calls=[self._call("bash", 0)]), runtime) + mw._apply(_make_state(tool_calls=[self._call("read_file", 0), self._call("bash", 1)]), runtime) + calls = [self._call("read_file", 1), self._call("bash", 2)] + if hard_stop_first: + calls.reverse() + + decision = mw._track_and_check(_make_state(tool_calls=calls), runtime) + + assert decision is not None and decision.hard_stop + assert decision.tool_names == ("bash",) + assert decision.count == 3 + assert "read_file" not in mw._tool_freq_warned["test-thread"] + + def test_identical_call_warning_cannot_mask_frequency_hard_stop(self): + mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=5, tool_freq_warn=2, tool_freq_hard_limit=3) + runtime = _make_runtime() + calls = [self._call("read_file", 0), self._call("read_file", 1)] + mw._apply(_make_state(tool_calls=calls), runtime) + + decision = mw._track_and_check(_make_state(tool_calls=calls), runtime) + + assert decision is not None and decision.hard_stop + assert decision.detection_layer == "tool_frequency" + assert decision.count == 3 + assert not mw._warned.get("test-thread") + + def test_warning_batch_counts_all_calls_and_only_marks_selected_warning(self): + mw = LoopDetectionMiddleware(tool_freq_warn=2, tool_freq_hard_limit=10) + runtime = _make_runtime() + calls = [self._call("read_file", i) for i in range(3)] + [self._call("bash", i) for i in range(2)] + + first = mw._track_and_check(_make_state(tool_calls=calls), runtime) + + assert first is not None and first.action == "warn" + assert first.tool_names == ("read_file",) + assert list(mw._tool_name_history["test-thread"]) == ["read_file"] * 3 + ["bash"] * 2 + assert dict(mw._tool_name_counter["test-thread"]) == {"read_file": 3, "bash": 2} + assert mw._tool_freq_warned["test-thread"] == {"read_file"} + + second = mw._track_and_check(_make_state(tool_calls=[self._call("bash", 2)]), runtime) + assert second is not None and second.action == "warn" + assert second.tool_names == ("bash",) + assert second.count == 3 + + def test_hash_warning_preserves_frequency_accounting_and_pending_warning(self): + mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=10, tool_freq_warn=2, tool_freq_hard_limit=5) + runtime = _make_runtime() + calls = [self._call("read_file", 0)] + assert mw._track_and_check(_make_state(tool_calls=calls), runtime) is None + + hash_warning = mw._track_and_check(_make_state(tool_calls=calls), runtime) + + assert hash_warning is not None and hash_warning.detection_layer == "identical_call_set" + assert mw._tool_name_counter["test-thread"]["read_file"] == 2 + assert not mw._tool_freq_warned.get("test-thread") + + freq_warning = mw._track_and_check(_make_state(tool_calls=[self._call("read_file", 1)]), runtime) + assert freq_warning is not None and freq_warning.detection_layer == "tool_frequency" + assert freq_warning.count == 3 + + def test_warning_that_decays_within_batch_can_warn_on_next_burst(self): + mw = LoopDetectionMiddleware(window_size=4, tool_freq_warn=2, tool_freq_hard_limit=3) + runtime = _make_runtime() + calls = [self._call(name, i) for i, name in enumerate(["a", "b", "a", "c", "d"])] + + first = mw._track_and_check(_make_state(tool_calls=calls), runtime) + + assert first is not None and first.tool_names == ("a",) + assert mw._tool_name_counter["test-thread"]["a"] == 1 + assert "a" not in mw._tool_freq_warned["test-thread"] + second = mw._track_and_check(_make_state(tool_calls=[self._call("a", 5)]), runtime) + assert second is not None and second.action == "warn" + assert second.tool_names == ("a",) + + def test_other_tool_eviction_rearms_frequency_warning(self): + mw = LoopDetectionMiddleware(window_size=3, tool_freq_warn=2, tool_freq_hard_limit=3) + runtime = _make_runtime() + + first = mw._track_and_check( + _make_state(tool_calls=[self._call(name, i) for i, name in enumerate(["a", "b", "a"])]), + runtime, + ) + assert first is not None and first.tool_names == ("a",) + + second = mw._track_and_check(_make_state(tool_calls=[self._call("b", 3)]), runtime) + assert second is not None and second.tool_names == ("b",) + assert mw._tool_name_counter["test-thread"]["a"] == 1 + assert "a" not in mw._tool_freq_warned["test-thread"] + + third = mw._track_and_check(_make_state(tool_calls=[self._call("a", 4)]), runtime) + assert third is not None and third.tool_names == ("a",) + + def test_cross_tool_eviction_rearms_at_evicted_tools_override_threshold(self): + mw = LoopDetectionMiddleware( + window_size=3, + tool_freq_warn=2, + tool_freq_hard_limit=5, + tool_freq_overrides={"bash": (3, 4)}, + ) + runtime = _make_runtime() + + first = mw._track_and_check( + _make_state(tool_calls=[self._call("bash", i) for i in range(3)]), + runtime, + ) + assert first is not None and first.tool_names == ("bash",) + assert mw._tool_freq_warned["test-thread"] == {"bash"} + + second = mw._track_and_check( + _make_state(tool_calls=[self._call("read_file", i) for i in range(3)]), + runtime, + ) + assert second is not None and second.tool_names == ("read_file",) + assert mw._tool_name_counter["test-thread"]["bash"] == 2 + assert "bash" not in mw._tool_freq_warned["test-thread"] + + third = mw._track_and_check( + _make_state(tool_calls=[self._call("bash", i) for i in range(3, 6)]), + runtime, + ) + assert third is not None and third.tool_names == ("bash",) + assert third.count == 3 + + @pytest.mark.asyncio + @pytest.mark.parametrize("use_async", [False, True]) + async def test_batch_hard_stop_prevents_tool_execution_in_real_agent_graph(self, use_async): + executed = [] + + @as_tool + def read_file(path: str) -> str: + """Read a fake file without touching the filesystem.""" + executed.append(path) + return path + + calls = [{"name": "read_file", "id": f"read_{i}", "args": {"path": f"/{i}.py"}} for i in range(3)] + mw = LoopDetectionMiddleware(tool_freq_warn=2, tool_freq_hard_limit=3) + model = _CapturingFakeMessagesListChatModel(responses=[AIMessage(content="", tool_calls=calls), AIMessage(content="done")]) + graph = create_agent(model=model, tools=[read_file], middleware=[mw]) + inputs = {"messages": [("user", "inspect these files")]} + kwargs = {"context": {"thread_id": "batch-thread", "run_id": "batch-run"}, "config": {"recursion_limit": 10}} + + result = await graph.ainvoke(inputs, **kwargs) if use_async else graph.invoke(inputs, **kwargs) + + assert executed == [] + assert len(model.seen_messages) == 1 + assert result["messages"][-1].tool_calls == [] + assert "FORCED STOP" in result["messages"][-1].content + assert mw.consume_stop_reason("batch-run") == "loop_capped" + assert not mw._pending_warnings + + class TestFromConfig: """Tests for LoopDetectionMiddleware.from_config — the sole validated construction path."""