feat: Enhance Supervisor and Worker Runtime with Internal Capabilities and Lifecycle Hooks

- Added internal capability for handler invocation in SupervisorRuntime.
- Implemented lifecycle hooks (on_start, on_stop) in Star class and PluginWorkerRuntime.
- Updated CapabilityRouter to support request_id in call and stream handlers.
- Refactored message handling in PeerRuntime to include request_id.
- Introduced tests for API contract, legacy adapter, and migration scenarios.
- Improved WebSocketServerTransport to manage connection state more effectively.
- Enhanced plugin loading and discovery to maintain compatibility with legacy systems.

.gitignore

@@ -36,3 +36,4 @@ plugins/.venv/
 .idea/
 .vscode/
 *.iml
+uv.lock

AGENTS.md

@@ -0,0 +1,5 @@
+# CLAUDE Notes
+
+- 2026-03-12: `refactor.md` on disk was empty, while the active collaboration context contained the full v4 refactor design. Treat the conversation-approved v4 design as source of truth unless a newer committed document replaces it.
+- 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
+- 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.

CLAUDE.md

@@ -1,3 +1,5 @@
 # CLAUDE Notes
 
 - 2026-03-12: `refactor.md` on disk was empty, while the active collaboration context contained the full v4 refactor design. Treat the conversation-approved v4 design as source of truth unless a newer committed document replaces it.
+- 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
+- 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.

src-new/astrbot_sdk/_legacy_api.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any
+
+from loguru import logger
+
+from .clients.llm import LLMResponse
+from .context import Context as NewContext
+
+MIGRATION_DOC_URL = "https://docs.astrbot.app/migration/v3"
+_warned_methods: set[str] = set()
+
+
+def _warn_once(old_name: str, replacement: str) -> None:
+    if old_name in _warned_methods:
+        return
+    _warned_methods.add(old_name)
+    logger.warning(
+        "[AstrBot] 警告：{} 已过时。请替换为：{}\n迁移文档：{}",
+        old_name,
+        replacement,
+        MIGRATION_DOC_URL,
+    )
+
+
+class LegacyConversationManager:
+    def __init__(self, parent: "LegacyContext") -> None:
+        self._parent = parent
+        self._counters: defaultdict[str, int] = defaultdict(int)
+
+    def _ctx(self) -> NewContext:
+        return self._parent.require_runtime_context()
+
+    async def new_conversation(
+        self,
+        unified_msg_origin: str,
+        platform_id: str | None = None,
+        content: list[dict] | None = None,
+        title: str | None = None,
+        persona_id: str | None = None,
+    ) -> str:
+        ctx = self._ctx()
+        self._counters[unified_msg_origin] += 1
+        conversation_id = f"{ctx.plugin_id}-conv-{self._counters[unified_msg_origin]}"
+        stored = await ctx.db.get("__compat_conversations__") or {}
+        stored[conversation_id] = {
+            "unified_msg_origin": unified_msg_origin,
+            "platform_id": platform_id,
+            "content": content or [],
+            "title": title,
+            "persona_id": persona_id,
+        }
+        await ctx.db.set("__compat_conversations__", stored)
+        return conversation_id
+
+
+class LegacyContext:
+    def __init__(self, plugin_id: str) -> None:
+        self.plugin_id = plugin_id
+        self._runtime_context: NewContext | None = None
+        self.conversation_manager = LegacyConversationManager(self)
+
+    def bind_runtime_context(self, runtime_context: NewContext) -> None:
+        self._runtime_context = runtime_context
+
+    def require_runtime_context(self) -> NewContext:
+        if self._runtime_context is None:
+            raise RuntimeError("LegacyContext 尚未绑定运行时 Context")
+        return self._runtime_context
+
+    async def llm_generate(
+        self,
+        chat_provider_id: str,
+        prompt: str | None = None,
+        image_urls: list[str] | None = None,
+        tools: Any | None = None,
+        system_prompt: str | None = None,
+        contexts: list[dict] | None = None,
+        **kwargs: Any,
+    ) -> LLMResponse:
+        _warn_once("context.llm_generate()", "ctx.llm.chat(prompt)")
+        ctx = self.require_runtime_context()
+        return await ctx.llm.chat_raw(
+            prompt or "",
+            system=system_prompt,
+            history=contexts or [],
+            image_urls=image_urls or [],
+            tools=tools,
+            **kwargs,
+        )
+
+    async def tool_loop_agent(
+        self,
+        chat_provider_id: str,
+        prompt: str | None = None,
+        image_urls: list[str] | None = None,
+        tools: Any | None = None,
+        system_prompt: str | None = None,
+        contexts: list[dict] | None = None,
+        max_steps: int = 30,
+        **kwargs: Any,
+    ) -> LLMResponse:
+        _warn_once("context.tool_loop_agent()", "ctx.llm.chat_raw(...)")
+        ctx = self.require_runtime_context()
+        return await ctx.llm.chat_raw(
+            prompt or "",
+            system=system_prompt,
+            history=contexts or [],
+            image_urls=image_urls or [],
+            tools=tools,
+            max_steps=max_steps,
+            **kwargs,
+        )
+
+    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))
+
+    async def add_llm_tools(self, *tools: Any) -> None:
+        _warn_once("context.add_llm_tools()", "ctx.llm.chat_raw(..., tools=...)")
+        return None
+
+    async def put_kv_data(self, key: str, value: dict[str, Any]) -> None:
+        _warn_once("context.put_kv_data()", "ctx.db.set(key, value)")
+        ctx = self.require_runtime_context()
+        await ctx.db.set(key, value)
+
+    async def get_kv_data(self, key: str) -> dict[str, Any] | None:
+        _warn_once("context.get_kv_data()", "ctx.db.get(key)")
+        ctx = self.require_runtime_context()
+        return await ctx.db.get(key)
+
+    async def delete_kv_data(self, key: str) -> None:
+        _warn_once("context.delete_kv_data()", "ctx.db.delete(key)")
+        ctx = self.require_runtime_context()
+        await ctx.db.delete(key)
+
+
+class CommandComponent:
+    @classmethod
+    def _astrbot_create_legacy_context(cls, plugin_id: str) -> LegacyContext:
+        # Loader 通过这个工厂拿到旧 Context，避免核心运行时直接依赖 compat 实现。
+        return LegacyContext(plugin_id)
+
+
+Context = LegacyContext
+
+__all__ = [
+    "CommandComponent",
+    "Context",
+    "LegacyContext",
+    "LegacyConversationManager",
+    "MIGRATION_DOC_URL",
+]

src-new/astrbot_sdk/api/components/command.py

@@ -1,3 +1,3 @@
-from ...compat import CommandComponent
+from ..._legacy_api import CommandComponent
 
 __all__ = ["CommandComponent"]

src-new/astrbot_sdk/api/star/context.py

@@ -1,3 +1,3 @@
-from ...compat import Context
+from ..._legacy_api import Context
 
 __all__ = ["Context"]

src-new/astrbot_sdk/clients/memory.py

@@ -13,8 +13,18 @@ class MemoryClient:
         output = await self._proxy.call("memory.search", {"query": query})
         return list(output.get("items", []))
 
-    async def save(self, key: str, value: dict[str, Any]) -> None:
-        await self._proxy.call("memory.save", {"key": key, "value": value})
+    async def save(
+        self,
+        key: str,
+        value: dict[str, Any] | None = None,
+        **extra: Any,
+    ) -> None:
+        if value is not None and not isinstance(value, dict):
+            raise TypeError("memory.save 的 value 必须是 dict")
+        payload = dict(value or {})
+        if extra:
+            payload.update(extra)
+        await self._proxy.call("memory.save", {"key": key, "value": payload})
 
     async def delete(self, key: str) -> None:
         await self._proxy.call("memory.delete", {"key": key})

src-new/astrbot_sdk/compat.py

@@ -1,143 +1,8 @@
-from __future__ import annotations
+from ._legacy_api import CommandComponent, Context, LegacyContext, LegacyConversationManager
 
