diff --git a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py index 706642224e..c328b44f79 100644 --- a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py +++ b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py @@ -228,6 +228,13 @@ async def send_streaming(self, generator, use_fallback: bool = False): source = ( self.message_obj.raw_message ) # 提前获取,避免 generator 为空时 NameError + # 累积已生成全文与已下发长度,用于断流兜底,避免回复被掐断 + full_text = "" + sent_len = 0 + + def _plain_of(chain: MessageChain) -> str: + return "".join(c.text for c in chain.chain if isinstance(c, Plain)) + try: async for chain in generator: source = self.message_obj.raw_message @@ -255,10 +262,15 @@ async def send_streaming(self, generator, use_fallback: bool = False): "reset": False, } last_edit_time = 0 + # 工具段结束后已下发前缀不再延续,重置全文追踪 + full_text = "" + sent_len = 0 continue # 累积内容(拷贝,避免上游复用 MessageChain 改写 buffer) self._append_stream_delta(chain) + # 追踪已生成全文(用于断流兜底) + full_text += _plain_of(chain) # 节流:按时间间隔发送中间分片 current_time = asyncio.get_running_loop().time() @@ -272,19 +284,49 @@ async def send_streaming(self, generator, use_fallback: bool = False): if ret_id is not None: stream_payload["id"] = ret_id last_edit_time = asyncio.get_running_loop().time() + # 记录已下发长度(buffer 已清空) + sent_len = len(full_text) self.send_buffer = None # 清空已发送的分片,避免下次重复发送旧内容 if isinstance(source, botpy.message.C2CMessage): - # 结束流式对话,发送 buffer 中剩余内容 - stream_payload["state"] = 10 - ret = await self._post_send(stream=stream_payload) + # 结束流式对话:把尚未下发的尾段以 state=10 补齐,避免回复被截断。 + # 若所有内容已通过节流分片完整下发(tail 为空),仍需发送一条 state=10 + # 收尾帧让 QQ 侧把流从「生成中」翻转为「完成」——空内容会被 _post_send + # 直接丢弃,故用 "\n" 作为收尾标记(仅多一个换行,不会重复正文)。 + tail = full_text[sent_len:] if full_text else "" + if stream_payload.get("id") is not None: + self.send_buffer = MessageChain(use_t2i_=False, type="segment") + self.send_buffer.chain.append(Plain(text=tail if tail else "\n")) + stream_payload["state"] = 10 + ret = await self._post_send(stream=stream_payload) + elif full_text: + # 未拿到流式消息 id(中间分片返回无 id):降级为普通消息补发全文 + self.send_buffer = MessageChain(use_t2i_=False, type="segment") + self.send_buffer.chain.append(Plain(text=full_text)) + ret = await self._post_send() else: ret = await self._post_send() except Exception as e: logger.error(f"发送流式消息时出错: {e}", exc_info=True) - # 避免累计内容在异常后被整包重复发送:仅清理缓存,不做非流式整包兜底 - # 如需兜底,应该只发送未发送 delta(后续可继续优化) + # 断流兜底:把已生成但未下发的尾段以 state=10 补齐到同一条流式消息, + # 保证用户看到完整回答,而不是冻结在最后一个分片。 + try: + tail = full_text[sent_len:] if full_text else "" + if isinstance(source, botpy.message.C2CMessage) and stream_payload.get( + "id" + ) is not None: + self.send_buffer = MessageChain(use_t2i_=False, type="segment") + self.send_buffer.chain.append(Plain(text=tail if tail else "\n")) + stream_payload["state"] = 10 + await self._post_send(stream=stream_payload) + elif full_text: + # 未拿到流式消息 id(非 C2C 或中间分片无返回 id):降级为普通消息补发全文 + self.send_buffer = MessageChain(use_t2i_=False, type="segment") + self.send_buffer.chain.append(Plain(text=full_text)) + await self._post_send() + except Exception as e2: + logger.error(f"断流兜底补发也失败: {e2}", exc_info=True) self.send_buffer = None return None diff --git a/tests/test_qqofficial_stream_finalize.py b/tests/test_qqofficial_stream_finalize.py new file mode 100644 index 0000000000..901222d84e --- /dev/null +++ b/tests/test_qqofficial_stream_finalize.py @@ -0,0 +1,165 @@ +"""Regression tests for QQ Official C2C streaming finalisation. + +Reported symptom: a streamed C2C reply grew to its full length and then, once +the platform-side stream timed out, the message reverted to the first fragment +(a few characters). The full reply was still stored in the conversation DB. + +Root cause: ``send_streaming`` clears ``self.send_buffer`` after every throttled +intermediate send. On normal completion the ``state=10`` final frame was sent +against an already-empty buffer, so ``_post_send`` returned early and the frame +was never delivered -> the C2C stream was never finalised. The ``except`` branch +additionally dropped any unsent tail. + +These tests pin the fixed behaviour: + * an intermediate-throttled send that consumed the whole buffer must still be + followed by one ``state=10`` frame; + * when throttling left an unsent tail, the final ``state=10`` frame must carry + exactly that tail (no duplication); + * when the upstream stream breaks mid-way, the unsent tail must still be + flushed with ``state=10`` on the same stream. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import patch + +import botpy.message +import pytest + +from astrbot.api.event import MessageChain +from astrbot.api.message_components import Plain +from astrbot.api.platform import ( + AstrBotMessage, + MessageMember, + MessageType, + PlatformMetadata, +) +from astrbot.core.platform.sources.qqofficial.qqofficial_message_event import ( + QQOfficialMessageEvent, +) + + +def _buffer_text(event: QQOfficialMessageEvent) -> str: + if not event.send_buffer: + return "" + return "".join(c.text for c in event.send_buffer.chain if isinstance(c, Plain)) + + +def _make_c2c_event() -> QQOfficialMessageEvent: + raw = botpy.message.C2CMessage( + api=None, + event_id="event-1", + data={ + "id": "msg-1", + "author": {"user_openid": "user-1"}, + "content": "ping", + "timestamp": "0", + }, + ) + abm = AstrBotMessage() + abm.message_id = "msg-1" + abm.session_id = "user-1" + abm.self_id = "bot-1" + abm.sender = MessageMember(user_id="user-1", nickname="u") + abm.type = MessageType.FRIEND_MESSAGE + abm.message_str = "ping" + abm.message = [] + abm.raw_message = raw + meta = PlatformMetadata(name="qq_official", description="t", id="qq_official") + bot = SimpleNamespace(api=SimpleNamespace()) + return QQOfficialMessageEvent( + message_str="ping", + message_obj=abm, + platform_meta=meta, + session_id="user-1", + bot=bot, # type: ignore[arg-type] + ) + + +def _capturing_post_send(event: QQOfficialMessageEvent): + """Return (side_effect, calls) capturing buffer text + stream payload.""" + calls: list[tuple[str, dict | None]] = [] + + async def fake_post_send(stream=None): + # NOTE: send_streaming mutates stream_payload in place, so snapshot it. + calls.append((_buffer_text(event), dict(stream) if stream else None)) + event.send_buffer = None + return {"id": f"stream-{len(calls)}"} + + return fake_post_send, calls + + +@pytest.mark.asyncio +async def test_c2c_stream_is_finalised_when_buffer_already_flushed() -> None: + """A throttled send that consumed everything must still emit state=10.""" + event = _make_c2c_event() + fake_post_send, calls = _capturing_post_send(event) + + async def gen(): + yield MessageChain().message("完整回复") + + with ( + patch.object(event, "_post_send", side_effect=fake_post_send), + patch("asyncio.get_running_loop") as mock_loop, + ): + # monotonic time is large, so the first delta triggers a throttled send + mock_loop.return_value.time.return_value = 100.0 + await event.send_streaming(gen()) + + # 1st = throttled state=1 with the whole text; 2nd = empty-tail state=10. + assert len(calls) == 2 + assert calls[0][1] is not None and calls[0][1]["state"] == 1 + assert calls[0][0] == "完整回复" + assert calls[1][1] is not None and calls[1][1]["state"] == 10 + # empty tail is delivered as a "\n" marker so _post_send does not drop it + assert calls[1][0] == "\n" + + +@pytest.mark.asyncio +async def test_c2c_stream_final_frame_carries_only_unsent_tail() -> None: + """Content throttled out mid-stream must appear exactly once, in state=10.""" + event = _make_c2c_event() + fake_post_send, calls = _capturing_post_send(event) + + async def gen(): + yield MessageChain().message("前半") + yield MessageChain().message("后半") + + with ( + patch.object(event, "_post_send", side_effect=fake_post_send), + patch("asyncio.get_running_loop") as mock_loop, + ): + # constant time: first delta flushes, second is throttled out (no send) + mock_loop.return_value.time.return_value = 100.0 + await event.send_streaming(gen()) + + assert len(calls) == 2 + assert calls[0][1] is not None and calls[0][1]["state"] == 1 + assert calls[0][0] == "前半" + assert calls[1][1] is not None and calls[1][1]["state"] == 10 + assert calls[1][0] == "后半" + + +@pytest.mark.asyncio +async def test_c2c_stream_flushes_tail_when_upstream_breaks() -> None: + """An aborted generator must not leave the stream stuck in state=1.""" + event = _make_c2c_event() + fake_post_send, calls = _capturing_post_send(event) + + async def gen(): + yield MessageChain().message("前半") + yield MessageChain().message("后半") + raise RuntimeError("upstream stream broke") + + with ( + patch.object(event, "_post_send", side_effect=fake_post_send), + patch("asyncio.get_running_loop") as mock_loop, + ): + mock_loop.return_value.time.return_value = 100.0 + await event.send_streaming(gen()) + + assert len(calls) == 2 + assert calls[0][1] is not None and calls[0][1]["state"] == 1 + assert calls[1][1] is not None and calls[1][1]["state"] == 10 + assert calls[1][0] == "后半"