feat: 增强 LegacyContext 和 WorkerSession 的消息处理，支持更灵活的参数传递和连接关闭管理

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=...)")

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:

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:

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)