diff --git a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py index 706642224e..31da9cba85 100644 --- a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py +++ b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py @@ -216,6 +216,27 @@ async def send(self, message: MessageChain) -> None: self.send_buffer = message await self._post_send() + async def _close_stream_segment(self, stream_payload: dict): + """以 state=10 收尾当前流式段；流已开但 buffer 恰好为空时补最小收尾帧。 + + QQ C2C 流式协议缺 state=10 会在超时后把整段回滚到首包（#10066）： + 中间分片已把全文发完、结尾没有剩余内容时也必须补一个 "\n" 收尾帧， + 否则客户端等不到结束帧，最终只显示首包几个字。 + """ + stream_payload["state"] = 10 + has_content = self.send_buffer is not None and any( + (isinstance(c, Plain) and c.text) or not isinstance(c, Plain) + for c in self.send_buffer.chain + ) + if not has_content: + # 只有空 Plain 的 buffer 也算空：_post_send_one 会拒掉空文本， + # 收尾帧照样缺席（#10069 review） + if stream_payload.get("id") is None: + # 从未发出任何分片，无流可收 + return None + self.send_buffer = MessageChain(chain=[Plain(text="\n")]) + return await self._post_send(stream=stream_payload) + async def send_streaming(self, generator, use_fallback: bool = False): """流式输出仅支持消息列表私聊（C2C），其他消息源退化为普通发送""" # 先标记事件层“已执行发送操作”，避免异常路径遗漏 @@ -241,9 +262,10 @@ async def send_streaming(self, generator, use_fallback: bool = False): # tool_call break 信号：工具开始执行，先把已有 buffer 以 state=10 结束当前流式段 if chain.type == "break": - if self.send_buffer: - stream_payload["state"] = 10 - ret = await self._post_send(stream=stream_payload) + if (self.send_buffer and self.send_buffer.chain) or ( + stream_payload.get("id") is not None + ): + ret = await self._close_stream_segment(stream_payload) ret_id = self._extract_response_message_id(ret) if ret_id is not None: stream_payload["id"] = ret_id @@ -275,9 +297,8 @@ async def send_streaming(self, generator, use_fallback: bool = False): self.send_buffer = None # 清空已发送的分片，避免下次重复发送旧内容 if isinstance(source, botpy.message.C2CMessage): - # 结束流式对话，发送 buffer 中剩余内容 - stream_payload["state"] = 10 - ret = await self._post_send(stream=stream_payload) + # 结束流式对话，发送 buffer 中剩余内容（空尾也要补收尾帧） + ret = await self._close_stream_segment(stream_payload) else: ret = await self._post_send() diff --git a/tests/test_qqofficial_stream_buffer_copy.py b/tests/test_qqofficial_stream_buffer_copy.py index 4a04f6ab1b..28c59859c4 100644 --- a/tests/test_qqofficial_stream_buffer_copy.py +++ b/tests/test_qqofficial_stream_buffer_copy.py @@ -279,6 +279,80 @@ async def gen(): assert sent_texts[0] == "不稀罕" +@pytest.mark.asyncio +async def test_c2c_stream_closes_with_state10_when_tail_buffer_empty() -> None: + """#10066: 中间分片把全文发完后生成器收尾时 buffer 为空，也必须补 state=10 + 收尾帧，否则 QQ 超时把整段回滚到首包几个字。""" + event = _make_c2c_event() + frames: list[tuple[int | None, str]] = [] + + async def fake_post_send(stream=None): + parts = [] + if event.send_buffer: + for c in event.send_buffer.chain: + if isinstance(c, Plain) and c.text: + parts.append(c.text) + frames.append((stream.get("state") if stream else None, "".join(parts))) + event.send_buffer = None + return {"id": "stream-1"} + + async def gen(): + yield MessageChain().message("不") + yield MessageChain().message("稀") + # 之后没有新 delta：生成器以空 buffer 收尾 + + from unittest.mock import patch + + with ( + patch.object(event, "_post_send", side_effect=fake_post_send), + patch("asyncio.get_running_loop") as mock_loop, + ): + # 第一个 delta 在 0.5s（不触发节流），第二个在 2.0s（触发中间分片并清空 buffer） + mock_loop.return_value.time.side_effect = [0.5, 2.0, 2.0, 2.0] + await event.send_streaming(gen()) + + # 中间分片带走全文后，收尾帧仍要以 state=10 发出（最小 "\n" 收尾） + assert (1, "不稀") in frames + assert frames[-1] == (10, "\n") + + +@pytest.mark.asyncio +async def test_c2c_stream_break_closes_open_segment_with_empty_buffer() -> None: + """#10066 同族：tool_call break 到达时 buffer 恰好为空但流已开，也要先补 + state=10 收尾再开新段，否则该段同样会被 QQ 超时回滚。""" + event = _make_c2c_event() + frames: list[tuple[int | None, str]] = [] + + async def fake_post_send(stream=None): + parts = [] + if event.send_buffer: + for c in event.send_buffer.chain: + if isinstance(c, Plain) and c.text: + parts.append(c.text) + frames.append((stream.get("state") if stream else None, "".join(parts))) + event.send_buffer = None + return {"id": "stream-1"} + + async def gen(): + yield MessageChain().message("首段文本") + yield MessageChain(type="break") + + from unittest.mock import patch + + with ( + patch.object(event, "_post_send", side_effect=fake_post_send), + patch("asyncio.get_running_loop") as mock_loop, + ): + # 2.0s 到达：首个 delta 立即触发中间分片并清空 buffer + mock_loop.return_value.time.side_effect = [2.0, 2.0, 2.0, 2.0] + await event.send_streaming(gen()) + + assert frames[0] == (1, "首段文本") + assert frames[1] == (10, "\n") + # break 后 buffer 空且新段未开：结尾不再多发收尾帧 + assert len(frames) == 2 + + @pytest.mark.asyncio async def test_group_stream_sends_once_after_all_deltas() -> None: event = _make_group_event() @@ -297,3 +371,40 @@ async def gen(): await event.send_streaming(gen()) assert calls == 1 + + +@pytest.mark.asyncio +async def test_c2c_stream_closes_when_tail_is_empty_plain() -> None: + """#10069 review: 结尾只剩空 Plain("") 的 buffer 也被视为空，照样补 + state=10 收尾帧；否则 _post_send_one 拒掉空文本，流照样被超时回滚。""" + event = _make_c2c_event() + frames: list[tuple[int | None, str]] = [] + + async def fake_post_send(stream=None): + parts = [] + if event.send_buffer: + for c in event.send_buffer.chain: + if isinstance(c, Plain) and c.text: + parts.append(c.text) + frames.append((stream.get("state") if stream else None, "".join(parts))) + event.send_buffer = None + return {"id": "stream-1"} + + async def gen(): + yield MessageChain().message("不") + yield MessageChain().message("稀") + yield MessageChain(chain=[Plain("")]) # 空 delta 收尾 + + from unittest.mock import patch + + with ( + patch.object(event, "_post_send", side_effect=fake_post_send), + patch("asyncio.get_running_loop") as mock_loop, + ): + # 2.0s 触发中间分片冲掉全文，之后只剩空 delta + mock_loop.return_value.time.side_effect = [0.5, 2.0, 2.0, 2.0] + await event.send_streaming(gen()) + + # 中间分片带走全文，空 Plain 尾也照样补 state=10 最小收尾帧 + assert frames[0] == (1, "不稀") + assert frames[-1] == (10, "\n")