feat: 更新文档和测试，确保 LegacyContext 共享和兼容性

AGENTS.md

@@ -8,6 +8,7 @@
 - 2026-03-13: `CapabilityRouter.register(..., stream_handler=...)` uses the signature `(request_id, payload, cancel_token)`, not the peer-level `(message, token)`. Reusing the peer handler shape in router tests causes an immediate failed stream event before the test logic runs.
 - 2026-03-13: `Peer` had an early-cancel race for inbound invokes: if `CancelMessage` arrived before the invoke task executed its first line, the task could be cancelled before sending any terminal event, leaving the caller's stream iterator waiting forever. Preserve a per-request start event and pre-check the cancel token at the top of `_handle_invoke`.
 - 2026-03-13: `src-new/astrbot_sdk/api` and several top-level module docstrings overstated v4 compatibility gaps by labeling migrated or compat-backed APIs as "missing". Treat `_legacy_api.py`, `astrbot_sdk.api.*` thin re-exports, and top-level `events.py` / `decorators.py` as a split compatibility surface; do not mechanically recreate the old tree from stale TODO comments.
+- 2026-03-13: Legacy components are expected to share one `LegacyContext` per plugin, matching the old `StarManager` behavior. Creating one compat context per component breaks `_register_component()` / `call_context_function()` cross-component registration chains and diverges from legacy semantics.
 
 # 开发命令
 

CLAUDE.md

@@ -8,6 +8,7 @@
 - 2026-03-13: `CapabilityRouter.register(..., stream_handler=...)` uses the signature `(request_id, payload, cancel_token)`, not the peer-level `(message, token)`. Reusing the peer handler shape in router tests causes an immediate failed stream event before the test logic runs.
 - 2026-03-13: `Peer` had an early-cancel race for inbound invokes: if `CancelMessage` arrived before the invoke task executed its first line, the task could be cancelled before sending any terminal event, leaving the caller's stream iterator waiting forever. Preserve a per-request start event and pre-check the cancel token at the top of `_handle_invoke`.
 - 2026-03-13: `src-new/astrbot_sdk/api` and several top-level module docstrings overstated v4 compatibility gaps by labeling migrated or compat-backed APIs as "missing". Treat `_legacy_api.py`, `astrbot_sdk.api.*` thin re-exports, and top-level `events.py` / `decorators.py` as a split compatibility surface; do not mechanically recreate the old tree from stale TODO comments.
+- 2026-03-13: Legacy components are expected to share one `LegacyContext` per plugin, matching the old `StarManager` behavior. Creating one compat context per component breaks `_register_component()` / `call_context_function()` cross-component registration chains and diverges from legacy semantics.
 
 # 开发命令
 

src-new/astrbot_sdk/_legacy_api.py

@@ -9,6 +9,7 @@
 # 【提供的兼容类型】
 # - LegacyContext: 旧版 Context 兼容实现
 #   - 提供 llm_generate(), tool_loop_agent(), send_message() 等方法
+#   - 提供 _register_component()/call_context_function() 兼容链路
 #   - 内部委托给新版 Context 的客户端
 #
 # - LegacyConversationManager: 旧版会话管理器兼容实现
@@ -30,7 +31,6 @@
 # =============================================================================
 #
 # 1. LegacyContext 方法不完整
-#    - 缺少 _register_component() 方法（旧版有）
 #    - add_llm_tools() 抛出 NotImplementedError（旧版支持）
 #
 # 2. LegacyConversationManager 方法不完整
@@ -51,7 +51,9 @@
 
 from __future__ import annotations
 
+import inspect
 from collections import defaultdict
+from collections.abc import Callable
 from typing import Any
 
 from loguru import logger
