fix: dispatch OnAfterMessageSentEvent after streaming delivery

astrbot/core/pipeline/respond/stage.py

@@ -127,6 +127,12 @@ async def _is_empty_message_chain(self, chain: list[BaseMessageComponent]) -> bo
         # 如果所有组件都为空
         return True
 
+    async def _after_sent_cleanup(self, event: AstrMessageEvent) -> bool:
+        if await call_event_hook(event, EventType.OnAfterMessageSentEvent):
+            return True
+        event.clear_result()
+        return False
+
     def is_seg_reply_required(self, event: AstrMessageEvent) -> bool:
         """检查是否需要分段回复"""
         if not self.enable_seg:
@@ -227,8 +233,11 @@ async def process(
                 == "realtime_segmenting"
             )
             logger.info(f"Applying streaming output ({event.get_platform_id()}).")
-            await event.send_streaming(result.async_stream, realtime_segmenting)
-            return
+            delivered = await event.send_streaming(result.async_stream, realtime_segmenting)
+            if delivered:
+                if await self._after_sent_cleanup(event):
+                    return
+                return
         if len(result.chain) > 0:
             # 检查路径映射
             if mappings := self.platform_settings.get("path_mapping", []):
@@ -325,7 +334,5 @@ async def process(
                             exc_info=True,
                         )
 
-        if await call_event_hook(event, EventType.OnAfterMessageSentEvent):
+        if await self._after_sent_cleanup(event):
             return
-
-        event.clear_result()

astrbot/core/platform/astr_message_event.py

@@ -286,7 +286,7 @@ async def send_streaming(
         self,
         generator: AsyncGenerator[MessageChain, None],
         use_fallback: bool = False,
-    ) -> None:
+    ) -> bool:
         """发送流式消息到消息平台，使用异步生成器。
         目前仅支持: telegram，qq official 私聊。
         Fallback仅支持 aiocqhttp。
@@ -295,6 +295,7 @@ async def send_streaming(
             Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name),
         )
         self._has_send_oper = True
+        return True
 
     async def send_typing(self) -> None:
         """发送输入中状态。

astrbot/core/platform/sources/aiocqhttp/aiocqhttp_message_event.py

@@ -210,7 +210,7 @@ async def send_streaming(
                 else:
                     buffer.chain.extend(chain.chain)
             if not buffer:
-                return None
+                return False
             buffer.squash_plain()
             await self.send(buffer)
             return await super().send_streaming(generator, use_fallback)

astrbot/core/platform/sources/dingtalk/dingtalk_event.py

@@ -37,7 +37,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
             else:
                 buffer.chain.extend(chain.chain)
         if not buffer:
-            return None
+            return False
         buffer.squash_plain()
         await self.send(buffer)
         return await super().send_streaming(generator, use_fallback)

astrbot/core/platform/sources/discord/discord_platform_event.py

@@ -118,7 +118,7 @@ async def send_streaming(
             else:
                 buffer.chain.extend(chain.chain)
         if not buffer:
-            return None
+            return False
         buffer.squash_plain()
         await self.send(buffer)
         return await super().send_streaming(generator, use_fallback)

astrbot/core/platform/sources/lark/lark_event.py

@@ -1294,7 +1294,7 @@ async def _flush_and_close_card() -> None:
                     )
                 )
                 self._has_send_oper = True
-            return
+            return False
 
         await _flush_and_close_card()
 
@@ -1303,3 +1303,4 @@ async def _flush_and_close_card() -> None:
             Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
         )
         self._has_send_oper = True
+        return True

astrbot/core/platform/sources/line/line_event.py

