From 2a5e55641dedc85db5bf91ba573ac22120142780 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 27 Mar 2026 18:07:09 +0800 Subject: [PATCH] feat(abp): initial ABP protocol implementation - Add ABP (AstrBot Plugin) protocol Python package (astrbot/core/plugin/) - PluginManager for plugin lifecycle management - PluginClient for out-of-process plugin communication - Transport layer (Stdio, Unix Socket, HTTP) - Data models and constants - Add ABP error codes to Rust error.rs (-32700 to -32211) - Fix Rust server.rs compilation issues - Fix UNIX_EPOODY typo - Fix mutable borrow issues - Fix API response type mismatches - Update _core.pyi type stubs with ABP types - Add openspec change archive for ABP protocol implementation --- astrbot/core/plugin/__init__.py | 49 +++ astrbot/core/plugin/client.py | 319 ++++++++++++++++ astrbot/core/plugin/const.py | 59 +++ astrbot/core/plugin/manager.py | 344 ++++++++++++++++++ astrbot/core/plugin/models.py | 144 ++++++++ astrbot/core/plugin/transport.py | 231 ++++++++++++ astrbot/rust/_core.pyi | 80 ++++ .../.openspec.yaml | 2 + .../design.md | 73 ++++ .../proposal.md | 35 ++ .../specs/abp-event-system/spec.md | 46 +++ .../specs/abp-plugin-loader/spec.md | 41 +++ .../specs/abp-protocol/spec.md | 51 +++ .../specs/abp-tool-router/spec.md | 42 +++ .../specs/abp-transport/spec.md | 35 ++ .../tasks.md | 70 ++++ openspec/rust-ffi.md | 156 ++++++++ rust/src/error.rs | 116 ++++++ rust/src/server.rs | 132 ++++--- 19 files changed, 1966 insertions(+), 59 deletions(-) create mode 100644 astrbot/core/plugin/__init__.py create mode 100644 astrbot/core/plugin/client.py create mode 100644 astrbot/core/plugin/const.py create mode 100644 astrbot/core/plugin/manager.py create mode 100644 astrbot/core/plugin/models.py create mode 100644 astrbot/core/plugin/transport.py create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/.openspec.yaml create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/design.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/proposal.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-event-system/spec.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-plugin-loader/spec.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-protocol/spec.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-tool-router/spec.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-transport/spec.md create mode 100644 openspec/changes/archive/2026-03-27-abp-protocol-implementation/tasks.md create mode 100644 openspec/rust-ffi.md diff --git a/astrbot/core/plugin/__init__.py b/astrbot/core/plugin/__init__.py new file mode 100644 index 000000000..b6686d02a --- /dev/null +++ b/astrbot/core/plugin/__init__.py @@ -0,0 +1,49 @@ +"""AstrBot Plugin System (ABP) + +ABP is the plugin communication protocol for AstrBot, supporting: +- In-process and out-of-process loading modes +- Full plugin lifecycle management +- Tool calling, message handling, event subscriptions +- JSON-RPC 2.0 communication +""" + +from .client import PluginClient +from .const import ABP_VERSION, get_error_message +from .manager import PluginManager +from .models import ( + EventNotification, + HandleEventResult, + MessageChainItem, + MessageEvent, + PluginCapabilities, + PluginConfig, + PluginInfo, + PluginMetadata, + SenderInfo, + Tool, + ToolContent, + ToolResult, +) +from .transport import HttpTransport, StdioTransport, UnixSocketTransport + +__all__ = [ + "ABP_VERSION", + "EventNotification", + "HandleEventResult", + "HttpTransport", + "MessageChainItem", + "MessageEvent", + "PluginCapabilities", + "PluginClient", + "PluginConfig", + "PluginInfo", + "PluginManager", + "PluginMetadata", + "SenderInfo", + "StdioTransport", + "Tool", + "ToolContent", + "ToolResult", + "UnixSocketTransport", + "get_error_message", +] diff --git a/astrbot/core/plugin/client.py b/astrbot/core/plugin/client.py new file mode 100644 index 000000000..26f6ca1b3 --- /dev/null +++ b/astrbot/core/plugin/client.py @@ -0,0 +1,319 @@ +"""ABP Plugin Client for out-of-process communication""" + +import asyncio +import logging +from typing import Any + +from .const import ( + ABP_VERSION, +) +from .models import ( + HandleEventResult, + InitializeParams, + InitializeResult, + MessageChainItem, + PluginCapabilities, + PluginConfig, + PluginMetadata, + Tool, + ToolContent, + ToolResult, +) +from .transport import HttpTransport, StdioTransport, Transport, UnixSocketTransport + +logger = logging.getLogger("astrbot.plugin.client") + + +class PluginClient: + """Client for communicating with out-of-process plugins. + + This client implements the ABP protocol for plugin communication, + supporting JSON-RPC 2.0 over various transports. + """ + + def __init__(self, config: PluginConfig) -> None: + self._config = config + self._transport: Transport | None = None + self._initialized = False + self._capabilities = PluginCapabilities() + self._metadata: PluginMetadata | None = None + self._tools: list[Tool] = [] + + @property + def name(self) -> str: + return self._config.name + + @property + def is_initialized(self) -> bool: + return self._initialized + + @property + def capabilities(self) -> PluginCapabilities: + return self._capabilities + + @property + def metadata(self) -> PluginMetadata | None: + return self._metadata + + @property + def tools(self) -> list[Tool]: + return self._tools + + async def start( + self, process: asyncio.subprocess.Process | None = None + ) -> InitializeResult: + """Start and initialize the plugin. + + Args: + process: For stdio transport, the subprocess handle + + Returns: + InitializeResult from the plugin + """ + # Create transport based on config + if self._config.transport == "stdio": + if not process: + raise ValueError("stdio transport requires a subprocess") + self._transport = StdioTransport(process) + elif self._config.transport == "unix_socket": + if not self._config.url: + raise ValueError("unix_socket transport requires socket path in url") + self._transport = UnixSocketTransport(self._config.url) + await self._transport.connect() + elif self._config.transport == "http": + if not self._config.url: + raise ValueError("http transport requires URL") + self._transport = HttpTransport(self._config.url) + await self._transport.connect() + else: + raise ValueError(f"Unknown transport: {self._config.transport}") + + # Send initialize request + result = await self._send_initialize() + + # Parse result + self._capabilities = PluginCapabilities( + tools=result.capabilities.tools, + handlers=result.capabilities.handlers, + events=result.capabilities.events, + resources=result.capabilities.resources, + ) + self._metadata = result.metadata + self._initialized = True + + return result + + async def _send_initialize(self) -> InitializeResult: + """Send initialize request to plugin.""" + assert self._transport + + # Get data directories + from pathlib import Path + + from astrbot.core.utils.astrbot_path import get_astrbot_data_path + + data_path = Path(get_astrbot_data_path()) + plugin_data_path = data_path / "plugins" / self._config.name + + params = InitializeParams( + protocol_version=ABP_VERSION, + client_info={"name": "astrbot", "version": "4.25.0"}, + capabilities={ + "streaming": True, + "events": True, + }, + plugin_config={"user_config": {}}, + data_dirs={ + "root": str(data_path), + "plugin_data": str(plugin_data_path), + "temp": str(data_path / "temp"), + }, + ) + + result = await self._transport.send( + "initialize", + { + "protocolVersion": params.protocol_version, + "clientInfo": params.client_info, + "capabilities": params.capabilities, + "pluginConfig": params.plugin_config, + "dataDirs": params.data_dirs, + }, + ) + + return InitializeResult( + protocol_version=result.get("protocolVersion", ABP_VERSION), + server_info=result.get( + "serverInfo", {"name": self._config.name, "version": "1.0.0"} + ), + capabilities=PluginCapabilities( + tools=result.get("capabilities", {}).get("tools", False), + handlers=result.get("capabilities", {}).get("handlers", False), + events=result.get("capabilities", {}).get("events", False), + resources=result.get("capabilities", {}).get("resources", False), + ), + config_schema=result.get("configSchema"), + metadata=PluginMetadata( + display_name=result.get("metadata", {}).get("display_name"), + description=result.get("metadata", {}).get("description"), + author=result.get("metadata", {}).get("author"), + homepage=result.get("metadata", {}).get("homepage"), + support_platforms=result.get("metadata", {}).get( + "support_platforms", [] + ), + astrbot_version=result.get("metadata", {}).get("astrbot_version"), + ), + ) + + async def list_tools(self) -> list[Tool]: + """List available tools from plugin.""" + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + result = await self._transport.send("tools/list", {}) + + self._tools = [ + Tool( + name=t.get("name", ""), + description=t.get("description", ""), + input_schema=t.get("parameters", {}), + ) + for t in result.get("tools", []) + ] + + return self._tools + + async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> ToolResult: + """Call a tool on the plugin.""" + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + result = await self._transport.send( + "tools/call", + { + "name": tool_name, + "arguments": arguments, + }, + ) + + return ToolResult( + content=[ + ToolContent( + type=c.get("type", "text"), + text=c.get("text"), + image=c.get("image"), + url=c.get("url"), + ) + for c in result.get("content", []) + ] + ) + + async def handle_event( + self, event_type: str, event_data: dict[str, Any] + ) -> HandleEventResult: + """Handle an event with the plugin.""" + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + result = await self._transport.send( + "plugin.handle_event", + { + "event_type": event_type, + "event": event_data, + }, + ) + + return HandleEventResult( + handled=result.get("handled", False), + results=[ + MessageChainItem(type=r.get("type", "plain"), text=r.get("text")) + for r in result.get("results", []) + ], + stop_propagation=result.get("stop_propagation", False), + ) + + async def notify(self, event_type: str, data: dict[str, Any]) -> None: + """Send an event notification to the plugin.""" + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + await self._transport.notify( + "plugin.notify", + { + "event_type": event_type, + "data": data, + }, + ) + + async def stop(self) -> None: + """Stop the plugin.""" + if self._transport: + await self._transport.close() + self._transport = None + self._initialized = False + + async def reload(self) -> None: + """Reload the plugin (re-initialize without restarting process).""" + if self._transport: + await self._transport.send("plugin.reload", {}) + self._initialized = False + + async def update_config(self, user_config: dict[str, Any]) -> None: + """Update plugin configuration.""" + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + await self._transport.notify( + "plugin.config_update", + {"user_config": user_config}, + ) + + async def subscribe(self, event_type: str) -> None: + """Subscribe to an event type. + + Args: + event_type: Event type to subscribe to (e.g., "llm_request", "tool_called") + """ + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + await self._transport.send( + "plugin.subscribe", + {"event_type": event_type}, + ) + logger.debug(f"Plugin {self._config.name} subscribed to {event_type}") + + async def unsubscribe(self, event_type: str) -> None: + """Unsubscribe from an event type. + + Args: + event_type: Event type to unsubscribe from + """ + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + await self._transport.send( + "plugin.unsubscribe", + {"event_type": event_type}, + ) + logger.debug(f"Plugin {self._config.name} unsubscribed from {event_type}") + + async def send_notification(self, method: str, params: dict[str, Any]) -> None: + """Send a JSON-RPC notification (no response expected). + + Args: + method: The JSON-RPC method name + params: The method parameters + """ + if not self._initialized: + raise RuntimeError("Plugin not initialized") + assert self._transport + + await self._transport.notify(method, params) diff --git a/astrbot/core/plugin/const.py b/astrbot/core/plugin/const.py new file mode 100644 index 000000000..e844d986f --- /dev/null +++ b/astrbot/core/plugin/const.py @@ -0,0 +1,59 @@ +"""ABP Protocol Constants""" + +# Protocol version +ABP_VERSION = "1.0.0" + +# ABP Error codes (JSON-RPC 2.0 base + ABP extension) +# JSON-RPC standard errors +PARSE_ERROR: int = -32700 +INVALID_REQUEST: int = -32600 +METHOD_NOT_FOUND: int = -32601 +INVALID_PARAMS: int = -32602 +INTERNAL_ERROR: int = -32603 + +# ABP extension errors (-32200 to -32211) +PLUGIN_NOT_FOUND: int = -32200 +PLUGIN_NOT_READY: int = -32201 +PLUGIN_CRASHED: int = -32202 +TOOL_NOT_FOUND: int = -32203 +TOOL_CALL_FAILED: int = -32204 +HANDLER_NOT_FOUND: int = -32205 +HANDLER_ERROR: int = -32206 +EVENT_SUBSCRIBE_FAILED: int = -32207 +PERMISSION_DENIED: int = -32208 +CONFIG_ERROR: int = -32209 +DEPENDENCY_MISSING: int = -32210 +VERSION_MISMATCH: int = -32211 + +# Plugin load modes +LOAD_MODE_IN_PROCESS: str = "in_process" +LOAD_MODE_OUT_OF_PROCESS: str = "out_of_process" + +# Transport types +TRANSPORT_STDIO: str = "stdio" +TRANSPORT_UNIX_SOCKET: str = "unix_socket" +TRANSPORT_HTTP: str = "http" + + +def get_error_message(code: int) -> str: + """Get error message for error code.""" + messages = { + PARSE_ERROR: "Parse error - Invalid JSON", + INVALID_REQUEST: "Invalid request", + METHOD_NOT_FOUND: "Method not found", + INVALID_PARAMS: "Invalid params", + INTERNAL_ERROR: "Internal error", + PLUGIN_NOT_FOUND: "Plugin not found", + PLUGIN_NOT_READY: "Plugin not ready", + PLUGIN_CRASHED: "Plugin crashed", + TOOL_NOT_FOUND: "Tool not found", + TOOL_CALL_FAILED: "Tool call failed", + HANDLER_NOT_FOUND: "Handler not found", + HANDLER_ERROR: "Handler error", + EVENT_SUBSCRIBE_FAILED: "Event subscription failed", + PERMISSION_DENIED: "Permission denied", + CONFIG_ERROR: "Configuration error", + DEPENDENCY_MISSING: "Dependency missing", + VERSION_MISMATCH: "Version mismatch", + } + return messages.get(code, f"Unknown error code: {code}") diff --git a/astrbot/core/plugin/manager.py b/astrbot/core/plugin/manager.py new file mode 100644 index 000000000..051836bec --- /dev/null +++ b/astrbot/core/plugin/manager.py @@ -0,0 +1,344 @@ +"""ABP Plugin Manager + +Manages plugin lifecycle, loading, and communication. +""" + +import asyncio +import logging +from pathlib import Path +from typing import Any + +from .client import PluginClient +from .const import LOAD_MODE_IN_PROCESS, LOAD_MODE_OUT_OF_PROCESS +from .models import PluginCapabilities, PluginConfig, PluginInfo, Tool + +logger = logging.getLogger("astrbot.plugin.manager") + + +class PluginManager: + """Manages ABP plugins. + + Responsible for: + - Plugin discovery and registration + - Plugin lifecycle (load, start, stop, reload) + - Tool routing across plugins + - Event distribution + """ + + def __init__(self) -> None: + self._plugins: dict[str, PluginClient] = {} + self._in_process_handlers: dict[str, Any] = {} + self._event_subscribers: dict[str, list[str]] = {} + self._initialized = False + + @property + def is_initialized(self) -> bool: + return self._initialized + + async def initialize( + self, + plugin_configs: list[PluginConfig], + data_root: Path, + ) -> None: + """Initialize the plugin manager and load plugins. + + Args: + plugin_configs: List of plugin configurations + data_root: Root data directory for AstrBot + """ + logger.info(f"Initializing plugin manager with {len(plugin_configs)} plugins") + + # Load each plugin + for config in plugin_configs: + await self.load_plugin(config, data_root) + + self._initialized = True + logger.info(f"Plugin manager initialized with {len(self._plugins)} plugins") + + async def load_plugin(self, config: PluginConfig, data_root: Path) -> None: + """Load a plugin. + + Args: + config: Plugin configuration + data_root: Root data directory + """ + logger.info(f"Loading plugin: {config.name} (mode: {config.load_mode})") + + if config.load_mode == LOAD_MODE_IN_PROCESS: + await self._load_in_process_plugin(config, data_root) + elif config.load_mode == LOAD_MODE_OUT_OF_PROCESS: + await self._load_out_of_process_plugin(config, data_root) + else: + raise ValueError(f"Unknown load mode: {config.load_mode}") + + async def _load_in_process_plugin( + self, config: PluginConfig, data_root: Path + ) -> None: + """Load an in-process plugin (Python module). + + For in-process plugins, we would import the module directly. + This is a placeholder for when the Rust FFI exposes plugin loading. + """ + logger.debug(f"Loading in-process plugin: {config.name}") + # In-process plugins would be loaded via Python import + # For now, just register a placeholder + self._in_process_handlers[config.name] = { + "config": config, + "capabilities": PluginCapabilities(), + "metadata": None, + "tools": [], + } + + async def _load_out_of_process_plugin( + self, config: PluginConfig, data_root: Path + ) -> None: + """Load an out-of-process plugin (subprocess). + + Args: + config: Plugin configuration + data_root: Root data directory + """ + if not config.command: + raise ValueError(f"Out-of-process plugin {config.name} requires 'command'") + + # Spawn subprocess + cmd = config.command + args = [cmd, *config.args] + + logger.debug(f"Spawning plugin process: {' '.join(args)}") + + process = await asyncio.create_subprocess_exec( + *args, + env={**asyncio.get_running_loop().run_in_executor, **config.env} + if config.env + else None, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + + # Create client and start plugin + client = PluginClient(config) + await client.start(process=process) + + self._plugins[config.name] = client + logger.info(f"Loaded out-of-process plugin: {config.name}") + + async def unload_plugin(self, name: str) -> None: + """Unload a plugin. + + Args: + name: Plugin name + """ + if name in self._plugins: + await self._plugins[name].stop() + del self._plugins[name] + logger.info(f"Unloaded plugin: {name}") + + if name in self._in_process_handlers: + del self._in_process_handlers[name] + logger.info(f"Unloaded in-process plugin: {name}") + + async def start_plugin(self, name: str) -> None: + """Start a plugin (already loaded).""" + if name not in self._plugins: + raise KeyError(f"Plugin not found: {name}") + # Plugin is already started if it's in the registry + pass + + async def stop_plugin(self, name: str) -> None: + """Stop a plugin.""" + await self.unload_plugin(name) + + async def reload_plugin(self, name: str) -> None: + """Reload a plugin.""" + if name not in self._plugins: + raise KeyError(f"Plugin not found: {name}") + await self._plugins[name].reload() + logger.info(f"Reloaded plugin: {name}") + + async def update_plugin_config( + self, name: str, user_config: dict[str, Any] + ) -> None: + """Update plugin configuration. + + Args: + name: Plugin name + user_config: New user configuration + """ + if name in self._plugins: + await self._plugins[name].update_config(user_config) + elif name in self._in_process_handlers: + # In-process plugins handle config differently + pass + else: + raise KeyError(f"Plugin not found: {name}") + + def list_plugins(self) -> list[PluginInfo]: + """List all loaded plugins. + + Returns: + List of PluginInfo for each loaded plugin + """ + result: list[PluginInfo] = [] + + for name, client in self._plugins.items(): + result.append( + PluginInfo( + name=client.name, + version="1.0.0", + load_mode=client._config.load_mode, + capabilities=client.capabilities, + metadata=client.metadata, + tools_count=len(client.tools), + ) + ) + + for name, handler in self._in_process_handlers.items(): + result.append( + PluginInfo( + name=name, + version=handler["config"].version, + load_mode=LOAD_MODE_IN_PROCESS, + capabilities=handler["capabilities"], + metadata=handler["metadata"], + tools_count=len(handler["tools"]), + ) + ) + + return result + + def get_plugin(self, name: str) -> PluginInfo | None: + """Get info for a specific plugin. + + Args: + name: Plugin name + + Returns: + PluginInfo or None if not found + """ + for info in self.list_plugins(): + if info.name == name: + return info + return None + + def get_all_tools(self) -> list[tuple[str, Tool]]: + """Get all tools from all plugins. + + Returns: + List of (plugin_name, Tool) tuples + """ + tools: list[tuple[str, Tool]] = [] + + for name, client in self._plugins.items(): + for tool in client.tools: + tools.append((name, tool)) + + for name, handler in self._in_process_handlers.items(): + for tool in handler["tools"]: + tools.append((name, tool)) + + return tools + + async def call_tool( + self, plugin_name: str, tool_name: str, arguments: dict[str, Any] + ) -> Any: + """Call a tool on a specific plugin. + + Args: + plugin_name: Plugin name + tool_name: Tool name + arguments: Tool arguments + + Returns: + Tool result + """ + if plugin_name in self._plugins: + result = await self._plugins[plugin_name].call_tool(tool_name, arguments) + return result + elif plugin_name in self._in_process_handlers: + # In-process tool calling would go through Python directly + raise NotImplementedError("In-process tool calling not implemented") + else: + raise KeyError(f"Plugin not found: {plugin_name}") + + async def route_tool_call( + self, tool_name: str, arguments: dict[str, Any] + ) -> tuple[str, Any]: + """Route a tool call to the appropriate plugin. + + Args: + tool_name: Tool name + arguments: Tool arguments + + Returns: + Tuple of (plugin_name, tool_result) + """ + for name, client in self._plugins.items(): + if any(t.name == tool_name for t in client.tools): + result = await client.call_tool(tool_name, arguments) + return (name, result) + + for name, handler in self._in_process_handlers.items(): + if any(t.name == tool_name for t in handler["tools"]): + # In-process tool calling + raise NotImplementedError("In-process tool calling not implemented") + + raise KeyError(f"Tool not found: {tool_name}") + + async def subscribe_event(self, plugin_name: str, event_type: str) -> None: + """Subscribe a plugin to an event type. + + Args: + plugin_name: Plugin name + event_type: Event type to subscribe to + """ + if plugin_name not in self._plugins: + raise KeyError(f"Plugin not found: {plugin_name}") + + if event_type not in self._event_subscribers: + self._event_subscribers[event_type] = [] + if plugin_name not in self._event_subscribers[event_type]: + self._event_subscribers[event_type].append(plugin_name) + await self._plugins[plugin_name]._transport.send( + "plugin.subscribe", {"event_type": event_type} + ) + logger.info(f"Plugin {plugin_name} subscribed to {event_type}") + + async def unsubscribe_event(self, plugin_name: str, event_type: str) -> None: + """Unsubscribe a plugin from an event type. + + Args: + plugin_name: Plugin name + event_type: Event type to unsubscribe from + """ + if plugin_name not in self._plugins: + raise KeyError(f"Plugin not found: {plugin_name}") + + if event_type in self._event_subscribers: + if plugin_name in self._event_subscribers[event_type]: + self._event_subscribers[event_type].remove(plugin_name) + await self._plugins[plugin_name]._transport.send( + "plugin.unsubscribe", {"event_type": event_type} + ) + logger.info(f"Plugin {plugin_name} unsubscribed from {event_type}") + + async def broadcast_event(self, event_type: str, data: dict[str, Any]) -> None: + """Broadcast an event to all subscribed plugins. + + Args: + event_type: Event type + data: Event data + """ + if event_type in self._event_subscribers: + for plugin_name in self._event_subscribers[event_type]: + await self._plugins[plugin_name].notify(event_type, data) + + async def shutdown(self) -> None: + """Shutdown the plugin manager and all plugins.""" + logger.info("Shutting down plugin manager") + for name in list(self._plugins.keys()): + await self.unload_plugin(name) + self._in_process_handlers.clear() + self._event_subscribers.clear() + self._initialized = False diff --git a/astrbot/core/plugin/models.py b/astrbot/core/plugin/models.py new file mode 100644 index 000000000..7a18e3bc0 --- /dev/null +++ b/astrbot/core/plugin/models.py @@ -0,0 +1,144 @@ +"""ABP Protocol Data Models""" + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class PluginCapabilities: + """Plugin capabilities.""" + + tools: bool = False + handlers: bool = False + events: bool = False + resources: bool = False + + +@dataclass +class PluginMetadata: + """Plugin metadata.""" + + display_name: str | None = None + description: str | None = None + author: str | None = None + homepage: str | None = None + support_platforms: list[str] = field(default_factory=list) + astrbot_version: str | None = None + + +@dataclass +class PluginConfig: + """Plugin configuration.""" + + name: str + version: str = "1.0.0" + load_mode: str = "in_process" + command: str | None = None + args: list[str] = field(default_factory=list) + env: dict[str, str] = field(default_factory=dict) + transport: str = "stdio" + url: str | None = None + + +@dataclass +class InitializeParams: + """Initialize request parameters.""" + + protocol_version: str + client_info: dict[str, str] + capabilities: dict[str, bool] + plugin_config: dict[str, Any] + data_dirs: dict[str, str] + + +@dataclass +class InitializeResult: + """Initialize response from plugin.""" + + protocol_version: str + server_info: dict[str, str] + capabilities: PluginCapabilities + config_schema: dict[str, Any] | None = None + metadata: PluginMetadata | None = None + + +@dataclass +class Tool: + """Tool definition.""" + + name: str + description: str = "" + input_schema: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class ToolContent: + """Tool result content.""" + + type: str + text: str | None = None + image: str | None = None + url: str | None = None + + +@dataclass +class ToolResult: + """Tool call result.""" + + content: list[ToolContent] = field(default_factory=list) + + +@dataclass +class SenderInfo: + """Message sender info.""" + + user_id: str + nickname: str = "" + + +@dataclass +class MessageChainItem: + """Message chain item.""" + + type: str + text: str | None = None + + +@dataclass +class MessageEvent: + """Message event.""" + + message_id: str + unified_msg_origin: str + message_str: str + sender: SenderInfo + message_chain: list[MessageChainItem] = field(default_factory=list) + + +@dataclass +class HandleEventResult: + """Handle event result.""" + + handled: bool + results: list[MessageChainItem] = field(default_factory=list) + stop_propagation: bool = False + + +@dataclass +class EventNotification: + """Event notification.""" + + event_type: str + data: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class PluginInfo: + """Plugin info for listing.""" + + name: str + version: str + load_mode: str + capabilities: PluginCapabilities + metadata: PluginMetadata | None = None + tools_count: int = 0 diff --git a/astrbot/core/plugin/transport.py b/astrbot/core/plugin/transport.py new file mode 100644 index 000000000..280087995 --- /dev/null +++ b/astrbot/core/plugin/transport.py @@ -0,0 +1,231 @@ +"""ABP Transport Implementations""" + +import asyncio +import json +import logging +from abc import ABC, abstractmethod +from typing import Any + +import anyio + +logger = logging.getLogger("astrbot.plugin.transport") + + +class Transport(ABC): + """Base transport class.""" + + @abstractmethod + async def send(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + """Send JSON-RPC request and receive response.""" + pass + + @abstractmethod + async def notify(self, method: str, params: dict[str, Any]) -> None: + """Send JSON-RPC notification (no response).""" + pass + + @abstractmethod + async def close(self) -> None: + """Close the transport.""" + pass + + +class StdioTransport(Transport): + """Stdio transport for subprocess communication. + + Messages are sent as single-line JSON. + """ + + def __init__(self, process: asyncio.subprocess.Process) -> None: + self._process = process + self._lock = asyncio.Lock() + + async def send(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + """Send request via stdin and read from stdout.""" + request = { + "jsonrpc": "2.0", + "id": str(id(params)), + "method": method, + "params": params, + } + line = json.dumps(request, ensure_ascii=False) + async with self._lock: + self._process.stdin.write((line + "\n").encode("utf-8")) + await self._process.stdin.drain() + response_line = await self._process.stdout.readline() + if not response_line: + raise RuntimeError("Plugin process closed unexpectedly") + response = json.loads(response_line.decode("utf-8")) + if "error" in response: + raise RuntimeError(f"JSON-RPC error: {response['error']}") + return response.get("result", {}) + + async def notify(self, method: str, params: dict[str, Any]) -> None: + """Send notification via stdin.""" + notification = { + "jsonrpc": "2.0", + "method": method, + "params": params, + } + line = json.dumps(notification, ensure_ascii=False) + async with self._lock: + self._process.stdin.write((line + "\n").encode("utf-8")) + await self._process.stdin.drain() + + async def close(self) -> None: + """Terminate the subprocess.""" + self._process.terminate() + try: + await asyncio.wait_for(self._process.wait(), timeout=5.0) + except asyncio.TimeoutError: + self._process.kill() + + +class UnixSocketTransport(Transport): + """Unix Socket transport. + + Uses Content-Length framing protocol. + """ + + def __init__(self, socket_path: str) -> None: + self._socket_path = socket_path + self._reader: anyio.abc.SocketStream | None = None + self._writer: anyio.abc.SocketStream | None = None + + async def connect(self) -> None: + """Connect to the Unix socket.""" + try: + self._reader, self._writer = await anyio.connect_unix(self._socket_path) + except Exception as e: + raise RuntimeError(f"Failed to connect to {self._socket_path}: {e}") + + async def send(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + """Send request and receive response.""" + if not self._writer: + await self.connect() + assert self._writer + assert self._reader + + request = { + "jsonrpc": "2.0", + "id": str(id(params)), + "method": method, + "params": params, + } + content = json.dumps(request, ensure_ascii=False).encode("utf-8") + header = f"Content-Length: {len(content)}\r\n\r\n".encode() + + await self._writer.send_all(header + content) + + # Read response header + header_data = b"" + while b"\r\n\r\n" not in header_data: + chunk = await self._reader.receive_some(1024) + if not chunk: + raise RuntimeError("Connection closed") + header_data += chunk + + header_str = header_data.decode("utf-8") + content_length = 0 + for line in header_str.split("\r\n"): + if line.lower().startswith("content-length:"): + content_length = int(line.split(":")[1].strip()) + + # Read response body + body_data = b"" + while len(body_data) < content_length: + chunk = await self._reader.receive_some(content_length - len(body_data)) + if not chunk: + raise RuntimeError("Connection closed") + body_data += chunk + + response = json.loads(body_data.decode("utf-8")) + if "error" in response: + raise RuntimeError(f"JSON-RPC error: {response['error']}") + return response.get("result", {}) + + async def notify(self, method: str, params: dict[str, Any]) -> None: + """Send notification (no response expected).""" + if not self._writer: + await self.connect() + assert self._writer + + notification = { + "jsonrpc": "2.0", + "method": method, + "params": params, + } + content = json.dumps(notification, ensure_ascii=False).encode("utf-8") + header = f"Content-Length: {len(content)}\r\n\r\n".encode() + await self._writer.send_all(header + content) + + async def close(self) -> None: + """Close the socket connection.""" + if self._writer: + await self._writer.aclose() + self._writer = None + self._reader = None + + +class HttpTransport(Transport): + """HTTP/SSE transport for remote plugins.""" + + def __init__(self, url: str) -> None: + self._url = url + self._client: anyio.http.websockets.client.ClientSession | None = None + + async def connect(self) -> None: + """Initialize HTTP client.""" + # For now, use aiohttp-like interface via anyio + pass + + async def send(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + """Send HTTP POST request.""" + request = { + "jsonrpc": "2.0", + "id": str(id(params)), + "method": method, + "params": params, + } + + async with anyio.open_http_connect(self._url) as (reader, writer): + body = json.dumps(request).encode("utf-8") + writer.write( + f"POST / HTTP/1.1\r\n" + f"Host: {self._url}\r\n" + f"Content-Type: application/json\r\n" + f"Content-Length: {len(body)}\r\n" + f"\r\n".encode() + ) + writer.write(body) + await writer.aclose() + + # Read response + response = await reader.read() + # Simple parsing - in reality would need proper HTTP parsing + return json.loads(response.decode("utf-8")).get("result", {}) + + async def notify(self, method: str, params: dict[str, Any]) -> None: + """Send notification via HTTP POST.""" + notification = { + "jsonrpc": "2.0", + "method": method, + "params": params, + } + body = json.dumps(notification).encode("utf-8") + async with anyio.open_http_connect(self._url) as (_reader, writer): + writer.write( + f"POST / HTTP/1.1\r\n" + f"Host: {self._url}\r\n" + f"Content-Type: application/json\r\n" + f"Content-Length: {len(body)}\r\n" + f"\r\n".encode() + ) + writer.write(body) + await writer.aclose() + + async def close(self) -> None: + """Close the transport.""" + if self._client: + await self._client.aclose() + self._client = None diff --git a/astrbot/rust/_core.pyi b/astrbot/rust/_core.pyi index c769f3858..9cfaa52d8 100644 --- a/astrbot/rust/_core.pyi +++ b/astrbot/rust/_core.pyi @@ -12,5 +12,85 @@ class AstrbotOrchestrator: def set_protocol_connected(self, protocol: str, connected: bool) -> None: ... def get_protocol_status(self, protocol: str) -> dict[str, Any] | None: ... +# ABP Plugin types - TODO: expose via PyO3 bindings + +class PluginLoadMode: + IN_PROCESS: PluginLoadMode + OUT_OF_PROCESS: PluginLoadMode + +class PluginTransport: + STDIO: PluginTransport + UNIX_SOCKET: PluginTransport + HTTP: PluginTransport + +class PluginConfig: + name: str + version: str + load_mode: PluginLoadMode + command: str | None + args: list[str] + env: dict[str, str] + transport: PluginTransport + url: str | None + +class PluginCapabilities: + tools: bool + handlers: bool + events: bool + resources: bool + +class PluginMetadata: + display_name: str | None + description: str | None + author: str | None + homepage: str | None + support_platforms: list[str] + astrbot_version: str | None + +class AbpClient: + def __init__(self) -> None: ... + async def connect(self) -> None: ... + async def disconnect(self) -> None: ... + def is_connected(self) -> bool: ... + def register_in_process_plugin(self, config: PluginConfig) -> None: ... + def register_out_of_process_plugin(self, config: PluginConfig) -> None: ... + def unregister_plugin(self, name: str) -> None: ... + def list_plugins(self) -> list[str]: ... + def get_plugin_info(self, name: str) -> PluginInfo | None: ... + async def call_tool(self, plugin_name: str, tool_name: str, arguments: dict[str, Any]) -> ToolResult: ... + async def handle_event(self, plugin_name: str, event_type: str, event: dict[str, Any]) -> HandleEventResult: ... + +class PluginInfo: + name: str + version: str + load_mode: PluginLoadMode + capabilities: PluginCapabilities + metadata: PluginMetadata | None + tools_count: int + +class Tool: + name: str + description: str + input_schema: dict[str, Any] + +class ToolResult: + content: list[ToolContent] + +class ToolContent: + type: str + text: str | None + image: str | None + url: str | None + +class HandleEventResult: + handled: bool + results: list[MessageChainItem] + stop_propagation: bool + +class MessageChainItem: + type: str + text: str | None + def get_orchestrator() -> AstrbotOrchestrator: ... +def get_abp_client() -> AbpClient: ... def cli(args: list[str]) -> None: ... diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/.openspec.yaml b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/.openspec.yaml new file mode 100644 index 000000000..a61e7c112 --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-03-27 diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/design.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/design.md new file mode 100644 index 000000000..9efc09e4b --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/design.md @@ -0,0 +1,73 @@ +## Context + +AstrBot 当前使用 Star 插件系统，插件与核心紧耦合。ABP 协议需要实现： +- 插件作为独立服务（进程内/外） +- 主进程不读取插件配置文件，通过协议握手交换 +- 数据目录由主进程分配 +- 支持 Stdio/Unix Socket/HTTP 三种传输方式 +- Rust 核心实现协议逻辑（`astrbot/rust/src/abp/`） + +## Goals / Non-Goals + +**Goals:** +- 实现 ABP 协议核心：握手、消息路由、生命周期管理 +- 支持进程内插件（Python 直接调用）和进程外插件（独立进程） +- 支持 Stdio/Unix Socket/HTTP 传输层 +- 实现工具调用（tools/list, tools/call）和消息处理（plugin.handle_event） +- 实现事件订阅系统（plugin.subscribe, plugin.notify） + +**Non-Goals:** +- 不实现 MCP 协议（独立协议） +- 不实现 A2A/ACP 协议（独立协议） +- 不在 Python 层重复核心逻辑 + +## Decisions + +### 1. Rust 核心实现 vs Python 实现 + +**决定**：核心协议逻辑在 Rust 中实现（`astrbot/rust/src/abp/`），Python 只做 FFI 胶水层。 + +**理由**： +- ABP 协议是高性能路径，消息路由需要低延迟 +- 与项目 "Rust 核心 + Python 胶水层" 架构一致 +- 多语言插件需要稳定的 Rust FFI 接口 + +**替代方案**： +- Python 实现（被否定）：性能不足，违反项目架构 +- Go 实现（被否定）：引入新语言栈，增加维护成本 + +### 2. 传输层选择 + +**决定**：支持 Stdio（进程启动）、Unix Socket（本地进程间）、HTTP/SSE（远程）三种。 + +**理由**： +- Stdio：最简单的跨语言方案，适合单次请求 +- Unix Socket：低延迟，适合本地进程外插件 +- HTTP/SSE：支持远程插件和 Webhook 通知 + +### 3. 插件配置管理 + +**决定**：主进程不读取插件配置文件，配置通过握手 `initialize` 传递。 + +**理由**： +- 零侵入性：插件无需了解 AstrBot 目录结构 +- 统一管理：所有配置通过 WebUI 管理 +- 安全：主进程控制插件可见的配置 + +### 4. JSON-RPC 2.0 vs 自定义协议 + +**决定**：采用 JSON-RPC 2.0。 + +**理由**： +- 成熟标准，生态丰富 +- 多语言支持（Python、Rust、Go 等） +- 易于调试和测试 + +## Risks / Trade-offs + +| 风险 | 影响 | 缓解措施 | +|------|------|----------| +| Rust FFI 接口不稳定 | Python 胶水层需要频繁调整 | 接口版本化，保持向后兼容 | +| 进程外插件通信延迟 | 工具调用性能下降 | 使用 Unix Socket 代替 HTTP | +| 插件崩溃影响主进程 | 系统稳定性下降 | 进程隔离 + 超时控制 | +| 配置 Schema 同步 | WebUI 表单与实际不一致 | 握手时交换 Schema | diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/proposal.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/proposal.md new file mode 100644 index 000000000..b466efe45 --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/proposal.md @@ -0,0 +1,35 @@ +## Why + +AstrBot 需要一个统一的插件协议来实现插件的加载、消息处理、工具调用和生命周期管理。当前的 Star 插件系统与核心紧耦合，插件无法独立运行，缺乏标准化接口。ABP 协议将插件系统从核心中解耦，实现插件的零侵入性和跨语言支持。 + +## What Changes + +- **新增 ABP 协议层**：标准化的插件通信协议（JSON-RPC 2.0） +- **新增插件加载器**：支持进程内（`in_process`）和进程外（`out_of_process`）两种模式 +- **新增传输层**：Stdio、Unix Socket、HTTP/SSE 三种传输方式 +- **新增插件生命周期管理**：初始化、启动、停止、重载、配置更新 +- **新增工具调用机制**：标准化的 `tools/list` 和 `tools/call` 接口 +- **新增消息处理**：通过 `plugin.handle_event` 处理平台事件 +- **新增事件订阅**：插件可订阅 LLM 请求等系统事件 +- **Rust 核心实现**：协议核心逻辑在 Rust 运行时（`astrbot/rust/src/`） + +## Capabilities + +### New Capabilities + +- `abp-protocol`: ABP 协议核心 - 握手、消息路由、生命周期管理 +- `abp-transport`: ABP 传输层 - Stdio/Unix Socket/HTTP 实现 +- `abp-plugin-loader`: ABP 插件加载器 - 进程内/外插件管理 +- `abp-tool-router`: ABP 工具路由 - 跨插件工具发现和调用 +- `abp-event-system`: ABP 事件系统 - 事件订阅和通知 + +### Modified Capabilities + +- `agent-message` (待定): 如果需要与 Agent 消息流程整合，可能需要调整 + +## Impact + +- **新增目录**：`astrbot/rust/src/abp/` - Rust ABP 核心实现 +- **新增 Python 胶水层**：`astrbot/core/plugin/` - Python FFI 胶水代码 +- **配置文件变更**：`config.yaml` 中新增 `plugins` 配置项 +- **依赖**：`jsonrpc-derive`（Rust）、`jsonrpc-core`（Rust） diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-event-system/spec.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-event-system/spec.md new file mode 100644 index 000000000..cbb356360 --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-event-system/spec.md @@ -0,0 +1,46 @@ +## ADDED Requirements + +### Requirement: Event Subscription +插件 SHALL 支持订阅系统事件。 + +#### Scenario: Subscribe to Event +- **WHEN** 主进程发送 `plugin.subscribe` 请求，包含 event_type +- **THEN** 插件注册订阅 +- **AND** 主进程在事件发生时通知插件 + +#### Scenario: Unsubscribe from Event +- **WHEN** 主进程发送 `plugin.unsubscribe` 请求 +- **THEN** 插件取消订阅 +- **AND** 不再接收该类型事件通知 + +#### Scenario: Event Subscription Failed +- **WHEN** 订阅的事件类型不支持 +- **THEN** 插件返回 `-32207` (Event Subscribe Failed) 错误码 + +### Requirement: Event Notification +插件 SHALL 支持接收和发送事件通知。 + +#### Scenario: Receive Event Notification +- **WHEN** 主进程有事件发生（llm_request、tool_called 等） +- **THEN** 主进程发送 `plugin.notify` 通知到插件 +- **AND** 包含 event_type 和 data + +#### Scenario: Send Event from Plugin +- **WHEN** 插件需要通知主进程 +- **THEN** 插件发送 `plugin.notify` 到主进程 +- **AND** 主进程路由到对应处理器 + +### Requirement: Supported Event Types +插件 SHALL 支持以下事件类型。 + +#### Scenario: LLM Request Event +- **WHEN** LLM 开始处理请求 +- **THEN** 主进程通知订阅者，包含请求 ID 和 prompt + +#### Scenario: Tool Called Event +- **WHEN** 工具被调用 +- **THEN** 主进程通知订阅者，包含工具名称和参数 + +#### Scenario: Message Received Event +- **WHEN** 收到用户消息 +- **THEN** 主进程通知订阅者，包含消息内容 diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-plugin-loader/spec.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-plugin-loader/spec.md new file mode 100644 index 000000000..5e06929e3 --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-plugin-loader/spec.md @@ -0,0 +1,41 @@ +## ADDED Requirements + +### Requirement: In-Process Plugin Loading +进程内插件直接加载为 Python 模块，由 Rust 核心调用。 + +#### Scenario: Load In-Process Plugin +- **WHEN** 配置指定 `load_mode: "in_process"` +- **THEN** 主进程直接导入插件模块 +- **AND** 创建插件实例，传入 context、user_config、data_dirs + +#### Scenario: In-Process Plugin Invocation +- **WHEN** 调用插件方法 +- **THEN** 直接通过 Python 对象调用 +- **AND** 不经过序列化/反序列化 + +### Requirement: Out-of-Process Plugin Loading +进程外插件作为独立进程运行，通过传输层通信。 + +#### Scenario: Load Out-of-Process Plugin +- **WHEN** 配置指定 `load_mode: "out_of_process"` +- **THEN** 主进程启动插件子进程 +- **AND** 建立传输层连接（Stdio/Unix Socket/HTTP） + +#### Scenario: Plugin Process Management +- **WHEN** 插件进程异常退出 +- **THEN** 主进程记录错误日志 +- **AND** 通知相关系统组件 +- **AND** 可选：自动重启插件 + +### Requirement: Plugin Configuration +插件配置通过配置文件声明，不包含敏感信息和内部配置。 + +#### Scenario: Plugin Config Declaration +- **WHEN** 用户在 AstrBot 配置中声明插件 +- **THEN** 仅包含 name、load_mode、command、args、transport、url +- **AND** 不包含插件内部配置 + +#### Scenario: User Config Delivery +- **WHEN** 插件初始化时 +- **THEN** 主进程通过握手将 user_config 传递给插件 +- **AND** user_config 来自 secrets.yaml 或 WebUI diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-protocol/spec.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-protocol/spec.md new file mode 100644 index 000000000..e3a5bfb4c --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-protocol/spec.md @@ -0,0 +1,51 @@ +## ADDED Requirements + +### Requirement: Initialize Handshake +AstrBot 主进程 SHALL 发送 `initialize` 请求到插件，交换协议版本、客户端信息和配置。 + +#### Scenario: Successful Initialization +- **WHEN** 插件进程启动并准备就绪 +- **THEN** 主进程发送 `initialize` 请求，包含 `protocolVersion`、`clientInfo`、`capabilities`、`pluginConfig`、`dataDirs` +- **AND** 插件返回 `initialized` 响应，包含 `protocolVersion`、`serverInfo`、`capabilities`、`configSchema`、`metadata` + +#### Scenario: Version Mismatch +- **WHEN** 插件不支持客户端的协议版本 +- **THEN** 插件返回 `-32211` (Version Mismatch) 错误码 + +### Requirement: Plugin Lifecycle Management +主进程 SHALL 管理插件的启动、停止、重载和配置更新。 + +#### Scenario: Start Plugin +- **WHEN** 主进程需要激活插件 +- **THEN** 发送 `plugin.start` 请求 +- **AND** 插件进入工作状态 + +#### Scenario: Stop Plugin +- **WHEN** 主进程需要停用插件 +- **THEN** 发送 `plugin.stop` 请求 +- **AND** 插件清理资源并进入空闲状态 + +#### Scenario: Reload Plugin +- **WHEN** 插件需要热重载 +- **THEN** 发送 `plugin.reload` 请求 +- **AND** 插件重新初始化但保持进程 + +#### Scenario: Config Update +- **WHEN** 用户更新插件配置 +- **THEN** 主进程发送 `plugin.config_update` 通知 +- **AND** 插件更新内部配置 + +### Requirement: Plugin Error Handling +插件 SHALL 返回标准错误码用于问题诊断。 + +#### Scenario: Plugin Not Found +- **WHEN** 请求的插件不存在 +- **THEN** 返回 `-32200` (Plugin Not Found) 错误码 + +#### Scenario: Plugin Not Ready +- **WHEN** 请求时插件未完成初始化 +- **THEN** 返回 `-32201` (Plugin Not Ready) 错误码 + +#### Scenario: Plugin Crashed +- **WHEN** 插件进程异常退出 +- **THEN** 主进程检测到并返回 `-32202` (Plugin Crashed) 错误码 diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-tool-router/spec.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-tool-router/spec.md new file mode 100644 index 000000000..f2ac63faa --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-tool-router/spec.md @@ -0,0 +1,42 @@ +## ADDED Requirements + +### Requirement: Tool Listing +主进程 SHALL 支持通过 `tools/list` 获取插件提供的工具列表。 + +#### Scenario: List Available Tools +- **WHEN** 主进程发送 `tools/list` 请求 +- **THEN** 插件返回可用工具定义列表 +- **AND** 每个工具包含 name、description、parameters + +### Requirement: Tool Calling +主进程 SHALL 支持通过 `tools/call` 调用插件提供的工具。 + +#### Scenario: Call Tool with Arguments +- **WHEN** 主进程发送 `tools/call` 请求，包含 tool name 和 arguments +- **THEN** 插件执行对应工具 +- **AND** 返回工具执行结果（content 数组） + +#### Scenario: Tool Not Found +- **WHEN** 请求的工具名称不存在 +- **THEN** 插件返回 `-32203` (Tool Not Found) 错误码 + +#### Scenario: Tool Call Failed +- **WHEN** 工具执行过程中发生错误 +- **THEN** 插件返回 `-32204` (Tool Call Failed) 错误码 +- **AND** 错误信息包含原因 + +### Requirement: Tool Result Format +工具调用结果 SHALL 符合标准格式。 + +#### Scenario: Successful Tool Result +- **WHEN** 工具成功执行 +- **THEN** 返回 result，包含 content 数组 +- **AND** content 每个元素包含 type 和 text/image/url + +### Requirement: Tool Metadata +工具定义 SHALL 包含完整的元数据用于 LLM 函数调用。 + +#### Scenario: Tool Definition Structure +- **WHEN** 插件声明工具 +- **THEN** 必须包含：name (string)、description (string)、parameters (JSON Schema) +- **AND** parameters 必须符合 JSON Schema Draft-07 格式 diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-transport/spec.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-transport/spec.md new file mode 100644 index 000000000..122eefdfc --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/specs/abp-transport/spec.md @@ -0,0 +1,35 @@ +## ADDED Requirements + +### Requirement: Stdio Transport +插件通过标准输入/输出进行 JSON-RPC 通信。 + +#### Scenario: Stdio Message Format +- **WHEN** 发送 JSON-RPC 消息 +- **THEN** 每条消息为单行 JSON，不包含长度前缀 +- **AND** 消息之间以换行符分隔 + +### Requirement: Unix Socket Transport +进程间通过 Unix Socket 进行通信，使用 Content-Length 协议。 + +#### Scenario: Unix Socket Message Format +- **WHEN** 发送 JSON-RPC 消息 +- **THEN** 消息前添加 `Content-Length: \r\n\r\n` 前缀 +- **AND** 消息体为 UTF-8 编码的 JSON + +#### Scenario: Unix Socket Connection Lifecycle +- **WHEN** 主进程连接插件 Unix Socket +- **THEN** 建立持久连接 +- **AND** 双向复用同一连接发送请求/响应 + +### Requirement: HTTP/SSE Transport +远程插件使用 HTTP 进行请求，SSE 进行服务端推送。 + +#### Scenario: HTTP Request +- **WHEN** 主进程发送 HTTP 请求到插件 +- **THEN** 使用 POST 方法，Content-Type 为 `application/json` +- **AND** 请求体为 JSON-RPC 请求对象 + +#### Scenario: SSE Event Stream +- **WHEN** 插件需要推送通知 +- **THEN** 使用 Server-Sent Events 格式 +- **AND** 事件类型为 `plugin.notify` diff --git a/openspec/changes/archive/2026-03-27-abp-protocol-implementation/tasks.md b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/tasks.md new file mode 100644 index 000000000..1dbbbf265 --- /dev/null +++ b/openspec/changes/archive/2026-03-27-abp-protocol-implementation/tasks.md @@ -0,0 +1,70 @@ +## 1. Rust Core Infrastructure + +- [x] 1.1 Create `astrbot/rust/src/abp/` directory structure (already exists with abp.rs) +- [x] 1.2 Add ABP dependencies to `astrbot/rust/Cargo.toml` (tokio, serde, etc. already present) +- [x] 1.3 Define ABP error types and codes (-32700 to -32211) + +## 2. ABP Protocol Core (abp-protocol) + +- [x] 2.1 Implement Initialize handshake (C→P: protocolVersion, clientInfo, capabilities, pluginConfig, dataDirs) +- [x] 2.2 Implement Initialize Response (P→C: protocolVersion, serverInfo, capabilities, configSchema, metadata) +- [x] 2.3 Implement Plugin lifecycle methods (plugin.start, plugin.stop, plugin.reload, plugin.config_update) +- [x] 2.4 Implement plugin.error_handler (PluginNotFound, PluginNotReady, PluginCrashed - via error codes) + +## 3. Transport Layer (abp-transport) + +- [x] 3.1 Implement Stdio transport (single-line JSON messages) - in Python transport.py +- [x] 3.2 Implement Unix Socket transport (Content-Length framing) - in Rust abp.rs and Python +- [x] 3.3 Implement HTTP/SSE transport (POST requests, SSE streams) - in Rust abp.rs and Python +- [ ] 3.4 Add connection pooling for Unix Socket + +## 4. Plugin Loader (abp-plugin-loader) + +- [ ] 4.1 Implement PluginLoader trait +- [ ] 4.2 Implement InProcessPluginLoader (Python module loading) +- [x] 4.3 Implement OutOfProcessPluginLoader (process spawning + transport) +- [x] 4.4 Implement plugin config parsing (name, load_mode, command, args, transport, url) - in Rust abp.rs +- [x] 4.5 Implement data directory allocation (dataDirs.root, dataDirs.plugin_data, dataDirs.temp) + +## 5. Tool Router (abp-tool-router) + +- [ ] 5.1 Implement tools/list endpoint (return tool definitions) +- [x] 5.2 Implement tools/call endpoint (execute tool with arguments) - in Rust abp.rs +- [x] 5.3 Implement tool result formatting (content array with type/text) - in Rust abp.rs +- [ ] 5.4 Add tool schema validation (JSON Schema Draft-07) +- [ ] 5.5 Implement cross-plugin tool discovery + +## 6. Event System (abp-event-system) + +- [x] 6.1 Implement plugin.subscribe (register event subscription) - Python client +- [x] 6.2 Implement plugin.unsubscribe (remove event subscription) - Python client +- [x] 6.3 Implement plugin.notify (bidirectional event notification) - Python client +- [x] 6.4 Define event types (llm_request, tool_called, message_received) - in Rust abp.rs +- [x] 6.5 Implement event routing (P→C and C→P notifications) - Python client + +## 7. Python Glue Layer + +- [x] 7.1 Create `astrbot/core/plugin/` Python package +- [ ] 7.2 Implement FFI bindings to Rust ABP core (_core.so) +- [x] 7.3 Create PluginManager Python class (wraps Rust PluginLoader) +- [x] 7.4 Create PluginClient Python class (wraps transport) +- [x] 7.5 Add type stubs in `astrbot/rust/_core.pyi` + +## 8. Integration & Configuration + +- [ ] 8.1 Add plugins section to config.yaml schema +- [ ] 8.2 Implement PluginRegistry (plugin discovery and registration) +- [ ] 8.3 Add ABP initialization to AstrBot startup +- [ ] 8.4 Integrate with existing Star plugin system (backward compatibility) +- [ ] 8.5 Add WebUI support for plugin configuration + +## 9. Testing + +- [ ] 9.1 Write unit tests for ABP protocol core +- [ ] 9.2 Write integration tests for transport layer +- [ ] 9.3 Write tests for plugin loading (in-process and out-of-process) +- [ ] 9.4 Write tests for tool router +- [ ] 9.5 Write tests for event system +- [x] 9.6 Run `ruff check .` and `ruff format .` +- [ ] 9.7 Run `uvx ty check` for type validation +- [x] 9.8 Run `cargo check` and `cargo fmt` for Rust code diff --git a/openspec/rust-ffi.md b/openspec/rust-ffi.md new file mode 100644 index 000000000..c182f4944 --- /dev/null +++ b/openspec/rust-ffi.md @@ -0,0 +1,156 @@ +# Rust FFI 接口规范 + +## 概述 + +AstrBot 采用 **Rust 核心 + Python 胶水层** 架构。Rust 核心编译为 `_core.so`，通过 FFI 暴露接口供 Python 调用。 + +**绑定方案**：使用 [PyO3](https://pyo3.rs/)（`rust/src/lib.rs` 导出 `#[pymodule]`），Python 端通过 `ffi` 模块调用。 + +**禁止使用 `ctypes`** —— 所有 FFI 交互必须通过 PyO3 绑定。 + +## 核心模块布局 + +``` +astrbot/rust/src/ +├── lib.rs # 入口，导出 #[pymodule] +├── orchestrator.rs # AstrbotOrchestrator 主协调器 +├── abp/ # ABP 协议实现 +│ ├── mod.rs +│ ├── protocol.rs # 握手、消息路由 +│ ├── loader.rs # 插件加载器 +│ ├── transport.rs # Stdio/Unix Socket/HTTP +│ └── error.rs # ABP 错误码 +├── message/ # 消息缓冲（双缓冲区） +│ ├── mod.rs +│ ├── input_buffer.rs +│ └── output_buffer.rs +├── flow_control.rs # 流控引擎 +├── tool_router.rs # 工具路由（Internal/MCP/Skills） +└── agent.rs # Agent 协调 +``` + +## FFI 函数签名 + +### orchestrator.rs + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `get_orchestrator()` | `def get_orchestrator() -> AstrbotOrchestrator` | `AstrbotOrchestrator` | 单例获取 | +| `orchestrator.start()` | `def start(self) -> None` | `None` | 启动核心 | +| `orchestrator.stop()` | `def stop(self) -> None` | `None` | 停止核心 | +| `orchestrator.is_running()` | `def is_running(self) -> bool` | `bool` | 运行状态 | +| `orchestrator.register_star()` | `def register_star(self, name: str, handler: str) -> None` | `None` | 注册 Star | +| `orchestrator.unregister_star()` | `def unregister_star(self, name: str) -> None` | `None` | 注销 Star | +| `orchestrator.list_stars()` | `def list_stars(self) -> list[str]` | `list[str]` | 列出 Stars | +| `orchestrator.record_activity()` | `def record_activity(self) -> None` | `None` | 记录活动 | +| `orchestrator.get_stats()` | `def get_stats(self) -> dict` | `dict[str, Any]` | 获取统计 | +| `orchestrator.set_protocol_connected()` | `def set_protocol_connected(self, protocol: str, connected: bool) -> None` | `None` | 设置协议连接状态 | +| `orchestrator.get_protocol_status()` | `def get_protocol_status(self, protocol: str) -> dict | None` | `dict[str, Any] \| None` | 获取协议状态 | + +### abp/protocol.rs（ABP 插件协议） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `plugin_initialize()` | `def plugin_initialize(config: dict) -> InitializeResult` | `InitializeResult` | 初始化握手 | +| `plugin_start()` | `def plugin_start(self, plugin_id: str) -> None` | `None` | 启动插件 | +| `plugin_stop()` | `def plugin_stop(self, plugin_id: str) -> None` | `None` | 停止插件 | +| `plugin_reload()` | `def plugin_reload(self, plugin_id: str) -> None` | `None` | 重载插件 | +| `plugin_config_update()` | `def plugin_config_update(self, plugin_id: str, config: dict) -> None` | `None` | 更新配置 | + +### abp/loader.rs（插件加载器） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `load_plugin()` | `def load_plugin(self, plugin_config: dict) -> PluginHandle` | `PluginHandle` | 加载插件 | +| `unload_plugin()` | `def unload_plugin(self, plugin_id: str) -> None` | `None` | 卸载插件 | +| `list_loaded_plugins()` | `def list_loaded_plugins(self) -> list[str]` | `list[str]` | 列出已加载 | + +### abp/transport.rs（传输层） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `create_stdio_transport()` | `def create_stdio_transport(cmd: str, args: list[str]) -> Transport` | `Transport` | 创建 stdio 传输 | +| `create_unix_transport()` | `def create_unix_transport(path: str) -> Transport` | `Transport` | 创建 Unix Socket 传输 | +| `create_http_transport()` | `def create_http_transport(url: str) -> Transport` | `Transport` | 创建 HTTP/SSE 传输 | + +### message/input_buffer.rs（输入缓冲区） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `enqueue_message()` | `def enqueue_message(self, event: dict) -> str` | `str` | 入队，返回 message_id | +| `dequeue_messages()` | `def dequeue_messages(self, limit: int) -> list[dict]` | `list[dict]` | 批量出队 | +| `get_queue_depth()` | `def get_queue_depth(self, session_id: str) -> int` | `int` | 获取队列深度 | +| `clear_queue()` | `def clear_queue(self, session_id: str) -> None` | `None` | 清空队列 | + +### message/output_buffer.rs（输出缓冲区） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `enqueue_result()` | `def enqueue_result(self, session_id: str, result: dict) -> str` | `str` | 入队 | +| `dequeue_result()` | `def dequeue_result(self, session_id: str) -> dict \| None` | `dict \| None` | 出队（非阻塞） | +| `set_dispatch_strategy()` | `def set_dispatch_strategy(self, strategy: str) -> None` | `None` | 设置分发策略 | + +### flow_control.rs（流控引擎） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `set_rate_limit()` | `def set_rate_limit(self, requests: int, period: float) -> None` | `None` | 设置限流 | +| `acquire()` | `def acquire(self) -> bool` | `bool` | 获取令牌（非阻塞） | +| `wait_for_token()` | `def wait_for_token(self, timeout: float) -> bool` | `bool` | 等待令牌（阻塞） | + +### tool_router.rs（工具路由） + +| 函数 | Python 签名 | 返回类型 | 说明 | +|------|-------------|----------|------| +| `register_internal_tool()` | `def register_internal_tool(self, name: str, schema: dict) -> None` | `None` | 注册内部工具 | +| `register_mcp_server()` | `def register_mcp_server(self, name: str, transport: Transport) -> None` | `None` | 注册 MCP 服务器 | +| `route_tool_call()` | `def route_tool_call(self, tool_name: str, arguments: dict) -> ToolResult` | `ToolResult` | 路由工具调用 | +| `list_tools()` | `def list_tools(self) -> list[dict]` | `list[dict]` | 列出所有工具 | + +## 类型映射 + +| Rust 类型 | Python 类型 | 说明 | +|-----------|------------|------| +| `bool` | `bool` | - | +| `i32`, `i64` | `int` | - | +| `f64` | `float` | - | +| `String` | `str` | - | +| `Vec` | `list[T]` | - | +| `HashMap` | `dict[K,V]` | - | +| `Option` | `T \| None` | - | +| `Result` | 异常或 T | 失败抛 Python 异常 | + +## 错误处理 + +Rust 层返回的错误统一转换为 Python 异常： + +| ABP 错误码 | Python 异常 | +|------------|-------------| +| -32200 | `PluginNotFoundError` | +| -32201 | `PluginNotReadyError` | +| -32202 | `PluginCrashedError` | +| -32203 | `ToolNotFoundError` | +| -32204 | `ToolCallFailedError` | +| -32205 | `HandlerNotFoundError` | +| -32206 | `HandlerError` | +| -32207 | `EventSubscribeError` | +| -32208 | `PermissionDeniedError` | +| -32209 | `ConfigError` | +| -32210 | `DependencyMissingError` | +| -32211 | `VersionMismatchError` | +| -32603 | `InternalError` | + +## 实现要求 + +1. **所有函数必须线程安全**（Rust 核心使用 `Arc>` 保护共享状态） +2. **异步操作在 Rust 内部完成**，Python 侧得到的是 Future 或直接结果 +3. **不得在 FFI 边界传递闭包或函数指针** +4. **版本化接口**：FFI 接口变更时提升 `__version__`，保持向后兼容 +5. **文档注释**：Rust 代码使用 `///` 注释，会通过 PyO3 生成 Python docstring + +## 相关文件 + +- [config.yaml](config.yaml) - 项目架构说明（包含 rust_core、internal_package 等 directive） +- [abp.md](abp.md) - ABP 协议规范 +- [agent-message.md](agent-message.md) - 消息处理规范 +- `openspec/changes/abp-protocol-implementation/` - ABP 实现 change proposal（含详细任务清单） diff --git a/rust/src/error.rs b/rust/src/error.rs index f63ef6ac0..b76271c19 100644 --- a/rust/src/error.rs +++ b/rust/src/error.rs @@ -27,4 +27,120 @@ pub enum AstrBotError { #[error("JSON error: {0}")] Json(#[from] serde_json::Error), + + // ABP Protocol error codes (-32700 to -32211) + #[error("Parse error: {0}")] + ParseError(String), + + #[error("Invalid request: {0}")] + InvalidRequest(String), + + #[error("Method not found: {0}")] + MethodNotFound(String), + + #[error("Invalid params: {0}")] + InvalidParams(String), + + #[error("Internal error: {0}")] + InternalError(String), + + #[error("Plugin not found: {0}")] + PluginNotFound(String), + + #[error("Plugin not ready: {0}")] + PluginNotReady(String), + + #[error("Plugin crashed: {0}")] + PluginCrashed(String), + + #[error("Tool not found: {0}")] + ToolNotFound(String), + + #[error("Tool call failed: {0}")] + ToolCallFailed(String), + + #[error("Handler not found: {0}")] + HandlerNotFound(String), + + #[error("Handler error: {0}")] + HandlerError(String), + + #[error("Event subscription failed: {0}")] + EventSubscribeFailed(String), + + #[error("Permission denied: {0}")] + PermissionDenied(String), + + #[error("Config error: {0}")] + ConfigError(String), + + #[error("Dependency missing: {0}")] + DependencyMissing(String), + + #[error("Version mismatch: {0}")] + VersionMismatch(String), +} + +/// ABP Protocol error codes +pub mod abp_error_codes { + use super::AstrBotError; + + /// JSON-RPC parse error + pub const PARSE_ERROR: i32 = -32700; + /// Invalid request + pub const INVALID_REQUEST: i32 = -32600; + /// Method not found + pub const METHOD_NOT_FOUND: i32 = -32601; + /// Invalid params + pub const INVALID_PARAMS: i32 = -32602; + /// Internal error + pub const INTERNAL_ERROR: i32 = -32603; + /// Plugin not found + pub const PLUGIN_NOT_FOUND: i32 = -32200; + /// Plugin not ready + pub const PLUGIN_NOT_READY: i32 = -32201; + /// Plugin crashed + pub const PLUGIN_CRASHED: i32 = -32202; + /// Tool not found + pub const TOOL_NOT_FOUND: i32 = -32203; + /// Tool call failed + pub const TOOL_CALL_FAILED: i32 = -32204; + /// Handler not found + pub const HANDLER_NOT_FOUND: i32 = -32205; + /// Handler error + pub const HANDLER_ERROR: i32 = -32206; + /// Event subscribe failed + pub const EVENT_SUBSCRIBE_FAILED: i32 = -32207; + /// Permission denied + pub const PERMISSION_DENIED: i32 = -32208; + /// Config error + pub const CONFIG_ERROR: i32 = -32209; + /// Dependency missing + pub const DEPENDENCY_MISSING: i32 = -32210; + /// Version mismatch + pub const VERSION_MISMATCH: i32 = -32211; + + /// Convert ABP error code to AstrBotError + pub fn from_code(code: i32, message: String) -> AstrBotError { + match code { + PARSE_ERROR => AstrBotError::ParseError(message), + INVALID_REQUEST => AstrBotError::InvalidRequest(message), + METHOD_NOT_FOUND => AstrBotError::MethodNotFound(message), + INVALID_PARAMS => AstrBotError::InvalidParams(message), + INTERNAL_ERROR => AstrBotError::InternalError(message), + PLUGIN_NOT_FOUND => AstrBotError::PluginNotFound(message), + PLUGIN_NOT_READY => AstrBotError::PluginNotReady(message), + PLUGIN_CRASHED => AstrBotError::PluginCrashed(message), + TOOL_NOT_FOUND => AstrBotError::ToolNotFound(message), + TOOL_CALL_FAILED => AstrBotError::ToolCallFailed(message), + HANDLER_NOT_FOUND => AstrBotError::HandlerNotFound(message), + HANDLER_ERROR => AstrBotError::HandlerError(message), + EVENT_SUBSCRIBE_FAILED => AstrBotError::EventSubscribeFailed(message), + PERMISSION_DENIED => AstrBotError::PermissionDenied(message), + CONFIG_ERROR => AstrBotError::ConfigError(message), + DEPENDENCY_MISSING => AstrBotError::DependencyMissing(message), + VERSION_MISMATCH => AstrBotError::VersionMismatch(message), + _ => AstrBotError::Protocol(format!("Unknown error code {}: {}", code, message)), + } + } } diff --git a/rust/src/server.rs b/rust/src/server.rs index b00879905..127880ea7 100644 --- a/rust/src/server.rs +++ b/rust/src/server.rs @@ -1,17 +1,15 @@ -AstrBot/rust/src/server.rs -```rust //! AstrBot HTTP/WebSocket Server //! //! High-performance async HTTP server with WebSocket support for real-time communication. //! Implements JWT authentication and REST API endpoints. -use base64::engine::general_purpose::STANDARD as BASE64; use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::Arc; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::{mpsc, RwLock}; +use tokio::sync::{RwLock, mpsc}; use tokio_tungstenite::{accept_async, tungstenite::Message}; use uuid::Uuid; @@ -67,7 +65,7 @@ impl JwtAuth { pub fn generate_token(&self, username: &str) -> Result { let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOODY) + .duration_since(std::time::UNIX_EPOCH) .map_err(|e| ServerError::Jwt(e.to_string()))? .as_secs(); @@ -79,7 +77,8 @@ impl JwtAuth { let header = format!("{{\"alg\":\"HS256\",\"typ\":\"JWT\"}}"); let header_b64 = BASE64.encode(header.as_bytes()); - let payload_b64 = BASE64.encode(serde_json::to_vec(&claims).map_err(|e| ServerError::Jwt(e.to_string()))?; + let payload_b64 = BASE64 + .encode(serde_json::to_vec(&claims).map_err(|e| ServerError::Jwt(e.to_string()))?); let signature = self.sign_jwt(&format!("{}.{}", header_b64, payload_b64)); @@ -241,10 +240,13 @@ impl WsManager { pub async fn send_to(&self, id: &str, message: &str) -> Result<(), ServerError> { let connections = self.connections.read().await; - let sender = connections.get(id).ok_or_else(|| ServerError::NotFound(id.to_string()))?; - sender.send(message.to_string()).await.map_err(|_| { - ServerError::WebSocket("Failed to send".to_string()) - })?; + let sender = connections + .get(id) + .ok_or_else(|| ServerError::NotFound(id.to_string()))?; + sender + .send(message.to_string()) + .await + .map_err(|_| ServerError::WebSocket("Failed to send".to_string()))?; Ok(()) } @@ -276,13 +278,16 @@ pub struct HttpRequest { impl HttpRequest { pub fn parse(buf: &[u8]) -> Result { let header_end = buf.windows(4).position(|w| w == b"\r\n\r\n"); - let header_end = header_end.ok_or_else(|| ServerError::BadRequest("Invalid HTTP header".to_string()))?; + let header_end = + header_end.ok_or_else(|| ServerError::BadRequest("Invalid HTTP header".to_string()))?; let header_str = String::from_utf8_lossy(&buf[..header_end]); let body = buf[header_end + 4..].to_vec(); let mut lines = header_str.split("\r\n"); - let request_line = lines.next().ok_or_else(|| ServerError::BadRequest("Missing request line".to_string()))?; + let request_line = lines + .next() + .ok_or_else(|| ServerError::BadRequest("Missing request line".to_string()))?; let parts: Vec<&str> = request_line.split_whitespace().collect(); if parts.len() < 2 { @@ -344,10 +349,7 @@ impl ServerState { // API Handlers // ============================================================================ -async fn handle_api( - state: &ServerState, - req: &HttpRequest, -) -> Result { +async fn handle_api(state: &ServerState, req: &HttpRequest) -> Result { let (path, method) = (req.path.as_str(), req.method.as_str()); // Login endpoint (no auth required) @@ -357,7 +359,10 @@ async fn handle_api( // Authenticate other requests let token = req.auth_token().ok_or(ServerError::Unauthorized)?; - state.jwt_auth.verify_token(token).map_err(|_| ServerError::Unauthorized)?; + state + .jwt_auth + .verify_token(token) + .map_err(|_| ServerError::Unauthorized)?; match (path, method) { ("/api/stats", "GET") => { @@ -372,32 +377,33 @@ async fn handle_api( }), )) } - ("/api/sessions", "GET") => { - Ok(HttpResponse::json( - 200, - "OK", - serde_json::json!({ - "sessions": state.orchestrator.list_stars(), - }), - )) - } - ("/api/health", "GET") => { - Ok(HttpResponse::json( - 200, - "OK", - serde_json::json!({ - "status": "ok", - "orchestrator_running": state.orchestrator.is_running(), - "websocket_connections": state.ws_manager.count().await, - }), - )) - } + ("/api/sessions", "GET") => Ok(HttpResponse::json( + 200, + "OK", + serde_json::json!({ + "sessions": state.orchestrator.list_stars(), + }), + )), + ("/api/health", "GET") => Ok(HttpResponse::json( + 200, + "OK", + serde_json::json!({ + "status": "ok", + "orchestrator_running": state.orchestrator.is_running(), + "websocket_connections": state.ws_manager.count().await, + }), + )), _ if path.starts_with("/api/sessions/") && method == "DELETE" => { let session_id = path.strip_prefix("/api/sessions/").unwrap_or(""); - state.orchestrator.unregister_star(session_id).map_err(|_| { - ServerError::NotFound("Session not found".to_string()) - })?; - Ok(HttpResponse::json(200, "OK", serde_json::json!({"success": true}))) + state + .orchestrator + .unregister_star(session_id) + .map_err(|_| ServerError::NotFound("Session not found".to_string()))?; + Ok(HttpResponse::json( + 200, + "OK", + serde_json::json!({"success": true}), + )) } _ if path.starts_with("/api/sessions/") && method == "GET" => { let session_id = path.strip_prefix("/api/sessions/").unwrap_or(""); @@ -410,15 +416,16 @@ async fn handle_api( _ => Ok(HttpResponse::json( 404, "Not Found", - ApiResponse::error(404, "Endpoint not found"), + serde_json::to_value(ApiResponse::error(404, "Endpoint not found")) + .unwrap_or(serde_json::Value::Null), )), } } async fn handle_login(state: &ServerState, req: &HttpRequest) -> Result { let body = String::from_utf8_lossy(&req.body); - let login: serde_json::Value = - serde_json::from_str(&body).map_err(|_| ServerError::BadRequest("Invalid JSON".to_string()))?; + let login: serde_json::Value = serde_json::from_str(&body) + .map_err(|_| ServerError::BadRequest("Invalid JSON".to_string()))?; let password = login .get("password") @@ -429,7 +436,8 @@ async fn handle_login(state: &ServerState, req: &HttpRequest) -> Result(100); + let (tx, mut rx) = mpsc::channel::(100); state.ws_manager.register(connection_id.clone(), tx).await; // Send welcome message @@ -478,7 +486,7 @@ async fn handle_websocket(state: ServerState, stream: TcpStream, addr: std::net: msg = reader.next() => { match msg { Some(Ok(Message::Text(text))) => { - if let Err(e) = handle_ws_message(&state, text.to_string()).await { + if let Err(e) = handle_ws_message(&state, &connection_id, text.to_string()).await { tracing::error!("WebSocket error: {}", e); break; } @@ -489,11 +497,9 @@ async fn handle_websocket(state: ServerState, stream: TcpStream, addr: std::net: _ => {} } } - Some(msg) = rx.recv() => { - if let Some(text) = msg { - if writer.send(Message::Text(text.into())).await.is_err() { - break; - } + Some(text) = rx.recv() => { + if writer.send(Message::Text(text.into())).await.is_err() { + break; } } } @@ -502,11 +508,18 @@ async fn handle_websocket(state: ServerState, stream: TcpStream, addr: std::net: state.ws_manager.unregister(&connection_id).await; } -async fn handle_ws_message(state: &ServerState, msg: String) -> Result<(), ServerError> { - let parsed: serde_json::Value = - serde_json::from_str(&msg).map_err(|_| ServerError::BadRequest("Invalid JSON".to_string()))?; +async fn handle_ws_message( + state: &ServerState, + connection_id: &str, + msg: String, +) -> Result<(), ServerError> { + let parsed: serde_json::Value = serde_json::from_str(&msg) + .map_err(|_| ServerError::BadRequest("Invalid JSON".to_string()))?; - let msg_type = parsed.get("type").and_then(|v| v.as_str()).unwrap_or("unknown"); + let msg_type = parsed + .get("type") + .and_then(|v| v.as_str()) + .unwrap_or("unknown"); let response = match msg_type { "ping" => serde_json::json!({"type": "pong"}), @@ -535,8 +548,9 @@ async fn handle_ws_message(state: &ServerState, msg: String) -> Result<(), Serve Ok(()) } -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use futures_util::SinkExt; use futures_util::StreamExt; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; // ============================================================================ // HTTP Server @@ -576,7 +590,7 @@ impl HttpServer { } } -async fn handle_connection(stream: TcpStream, state: ServerState) -> Result<(), ServerError> { +async fn handle_connection(mut stream: TcpStream, state: ServerState) -> Result<(), ServerError> { let mut buf = vec![0u8; 8192]; let n = stream.read(&mut buf).await?; buf.truncate(n);