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 <willem.jiang@gmail.com>

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:

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.

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:

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."""