-from collections import defaultdict
-from typing import Any
-
-from loguru import logger
-
-from .clients.llm import LLMResponse
-from .context import Context as NewContext
-
-_warned_methods: set[str] = set()
-
-
-def _warn_once(old_name: str, replacement: str) -> None:
-    if old_name in _warned_methods:
-        return
-    _warned_methods.add(old_name)
-    logger.warning(
-        "[AstrBot] 警告：{} 已过时。请替换为：{}",
-        old_name,
-        replacement,
-    )
-
-
-class LegacyConversationManager:
-    def __init__(self, parent: "LegacyContext") -> None:
-        self._parent = parent
-        self._counters: defaultdict[str, int] = defaultdict(int)
-
-    def _ctx(self) -> NewContext:
-        return self._parent.require_runtime_context()
-
-    async def new_conversation(
-        self,
-        unified_msg_origin: str,
-        platform_id: str | None = None,
-        content: list[dict] | None = None,
-        title: str | None = None,
-        persona_id: str | None = None,
-    ) -> str:
-        ctx = self._ctx()
-        self._counters[unified_msg_origin] += 1
-        conversation_id = f"{ctx.plugin_id}-conv-{self._counters[unified_msg_origin]}"
-        stored = await ctx.db.get("__compat_conversations__") or {}
-        stored[conversation_id] = {
-            "unified_msg_origin": unified_msg_origin,
-            "platform_id": platform_id,
-            "content": content or [],
-            "title": title,
-            "persona_id": persona_id,
-        }
-        await ctx.db.set("__compat_conversations__", stored)
-        return conversation_id
-
-
-class LegacyContext:
-    def __init__(self, plugin_id: str) -> None:
-        self.plugin_id = plugin_id
-        self._runtime_context: NewContext | None = None
-        self.conversation_manager = LegacyConversationManager(self)
-
-    def bind_runtime_context(self, runtime_context: NewContext) -> None:
-        self._runtime_context = runtime_context
-
-    def require_runtime_context(self) -> NewContext:
-        if self._runtime_context is None:
-            raise RuntimeError("LegacyContext 尚未绑定运行时 Context")
-        return self._runtime_context
-
-    async def llm_generate(
-        self,
-        chat_provider_id: str,
-        prompt: str | None = None,
-        image_urls: list[str] | None = None,
-        tools: Any | None = None,
-        system_prompt: str | None = None,
-        contexts: list[dict] | None = None,
-        **kwargs: Any,
-    ) -> LLMResponse:
-        _warn_once("context.llm_generate()", "ctx.llm.chat(prompt)")
-        ctx = self.require_runtime_context()
-        return await ctx.llm.chat_raw(
-            prompt or "",
-            system=system_prompt,
-            history=contexts or [],
-            image_urls=image_urls or [],
-            tools=tools,
-            **kwargs,
-        )
-
-    async def tool_loop_agent(
-        self,
-        chat_provider_id: str,
-        prompt: str | None = None,
-        image_urls: list[str] | None = None,
-        tools: Any | None = None,
-        system_prompt: str | None = None,
-        contexts: list[dict] | None = None,
-        max_steps: int = 30,
-        **kwargs: Any,
-    ) -> LLMResponse:
-        _warn_once("context.tool_loop_agent()", "ctx.llm.chat_raw(...)")
-        ctx = self.require_runtime_context()
-        return await ctx.llm.chat_raw(
-            prompt or "",
-            system=system_prompt,
-            history=contexts or [],
-            image_urls=image_urls or [],
-            tools=tools,
-            max_steps=max_steps,
-            **kwargs,
-        )
-
-    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))
-
-    async def add_llm_tools(self, *tools: Any) -> None:
-        _warn_once("context.add_llm_tools()", "ctx.llm.chat_raw(..., tools=...)")
-        return None
-
-    async def put_kv_data(self, key: str, value: dict[str, Any]) -> None:
-        _warn_once("context.put_kv_data()", "ctx.db.set(key, value)")
-        ctx = self.require_runtime_context()
-        await ctx.db.set(key, value)
-
-    async def get_kv_data(self, key: str) -> dict[str, Any] | None:
-        _warn_once("context.get_kv_data()", "ctx.db.get(key)")
-        ctx = self.require_runtime_context()
-        return await ctx.db.get(key)
-
-    async def delete_kv_data(self, key: str) -> None:
-        _warn_once("context.delete_kv_data()", "ctx.db.delete(key)")
-        ctx = self.require_runtime_context()
-        await ctx.db.delete(key)
-
-
-class CommandComponent:
-    pass
-
-
-Context = LegacyContext
+__all__ = [
+    "CommandComponent",
+    "Context",
+    "LegacyContext",
+    "LegacyConversationManager",
+]

src-new/astrbot_sdk/protocol/legacy_adapter.py

@@ -1,90 +1,579 @@
 from __future__ import annotations
 
-from typing import Any
+import json
+from typing import Any, Literal
 
