fix(clarification): keep each question when a provider reuses a call ID (#7762)

* fix(clarification): keep each question when a provider reuses a call ID

The clarification ToolMessage ID, which is also the human-input
request_id, was clarification:{tool_call_id}. Providers that number
calls per response reuse call_0 across turns, so a later question took
the earlier one's ID: add_messages replaced the earlier question in
place and left the new call without a result.

Add a digest of the issuing AIMessage ID. A retry of the same call comes
from the same AIMessage and keeps its ID; without that message in state
the ID keeps its previous form.

* docs(changelog): reference #7762 for the clarification call ID fix

CHANGELOG.md

@@ -19,6 +19,14 @@ This section accumulates work toward the **2.3.0** milestone
 
 ### Fixed
 
+- **agents:** A clarification no longer overwrites an earlier question when the
+  provider reuses a tool-call ID across turns (for example `call_0` on every
+  response). Its message ID, which is also the human-input `request_id`, named
+  only the call ID, so the later question replaced the earlier one in place and
+  the later call was left without a result. The ID now also names the
+  AIMessage that issued the call; a retry of the same call keeps its ID.
+  ([#7762])
+
 - **projects:** The trash retention sweep no longer deletes live document
   originals when the data directory, `users/`, a user's directory or a
   user's `projects/` is a symlink. It protected each document by its resolved
@@ -15957,3 +15965,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7727]: https://github.com/bytedance/deer-flow/pull/7727
 [#7729]: https://github.com/bytedance/deer-flow/pull/7729
 [#7732]: https://github.com/bytedance/deer-flow/pull/7732
+[#7762]: https://github.com/bytedance/deer-flow/pull/7762

CHANGELOG_zh.md

@@ -20,6 +20,11 @@
 
 ### 修复
 
+- **智能体：** 当模型提供方在不同轮次复用同一个工具调用 ID（例如每次响应都是 `call_0`）时，
+  澄清提问不再覆盖之前的问题。此前其消息 ID（同时也是人工输入的 `request_id`）只包含调用 ID，
+  后一个问题会原地替换前一个问题，后一次调用也因此缺少结果。现在 ID 还会包含发起该调用的
+  AIMessage；同一调用的重试仍保持原 ID。([#7762])
+
 - **项目：** 当数据目录、`users/`、某个用户目录或用户的 `projects/` 是符号链接时，
   回收站保留期清扫不再删除仍在使用的文档原件。此前它按解析后的路径保护每个文档，
   却遍历未解析的目录树，导致没有任何文档能匹配上，所有超过 24 小时的原件都被当作
@@ -12719,3 +12724,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7727]: https://github.com/bytedance/deer-flow/pull/7727
 [#7729]: https://github.com/bytedance/deer-flow/pull/7729
 [#7732]: https://github.com/bytedance/deer-flow/pull/7732
+[#7762]: https://github.com/bytedance/deer-flow/pull/7762

backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py

@@ -3,7 +3,7 @@
 import json
 import logging
 import re
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from hashlib import sha256
 from typing import Any, override
 
@@ -96,12 +96,42 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]):
 
     state_schema = ClarificationMiddlewareState
 
-    def _stable_message_id(self, tool_call_id: str, formatted_message: str) -> str:
-        """Build a deterministic message ID so retried clarification calls replace, not append."""
-        if tool_call_id:
-            return f"clarification:{tool_call_id}"
-        digest = sha256(formatted_message.encode("utf-8")).hexdigest()[:16]
-        return f"clarification:{digest}"
+    def _stable_message_id(self, tool_call_id: str, formatted_message: str, issuing_message_id: str | None = None) -> str:
+        """Build a deterministic message ID so retried clarification calls replace, not append.
+
+        The ID doubles as the human-input ``request_id``. Providers may reuse
+        call IDs across turns, so the ID also names the AIMessage that issued
+        the call: a later question must not take an earlier one's ID, or
+        ``add_messages`` would replace the earlier question in place and leave
+        the new call without a result. A retry of the same call comes from the
+        same AIMessage and keeps its ID.
+        """
+        base = tool_call_id or sha256(formatted_message.encode("utf-8")).hexdigest()[:16]
+        if not issuing_message_id:
+            return f"clarification:{base}"
+        # A digest rather than the raw AIMessage ID keeps the request ID short
+        # enough for keys built from it, such as the 128-character
+        # notification source key.
+        return f"clarification:{base}:{sha256(issuing_message_id.encode('utf-8')).hexdigest()[:12]}"
+
+    @staticmethod
+    def _issuing_message_id(request: ToolCallRequest) -> str | None:
+        """ID of the latest AIMessage in state that carries this tool call, if any.
+
+        Message IDs are unique within state, since ``add_messages`` merges
+        messages that share one, so this identifies the call's turn.
+        """
+        state = getattr(request, "state", None)
+        if not isinstance(state, Mapping):
+            return None
+        messages = state.get("messages")
+        if not isinstance(messages, list):
+            return None
+        tool_call_id = request.tool_call.get("id") or ""
+        for message in reversed(messages):
+            if isinstance(message, AIMessage) and any((tool_call.get("id") or "") == tool_call_id for tool_call in message.tool_calls):
+                return message.id or None
+        return None
 
     def _normalize_options(self, raw_options: Any) -> list[str]:
         """Normalize tool-provided options into displayable string values."""
@@ -457,7 +487,7 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]):
         tool_call_id = request.tool_call.get("id", "")
         logger.info("ask_clarification suppressed by run interaction policy; applying unattended risk guidance")
         return ToolMessage(
-            id=self._stable_message_id(tool_call_id, "proceed-without-clarification"),
+            id=self._stable_message_id(tool_call_id, "proceed-without-clarification", self._issuing_message_id(request)),
             content=(
                 "Clarification is disabled in this context — the human is not present "
                 "to answer synchronously. Do not ask for confirmation or wait for a human response. "
@@ -497,7 +527,7 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]):
         # Get the tool call ID
         tool_call_id = request.tool_call.get("id", "")
 
-        request_id = self._stable_message_id(tool_call_id, formatted_message)
+        request_id = self._stable_message_id(tool_call_id, formatted_message, self._issuing_message_id(request))
         human_input_payload = self._build_human_input_payload(args, tool_call_id=tool_call_id, request_id=request_id, fields=fields)
 
         # Create a ToolMessage with the formatted question

backend/tests/test_clarification_middleware.py

@@ -4,6 +4,7 @@ import json
 from types import SimpleNamespace
 
 import pytest
+from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.graph.message import add_messages
 
 from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
@@ -674,6 +675,54 @@ class TestClarificationCommandIdempotency:
         assert merged[0].content == first_message.content
         assert merged[0].artifact == first_message.artifact
 
+    @staticmethod
+    def _turn(ai_message_id, call_id, question):
+        return AIMessage(
+            id=ai_message_id,
+            content="",
+            tool_calls=[{"name": "ask_clarification", "id": call_id, "args": {"question": question, "clarification_type": "missing_info"}}],
+        )
+
+    @staticmethod
+    def _result(middleware, messages, *, disabled):
+        call = messages[-1].tool_calls[0]
+        request = SimpleNamespace(
+            tool_call=call,
+            state={"messages": messages},
+            runtime=SimpleNamespace(context={"disable_clarification": True} if disabled else {}),
+        )
+        result = middleware.wrap_tool_call(request, lambda _req: pytest.fail("handler should not be called"))
+        return result if disabled else result.update["messages"][0]
+
+    @pytest.mark.parametrize("disabled", [False, True], ids=["interactive", "unattended"])
+    def test_retry_from_the_same_ai_message_keeps_its_message_id(self, middleware, disabled):
+        messages = [HumanMessage(content="Plan a trip"), self._turn("ai-1", "call_0", "Which city?")]
+
+        first = self._result(middleware, messages, disabled=disabled)
+        second = self._result(middleware, messages, disabled=disabled)
+
+        assert second.id == first.id
+        assert len(add_messages(add_messages(messages, [first]), [second])) == 3
+
+    @pytest.mark.parametrize("call_id", ["call_0", ""], ids=["reused-id", "missing-id"])
+    @pytest.mark.parametrize("disabled", [False, True], ids=["interactive", "unattended"])
+    def test_reused_call_id_in_a_later_turn_gets_its_own_message_id(self, middleware, call_id, disabled):
+        """Providers that number calls per response reuse ``call_0`` across turns."""
+        history = [HumanMessage(content="Plan a trip"), self._turn("ai-1", call_id, "Which city?")]
+        first = self._result(middleware, history, disabled=disabled)
+        history = add_messages(history, [first, HumanMessage(content="Seoul"), self._turn("ai-2", call_id, "Which city?")])
+
+        second = self._result(middleware, history, disabled=disabled)
+        merged = add_messages(history, [second])
+
+        assert second.id != first.id
+        # The earlier question stays in place and the new call gets its result.
+        assert [message.id for message in merged] == [merged[0].id, "ai-1", first.id, merged[3].id, "ai-2", second.id]
+        assert merged[2].content == first.content
+        if not disabled:
+            assert first.artifact["human_input"]["request_id"] == first.id
+            assert second.artifact["human_input"]["request_id"] == second.id
+
     def test_tool_message_model_dump_preserves_human_input_artifact(self, middleware):
         request = SimpleNamespace(
             tool_call={

backend/tests/test_clarification_reused_call_id_graph_integration.py

@@ -0,0 +1,124 @@
+"""Graph-level test for clarifications whose provider reuses a call ID.
+
+Some providers number tool calls per response (``call_0``), so two turns can
+issue ``ask_clarification`` under the same call ID. Each question's ToolMessage
+must keep its own ID in a real ``create_agent`` graph with a checkpointer:
+``add_messages`` replaces a message that shares an ID in place, which would
+overwrite the earlier question and leave the later call without a result.
+"""
+
+from __future__ import annotations
+
+from langchain.agents import create_agent
+from langchain_core.language_models import BaseChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langgraph.checkpoint.memory import InMemorySaver
+
+from deerflow.agents.human_input import HUMAN_INPUT_RESPONSE_KEY, extract_unanswered_human_input_request
+from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
+from deerflow.tools.builtins.clarification_tool import ask_clarification_tool
+
+_REUSED_CALL_ID = "call_0"
+
+
+def _ask(question: str) -> AIMessage:
+    return AIMessage(
+        content="",
+        tool_calls=[
+            {
+                "id": _REUSED_CALL_ID,
+                "name": "ask_clarification",
+                "args": {"question": question, "clarification_type": "missing_info"},
+            }
+        ],
+    )
+
+
+class _ScriptedModel(BaseChatModel):
+    """Replays ``replies`` in order and records the messages of each call."""
+
+    replies: list[AIMessage]
+    seen: list[list] = []
+
+    @property
+    def _llm_type(self) -> str:
+        return "fake-reused-call-id"
+
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        self.seen.append(list(messages))
+        return ChatResult(generations=[ChatGeneration(message=self.replies[len(self.seen) - 1])])
+
+    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
+        return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+def _answer(request: dict[str, str], value: str) -> HumanMessage:
+    return HumanMessage(
+        content=value,
+        additional_kwargs={
+            HUMAN_INPUT_RESPONSE_KEY: {
+                "version": 1,
+                "kind": "human_input_response",
+                "source": "ask_clarification",
+                "request_id": request["request_id"],
+                "response_kind": "text",
+                "value": value,
+            }
+        },
+    )
+
+
+def _assert_every_call_answered_in_place(messages: list) -> None:
+    """Each tool call is followed by its own result, as providers require."""
+    for index, message in enumerate(messages):
+        if isinstance(message, AIMessage) and message.tool_calls:
+            result = messages[index + 1]
+            assert isinstance(result, ToolMessage), f"call at {index} has no result: {messages}"
+            assert result.tool_call_id == _REUSED_CALL_ID
+
+
+def test_reused_call_id_keeps_each_clarification_question():
+    model = _ScriptedModel(replies=[_ask("Which city?"), _ask("How many days?"), AIMessage(content="done")], seen=[])
+    agent = create_agent(model=model, tools=[ask_clarification_tool], middleware=[ClarificationMiddleware()], checkpointer=InMemorySaver())
+    config = {"configurable": {"thread_id": "reused-call-id"}}
+
+    agent.invoke({"messages": [HumanMessage(content="Plan a trip")]}, config)
+    first = extract_unanswered_human_input_request(agent.get_state(config).values["messages"])
+    assert first is not None and first["question"] == "Which city?"
+
+    agent.invoke({"messages": [_answer(first, "Seoul")]}, config)
+    second = extract_unanswered_human_input_request(agent.get_state(config).values["messages"])
+    assert second is not None and second["question"] == "How many days?"
+    assert second["request_id"] != first["request_id"]
+
+    agent.invoke({"messages": [_answer(second, "3 days")]}, config)
+
+    messages = agent.get_state(config).values["messages"]
+    _assert_every_call_answered_in_place(messages)
+    questions = [message.artifact["human_input"]["question"] for message in messages if isinstance(message, ToolMessage)]
+    assert questions == ["Which city?", "How many days?"]
+    # The model sees both questions next to the answers given to them.
+    final_call = model.seen[-1]
+    _assert_every_call_answered_in_place(final_call)
+    assert [message.content for message in final_call if isinstance(message, HumanMessage)] == ["Plan a trip", "Seoul", "3 days"]
+
+
+def test_reused_call_id_keeps_each_unattended_clarification_result():
+    model = _ScriptedModel(replies=[_ask("Which city?"), _ask("How many days?"), AIMessage(content="done")], seen=[])
+    agent = create_agent(model=model, tools=[ask_clarification_tool], middleware=[ClarificationMiddleware()], checkpointer=InMemorySaver())
+    config = {"configurable": {"thread_id": "reused-call-id-unattended"}}
+    unattended = {"disable_clarification": True}
+
+    # ``ask_clarification`` returns directly, so each unattended run ends at
+    # its "proceed" result and the next run continues the thread.
+    for text in ["Plan a trip", "Continue", "Continue"]:
+        agent.invoke({"messages": [HumanMessage(content=text)]}, config, context=unattended)
+
+    messages = agent.get_state(config).values["messages"]
+    _assert_every_call_answered_in_place(messages)
+    assert len([message for message in messages if isinstance(message, ToolMessage)]) == 2
+    _assert_every_call_answered_in_place(model.seen[-1])