fix(qqofficial): finalize C2C stream to prevent replies reverting to the first fragment

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

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] == "后半"