diff --git a/src-new/astrbot_sdk/_legacy_api.py b/src-new/astrbot_sdk/_legacy_api.py index 292f71af6..291a3a91d 100644 --- a/src-new/astrbot_sdk/_legacy_api.py +++ b/src-new/astrbot_sdk/_legacy_api.py @@ -117,7 +117,15 @@ class LegacyContext: async def send_message(self, session: str, message_chain: Any) -> None: _warn_once("context.send_message()", "ctx.platform.send(session, text)") ctx = self.require_runtime_context() - await ctx.platform.send(session, str(message_chain)) + # 正确序列化 MessageChain 对象 + # 优先使用 get_plain_text() 方法（旧版 MessageChain） + if hasattr(message_chain, "get_plain_text") and callable(message_chain.get_plain_text): + text = message_chain.get_plain_text() + elif hasattr(message_chain, "to_text") and callable(message_chain.to_text): + text = message_chain.to_text() + else: + text = str(message_chain) + await ctx.platform.send(session, text) async def add_llm_tools(self, *tools: Any) -> None: _warn_once("context.add_llm_tools()", "ctx.llm.chat_raw(..., tools=...)") diff --git a/src-new/astrbot_sdk/runtime/bootstrap.py b/src-new/astrbot_sdk/runtime/bootstrap.py index 7745d3cc6..921515c56 100644 --- a/src-new/astrbot_sdk/runtime/bootstrap.py +++ b/src-new/astrbot_sdk/runtime/bootstrap.py @@ -5,6 +5,7 @@ import inspect import os import signal import sys +from collections.abc import Callable from pathlib import Path from typing import IO, Any @@ -71,11 +72,13 @@ class WorkerSession: repo_root: Path, env_manager: PluginEnvironmentManager, capability_router: CapabilityRouter, + on_closed: Callable[[], None] | None = None, ) -> None: self.plugin = plugin self.repo_root = repo_root.resolve() self.env_manager = env_manager self.capability_router = capability_router + self.on_closed = on_closed self.peer: Peer | None = None self.handlers = [] @@ -110,12 +113,42 @@ class WorkerSession: self.peer.set_invoke_handler(self._handle_capability_invoke) try: await self.peer.start() - await self.peer.wait_until_remote_initialized() + # 同时监听初始化完成和连接关闭，避免 worker 崩溃时等满超时 + init_task = asyncio.create_task(self.peer.wait_until_remote_initialized(timeout=None)) + closed_task = asyncio.create_task(self.peer.wait_closed()) + done, pending = await asyncio.wait( + {init_task, closed_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + for task in pending: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + if closed_task in done: + raise RuntimeError(f"插件 {self.plugin.name} worker 进程在初始化阶段退出") + self.handlers = list(self.peer.remote_handlers) + + # 启动后台任务监听连接关闭 + if self.on_closed is not None: + asyncio.create_task(self._watch_connection()) except Exception: await self.stop() raise + async def _watch_connection(self) -> None: + """监听 Worker 连接关闭，触发清理回调""" + if self.peer is not None: + await self.peer.wait_closed() + if self.on_closed is not None: + try: + self.on_closed() + except Exception: + logger.exception("on_closed callback failed for plugin {}", self.plugin.name) + async def stop(self) -> None: if self.peer is not None: await self.peer.stop() @@ -220,6 +253,7 @@ class SupervisorRuntime: repo_root=self.repo_root, env_manager=self.env_manager, capability_router=self.capability_router, + on_closed=lambda name=plugin.name: self._handle_worker_closed(name), ) try: await session.start() @@ -248,6 +282,19 @@ class SupervisorRuntime: await self.stop() raise + def _handle_worker_closed(self, plugin_name: str) -> None: + """Worker 连接关闭时的清理回调""" + session = self.worker_sessions.pop(plugin_name, None) + if session is None: + return + # 从 handler_to_worker 中移除该 worker 的所有 handlers + for handler in session.handlers: + self.handler_to_worker.pop(handler.id, None) + # 从 loaded_plugins 中移除 + if plugin_name in self.loaded_plugins: + self.loaded_plugins.remove(plugin_name) + logger.warning("插件 {} worker 连接已关闭，已清理相关 handlers", plugin_name) + async def stop(self) -> None: for session in list(self.worker_sessions.values()): await session.stop() @@ -312,7 +359,12 @@ class PluginWorkerRuntime: [item.descriptor for item in self.loaded_plugin.handlers], metadata={"plugin_id": self.plugin.name}, ) - await self._run_lifecycle("on_start") + try: + await self._run_lifecycle("on_start") + except Exception: + # on_start 失败时，通知 Supervisor 并退出 + await self.peer.stop() + raise async def stop(self) -> None: try: diff --git a/src-new/astrbot_sdk/runtime/handler_dispatcher.py b/src-new/astrbot_sdk/runtime/handler_dispatcher.py index 358da24f6..3c4df8238 100644 --- a/src-new/astrbot_sdk/runtime/handler_dispatcher.py +++ b/src-new/astrbot_sdk/runtime/handler_dispatcher.py @@ -29,7 +29,10 @@ class HandlerDispatcher: if loaded.legacy_context is not None: loaded.legacy_context.bind_runtime_context(ctx) - task = asyncio.create_task(self._run_handler(loaded, event, ctx)) + # 提取 legacy args 用于兼容旧版 handler 签名 + legacy_args = message.input.get("args") or {} + + task = asyncio.create_task(self._run_handler(loaded, event, ctx, legacy_args)) self._active[message.id] = (task, cancel_token) try: await task @@ -50,9 +53,10 @@ class HandlerDispatcher: loaded: LoadedHandler, event: MessageEvent, ctx: Context, + legacy_args: dict[str, Any] | None = None, ) -> None: try: - result = loaded.callable(*self._build_args(loaded.callable, event, ctx)) + result = loaded.callable(*self._build_args(loaded.callable, event, ctx, legacy_args)) if inspect.isasyncgen(result): async for item in result: await self._consume_legacy_result(item, event) @@ -65,9 +69,16 @@ class HandlerDispatcher: await self._handle_error(loaded.owner, exc, event, ctx) raise - def _build_args(self, handler, event: MessageEvent, ctx: Context) -> list[Any]: + def _build_args( + self, + handler, + event: MessageEvent, + ctx: Context, + legacy_args: dict[str, Any] | None = None, + ) -> list[Any]: signature = inspect.signature(handler) args: list[Any] = [] + legacy_args = legacy_args or {} for parameter in signature.parameters.values(): if parameter.kind not in ( inspect.Parameter.POSITIONAL_ONLY, @@ -78,6 +89,9 @@ class HandlerDispatcher: args.append(event) elif parameter.name in {"ctx", "context"}: args.append(ctx) + elif parameter.name in legacy_args: + # 支持从 legacy args 中注入参数（如命令参数、regex 捕获组等） + args.append(legacy_args[parameter.name]) return args async def _consume_legacy_result(self, item: Any, event: MessageEvent) -> None: diff --git a/src-new/astrbot_sdk/runtime/peer.py b/src-new/astrbot_sdk/runtime/peer.py index 531f348cf..0f7603afe 100644 --- a/src-new/astrbot_sdk/runtime/peer.py +++ b/src-new/astrbot_sdk/runtime/peer.py @@ -67,6 +67,22 @@ class Peer: async def stop(self) -> None: self._closed = True + # 终止所有挂起的 RPC，避免调用方永久挂起 + for future in list(self._pending_results.values()): + if not future.done(): + future.set_exception(AstrBotError.internal_error("连接已关闭")) + self._pending_results.clear() + + for queue in list(self._pending_streams.values()): + await queue.put(AstrBotError.internal_error("连接已关闭")) + self._pending_streams.clear() + + # 取消所有入站任务 + for task, token in list(self._inbound_tasks.values()): + token.cancel() + task.cancel() + self._inbound_tasks.clear() + await self.transport.stop() async def wait_closed(self) -> None: @@ -296,13 +312,15 @@ class Peer: if queue is not None: await queue.put(AstrBotError.protocol_error("stream=true 调用不应收到 result")) return - future.set_result(message) + # 检查 future 是否已完成（可能被调用方取消） + if not future.done(): + future.set_result(message) async def _handle_event(self, message: EventMessage) -> None: queue = self._pending_streams.get(message.id) if queue is None: future = self._pending_results.get(message.id) - if future is not None: + if future is not None and not future.done(): future.set_exception(AstrBotError.protocol_error("stream=false 调用不应收到 event")) return await queue.put(message)