fix(pipeline): 流式输出时保留发送期装饰（引用 / @）

astrbot/core/pipeline/result_decorate/stage.py

@@ -123,15 +123,50 @@ def _split_text_by_words(self, text: str) -> list[str]:
                 result.append(seg)
         return result if result else [text]
 
+    def _stash_streaming_decorations(self, event: AstrMessageEvent) -> None:
+        """记录本次流式发送所需的发送期装饰（引用 / at）。
+
+        流式正文由 result.async_stream 逐段发出，result.chain 在 respond 阶段被忽略，
+        因此这里不做任何链修改，只把决策结果交给平台适配器，保持共享阶段的平台无关性。
+        """
+        try:
+            event.set_extra(
+                "_streaming_decorations",
+                {
+                    "quote": bool(self.reply_with_quote),
+                    "mention": bool(
+                        self.reply_with_mention
+                        and event.get_message_type() != MessageType.FRIEND_MESSAGE
+                    ),
+                },
+            )
+        except Exception:
+            logger.debug("stash streaming decorations failed", exc_info=True)
+
     async def process(
         self,
         event: AstrMessageEvent,
     ) -> None | AsyncGenerator[None, None]:
         result = event.get_result()
-        if result is None or not result.chain:
+        if result is None:
             return
 
-        if result.result_content_type == ResultContentType.STREAMING_RESULT:
+        # 流式结果的正文在 result.async_stream 中逐段发出，result.chain 在 respond
+        # 阶段会被忽略（respond 直接使用 async_stream），因此这里修改 chain 不会生效。
+        # 为了不改动共享阶段的接口、也不替各平台决定发送细节，只把「本次需要哪些
+        # 发送期装饰」记录到事件扩展字段上，由平台自身的 send_streaming 按需应用。
+        #
+        # 注意必须放在下面的 chain 判空之前：流式场景下 chain 恒为空，否则会提前返回，
+        # 装饰决策就无处安放（这也正是此前流式输出丢失引用 / @ 的原因）。
+        if result.result_content_type in (
+            ResultContentType.STREAMING_RESULT,
+            ResultContentType.STREAMING_FINISH,
+        ):
+            self._stash_streaming_decorations(event)
+            if result.result_content_type == ResultContentType.STREAMING_RESULT:
+                return
+
+        if not result.chain:
             return
 
         is_stream = result.result_content_type == ResultContentType.STREAMING_FINISH

astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py

@@ -27,7 +27,7 @@
 
 from astrbot.api import logger
 from astrbot.api.event import AstrMessageEvent, MessageChain
