From 2f91e54d8a90a5a81156c3d1ef6c4684ad19d080 Mon Sep 17 00:00:00 2001 From: bcbcbc Date: Thu, 17 Sep 2026 21:09:01 +0800 Subject: [PATCH 1/2] fix: dispatch OnAfterMessageSentEvent after streaming delivery --- astrbot/core/pipeline/respond/stage.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/astrbot/core/pipeline/respond/stage.py b/astrbot/core/pipeline/respond/stage.py index 66c6ba5419..e6464feb53 100644 --- a/astrbot/core/pipeline/respond/stage.py +++ b/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: @@ -228,6 +234,8 @@ async def process( ) logger.info(f"Applying streaming output ({event.get_platform_id()}).") await event.send_streaming(result.async_stream, realtime_segmenting) + if await self._after_sent_cleanup(event): + return return if len(result.chain) > 0: # 检查路径映射 @@ -325,7 +333,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() From 450b02c4b35eb4759caed3c3611c893c48762f4b Mon Sep 17 00:00:00 2001 From: root Date: Fri, 18 Sep 2026 16:43:22 +0800 Subject: [PATCH 2/2] fix: only dispatch OnAfterMessageSentEvent when streaming actually delivered --- astrbot/core/pipeline/respond/stage.py | 7 ++++--- astrbot/core/platform/astr_message_event.py | 3 ++- .../sources/aiocqhttp/aiocqhttp_message_event.py | 2 +- .../core/platform/sources/dingtalk/dingtalk_event.py | 2 +- .../platform/sources/discord/discord_platform_event.py | 2 +- astrbot/core/platform/sources/lark/lark_event.py | 3 ++- astrbot/core/platform/sources/line/line_event.py | 2 +- .../platform/sources/mattermost/mattermost_event.py | 2 +- astrbot/core/platform/sources/misskey/misskey_event.py | 2 +- .../sources/qqofficial/qqofficial_message_event.py | 2 +- astrbot/core/platform/sources/slack/slack_event.py | 2 +- astrbot/core/platform/sources/telegram/tg_event.py | 1 + astrbot/core/platform/sources/webchat/webchat_event.py | 2 +- astrbot/core/platform/sources/wecom/wecom_event.py | 2 +- .../platform/sources/wecom_ai_bot/wecomai_event.py | 10 +++++----- .../core/platform/sources/weixin_oc/weixin_oc_event.py | 2 +- .../weixin_official_account/weixin_offacc_event.py | 2 +- 17 files changed, 26 insertions(+), 22 deletions(-) diff --git a/astrbot/core/pipeline/respond/stage.py b/astrbot/core/pipeline/respond/stage.py index e6464feb53..eb5beb0438 100644 --- a/astrbot/core/pipeline/respond/stage.py +++ b/astrbot/core/pipeline/respond/stage.py @@ -233,10 +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) - if await self._after_sent_cleanup(event): + delivered = await event.send_streaming(result.async_stream, realtime_segmenting) + if delivered: + if await self._after_sent_cleanup(event): + return return - return if len(result.chain) > 0: # 检查路径映射 if mappings := self.platform_settings.get("path_mapping", []): diff --git a/astrbot/core/platform/astr_message_event.py b/astrbot/core/platform/astr_message_event.py index 0c8ad246e0..81fdf3a31b 100644 --- a/astrbot/core/platform/astr_message_event.py +++ b/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: """发送输入中状态。 diff --git a/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_message_event.py b/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_message_event.py index 8e2b008eb6..720595c5c3 100644 --- a/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_message_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/dingtalk/dingtalk_event.py b/astrbot/core/platform/sources/dingtalk/dingtalk_event.py index 3331c51476..7aaed10eba 100644 --- a/astrbot/core/platform/sources/dingtalk/dingtalk_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/discord/discord_platform_event.py b/astrbot/core/platform/sources/discord/discord_platform_event.py index d622b724d8..816b64eecb 100644 --- a/astrbot/core/platform/sources/discord/discord_platform_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/lark/lark_event.py b/astrbot/core/platform/sources/lark/lark_event.py index 6227d38f83..f6198bcee5 100644 --- a/astrbot/core/platform/sources/lark/lark_event.py +++ b/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 diff --git a/astrbot/core/platform/sources/line/line_event.py b/astrbot/core/platform/sources/line/line_event.py index c107207451..9fedccb5b5 100644 --- a/astrbot/core/platform/sources/line/line_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/mattermost/mattermost_event.py b/astrbot/core/platform/sources/mattermost/mattermost_event.py index 850274c2b1..1b0fff448e 100644 --- a/astrbot/core/platform/sources/mattermost/mattermost_event.py +++ b/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 diff --git a/astrbot/core/platform/sources/misskey/misskey_event.py b/astrbot/core/platform/sources/misskey/misskey_event.py index 025d291cdf..b5e128a4e3 100644 --- a/astrbot/core/platform/sources/misskey/misskey_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py b/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py index 31da9cba85..ef59d180ed 100644 --- a/astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py +++ b/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). diff --git a/astrbot/core/platform/sources/slack/slack_event.py b/astrbot/core/platform/sources/slack/slack_event.py index 77b21e814e..9e3a276303 100644 --- a/astrbot/core/platform/sources/slack/slack_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/telegram/tg_event.py b/astrbot/core/platform/sources/telegram/tg_event.py index c1eaaf135b..276dbfc943 100644 --- a/astrbot/core/platform/sources/telegram/tg_event.py +++ b/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, diff --git a/astrbot/core/platform/sources/webchat/webchat_event.py b/astrbot/core/platform/sources/webchat/webchat_event.py index 11c2d35128..754d99978e 100644 --- a/astrbot/core/platform/sources/webchat/webchat_event.py +++ b/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: diff --git a/astrbot/core/platform/sources/wecom/wecom_event.py b/astrbot/core/platform/sources/wecom/wecom_event.py index 265b41014f..c4f07f4396 100644 --- a/astrbot/core/platform/sources/wecom/wecom_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/wecom_ai_bot/wecomai_event.py b/astrbot/core/platform/sources/wecom_ai_bot/wecomai_event.py index 74d120f5f3..6191146148 100644 --- a/astrbot/core/platform/sources/wecom_ai_bot/wecomai_event.py +++ b/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,7 +308,7 @@ 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 = "" @@ -316,7 +316,7 @@ async def send_streaming(self, generator, use_fallback=False) -> None: async def enqueue_stream_plain(text: str) -> None: if not text: - return + return True await back_queue.put( { "type": "plain", diff --git a/astrbot/core/platform/sources/weixin_oc/weixin_oc_event.py b/astrbot/core/platform/sources/weixin_oc/weixin_oc_event.py index 84a19a9e7b..16a8bdc24d 100644 --- a/astrbot/core/platform/sources/weixin_oc/weixin_oc_event.py +++ b/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) diff --git a/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_event.py b/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_event.py index ae536593c5..5bd95c9d28 100644 --- a/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_event.py +++ b/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)