From 61d76d9a6a4524e7d350e568993218d7d787e4f7 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Fri, 6 Nov 2026 22:36:19 +0900 Subject: [PATCH] fix(channels): key WeCom group conversations by chatid, not the sender (#7794) * fix(channels): key WeCom group conversations by chatid, not the sender Group messages used the sender's userid as chat_id and topic_id, the same key as the sender's private chat. Wherever no group connection kept them apart (auth-disabled mode, require_bound_identity: false, legacy group bindings), a member's private chat and every group they wrote in ran in one DeerFlow thread. Groups are now keyed by chatid with one thread per member. A group-scoped connection carries its sender-keyed thread over once; a legacy binding, which spans every group of the bot, does not. * docs(changelog): reference #7794 for the WeCom group conversation fix --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 13 ++ CHANGELOG_zh.md | 9 + backend/app/channels/wecom.py | 47 ++++- backend/docs/IM_CHANNEL_CONNECTIONS.md | 6 + backend/tests/test_wecom_conversation_keys.py | 164 ++++++++++++++++++ backend/tests/test_wecom_group_bindings.py | 9 +- 6 files changed, 239 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_wecom_conversation_keys.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d5d343774a..d3d0fb0c74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,18 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **channels:** WeCom group messages now have their own conversation. Every + message used the sender's userid as both chat and topic ID, so a member's + private chat and every group they wrote in shared one DeerFlow thread + wherever no group binding kept them apart: in auth-disabled mode, with + `require_bound_identity: false`, and under legacy group bindings. A group + reply could then draw on the member's private chat or another group, and a + reply sent without its frame went to the private chat. Groups are now keyed + by `chatid` with one thread per member, and private chats keep their + threads. A group bound with its own `/connect` keeps its thread; a legacy + group binding, which covers every group of the bot, starts a new one in + each group. ([#7794]) + - **sandbox:** Fresh lazy AIO runs with an explicit authorized skill view no longer create or acquire a sandbox before the first sandbox-backed tool call. Existing sandbox bindings and providers that require skill uploads @@ -16038,6 +16050,7 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7781]: https://github.com/bytedance/deer-flow/pull/7781 [#7782]: https://github.com/bytedance/deer-flow/pull/7782 [#7792]: https://github.com/bytedance/deer-flow/pull/7792 +[#7794]: https://github.com/bytedance/deer-flow/pull/7794 [#7831]: https://github.com/bytedance/deer-flow/pull/7831 [#7892]: https://github.com/bytedance/deer-flow/pull/7892 [#7907]: https://github.com/bytedance/deer-flow/pull/7907 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 482ecbeae5..403e2cc559 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -23,6 +23,14 @@ ### 修复 +- **渠道：** 企业微信群消息现在拥有独立的会话。此前每条消息都把发送者的 userid + 同时用作会话 ID 和话题 ID，因此在没有群绑定加以区分的情况下（免认证模式、 + `require_bound_identity: false`，以及旧版群绑定），成员的私聊和其发过言的所有群 + 共用同一个 DeerFlow 线程：群内回复可能用到该成员私聊或其他群的内容，失去原始 + frame 的回复也会被发到私聊。现在群以 `chatid` 为键、每位成员一个线程，私聊线程 + 保持不变。通过各自 `/connect` 绑定的群保留原有线程；覆盖该机器人所有群的旧版 + 群绑定会在每个群中开始新线程。([#7794]) + - **沙箱：** 使用显式授权技能视图的新 AIO 运行，在启用惰性初始化时，不再于首次调用 沙箱工具前创建或获取沙箱。已有沙箱绑定和需要上传技能的提供方继续保留原有的即时同步行为。 ([#7782]) @@ -12761,5 +12769,6 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 [#7781]: https://github.com/bytedance/deer-flow/pull/7781 [#7792]: https://github.com/bytedance/deer-flow/pull/7792 +[#7794]: https://github.com/bytedance/deer-flow/pull/7794 [#7831]: https://github.com/bytedance/deer-flow/pull/7831 [#7782]: https://github.com/bytedance/deer-flow/pull/7782 diff --git a/backend/app/channels/wecom.py b/backend/app/channels/wecom.py index 9f7c0bd458..1002e7a759 100644 --- a/backend/app/channels/wecom.py +++ b/backend/app/channels/wecom.py @@ -40,6 +40,10 @@ else: _WECOM_ACK_REJECTION_PREFIX = "Reply ack error" +def _group_workspace_id(aibotid: str | None, chatid: str) -> str: + return f"{aibotid}:group:{chatid}" if aibotid else f"group:{chatid}" + + def _file_md5(path: str) -> str: md5_hasher = hashlib.md5() with open(path, "rb") as file_obj: @@ -587,8 +591,12 @@ class WeComChannel(Channel): } if body.get("chatid") is not None: inbound_metadata["chatid"] = body.get("chatid") + # A group's conversation is the group, a private chat's the sender. + # Keyed by sender alone, a member's private chat and every group they + # wrote in shared one thread wherever no group connection kept them apart. + group_chatid = str(body.get("chatid") or "").strip() if body.get("chattype") == "group" else "" inbound = self._make_inbound( - chat_id=user_id, # keep user's conversation in memory + chat_id=group_chatid or user_id, user_id=user_id, text=text, msg_type=inbound_type, @@ -596,7 +604,7 @@ class WeComChannel(Channel): files=files or [], metadata=inbound_metadata, ) - inbound.topic_id = user_id # keep the same thread + inbound.topic_id = user_id # one thread per member in each conversation stream_id = generate_req_id("stream") self._ws_frames[msg_id] = frame @@ -614,6 +622,8 @@ class WeComChannel(Channel): pass inbound = await self._attach_connection_identity(inbound) + if group_chatid: + await self._carry_over_sender_keyed_group_thread(inbound, group_chatid) self._commit_reserved_inbound(reservation, inbound) finally: reservation.release() @@ -625,7 +635,7 @@ class WeComChannel(Channel): if chattype == "group": if not chatid or self._connection_repo is None: return inbound - workspace_id = f"{aibotid}:group:{chatid}" if aibotid else f"group:{chatid}" + workspace_id = _group_workspace_id(aibotid, chatid) inbound = await attach_connection_identity(inbound, repo=self._connection_repo, provider="wecom", workspace_id=workspace_id, fallback_without_workspace=False) if inbound.connection_id: return inbound @@ -647,6 +657,35 @@ class WeComChannel(Channel): fallback_without_workspace=True, ) + async def _carry_over_sender_keyed_group_thread(self, inbound: InboundMessage, chatid: str) -> None: + """Rebind a group connection's thread from the sender key it used to have. + + Group messages were once keyed like private ones (``userid``, ``userid``). + A group connection belongs to one group, so its thread under that key is + this group's and moves to the group key once. A legacy group binding + spans every group of the bot; its old thread is left behind rather than + continued in whichever group writes first. + """ + repo = self._connection_repo + if repo is None or not inbound.connection_id or not inbound.owner_user_id: + return + aibotid = str(inbound.metadata.get("aibotid") or "").strip() or None + if inbound.workspace_id != _group_workspace_id(aibotid, chatid): + return + if await repo.get_thread_id(inbound.connection_id, inbound.chat_id, inbound.topic_id) is not None: + return + previous = await repo.get_thread_id(inbound.connection_id, inbound.user_id, inbound.user_id) + if previous is None: + return + await repo.claim_thread_id( + connection_id=inbound.connection_id, + owner_user_id=inbound.owner_user_id, + provider=self.name, + external_conversation_id=inbound.chat_id, + external_topic_id=inbound.topic_id, + thread_id=previous, + ) + async def _bind_connection_from_connect_code(self, *, frame: dict[str, Any], user_id: str, code: str) -> bool: if self._connection_repo is None or not code: return False @@ -669,7 +708,7 @@ class WeComChannel(Channel): if not chatid: await self._send_connection_reply(frame, await self._connect_reply_text("incomplete", owner_user_id=state["owner_user_id"])) return True - workspace_id = f"{aibotid}:group:{chatid}" if aibotid else f"group:{chatid}" + workspace_id = _group_workspace_id(aibotid, chatid) raw_group_name = str(body.get("chatname") or body.get("group_name") or "").strip() workspace_name = notification_group_display_name(state.get("metadata")) or raw_group_name[:256] or None metadata: dict[str, Any] = { diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 51775dbe1c..fc7137490e 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -828,6 +828,12 @@ chat identity. That compatibility applies only to the old group-binding shape; a new private binding never authorizes messages from an unbound group. Bind a specific group again with a name to make it available as a notification target. +Each WeCom group is its own conversation, with one DeerFlow thread per member; +a private chat is keyed by the sender alone. A group bound with its own +`/connect` keeps the thread it had before groups got their own conversation. A +legacy group binding covers every group of the bot, so its earlier thread is +not continued in any of them; each group starts a new one. + ### WeChat private notifications Use the existing iLink QR/bot setup and bind in a private chat. Send one private diff --git a/backend/tests/test_wecom_conversation_keys.py b/backend/tests/test_wecom_conversation_keys.py new file mode 100644 index 0000000000..c0e7fa2f2f --- /dev/null +++ b/backend/tests/test_wecom_conversation_keys.py @@ -0,0 +1,164 @@ +"""WeCom conversations: a group is keyed by its chatid, a private chat by the sender. + +Group messages used to be keyed like private ones (``chat_id`` and ``topic_id`` +both the sender's ``userid``). Wherever no group connection kept them apart +(auth-disabled mode, ``require_bound_identity: false``, legacy group +bindings), a member's private chat and every group they wrote in shared one +DeerFlow thread, and a reply in one group could draw on another chat. +""" + +from __future__ import annotations + +import tempfile +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio + +from app.channels.connection_identity import lookup_thread_id +from app.channels.manager import ChannelManager +from app.channels.message_bus import InboundMessage, MessageBus, OutboundMessage +from app.channels.store import JsonChannelStore +from app.channels.wecom import WeComChannel +from deerflow.persistence.channel_connections import ChannelConnectionRepository +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + +pytestmark = pytest.mark.asyncio + + +@pytest_asyncio.fixture +async def repo(tmp_path): + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'wecom.db'}", sqlite_dir=str(tmp_path)) + try: + yield ChannelConnectionRepository(get_session_factory()) + finally: + await close_engine() + + +def _channel(bus: MessageBus, repo: ChannelConnectionRepository | None = None) -> WeComChannel: + channel = WeComChannel(bus, {"connection_repo": repo} if repo is not None else {}) + channel._ws_client = SimpleNamespace(reply=AsyncMock(), reply_stream=AsyncMock(), send_message=AsyncMock()) + return channel + + +async def _receive(channel: WeComChannel, bus: MessageBus, msgid: str, *, chatid: str | None = None, user: str = "alice") -> InboundMessage: + body = {"aibotid": "bot-1", "chattype": "group" if chatid else "single", "msgid": msgid, "from": {"userid": user}, "text": {"content": "hi"}} + if chatid: + body["chatid"] = chatid + await channel._publish_ws_inbound({"body": body}, "hi") + return bus.get_inbound_nowait() + + +async def test_group_and_private_messages_get_their_own_conversation_keys(): + bus = MessageBus() + channel = _channel(bus) + + private = await _receive(channel, bus, "m1") + group_a = await _receive(channel, bus, "m2", chatid="wr-team-a") + group_b = await _receive(channel, bus, "m3", chatid="wr-team-b") + other_member = await _receive(channel, bus, "m4", chatid="wr-team-a", user="bob") + + assert (private.chat_id, private.topic_id) == ("alice", "alice") + assert (group_a.chat_id, group_a.topic_id) == ("wr-team-a", "alice") + assert (group_b.chat_id, group_b.topic_id) == ("wr-team-b", "alice") + assert (other_member.chat_id, other_member.topic_id) == ("wr-team-a", "bob") + + +async def test_reply_without_its_frame_goes_to_the_group_not_the_sender(): + # The frame is gone after a Gateway restart, so a recovered run's reply + # is pushed to the conversation's chat_id instead. + bus = MessageBus() + channel = _channel(bus) + message = await _receive(channel, bus, "m1", chatid="wr-team-a") + channel._ws_frames.clear() + + await channel.send(OutboundMessage(channel_name="wecom", chat_id=message.chat_id, thread_id="thread-1", text="answer", thread_ts=message.thread_ts, is_final=True)) + + channel._ws_client.send_message.assert_awaited_once() + assert channel._ws_client.send_message.await_args.args[0] == "wr-team-a" + + +def _client() -> MagicMock: + created = iter(f"thread-{index}" for index in range(1, 10)) + client = MagicMock() + client.threads.create = AsyncMock(side_effect=lambda **_kwargs: {"thread_id": next(created)}) + client.threads.update = AsyncMock(return_value={}) + client.threads.get = AsyncMock(side_effect=lambda thread_id, **_kwargs: {"thread_id": thread_id}) + + def stream(*_args, **_kwargs): + async def events(): + yield SimpleNamespace(event="values", data={"messages": [{"type": "ai", "content": "ok"}]}) + + return events() + + client.runs.stream = MagicMock(side_effect=stream) + return client + + +async def test_unbound_private_chat_and_groups_run_in_separate_threads(monkeypatch): + # Auth-disabled mode: no connection scopes the conversation, only its key. + monkeypatch.setattr("app.channels.manager._auth_disabled_owner_user_id", lambda: "default") + bus = MessageBus() + channel = _channel(bus) + manager = ChannelManager(bus=bus, store=JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"), max_concurrency=1, shutdown_grace_period_seconds=0) + manager._client = client = _client() + + for msgid, chatid in (("m1", None), ("m2", "wr-team-a"), ("m3", "wr-team-b"), ("m4", "wr-team-a"), ("m5", None)): + await manager._handle_message(await _receive(channel, bus, msgid, chatid=chatid)) + + threads = [call.args[0] for call in client.runs.stream.call_args_list] + assert threads == ["thread-1", "thread-2", "thread-3", "thread-2", "thread-1"] + + +async def _bind_group(repo: ChannelConnectionRepository, chatid: str) -> dict: + return await repo.upsert_connection( + owner_user_id="owner", + provider="wecom", + external_account_id="alice", + workspace_id=f"bot-1:group:{chatid}", + metadata={"aibotid": "bot-1", "chattype": "group", "chatid": chatid}, + ) + + +async def _bind_sender_keyed_thread(repo: ChannelConnectionRepository, connection: dict, thread_id: str) -> None: + await repo.set_thread_id(connection_id=connection["id"], owner_user_id="owner", provider="wecom", external_conversation_id="alice", external_topic_id="alice", thread_id=thread_id) + + +async def test_bound_group_keeps_the_thread_it_had_under_the_sender_key(repo): + group_a = await _bind_group(repo, "wr-team-a") + group_b = await _bind_group(repo, "wr-team-b") + await _bind_sender_keyed_thread(repo, group_a, "team-a-thread") + await _bind_sender_keyed_thread(repo, group_b, "team-b-thread") + bus = MessageBus() + channel = _channel(bus, repo) + + first_a = await _receive(channel, bus, "m1", chatid="wr-team-a") + first_b = await _receive(channel, bus, "m2", chatid="wr-team-b") + + assert first_a.connection_id == group_a["id"] + assert await lookup_thread_id(first_a, repo=repo, store=None) == "team-a-thread" + assert await lookup_thread_id(first_b, repo=repo, store=None) == "team-b-thread" + + # A rebinding under the group key (``/new``) is never overwritten. + await repo.set_thread_id(connection_id=group_a["id"], owner_user_id="owner", provider="wecom", external_conversation_id="wr-team-a", external_topic_id="alice", thread_id="fresh-thread") + later_a = await _receive(channel, bus, "m3", chatid="wr-team-a") + assert await lookup_thread_id(later_a, repo=repo, store=None) == "fresh-thread" + + +async def test_legacy_group_binding_does_not_continue_its_sender_keyed_thread(repo): + legacy = await repo.upsert_connection(owner_user_id="owner", provider="wecom", external_account_id="alice", workspace_id="bot-1", metadata={"aibotid": "bot-1", "chattype": "group"}) + await _bind_sender_keyed_thread(repo, legacy, "existing-thread") + bus = MessageBus() + channel = _channel(bus, repo) + + group_a = await _receive(channel, bus, "m1", chatid="wr-team-a") + group_b = await _receive(channel, bus, "m2", chatid="wr-team-b") + + # The binding still authorizes its owner in every group of the bot, but + # its thread may hold any of them, so no group continues it. + assert group_a.connection_id == group_b.connection_id == legacy["id"] + assert group_a.owner_user_id == "owner" + assert await lookup_thread_id(group_a, repo=repo, store=None) is None + assert await lookup_thread_id(group_b, repo=repo, store=None) is None diff --git a/backend/tests/test_wecom_group_bindings.py b/backend/tests/test_wecom_group_bindings.py index 00476c7324..dcf0d824fc 100644 --- a/backend/tests/test_wecom_group_bindings.py +++ b/backend/tests/test_wecom_group_bindings.py @@ -7,7 +7,6 @@ from unittest.mock import AsyncMock import pytest import pytest_asyncio -from app.channels.connection_identity import lookup_thread_id from app.channels.message_bus import InboundMessage, MessageBus from app.channels.wecom import WeComChannel from deerflow.persistence.channel_connections import ChannelConnectionRepository @@ -24,7 +23,7 @@ async def repo(tmp_path): def inbound(group="wr-group", bot="bot-1"): - return InboundMessage(channel_name="wecom", chat_id="alice", user_id="alice", text="group follow-up", topic_id="alice", metadata={"aibotid": bot, "chattype": "group", "chatid": group}) + return InboundMessage(channel_name="wecom", chat_id=group, user_id="alice", text="group follow-up", topic_id="alice", metadata={"aibotid": bot, "chattype": "group", "chatid": group}) @pytest.mark.asyncio @@ -38,14 +37,14 @@ async def test_private_binding_does_not_authorize_an_unbound_group(repo, private @pytest.mark.asyncio -async def test_legacy_group_binding_retains_owner_and_existing_thread(repo): +async def test_legacy_group_binding_retains_owner(repo): + # The thread it had under the sender key is not continued; see + # test_wecom_conversation_keys.py. legacy = await repo.upsert_connection(owner_user_id="owner", provider="wecom", external_account_id="alice", workspace_id="bot-1", metadata={"aibotid": "bot-1", "chattype": "group"}) - await repo.set_thread_id(connection_id=legacy["id"], owner_user_id="owner", provider="wecom", external_conversation_id="alice", external_topic_id="alice", thread_id="existing-thread") channel = WeComChannel(MessageBus(), {"connection_repo": repo}) message = await channel._attach_connection_identity(inbound()) assert message.connection_id == legacy["id"] assert message.owner_user_id == "owner" - assert await lookup_thread_id(message, repo=repo, store=None) == "existing-thread" other_bot = await channel._attach_connection_identity(inbound(bot="bot-2")) assert other_bot.connection_id is None await repo.disconnect_connection(connection_id=legacy["id"], owner_user_id="owner")