From ec71ec4a0b5971e5750d30d0edd32e3b487a0ed5 Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Tue, 6 Oct 2026 21:12:52 +0800 Subject: [PATCH] fix(sdk-bridge): mark in-flight events on proactive send In-process event.send() sets _has_send_oper, which makes the pipeline skip the LLM stage after a handler replies directly. The isolated path lost that signal because the compat send went through the message.send RPC without touching the host event, so isolated plugins fell through to the LLM after replying (e.g. astrna's astrna_issue_latest guard message). Track core events being served by the Runner (keyed by umo) in the bridge, and have MessageSendService mark them via a callback after a successful send, covering message, command, and tool stubs. --- astrbot/core/star/sdk_bridge/bridge.py | 80 ++++++++++++++----- .../core/star/sdk_bridge/services/messages.py | 13 ++- tests/unit/test_sdk_bridge.py | 5 ++ 3 files changed, 76 insertions(+), 22 deletions(-) diff --git a/astrbot/core/star/sdk_bridge/bridge.py b/astrbot/core/star/sdk_bridge/bridge.py index db7a36796..a128ea1ca 100644 --- a/astrbot/core/star/sdk_bridge/bridge.py +++ b/astrbot/core/star/sdk_bridge/bridge.py @@ -26,7 +26,7 @@ from astrbot.core.star.star_handler import ( star_handlers_registry, ) -from .convert import apply_sdk_result, to_sdk_event +from .convert import apply_sdk_result, to_sdk_event, to_umo_string from .detect import _BRIDGE_MODULE_PREFIX from .detect import is_isolated_legacy_dir as is_isolated_legacy_dir from .detect import is_sdk_plugin_dir as is_sdk_plugin_dir @@ -202,6 +202,10 @@ class SDKPluginBridge: self._pending_command_links: list[tuple[str, Any]] = [] self._web_routes: list[tuple[str, tuple[str, ...]]] = [] self._views_manifest: dict = {} + # Core events currently being served by the Runner, keyed by umo + # string. Proactive sends mark them so the pipeline skips the LLM + # stage, mirroring in-process event.send(). + self._inflight_events: dict[str, list[Any]] = {} async def start(self) -> None: """Start the Runner subprocess and register pipeline handlers. @@ -652,6 +656,11 @@ class SDKPluginBridge: ) return filters + def _mark_event_sent(self, umo: Any) -> None: + """Mirror in-process event.send() on every in-flight event for umo.""" + for event in self._inflight_events.get(to_umo_string(umo), ()): + event._has_send_oper = True + def _make_message_stub(self, handler_id: str) -> Any: """Create the async generator bridging pipeline and message invocation. @@ -666,14 +675,23 @@ class SDKPluginBridge: async def stub(event, **_kwargs): client = bridge._require_client() sdk_event = to_sdk_event(event) - async for result in client.invoke(handler_id, sdk_event): - if result is not None: - apply_sdk_result( - event, - result, - resolve_asset=bridge._require_asset_store().resolve, - ) - yield + key = to_umo_string(sdk_event.umo) + bridge._inflight_events.setdefault(key, []).append(event) + try: + async for result in client.invoke(handler_id, sdk_event): + if result is not None: + apply_sdk_result( + event, + result, + resolve_asset=bridge._require_asset_store().resolve, + ) + yield + finally: + events = bridge._inflight_events.get(key) + if events and event in events: + events.remove(event) + if not events: + bridge._inflight_events.pop(key, None) return stub @@ -823,7 +841,18 @@ class SDKPluginBridge: umo=sdk_event.umo if sdk_event is not None else None, event=sdk_event, ) - return await client.invoke_tool(handler_id, call, kwargs) + key = to_umo_string(sdk_event.umo) if sdk_event is not None else None + if key is not None: + bridge._inflight_events.setdefault(key, []).append(event) + try: + return await client.invoke_tool(handler_id, call, kwargs) + finally: + if key is not None: + events = bridge._inflight_events.get(key) + if events and event in events: + events.remove(event) + if not events: + bridge._inflight_events.pop(key, None) return stub @@ -846,16 +875,25 @@ class SDKPluginBridge: command_path=command_path, arguments=kwargs, ) - async for result in client.invoke(handler_id, sdk_event, **kwargs): - if result is not None: - apply_sdk_result( - event, - result, - resolve_asset=bridge._require_asset_store().resolve, - ) - # Each resume of this generator advances the client stream, - # which acknowledges the previous plugin yield. - yield + key = to_umo_string(sdk_event.umo) + bridge._inflight_events.setdefault(key, []).append(event) + try: + async for result in client.invoke(handler_id, sdk_event, **kwargs): + if result is not None: + apply_sdk_result( + event, + result, + resolve_asset=bridge._require_asset_store().resolve, + ) + # Each resume of this generator advances the client stream, + # which acknowledges the previous plugin yield. + yield + finally: + events = bridge._inflight_events.get(key) + if events and event in events: + events.remove(event) + if not events: + bridge._inflight_events.pop(key, None) return stub @@ -889,7 +927,7 @@ class SDKPluginBridge: store = self._require_asset_store() services: tuple[HostService, ...] = ( StorageService(self.plugin_id), - MessageSendService(self.context, store), + MessageSendService(self.context, store, mark_sent=self._mark_event_sent), ConversationReadService(self.context), ConversationWriteService(self.context), AssetTransferService(store), diff --git a/astrbot/core/star/sdk_bridge/services/messages.py b/astrbot/core/star/sdk_bridge/services/messages.py index 55ddbe349..e0e662217 100644 --- a/astrbot/core/star/sdk_bridge/services/messages.py +++ b/astrbot/core/star/sdk_bridge/services/messages.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Callable from typing import TYPE_CHECKING, Any from astrbot_sdk.errors import InvalidRequest, NotFound @@ -19,15 +20,23 @@ class MessageSendService: capability_id = "message.send" - def __init__(self, context: Context, store: AssetStore) -> None: + def __init__( + self, + context: Context, + store: AssetStore, + mark_sent: Callable[[Any], None] | None = None, + ) -> None: """Initialize the service. Args: context: AstrBot star context used for platform access. store: Asset store used to resolve asset references. + mark_sent: Callback marking in-flight events of the session as + having sent, mirroring in-process event.send(). """ self._context = context self._store = store + self._mark_sent = mark_sent async def handle(self, operation: str, payload: dict[str, Any]) -> Any: """Serve the send operation.""" @@ -50,4 +59,6 @@ class MessageSendService: ) if not sent: raise NotFound(f"platform not found: {umo.platform_id}") + if self._mark_sent is not None: + self._mark_sent(umo) return {"message_id": ""} diff --git a/tests/unit/test_sdk_bridge.py b/tests/unit/test_sdk_bridge.py index 9e08db58e..06991c192 100644 --- a/tests/unit/test_sdk_bridge.py +++ b/tests/unit/test_sdk_bridge.py @@ -48,6 +48,7 @@ class FakeCoreEvent: self._extras: dict[str, Any] = {} self._result = None self._stopped = False + self._has_send_oper = False def get_platform_name(self): return "webchat" @@ -629,6 +630,10 @@ class TestPlugin(Plugin): # ctx.messages.send 经由 Context.send_message 主动发送 context.send_message.assert_awaited_once() + # 镜像进程内 event.send()：在飞事件被标记为已发送，管线跳过 LLM 阶段 + assert event._has_send_oper is True + # 处理结束后在飞事件表清空 + assert bridge._inflight_events == {} session, chain = context.send_message.await_args.args assert session.platform_id == "webchat" assert session.message_type is CoreMessageType.FRIEND_MESSAGE