-from astrbot.api.message_components import At, File, Image, Plain, Record, Video
+from astrbot.api.message_components import At, File, Image, Plain, Record, Reply, Video
 from astrbot.api.platform import AstrBotMessage, Group, PlatformMetadata
 from astrbot.core.platform.sources.qqofficial.qqofficial_chunked_upload import (
     QQOFFICIAL_CHUNKED_UPLOAD_THRESHOLD,
@@ -237,6 +237,46 @@ async def _close_stream_segment(self, stream_payload: dict):
             self.send_buffer = MessageChain(chain=[Plain(text="\n")])
         return await self._post_send(stream=stream_payload)
 
+    def _apply_streaming_decorations(self, chain: MessageChain) -> MessageChain:
+        """为流式输出的首个分片返回一个带发送期装饰的新消息链。
+
+        流式正文由 result.async_stream 逐段发出，共享阶段无法修改 result.chain，
+        因此由平台在这里按需应用，避免改动共享阶段接口。
+
+        注意：传入的 chain 由上游生成器拥有，且可能在多个分片间复用，
+        因此这里**不修改它**，而是复制后再插入装饰；否则已插入的 Reply / At
+        会在后续分片被再次累积，导致重复引用。装饰只会应用一次
+        （取走决策后立即清空）。
+        """
+        try:
+            decorations = self.get_extra("_streaming_decorations")
+        except Exception:
+            return chain
+        if not isinstance(decorations, dict):
+            return chain
+        try:
+            self.set_extra("_streaming_decorations", None)
+        except Exception:
+            pass
+        # 与 result_decorate 的非流式路径保持一致：仅纯文本 / 图文消息可装饰。
+        if not all(isinstance(c, (Plain, Image)) for c in chain.chain):
+            return chain
+        try:
+            decorated = copy.deepcopy(chain)
+            if decorations.get("mention"):
+                decorated.chain.insert(
+                    0,
+                    At(qq=self.get_sender_id(), name=self.get_sender_name()),
+                )
+                if len(decorated.chain) > 1 and isinstance(decorated.chain[1], Plain):
+                    decorated.chain[1].text = chr(10) + decorated.chain[1].text
+            if decorations.get("quote"):
+                decorated.chain.insert(0, Reply(id=self.message_obj.message_id))
+            return decorated
+        except Exception:
+            logger.debug("apply streaming decorations failed", exc_info=True)
+            return chain
+
     async def send_streaming(self, generator, use_fallback: bool = False):
         """流式输出仅支持消息列表私聊（C2C），其他消息源退化为普通发送"""
         # 先标记事件层“已执行发送操作”，避免异常路径遗漏
@@ -252,6 +292,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
         try:
             async for chain in generator:
                 source = self.message_obj.raw_message
+                chain = self._apply_streaming_decorations(chain)
 
                 if not isinstance(source, botpy.message.C2CMessage):
                     # 非 C2C 场景：直接累积，最后统一发（拷贝 delta，避免引用丢首字）

tests/test_streaming_result_decoration.py

@@ -0,0 +1,181 @@
+"""流式输出的发送期装饰（引用 / @）回归测试。
+
+背景：result_decorate 在流式场景下无法修改 result.chain —— respond 阶段直接使用
+result.async_stream，且流式结果的 chain 恒为空。此前 STREAMING_RESULT 分支位于
+「chain 判空」之后，导致装饰决策无处安放，开启流式后 reply_with_quote /
+reply_with_mention 全部失效。
+
+修复：把决策记录提前到 chain 判空之前，写入 event 扩展字段，由平台适配器在
+send_streaming 的首个分片上按需应用。
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from astrbot.api.event import MessageChain
+from astrbot.api.message_components import Plain, Reply
+from astrbot.api.platform import (
+    AstrBotMessage,
+    MessageMember,
+    MessageType,
+    PlatformMetadata,
+)
+from astrbot.core.message.message_event_result import (
+    MessageEventResult,
+    ResultContentType,
+)
+from astrbot.core.pipeline.result_decorate.stage import ResultDecorateStage
+from astrbot.core.platform.sources.qqofficial.qqofficial_message_event import (
+    QQOfficialMessageEvent,
+)
+from astrbot.core.platform.sources.qqofficial.qqofficial_platform_adapter import (
+    PatchedGroupMessage,
+)
+
+REF_IDX = "REFIDX_streaming+/=="
+
+
+def _make_stage(quote: bool = True, mention: bool = False) -> ResultDecorateStage:
+    """只构造判定装饰所需的属性，避免拉起完整流水线。"""
+    stage = object.__new__(ResultDecorateStage)
+    stage.reply_with_quote = quote
+    stage.reply_with_mention = mention
+    return stage
+
+
+def _make_event() -> QQOfficialMessageEvent:
+    data: dict = {
+        "id": "ROBOT1.0_message-id",
+        "author": {"member_openid": "member-1"},
+        "group_openid": "group-1",
+        "content": "ping",
+        "timestamp": "0",
+        "message_scene": {"source": "default", "ext": [f"msg_idx={REF_IDX}"]},
+    }
+    raw = PatchedGroupMessage(api=None, event_id="event-1", data=data)
+
+    abm = AstrBotMessage()
+    abm.message_id = "ROBOT1.0_message-id"
+    abm.session_id = "group-1"
+    abm.group_id = "group-1"
+    abm.self_id = "bot-1"
+    abm.sender = MessageMember(user_id="member-1", nickname="u")
+    abm.type = MessageType.GROUP_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(post_group_message=AsyncMock()))
+    return QQOfficialMessageEvent(
+        message_str="ping",
+        message_obj=abm,
+        platform_meta=meta,
+        session_id="group-1",
+        bot=bot,  # type: ignore[arg-type]
+    )
+
+
+@pytest.mark.asyncio
+async def test_decorations_stashed_for_streaming_result_with_empty_chain() -> None:
+    """核心回归：流式结果 chain 为空时也必须记录装饰决策。"""
+    event = _make_event()
+    result = MessageEventResult().set_result_content_type(
+        ResultContentType.STREAMING_RESULT
+    )
+    assert not result.chain
+    event.set_result(result)
+
+    async for _ in _make_stage(quote=True).process(event):
+        pass
+
+    assert event.get_extra("_streaming_decorations") == {
+        "quote": True,
+        "mention": False,
+    }
+
+
+@pytest.mark.asyncio
+async def test_decorations_stashed_for_streaming_finish() -> None:
+    event = _make_event()
+    result = MessageEventResult().set_result_content_type(
+        ResultContentType.STREAMING_FINISH
+    )
+    event.set_result(result)
+
+    async for _ in _make_stage(quote=True).process(event):
+        pass
+
+    assert (event.get_extra("_streaming_decorations") or {}).get("quote") is True
+
+
+@pytest.mark.asyncio
+async def test_decoration_flags_follow_config() -> None:
+    event = _make_event()
+    event.set_result(
+        MessageEventResult().set_result_content_type(ResultContentType.STREAMING_RESULT)
+    )
+
+    async for _ in _make_stage(quote=False, mention=False).process(event):
+        pass
+
+    decorations = event.get_extra("_streaming_decorations")
+    assert decorations == {"quote": False, "mention": False}
+
+
+def test_apply_returns_decorated_copy() -> None:
+    """首个分片应被插入 Reply；决策被取走，避免后续分片重复引用。"""
+    event = _make_event()
+    event.set_extra("_streaming_decorations", {"quote": True, "mention": False})
+    chain = MessageChain(chain=[Plain("hi")])
+
+    decorated = event._apply_streaming_decorations(chain)
+
+    assert isinstance(decorated.chain[0], Reply)
+    assert decorated.chain[0].id == "ROBOT1.0_message-id"
+    assert event.get_extra("_streaming_decorations") is None
+
+
+def test_apply_does_not_mutate_source_chain() -> None:
+    """上游可能跨分片复用 chain，因此必须返回副本、不改动传入对象。"""
+    event = _make_event()
+    event.set_extra("_streaming_decorations", {"quote": True, "mention": False})
+    chain = MessageChain(chain=[Plain("hi")])
+
+    decorated = event._apply_streaming_decorations(chain)
+
+    assert decorated is not chain
+    assert len(chain.chain) == 1
+    assert isinstance(chain.chain[0], Plain)
+
+    # 复用同一个 chain 再走一遍：决策已被取走，不应再次装饰
+    again = event._apply_streaming_decorations(chain)
+    assert again is chain
+    assert len(again.chain) == 1
+
+
+def test_apply_is_noop_without_stashed_decorations() -> None:
+    event = _make_event()
+    chain = MessageChain(chain=[Plain("hi")])
+
+    result = event._apply_streaming_decorations(chain)
+
+    assert result is chain
+    assert len(chain.chain) == 1
+    assert isinstance(chain.chain[0], Plain)
+
+
+def test_apply_skips_chain_with_non_plain_components() -> None:
+    """与 result_decorate 的非流式路径一致：仅纯文本 / 图文消息可装饰。"""
+
+    event = _make_event()
+    event.set_extra("_streaming_decorations", {"quote": True, "mention": False})
+    chain = MessageChain(chain=[Reply(id="ROBOT1.0_other"), Plain("hi")])
+
+    result = event._apply_streaming_decorations(chain)
+
+    assert result is chain
+    assert len(chain.chain) == 2
+    assert chain.chain[0].id == "ROBOT1.0_other"