diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7da883a0c..085be63c3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -582,6 +582,15 @@ This section accumulates work toward the **2.1.0** milestone
 
 ### Fixed
 
+- **middleware:** Stop a guard that removes tool calls from breaking every later
+  turn of a Claude or OpenAI Responses thread. Token-budget and loop-detection
+  hard stops, subagent-limit truncation, and safety suppression cleared
+  `tool_calls` but left the provider's own tool-call blocks in the message
+  content. Anthropic and the Responses API resend those blocks, so the next
+  request carried a tool call with no result and the provider rejected it, and
+  a hard stop saved to the checkpoint kept failing on each new message. All
+  guards now remove the matching content blocks through one shared helper,
+  which also keeps a Responses call that clarification retains. ([#5447])
 - **sandbox:** Stop remote `glob` and `grep` from reporting "no matches" when
   their output was cut off. BoxLite, Tenki, E2B, and OpenSandbox cap the
   search's raw output and then filter it in Python (ignored directories such as
@@ -2858,4 +2867,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#5419]: https://github.com/bytedance/deer-flow/pull/5419
 [#5427]: https://github.com/bytedance/deer-flow/pull/5427
 [#5431]: https://github.com/bytedance/deer-flow/pull/5431
+[#5447]: https://github.com/bytedance/deer-flow/pull/5447
 
diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md
index ecf184fe0..d3055b3f7 100644
--- a/CHANGELOG_zh.md
+++ b/CHANGELOG_zh.md
@@ -397,6 +397,12 @@
 
 ### 修复
 
+- **中间件：** 移除工具调用的守卫不再导致 Claude 或 OpenAI Responses 线程之后的每一轮都失败。
+  token 预算与循环检测的硬停止、subagent 数量限制的截断以及安全终止抑制只清空了 `tool_calls`，
+  却把 provider 自身的工具调用块留在消息 content 中。Anthropic 与 Responses API 会重新发送这些块，
+  导致下一次请求带着没有结果的工具调用而被 provider 拒绝；写入 checkpoint 的硬停止消息还会让之后
+  每条新消息都失败。现在所有守卫都通过同一个共享 helper 删除对应的 content 块，该 helper 也会保留
+  clarification 所保留的 Responses 调用。([#5447])
 - **沙箱：** 远程 `glob` 与 `grep` 的输出被截断时，不再报告"没有匹配"。BoxLite、Tenki、E2B 与
   OpenSandbox 会先限制搜索的原始输出行数，再在 Python 中过滤（`node_modules` 等忽略目录、匹配模式或 `glob`
   范围），但只有达到 `max_results` 时才报告 `truncated`。若被截取的行全部被过滤掉，截断位置之后仍有
@@ -2184,3 +2190,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#5419]: https://github.com/bytedance/deer-flow/pull/5419
 [#5427]: https://github.com/bytedance/deer-flow/pull/5427
 [#5431]: https://github.com/bytedance/deer-flow/pull/5431
+[#5447]: https://github.com/bytedance/deer-flow/pull/5447
diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
index 1f55f00ca..a2c85e905 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
+++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
@@ -45,6 +45,10 @@ Continuity history readers share shape validation, including the capture failure
 path and DurableContext rendering, so malformed persisted metadata cannot abort
 ordinary compaction or a model call.
 
+**Removing tool calls.** Use `clone_ai_message_with_tool_calls`, not a bare
+`tool_calls` update: adapters resend stale `content` tool-call blocks, which
+strict providers reject.
+
 **Shared runtime base** (`build_lead_runtime_middlewares`; subagents reuse most of this via `build_subagent_runtime_middlewares`):
 
 1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages. `additional_kwargs.original_user_content` is server-owned provenance: Gateway strips caller-supplied values for non-internal run requests, trusted IM calls may carry the string they captured before adding transport/file context, and the middleware replaces any non-string value before wrapping. Uploads and sanitization retain first-writer-wins only for validated strings. Caller markers are marked `untrusted_input`, never stripped; scope is every turn.
@@ -107,7 +111,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
 26. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)
 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
+29. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears structured, raw, and content-block tool calls 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
    State is run-scoped: new user runs get fresh budgets; same-run goal
    continuations share history. Keep sibling warnings isolated and lifecycle
    hooks topology-stable. Before changing this guard, read
diff --git a/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py
index 4bac91f79..623ec821a 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py
@@ -71,30 +71,6 @@ class ClarificationMiddlewareState(AgentState):
     pass
 
 
-def _filter_content_tool_use(content: Any, kept_ids: set[str], kept_names: set[str]) -> Any:
-    """Drop provider tool-use blocks that were stripped from ``tool_calls``.
-
-    Anthropic ``tool_use`` blocks carry an ``id`` that matches ``tool_calls``.
-    Gemini-style ``function_call`` blocks often have no ``id`` (langchain
-    synthesizes ids onto ``tool_calls`` only), so those are matched by ``name``.
-    """
-    if not isinstance(content, list):
-        return content
-    filtered: list[Any] = []
-    for block in content:
-        if isinstance(block, dict) and block.get("type") in {"tool_use", "function_call"}:
-            block_id = block.get("id")
-            if isinstance(block_id, str) and block_id:
-                if block_id not in kept_ids:
-                    continue
-            elif block.get("type") == "function_call":
-                name = block.get("name")
-                if not isinstance(name, str) or name not in kept_names:
-                    continue
-        filtered.append(block)
-    return filtered
-
-
 class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]):
     """Intercepts clarification tool calls and interrupts execution to present questions to the user.
 
@@ -464,15 +440,9 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]):
             dropped_names,
         )
 
-        kept_for_content = clarification_calls + invalid_clarification_calls
-        kept_ids = {tc["id"] for tc in kept_for_content if isinstance(tc.get("id"), str) and tc["id"]}
-        kept_names = {str(tc["name"]) for tc in kept_for_content if isinstance(tc.get("name"), str) and tc["name"]}
-        new_content = _filter_content_tool_use(last.content, kept_ids, kept_names)
-        patched = clone_ai_message_with_tool_calls(
-            last,
-            clarification_calls,
-            content=new_content if new_content is not last.content else None,
-        )
+        # The clone also drops the siblings' provider content blocks, keeping
+        # blocks for calls that remain on tool_calls or invalid_tool_calls.
+        patched = clone_ai_message_with_tool_calls(last, clarification_calls)
         return {"messages": [patched]}
 
     def _handle_disabled_clarification(self, request: ToolCallRequest) -> ToolMessage:
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 f539a2b26..d9c1b0b15 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py
@@ -68,7 +68,6 @@ import threading
 import uuid
 from collections import Counter, OrderedDict, defaultdict, deque
 from collections.abc import Awaitable, Callable
-from copy import deepcopy
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal, override
 
@@ -83,6 +82,7 @@ from deerflow.agents.middlewares.audit_context import (
     LOOP_DETECTION_RECORDER_CONTEXT_KEY,
     resolve_audit_recorder,
 )
+from deerflow.agents.middlewares.tool_call_metadata import clone_ai_message_with_tool_calls
 from deerflow.runtime.events.catalog import MIDDLEWARE_LOOP_DETECTION_TAG
 
 if TYPE_CHECKING:
@@ -708,26 +708,6 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
         # Fallback: coerce unexpected types to str to avoid TypeError
         return str(content) + f"\n\n{text}"
 
-    @staticmethod
-    def _build_hard_stop_update(last_msg, content: str | list) -> dict:
-        """Clear tool-call metadata so forced-stop messages serialize as plain assistant text."""
-        update = {
-            "tool_calls": [],
-            "content": content,
-        }
-
-        additional_kwargs = dict(getattr(last_msg, "additional_kwargs", {}) or {})
-        for key in ("tool_calls", "function_call"):
-            additional_kwargs.pop(key, None)
-        update["additional_kwargs"] = additional_kwargs
-
-        response_metadata = deepcopy(getattr(last_msg, "response_metadata", {}) or {})
-        if response_metadata.get("finish_reason") == "tool_calls":
-            response_metadata["finish_reason"] = "stop"
-        update["response_metadata"] = response_metadata
-
-        return update
-
     def _record_audit_event(
         self,
         decision: _LoopDecision,
@@ -791,14 +771,15 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
             ctx = getattr(runtime, "context", None)
             if isinstance(ctx, dict):
                 ctx["stop_reason"] = "loop_capped"
-            # Strip tool_calls from the last AIMessage to force text output.
-            # Once tool_calls are stripped, the AIMessage no longer requires
-            # matching ToolMessage responses, so mutating it in place here
-            # is safe for OpenAI/Moonshot pairing validators.
+            # Strip tool calls from every provider surface of the last
+            # AIMessage (structured, raw, and content blocks) to force text
+            # output. With no call left on any surface, the AIMessage no
+            # longer requires matching ToolMessage responses, so replacing it
+            # here is safe for strict provider pairing validators.
             messages = state.get("messages", [])
             last_msg = messages[-1]
             content = self._append_text(last_msg.content, warning or _HARD_STOP_MSG)
-            stripped_msg = last_msg.model_copy(update=self._build_hard_stop_update(last_msg, content))
+            stripped_msg = clone_ai_message_with_tool_calls(last_msg, [], content=content)
             return {"messages": [stripped_msg]}
 
         if decision.action == "warn":
diff --git a/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py
index 19d25f1a6..767840c73 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py
@@ -201,7 +201,8 @@ class SafetyFinishReasonMiddleware(AgentMiddleware[AgentState]):
         new_content = self._append_user_message(message.content, explanation)
 
         # clone_ai_message_with_tool_calls handles structured tool_calls,
-        # raw additional_kwargs.tool_calls, and function_call in one shot.
+        # raw additional_kwargs.tool_calls, function_call, and provider
+        # tool-call content blocks in one shot.
         # It only rewrites finish_reason when the old value was "tool_calls",
         # which is not our case — content_filter / refusal / SAFETY stay put
         # so downstream SSE / converters keep seeing the real provider reason.
diff --git a/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py
index 0c867de81..9a5154f83 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py
@@ -9,7 +9,8 @@ Detection strategy:
      history.
   2. If the highest fraction (input, output, or total) >= warn_threshold,
      queue a warning.
-  3. If the highest fraction >= hard_stop_threshold, strip tool_calls.
+  3. If the highest fraction >= hard_stop_threshold, strip tool calls from
+     every provider surface (structured, raw, and content blocks).
 Warning injection uses the deferred pattern:
   - after_model queues the warning (does NOT mutate state).
   - wrap_model_call injects it as a HumanMessage at the next model call.
@@ -55,6 +56,7 @@ from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.runtime import Runtime
 
 from deerflow.agents.middlewares._bounded_dict import BoundedDict
+from deerflow.agents.middlewares.tool_call_metadata import clone_ai_message_with_tool_calls
 from deerflow.config.token_budget_config import TokenBudgetConfig
 
 logger = logging.getLogger(__name__)
@@ -235,19 +237,7 @@ class TokenBudgetMiddleware(AgentMiddleware[AgentState]):
 
     def _build_hard_stop_update(self, msg: AIMessage, stop_msg: str) -> dict[str, Any]:
         """Build the state update dictionary for a hard stop."""
-        updated_content = self._append_text(msg.content, stop_msg)
-        kwargs = dict(msg.additional_kwargs) if msg.additional_kwargs else {}
-        if "tool_calls" in kwargs:
-            del kwargs["tool_calls"]
-        if "function_call" in kwargs:
-            del kwargs["function_call"]
-
-        response_metadata = dict(getattr(msg, "response_metadata", {}) or {})
-
-        if response_metadata.get("finish_reason") == "tool_calls":
-            response_metadata["finish_reason"] = "stop"
-
-        stopped_msg = msg.model_copy(update={"content": updated_content, "tool_calls": [], "additional_kwargs": kwargs, "response_metadata": response_metadata})
+        stopped_msg = clone_ai_message_with_tool_calls(msg, [], content=self._append_text(msg.content, stop_msg))
         return {"messages": [stopped_msg]}
 
     def _apply(self, state: AgentState, runtime: Runtime) -> dict | None:
diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_call_metadata.py b/backend/packages/harness/deerflow/agents/middlewares/tool_call_metadata.py
index f0845622b..0ad4a7e62 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/tool_call_metadata.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/tool_call_metadata.py
@@ -2,10 +2,29 @@
 
 from __future__ import annotations
 
+from collections import Counter
 from typing import Any
 
 from langchain_core.messages import AIMessage
 
+# Content block types that provider adapters re-serialize as tool calls,
+# mapped to the keys (in priority order) holding the id that pairs with
+# ``AIMessage.tool_calls[*]["id"]``. These mirror the content surfaces in
+# ``tool_call_args``. Anthropic re-emits a ``tool_use`` block whose id is
+# absent from ``tool_calls``; the OpenAI Responses input builder re-emits
+# every ``function_call``/``custom_tool_call`` block, whose ``id`` is the
+# ``fc_…`` item id and whose ``call_id`` is the tool-call id; Google GenAI
+# ``function_call`` blocks carry the tool-call id in ``id``, or none at all;
+# LangChain v1 ``tool_call``/``tool_call_chunk`` blocks convert back to any of
+# these shapes.
+_CONTENT_TOOL_CALL_ID_KEYS: dict[str, tuple[str, ...]] = {
+    "tool_use": ("id",),
+    "function_call": ("call_id", "id"),
+    "custom_tool_call": ("call_id",),
+    "tool_call": ("id",),
+    "tool_call_chunk": ("id",),
+}
+
 
 def _raw_tool_call_id(raw_tool_call: Any) -> str | None:
     if not isinstance(raw_tool_call, dict):
@@ -15,18 +34,78 @@ def _raw_tool_call_id(raw_tool_call: Any) -> str | None:
     return raw_id if isinstance(raw_id, str) and raw_id else None
 
 
+def _content_block_call_id(block: dict[str, Any], id_keys: tuple[str, ...]) -> str | None:
+    for key in id_keys:
+        value = block.get(key)
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _tool_call_block_id_keys(block: Any) -> tuple[str, ...] | None:
+    block_type = block.get("type") if isinstance(block, dict) else None
+    return _CONTENT_TOOL_CALL_ID_KEYS.get(block_type) if isinstance(block_type, str) else None
+
+
+def _sync_content_tool_call_blocks(content: Any, retained_calls: list[dict[str, Any]]) -> Any:
+    """Drop content tool-call blocks whose call is no longer on the message.
+
+    A block left behind is sent as a tool call with no matching tool result,
+    which Anthropic and the OpenAI Responses API reject on every later request
+    for the thread. Blocks without an id (Gemini-style ``function_call``) pair
+    by name, in order, with the retained calls that no id-bearing block already
+    matched. Returns ``content`` itself when no block is dropped.
+    """
+    if not isinstance(content, list):
+        return content
+
+    retained_ids = {call["id"] for call in retained_calls if isinstance(call.get("id"), str) and call["id"]}
+    # (block, is a tool-call block, the call id it carries or None when id-less)
+    entries: list[tuple[Any, bool, str | None]] = []
+    for block in content:
+        id_keys = _tool_call_block_id_keys(block)
+        entries.append((block, id_keys is not None, _content_block_call_id(block, id_keys) if id_keys is not None else None))
+    matched_ids = {call_id for _, _, call_id in entries if call_id in retained_ids}
+    idless_budget = Counter(call["name"] for call in retained_calls if isinstance(call.get("name"), str) and not (isinstance(call.get("id"), str) and call["id"] in matched_ids))
+
+    synced: list[Any] = []
+    for block, is_tool_call_block, call_id in entries:
+        if not is_tool_call_block:
+            synced.append(block)
+            continue
+        if call_id is not None:
+            if call_id in retained_ids:
+                synced.append(block)
+            continue
+        name = block.get("name")
+        if isinstance(name, str) and idless_budget[name] > 0:
+            idless_budget[name] -= 1
+            synced.append(block)
+    return content if len(synced) == len(content) else synced
+
+
 def clone_ai_message_with_tool_calls(
     message: AIMessage,
     tool_calls: list[dict[str, Any]],
     *,
     content: Any | None = None,
 ) -> AIMessage:
-    """Clone an AIMessage while keeping raw provider tool-call metadata in sync."""
+    """Clone an AIMessage while keeping every provider tool-call surface in sync.
+
+    Besides ``tool_calls``, the raw ``additional_kwargs`` payload and the
+    provider tool-call blocks in ``content`` (including a caller-supplied
+    ``content``) are trimmed to the retained calls. Blocks for calls still on
+    ``invalid_tool_calls`` are kept, because ``DanglingToolCallMiddleware``
+    answers those calls with placeholder tool results.
+    """
     kept_ids = {tc["id"] for tc in tool_calls if isinstance(tc.get("id"), str) and tc["id"]}
 
     update: dict[str, Any] = {"tool_calls": tool_calls}
-    if content is not None:
-        update["content"] = content
+    invalid_tool_calls = [tc for tc in (getattr(message, "invalid_tool_calls", None) or []) if isinstance(tc, dict)]
+    source_content = content if content is not None else message.content
+    synced_content = _sync_content_tool_call_blocks(source_content, [*tool_calls, *invalid_tool_calls])
+    if content is not None or synced_content is not message.content:
+        update["content"] = synced_content
 
     additional_kwargs = dict(getattr(message, "additional_kwargs", {}) or {})
     raw_tool_calls = additional_kwargs.get("tool_calls")
diff --git a/backend/tests/test_clarification_middleware.py b/backend/tests/test_clarification_middleware.py
index 97f24a227..0c43ff77b 100644
--- a/backend/tests/test_clarification_middleware.py
+++ b/backend/tests/test_clarification_middleware.py
@@ -1039,6 +1039,41 @@ class TestDropParallelSiblingTools:
         ]
         assert [tc["name"] for tc in patched.tool_calls] == ["ask_clarification"]
 
+    def test_keeps_openai_responses_clarification_block_matched_by_call_id(self, middleware):
+        # Responses blocks carry the fc_ item id in ``id`` and the tool-call id
+        # in ``call_id``; matching on ``id`` would drop the kept call's block.
+        clarify = {"type": "function_call", "id": "fc_1", "call_id": "c1", "name": "ask_clarification", "arguments": "{}"}
+        sibling = {"type": "function_call", "id": "fc_2", "call_id": "b1", "name": "bash", "arguments": "{}"}
+        msg = self._ai(
+            [
+                {"id": "c1", "name": "ask_clarification", "args": {"question": "q?"}},
+                {"id": "b1", "name": "bash", "args": {"command": "ls"}},
+            ],
+            content=[clarify, sibling],
+        )
+        patched = middleware.after_model({"messages": [msg]}, self._runtime())["messages"][0]
+        assert patched.content == [clarify]
+
+    def test_keeps_content_block_for_invalid_sibling_left_on_message(self, middleware):
+        # The invalid sibling stays on invalid_tool_calls and is answered by
+        # DanglingToolCallMiddleware, so its block must stay to pair with it.
+        content = [
+            {"type": "tool_use", "id": "c1", "name": "ask_clarification", "input": {"question": "q?"}},
+            {"type": "tool_use", "id": "b1", "name": "bash", "input": {"command": "ls"}},
+            {"type": "tool_use", "id": "w1", "name": "write_file", "input": {}},
+        ]
+        msg = self._ai(
+            [
+                {"id": "c1", "name": "ask_clarification", "args": {"question": "q?"}},
+                {"id": "b1", "name": "bash", "args": {"command": "ls"}},
+            ],
+            content=content,
+            invalid_tool_calls=[{"id": "w1", "name": "write_file", "args": "{", "error": "parse", "type": "invalid_tool_call"}],
+        )
+        patched = middleware.after_model({"messages": [msg]}, self._runtime())["messages"][0]
+        assert [block["id"] for block in patched.content] == ["c1", "w1"]
+        assert [tc["id"] for tc in patched.invalid_tool_calls] == ["w1"]
+
     def test_aafter_model_matches_sync(self, middleware):
         import asyncio
 
diff --git a/backend/tests/test_loop_detection_middleware.py b/backend/tests/test_loop_detection_middleware.py
index 6806e6856..827f26b8d 100644
--- a/backend/tests/test_loop_detection_middleware.py
+++ b/backend/tests/test_loop_detection_middleware.py
@@ -1381,6 +1381,24 @@ class TestHardStopWithListContent:
         assert msg.content[2]["type"] == "text"
         assert _HARD_STOP_MSG in msg.content[2]["text"]
 
+    def test_hard_stop_drops_provider_tool_use_blocks(self):
+        """A stripped call's Anthropic tool_use block must not outlive it in content."""
+        mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=4)
+        runtime = _make_runtime()
+        call = [_bash_call("ls")]
+        list_content = [
+            {"type": "text", "text": "I'll run ls"},
+            {"type": "tool_use", "id": "call_ls", "name": "bash", "input": {"command": "ls"}},
+        ]
+
+        for _ in range(3):
+            mw._apply(_make_state(tool_calls=call, content=list_content), runtime)
+        result = mw._apply(_make_state(tool_calls=call, content=list_content), runtime)
+
+        msg = result["messages"][0]
+        assert [block["type"] for block in msg.content] == ["text", "text"]
+        assert _HARD_STOP_MSG in msg.content[-1]["text"]
+
     def test_hard_stop_with_none_content(self):
         """Hard stop on None content should produce a plain string."""
         mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=4)
diff --git a/backend/tests/test_safety_finish_reason_middleware.py b/backend/tests/test_safety_finish_reason_middleware.py
index 2b9618aad..622b372af 100644
--- a/backend/tests/test_safety_finish_reason_middleware.py
+++ b/backend/tests/test_safety_finish_reason_middleware.py
@@ -422,6 +422,27 @@ class TestMessageRewrite:
         assert patched.content[-1]["type"] == "text"
         assert "safety-related signal" in patched.content[-1]["text"]
 
+    def test_drops_suppressed_calls_provider_tool_use_blocks(self):
+        mw = SafetyFinishReasonMiddleware()
+        call = _write_call()
+        state = {
+            "messages": [
+                _ai(
+                    content=[
+                        {"type": "text", "text": "partial answer"},
+                        {"type": "tool_use", "id": call["id"], "name": call["name"], "input": call["args"]},
+                    ],
+                    tool_calls=[call],
+                    response_metadata={"stop_reason": "refusal"},
+                )
+            ]
+        }
+
+        patched = mw.after_model(state, _runtime())["messages"][0]
+
+        assert [block["type"] for block in patched.content] == ["text", "text"]
+        assert "safety-related signal" in patched.content[-1]["text"]
+
     def test_idempotent_on_already_cleared_message(self):
         # Re-running the middleware on a message we already cleared must not
         # re-trigger (tool_calls is now empty → fast passthrough).
diff --git a/backend/tests/test_subagent_limit_middleware.py b/backend/tests/test_subagent_limit_middleware.py
index 2940824b0..2ce60c1f6 100644
--- a/backend/tests/test_subagent_limit_middleware.py
+++ b/backend/tests/test_subagent_limit_middleware.py
@@ -172,6 +172,35 @@ class TestTruncateTaskCalls:
         assert [tc["id"] for tc in updated_msg.additional_kwargs["tool_calls"]] == ["t1", "t2"]
         assert updated_msg.response_metadata["finish_reason"] == "tool_calls"
 
+    def test_truncation_syncs_provider_tool_use_content_blocks(self):
+        # The tools node answers only the kept calls, so a dropped call's
+        # Anthropic tool_use block would reach the next model request unpaired.
+        mw = SubagentLimitMiddleware(max_concurrent=2)
+        msg = AIMessage(
+            content=[{"type": "tool_use", "id": call_id, "name": "task", "input": {"prompt": "p"}} for call_id in ("t1", "t2", "t3")],
+            tool_calls=[_task_call("t1"), _task_call("t2"), _task_call("t3")],
+        )
+
+        result = mw.after_model({"messages": [msg]}, _make_runtime())
+
+        updated_msg = result["messages"][0]
+        assert [block["id"] for block in updated_msg.content] == ["t1", "t2"]
+
+    def test_total_limit_reached_drops_provider_tool_use_content_blocks(self):
+        mw = SubagentLimitMiddleware(max_concurrent=3, max_total=1)
+        msg = AIMessage(
+            content=[{"type": "tool_use", "id": "t2", "name": "task", "input": {"prompt": "p"}}],
+            tool_calls=[_task_call("t2")],
+        )
+        state = {"messages": [msg], "delegations": [_delegation("t1", run_id="run-1")]}
+
+        result = mw.after_model(state, _make_runtime())
+
+        updated_msg = result["messages"][0]
+        assert updated_msg.tool_calls == []
+        assert [block["type"] for block in updated_msg.content] == ["text"]
+        assert "subagent delegation limit" in updated_msg.content[0]["text"]
+
     def test_total_limit_counts_prior_delegations(self):
         mw = SubagentLimitMiddleware(max_concurrent=3, max_total=4)
         msg = AIMessage(
diff --git a/backend/tests/test_token_budget_middleware.py b/backend/tests/test_token_budget_middleware.py
index b302c4d16..f80ae708b 100644
--- a/backend/tests/test_token_budget_middleware.py
+++ b/backend/tests/test_token_budget_middleware.py
@@ -10,6 +10,7 @@ from langgraph.checkpoint.memory import InMemorySaver
 
 from deerflow.agents.middlewares.token_budget_middleware import TokenBudgetMiddleware
 from deerflow.config.token_budget_config import TokenBudgetConfig
+from deerflow.models.claude_provider import ClaudeChatModel
 
 
 def _make_runtime(thread_id="test-thread", run_id="test-run"):
@@ -219,6 +220,23 @@ class TestTokenBudgetHardStop:
         assert "Thinking" in msgs[0].content
         assert "TOKEN BUDGET EXCEEDED" in msgs[0].content
 
+    def test_hard_stop_drops_provider_tool_call_content_blocks(self):
+        # Anthropic keeps tool_use blocks in content; one left behind without a
+        # tool_result makes every later request on the thread fail with a 400.
+        config = TokenBudgetConfig(max_tokens=100000, hard_stop_threshold=1.0, enabled=True)
+        mw = TokenBudgetMiddleware.from_config(config)
+        tool_calls = [{"name": "bash", "args": {"command": "ls"}, "id": "toolu_1"}]
+        content = [
+            {"type": "text", "text": "Listing"},
+            {"type": "tool_use", "id": "toolu_1", "name": "bash", "input": {"command": "ls"}},
+        ]
+
+        res = mw._apply(_make_state_with_usage(total=105000, tool_calls=tool_calls, content=content), _make_runtime())
+
+        stopped = res["messages"][0]
+        assert [block["type"] for block in stopped.content] == ["text", "text"]
+        assert "TOKEN BUDGET EXCEEDED" in stopped.content[-1]["text"]
+
     def test_hard_stop_stamps_token_capped_stop_reason_consumed_once(self):
         """#3875 Phase 2: a hard-stop stamps ``token_capped`` on a per-run
         accessor the executor reads post-run. It pops on read so a second read
@@ -350,6 +368,42 @@ class TestTokenBudgetAgentGraph:
         graph.invoke({"messages": [HumanMessage("next question")]}, config=config, context={"thread_id": "goal-thread", "run_id": "run-2"})
         assert executed == ["a", "b", "d"]
 
+    def test_checkpointed_hard_stop_leaves_next_anthropic_turn_well_formed(self):
+        """The stopped message is checkpointed and replayed; its tool_use must not reach the next request unpaired."""
+        executed: list[str] = []
+
+        @as_tool
+        def bash(command: str) -> str:
+            """Run a fake shell command."""
+            executed.append(command)
+            return "ok"
+
+        over_budget_call = AIMessage(
+            content=[
+                {"type": "text", "text": "Listing files."},
+                {"type": "tool_use", "id": "toolu_ls", "name": "bash", "input": {"command": "ls"}},
+            ],
+            id="ai-over-budget",
+            tool_calls=[{"name": "bash", "id": "toolu_ls", "args": {"command": "ls"}}],
+            usage_metadata={"input_tokens": 12_000, "output_tokens": 0, "total_tokens": 12_000},
+        )
+        answer = AIMessage(content="second answer", id="ai-answer", usage_metadata={"input_tokens": 100, "output_tokens": 0, "total_tokens": 100})
+        model = _RecordingToolCallingFakeModel(responses=[over_budget_call, answer])
+        mw = TokenBudgetMiddleware(TokenBudgetConfig(enabled=True, max_tokens=10_000))
+        graph = create_agent(model=model, tools=[bash], middleware=[mw], checkpointer=InMemorySaver())
+        config = {"configurable": {"thread_id": "stopped-thread"}}
+
+        graph.invoke({"messages": [HumanMessage("list files")]}, config=config, context={"thread_id": "stopped-thread", "run_id": "run-1"})
+        graph.invoke({"messages": [HumanMessage("continue")]}, config=config, context={"thread_id": "stopped-thread", "run_id": "run-2"})
+
+        assert executed == []
+        payload = ClaudeChatModel(model="claude-sonnet-4-5", anthropic_api_key="sk-ant-offline")._get_request_payload(model.requests[1])
+        turns = [turn["content"] if isinstance(turn["content"], list) else [] for turn in payload["messages"]]
+        for index, blocks in enumerate(turns):
+            tool_use_ids = {block["id"] for block in blocks if block["type"] == "tool_use"}
+            following = turns[index + 1] if index + 1 < len(turns) else []
+            assert tool_use_ids <= {block["tool_use_id"] for block in following if block["type"] == "tool_result"}
+
     @pytest.mark.parametrize("context", [{"thread_id": "no-run-id"}, {"thread_id": "no-run-id", "run_id": None}])
     def test_invocation_without_run_id_keeps_one_budget_across_graph_nodes(self, context):
         """LangGraph hands each node its own Runtime, so an invocation without a run_id can't be keyed by id(runtime)."""
diff --git a/backend/tests/test_tool_call_metadata.py b/backend/tests/test_tool_call_metadata.py
new file mode 100644
index 000000000..1e6e8f88d
--- /dev/null
+++ b/backend/tests/test_tool_call_metadata.py
@@ -0,0 +1,194 @@
+"""Tests for keeping AIMessage tool-call surfaces consistent when calls are removed.
+
+Provider adapters re-serialize tool calls from ``AIMessage.content`` blocks, not
+only from ``tool_calls``. A guard that strips a call from ``tool_calls`` but
+leaves its content block behind sends the provider a tool call with no matching
+tool result, which Anthropic and the OpenAI Responses API reject on every later
+request for that thread.
+"""
+
+import pytest
+from langchain_anthropic import ChatAnthropic
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_openai import ChatOpenAI
+
+from deerflow.agents.middlewares.tool_call_metadata import clone_ai_message_with_tool_calls
+from deerflow.models.claude_provider import ClaudeChatModel
+
+# DeerFlow deployments use ClaudeChatModel, which post-processes the payload
+# built by the upstream ChatAnthropic formatter; pin both.
+_ANTHROPIC_MODEL_CLASSES = pytest.mark.parametrize("model_class", [ChatAnthropic, ClaudeChatModel], ids=["ChatAnthropic", "ClaudeChatModel"])
+
+
+def _call(call_id: str, name: str = "bash") -> dict:
+    return {"id": call_id, "name": name, "args": {}}
+
+
+def _tool_use(call_id: str, name: str = "bash") -> dict:
+    return {"type": "tool_use", "id": call_id, "name": name, "input": {}}
+
+
+class TestContentToolCallBlockSync:
+    def test_drops_anthropic_tool_use_blocks_for_removed_calls(self):
+        message = AIMessage(
+            content=[{"type": "text", "text": "running"}, _tool_use("a"), _tool_use("b")],
+            tool_calls=[_call("a"), _call("b")],
+        )
+
+        cloned = clone_ai_message_with_tool_calls(message, [_call("a")])
+
+        assert cloned.content == [{"type": "text", "text": "running"}, _tool_use("a")]
+        assert [tc["id"] for tc in cloned.tool_calls] == ["a"]
+
+    def test_clearing_every_call_keeps_non_tool_call_blocks(self):
+        thinking = {"type": "thinking", "thinking": "hmm", "signature": "sig"}
+        server_tool = {"type": "server_tool_use", "id": "srvtoolu_1", "name": "web_search", "input": {}}
+        message = AIMessage(content=[thinking, server_tool, _tool_use("a")], tool_calls=[_call("a")])
+
+        cloned = clone_ai_message_with_tool_calls(message, [])
+
+        assert cloned.content == [thinking, server_tool]
+
+    def test_filters_caller_supplied_content(self):
+        message = AIMessage(content=[_tool_use("a")], tool_calls=[_call("a")])
+        stop_note = {"type": "text", "text": "stopped"}
+
+        cloned = clone_ai_message_with_tool_calls(message, [], content=[*message.content, stop_note])
+
+        assert cloned.content == [stop_note]
+
+    def test_openai_responses_blocks_match_by_call_id_not_item_id(self):
+        kept = {"type": "function_call", "id": "fc_1", "call_id": "a", "name": "bash", "arguments": "{}"}
+        dropped = {"type": "function_call", "id": "fc_2", "call_id": "b", "name": "bash", "arguments": "{}"}
+        dropped_custom = {"type": "custom_tool_call", "id": "ctc_3", "call_id": "c", "name": "patch", "input": "x"}
+        message = AIMessage(
+            content=[kept, dropped, dropped_custom],
+            tool_calls=[_call("a"), _call("b"), _call("c", "patch")],
+        )
+
+        cloned = clone_ai_message_with_tool_calls(message, [_call("a")])
+
+        assert cloned.content == [kept]
+
+    def test_langchain_standard_tool_call_blocks_match_by_id(self):
+        kept = {"type": "tool_call", "id": "a", "name": "bash", "args": {}}
+        dropped = {"type": "tool_call", "id": "b", "name": "bash", "args": {}}
+        dropped_chunk = {"type": "tool_call_chunk", "id": "c", "name": "bash", "args": "{}", "index": 2}
+        message = AIMessage(content=[kept, dropped, dropped_chunk], tool_calls=[_call("a"), _call("b"), _call("c")])
+
+        cloned = clone_ai_message_with_tool_calls(message, [_call("a")])
+
+        assert cloned.content == [kept]
+
+    def test_google_genai_function_call_blocks_match_by_id(self):
+        # Keep the second of two same-named calls: an id match keeps its own
+        # block, where falling back to name order would keep the first one.
+        dropped = {"type": "function_call", "id": "a", "name": "bash", "args": {}}
+        kept = {"type": "function_call", "id": "b", "name": "bash", "args": {}}
+        message = AIMessage(content=[dropped, kept], tool_calls=[_call("a"), _call("b")])
+
+        cloned = clone_ai_message_with_tool_calls(message, [_call("b")])
+
+        assert cloned.content == [kept]
+
+    def test_idless_blocks_keep_only_as_many_per_name_as_retained_calls(self):
+        # Gemini-style blocks carry no id, so they pair with retained calls by
+        # name, in order: truncating four same-named calls to three keeps three.
+        blocks = [{"type": "function_call", "name": "task", "args": {"n": n}} for n in range(4)]
+        calls = [_call(f"t{n}", "task") for n in range(4)]
+        message = AIMessage(content=list(blocks), tool_calls=calls)
+
+        cloned = clone_ai_message_with_tool_calls(message, calls[:3])
+
+        assert cloned.content == blocks[:3]
+
+    def test_idless_budget_skips_calls_already_paired_by_id(self):
+        # A retained call whose own id-bearing block matched must not also let
+        # a same-named id-less block survive on the name budget.
+        with_id = {"type": "function_call", "id": "a", "name": "bash", "args": {}}
+        idless = {"type": "function_call", "name": "bash", "args": {}}
+        message = AIMessage(content=[with_id, idless], tool_calls=[_call("a"), _call("b")])
+
+        assert clone_ai_message_with_tool_calls(message, [_call("a")]).content == [with_id]
+        assert clone_ai_message_with_tool_calls(message, [_call("a"), _call("b")]).content is message.content
+
+    def test_keeps_blocks_for_calls_that_remain_invalid_tool_calls(self):
+        # DanglingToolCallMiddleware answers invalid_tool_calls with a placeholder
+        # ToolMessage, so their content block must stay to pair with it.
+        message = AIMessage(
+            content=[_tool_use("bad"), _tool_use("a")],
+            tool_calls=[_call("a")],
+            invalid_tool_calls=[{"type": "invalid_tool_call", "id": "bad", "name": "bash", "args": "{", "error": "parse"}],
+        )
+
+        cloned = clone_ai_message_with_tool_calls(message, [])
+
+        assert cloned.content == [_tool_use("bad")]
+        assert [tc["id"] for tc in cloned.invalid_tool_calls] == ["bad"]
+
+    def test_leaves_content_untouched_when_every_block_is_still_paired(self):
+        content = [{"type": "text", "text": "hi"}, _tool_use("a")]
+        message = AIMessage(content=content, tool_calls=[_call("a")])
+
+        cloned = clone_ai_message_with_tool_calls(message, [_call("a")])
+
+        assert cloned.content is message.content
+
+    def test_string_content_is_unchanged(self):
+        message = AIMessage(content="plain", tool_calls=[_call("a")])
+
+        assert clone_ai_message_with_tool_calls(message, []).content == "plain"
+
+
+def _anthropic_payload_turns(model_class: type[ChatAnthropic], messages: list) -> list[tuple[str, list[str]]]:
+    payload = model_class(model="claude-sonnet-4-5", anthropic_api_key="sk-ant-offline")._get_request_payload(messages)
+    turns = []
+    for turn in payload["messages"]:
+        blocks = turn["content"] if isinstance(turn["content"], list) else [{"type": "text"}]
+        turns.append((turn["role"], [block.get("id") or block.get("tool_use_id") or block["type"] for block in blocks if block["type"] in ("tool_use", "tool_result")]))
+    return turns
+
+
+class TestProviderRequestContract:
+    """Drive the real provider request builders offline; no network is used."""
+
+    @_ANTHROPIC_MODEL_CLASSES
+    def test_anthropic_request_pairs_every_tool_use_after_calls_are_cleared(self, model_class):
+        message = AIMessage(
+            content=[{"type": "text", "text": "reading"}, _tool_use("toolu_a")],
+            tool_calls=[_call("toolu_a")],
+            response_metadata={"model_provider": "anthropic"},
+        )
+        stopped = clone_ai_message_with_tool_calls(message, [], content=[*message.content, {"type": "text", "text": "stopped"}])
+
+        turns = _anthropic_payload_turns(model_class, [HumanMessage("hi"), stopped, HumanMessage("continue")])
+
+        assert turns == [("user", []), ("assistant", []), ("user", [])]
+
+    @_ANTHROPIC_MODEL_CLASSES
+    def test_anthropic_request_pairs_every_tool_use_after_calls_are_truncated(self, model_class):
+        calls = [_call(f"toolu_{n}", "task") for n in range(3)]
+        message = AIMessage(
+            content=[_tool_use(call["id"], "task") for call in calls],
+            tool_calls=calls,
+            response_metadata={"model_provider": "anthropic"},
+        )
+        truncated = clone_ai_message_with_tool_calls(message, calls[:2])
+        results = [ToolMessage(content="ok", tool_call_id=call["id"]) for call in truncated.tool_calls]
+
+        turns = _anthropic_payload_turns(model_class, [HumanMessage("go"), truncated, *results])
+
+        assert turns == [("user", []), ("assistant", ["toolu_0", "toolu_1"]), ("user", ["toolu_0", "toolu_1"])]
+
+    def test_openai_responses_request_drops_function_call_after_calls_are_cleared(self):
+        message = AIMessage(
+            content=[{"type": "function_call", "id": "fc_1", "call_id": "call_a", "name": "bash", "arguments": "{}"}],
+            tool_calls=[_call("call_a")],
+            response_metadata={"model_provider": "openai"},
+        )
+        stopped = clone_ai_message_with_tool_calls(message, [], content=[*message.content, {"type": "text", "text": "stopped"}])
+        llm = ChatOpenAI(model="gpt-5", api_key="sk-offline", use_responses_api=True, output_version="responses/v1")
+
+        payload = llm._get_request_payload([HumanMessage("hi"), stopped, HumanMessage("continue")])
+
+        assert [item for item in payload["input"] if item.get("type") == "function_call"] == []