Refactor legacy runtime execution boundary

AGENTS.md

@@ -31,6 +31,7 @@
 - 2026-03-13: Legacy AI compat methods must return `astrbot_sdk.api.provider.entities.LLMResponse`, not the v4 `clients.llm.LLMResponse`. Old plugins inspect compat fields like `completion_text`, `tools_call_name`, and `to_openai_tool_calls()`, so returning the new client model is a silent behavior regression.
 - 2026-03-13: `filter.llm_tool()` must resolve deferred annotations from `from __future__ import annotations` when inferring JSON schema. Reading `inspect.Parameter.annotation` directly degrades typed params like `a: int` into `"int"` strings and silently turns tool argument schemas into generic strings.
 - 2026-03-13: Legacy result hooks must reuse the same `AstrMessageEvent` instance across `on_decorating_result` and `after_message_sent`. Re-wrapping the original v4 `MessageEvent` for the second hook drops decorated `event.set_result(...)` mutations and makes post-send hooks observe an empty result.
+- 2026-03-13: `src-new/astrbot_sdk/_legacy_runtime.py` already exists as the intended compat execution boundary. When cleaning runtime architecture, wire `loader` / `handler_dispatcher` / `bootstrap` through that adapter instead of adding new direct `legacy_context` branches in runtime files, or the compat logic will spread again.
 
 # 开发命令
 

CLAUDE.md

@@ -31,6 +31,7 @@
 - 2026-03-13: Legacy AI compat methods must return `astrbot_sdk.api.provider.entities.LLMResponse`, not the v4 `clients.llm.LLMResponse`. Old plugins inspect compat fields like `completion_text`, `tools_call_name`, and `to_openai_tool_calls()`, so returning the new client model is a silent behavior regression.
 - 2026-03-13: `filter.llm_tool()` must resolve deferred annotations from `from __future__ import annotations` when inferring JSON schema. Reading `inspect.Parameter.annotation` directly degrades typed params like `a: int` into `"int"` strings and silently turns tool argument schemas into generic strings.
 - 2026-03-13: Legacy result hooks must reuse the same `AstrMessageEvent` instance across `on_decorating_result` and `after_message_sent`. Re-wrapping the original v4 `MessageEvent` for the second hook drops decorated `event.set_result(...)` mutations and makes post-send hooks observe an empty result.
+- 2026-03-13: `src-new/astrbot_sdk/_legacy_runtime.py` already exists as the intended compat execution boundary. When cleaning runtime architecture, wire `loader` / `handler_dispatcher` / `bootstrap` through that adapter instead of adding new direct `legacy_context` branches in runtime files, or the compat logic will spread again.
 
 # 开发命令
 

src-new/astrbot_sdk/_legacy_runtime.py