@@ -85,6 +87,8 @@ class LegacyConversationManager:
     注意：此实现不提供持久化保证，会话数据仅在当前运行时有效。
     """
 
+    __compat_component_name__ = "ConversationManager"
+
     def __init__(self, parent: "LegacyContext") -> None:
         self._parent = parent
         self._counters: defaultdict[str, int] = defaultdict(int)
@@ -399,7 +403,10 @@ class LegacyContext:
     def __init__(self, plugin_id: str) -> None:
         self.plugin_id = plugin_id
         self._runtime_context: NewContext | None = None
+        self._registered_managers: dict[str, Any] = {}
+        self._registered_functions: dict[str, Callable[..., Any]] = {}
         self.conversation_manager = LegacyConversationManager(self)
+        self._register_component(self.conversation_manager)
 
     def bind_runtime_context(self, runtime_context: NewContext) -> None:
         self._runtime_context = runtime_context
@@ -409,6 +416,59 @@ class LegacyContext:
             raise RuntimeError("LegacyContext 尚未绑定运行时 Context")
         return self._runtime_context
 
+    @staticmethod
+    def _component_names(component: Any) -> list[str]:
+        names = [component.__class__.__name__]
+        compat_name = getattr(component, "__compat_component_name__", None)
+        if isinstance(compat_name, str) and compat_name and compat_name not in names:
+            names.insert(0, compat_name)
+        return names
+
+    def _register_component(self, *components: Any) -> None:
+        """保留旧版按名称暴露组件方法的兼容链路。"""
+        for component in components:
+            for class_name in self._component_names(component):
+                self._registered_managers[class_name] = component
+                for attr_name in dir(component):
+                    if attr_name.startswith("_"):
+                        continue
+                    try:
+                        attr = getattr(component, attr_name)
+                    except Exception:
+                        continue
+                    if callable(attr):
+                        self._registered_functions[f"{class_name}.{attr_name}"] = attr
+
+    async def execute_registered_function(
+        self,
+        func_full_name: str,
+        args: dict[str, Any] | None = None,
+    ) -> Any:
+        if args is None:
+            call_args: dict[str, Any] = {}
+        elif isinstance(args, dict):
+            call_args = args
+        else:
+            raise TypeError("LegacyContext 调用参数必须是 dict")
+
+        func = self._registered_functions.get(func_full_name)
+        if func is None:
+            raise ValueError(f"Function not found: {func_full_name}")
+
+        result = func(**call_args)
+        if inspect.isawaitable(result):
+            return await result
+        return result
+
+    async def call_context_function(
+        self,
+        func_full_name: str,
+        args: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        return {
+            "data": await self.execute_registered_function(func_full_name, args),
+        }
+
     async def llm_generate(
         self,
         chat_provider_id: str,

src-new/astrbot_sdk/runtime/loader.py

@@ -372,6 +372,7 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin:
 
     instances: list[Any] = []
     handlers: list[LoadedHandler] = []
+    shared_legacy_context = None
     for component in plugin.manifest_data.get("components", []):
         class_path = component.get("class")
         if not isinstance(class_path, str) or ":" not in class_path:
@@ -381,7 +382,12 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin:
         if _is_new_star_component(component_cls):
             instance = component_cls()
         else:
-            legacy_context = _create_legacy_context(component_cls, plugin.name)
+            if shared_legacy_context is None:
+                # 旧版 StarManager 为同一插件复用一个 Context 实例。
+                shared_legacy_context = _create_legacy_context(
+                    component_cls, plugin.name
+                )
+            legacy_context = shared_legacy_context
             try:
                 instance = component_cls(legacy_context)
             except TypeError:

tests_v4/test_api_legacy_context.py

@@ -52,6 +52,51 @@ class TestLegacyContext:
         """Context should be an alias for LegacyContext."""
         assert Context is LegacyContext
 
+    def test_auto_registers_conversation_manager_with_legacy_name(self):
+        """LegacyContext should expose ConversationManager.* legacy names."""
+        ctx = LegacyContext("test_plugin")
+
+        assert (
+            ctx._registered_managers["ConversationManager"] is ctx.conversation_manager
+        )
+        assert "ConversationManager.new_conversation" in ctx._registered_functions
+
+    @pytest.mark.asyncio
+    async def test_call_context_function_wraps_registered_result(self):
+        """call_context_function() should preserve the legacy {data: ...} shape."""
+
+        class SyncComponent:
+            def greet(self, name: str) -> str:
+                return f"hello {name}"
+
+        ctx = LegacyContext("test_plugin")
+        ctx._register_component(SyncComponent())
+
+        result = await ctx.call_context_function(
+            "SyncComponent.greet",
+            {"name": "astrbot"},
+        )
+
+        assert result == {"data": "hello astrbot"}
+
+    @pytest.mark.asyncio
+    async def test_execute_registered_function_supports_async_methods(self):
+        """execute_registered_function() should await async component methods."""
+
+        class AsyncComponent:
+            async def double(self, value: int) -> int:
+                return value * 2
+
+        ctx = LegacyContext("test_plugin")
+        ctx._register_component(AsyncComponent())
+
+        result = await ctx.execute_registered_function(
+            "AsyncComponent.double",
+            {"value": 21},
+        )
+
+        assert result == 42
+
 
 class TestLegacyConversationManager:
     """Tests for LegacyConversationManager."""
@@ -327,6 +372,7 @@ class TestLegacyContextLLMMethods:
         mock_llm.chat_raw.assert_called_once()
         call_kwargs = mock_llm.chat_raw.call_args[1]
         assert call_kwargs["max_steps"] == 10
+        assert result.text == "response"
 
     @pytest.mark.asyncio
     async def test_add_llm_tools_raises_not_implemented(self):

tests_v4/test_loader.py

@@ -654,6 +654,75 @@ class TestLoadPlugin:
                 if str(plugin_dir) in sys.path:
                     sys.path.remove(str(plugin_dir))
 
+    @pytest.mark.asyncio
+    async def test_load_plugin_shares_legacy_context_between_components(self):
+        """Legacy components in one plugin should share the same LegacyContext."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir) / "test_plugin"
+            plugin_dir.mkdir()
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+            module_dir = plugin_dir / "legacy_pkg"
+            module_dir.mkdir()
+            (module_dir / "__init__.py").write_text("", encoding="utf-8")
+            (module_dir / "components.py").write_text(
+                textwrap.dedent(
+                    """\
+                    from astrbot_sdk.api.components.command import CommandComponent
+                    from astrbot_sdk.api.star.context import Context
+
+
+                    class FirstComponent(CommandComponent):
+                        def __init__(self, context: Context):
+                            self.context = context
+                            context._register_component(self)
+
+                        def echo(self, text: str) -> str:
+                            return f"first:{text}"
+
+
+                    class SecondComponent(CommandComponent):
+                        def __init__(self, context: Context):
+                            self.context = context
+                    """
+                ),
+                encoding="utf-8",
+            )
+
+            manifest_path.write_text(
+                yaml.dump(
+                    {
+                        "name": "test_plugin",
+                        "runtime": {"python": "3.12"},
+                        "components": [
+                            {"class": "legacy_pkg.components:FirstComponent"},
+                            {"class": "legacy_pkg.components:SecondComponent"},
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            spec = load_plugin_spec(plugin_dir)
+
+            if str(plugin_dir) not in sys.path:
+                sys.path.insert(0, str(plugin_dir))
+
+            try:
+                loaded = load_plugin(spec)
+
+                assert len(loaded.instances) == 2
+                assert loaded.instances[0].context is loaded.instances[1].context
+                result = await loaded.instances[1].context.call_context_function(
+                    "FirstComponent.echo",
+                    {"text": "hi"},
+                )
+                assert result == {"data": "first:hi"}
+            finally:
+                if str(plugin_dir) in sys.path:
+                    sys.path.remove(str(plugin_dir))
+
 
 class TestStateFileConstant:
     """Tests for STATE_FILE_NAME constant."""