feat: support remote websocket workers in sdk runtime

Author: whatevertogo

astrbot-sdk/src/astrbot_sdk/cli.py

@@ -1144,6 +1144,39 @@ def _build_plugin(plugin_dir: Path, output_dir: Path | None) -> None:
     click.echo(f"artifact: {archive_path}")
 
 
+def _run_websocket_worker_entrypoint(
+    *,
+    worker_id: str | None,
+    plugin_dirs: tuple[Path, ...],
+    host: str,
+    port: int,
+    path: str,
+    tls_ca_file: Path,
+    tls_cert_file: Path,
+    tls_key_file: Path,
+) -> None:
+    resolved_plugin_dirs = list(plugin_dirs) if plugin_dirs else [Path.cwd()]
+    _run_async_entrypoint(
+        run_websocket_server(
+            worker_id=worker_id,
+            plugin_dirs=resolved_plugin_dirs,
+            host=host,
+            port=port,
+            path=path,
+            tls_ca_file=tls_ca_file,
+            tls_cert_file=tls_cert_file,
+            tls_key_file=tls_key_file,
+        ),
+        log_message=f"启动 WebSocket Worker，端口：{port}",
+        context={
+            "worker_id": worker_id,
+            "plugin_dirs": resolved_plugin_dirs,
+            "port": port,
+            "path": path,
+        },
+    )
+
+
 @click.group()
 @click.option("-v", "--verbose", is_flag=True, help="Enable verbose output")
 @click.pass_context
@@ -1161,20 +1194,37 @@ def cli(ctx, verbose: bool) -> None:
     type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
     help="Directory containing plugin folders",
 )
+@click.option(
+    "--workers-manifest",
+    default=None,
+    type=click.Path(file_okay=True, dir_okay=False, path_type=Path),
+    help="Supervisor manifest describing remote websocket workers",
+)
 @click.option(
     "--protocol-stdout",
     default=None,
     type=str,
     help="Redirect runtime protocol stdout to console, silent, or a file path",
 )
-def run(plugins_dir: Path, protocol_stdout: str | None) -> None:
+def run(
+    plugins_dir: Path,
+    workers_manifest: Path | None,
+    protocol_stdout: str | None,
+) -> None:
     """Start the plugin supervisor over stdio."""
     transport_stdout, opened_stdout = _resolve_protocol_stdout(protocol_stdout)
     try:
         _run_async_entrypoint(
-            run_supervisor(plugins_dir=plugins_dir, stdout=transport_stdout),
+            run_supervisor(
+                plugins_dir=plugins_dir,
+                stdout=transport_stdout,
+                workers_manifest=workers_manifest,
+            ),
             log_message=f"启动插件主管进程，插件目录：{plugins_dir}",
-            context={"plugins_dir": plugins_dir},
+            context={
+                "plugins_dir": plugins_dir,
+                "workers_manifest": workers_manifest,
+            },
         )
     finally:
         if opened_stdout is not None:
