# NousResearch/hermes-agent · WeCom WebSocket reconnect silent group message loss ## Open now - Nothing open on GitHub. A cluster of bug reports and fixes addressing WeCom AI Bot WebSocket reconnections that cause group messages to be silently dropped, including zombie session handling, watchdog blind spots where DM traffic masks group routing failures, and resetting message timestamps on reconnect. ## TL;DR - 4 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 1 merged PR(s), 2 PR(s) closed without merging. - A fix was merged to main in PR #11897. Verify it addresses your case. - **Recommended:** read PR #11897 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `resolved` · deciding PR(s): `pr-11897`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference. - Pairwise link density `0.50`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #11554: \[Bug\]: WeCom AI Bot WebSocket reconnects cause group message delivery to stop silently ## Prior & competing PRs - **#11656: fix(wecom): watchdog + reset _last_msg_at on reconnect to fix silent group message loss**. closed without merge · fix tier `suggested` - maintainer direction: category `merged_via`; implication `superseded`; refs `#11572` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#11897: fix(wecom): resolve WebSocket zombie sessions and group chat 600039 errors (#11554)**. merged to main · fix tier `verified` - maintainer direction: category `merged` - files: `gateway/platforms/wecom.py`, `tests/gateway/test_wecom.py` - fix hunks: - `gateway/platforms/wecom.py` ```diff @@ -180,6 +180,8 @@ def __init__(self, config: PlatformConfig): self._text_batch_split_delay_seconds = float(os.getenv("HERMES_WECOM_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0")) self._pending_text_batches: Dict[str, MessageEvent] = {} self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {} + self._device_id = uuid.uuid4().hex + self._last_chat_req_ids: Dict[str, str] = {} # ------------------------------------------------------------------ # Connection lifecycle ``` ```diff @@ -277,7 +279,11 @@ async def _open_connection(self) -> None: { "cmd": APP_CMD_SUBSCRIBE, "headers": {"req_id": req_id}, - "body": {"bot_id": self._bot_id, "secret": [redacted:secret-assignment] + "body": { + "bot_id": self._bot_id, + "secret": [redacted:secret-assignment] + "device_id": self._device_id, + }, } ) ``` ```diff @@ -496,6 +502,11 @@ async def _on_message(self, payload: Dict[str, Any]) -> None: logger.debug("[%s] DM sender %s blocked by policy", self.name, sender_id) return + # Cache the inbound req_id after policy checks so proactive sends to + # this chat can fall back to APP_CMD_RESPONSE (required for groups — + # WeCom AI Bots cannot initiate APP_CMD_SEND in group chats). + self._remember_chat_req_id(chat_id, self._payload_req_id(payload)) + text, reply_text = self._extract_text(body) media_urls, media_types = await self._extract_media(body) message_type = self._derive_message_type(body, text, media_types) ``` ```diff @@ -847,6 +858,23 @@ def _remember_reply_req_id(self, message_id: str, req_id: str) -> None: while len(self._reply_req_ids) > DEDUP_MAX_SIZE: self._reply_req_ids.pop(next(iter(self._reply_req_ids))) + def _remember_chat_req_id(self, chat_id: str, req_id: str) -> None: + """Cache the most recent inbound req_id per chat. + + Used as a fallback reply target when we need to send into a group + without an explicit ``reply_to`` — WeCom AI Bots are blocked from + APP_CMD_SEND in groups and must use APP_CMD_RESPONSE bound to some + prior req_id. Bounded like _reply_req_ids so long-running gateways + don't leak memory across many chats. + """ + normalized_chat_id = str(chat_id or "").strip() + normalized_req_id = str(req_id or "").strip() + if not normalized_chat_id or not normalized_req_id: + return + self._last_chat_req_ids[normalized_chat_id] = normalized_req_id + while len(self._last_chat_req_ids) > DEDUP_MAX_SIZE: + self._last_chat_req_ids.pop(next(iter(self._last_chat_req_ids))) + def _reply_req_id_for_message(self, reply_to: Optional[str]) -> Optional[str]: normalized = str(reply_to or "").strip() if not normalized or normalized.startswith("quote:"): ``` ```diff @@ -1163,19 +1191,15 @@ async def _send_media_message(self, chat_id: str, media_type: str, media_id: str self._raise_for_wecom_error(response, "send media message") return response - async def _send_reply_stream(self, reply_req_id: str, content: str) -> Dict[str, Any]: + async def _send_reply_markdown(self, reply_req_id: str, content: str) -> Dict[str, Any]: response = await self._send_reply_request( reply_req_id, { - "msgtype": "stream", - "stream": { - "id": self._new_req_id("stream"), - "finish": True, - "content": content[:self.MAX_MESSAGE_LENGTH], - }, + "msgtype": "markdown", + "markdown": {"content": content[:self.MAX_MESSAGE_LENGTH]}, }, ) - self._raise_for_wecom_error(response, "send reply stream") + self._raise_for_wecom_error(response, "send reply markdown") return response async def _send_reply_media_message( ``` ```diff @@ -1235,6 +1259,9 @@ async def _send_media_source( return SendResult(success=False, error=prepared["reject_reason"]) reply_req_id = self._reply_req_id_for_message(reply_to) + if not reply_req_id and chat_id in self._last_chat_req_ids: + reply_req_id = self._last_chat_req_ids[chat_id] + try: upload_result = await self._upload_media_bytes( prepared["data"], ``` ```diff @@ -1302,8 +1329,12 @@ async def send( try: reply_req_id = self._reply_req_id_for_message(reply_to) + + if not reply_req_id and chat_id in self._last_chat_req_ids: + reply_req_id = self._last_chat_req_ids[chat_id] + if reply_req_id: - response = await self._send_reply_stream(reply_req_id, content) + response = await self._send_reply_markdown(reply_req_id, content) else: response = await self._send_request( APP_CMD_SEND, ``` - `tests/gateway/test_wecom.py` ```diff @@ -119,7 +119,7 @@ async def aclose(self): class TestWeComReplyMode: @pytest.mark.asyncio - async def test_send_uses_passive_reply_stream_when_reply_context_exists(self): + async def test_send_uses_passive_reply_markdown_when_reply_context_exists(self): from gateway.platforms.wecom import WeComAdapter adapter = WeComAdapter(PlatformConfig(enabled=True)) ``` ```diff @@ -134,9 +134,10 @@ async def test_send_uses_passive_reply_stream_when_reply_context_exists(self): adapter._send_reply_request.assert_awaited_once() args = adapter._send_reply_request.await_args.args assert args[0] == "req-1" - assert args[1]["msgtype"] == "stream" - assert args[1]["stream"]["finish"] is True - assert args[1]["stream"]["content"] == "hello from reply" + # msgtype: stream triggers WeCom errcode 600039 on many mobile clients + # (unsupported type). Markdown renders everywhere. + assert args[1]["msgtype"] == "markdown" + assert args[1]["markdown"]["content"] == "hello from reply" @pytest.mark.asyncio async def test_send_image_file_uses_passive_reply_media_when_reply_context_exists(self): ``` ```diff @@ -593,3 +594,193 @@ async def test_on_message_respects_group_policy(self): await adapter._on_message(payload) adapter.handle_message.assert_not_awaited() + +class TestWeComZombieSessionFix: + """Tests for PR #11572 — device_id, markdown reply, group req_id fallback.""" + + def test_adapter_generates_stable_device_id_per_instance(self): + from gateway.platforms.wecom import WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + assert isinstance(adapter._device_id, str) + assert len(adapter._device_id) > 0 + # Second snapshot on the same adapter must be identical — only a fresh + # adapter instance should get a new device_id (one-per-reconnect is the + # zombie-session footgun we're fixing). + assert adapter._device_id == adapter._device_id + + def test_different_adapter_instances_get_distinct_device_ids(self): + from gateway.platforms.wecom import WeComAdapter + + a = WeComAdapter(PlatformConfig(enabled=True)) + b = WeComAdapter(PlatformConfig(enabled=True)) + assert a._device_id != b._device_id + + @pytest.mark.asyncio + async def test_open_connection_includes_device_id_in_subscribe(self): + from gateway.platforms.wecom import APP_CMD_SUBSCRIBE, WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + adapter._bot_id = "test-bot" + adapter._secret = "[redacted:secret-assignment]" + + sent_payloads = [] + + class _FakeWS: + closed = False + + async def send_json(self, payload): + sent_payloads.append(payload) + + async def close(self): + return None + + class _FakeSession: + def __init__(self, *args, **kwargs): + pass + + async def ws_connect(self, *args, **kwargs): + return _FakeWS() + + async def close(self): + return None + + async def _fake_cleanup(): + return None + + async def _fake_handshake(req_id): + return {"errcode": 0, "headers": {"req_id": req_id}} + + adapter._cleanup_ws = _fake_cleanup + adapter._wait_for_handshake = _fake_handshake + + with patch("gateway.platforms.wecom.aiohttp.ClientSession", _FakeSession): + await adapter._open_connection() + + assert len(sent_payloads) == 1 + subscribe = sent_payloads[0] + assert subscribe["cmd"] == APP_CMD_SUBSCRIBE + assert subscribe["body"]["bot_id"] == "test-bot" + assert subscribe["body"]["secret"] == "test-secret" + assert subscribe["body"]["device_id"] == adapter._device_id + + @pytest.mark.asyncio + async def test_on_message_caches_last_req_id_per_chat(self): + from gateway.platforms.wecom import WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + adapter._text_batch_delay_seconds = 0 + adapter.handle_message = AsyncMock() + adapter._extract_media = AsyncMock(return_value=([], [])) + + payload = { + "cmd": "aibot_msg_callback", + "headers": {"req_id": "req-abc"}, + "body": { + "msgid": "msg-1", + "chatid": "group-1", + "chattype": "group", + "from": {"userid": "user-1"}, + "msgtype": "text", + "text": {"content": "hi"}, + }, + } + + await adapter._on_message(payload) + assert adapter._last_chat_req_ids["group-1"] == "req-abc" + + @pytest.mark.asyncio + async def test_on_message_does_not_cache_blocked_sender_req_id(self): + """Blocked chats shouldn't populate the proactive-send fallback cache.""" + from gateway.platforms.wecom import WeComAdapter + + adapter = WeComAdapter( + PlatformConfig( + enabled=True, + extra={"group_policy": "allowlist", "group_allow_from": ["group-ok"]}, + ) + ) + adapter.handle_message = AsyncMock() + adapter._extract_media = AsyncMock(return_value=([], [])) + + payload = { + "cmd": "aibot_msg_callback", + "headers": {"req_id": "req-abc"}, + "body": { + "msgid": "msg-1", + "chatid": "group-blocked", + "chattype": "group", + "from": {"userid": "user-1"}, + "msgtype": "text", + "text": {"content": "hi"}, + }, + } + + await adapter._on_message(payload) + adapter.handle_message.assert_not_awaited() + assert "group-blocked" not in adapter._last_chat_req_ids + + def test_remember_chat_req_id_is_bounded(self): + from gateway.platforms.wecom import DEDUP_MAX_SIZE, WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + for i in range(DEDUP_MAX_SIZE + 50): + adapter._remember_chat_req_id(f"chat-{i}", f"req-{i}") + assert len(adapter._last_chat_req_ids) <= DEDUP_MAX_SIZE + # The most recently remembered chat must still be present. + latest = f"chat-{DEDUP_MAX_SIZE + 49}" + assert adapter._last_chat_req_ids[latest] == f"req-{DEDUP_MAX_SIZE + 49}" + + def test_remember_chat_req_id_ignores_empty_values(self): + from gateway.platforms.wecom import WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + adapter._remember_chat_req_id("", "req-1") + adapter._remember_chat_req_id("chat-1", "") + adapter._remember_chat_req_id(" ", " ") + assert adapter._last_chat_req_ids == {} + + @pytest.mark.asyncio + async def test_proactive_group_send_falls_back_to_cached_req_id(self): + """Sending into a group without reply_to should use the last cached + req_id via APP_CMD_RESPONSE — WeCom AI Bots cannot initiate APP_CMD_SEND + in group chats (errcode 600039).""" + from gateway.platforms.wecom import WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + adapter._last_chat_req_ids["group-1"] = "inbound-req-42" + adapter._send_reply_request = AsyncMock( + return_value={"headers": {"req_id": "inbound-req-42"}, "errcode": 0} + ) + adapter._send_request = AsyncMock( + return_value={"headers": {"req_id": "new"}, "errcode": 0} + ) + + result = await adapter.send("group-1", "ping", reply_to=None) + + assert result.success is True + # Must route through reply (APP_CMD_RESPONSE), not proactive send. + adapter._send_reply_request.assert_awaited_once() + adapter._send_request.assert_not_awaited() + args = adapter._send_reply_request.await_args.args + assert args[0] == "inbound-req-42" + assert args[1]["msgtype"] == "markdown" + assert args[1]["markdown"]["content"] == "ping" + + @pytest.mark.asyncio + async def test_proactive_send_without_cached_req_id_uses_app_cmd_send(self): + """When we have no prior req_id (fresh DM target), APP_CMD_SEND is used.""" + from gateway.platforms.wecom import APP_CMD_SEND, WeComAdapter + + adapter = WeComAdapter(PlatformConfig(enabled=True)) + adapter._send_request = AsyncMock( + return_value={"headers": {"req_id": "new"}, "errcode": 0} + ) + + result = await adapter.send("fresh-dm-chat", "ping", reply_to=None) + + assert result.success is True + adapter._send_request.assert_awaited_once() + cmd = adapter._send_request.await_args.args[0] + assert cmd == APP_CMD_SEND + ``` - **#11572: fix(wecom): resolve WebSocket zombie sessions and group chat 600039 e…**. closed without merge - maintainer direction: category `merged_via`; implication `superseded` ## Fix status - Merged to main in PR #11897 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-04-18`. This confirmation is old and may be out of date (a revert/reopen would not have invalidated it here); re-verify.