@@ -260,7 +260,7 @@ async def send_streaming(
                 else:
                     buffer.chain.extend(chain.chain)
             if not buffer:
-                return None
+                return False
             buffer.squash_plain()
             await self.send(buffer)
             return await super().send_streaming(generator, use_fallback)

astrbot/core/platform/sources/mattermost/mattermost_event.py

@@ -43,7 +43,7 @@ async def send_streaming(
                 else:
                     message_buffer.chain.extend(chain.chain)
             if not message_buffer:
-                return None
+                return False
             message_buffer.squash_plain()
             await self.send(message_buffer)
             return None

astrbot/core/platform/sources/misskey/misskey_event.py

@@ -139,7 +139,7 @@ async def send_streaming(
                 else:
                     buffer.chain.extend(chain.chain)
             if not buffer:
-                return None
+                return False
             buffer.squash_plain()
             await self.send(buffer)
             return await super().send_streaming(generator, use_fallback)

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

@@ -308,7 +308,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
             # 如需兜底，应该只发送未发送 delta（后续可继续优化）
             self.send_buffer = None
 
-        return None
+        return True
 
     def _append_stream_delta(self, chain: MessageChain) -> None:
         """Append stream delta into an owned buffer (copy components).

astrbot/core/platform/sources/slack/slack_event.py

@@ -186,7 +186,7 @@ async def send_streaming(
                 else:
                     buffer.chain.extend(chain.chain)
             if not buffer:
-                return None
+                return False
             buffer.squash_plain()
             await self.send(buffer)
             return await super().send_streaming(generator, use_fallback)

astrbot/core/platform/sources/telegram/tg_event.py

@@ -622,6 +622,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
             Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name),
         )
         self._has_send_oper = True
+        return True
 
     async def _send_streaming_draft(
         self,

astrbot/core/platform/sources/webchat/webchat_event.py

@@ -222,7 +222,7 @@ async def send_streaming(self, generator, use_fallback: bool = False) -> None:
 
                 accepted = await webchat_queue_mgr.put_back_queue(request_id, payload)
                 if not accepted:
-                    return
+                    return False
                 continue
 
             # if chain.type == "break" and final_data:

astrbot/core/platform/sources/wecom/wecom_event.py

@@ -306,7 +306,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
             else:
                 buffer.chain.extend(chain.chain)
         if not buffer:
-            return None
+            return False
         buffer.squash_plain()
         await self.send(buffer)
         return await super().send_streaming(generator, use_fallback)

astrbot/core/platform/sources/wecom_ai_bot/wecomai_event.py

@@ -214,7 +214,7 @@ async def send(self, message: MessageChain | None) -> None:
         )
         await super().send(MessageChain([]))
 
-    async def send_streaming(self, generator, use_fallback=False) -> None:
+    async def send_streaming(self, generator, use_fallback=False) -> bool:
         """流式发送消息，参考webchat的send_streaming设计"""
         final_data = ""
         raw = self.message_obj.raw_message
@@ -253,7 +253,7 @@ async def send_streaming(self, generator, use_fallback=False) -> None:
                     },
                 )
                 await super().send_streaming(generator, use_fallback)
-                return
+                return True
 
             increment_plain = ""
             last_stream_update_time = 0.0
@@ -298,7 +298,7 @@ async def send_streaming(self, generator, use_fallback=False) -> None:
                 },
             )
             await super().send_streaming(generator, use_fallback)
-            return
+            return True
 
         if self.only_use_webhook_url_to_send and self.webhook_client:
             merged_chain = MessageChain([])
@@ -308,15 +308,15 @@ async def send_streaming(self, generator, use_fallback=False) -> None:
             await self.webhook_client.send_message_chain(merged_chain)
             await self._mark_stream_complete(stream_id)
             await super().send_streaming(generator, use_fallback)
-            return
+            return True
 
         # 企业微信智能机器人不支持增量发送，因此我们需要在这里将增量内容累积起来，按间隔推送
         increment_plain = ""
         last_stream_update_time = 0.0
 
         async def enqueue_stream_plain(text: str) -> None:
             if not text:
-                return
+                return True
             await back_queue.put(
                 {
                     "type": "plain",

astrbot/core/platform/sources/weixin_oc/weixin_oc_event.py

@@ -86,7 +86,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
                 else:
                     buffer.chain.extend(chain.chain)
             if not buffer:
-                return None
+                return False
             await self.send(buffer)
             return await super().send_streaming(generator, use_fallback)
 

astrbot/core/platform/sources/weixin_official_account/weixin_offacc_event.py

@@ -183,7 +183,7 @@ async def send_streaming(self, generator, use_fallback: bool = False):
             else:
                 buffer.chain.extend(chain.chain)
         if not buffer:
-            return None
+            return False
         buffer.squash_plain()
         await self.send(buffer)
         return await super().send_streaming(generator, use_fallback)