@@ -0,0 +1,171 @@
+"""legacy 运行时执行适配。
+
+这个模块把 compat 执行细节从 runtime 主干中收口出来：
+
+- 旧自定义过滤器执行
+- 旧结果装饰与发送后 hook
+- 旧插件错误 hook
+- worker 生命周期中的 compat hook 调用
+
+v4 主干只与这个适配层交互，不直接展开 legacy 事件包装和 hook 名称。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass, field
+from typing import Any
+
+from .api.event import AstrMessageEvent
+from .api.event.event_result import MessageEventResult
+from .api.message.chain import MessageChain
+from .context import Context
+from .events import MessageEvent
+
+
+@dataclass(slots=True)
+class LegacyPreparedResult:
+    item: Any
+    compat_event: AstrMessageEvent | None = None
+    stopped: bool = False
+
+
+@dataclass(slots=True)
+class LegacyRuntimeAdapter:
+    legacy_context: Any
+    filters: list[Any] = field(default_factory=list)
+
+    @classmethod
+    def from_handler(cls, legacy_context: Any, handler: Any) -> "LegacyRuntimeAdapter":
+        from .api.event.filter import get_compat_custom_filters
+
+        return cls(
+            legacy_context=legacy_context,
+            filters=list(get_compat_custom_filters(handler)),
+        )
+
+    def bind_runtime_context(self, runtime_context: Context) -> None:
+        binder = getattr(self.legacy_context, "bind_runtime_context", None)
+        if callable(binder):
+            binder(runtime_context)
+
+    def register_component(self, component: Any) -> None:
+        register = getattr(self.legacy_context, "_register_compat_component", None)
+        if callable(register):
+            register(component)
+
+    async def run_hook(self, hook_name: str, **kwargs: Any) -> list[Any]:
+        runner = getattr(self.legacy_context, "_run_compat_hook", None)
+        if not callable(runner):
+            return []
+        return await runner(
+            hook_name,
+            legacy_context=self.legacy_context,
+            **kwargs,
+        )
+
+    def runtime_config(self) -> Any:
+        config_getter = getattr(self.legacy_context, "_runtime_config", None)
+        if callable(config_getter):
+            return config_getter()
+        return None
+
+    async def passes_filters(self, event: MessageEvent) -> bool:
+        if not self.filters:
+            return True
+        compat_event = AstrMessageEvent.from_message_event(event)
+        cfg = self.runtime_config()
+        for filter_obj in self.filters:
+            if not filter_obj.filter(compat_event, cfg):
+                return False
+        return True
+
+    async def prepare_result(
+        self,
+        item: Any,
+        event: MessageEvent,
+        ctx: Context | None,
+    ) -> LegacyPreparedResult:
+        compat_event = AstrMessageEvent.from_message_event(event)
+        if isinstance(item, (MessageEventResult, MessageChain, str)):
+            compat_event.set_result(item)
+            await self.run_hook(
+                "on_decorating_result",
+                event=compat_event,
+                context=ctx,
+                result=compat_event.get_result(),
+            )
+            if compat_event.is_stopped():
+                return LegacyPreparedResult(
+                    item=item,
+                    compat_event=compat_event,
+                    stopped=True,
+                )
+            item = compat_event.get_result() or item
+        return LegacyPreparedResult(item=item, compat_event=compat_event)
+
+    async def after_send(
+        self,
+        compat_event: AstrMessageEvent | None,
+        ctx: Context | None,
+    ) -> None:
+        if compat_event is None:
+            return
+        await self.run_hook(
+            "after_message_sent",
+            event=compat_event,
+            context=ctx,
+        )
+
+    async def handle_error(
+        self,
+        *,
+        plugin_id: str,
+        handler_name: str,
+        exc: Exception,
+        event: MessageEvent,
+        ctx: Context,
+        traceback_text: str,
+    ) -> None:
+        await self.run_hook(
+            "on_plugin_error",
+            event=AstrMessageEvent.from_message_event(event),
+            context=ctx,
+            plugin_name=plugin_id,
+            handler_name=handler_name,
+            error=exc,
+            traceback_text=traceback_text,
+        )
+
+
+def get_legacy_runtime_adapter(loaded: Any) -> LegacyRuntimeAdapter | None:
+    adapter = getattr(loaded, "legacy_runtime", None)
+    if adapter is not None:
+        return adapter
+    legacy_context = getattr(loaded, "legacy_context", None)
+    if legacy_context is None:
+        return None
+    filters = list(getattr(loaded, "compat_filters", ()))
+    adapter = LegacyRuntimeAdapter(legacy_context=legacy_context, filters=filters)
+    try:
+        setattr(loaded, "legacy_runtime", adapter)
+    except AttributeError:
+        pass
+    return adapter
+
+
+def iter_unique_legacy_runtime_adapters(
+    loaded_items: Iterable[Any],
+) -> list[LegacyRuntimeAdapter]:
+    seen_contexts: set[int] = set()
+    adapters: list[LegacyRuntimeAdapter] = []
+    for loaded in loaded_items:
+        adapter = get_legacy_runtime_adapter(loaded)
+        if adapter is None:
+            continue
+        marker = id(adapter.legacy_context)
+        if marker in seen_contexts:
+            continue
+        seen_contexts.add(marker)
+        adapters.append(adapter)
+    return adapters

src-new/astrbot_sdk/runtime/bootstrap.py

@@ -96,6 +96,7 @@ from typing import IO, Any
 
 from loguru import logger
 
+from .._legacy_runtime import iter_unique_legacy_runtime_adapters
 from ..context import Context as RuntimeContext
 from ..errors import AstrBotError
 from ..protocol.descriptors import CapabilityDescriptor
@@ -729,40 +730,20 @@ class PluginWorkerRuntime:
                 await result
 
     def _bind_legacy_runtime_contexts(self, runtime_context: RuntimeContext) -> None:
-        seen: set[int] = set()
-        for loaded in [
-            *self.loaded_plugin.handlers,
-            *self.loaded_plugin.capabilities,
-        ]:
-            legacy_context = getattr(loaded, "legacy_context", None)
-            if legacy_context is None:
-                continue
-            marker = id(legacy_context)
-            if marker in seen:
-                continue
-            seen.add(marker)
-            bind_runtime_context = getattr(legacy_context, "bind_runtime_context", None)
-            if callable(bind_runtime_context):
-                bind_runtime_context(runtime_context)
+        for adapter in iter_unique_legacy_runtime_adapters(
+            [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities]
+        ):
+            adapter.bind_runtime_context(runtime_context)
 
     async def _run_compat_context_hook(self, hook_name: str, **kwargs: Any) -> None:
-        seen: set[int] = set()
-        for loaded in [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities]:
-            legacy_context = getattr(loaded, "legacy_context", None)
-            if legacy_context is None:
-                continue
-            marker = id(legacy_context)
-            if marker in seen:
-                continue
-            seen.add(marker)
-            run_hook = getattr(legacy_context, "_run_compat_hook", None)
-            if callable(run_hook):
-                await run_hook(
-                    hook_name,
-                    context=self._lifecycle_context,
-                    legacy_context=legacy_context,
-                    **kwargs,
-                )
+        for adapter in iter_unique_legacy_runtime_adapters(
+            [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities]
+        ):
+            await adapter.run_hook(
+                hook_name,
+                context=self._lifecycle_context,
+                **kwargs,
+            )
 
     @staticmethod
     def _resolve_lifecycle_hook(instance: Any, method_name: str):

src-new/astrbot_sdk/runtime/handler_dispatcher.py

@@ -73,6 +73,7 @@ import typing
 from collections.abc import AsyncIterator
 from typing import Any, get_type_hints
 
+from .._legacy_runtime import get_legacy_runtime_adapter
 from .._session_waiter import SessionWaiterManager
 from ..context import CancelToken, Context
 from ..errors import AstrBotError
@@ -104,9 +105,10 @@ class HandlerDispatcher:
         event.bind_reply_handler(self._create_reply_handler(ctx, event))
         if await self._session_waiters.dispatch(event):
             return {}
-        if loaded.legacy_context is not None:
-            loaded.legacy_context.bind_runtime_context(ctx)
-            if not await self._passes_compat_filters(loaded, event):
+        legacy_runtime = get_legacy_runtime_adapter(loaded)
+        if legacy_runtime is not None:
+            legacy_runtime.bind_runtime_context(ctx)
+            if not await legacy_runtime.passes_filters(event):
                 return {}
 
         # 提取 legacy args 用于兼容旧版 handler 签名
@@ -149,6 +151,7 @@ class HandlerDispatcher:
         ctx: Context,
         legacy_args: dict[str, Any] | None = None,
     ) -> None:
+        legacy_runtime = get_legacy_runtime_adapter(loaded)
         try:
             result = loaded.callable(
                 *self._build_args(loaded.callable, event, ctx, legacy_args)
@@ -159,7 +162,7 @@ class HandlerDispatcher:
                         item,
                         event,
                         ctx,
-                        legacy_context=loaded.legacy_context,
+                        legacy_runtime=legacy_runtime,
                     )
                 return
             if inspect.isawaitable(result):
@@ -169,7 +172,7 @@ class HandlerDispatcher:
                     result,
                     event,
                     ctx,
-                    legacy_context=loaded.legacy_context,
+                    legacy_runtime=legacy_runtime,
                 )
         except Exception as exc:
             await self._handle_error(
@@ -177,7 +180,7 @@ class HandlerDispatcher:
                 exc,
                 event,
                 ctx,
-                legacy_context=loaded.legacy_context,
+                legacy_runtime=legacy_runtime,
                 handler_name=loaded.callable.__name__,
             )
             raise
@@ -306,27 +309,18 @@ class HandlerDispatcher:
         event: MessageEvent,
         ctx: Context | None = None,
         *,
-        legacy_context=None,
+        legacy_runtime=None,
     ) -> None:
         from ..api.event.event_result import MessageEventResult
-        from ..api.event import AstrMessageEvent
         from ..api.message.chain import MessageChain
 
         compat_event = None
-        if legacy_context is not None:
-            compat_event = AstrMessageEvent.from_message_event(event)
-            if isinstance(item, (MessageEventResult, MessageChain, str)):
-                compat_event.set_result(item)
-                await legacy_context._run_compat_hook(
-                    "on_decorating_result",
-                    event=compat_event,
-                    context=ctx,
-                    legacy_context=legacy_context,
-                    result=compat_event.get_result(),
-                )
-                if compat_event.is_stopped():
-                    return
-                item = compat_event.get_result() or item
+        if legacy_runtime is not None:
+            prepared = await legacy_runtime.prepare_result(item, event, ctx)
+            compat_event = prepared.compat_event
+            if prepared.stopped:
+                return
+            item = prepared.item
 
         if isinstance(item, MessageEventResult):
             if item.chain and ctx is not None and not item.is_plain_text_only():
@@ -334,24 +328,14 @@ class HandlerDispatcher:
                     event.session_ref or event.session_id,
                     item.to_payload(),
                 )
-                if legacy_context is not None:
-                    await legacy_context._run_compat_hook(
-                        "after_message_sent",
-                        event=compat_event,
-                        context=ctx,
-                        legacy_context=legacy_context,
-                    )
+                if legacy_runtime is not None:
+                    await legacy_runtime.after_send(compat_event, ctx)
                 return
             plain_text = item.get_plain_text()
             if plain_text:
                 await event.reply(plain_text)
-                if legacy_context is not None:
-                    await legacy_context._run_compat_hook(
-                        "after_message_sent",
-                        event=compat_event,
-                        context=ctx,
-                        legacy_context=legacy_context,
-                    )
+                if legacy_runtime is not None:
+                    await legacy_runtime.after_send(compat_event, ctx)
             return
         if isinstance(item, MessageChain):
             if item.chain and ctx is not None and not item.is_plain_text_only():
@@ -359,70 +343,29 @@ class HandlerDispatcher:
                     event.session_ref or event.session_id,
                     item.to_payload(),
                 )
-                if legacy_context is not None:
-                    await legacy_context._run_compat_hook(
-                        "after_message_sent",
-                        event=compat_event,
-                        context=ctx,
-                        legacy_context=legacy_context,
-                    )
+                if legacy_runtime is not None:
+                    await legacy_runtime.after_send(compat_event, ctx)
                 return
             plain_text = item.get_plain_text()
             if plain_text:
                 await event.reply(plain_text)
-                if legacy_context is not None:
-                    await legacy_context._run_compat_hook(
-                        "after_message_sent",
-                        event=compat_event,
-                        context=ctx,
-                        legacy_context=legacy_context,
-                    )
+                if legacy_runtime is not None:
+                    await legacy_runtime.after_send(compat_event, ctx)
             return
         if isinstance(item, PlainTextResult):
             await event.reply(item.text)
-            if legacy_context is not None:
-                await legacy_context._run_compat_hook(
-                    "after_message_sent",
-                    event=compat_event or AstrMessageEvent.from_message_event(event),
-                    context=ctx,
-                    legacy_context=legacy_context,
-                )
+            if legacy_runtime is not None:
+                await legacy_runtime.after_send(compat_event, ctx)
             return
         if isinstance(item, str):
             await event.reply(item)
-            if legacy_context is not None:
-                await legacy_context._run_compat_hook(
-                    "after_message_sent",
-                    event=compat_event or AstrMessageEvent.from_message_event(event),
-                    context=ctx,
-                    legacy_context=legacy_context,
-                )
+            if legacy_runtime is not None:
+                await legacy_runtime.after_send(compat_event, ctx)
             return
         if isinstance(item, dict) and "text" in item:
             await event.reply(str(item["text"]))
-            if legacy_context is not None:
-                await legacy_context._run_compat_hook(
-                    "after_message_sent",
-                    event=compat_event or AstrMessageEvent.from_message_event(event),
-                    context=ctx,
-                    legacy_context=legacy_context,
-                )
-
-    async def _passes_compat_filters(
-        self,
-        loaded: LoadedHandler,
-        event: MessageEvent,
-    ) -> bool:
-        if not loaded.compat_filters or loaded.legacy_context is None:
-            return True
-        from ..api.event import AstrMessageEvent
-
-        compat_event = AstrMessageEvent.from_message_event(event)
-        cfg = loaded.legacy_context._runtime_config()
-        for filter_obj in loaded.compat_filters:
-            if not filter_obj.filter(compat_event, cfg):
-                return False
-        return True
+            if legacy_runtime is not None:
+                await legacy_runtime.after_send(compat_event, ctx)
 
     async def _handle_error(
         self,
@@ -431,20 +374,16 @@ class HandlerDispatcher:
         event: MessageEvent,
         ctx: Context,
         *,
-        legacy_context=None,
+        legacy_runtime=None,
         handler_name: str = "",
     ) -> None:
-        if legacy_context is not None:
-            from ..api.event import AstrMessageEvent
-
-            await legacy_context._run_compat_hook(
-                "on_plugin_error",
-                event=AstrMessageEvent.from_message_event(event),
-                context=ctx,
-                legacy_context=legacy_context,
-                plugin_name=self._plugin_id,
+        if legacy_runtime is not None:
+            await legacy_runtime.handle_error(
+                plugin_id=self._plugin_id,
                 handler_name=handler_name,
-                error=exc,
+                exc=exc,
+                event=event,
+                ctx=ctx,
                 traceback_text="".join(
                     traceback.TracebackException.from_exception(exc).format()
                 ),
@@ -484,8 +423,9 @@ class CapabilityDispatcher:
             plugin_id=self._plugin_id,
             cancel_token=cancel_token,
         )
-        if loaded.legacy_context is not None:
-            loaded.legacy_context.bind_runtime_context(ctx)
+        legacy_runtime = get_legacy_runtime_adapter(loaded)
+        if legacy_runtime is not None:
+            legacy_runtime.bind_runtime_context(ctx)
 
         task = asyncio.create_task(
             self._run_capability(

src-new/astrbot_sdk/runtime/loader.py

@@ -100,9 +100,7 @@ from typing import Any
 
 import yaml
 
-from ..api.event.filter import (
-    get_compat_custom_filters,
-)
+from .._legacy_runtime import LegacyRuntimeAdapter
 from ..api.basic import AstrBotConfig
 from ..decorators import get_capability_meta, get_handler_meta
 from ..protocol.descriptors import CapabilityDescriptor, HandlerDescriptor
@@ -155,6 +153,21 @@ class LoadedHandler:
     owner: Any
     legacy_context: Any | None = None
     compat_filters: list[Any] = field(default_factory=list)
+    legacy_runtime: LegacyRuntimeAdapter | None = field(
+        init=False, default=None, repr=False
+    )
+
+    def __post_init__(self) -> None:
+        if self.legacy_context is None:
+            return
+        if not self.compat_filters:
+            from ..api.event.filter import get_compat_custom_filters
+
+            self.compat_filters = list(get_compat_custom_filters(self.callable))
+        self.legacy_runtime = LegacyRuntimeAdapter(
+            legacy_context=self.legacy_context,
+            filters=list(self.compat_filters),
+        )
 
 
 @dataclass(slots=True)
@@ -163,6 +176,16 @@ class LoadedCapability:
     callable: Any
     owner: Any
     legacy_context: Any | None = None
+    legacy_runtime: LegacyRuntimeAdapter | None = field(
+        init=False, default=None, repr=False
+    )
+
+    def __post_init__(self) -> None:
+        if self.legacy_context is None:
+            return
+        self.legacy_runtime = LegacyRuntimeAdapter(
+            legacy_context=self.legacy_context,
+        )
 
 
 @dataclass(slots=True)
@@ -700,11 +723,9 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin:
                 and getattr(instance, "config", None) is None
             ):
                 setattr(instance, "config", component_config)
-            register_compat_component = getattr(
-                legacy_context, "_register_compat_component", None
+            LegacyRuntimeAdapter(legacy_context=legacy_context).register_component(
+                instance
             )
-            if callable(register_compat_component):
-                register_compat_component(instance)
         instances.append(instance)
         for name in _iter_discoverable_names(instance):
             resolved = _resolve_handler_candidate(instance, name)
@@ -737,7 +758,6 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin:
                     callable=bound,
                     owner=instance,
                     legacy_context=legacy_context,
-                    compat_filters=list(get_compat_custom_filters(bound)),
                 )
             )
     return LoadedPlugin(

tests_v4/test_bootstrap.py

@@ -16,6 +16,7 @@ import pytest
 import yaml
 
 from astrbot_sdk._legacy_api import LegacyContext
+from astrbot_sdk._legacy_runtime import LegacyRuntimeAdapter
 from astrbot_sdk.context import CancelToken
 from astrbot_sdk.errors import AstrBotError
 from astrbot_sdk.protocol.descriptors import (
@@ -1064,10 +1065,14 @@ class DemoComponent(Star):
             )
             legacy_context = LegacyContext("test_plugin")
             runtime.loaded_plugin.handlers.append(
-                SimpleNamespace(legacy_context=legacy_context)
+                SimpleNamespace(
+                    legacy_runtime=LegacyRuntimeAdapter(legacy_context=legacy_context)
+                )
             )
             runtime.loaded_plugin.capabilities.append(
-                SimpleNamespace(legacy_context=legacy_context)
+                SimpleNamespace(
+                    legacy_runtime=LegacyRuntimeAdapter(legacy_context=legacy_context)
+                )
             )
 
             runtime._bind_legacy_runtime_contexts(runtime._lifecycle_context)

tests_v4/test_handler_dispatcher.py

@@ -16,6 +16,7 @@ from astrbot.core.utils.session_waiter import (
     session_waiter,
 )
 from astrbot_sdk._legacy_api import LegacyContext
+from astrbot_sdk._legacy_runtime import LegacyRuntimeAdapter
 from astrbot_sdk.api.event import AstrMessageEvent
 from astrbot_sdk.api.event.filter import (
     CustomFilter,
@@ -580,6 +581,60 @@ class TestHandlerDispatcherInvoke:
 class TestHandlerDispatcherLegacyCompat:
     """Tests for legacy compat hooks and filters during dispatch."""
 
+    @pytest.mark.asyncio
+    async def test_prebuilt_legacy_runtime_adapter_can_drive_filtering(self):
+        """Dispatcher should prefer the adapter boundary instead of reading raw legacy fields."""
+        peer = MockPeer()
+        legacy_context = LegacyContext("test_plugin")
+        called = []
+
+        class RejectAll(CustomFilter):
+            def filter(self, event: AstrMessageEvent, cfg) -> bool:
+                return False
+
+        async def handler_func(event: AstrMessageEvent):
+            called.append(event.message_str)
+
+        descriptor = HandlerDescriptor(
+            id="legacy.adapter.filtered",
+            trigger=CommandTrigger(command="ask"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=handler_func,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+        handler.legacy_runtime = LegacyRuntimeAdapter(
+            legacy_context=legacy_context,
+            filters=[RejectAll()],
+        )
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        message = InvokeMessage(
+            id="msg_filtered_adapter",
+            capability="handler.invoke",
+            input={
+                "handler_id": "legacy.adapter.filtered",
+                "event": {
+                    "text": "ask",
+                    "session_id": "session-filtered",
+                    "user_id": "user-1",
+                    "platform": "test",
+                },
+            },
+        )
+
+        result = await dispatcher.invoke(message, CancelToken())
+
+        assert result == {}
+        assert called == []
+        assert peer.sent_messages == []
+
     @pytest.mark.asyncio
     async def test_custom_filter_can_skip_legacy_handler_invocation(self):
         """Legacy custom filters should prevent handler execution when they reject the event."""

tests_v4/test_loader.py

@@ -14,6 +14,8 @@ from unittest.mock import MagicMock, patch
 import pytest
 import yaml
 
+from astrbot_sdk._legacy_api import LegacyContext
+from astrbot_sdk._legacy_runtime import LegacyRuntimeAdapter
 from astrbot_sdk.protocol.descriptors import CommandTrigger, HandlerDescriptor
 from astrbot_sdk.runtime.environment_groups import (
     GROUP_STATE_FILE_NAME,
@@ -159,6 +161,28 @@ class TestLoadedHandler:
         assert loaded.callable == handler_func
         assert loaded.owner == owner
         assert loaded.legacy_context is None
+        assert loaded.legacy_runtime is None
+
+    def test_init_builds_legacy_runtime_adapter(self):
+        """LoadedHandler should prebuild the legacy runtime adapter for runtime use."""
+        descriptor = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+        legacy_context = LegacyContext("test_plugin")
+
+        def handler_func():
+            pass
+
+        loaded = LoadedHandler(
+            descriptor=descriptor,
+            callable=handler_func,
+            owner=MagicMock(),
+            legacy_context=legacy_context,
+        )
+
+        assert isinstance(loaded.legacy_runtime, LegacyRuntimeAdapter)
+        assert loaded.legacy_runtime.legacy_context is legacy_context
 
 
 class TestLoadedPlugin: