fix(qqofficial): always send the state=10 closing frame in C2C streaming

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()
 

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")