diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index 7c5ef23ca29da..39891c23acde9 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -871,9 +871,13 @@ async def _process_message(self, message: Dict[str, Any]) -> None: # Secondary content-fingerprint dedup: upstream re-sends identical text under new message_ids. item_list = message.get("item_list") or [] text = _extract_text(item_list) - if text and self._dedup.is_duplicate(f"content:{sender_id}:{hashlib.md5(text.encode()).hexdigest()}"): - logger.debug("[%s] Content-dedup: skipping duplicate message from %s", self.name, sender_id) - return + content_keys: set = set() + if text: + content_key = f"content:{sender_id}:{hashlib.md5(text.encode()).hexdigest()}" + if self._dedup.is_duplicate(content_key): + logger.debug("[%s] Content-dedup: skipping duplicate message from %s", self.name, sender_id) + return + content_keys.add(content_key) chat_type, effective_chat_id = _guess_chat_type(message, self._account_id) if chat_type == "group": if not self._is_group_allowed(effective_chat_id): @@ -898,10 +902,39 @@ async def _process_message(self, message: Dict[str, Any]) -> None: message_id=message_id or None, media_urls=media_paths, media_types=media_types, timestamp=datetime.now()) logger.info("[%s] inbound from=%s type=%s media=%d", self.name, _safe_id(sender_id), source.chat_type, len(media_paths)) if event.message_type == MessageType.TEXT: + # Content fingerprints ride on the event so the dispatched batch can release + # them afterwards (_dispatch_text_batch) — the same text sent again once the + # batch has been dispatched is a new intent, not an in-flight duplicate (#36750). + event._dedup_content_keys = content_keys # type: ignore[attr-defined] self._enqueue_text_event(event) else: await self.handle_message(event) + def _enqueue_text_event(self, event: MessageEvent) -> None: + # Batching merges later chunks into the first pending event, so their content-dedup + # keys must ride on the surviving event or a merged text stays claimed for the + # whole TTL after the flush (#36750). + pending = self._pending_text_batches.get(self._text_batch_key(event)) + incoming = getattr(event, "_dedup_content_keys", None) + if pending is not None and incoming: + existing = getattr(pending, "_dedup_content_keys", None) + if existing is not None: + existing |= incoming + else: + pending._dedup_content_keys = set(incoming) # type: ignore[attr-defined] + super()._enqueue_text_event(event) + + async def _dispatch_text_batch(self, event: MessageEvent) -> None: + # The batch is dispatched: release every content fingerprint it carried. An upstream + # retry that races this boundary still hits the message_id dedup; only the replay of + # the same text as a fresh send (new message_id) is re-accepted (#36750). + keys = getattr(event, "_dedup_content_keys", None) + try: + await super()._dispatch_text_batch(event) + finally: + for key in keys or (): + self._dedup.discard(key) + async def _collect_media(self, item: Dict[str, Any], media_paths: List[str], media_types: List[str]) -> None: spec = _INBOUND_MEDIA.get(item.get("type")) path, mime = await self._download_media(item, spec) if spec else (None, "") diff --git a/tests/gateway/test_weixin.py b/tests/gateway/test_weixin.py index 9a5575b9aaddb..a367063a4725a 100644 --- a/tests/gateway/test_weixin.py +++ b/tests/gateway/test_weixin.py @@ -663,6 +663,80 @@ async def _drive(): event = adapter.handle_message.await_args[0][0] assert event.text == "hello world" + def test_same_text_accepted_after_previous_dispatch(self): + """Regression test for Issue #36750. + + Once a text batch has been dispatched, its content-fingerprint key is + released, so the same text sent again (a fresh message_id) is accepted + instead of being silently dropped for the whole dedup TTL. + """ + adapter = _make_adapter() + adapter._poll_session = object() + adapter.handle_message = AsyncMock() + adapter._text_batch_delay_seconds = 0.05 + adapter._text_batch_split_delay_seconds = 0.05 + + msg = { + "from_user_id": "wxid_user1", + "item_list": [{"type": 1, "text_item": {"text": "怎么样了"}}], + } + + async def _drive(): + # First send is accepted and flushed. + await adapter._process_message({**msg, "message_id": "msg-1"}) + await asyncio.sleep(0.2) + assert adapter.handle_message.await_count == 1 + # Same text again under a new message_id after the previous batch + # was dispatched — must be accepted, not suppressed by the TTL cache. + await adapter._process_message({**msg, "message_id": "msg-2"}) + await asyncio.sleep(0.2) + + asyncio.run(_drive()) + + assert adapter.handle_message.await_count == 2 + + def test_merged_batch_releases_every_content_key(self): + """A flushed batch must release every merged chunk's content key (#36750 review). + + Two distinct texts accepted within one debounce window merge into a + single dispatched event; both fingerprints must be released on dispatch + so re-sending either text afterwards is accepted. + """ + adapter = _make_adapter() + adapter._poll_session = object() + adapter.handle_message = AsyncMock() + adapter._text_batch_delay_seconds = 0.05 + adapter._text_batch_split_delay_seconds = 0.05 + + def _msg(text: str, message_id: str) -> dict: + return { + "from_user_id": "wxid_user1", + "item_list": [{"type": 1, "text_item": {"text": text}}], + "message_id": message_id, + } + + async def _drive(): + # A and B are distinct texts landing in the same pending batch. + await adapter._process_message(_msg("第一句", "msg-1")) + await adapter._process_message(_msg("第二句", "msg-2")) + await asyncio.sleep(0.2) + assert adapter.handle_message.await_count == 1 + assert adapter.handle_message.await_args[0][0].text == "第一句\n第二句" + # Re-send both texts under fresh message_ids after the dispatch. + await adapter._process_message(_msg("第一句", "msg-3")) + await adapter._process_message(_msg("第二句", "msg-4")) + await asyncio.sleep(0.2) + + asyncio.run(_drive()) + + # The re-sends were accepted and merged into a second dispatched batch. + # If only the first chunk's key had been released, this second batch + # would carry just one of the two texts; if neither, no second dispatch. + assert adapter.handle_message.await_count == 2 + second_batch = adapter.handle_message.await_args_list[1].args[0].text + assert "第一句" in second_batch + assert "第二句" in second_batch + class TestWeixinTextDebounce: """Text-debounce batching for rapid multi-message bursts (issue #35301).