diff --git a/CHANGELOG.md b/CHANGELOG.md index 54ba21c039..ec81fa0fa7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 5c6890b30d..989b8867d2 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index dbfe34faf3..b305dcc34e 100644 --- a/backend/app/channels/manager.py +++ b/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: diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 698a291493..f02e028fae 100644 --- a/backend/tests/test_channels.py +++ b/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):