From 6878f719f7bd6233193b81b7eb72e1334675dd09 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Mon, 14 Sep 2026 01:00:05 +0800 Subject: [PATCH 1/2] fix(qqofficial): always send the state=10 closing frame in C2C streaming QQ's C2C streaming protocol ends a stream with a state=10 frame whose content must end with \n. AstrBot only sent it while unsent text remained in the buffer: once the throttled middle frames had flushed the full reply and the generator ended with an empty tail, no closing frame went out at all. QQ then timed the stream out and rolled the whole message back to the first packet, which is why a full 850-char reply ended up displaying only its first few characters (#10066), and why some streams hang in "generating" forever. Close every open segment: when the stream is open (a frame id exists) but the tail buffer is empty, send a minimal "\n" closing frame; when nothing was ever sent (no id), keep sending nothing. The tool_call break path closes the same way through a shared helper so an empty segment cannot orphan its stream either. Regression tests pin both paths against a fake _post_send: empty tail after a middle flush, and break arriving on an empty but open segment. --- .../qqofficial/qqofficial_message_event.py | 27 +++++-- tests/test_qqofficial_stream_buffer_copy.py | 74 +++++++++++++++++++ 2 files changed, 95 insertions(+), 6 deletions(-) diff --git a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py index 806034bb94..0680fc0b45 100644 --- a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py +++ b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py @@ -107,6 +107,21 @@ 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 + if not self.send_buffer or not self.send_buffer.chain: + 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），其他消息源退化为普通发送""" # 先标记事件层“已执行发送操作”，避免异常路径遗漏 @@ -132,9 +147,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 @@ -166,9 +182,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..33aa11992f 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() From 1db444cd3d5a01bacdd5dba18d662bca19a11116 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Mon, 14 Sep 2026 12:28:46 +0800 Subject: [PATCH 2/2] fix(qqofficial): treat an empty-Plain tail as empty for the closing frame Review on #10069: a buffer holding only Plain("") components has a truthy chain, so the "\n" fallback was skipped, _post_send_one rejected the empty parsed content, and no state=10 frame went out for an opened stream. Check for sendable content (non-empty text or a non-Plain component) instead of chain non-emptiness. Regression test pins the empty-Plain tail after a middle flush. --- .../qqofficial/qqofficial_message_event.py | 8 +++- tests/test_qqofficial_stream_buffer_copy.py | 37 +++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py index 2a3371ca6c..31da9cba85 100644 --- a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py +++ b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py @@ -224,7 +224,13 @@ async def _close_stream_segment(self, stream_payload: dict): 否则客户端等不到结束帧，最终只显示首包几个字。 """ stream_payload["state"] = 10 - if not self.send_buffer or not self.send_buffer.chain: + 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 diff --git a/tests/test_qqofficial_stream_buffer_copy.py b/tests/test_qqofficial_stream_buffer_copy.py index 33aa11992f..28c59859c4 100644 --- a/tests/test_qqofficial_stream_buffer_copy.py +++ b/tests/test_qqofficial_stream_buffer_copy.py @@ -371,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")