@@ -1362,12 +1412,101 @@ def worker(
             opened_stdout.close()
 
 
+@cli.command("serve-worker")
+@click.option("--worker-id", default=None, type=str, help="Stable websocket worker id")
+@click.option(
+    "--plugin-dir",
+    "plugin_dirs",
+    multiple=True,
+    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
+    help="Plugin directory to serve; repeat to host multiple plugins in one worker",
+)
+@click.option("--host", default="127.0.0.1", show_default=True)
+@click.option("--port", default=8765, type=int, show_default=True)
+@click.option("--path", default="/", show_default=True)
+@click.option(
+    "--tls-ca-file",
+    required=True,
+    type=click.Path(file_okay=True, dir_okay=False, exists=True, path_type=Path),
+)
+@click.option(
+    "--tls-cert-file",
+    required=True,
+    type=click.Path(file_okay=True, dir_okay=False, exists=True, path_type=Path),
+)
+@click.option(
+    "--tls-key-file",
+    required=True,
+    type=click.Path(file_okay=True, dir_okay=False, exists=True, path_type=Path),
+)
+def serve_worker(
+    worker_id: str | None,
+    plugin_dirs: tuple[Path, ...],
+    host: str,
+    port: int,
+    path: str,
+    tls_ca_file: Path,
+    tls_cert_file: Path,
+    tls_key_file: Path,
+) -> None:
+    """Serve one or more plugins as a standalone websocket worker."""
+    _run_websocket_worker_entrypoint(
+        worker_id=worker_id,
+        plugin_dirs=plugin_dirs,
+        host=host,
+        port=port,
+        path=path,
+        tls_ca_file=tls_ca_file,
+        tls_cert_file=tls_cert_file,
+        tls_key_file=tls_key_file,
+    )
+
+
 @cli.command(hidden=True)
-@click.option("--port", default=8765, type=int, help="WebSocket server port")
-def websocket(port: int) -> None:
-    """WebSocket runtime entrypoint kept for standalone bridge scenarios."""
-    _run_async_entrypoint(
-        run_websocket_server(port=port),
-        log_message=f"启动 WebSocket 服务器，端口：{port}",
-        context={"port": port},
+@click.option("--worker-id", default=None, type=str)
+@click.option(
+    "--plugin-dir",
+    "plugin_dirs",
+    multiple=True,
+    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
+)
+@click.option("--host", default="127.0.0.1", show_default=True)
+@click.option("--port", default=8765, type=int, show_default=True)
+@click.option("--path", default="/", show_default=True)
+@click.option(
+    "--tls-ca-file",
+    required=True,
+    type=click.Path(file_okay=True, dir_okay=False, exists=True, path_type=Path),
+)
+@click.option(
+    "--tls-cert-file",
+    required=True,
+    type=click.Path(file_okay=True, dir_okay=False, exists=True, path_type=Path),
+)
+@click.option(
+    "--tls-key-file",
+    required=True,
+    type=click.Path(file_okay=True, dir_okay=False, exists=True, path_type=Path),
+)
+def websocket(
+    worker_id: str | None,
+    plugin_dirs: tuple[Path, ...],
+    host: str,
+    port: int,
+    path: str,
+    tls_ca_file: Path,
+    tls_cert_file: Path,
+    tls_key_file: Path,
+) -> None:
+    """Deprecated websocket runtime wrapper for standalone worker scenarios."""
+    logger.warning("'astr websocket' is deprecated; use 'astr serve-worker' instead")
+    _run_websocket_worker_entrypoint(
+        worker_id=worker_id,
+        plugin_dirs=plugin_dirs,
+        host=host,
+        port=port,
+        path=path,
+        tls_ca_file=tls_ca_file,
+        tls_cert_file=tls_cert_file,
+        tls_key_file=tls_key_file,
     )

astrbot-sdk/src/astrbot_sdk/llm/agents.py

@@ -28,9 +28,9 @@ def from_payload(cls, payload: dict[str, Any]) -> AgentSpec:
 
 
 class BaseAgentRunner(ABC):
-    """P0.5 agent registration surface.
+    """ agent registration surface.
 
-    P0.5 only supports agent registration metadata. Actual execution remains
+     only supports agent registration metadata. Actual execution remains
     owned by the core tool loop and is not directly callable from SDK plugins.
     """
 

astrbot-sdk/src/astrbot_sdk/runtime/bootstrap.py

@@ -28,8 +28,12 @@
     _sdk_source_dir,
     _wait_for_shutdown,
 )
-from .transport import StdioTransport, WebSocketServerTransport
-from .worker import GroupWorkerRuntime, PluginWorkerRuntime
+from .transport import (
+    StdioTransport,
+    WebSocketServerTransport,
+    build_websocket_server_ssl_context,
+)
+from .worker import GroupWorkerRuntime, PluginWorkerRuntime, _load_plugin_specs
 
 __all__ = [
     "GroupWorkerRuntime",
@@ -52,6 +56,7 @@ async def run_supervisor(
     stdin: IO[str] | None = None,
     stdout: IO[str] | None = None,
     env_manager: PluginEnvironmentManager | None = None,
+    workers_manifest: Path | None = None,
 ) -> None:
     transport_stdin, transport_stdout, original_stdout = _prepare_stdio_transport(
         stdin,
@@ -62,6 +67,7 @@ async def run_supervisor(
         transport=transport,
         plugins_dir=plugins_dir,
         env_manager=env_manager,
+        workers_manifest=workers_manifest,
     )
 
     try:
@@ -115,15 +121,47 @@ async def run_plugin_worker(
 
 async def run_websocket_server(
     *,
+    worker_id: str | None = None,
     host: str = "127.0.0.1",
     port: int = 8765,
     path: str = "/",
-    plugin_dir: Path | None = None,
+    plugin_dirs: list[Path] | None = None,
+    tls_ca_file: Path | None = None,
+    tls_cert_file: Path | None = None,
+    tls_key_file: Path | None = None,
 ) -> None:
-    runtime = PluginWorkerRuntime(
-        plugin_dir=plugin_dir or Path.cwd(),
-        transport=WebSocketServerTransport(host=host, port=port, path=path),
+    resolved_plugin_dirs = [path.resolve() for path in (plugin_dirs or [Path.cwd()])]
+    if tls_ca_file is None or tls_cert_file is None or tls_key_file is None:
+        raise ValueError(
+            "tls_ca_file, tls_cert_file, and tls_key_file are required for websocket workers"
+        )
+    transport = WebSocketServerTransport(
+        host=host,
+        port=port,
+        path=path,
+        ssl_context=build_websocket_server_ssl_context(
+            ca_file=tls_ca_file,
+            cert_file=tls_cert_file,
+            key_file=tls_key_file,
+        ),
     )
+    resolved_worker_id = worker_id
+    if resolved_worker_id is None and len(resolved_plugin_dirs) == 1:
+        resolved_worker_id = _load_plugin_specs([resolved_plugin_dirs[0]])[0].name
+    if len(resolved_plugin_dirs) == 1:
+        runtime = PluginWorkerRuntime(
+            plugin_dir=resolved_plugin_dirs[0],
+            worker_id=resolved_worker_id,
+            transport=transport,
+        )
+    else:
+        if resolved_worker_id is None:
+            raise ValueError("worker_id is required when serving multiple plugins")
+        runtime = GroupWorkerRuntime(
+            plugin_dirs=resolved_plugin_dirs,
+            worker_id=resolved_worker_id,
+            transport=transport,
+        )
     try:
         await runtime.start()
         stop_event = asyncio.Event()

astrbot-sdk/src/astrbot_sdk/runtime/capability_router.py

@@ -346,17 +346,26 @@ def upsert_plugin(
         normalized_metadata.setdefault("support_platforms", [])
         normalized_metadata.setdefault("astrbot_version", None)
         local_mcp_servers = normalized_metadata.pop("local_mcp_servers", {})
-        self._plugins[name] = _RegisteredPlugin(
-            metadata=normalized_metadata,
-            config=dict(config or {}),
-            handlers=[],
-            local_mcp_servers={
+        normalized_servers = (
+            {
                 str(server_name): dict(server_payload)
                 for server_name, server_payload in local_mcp_servers.items()
                 if str(server_name).strip() and isinstance(server_payload, dict)
             }
             if isinstance(local_mcp_servers, dict)
-            else {},
+            else {}
+        )
+        existing = self._plugins.get(name)
+        if existing is not None:
+            existing.metadata = normalized_metadata
+            existing.config = dict(config or {})
+            existing.local_mcp_servers = normalized_servers
+            return
+        self._plugins[name] = _RegisteredPlugin(
+            metadata=normalized_metadata,
+            config=dict(config or {}),
+            handlers=[],
+            local_mcp_servers=normalized_servers,
         )
 
     def set_plugin_handlers(

astrbot-sdk/src/astrbot_sdk/runtime/peer.py

@@ -133,6 +133,12 @@ def _select_negotiated_protocol_version(
     remote_metadata: dict[str, Any],
     local_supported_versions: Sequence[str],
 ) -> str | None:
+    """从双方支持的版本中选出最佳兼容版本。
+
+    协商策略：优先精确匹配，否则在同主版本号范围内选双方都支持的最高版本。
+    排除比请求版本更高的候选，因为远端能提供高于我们请求的版本说明我们本地
+    尚未实现该版本协议，无法正确处理对应的协议消息。
+    """
     if requested_version in local_supported_versions:
         return requested_version
     requested_key = _parse_protocol_version(requested_version)
@@ -241,7 +247,15 @@ async def start(self) -> None:
         self._transport_watch_task = asyncio.create_task(self._watch_transport_closed())
 
     async def stop(self) -> None:
-        """关闭 `Peer` 并清理所有挂起中的请求、流和入站任务。"""
+        """关闭 `Peer` 并清理所有挂起中的请求、流和入站任务。
+
+        重入安全性：transport.stop() 关闭底层连接时会触发原始消息处理器的
+        异常路径，该路径调用 _fail_connection() -> _schedule_stop() -> stop()，
+        形成间接递归。_stopping 标志和 _stop_task 引用共同防止重复清理资源。
+        使用 asyncio.shield 等待是因为：如果当前任务在等待另一个 stop() 完成
+        期间被取消，shield 保护内部 stop_task 不被连带取消，避免 Peer 停留在
+        半关闭状态。
+        """
         if self._closed.is_set():
             return
         current_task = asyncio.current_task()
@@ -275,7 +289,9 @@ async def stop(self) -> None:
             await self.transport.stop()
             self._closed.set()
         finally:
-            # 只在当前 task 就是 stop_task 时才清除引用，避免误清其他 task 的记录
+            # 只在当前 task 就是 stop_task 时才清除引用，避免误清其他 task 的记录。
+            # 场景：A 任务正在 stop() 中，B 任务也进入了 stop() 并等待 A 完成，
+            # 如果 B 在 finally 中清除了 _stop_task，A 还未执行完就会失去引用。
             if self._stop_task is current_task:
                 self._stop_task = None
 
@@ -564,8 +580,12 @@ def _on_invoke_done(
                     exc = _task.exception()
                     if exc is None:
                         return
-                    # 后台 invoke 理论上应把错误编码成协议消息；若异常仍逃逸，通常说明
-                    # 回复发送失败或连接状态异常，必须立刻标记连接失效，避免对端永久等待。
+                    # 为什么整个连接都要失败？正常情况下 invoke handler 会把错误编码成
+                    # ResultMessage 发回给对端。如果异常仍然逃逸，说明要么回复发送失败
+                    # （transport 已断），要么 handler 实现有 bug。无论哪种情况，连接的
+                    # 消息交换契约已不可靠，继续使用可能导致对端无限等待或数据丢失。
+                    # 采用"单点故障 → 全连接失败"策略而非隔离单个 handler，是因为协议层
+                    # 无法保证后续消息的正确性。
                     logger.error(
                         "Peer inbound invoke task crashed unexpectedly: "
                         "request_id={} error={!r}",