-from .messages import CancelMessage, EventMessage, InvokeMessage, ResultMessage
+from pydantic import BaseModel, ConfigDict, Field
+
+from .descriptors import EventTrigger, HandlerDescriptor, Permissions
+from .messages import (
+    CancelMessage,
+    ErrorPayload,
+    EventMessage,
+    InitializeMessage,
+    InvokeMessage,
+    PeerInfo,
+    ProtocolMessage,
+    ResultMessage,
+)
+
+LEGACY_JSONRPC_VERSION = "2.0"
+LEGACY_CONTEXT_CAPABILITY = "internal.legacy.call_context_function"
+LEGACY_HANDSHAKE_METADATA_KEY = "legacy_handshake_payload"
+LEGACY_PLUGIN_KEYS_METADATA_KEY = "legacy_plugin_keys"
+LEGACY_ADAPTER_MESSAGE_EVENT = 3
+
+
+class _LegacyMessageBase(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class LegacyErrorData(_LegacyMessageBase):
+    code: int = -32000
+    message: str
+    data: Any | None = None
+
+
+class LegacyRequest(_LegacyMessageBase):
+    jsonrpc: Literal["2.0"] = LEGACY_JSONRPC_VERSION
+    id: str | None = None
+    method: str
+    params: dict[str, Any] = Field(default_factory=dict)
+
+
+class _LegacyResponse(_LegacyMessageBase):
+    jsonrpc: Literal["2.0"] = LEGACY_JSONRPC_VERSION
+    id: str | None = None
+
+
+class LegacySuccessResponse(_LegacyResponse):
+    result: Any = Field(default_factory=dict)
+
+
+class LegacyErrorResponse(_LegacyResponse):
+    error: LegacyErrorData
+
+
+LegacyMessage = LegacyRequest | LegacySuccessResponse | LegacyErrorResponse
+LegacyToV4Message = InitializeMessage | InvokeMessage | ResultMessage | EventMessage | CancelMessage
+
+
+class LegacyAdapter:
+    def __init__(
+        self,
+        *,
+        protocol_version: str = "1.0",
+        legacy_peer_name: str = "legacy-peer",
+        legacy_peer_role: Literal["plugin", "core"] = "plugin",
+        legacy_peer_version: str | None = None,
+    ) -> None:
+        self.protocol_version = protocol_version
+        self.legacy_peer_name = legacy_peer_name
+        self.legacy_peer_role = legacy_peer_role
+        self.legacy_peer_version = legacy_peer_version
+        self._handler_names_by_request_id: dict[str, str] = {}
+        self._pending_handshake_ids: set[str] = set()
+
+    def track_handler(self, request_id: str, handler_full_name: str) -> None:
+        if request_id:
+            self._handler_names_by_request_id[request_id] = handler_full_name
+
+    def legacy_to_v4(
+        self,
+        payload: str | bytes | dict[str, Any] | LegacyMessage,
+    ) -> LegacyToV4Message:
+        message = parse_legacy_message(payload)
+        if isinstance(message, LegacyRequest):
+            return self.legacy_request_to_message(message)
+        if isinstance(message, LegacySuccessResponse):
+            return self.legacy_response_to_message(message)
+        return self.legacy_error_to_result(message)
+
+    def legacy_request_to_message(
+        self,
+        payload: LegacyRequest | dict[str, Any],
+    ) -> InitializeMessage | InvokeMessage | EventMessage | CancelMessage:
+        message = payload if isinstance(payload, LegacyRequest) else LegacyRequest.model_validate(payload)
+        params = message.params or {}
+        method = message.method
+
+        if method == "handshake":
+            request_id = self._request_id(message.id, "legacy-handshake")
+            self._pending_handshake_ids.add(request_id)
+            return InitializeMessage(
+                id=request_id,
+                protocol_version=self.protocol_version,
+                peer=PeerInfo(
+                    name=self.legacy_peer_name,
+                    role=self.legacy_peer_role,
+                    version=self.legacy_peer_version,
+                ),
+                handlers=[],
+                metadata={"legacy_handshake": True},
+            )
+
+        if method == "call_handler":
+            request_id = self._request_id(message.id, "legacy-call-handler")
+            handler_full_name = str(params.get("handler_full_name", ""))
+            self.track_handler(request_id, handler_full_name)
+            return InvokeMessage(
+                id=request_id,
+                capability="handler.invoke",
+                input={
+                    "handler_id": handler_full_name,
+                    "event": self._as_dict(params.get("event"), field_name="data"),
+                    "args": self._as_dict(params.get("args"), field_name="value"),
+                },
+                stream=False,
+            )
+
+        if method == "call_context_function":
+            request_id = self._request_id(message.id, "legacy-context")
+            return InvokeMessage(
+                id=request_id,
+                capability=LEGACY_CONTEXT_CAPABILITY,
+                input={
+                    "name": str(params.get("name", "")),
+                    "args": self._as_dict(params.get("args"), field_name="value"),
+                },
+                stream=False,
+            )
+
+        if method == "handler_stream_start":
+            request_id = self._request_id(params.get("id"), "legacy-stream")
+            handler_full_name = str(params.get("handler_full_name", ""))
+            self.track_handler(request_id, handler_full_name)
+            return EventMessage(id=request_id, phase="started")
+
+        if method == "handler_stream_update":
+            request_id = self._request_id(params.get("id"), "legacy-stream")
+            handler_full_name = str(params.get("handler_full_name", ""))
+            self.track_handler(request_id, handler_full_name)
+            return EventMessage(
+                id=request_id,
+                phase="delta",
+                data=self._as_dict(params.get("data"), field_name="value"),
+            )
+
+        if method == "handler_stream_end":
+            request_id = self._request_id(params.get("id"), "legacy-stream")
+            handler_full_name = str(params.get("handler_full_name", ""))
+            self.track_handler(request_id, handler_full_name)
+            error = params.get("error")
+            if isinstance(error, dict):
+                return EventMessage(
+                    id=request_id,
+                    phase="failed",
+                    error=ErrorPayload.model_validate(self._coerce_error_payload(error)),
+                )
+            return EventMessage(id=request_id, phase="completed")
+
+        if method == "cancel":
+            return CancelMessage(
+                id=self._request_id(message.id, "legacy-cancel"),
+                reason=str(params.get("reason", "user_cancelled")),
+            )
+
+        return InvokeMessage(
+            id=self._request_id(message.id, "legacy-invoke"),
+            capability=method,
+            input=self._as_dict(params, field_name="data"),
+            stream=False,
+        )
+
+    def legacy_response_to_message(
+        self,
+        payload: LegacySuccessResponse | dict[str, Any],
+    ) -> InitializeMessage | ResultMessage:
+        message = payload if isinstance(payload, LegacySuccessResponse) else LegacySuccessResponse.model_validate(payload)
+        request_id = self._request_id(message.id, "legacy-result")
+
+        if request_id in self._pending_handshake_ids or self._looks_like_handshake_payload(message.result):
+            self._pending_handshake_ids.discard(request_id)
+            payload_dict = self._as_dict(message.result, field_name="data")
+            peer_name, peer_version = self._legacy_peer_from_handshake_payload(payload_dict)
+            return InitializeMessage(
+                id=request_id,
+                protocol_version=self.protocol_version,
+                peer=PeerInfo(
+                    name=peer_name,
+                    role=self.legacy_peer_role,
+                    version=peer_version,
+                ),
+                handlers=self._legacy_handlers_to_descriptors(payload_dict),
+                metadata={
+                    LEGACY_HANDSHAKE_METADATA_KEY: payload_dict,
+                    LEGACY_PLUGIN_KEYS_METADATA_KEY: sorted(payload_dict.keys()),
+                },
+            )
+
+        return ResultMessage(
+            id=request_id,
+            success=True,
+            output=self._as_dict(message.result, field_name="data"),
+        )
+
+    def legacy_error_to_result(
+        self,
+        payload: LegacyErrorResponse | dict[str, Any],
+    ) -> ResultMessage:
+        message = payload if isinstance(payload, LegacyErrorResponse) else LegacyErrorResponse.model_validate(payload)
+        request_id = self._request_id(message.id, "legacy-error")
+        kind = None
+        if request_id in self._pending_handshake_ids:
+            self._pending_handshake_ids.discard(request_id)
+            kind = "initialize_result"
+        return ResultMessage(
+            id=request_id,
+            kind=kind,
+            success=False,
+            error=ErrorPayload.model_validate(self._legacy_error_to_payload(message.error)),
+        )
+
+    def build_legacy_handshake_request(self, request_id: str) -> dict[str, Any]:
+        self._pending_handshake_ids.add(request_id)
+        return {
+            "jsonrpc": LEGACY_JSONRPC_VERSION,
+            "id": request_id,
+            "method": "handshake",
+            "params": {},
+        }
+
+    def initialize_to_legacy_handshake_response(
+        self,
+        message: InitializeMessage,
+        *,
+        request_id: str | None = None,
+    ) -> dict[str, Any]:
+        response_id = request_id or message.id
+        payload = self._legacy_handshake_payload_from_initialize(message)
+        return {
+            "jsonrpc": LEGACY_JSONRPC_VERSION,
+            "id": response_id,
+            "result": payload,
+        }
+
+    def invoke_to_legacy_request(self, message: InvokeMessage) -> dict[str, Any]:
+        if message.capability == "handler.invoke":
+            handler_full_name = str(message.input.get("handler_id", ""))
+            self.track_handler(message.id, handler_full_name)
+            return {
+                "jsonrpc": LEGACY_JSONRPC_VERSION,
+                "id": message.id,
+                "method": "call_handler",
+                "params": {
+                    "handler_full_name": handler_full_name,
+                    "event": self._as_dict(message.input.get("event"), field_name="data"),
+                    "args": self._as_dict(message.input.get("args"), field_name="value"),
+                },
+            }
+
+        if message.capability == LEGACY_CONTEXT_CAPABILITY:
+            return {
+                "jsonrpc": LEGACY_JSONRPC_VERSION,
+                "id": message.id,
+                "method": "call_context_function",
+                "params": {
+                    "name": str(message.input.get("name", "")),
+                    "args": self._as_dict(message.input.get("args"), field_name="value"),
+                },
+            }
+
+        return {
+            "jsonrpc": LEGACY_JSONRPC_VERSION,
+            "id": message.id,
+            "method": message.capability,
+            "params": self._as_dict(message.input, field_name="data"),
+        }
+
+    def result_to_legacy_response(self, message: ResultMessage) -> dict[str, Any]:
+        self._handler_names_by_request_id.pop(message.id, None)
+        if message.success:
+            return {
+                "jsonrpc": LEGACY_JSONRPC_VERSION,
+                "id": message.id,
+                "result": message.output,
+            }
+        return {
+            "jsonrpc": LEGACY_JSONRPC_VERSION,
+            "id": message.id,
+            "error": {
+                "code": -32000,
+                "message": message.error.message if message.error else "unknown error",
+                "data": message.error.model_dump() if message.error else None,
+            },
+        }
+
+    def event_to_legacy_notification(self, message: EventMessage) -> dict[str, Any]:
+        method = {
+            "started": "handler_stream_start",
+            "delta": "handler_stream_update",
+            "completed": "handler_stream_end",
+            "failed": "handler_stream_end",
+        }[message.phase]
+        params: dict[str, Any] = {
+            "id": message.id,
+            "handler_full_name": self._handler_names_by_request_id.get(message.id, ""),
+        }
+        if message.phase == "delta":
+            params["data"] = message.data
+        if message.phase == "failed" and message.error is not None:
+            params["error"] = message.error.model_dump()
+        return {
+            "jsonrpc": LEGACY_JSONRPC_VERSION,
+            "method": method,
+            "params": params,
+        }
+
+    def cancel_to_legacy_request(self, message: CancelMessage) -> dict[str, Any]:
+        return {
+            "jsonrpc": LEGACY_JSONRPC_VERSION,
+            "id": message.id,
+            "method": "cancel",
+            "params": {"reason": message.reason},
+        }
+
+    @staticmethod
+    def _request_id(value: Any, fallback: str) -> str:
+        text = "" if value is None else str(value)
+        return text or fallback
+
+    @staticmethod
+    def _as_dict(value: Any, *, field_name: str) -> dict[str, Any]:
+        if isinstance(value, dict):
+            return value
+        if value is None:
+            return {}
+        return {field_name: value}
+
+    @staticmethod
+    def _looks_like_handshake_payload(value: Any) -> bool:
+        if not isinstance(value, dict) or not value:
+            return False
+        return all(isinstance(item, dict) and "handlers" in item for item in value.values())
+
+    @staticmethod
+    def _coerce_error_payload(value: dict[str, Any]) -> dict[str, Any]:
+        if {"code", "message"}.issubset(value):
+            return {
+                "code": str(value.get("code", "legacy_rpc_error")),
+                "message": str(value.get("message", "legacy error")),
+                "hint": str(value.get("hint", "")),
+                "retryable": bool(value.get("retryable", False)),
+            }
+        return {
+            "code": "legacy_rpc_error",
+            "message": str(value.get("message", "legacy error")),
+            "hint": "",
+            "retryable": False,
+        }
+
+    @staticmethod
+    def _legacy_error_to_payload(error: LegacyErrorData) -> dict[str, Any]:
+        if isinstance(error.data, dict) and {"code", "message"}.issubset(error.data):
+            return LegacyAdapter._coerce_error_payload(error.data)
+        return {
+            "code": "legacy_rpc_error",
+            "message": error.message,
+            "hint": "",
+            "retryable": False,
+        }
+
+    def _legacy_handlers_to_descriptors(
+        self,
+        payload: dict[str, Any],
+    ) -> list[HandlerDescriptor]:
+        handlers: list[HandlerDescriptor] = []
+        for star_info in payload.values():
+            star_handlers = star_info.get("handlers") or []
+            if not isinstance(star_handlers, list):
+                continue
+            for handler_data in star_handlers:
+                if isinstance(handler_data, dict):
+                    handlers.append(self._legacy_handler_to_descriptor(handler_data))
+        return handlers
+
+    @staticmethod
+    def _legacy_handler_to_descriptor(handler_data: dict[str, Any]) -> HandlerDescriptor:
+        extras_configs = handler_data.get("extras_configs")
+        extras = extras_configs if isinstance(extras_configs, dict) else {}
+        handler_id = str(
+            handler_data.get("handler_full_name")
+            or f"{handler_data.get('handler_module_path', 'legacy')}.{handler_data.get('handler_name', 'handler')}"
+        )
+        event_type = handler_data.get("event_type", LEGACY_ADAPTER_MESSAGE_EVENT)
+        permissions = Permissions(
+            require_admin=bool(extras.get("require_admin", False)),
+            level=int(extras.get("level", 0) or 0),
+        )
+        return HandlerDescriptor(
+            id=handler_id,
+            trigger=EventTrigger(event_type=str(event_type)),
+            priority=int(extras.get("priority", 0) or 0),
+            permissions=permissions,
+        )
+
+    def _legacy_handshake_payload_from_initialize(
+        self,
+        message: InitializeMessage,
+    ) -> dict[str, Any]:
+        raw_payload = message.metadata.get(LEGACY_HANDSHAKE_METADATA_KEY)
+        if isinstance(raw_payload, dict) and raw_payload:
+            return raw_payload
+
+        plugin_name = str(message.metadata.get("plugin_id") or message.peer.name)
+        display_name = str(message.metadata.get("display_name") or plugin_name)
+        module_path = str(message.metadata.get("module_path") or f"{plugin_name}.main")
+        root_dir_name = str(message.metadata.get("root_dir_name") or plugin_name)
+        handlers = [self._descriptor_to_legacy_handler(item) for item in message.handlers]
+        return {
+            module_path: {
+                "name": plugin_name,
+                "author": message.metadata.get("author", "legacy-adapter"),
+                "desc": message.metadata.get("desc", ""),
+                "version": message.peer.version,
+                "repo": message.metadata.get("repo"),
+                "module_path": module_path,
+                "root_dir_name": root_dir_name,
+                "reserved": bool(message.metadata.get("reserved", False)),
+                "activated": bool(message.metadata.get("activated", True)),
+                "config": message.metadata.get("config"),
+                "star_handler_full_names": [item["handler_full_name"] for item in handlers],
+                "display_name": display_name,
+                "logo_path": message.metadata.get("logo_path"),
+                "handlers": handlers,
+            }
+        }
+
+    @staticmethod
+    def _descriptor_to_legacy_handler(descriptor: HandlerDescriptor) -> dict[str, Any]:
+        module_path, _, handler_name = descriptor.id.rpartition(".")
+        if not module_path:
+            module_path = descriptor.id
+            handler_name = descriptor.id
+        desc = getattr(descriptor.trigger, "description", None)
+        event_type: int | str = LEGACY_ADAPTER_MESSAGE_EVENT
+        if isinstance(descriptor.trigger, EventTrigger):
+            event_type = (
+                int(descriptor.trigger.event_type)
+                if descriptor.trigger.event_type.isdigit()
+                else descriptor.trigger.event_type
+            )
+        return {
+            "event_type": event_type,
+            "handler_full_name": descriptor.id,
+            "handler_name": handler_name,
+            "handler_module_path": module_path,
+            "desc": desc or "",
+            "extras_configs": {
+                "priority": descriptor.priority,
+                "require_admin": descriptor.permissions.require_admin,
+                "level": descriptor.permissions.level,
+            },
+        }
+
+    def _legacy_peer_from_handshake_payload(
+        self,
+        payload: dict[str, Any],
+    ) -> tuple[str, str | None]:
+        first_star = next(iter(payload.values()), {})
+        if not isinstance(first_star, dict):
+            return self.legacy_peer_name, self.legacy_peer_version
+        peer_name = str(first_star.get("name") or self.legacy_peer_name)
+        version_value = first_star.get("version")
+        peer_version = None if version_value is None else str(version_value)
+        return peer_name, peer_version or self.legacy_peer_version
+
+
+def parse_legacy_message(
+    payload: str | bytes | dict[str, Any] | LegacyMessage,
+) -> LegacyMessage:
+    if isinstance(payload, (LegacyRequest, LegacySuccessResponse, LegacyErrorResponse)):
+        return payload
+    if isinstance(payload, bytes):
+        payload = payload.decode("utf-8")
+    if isinstance(payload, str):
+        payload = json.loads(payload)
+    if "method" in payload:
+        return LegacyRequest.model_validate(payload)
+    if "result" in payload:
+        return LegacySuccessResponse.model_validate(payload)
+    if "error" in payload:
+        return LegacyErrorResponse.model_validate(payload)
+    raise ValueError("未知 legacy JSON-RPC 消息类型")
+
+
+def legacy_message_to_v4(
+    payload: str | bytes | dict[str, Any] | LegacyMessage,
+) -> LegacyToV4Message:
+    return LegacyAdapter().legacy_to_v4(payload)
 
 
 def legacy_request_to_invoke(payload: dict[str, Any]) -> InvokeMessage:
-    method = str(payload.get("method", ""))
-    params = payload.get("params") or {}
-    request_id = str(payload.get("id", ""))
-    if method == "call_handler":
-        return InvokeMessage(
-            id=request_id,
-            capability="handler.invoke",
-            input={
-                "handler_id": params.get("handler_full_name", ""),
-                "event": params.get("event", {}),
-                "args": params.get("args", {}),
-            },
-            stream=False,
-        )
-    return InvokeMessage(
-        id=request_id,
-        capability=method,
-        input=params if isinstance(params, dict) else {},
-        stream=False,
+    message = LegacyAdapter().legacy_request_to_message(payload)
+    if not isinstance(message, InvokeMessage):
+        raise ValueError("legacy request 不能直接映射为 invoke")
+    return message
+
+
+def legacy_response_to_message(
+    payload: dict[str, Any],
+) -> InitializeMessage | ResultMessage:
+    message = LegacyAdapter().legacy_response_to_message(payload)
+    return message
+
+
+def initialize_to_legacy_handshake_response(
+    message: InitializeMessage,
+    *,
+    request_id: str | None = None,
+) -> dict[str, Any]:
+    return LegacyAdapter().initialize_to_legacy_handshake_response(
+        message,
+        request_id=request_id,
     )
 
 
 def invoke_to_legacy_request(message: InvokeMessage) -> dict[str, Any]:
-    if message.capability == "handler.invoke":
-        return {
-            "jsonrpc": "2.0",
-            "id": message.id,
-            "method": "call_handler",
-            "params": {
-                "handler_full_name": message.input.get("handler_id"),
-                "event": message.input.get("event", {}),
-                "args": message.input.get("args", {}),
-            },
-        }
-    return {
-        "jsonrpc": "2.0",
-        "id": message.id,
-        "method": message.capability,
-        "params": message.input,
-    }
+    return LegacyAdapter().invoke_to_legacy_request(message)
 
 
 def result_to_legacy_response(message: ResultMessage) -> dict[str, Any]:
-    if message.success:
-        return {
-            "jsonrpc": "2.0",
-            "id": message.id,
-            "result": message.output,
-        }
-    return {
-        "jsonrpc": "2.0",
-        "id": message.id,
-        "error": {
-            "code": -32000,
-            "message": message.error.message if message.error else "unknown error",
-            "data": message.error.model_dump() if message.error else None,
-        },
-    }
+    return LegacyAdapter().result_to_legacy_response(message)
 
 
-def event_to_legacy_notification(message: EventMessage) -> dict[str, Any]:
-    method = {
-        "started": "handler_stream_start",
-        "delta": "handler_stream_update",
-        "completed": "handler_stream_end",
-        "failed": "handler_stream_end",
-    }[message.phase]
-    params: dict[str, Any] = {"id": message.id}
-    if message.phase == "delta":
-        params["data"] = message.data
-    if message.phase == "failed" and message.error is not None:
-        params["error"] = message.error.model_dump()
-    return {"jsonrpc": "2.0", "method": method, "params": params}
+def event_to_legacy_notification(
+    message: EventMessage,
+    *,
+    handler_full_name: str | None = None,
+) -> dict[str, Any]:
+    adapter = LegacyAdapter()
+    if handler_full_name:
+        adapter.track_handler(message.id, handler_full_name)
+    return adapter.event_to_legacy_notification(message)
 
 
 def cancel_to_legacy_request(message: CancelMessage) -> dict[str, Any]:
-    return {
-        "jsonrpc": "2.0",
-        "id": message.id,
-        "method": "cancel",
-        "params": {"reason": message.reason},
-    }
+    return LegacyAdapter().cancel_to_legacy_request(message)
+
+
+__all__ = [
+    "LEGACY_ADAPTER_MESSAGE_EVENT",
+    "LEGACY_CONTEXT_CAPABILITY",
+    "LEGACY_HANDSHAKE_METADATA_KEY",
+    "LEGACY_PLUGIN_KEYS_METADATA_KEY",
+    "LegacyAdapter",
+    "LegacyErrorData",
+    "LegacyErrorResponse",
+    "LegacyRequest",
+    "LegacySuccessResponse",
+    "cancel_to_legacy_request",
+    "event_to_legacy_notification",
+    "initialize_to_legacy_handshake_response",
+    "invoke_to_legacy_request",
+    "legacy_message_to_v4",
+    "legacy_request_to_invoke",
+    "legacy_response_to_message",
+    "parse_legacy_message",
+    "result_to_legacy_response",
+]

src-new/astrbot_sdk/runtime/bootstrap.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import os
 import signal
 import sys
@@ -9,7 +10,9 @@ from typing import IO, Any
 
 from loguru import logger
 
+from ..context import Context as RuntimeContext
 from ..errors import AstrBotError
+from ..protocol.descriptors import CapabilityDescriptor
 from ..protocol.messages import InitializeOutput, PeerInfo
 from .capability_router import CapabilityRouter
 from .handler_dispatcher import HandlerDispatcher
@@ -153,6 +156,7 @@ class WorkerSession:
             message.input,
             stream=message.stream,
             cancel_token=cancel_token,
+            request_id=message.id,
         )
 
 
@@ -180,6 +184,31 @@ class SupervisorRuntime:
         self.active_requests: dict[str, WorkerSession] = {}
         self.loaded_plugins: list[str] = []
         self.skipped_plugins: dict[str, str] = {}
+        self._register_internal_capabilities()
+
+    def _register_internal_capabilities(self) -> None:
+        self.capability_router.register(
+            CapabilityDescriptor(
+                name="handler.invoke",
+                description="框架内部：转发到插件 handler",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "handler_id": {"type": "string"},
+                        "event": {"type": "object"},
+                    },
+                    "required": ["handler_id", "event"],
+                },
+                output_schema={
+                    "type": "object",
+                    "properties": {},
+                    "required": [],
+                },
+                cancelable=True,
+            ),
+            call_handler=self._route_handler_invoke,
+            exposed=False,
+        )
 
     async def start(self) -> None:
         discovery = discover_plugins(self.plugins_dir)
