From 5f8a3e9114875f7ae75a6273041dbf85a15697a9 Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Mon, 29 Jun 2026 01:33:27 -0400 Subject: [PATCH] fix(qqofficial): allocate fresh msg_seq per inbound msg_id to avoid 40054005 dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit QQ Official v2 API deduplicates passive messages by (msg_id, msg_seq). The adapter reused msg_seq across multiple sends under the same inbound msg_id, so multi-part text replies, rich media, and especially streaming chunks (msg_seq pinned at 1) were rejected with: 40054005 消息被去重，请检查请求msgseq Add a per-inbound-msg_id sequence allocator (next_reply_msg_seq) on QQOfficialClient, backed by a bounded OrderedDict + asyncio lock, and use it for C2C/group text sends and rich media. Streaming now advances ctx['msg_seq'] per chunk. Proactive sends (no msg_id) fall back to the existing global counter. Closes #2290 --- src/langbot/libs/qq_official_api/api.py | 32 ++++++- .../platform/adapters/qqofficial/adapter.py | 3 + .../platform/test_qqofficial_eba_adapter.py | 89 +++++++++++++++++++ 3 files changed, 122 insertions(+), 2 deletions(-) diff --git a/src/langbot/libs/qq_official_api/api.py b/src/langbot/libs/qq_official_api/api.py index db3194b6c..a42055bfe 100644 --- a/src/langbot/libs/qq_official_api/api.py +++ b/src/langbot/libs/qq_official_api/api.py @@ -1,6 +1,7 @@ import re import time import asyncio +import collections from quart import request import httpx from quart import Quart @@ -35,6 +36,11 @@ class QQOfficialClient: self.access_token_expiry_time = None self.logger = logger self._msg_seq_counter = 0 + # Per-inbound-msg_id 回复序号分配器。QQ 官方 v2 API 通过 (msg_id, msg_seq) + # 对被动消息去重，同一 msg_id 复用 msg_seq 会被拒绝 (40054005 消息被去重)。 + self._reply_msg_seq: 'collections.OrderedDict[str, int]' = collections.OrderedDict() + self._reply_msg_seq_lock = asyncio.Lock() + self._reply_msg_seq_max_keys = 1024 self._token_refresh_task: Optional[asyncio.Task] = None async def check_access_token(self): @@ -177,6 +183,27 @@ class QQOfficialClient: content_type = attachment.get('content_type', '') return content_type.startswith('image/') + async def next_reply_msg_seq(self, msg_id: Optional[str]) -> int: + """为同一条入站 msg_id 关联的每次被动下发分配递增的 msg_seq。 + + QQ 官方 v2 API 通过 (msg_id, msg_seq) 去重，同一 msg_id 复用 msg_seq 会被 + 拒绝 (40054005 消息被去重)。多段文本、富媒体、重试等场景下每次发送都需要 + 新的 msg_seq。无 msg_id（主动消息等）时回退到全局自增计数器。 + """ + if not msg_id: + self._msg_seq_counter += 1 + return self._msg_seq_counter + + key = str(msg_id) + async with self._reply_msg_seq_lock: + seq = self._reply_msg_seq.get(key, 0) + 1 + self._reply_msg_seq[key] = seq + self._reply_msg_seq.move_to_end(key) + # 限制字典大小，避免长期运行的内存增长 + while len(self._reply_msg_seq) > self._reply_msg_seq_max_keys: + self._reply_msg_seq.popitem(last=False) + return seq + async def send_private_text_msg(self, user_openid: str, content: str, msg_id: str): """发送私聊消息""" if not await self.check_access_token(): @@ -192,6 +219,7 @@ class QQOfficialClient: 'content': content, 'msg_type': 0, 'msg_id': msg_id, + 'msg_seq': await self.next_reply_msg_seq(msg_id), } response = await client.post(url, headers=headers, json=data) response_data = response.json() @@ -216,6 +244,7 @@ class QQOfficialClient: 'content': content, 'msg_type': 0, 'msg_id': msg_id, + 'msg_seq': await self.next_reply_msg_seq(msg_id), } response = await client.post(url, headers=headers, json=data) if response.status_code == 200: @@ -363,8 +392,7 @@ class QQOfficialClient: else: raise ValueError(f'Unsupported target_type: {target_type}') - self._msg_seq_counter += 1 - msg_seq = self._msg_seq_counter + msg_seq = await self.next_reply_msg_seq(msg_id) body = { 'msg_type': 7, 'media': {'file_info': file_info}, diff --git a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py index 47831a389..20a8913e0 100644 --- a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py +++ b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py @@ -280,6 +280,9 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl ctx['stream_msg_id'] = resp['id'] ctx['sent_length'] = len(ctx['accumulated_text']) ctx['index'] += 1 + # 每个 chunk 单独 HTTP 下发，msg_seq 必须递增，否则后续 chunk 会被 QQ + # 以 (msg_id, msg_seq) 去重 (40054005 消息被去重)。 + ctx['msg_seq'] += 1 if is_final: self._stream_ctx.pop(message_id, None) self._stream_ctx_ts.pop(message_id, None) diff --git a/tests/unit_tests/platform/test_qqofficial_eba_adapter.py b/tests/unit_tests/platform/test_qqofficial_eba_adapter.py index d979a033e..4e6d5df46 100644 --- a/tests/unit_tests/platform/test_qqofficial_eba_adapter.py +++ b/tests/unit_tests/platform/test_qqofficial_eba_adapter.py @@ -346,3 +346,92 @@ async def test_qqofficial_send_reply_stream_platform_api_and_unsupported(): with pytest.raises(NotSupportedError): await adapter.call_platform_api('missing', {}) + + +@pytest.mark.asyncio +async def test_qqofficial_client_next_reply_msg_seq_allocator(): + """msg_seq must increment per inbound msg_id to avoid 40054005 dedup (issue #2290).""" + from langbot.libs.qq_official_api.api import QQOfficialClient + + client = QQOfficialClient(secret='s', token='t', app_id='a', logger=DummyLogger()) + + # same msg_id -> strictly increasing 1, 2, 3 + assert [await client.next_reply_msg_seq('m1') for _ in range(3)] == [1, 2, 3] + # a different msg_id is tracked independently + assert await client.next_reply_msg_seq('m2') == 1 + # missing msg_id falls back to the global proactive counter, independent of per-id seqs + assert [await client.next_reply_msg_seq(None) for _ in range(2)] == [1, 2] + # concurrent allocations under the same msg_id are unique and gap-free + import asyncio as _asyncio + + concurrent = await _asyncio.gather(*[client.next_reply_msg_seq('m3') for _ in range(50)]) + assert sorted(concurrent) == list(range(1, 51)) + + +@pytest.mark.asyncio +async def test_qqofficial_text_send_carries_incrementing_msg_seq(monkeypatch): + """Multiple text replies under one inbound msg_id get distinct msg_seq values.""" + from langbot.libs.qq_official_api.api import QQOfficialClient + + client = QQOfficialClient(secret='s', token='t', app_id='a', logger=DummyLogger()) + client.access_token = 'token' + client.access_token_expiry_time = 9999999999 + + captured = [] + + class _Resp: + status_code = 200 + + def json(self): + return {'id': 'ok'} + + class _DummyAsyncClient: + def __init__(self, *a, **k): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, url, headers=None, json=None): + captured.append(json) + return _Resp() + + monkeypatch.setattr('langbot.libs.qq_official_api.api.httpx.AsyncClient', _DummyAsyncClient) + + await client.send_private_text_msg('user-1', 'first', 'inbound-1') + await client.send_private_text_msg('user-1', 'second', 'inbound-1') + await client.send_group_text_msg('group-1', 'third', 'inbound-1') + + seqs = [body['msg_seq'] for body in captured] + assert seqs == [1, 2, 3] + assert all(body['msg_id'] == 'inbound-1' for body in captured) + + +@pytest.mark.asyncio +async def test_qqofficial_stream_chunks_increment_msg_seq(): + """Each stream chunk must advance msg_seq so QQ does not dedup later chunks (issue #2290).""" + adapter = make_adapter() + source_event = await QQOfficialEventConverter().target2yiri(qq_event('C2C_MESSAGE_CREATE')) + + assert await adapter.create_message_card('msg-1', source_event) is True + bot_message = {'resp_message_id': 'msg-1'} + + for idx, text in enumerate(['hello ', 'world ', 'done']): + # reset the per-chunk rate limiter so each chunk actually sends + adapter._stream_ctx['msg-1']['last_update_ts'] = 0 + await adapter.reply_message_chunk( + source_event, + bot_message, + platform_message.MessageChain([platform_message.Plain(text=text)]), + is_final=(idx == 2), + ) + + stream_calls = [call[1] for call in adapter.bot.sent if call[0] == 'stream'] + seqs = [kwargs['msg_seq'] for kwargs in stream_calls] + # strictly increasing, no duplicates + assert seqs == sorted(set(seqs)) + assert len(seqs) == len(set(seqs)) + assert seqs[0] == 1