fix(channels): give channel run input messages an id (#6401)

* fix(channels): give channel run input messages an id

Channel runs sent their human message without an id. The Gateway stores
run input as sent, so the run record kept it id-less, while add_messages
gave the checkpoint copy a fresh uuid. When a web client opens the thread
while the run is still going, its reconnect hydration rebuilds the input
from both copies and matches them by id, so the user's message showed up
twice until the run finished and the view fell back to the checkpoint.

_human_input_message now stamps a uuid on every message it builds, so the
run record and the checkpoint carry the same id. It covers both channel
run paths: regular turns and the coalesced follow-up drain.

* docs(changelog): reference #6401

CHANGELOG.md

@@ -697,6 +697,13 @@ This release closes that milestone with **439 merged pull requests**.
 
 ### Fixed
 
+- **channels:** Opening an IM-channel conversation on the web while its run is still
+  going no longer shows the user's message twice. Channel run input carried no
+  message id, so the Gateway stored it id-less in the run record while the
+  checkpoint copy received a generated uuid, and the web client's reconnect
+  hydration, which matches the two copies by id, kept both until the run
+  finished. Channel human messages now carry their own id, so both copies share
+  it. ([#6401])
 - **deploy:** `make up`, `make down` and `make prod-logs` no longer stop on a
   fresh checkout with `env file .../.env not found`. `.env` and `frontend/.env` are gitignored and
   `make up` does not create them, but the production compose file required
@@ -8952,3 +8959,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6370]: https://github.com/bytedance/deer-flow/pull/6370
 [#6378]: https://github.com/bytedance/deer-flow/pull/6378
 [#6388]: https://github.com/bytedance/deer-flow/pull/6388
+[#6401]: https://github.com/bytedance/deer-flow/pull/6401

CHANGELOG_zh.md

@@ -615,6 +615,9 @@
 
 ### 修复
 
+- **渠道：** 在 IM 渠道会话的运行尚未结束时从网页打开它，用户消息不再显示两次。渠道运行输入此前不带消息 id，
+  Gateway 在运行记录中按原样保存无 id 的消息，而检查点中的副本会被分配新的 uuid；网页客户端重连时按 id
+  合并这两份副本，因此在运行结束前两条都会显示。现在渠道发送的人类消息自带 id，两份副本共享同一 id。([#6401])
 - **部署：** 在全新检出的仓库上，`make up`、`make down` 与 `make prod-logs` 
   不再因 `env file .../.env not found` 而中止。`.env` 与
   `frontend/.env` 已被 gitignore，`make up` 也不会创建它们，但生产 compose 文件此前要求两者都存在。
@@ -7389,3 +7392,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6370]: https://github.com/bytedance/deer-flow/pull/6370
 [#6378]: https://github.com/bytedance/deer-flow/pull/6378
 [#6388]: https://github.com/bytedance/deer-flow/pull/6388
+[#6401]: https://github.com/bytedance/deer-flow/pull/6401

backend/app/channels/manager.py

@@ -9,6 +9,7 @@ import mimetypes
 import re
 import stat
 import time
+import uuid
 from collections import OrderedDict
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
@@ -911,7 +912,10 @@ def _unknown_command_reply(command: str | None = None) -> str:
 
 
 def _human_input_message(content: str, *, original_content: str | None = None, files: list[dict[str, Any]] | None = None) -> dict[str, Any]:
-    message: dict[str, Any] = {"role": "human", "content": content}
+    # The Gateway records run input as sent, but add_messages gives an id-less
+    # message a fresh uuid in the checkpoint. A web client reconnecting to the
+    # run rebuilds its input from both copies and can only match them by id.
+    message: dict[str, Any] = {"id": str(uuid.uuid4()), "role": "human", "content": content}
     if original_content is not None and original_content != content or files:
         additional_kwargs: dict[str, Any] = {}
         if original_content is not None and original_content != content:

backend/tests/test_channels.py

@@ -1197,6 +1197,59 @@ class TestChannelManager:
 
         _run(go())
 
+    def test_run_input_message_has_a_unique_id(self):
+        """Each run's human message carries its own id.
+
+        The Gateway stores run input as sent, while ``add_messages`` gives an
+        id-less message a fresh uuid in the checkpoint. A web client that
+        reconnects to a running channel thread rebuilds the input from both
+        copies and can only match them by id.
+        """
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            outbound_received = []
+
+            async def capture_outbound(msg):
+                outbound_received.append(msg)
+
+            bus.subscribe_outbound(capture_outbound)
+
+            mock_client = _make_mock_langgraph_client()
+            manager._client = mock_client
+
+            await manager.start()
+            for text in ("first", "second"):
+                await bus.publish_inbound(InboundMessage(channel_name="test", chat_id="chat1", user_id="user1", text=text))
+            await _wait_for(lambda: len(outbound_received) >= 2)
+            await manager.stop()
+
+            ids = [call.kwargs["input"]["messages"][0].get("id") for call in mock_client.runs.wait.call_args_list]
+            assert len(ids) == 2
+            assert all(isinstance(message_id, str) and message_id for message_id in ids)
+            assert ids[0] != ids[1]
+
+        _run(go())
+
+    def test_run_input_message_id_matches_its_checkpoint_copy(self):
+        """The id a channel sends survives run admission into both copies."""
+        from langgraph.graph.message import add_messages
+
+        from app.channels.manager import _human_input_message
+        from app.gateway.services import _canonical_run_record_input, normalize_input
+
+        raw_input = {"messages": [_human_input_message("hello", original_content="/skill hello")]}
+        graph_input = normalize_input(raw_input, trusted_internal=True)
+
+        record = _canonical_run_record_input(raw_input, graph_input)
+        checkpoint_messages = add_messages([], graph_input["messages"])
+
+        assert record["messages"][0]["id"] == checkpoint_messages[0].id == raw_input["messages"][0]["id"]
+
     def test_worker_pool_dedupes_stable_provider_message_id(self, tmp_path):
         from app.channels.manager import ChannelManager
 
@@ -4810,7 +4863,9 @@ class TestGithubFollowupBuffer:
             await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
 
             mock_client.runs.create.assert_called_once()
-            drained_text = mock_client.runs.create.call_args[1]["input"]["messages"][0]["content"]
+            drained_message = mock_client.runs.create.call_args[1]["input"]["messages"][0]
+            drained_text = drained_message["content"]
+            assert isinstance(drained_message.get("id"), str) and drained_message["id"]
             for i in range(FOLLOWUP_DRAIN_BATCH_SIZE):
                 assert f"comment {i}" in drained_text
             for i in range(FOLLOWUP_DRAIN_BATCH_SIZE, 15):