@@ -224,22 +253,34 @@ class SupervisorRuntime:
             await session.stop()
         await self.peer.stop()
 
-    async def _handle_upstream_invoke(self, message, _cancel_token):
-        if message.capability != "handler.invoke":
-            raise AstrBotError.capability_not_found(message.capability)
-        handler_id = str(message.input.get("handler_id", ""))
+    async def _handle_upstream_invoke(self, message, cancel_token):
+        return await self.capability_router.execute(
+            message.capability,
+            message.input,
+            stream=message.stream,
+            cancel_token=cancel_token,
+            request_id=message.id,
+        )
+
+    async def _route_handler_invoke(
+        self,
+        request_id: str,
+        payload: dict[str, Any],
+        _cancel_token,
+    ) -> dict[str, Any]:
+        handler_id = str(payload.get("handler_id", ""))
         session = self.handler_to_worker.get(handler_id)
         if session is None:
             raise AstrBotError.invalid_input(f"handler not found: {handler_id}")
-        self.active_requests[message.id] = session
+        self.active_requests[request_id] = session
         try:
             return await session.invoke_handler(
                 handler_id,
-                message.input.get("event", {}),
-                request_id=message.id,
+                payload.get("event", {}),
+                request_id=request_id,
             )
         finally:
-            self.active_requests.pop(message.id, None)
+            self.active_requests.pop(request_id, None)
 
     async def _handle_upstream_cancel(self, request_id: str) -> None:
         session = self.active_requests.get(request_id)
@@ -261,6 +302,7 @@ class PluginWorkerRuntime:
             peer=self.peer,
             handlers=self.loaded_plugin.handlers,
         )
+        self._lifecycle_context = RuntimeContext(peer=self.peer, plugin_id=self.plugin.name)
         self.peer.set_invoke_handler(self._handle_invoke)
         self.peer.set_cancel_handler(self.dispatcher.cancel)
 
@@ -270,15 +312,44 @@ class PluginWorkerRuntime:
             [item.descriptor for item in self.loaded_plugin.handlers],
             metadata={"plugin_id": self.plugin.name},
         )
+        await self._run_lifecycle("on_start")
 
     async def stop(self) -> None:
-        await self.peer.stop()
+        try:
+            await self._run_lifecycle("on_stop")
+        finally:
+            await self.peer.stop()
 
     async def _handle_invoke(self, message, cancel_token):
         if message.capability != "handler.invoke":
             raise AstrBotError.capability_not_found(message.capability)
         return await self.dispatcher.invoke(message, cancel_token)
 
+    async def _run_lifecycle(self, method_name: str) -> None:
+        for instance in self.loaded_plugin.instances:
+            hook = getattr(instance, method_name, None)
+            if hook is None or not callable(hook):
+                continue
+            args = []
+            try:
+                signature = inspect.signature(hook)
+            except (TypeError, ValueError):
+                signature = None
+            if signature is not None:
+                positional_params = [
+                    parameter
+                    for parameter in signature.parameters.values()
+                    if parameter.kind in (
+                        inspect.Parameter.POSITIONAL_ONLY,
+                        inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                    )
+                ]
+                if positional_params:
+                    args.append(self._lifecycle_context)
+            result = hook(*args)
+            if inspect.isawaitable(result):
+                await result
+
 
 async def run_supervisor(
     *,

src-new/astrbot_sdk/runtime/capability_router.py

@@ -9,8 +9,8 @@ from typing import Any
 from ..errors import AstrBotError
 from ..protocol.descriptors import CapabilityDescriptor
 
-CallHandler = Callable[[dict[str, Any], object], Awaitable[dict[str, Any]]]
-StreamHandler = Callable[[dict[str, Any], object], AsyncIterator[dict[str, Any]]]
+CallHandler = Callable[[str, dict[str, Any], object], Awaitable[dict[str, Any]]]
+StreamHandler = Callable[[str, dict[str, Any], object], AsyncIterator[dict[str, Any]]]
 FinalizeHandler = Callable[[list[dict[str, Any]]], dict[str, Any]]
 
 
@@ -26,6 +26,7 @@ class _CapabilityRegistration:
     call_handler: CallHandler | None = None
     stream_handler: StreamHandler | None = None
     finalize: FinalizeHandler | None = None
+    exposed: bool = True
 
 
 class CapabilityRouter:
@@ -37,7 +38,11 @@ class CapabilityRouter:
         self._register_builtin_capabilities()
 
     def descriptors(self) -> list[CapabilityDescriptor]:
-        return [entry.descriptor for entry in self._registrations.values()]
+        return [
+            entry.descriptor
+            for entry in self._registrations.values()
+            if entry.exposed
+        ]
 
     def register(
         self,
@@ -46,12 +51,14 @@ class CapabilityRouter:
         call_handler: CallHandler | None = None,
         stream_handler: StreamHandler | None = None,
         finalize: FinalizeHandler | None = None,
+        exposed: bool = True,
     ) -> None:
         self._registrations[descriptor.name] = _CapabilityRegistration(
             descriptor=descriptor,
             call_handler=call_handler,
             stream_handler=stream_handler,
             finalize=finalize,
+            exposed=exposed,
         )
 
     async def execute(
@@ -61,6 +68,7 @@ class CapabilityRouter:
         *,
         stream: bool,
         cancel_token,
+        request_id: str,
     ) -> dict[str, Any] | StreamExecution:
         registration = self._registrations.get(capability)
         if registration is None:
@@ -72,13 +80,13 @@ class CapabilityRouter:
                 raise AstrBotError.invalid_input(f"{capability} 不支持 stream=true")
             finalize = registration.finalize or (lambda chunks: {"items": chunks})
             return StreamExecution(
-                iterator=registration.stream_handler(payload, cancel_token),
+                iterator=registration.stream_handler(request_id, payload, cancel_token),
                 finalize=finalize,
             )
 
         if registration.call_handler is None:
             raise AstrBotError.invalid_input(f"{capability} 只能以 stream=true 调用")
-        output = await registration.call_handler(payload, cancel_token)
+        output = await registration.call_handler(request_id, payload, cancel_token)
         self._validate_schema(registration.descriptor.output_schema, output)
         return output
 
@@ -90,11 +98,11 @@ class CapabilityRouter:
                 "required": required,
             }
 
-        async def llm_chat(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def llm_chat(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             prompt = str(payload.get("prompt", ""))
             return {"text": f"Echo: {prompt}"}
 
-        async def llm_chat_raw(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def llm_chat_raw(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             prompt = str(payload.get("prompt", ""))
             text = f"Echo: {prompt}"
             return {
@@ -108,6 +116,7 @@ class CapabilityRouter:
             }
 
         async def llm_stream(
+            _request_id: str,
             payload: dict[str, Any],
             token,
         ) -> AsyncIterator[dict[str, Any]]:
@@ -117,7 +126,7 @@ class CapabilityRouter:
                 await asyncio.sleep(0)
                 yield {"text": char}
 
-        async def memory_search(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def memory_search(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             query = str(payload.get("query", ""))
             items = [
                 {"key": key, "value": value}
@@ -126,7 +135,7 @@ class CapabilityRouter:
             ]
             return {"items": items}
 
-        async def memory_save(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def memory_save(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             key = str(payload.get("key", ""))
             value = payload.get("value")
             if not isinstance(value, dict):
@@ -134,14 +143,14 @@ class CapabilityRouter:
             self.memory_store[key] = value
             return {}
 
-        async def memory_delete(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def memory_delete(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             self.memory_store.pop(str(payload.get("key", "")), None)
             return {}
 
-        async def db_get(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def db_get(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             return {"value": self.db_store.get(str(payload.get("key", "")))}
 
-        async def db_set(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def db_set(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             key = str(payload.get("key", ""))
             value = payload.get("value")
             if not isinstance(value, dict):
@@ -149,18 +158,18 @@ class CapabilityRouter:
             self.db_store[key] = value
             return {}
 
-        async def db_delete(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def db_delete(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             self.db_store.pop(str(payload.get("key", "")), None)
             return {}
 
-        async def db_list(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def db_list(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             prefix = payload.get("prefix")
             keys = sorted(self.db_store.keys())
             if isinstance(prefix, str):
                 keys = [item for item in keys if item.startswith(prefix)]
             return {"keys": keys}
 
-        async def platform_send(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def platform_send(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             session = str(payload.get("session", ""))
             text = str(payload.get("text", ""))
             message_id = f"msg_{len(self.sent_messages) + 1}"
@@ -173,7 +182,7 @@ class CapabilityRouter:
             )
             return {"message_id": message_id}
 
-        async def platform_send_image(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def platform_send_image(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             session = str(payload.get("session", ""))
             image_url = str(payload.get("image_url", ""))
             message_id = f"img_{len(self.sent_messages) + 1}"
@@ -186,7 +195,7 @@ class CapabilityRouter:
             )
             return {"message_id": message_id}
 
-        async def platform_get_members(payload: dict[str, Any], _token) -> dict[str, Any]:
+        async def platform_get_members(_request_id: str, payload: dict[str, Any], _token) -> dict[str, Any]:
             session = str(payload.get("session", ""))
             return {
                 "members": [

src-new/astrbot_sdk/runtime/loader.py

@@ -13,7 +13,6 @@ from typing import Any
 
 import yaml
 
-from ..compat import LegacyContext
 from ..decorators import get_handler_meta
 from ..protocol.descriptors import HandlerDescriptor
 from ..star import Star
@@ -48,7 +47,7 @@ class LoadedHandler:
     descriptor: HandlerDescriptor
     callable: Any
     owner: Any
-    legacy_context: LegacyContext | None = None
+    legacy_context: Any | None = None
 
 
 @dataclass(slots=True)
@@ -58,6 +57,15 @@ class LoadedPlugin:
     instances: list[Any]
 
 
+def _create_legacy_context(component_cls: Any, plugin_name: str) -> Any:
+    factory = getattr(component_cls, "_astrbot_create_legacy_context", None)
+    if callable(factory):
+        return factory(plugin_name)
+    from ..api.star.context import Context as LegacyContext
+
+    return LegacyContext(plugin_name)
+
+
 def load_plugin_spec(plugin_dir: Path) -> PluginSpec:
     plugin_dir = plugin_dir.resolve()
     manifest_path = plugin_dir / "plugin.yaml"
@@ -264,11 +272,11 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin:
         if not isinstance(class_path, str) or ":" not in class_path:
             continue
         component_cls = import_string(class_path)
-        legacy_context: LegacyContext | None = None
+        legacy_context = None
         if isinstance(component_cls, type) and issubclass(component_cls, Star):
             instance = component_cls()
         else:
-            legacy_context = LegacyContext(plugin.name)
+            legacy_context = _create_legacy_context(component_cls, plugin.name)
             try:
                 instance = component_cls(legacy_context)
             except TypeError:

src-new/astrbot_sdk/runtime/transport.py

@@ -161,9 +161,11 @@ class WebSocketServerTransport(Transport):
         self._site: web.TCPSite | None = None
         self._ws: web.WebSocketResponse | None = None
         self._write_lock = asyncio.Lock()
+        self._connected = asyncio.Event()
 
     async def start(self) -> None:
         self._closed.clear()
+        self._connected.clear()
         self._app = web.Application()
         self._app.router.add_get(self._path, self._handle_socket)
         self._runner = web.AppRunner(self._app)
@@ -175,6 +177,7 @@ class WebSocketServerTransport(Transport):
             self._actual_port = socket.getsockname()[1]
 
     async def stop(self) -> None:
+        self._connected.clear()
         if self._ws is not None and not self._ws.closed:
             await self._ws.close()
         if self._site is not None:
@@ -186,6 +189,8 @@ class WebSocketServerTransport(Transport):
         self._closed.set()
 
     async def send(self, payload: str) -> None:
+        if self._ws is None or self._ws.closed:
+            await asyncio.wait_for(self._connected.wait(), timeout=30.0)
         if self._ws is None or self._ws.closed:
             raise RuntimeError("WebSocket 尚未连接")
         async with self._write_lock:
@@ -203,6 +208,7 @@ class WebSocketServerTransport(Transport):
         )
         await ws.prepare(request)
         self._ws = ws
+        self._connected.set()
         try:
             async for msg in ws:
                 if msg.type == aiohttp.WSMsgType.TEXT:
@@ -213,6 +219,7 @@ class WebSocketServerTransport(Transport):
                     logger.error("websocket server error: {}", ws.exception())
                     break
         finally:
+            self._connected.clear()
             self._closed.set()
             self._ws = None
         return ws

src-new/astrbot_sdk/star.py

@@ -9,6 +9,12 @@ from .errors import AstrBotError
 
 
 class Star:
+    async def on_start(self, ctx: Any | None = None) -> None:
+        return None
+
+    async def on_stop(self, ctx: Any | None = None) -> None:
+        return None
+
     async def on_error(self, error: Exception, event, ctx) -> None:
         if isinstance(error, AstrBotError):
             if error.retryable:

tests_v4/test_api_contract.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from astrbot_sdk._legacy_api import MIGRATION_DOC_URL, LegacyContext, _warned_methods
+from astrbot_sdk.clients._proxy import CapabilityProxy
+from astrbot_sdk.clients.memory import MemoryClient
+from astrbot_sdk.star import Star
+
+
+class _DummyPeer:
+    def __init__(self) -> None:
+        self.remote_capability_map = {}
+        self.calls: list[tuple[str, dict]] = []
+
+    async def invoke(self, name: str, payload: dict, *, stream: bool = False) -> dict:
+        self.calls.append((name, payload))
+        return {}
+
+
+class ApiContractTest(unittest.IsolatedAsyncioTestCase):
+    async def test_star_lifecycle_hooks_exist(self) -> None:
+        star = Star()
+        self.assertIsNone(await star.on_start())
+        self.assertIsNone(await star.on_stop())
+
+    async def test_memory_client_save_accepts_expanded_keyword_payload(self) -> None:
+        peer = _DummyPeer()
+        client = MemoryClient(CapabilityProxy(peer))
+
+        await client.save("memory-key", foo="bar", score=3)
+
+        self.assertEqual(
+            peer.calls,
+            [
+                (
+                    "memory.save",
+                    {
+                        "key": "memory-key",
+                        "value": {"foo": "bar", "score": 3},
+                    },
+                )
+            ],
+        )
+
+    async def test_compat_warning_includes_migration_doc_url(self) -> None:
+        _warned_methods.clear()
+        legacy_context = LegacyContext("compat-plugin")
+        with patch("astrbot_sdk._legacy_api.logger.warning") as warning:
+            await legacy_context.add_llm_tools()
+
+        warning.assert_called_once()
+        self.assertEqual(warning.call_args.args[-1], MIGRATION_DOC_URL)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests_v4/test_legacy_adapter.py

@@ -0,0 +1,239 @@
+from __future__ import annotations
+
+import unittest
+
+from astrbot_sdk.protocol.descriptors import CommandTrigger, HandlerDescriptor
+from astrbot_sdk.protocol.legacy_adapter import (
+    LEGACY_CONTEXT_CAPABILITY,
+    LEGACY_HANDSHAKE_METADATA_KEY,
+    LegacyAdapter,
+)
+from astrbot_sdk.protocol.messages import EventMessage, InitializeMessage, PeerInfo, ResultMessage
+
+
+class LegacyAdapterTest(unittest.TestCase):
+    def test_call_handler_roundtrip_preserves_handler_name_in_stream_notifications(self) -> None:
+        adapter = LegacyAdapter()
+
+        invoke = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "id": "call-1",
+                "method": "call_handler",
+                "params": {
+                    "handler_full_name": "commands.demo:MyPlugin.handle",
+                    "event": {"text": "/hello"},
+                    "args": {},
+                },
+            }
+        )
+
+        self.assertEqual(invoke.capability, "handler.invoke")
+        self.assertEqual(invoke.input["handler_id"], "commands.demo:MyPlugin.handle")
+
+        started = adapter.event_to_legacy_notification(
+            EventMessage(id="call-1", phase="started")
+        )
+        delta = adapter.event_to_legacy_notification(
+            EventMessage(id="call-1", phase="delta", data={"text": "hi"})
+        )
+        completed = adapter.event_to_legacy_notification(
+            EventMessage(id="call-1", phase="completed", output={"text": "hi"})
+        )
+        response = adapter.result_to_legacy_response(
+            ResultMessage(id="call-1", success=True, output={"handled_by": "demo"})
+        )
+
+        self.assertEqual(started["method"], "handler_stream_start")
+        self.assertEqual(delta["params"]["handler_full_name"], "commands.demo:MyPlugin.handle")
+        self.assertEqual(delta["params"]["data"], {"text": "hi"})
+        self.assertEqual(completed["method"], "handler_stream_end")
+        self.assertEqual(response["result"], {"handled_by": "demo"})
+
+    def test_call_context_function_maps_to_internal_capability(self) -> None:
+        adapter = LegacyAdapter()
+        message = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "id": "ctx-1",
+                "method": "call_context_function",
+                "params": {
+                    "name": "ConversationManager.new_conversation",
+                    "args": {"unified_msg_origin": "session-1"},
+                },
+            }
+        )
+
+        self.assertEqual(message.capability, LEGACY_CONTEXT_CAPABILITY)
+        self.assertEqual(message.input["name"], "ConversationManager.new_conversation")
+        self.assertEqual(
+            message.input["args"],
+            {"unified_msg_origin": "session-1"},
+        )
+
+        legacy_request = adapter.invoke_to_legacy_request(message)
+        self.assertEqual(legacy_request["method"], "call_context_function")
+        self.assertEqual(
+            legacy_request["params"]["name"],
+            "ConversationManager.new_conversation",
+        )
+
+    def test_legacy_handler_stream_notifications_map_back_to_v4_events(self) -> None:
+        adapter = LegacyAdapter()
+
+        started = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "method": "handler_stream_start",
+                "params": {
+                    "id": "call-2",
+                    "handler_full_name": "commands.demo:MyPlugin.handle",
+                },
+            }
+        )
+        delta = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "method": "handler_stream_update",
+                "params": {
+                    "id": "call-2",
+                    "handler_full_name": "commands.demo:MyPlugin.handle",
+                    "data": {"text": "partial"},
+                },
+            }
+        )
+        failed = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "method": "handler_stream_end",
+                "params": {
+                    "id": "call-2",
+                    "handler_full_name": "commands.demo:MyPlugin.handle",
+                    "error": {
+                        "code": "cancelled",
+                        "message": "调用被取消",
+                        "hint": "",
+                        "retryable": False,
+                    },
+                },
+            }
+        )
+
+        self.assertEqual(started.phase, "started")
+        self.assertEqual(delta.phase, "delta")
+        self.assertEqual(delta.data, {"text": "partial"})
+        self.assertEqual(failed.phase, "failed")
+        self.assertEqual(failed.error.code, "cancelled")
+
+    def test_handshake_payload_maps_to_initialize_and_roundtrips(self) -> None:
+        adapter = LegacyAdapter(legacy_peer_name="legacy-plugin")
+        legacy_payload = {
+            "plugin_one.main": {
+                "name": "plugin_one",
+                "author": "tester",
+                "desc": "legacy",
+                "version": "0.1.0",
+                "repo": None,
+                "module_path": "plugin_one.main",
+                "root_dir_name": "plugin_one",
+                "reserved": False,
+                "activated": True,
+                "config": None,
+                "star_handler_full_names": ["commands.plugin_one:SampleCommand.handle_plugin_one"],
+                "display_name": "plugin_one",
+                "logo_path": None,
+                "handlers": [
+                    {
+                        "event_type": 3,
+                        "handler_full_name": "commands.plugin_one:SampleCommand.handle_plugin_one",
+                        "handler_name": "handle_plugin_one",
+                        "handler_module_path": "commands.plugin_one:SampleCommand",
+                        "desc": "",
+                        "extras_configs": {"priority": 7},
+                    }
+                ],
+            }
+        }
+
+        initialize = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "id": "handshake-1",
+                "result": legacy_payload,
+            }
+        )
+
+        self.assertIsInstance(initialize, InitializeMessage)
+        self.assertEqual(initialize.peer.name, "plugin_one")
+        self.assertEqual(initialize.handlers[0].id, "commands.plugin_one:SampleCommand.handle_plugin_one")
+        self.assertEqual(initialize.handlers[0].priority, 7)
+        self.assertEqual(
+            initialize.metadata[LEGACY_HANDSHAKE_METADATA_KEY],
+            legacy_payload,
+        )
+
+        roundtrip = adapter.initialize_to_legacy_handshake_response(
+            initialize,
+            request_id="handshake-1",
+        )
+        self.assertEqual(roundtrip["result"], legacy_payload)
+
+    def test_handshake_error_becomes_initialize_failure(self) -> None:
+        adapter = LegacyAdapter()
+        adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "id": "handshake-2",
+                "method": "handshake",
+                "params": {},
+            }
+        )
+
+        result = adapter.legacy_to_v4(
+            {
+                "jsonrpc": "2.0",
+                "id": "handshake-2",
+                "error": {
+                    "code": -32000,
+                    "message": "boom",
+                },
+            }
+        )
+
+        self.assertIsInstance(result, ResultMessage)
+        self.assertEqual(result.kind, "initialize_result")
+        self.assertFalse(result.success)
+        self.assertEqual(result.error.code, "legacy_rpc_error")
+
+    def test_initialize_can_synthesize_legacy_handshake_payload(self) -> None:
+        adapter = LegacyAdapter()
+        initialize = InitializeMessage(
+            id="msg_001",
+            protocol_version="1.0",
+            peer=PeerInfo(name="v4-plugin", role="plugin", version="1.2.0"),
+            handlers=[
+                HandlerDescriptor(
+                    id="commands.sample:MyPlugin.hello",
+                    trigger=CommandTrigger(command="hello", description="hello"),
+                    priority=3,
+                )
+            ],
+            metadata={"plugin_id": "v4-plugin"},
+        )
+
+        payload = adapter.initialize_to_legacy_handshake_response(
+            initialize,
+            request_id="handshake-3",
+        )
+        star_payload = payload["result"]["v4-plugin.main"]
+
+        self.assertEqual(star_payload["name"], "v4-plugin")
+        self.assertEqual(
+            star_payload["handlers"][0]["handler_full_name"],
+            "commands.sample:MyPlugin.hello",
+        )
+        self.assertEqual(star_payload["handlers"][0]["extras_configs"]["priority"], 3)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests_v4/test_peer.py

@@ -37,6 +37,7 @@ class PeerRuntimeTest(unittest.IsolatedAsyncioTestCase):
                 message.input,
                 stream=message.stream,
                 cancel_token=token,
+                request_id=message.id,
             )
         )
 
@@ -172,6 +173,7 @@ class PeerRuntimeTest(unittest.IsolatedAsyncioTestCase):
                 message.input,
                 stream=message.stream,
                 cancel_token=token,
+                request_id=message.id,
             )
         )
 

tests_v4/test_script_migrations.py

@@ -0,0 +1,274 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import sys
+import tempfile
+import textwrap
+import time
+import unittest
+from pathlib import Path
+
+from astrbot_sdk.protocol.messages import InitializeOutput, PeerInfo
+from astrbot_sdk.runtime.bootstrap import PluginWorkerRuntime, SupervisorRuntime
+from astrbot_sdk.runtime.capability_router import CapabilityRouter
+from astrbot_sdk.runtime.peer import Peer
+from astrbot_sdk.runtime.transport import WebSocketClientTransport, WebSocketServerTransport
+
+from tests_v4.helpers import FakeEnvManager, make_transport_pair
+
+
+def write_websocket_plugin(plugin_root: Path) -> None:
+    (plugin_root / "commands").mkdir(parents=True, exist_ok=True)
+    (plugin_root / "commands" / "__init__.py").write_text("", encoding="utf-8")
+    (plugin_root / "requirements.txt").write_text("", encoding="utf-8")
+    (plugin_root / "plugin.yaml").write_text(
+        textwrap.dedent(
+            f"""\
+            _schema_version: 2
+            name: websocket_plugin
+            display_name: WebSocket Plugin
+            desc: websocket test
+            author: tester
+            version: 0.1.0
+            runtime:
+              python: "{sys.version_info.major}.{sys.version_info.minor}"
+            components:
+              - class: commands.sample:MyPlugin
+                type: command
+                name: hello
+                description: hello
+            """
+        ),
+        encoding="utf-8",
+    )
+    (plugin_root / "commands" / "sample.py").write_text(
+        textwrap.dedent(
+            """\
+            from astrbot_sdk import Context, MessageEvent, Star, on_command
+
+
+            class MyPlugin(Star):
+                @on_command("hello")
+                async def hello(self, event: MessageEvent, ctx: Context):
+                    await event.reply(f"ws:{event.text}")
+            """
+        ),
+        encoding="utf-8",
+    )
+
+
+def write_benchmark_plugin(plugins_dir: Path, index: int) -> None:
+    plugin_name = f"plugin_{index:03d}"
+    command_name = f"bench_{index:03d}"
+    plugin_dir = plugins_dir / plugin_name
+    commands_dir = plugin_dir / "commands"
+    commands_dir.mkdir(parents=True, exist_ok=True)
+    (commands_dir / "__init__.py").write_text("", encoding="utf-8")
+    (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+    (plugin_dir / "plugin.yaml").write_text(
+        textwrap.dedent(
+            f"""\
+            _schema_version: 2
+            name: {plugin_name}
+            display_name: {plugin_name}
+            desc: benchmark plugin {index}
+            author: tester
+            version: 0.1.0
+            runtime:
+              python: "{sys.version_info.major}.{sys.version_info.minor}"
+            components:
+              - class: commands.plugin_{index:03d}:BenchmarkCommand{index:03d}
+                type: command
+                name: {command_name}
+                description: {command_name}
+            """
+        ),
+        encoding="utf-8",
+    )
+    (commands_dir / f"plugin_{index:03d}.py").write_text(
+        textwrap.dedent(
+            f"""\
+            from astrbot_sdk.api.components.command import CommandComponent
+            from astrbot_sdk.api.event import AstrMessageEvent, filter
+            from astrbot_sdk.api.star.context import Context
+
+
+            class BenchmarkCommand{index:03d}(CommandComponent):
+                def __init__(self, context: Context):
+                    self.context = context
+
+                @filter.command("{command_name}")
+                async def handle(self, event: AstrMessageEvent):
+                    yield event.plain_result("{plugin_name}:{command_name}")
+            """
+        ),
+        encoding="utf-8",
+    )
+
+
+class StartClientMigrationTest(unittest.IsolatedAsyncioTestCase):
+    async def test_websocket_plugin_worker_supports_handshake_and_handler_invoke(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_root = Path(temp_dir) / "websocket_plugin"
+            write_websocket_plugin(plugin_root)
+
+            server_transport = WebSocketServerTransport(host="127.0.0.1", port=0, path="/ws")
+            runtime = PluginWorkerRuntime(plugin_dir=plugin_root, transport=server_transport)
+            runtime_task = asyncio.create_task(runtime.start())
+
+            try:
+                for _ in range(100):
+                    if server_transport.port != 0:
+                        break
+                    await asyncio.sleep(0.02)
+
+                core_router = CapabilityRouter()
+                core_peer = Peer(
+                    transport=WebSocketClientTransport(url=server_transport.url),
+                    peer_info=PeerInfo(name="websocket-core", role="core", version="v4"),
+                )
+                core_peer.set_initialize_handler(
+                    lambda _message: asyncio.sleep(
+                        0,
+                        result=InitializeOutput(
+                            peer=PeerInfo(name="websocket-core", role="core", version="v4"),
+                            capabilities=core_router.descriptors(),
+                            metadata={},
+                        ),
+                    )
+                )
+                core_peer.set_invoke_handler(
+                    lambda message, cancel_token: core_router.execute(
+                        message.capability,
+                        message.input,
+                        stream=message.stream,
+                        cancel_token=cancel_token,
+                        request_id=message.id,
+                    )
+                )
+                await core_peer.start()
+                try:
+                    await asyncio.wait_for(runtime_task, timeout=5)
+                    await core_peer.wait_until_remote_initialized()
+                    handler_id = core_peer.remote_handlers[0].id
+                    self.assertEqual(core_peer.remote_metadata["plugin_id"], "websocket_plugin")
+                    self.assertEqual(core_peer.remote_handlers[0].trigger.command, "hello")
+
+                    await core_peer.invoke(
+                        "handler.invoke",
+                        {
+                            "handler_id": handler_id,
+                            "event": {
+                                "text": "hello-websocket",
+                                "session_id": "session-ws",
+                                "user_id": "user-1",
+                                "platform": "test",
+                            },
+                        },
+                        request_id="call-ws",
+                    )
+
+                    self.assertEqual(
+                        [item.get("text") for item in core_router.sent_messages],
+                        ["ws:hello-websocket"],
+                    )
+                finally:
+                    await core_peer.stop()
+            finally:
+                if not runtime_task.done():
+                    runtime_task.cancel()
+                    with contextlib.suppress(asyncio.CancelledError):
+                        await runtime_task
+                await runtime.stop()
+
+
+class BenchmarkMigrationTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self.left, self.right = make_transport_pair()
+        self.core = Peer(
+            transport=self.left,
+            peer_info=PeerInfo(name="benchmark-core", role="core", version="v4"),
+        )
+        self.core.set_initialize_handler(
+            lambda _message: asyncio.sleep(
+                0,
+                result=InitializeOutput(
+                    peer=PeerInfo(name="benchmark-core", role="core", version="v4"),
+                    capabilities=[],
+                    metadata={},
+                ),
+            )
+        )
+        await self.core.start()
+
+    async def asyncTearDown(self) -> None:
+        await self.core.stop()
+
+    async def test_benchmark_style_runtime_report_covers_multi_plugin_workers(self) -> None:
+        plugin_count = 8
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir) / "plugins"
+            for index in range(plugin_count):
+                write_benchmark_plugin(plugins_dir, index)
+
+            runtime = SupervisorRuntime(
+                transport=self.right,
+                plugins_dir=plugins_dir,
+                env_manager=FakeEnvManager(),
+            )
+            started_at = time.perf_counter()
+            try:
+                await runtime.start()
+                await self.core.wait_until_remote_initialized()
+                measured_at = time.perf_counter()
+
+                worker_pids = sorted(
+                    process.pid
+                    for session in runtime.worker_sessions.values()
+                    if (process := getattr(getattr(session.peer, "transport", None), "_process", None)) is not None
+                    and process.returncode is None
+                )
+                report = {
+                    "plugin_count": plugin_count,
+                    "loaded_plugin_count": len(runtime.loaded_plugins),
+                    "loaded_plugins": sorted(runtime.loaded_plugins),
+                    "aggregated_handler_ids": list(self.core.remote_metadata["aggregated_handler_ids"]),
+                    "startup_total_duration_ms": round((measured_at - started_at) * 1000, 2),
+                    "worker_pids": worker_pids,
+                }
+
+                handler_id = next(
+                    item.id
+                    for item in self.core.remote_handlers
+                    if item.id.startswith("plugin_002:")
+                )
+                await self.core.invoke(
+                    "handler.invoke",
+                    {
+                        "handler_id": handler_id,
+                        "event": {
+                            "text": "/bench_002",
+                            "session_id": "bench-session",
+                            "user_id": "user-1",
+                            "platform": "test",
+                        },
+                    },
+                    request_id="bench-call",
+                )
+
+                self.assertEqual(report["loaded_plugin_count"], plugin_count)
+                self.assertEqual(len(report["worker_pids"]), plugin_count)
+                self.assertEqual(len(report["aggregated_handler_ids"]), plugin_count)
+                self.assertEqual(
+                    report["loaded_plugins"],
+                    [f"plugin_{index:03d}" for index in range(plugin_count)],
+                )
+                self.assertGreaterEqual(report["startup_total_duration_ms"], 0)
+                self.assertIn("plugin_002:bench_002", runtime.capability_router.sent_messages[-1]["text"])
+            finally:
+                await runtime.stop()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests_v4/test_supervisor_migration.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+import tempfile
+import textwrap
+import unittest
+from pathlib import Path
+
+from astrbot_sdk.protocol.messages import InitializeOutput, PeerInfo
+from astrbot_sdk.runtime.bootstrap import SupervisorRuntime
+from astrbot_sdk.runtime.loader import discover_plugins
+from astrbot_sdk.runtime.peer import Peer
+
+from tests_v4.helpers import FakeEnvManager, make_transport_pair
+
+
+def write_plugin(
+    root: Path,
+    folder_name: str,
+    *,
+    plugin_name: str | None = None,
+    python_version: str | None = None,
+    include_requirements: bool = True,
+    reply_text: str | None = None,
+) -> Path:
+    plugin_dir = root / folder_name
+    commands_dir = plugin_dir / "commands"
+    commands_dir.mkdir(parents=True, exist_ok=True)
+    (commands_dir / "__init__.py").write_text("", encoding="utf-8")
+
+    if python_version is None:
+        python_version = f"{sys.version_info.major}.{sys.version_info.minor}"
+
+    manifest_lines = [
+        "_schema_version: 2",
+        f"name: {plugin_name or folder_name}",
+        f"display_name: {folder_name}",
+        "desc: test plugin",
+        "author: tester",
+        "version: 0.1.0",
+    ]
+    if python_version != "__missing__":
+        manifest_lines.extend(
+            [
+                "runtime:",
+                f"  python: \"{python_version}\"",
+            ]
+        )
+    manifest_lines.extend(
+        [
+            "components:",
+            "  - class: commands.sample:SamplePlugin",
+            "    type: command",
+            "    name: hello",
+            "    description: hello",
+        ]
+    )
+    (plugin_dir / "plugin.yaml").write_text("\n".join(manifest_lines) + "\n", encoding="utf-8")
+    if include_requirements:
+        (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+    text = reply_text or f"{plugin_name or folder_name} handled"
+    (commands_dir / "sample.py").write_text(
+        textwrap.dedent(
+            f"""\
+            from astrbot_sdk import Context, MessageEvent, Star, on_command
+
+
+            class SamplePlugin(Star):
+                @on_command("hello")
+                async def hello(self, event: MessageEvent, ctx: Context):
+                    await event.reply({text!r})
+            """
+        ),
+        encoding="utf-8",
+    )
+    return plugin_dir
+
+
+class PluginDiscoveryMigrationTest(unittest.TestCase):
+    def test_discover_plugins_keeps_old_supervisor_filtering_rules(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            write_plugin(root, "plugin_one", plugin_name="plugin_one")
+            write_plugin(
+                root,
+                "plugin_two",
+                plugin_name="plugin_two",
+                python_version="__missing__",
+            )
+            write_plugin(
+                root,
+                "plugin_three",
+                plugin_name="plugin_three",
+                include_requirements=False,
+            )
+
+            discovery = discover_plugins(root)
+
+        self.assertEqual([plugin.name for plugin in discovery.plugins], ["plugin_one"])
+        self.assertIn("plugin_two", discovery.skipped_plugins)
+        self.assertIn("plugin_three", discovery.skipped_plugins)
+
+
+class SupervisorMigrationTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self.left, self.right = make_transport_pair()
+        self.core = Peer(
+            transport=self.left,
+            peer_info=PeerInfo(name="test-core", role="core", version="v4"),
+        )
+        self.core.set_initialize_handler(
+            lambda _message: asyncio.sleep(
+                0,
+                result=InitializeOutput(
+                    peer=PeerInfo(name="test-core", role="core", version="v4"),
+                    capabilities=[],
+                    metadata={},
+                ),
+            )
+        )
+        await self.core.start()
+
+    async def asyncTearDown(self) -> None:
+        await self.core.stop()
+
+    async def test_supervisor_aggregates_handlers_and_routes_target_plugin(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir) / "plugins"
+            write_plugin(
+                plugins_dir,
+                "plugin_one",
+                plugin_name="plugin_one",
+                reply_text="plugin_one handled",
+            )
+            write_plugin(
+                plugins_dir,
+                "plugin_two",
+                plugin_name="plugin_two",
+                reply_text="plugin_two handled",
+            )
+
+            runtime = SupervisorRuntime(
+                transport=self.right,
+                plugins_dir=plugins_dir,
+                env_manager=FakeEnvManager(),
+            )
+            try:
+                await runtime.start()
+                await self.core.wait_until_remote_initialized()
+
+                self.assertEqual(sorted(runtime.loaded_plugins), ["plugin_one", "plugin_two"])
+                self.assertEqual(self.core.remote_metadata["plugins"], ["plugin_one", "plugin_two"])
+                self.assertEqual(len(self.core.remote_handlers), 2)
+
+                handler_id = next(
+                    item.id
+                    for item in self.core.remote_handlers
+                    if item.id.startswith("plugin_two:")
+                )
+                await self.core.invoke(
+                    "handler.invoke",
+                    {
+                        "handler_id": handler_id,
+                        "event": {
+                            "text": "/hello",
+                            "session_id": "session-1",
+                            "user_id": "user-1",
+                            "platform": "test",
+                        },
+                    },
+                    request_id="call-route",
+                )
+
+                texts = [item.get("text") for item in runtime.capability_router.sent_messages]
+                self.assertEqual(texts, ["plugin_two handled"])
+            finally:
+                await runtime.stop()
+
+
+if __name__ == "__main__":
+    unittest.main()