diff --git a/tests_v4/test_api_contract.py b/tests_v4/test_api_contract.py
index db668f4e4..8bc06693b 100644
--- a/tests_v4/test_api_contract.py
+++ b/tests_v4/test_api_contract.py
@@ -71,14 +71,15 @@ class ApiContractTest(unittest.IsolatedAsyncioTestCase):
             ],
         )
 
-    async def test_compat_warning_includes_migration_doc_url(self) -> None:
+    async def test_add_llm_tools_raises_not_implemented(self) -> None:
+        """add_llm_tools() should raise NotImplementedError in v4."""
         _warned_methods.clear()
         legacy_context = LegacyContext("compat-plugin")
-        with patch("astrbot_sdk._legacy_api.logger.warning") as warning:
+        with self.assertRaises(NotImplementedError) as context:
             await legacy_context.add_llm_tools()
 
-        warning.assert_called_once()
-        self.assertEqual(warning.call_args.args[-1], MIGRATION_DOC_URL)
+        self.assertIn("add_llm_tools", str(context.exception))
+        self.assertIn(MIGRATION_DOC_URL, str(context.exception))
 
     async def test_compat_llm_generate_warning_matches_chat_raw_mapping(self) -> None:
         class _DummyLLM:
diff --git a/tests_v4/test_api_legacy_context.py b/tests_v4/test_api_legacy_context.py
index f10eda757..fdedab7f2 100644
--- a/tests_v4/test_api_legacy_context.py
+++ b/tests_v4/test_api_legacy_context.py
@@ -328,13 +328,15 @@ class TestLegacyContextLLMMethods:
         assert call_kwargs["max_steps"] == 10
 
     @pytest.mark.asyncio
-    async def test_add_llm_tools_returns_none(self):
-        """add_llm_tools() should return None (deprecated)."""
+    async def test_add_llm_tools_raises_not_implemented(self):
+        """add_llm_tools() should raise NotImplementedError in v4."""
         legacy_ctx = LegacyContext("test_plugin")
 
-        result = await legacy_ctx.add_llm_tools("tool1", "tool2")
+        with pytest.raises(NotImplementedError) as exc_info:
+            await legacy_ctx.add_llm_tools("tool1", "tool2")
 
-        assert result is None
+        assert "add_llm_tools" in str(exc_info.value)
+        assert MIGRATION_DOC_URL in str(exc_info.value)
 
 
 class TestCommandComponent:
diff --git a/tests_v4/test_bootstrap.py b/tests_v4/test_bootstrap.py
new file mode 100644
index 000000000..08ae06b3d
--- /dev/null
+++ b/tests_v4/test_bootstrap.py
@@ -0,0 +1,657 @@
+"""
+Tests for runtime/bootstrap.py - Bootstrap and runtime classes.
+"""
+from __future__ import annotations
+
+import asyncio
+import signal
+import sys
+import tempfile
+import textwrap
+from io import StringIO
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import yaml
+
+from astrbot_sdk.context import CancelToken
+from astrbot_sdk.errors import AstrBotError
+from astrbot_sdk.protocol.descriptors import CapabilityDescriptor, CommandTrigger, HandlerDescriptor
+from astrbot_sdk.protocol.messages import InitializeMessage, InitializeOutput, InvokeMessage, PeerInfo
+from astrbot_sdk.runtime.bootstrap import (
+    PluginWorkerRuntime,
+    SupervisorRuntime,
+    WorkerSession,
+    _install_signal_handlers,
+    _prepare_stdio_transport,
+    _wait_for_shutdown,
+)
+from astrbot_sdk.runtime.capability_router import CapabilityRouter
+from astrbot_sdk.runtime.loader import PluginSpec
+from astrbot_sdk.runtime.peer import Peer
+
+from tests_v4.helpers import FakeEnvManager, MemoryTransport, make_transport_pair
+
+
+async def start_test_core_peer(transport: MemoryTransport) -> Peer:
+    """Provide an initialize responder so transport-pair startup tests do not deadlock."""
+    core = Peer(
+        transport=transport,
+        peer_info=PeerInfo(name="test-core", role="core", version="v4"),
+    )
+    core.set_initialize_handler(
+        lambda _message: asyncio.sleep(
+            0,
+            result=InitializeOutput(
+                peer=PeerInfo(name="test-core", role="core", version="v4"),
+                capabilities=[],
+                metadata={},
+            ),
+        )
+    )
+    await core.start()
+    return core
+
+
+class TestInstallSignalHandlers:
+    """Tests for _install_signal_handlers function."""
+
+    def test_installs_handlers(self):
+        """_install_signal_handlers should install signal handlers."""
+        stop_event = asyncio.Event()
+
+        async def run_test():
+            _install_signal_handlers(stop_event)
+            # Just verify it doesn't raise on platforms that support it
+
+        asyncio.get_event_loop().run_until_complete(run_test())
+
+    def test_handles_not_implemented(self):
+        """_install_signal_handlers should handle NotImplementedError."""
+        stop_event = asyncio.Event()
+
+        async def run_test():
+            with patch.object(
+                asyncio.get_running_loop(),
+                "add_signal_handler",
+                side_effect=NotImplementedError,
+            ):
+                _install_signal_handlers(stop_event)
+
+        asyncio.get_event_loop().run_until_complete(run_test())
+
+
+class TestPrepareStdioTransport:
+    """Tests for _prepare_stdio_transport function."""
+
+    def test_with_both_streams(self):
+        """_prepare_stdio_transport should use provided streams."""
+        stdin = StringIO()
+        stdout = StringIO()
+
+        in_stream, out_stream, original = _prepare_stdio_transport(stdin, stdout)
+
+        assert in_stream is stdin
+        assert out_stream is stdout
+        assert original is None
+
+    def test_without_streams(self):
+        """_prepare_stdio_transport should use sys.stdin/stdout."""
+        original_stdout = sys.stdout
+
+        in_stream, out_stream, original = _prepare_stdio_transport(None, None)
+
+        assert in_stream is sys.stdin
+        assert out_stream is sys.stdout
+        assert original is original_stdout
+
+    def test_redirects_stdout(self):
+        """_prepare_stdio_transport should redirect sys.stdout to stderr."""
+        original_stdout = sys.stdout
+
+        _prepare_stdio_transport(None, None)
+
+        assert sys.stdout is sys.stderr
+
+        # Restore
+        sys.stdout = original_stdout
+
+
+class TestWaitForShutdown:
+    """Tests for _wait_for_shutdown function."""
+
+    @pytest.mark.asyncio
+    async def test_waits_for_stop_event(self):
+        """_wait_for_shutdown should wait for stop_event."""
+        left, right = make_transport_pair()
+        peer = MagicMock()
+        peer.wait_closed = AsyncMock()
+
+        stop_event = asyncio.Event()
+
+        async def set_event():
+            await asyncio.sleep(0.05)
+            stop_event.set()
+
+        asyncio.create_task(set_event())
+
+        await _wait_for_shutdown(peer, stop_event)
+
+        assert stop_event.is_set()
+
+    @pytest.mark.asyncio
+    async def test_waits_for_peer_closed(self):
+        """_wait_for_shutdown should wait for peer.wait_closed()."""
+        left, right = make_transport_pair()
+        peer = MagicMock()
+        peer.wait_closed = AsyncMock()
+
+        stop_event = asyncio.Event()
+
+        # Set wait_closed to complete
+        async def complete_wait_closed():
+            await asyncio.sleep(0.05)
+
+        peer.wait_closed.return_value = complete_wait_closed()
+
+        await _wait_for_shutdown(peer, stop_event)
+
+        peer.wait_closed.assert_called_once()
+
+
+class TestWorkerSessionInit:
+    """Tests for WorkerSession initialization."""
+
+    def test_init(self):
+        """WorkerSession should store all parameters."""
+        plugin = PluginSpec(
+            name="test",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+        router = CapabilityRouter()
+        env_manager = FakeEnvManager()
+
+        session = WorkerSession(
+            plugin=plugin,
+            repo_root=Path("/repo"),
+            env_manager=env_manager,
+            capability_router=router,
+        )
+
+        assert session.plugin == plugin
+        assert session.capability_router == router
+        assert session.peer is None
+        assert session.handlers == []
+
+
+class TestWorkerSessionMethods:
+    """Tests for WorkerSession methods."""
+
+    @pytest.mark.asyncio
+    async def test_invoke_handler_without_peer_raises(self):
+        """invoke_handler should raise if peer is None."""
+        plugin = PluginSpec(
+            name="test",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+        session = WorkerSession(
+            plugin=plugin,
+            repo_root=Path("/tmp"),
+            env_manager=FakeEnvManager(),
+            capability_router=CapabilityRouter(),
+        )
+
+        with pytest.raises(RuntimeError, match="not running"):
+            await session.invoke_handler("handler.id", {}, request_id="req-1")
+
+    @pytest.mark.asyncio
+    async def test_cancel_without_peer_does_nothing(self):
+        """cancel should do nothing if peer is None."""
+        plugin = PluginSpec(
+            name="test",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+        session = WorkerSession(
+            plugin=plugin,
+            repo_root=Path("/tmp"),
+            env_manager=FakeEnvManager(),
+            capability_router=CapabilityRouter(),
+        )
+
+        # Should not raise
+        await session.cancel("req-1")
+
+    @pytest.mark.asyncio
+    async def test_handle_initialize(self):
+        """_handle_initialize should return InitializeOutput."""
+        plugin = PluginSpec(
+            name="test_plugin",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+        router = CapabilityRouter()
+        session = WorkerSession(
+            plugin=plugin,
+            repo_root=Path("/tmp"),
+            env_manager=FakeEnvManager(),
+            capability_router=router,
+        )
+
+        message = InitializeMessage(
+            id="init-1",
+            protocol_version="1.0",
+            peer=PeerInfo(name="test", role="plugin"),
+        )
+
+        output = await session._handle_initialize(message)
+
+        assert output.peer.name == "astrbot-supervisor"
+        assert output.peer.role == "core"
+        assert len(output.capabilities) > 0
+
+    @pytest.mark.asyncio
+    async def test_handle_capability_invoke(self):
+        """_handle_capability_invoke should route to capability_router."""
+        plugin = PluginSpec(
+            name="test",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+        router = CapabilityRouter()
+        session = WorkerSession(
+            plugin=plugin,
+            repo_root=Path("/tmp"),
+            env_manager=FakeEnvManager(),
+            capability_router=router,
+        )
+
+        message = InvokeMessage(
+            id="invoke-1",
+            capability="llm.chat",
+            input={"prompt": "hello"},
+        )
+        token = CancelToken()
+
+        result = await session._handle_capability_invoke(message, token)
+
+        assert result["text"] == "Echo: hello"
+
+
+class TestSupervisorRuntimeInit:
+    """Tests for SupervisorRuntime initialization."""
+
+    def test_init(self):
+        """SupervisorRuntime should initialize correctly."""
+        transport = MemoryTransport()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=transport,
+                plugins_dir=Path(temp_dir),
+            )
+
+            assert runtime.transport is transport
+            assert runtime.worker_sessions == {}
+            assert runtime.handler_to_worker == {}
+            assert runtime.active_requests == {}
+            assert runtime.loaded_plugins == []
+            assert isinstance(runtime.capability_router, CapabilityRouter)
+
+    def test_registers_internal_capabilities(self):
+        """SupervisorRuntime should register internal capabilities."""
+        transport = MemoryTransport()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=transport,
+                plugins_dir=Path(temp_dir),
+            )
+
+            # handler.invoke should be registered (but not exposed)
+            assert "handler.invoke" in runtime.capability_router._registrations
+            # Should not be in descriptors (exposed=False)
+            names = [d.name for d in runtime.capability_router.descriptors()]
+            assert "handler.invoke" not in names
+
+
+class TestSupervisorRuntimeMethods:
+    """Tests for SupervisorRuntime methods."""
+
+    @pytest.mark.asyncio
+    async def test_start_with_empty_plugins_dir(self):
+        """SupervisorRuntime.start should work with empty plugins dir."""
+        left, right = make_transport_pair()
+        core = await start_test_core_peer(left)
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=right,
+                plugins_dir=Path(temp_dir),
+                env_manager=FakeEnvManager(),
+            )
+
+            try:
+                await runtime.start()
+                await core.wait_until_remote_initialized()
+                assert runtime.loaded_plugins == []
+                assert runtime.skipped_plugins == {}
+            finally:
+                await runtime.stop()
+                await core.stop()
+
+    @pytest.mark.asyncio
+    async def test_route_handler_invoke_missing_handler(self):
+        """_route_handler_invoke should raise for missing handler."""
+        transport = MemoryTransport()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=transport,
+                plugins_dir=Path(temp_dir),
+            )
+
+            with pytest.raises(AstrBotError, match="handler not found"):
+                await runtime._route_handler_invoke(
+                    "req-1",
+                    {"handler_id": "missing.handler", "event": {}},
+                    CancelToken(),
+                )
+
+    @pytest.mark.asyncio
+    async def test_handle_worker_closed_removes_session(self):
+        """_handle_worker_closed should remove session and handlers."""
+        left, right = make_transport_pair()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=right,
+                plugins_dir=Path(temp_dir),
+                env_manager=FakeEnvManager(),
+            )
+
+            # Add fake session
+            mock_session = MagicMock()
+            mock_session.handlers = [
+                HandlerDescriptor(id="test.handler", trigger=CommandTrigger(command="test"))
+            ]
+
+            runtime.worker_sessions["test_plugin"] = mock_session
+            runtime.handler_to_worker["test.handler"] = mock_session
+            runtime._handler_sources["test.handler"] = "test_plugin"
+            runtime.loaded_plugins.append("test_plugin")
+
+            runtime._handle_worker_closed("test_plugin")
+
+            assert "test_plugin" not in runtime.worker_sessions
+            assert "test.handler" not in runtime.handler_to_worker
+            assert "test.handler" not in runtime._handler_sources
+            assert "test_plugin" not in runtime.loaded_plugins
+
+    @pytest.mark.asyncio
+    async def test_handle_worker_closed_unknown_plugin(self):
+        """_handle_worker_closed should handle unknown plugin."""
+        transport = MemoryTransport()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=transport,
+                plugins_dir=Path(temp_dir),
+            )
+
+            # Should not raise for unknown plugin
+            runtime._handle_worker_closed("unknown_plugin")
+
+
+class TestPluginWorkerRuntimeInit:
+    """Tests for PluginWorkerRuntime initialization."""
+
+    def test_init_with_valid_plugin(self):
+        """PluginWorkerRuntime should initialize with valid plugin."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            transport = MemoryTransport()
+
+            # This should work if the plugin is valid
+            runtime = PluginWorkerRuntime(plugin_dir=plugin_dir, transport=transport)
+
+            assert runtime.plugin.name == "test_plugin"
+            assert runtime.peer is not None
+            assert runtime.dispatcher is not None
+
+
+class TestPluginWorkerRuntimeMethods:
+    """Tests for PluginWorkerRuntime methods."""
+
+    @pytest.mark.asyncio
+    async def test_handle_invoke_wrong_capability(self):
+        """_handle_invoke should raise for wrong capability."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            transport = MemoryTransport()
+            runtime = PluginWorkerRuntime(plugin_dir=plugin_dir, transport=transport)
+
+            message = InvokeMessage(
+                id="invoke-1",
+                capability="wrong.capability",
+                input={},
+            )
+            token = CancelToken()
+
+            with pytest.raises(AstrBotError, match="未找到能力"):
+                await runtime._handle_invoke(message, token)
+
+    @pytest.mark.asyncio
+    async def test_run_lifecycle_sync_hook(self):
+        """_run_lifecycle should call sync hooks."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            transport = MemoryTransport()
+            runtime = PluginWorkerRuntime(plugin_dir=plugin_dir, transport=transport)
+
+            # Add mock instance with sync hook
+            called = []
+
+            class MockInstance:
+                def on_start(self, ctx):
+                    called.append("on_start")
+
+            runtime.loaded_plugin.instances.append(MockInstance())
+
+            await runtime._run_lifecycle("on_start")
+
+            assert "on_start" in called
+
+    @pytest.mark.asyncio
+    async def test_run_lifecycle_async_hook(self):
+        """_run_lifecycle should call async hooks."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            transport = MemoryTransport()
+            runtime = PluginWorkerRuntime(plugin_dir=plugin_dir, transport=transport)
+
+            # Add mock instance with async hook
+            called = []
+
+            class MockInstance:
+                async def on_stop(self, ctx):
+                    called.append("on_stop")
+
+            runtime.loaded_plugin.instances.append(MockInstance())
+
+            await runtime._run_lifecycle("on_stop")
+
+            assert "on_stop" in called
+
+    @pytest.mark.asyncio
+    async def test_run_lifecycle_missing_method(self):
+        """_run_lifecycle should skip missing methods."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            transport = MemoryTransport()
+            runtime = PluginWorkerRuntime(plugin_dir=plugin_dir, transport=transport)
+
+            # Add mock instance without the method
+            class MockInstance:
+                pass
+
+            runtime.loaded_plugin.instances.append(MockInstance())
+
+            # Should not raise
+            await runtime._run_lifecycle("on_start")
+
+
+class TestIntegrationWithTransportPair:
+    """Integration tests using transport pairs."""
+
+    @pytest.mark.asyncio
+    async def test_supervisor_responds_to_initialize(self):
+        """SupervisorRuntime should respond to initialize messages."""
+        left, right = make_transport_pair()
+        core = await start_test_core_peer(left)
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=right,
+                plugins_dir=Path(temp_dir),
+                env_manager=FakeEnvManager(),
+            )
+
+            try:
+                await runtime.start()
+                await core.wait_until_remote_initialized()
+                assert core.remote_peer is not None
+                assert core.remote_peer.name == "astrbot-supervisor"
+                assert core.remote_metadata["plugins"] == []
+                assert core.remote_metadata["skipped_plugins"] == {}
+
+            finally:
+                await runtime.stop()
+                await core.stop()
+
+    @pytest.mark.asyncio
+    async def test_worker_session_lifecycle(self):
+        """WorkerSession should start and stop cleanly."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Create a minimal plugin
+            plugin_dir = Path(temp_dir) / "plugins" / "test_plugin"
+            plugin_dir.mkdir(parents=True)
+
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": f"{sys.version_info.major}.{sys.version_info.minor}"},
+                    "components": [],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            plugin = PluginSpec(
+                name="test_plugin",
+                plugin_dir=plugin_dir,
+                manifest_path=manifest_path,
+                requirements_path=requirements_path,
+                python_version=f"{sys.version_info.major}.{sys.version_info.minor}",
+                manifest_data={"name": "test_plugin"},
+            )
+
+            left, right = make_transport_pair()
+
+            session = WorkerSession(
+                plugin=plugin,
+                repo_root=Path(temp_dir),
+                env_manager=FakeEnvManager(),
+                capability_router=CapabilityRouter(),
+            )
+
+            # Note: Full start would require subprocess, skip for unit test
+            # Just verify the session can be created and stopped
+            await session.stop()
diff --git a/tests_v4/test_handler_dispatcher.py b/tests_v4/test_handler_dispatcher.py
new file mode 100644
index 000000000..a64b50443
--- /dev/null
+++ b/tests_v4/test_handler_dispatcher.py
@@ -0,0 +1,707 @@
+"""
+Tests for runtime/handler_dispatcher.py - HandlerDispatcher implementation.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from astrbot_sdk.context import CancelToken, Context
+from astrbot_sdk.events import MessageEvent, PlainTextResult
+from astrbot_sdk.protocol.descriptors import CommandTrigger, HandlerDescriptor, Permissions
+from astrbot_sdk.protocol.messages import InvokeMessage
+from astrbot_sdk.runtime.handler_dispatcher import HandlerDispatcher
+from astrbot_sdk.runtime.loader import LoadedHandler
+
+
+class MockPeer:
+    """Mock peer for testing."""
+
+    def __init__(self):
+        self.sent_messages: list[dict[str, Any]] = []
+
+
+def create_mock_handler(
+    handler_id: str = "test.handler",
+    command: str = "hello",
+) -> LoadedHandler:
+    """Create a mock loaded handler."""
+    descriptor = HandlerDescriptor(
+        id=handler_id,
+        trigger=CommandTrigger(command=command),
+    )
+
+    async def handler_func(event: MessageEvent, ctx: Context):
+        await event.reply("Hello!")
+        return None
+
+    handler_func.__func__ = handler_func  # Simulate bound method
+
+    return LoadedHandler(
+        descriptor=descriptor,
+        callable=handler_func,
+        owner=MagicMock(),
+        legacy_context=None,
+    )
+
+
+def create_invoke_message(
+    message_id: str = "msg_001",
+    handler_id: str = "test.handler",
+    event_data: dict[str, Any] | None = None,
+    args: dict[str, Any] | None = None,
+) -> InvokeMessage:
+    """Create a mock invoke message."""
+    input_data = {"handler_id": handler_id, "event": event_data or {}}
+    if args:
+        input_data["args"] = args
+    return InvokeMessage(
+        id=message_id,
+        capability="handler.invoke",
+        input=input_data,
+    )
+
+
+def create_message_event() -> MessageEvent:
+    """Create a mock message event."""
+    return MessageEvent(
+        session_id="session-1",
+        user_id="user-1",
+        platform="test",
+        text="hello world",
+    )
+
+
+class TestHandlerDispatcherInit:
+    """Tests for HandlerDispatcher initialization."""
+
+    def test_init(self):
+        """HandlerDispatcher should initialize with handlers."""
+        peer = MockPeer()
+        handler = create_mock_handler()
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        assert dispatcher._plugin_id == "test_plugin"
+        assert dispatcher._peer is peer
+        assert "test.handler" in dispatcher._handlers
+        assert dispatcher._active == {}
+
+    def test_handlers_indexed_by_id(self):
+        """HandlerDispatcher should index handlers by id."""
+        peer = MockPeer()
+        handlers = [
+            create_mock_handler("handler.one", "cmd1"),
+            create_mock_handler("handler.two", "cmd2"),
+        ]
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=handlers,
+        )
+
+        assert "handler.one" in dispatcher._handlers
+        assert "handler.two" in dispatcher._handlers
+
+
+class TestHandlerDispatcherInvoke:
+    """Tests for HandlerDispatcher.invoke method."""
+
+    @pytest.mark.asyncio
+    async def test_invoke_calls_handler(self):
+        """invoke should call the registered handler."""
+        peer = MockPeer()
+        reply_called = []
+        event = create_message_event()
+        event._reply_handler = lambda text: reply_called.append(text)
+
+        handler_called = []
+
+        async def handler_func(e: MessageEvent, ctx: Context):
+            handler_called.append(e)
+            await e.reply("response")
+
+        descriptor = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=handler_func,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        message = InvokeMessage(
+            id="msg_001",
+            capability="handler.invoke",
+            input={"handler_id": "test.handler", "event": event.model_dump()},
+        )
+
+        cancel_token = CancelToken()
+        result = await dispatcher.invoke(message, cancel_token)
+
+        assert result == {}
+        assert len(handler_called) == 1
+        assert "response" in reply_called
+
+    @pytest.mark.asyncio
+    async def test_invoke_missing_handler_raises(self):
+        """invoke should raise LookupError for missing handler."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        message = InvokeMessage(
+            id="msg_001",
+            capability="handler.invoke",
+            input={"handler_id": "nonexistent.handler", "event": {}},
+        )
+
+        cancel_token = CancelToken()
+
+        with pytest.raises(LookupError, match="handler not found"):
+            await dispatcher.invoke(message, cancel_token)
+
+    @pytest.mark.asyncio
+    async def test_invoke_with_legacy_args(self):
+        """invoke should pass legacy args to handler."""
+        peer = MockPeer()
+
+        received_args = []
+
+        async def handler_func(event: MessageEvent, ctx: Context, name: str):
+            received_args.append(name)
+            return None
+
+        descriptor = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=handler_func,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        message = InvokeMessage(
+            id="msg_001",
+            capability="handler.invoke",
+            input={
+                "handler_id": "test.handler",
+                "event": {
+                    "type": "message",
+                    "session_id": "s1",
+                    "user_id": "u1",
+                    "platform": "test",
+                },
+                "args": {"name": "test_name"},
+            },
+        )
+
+        cancel_token = CancelToken()
+        await dispatcher.invoke(message, cancel_token)
+
+        assert "test_name" in received_args
+
+    @pytest.mark.asyncio
+    async def test_invoke_tracks_active_task(self):
+        """invoke should track active task."""
+        peer = MockPeer()
+
+        async def slow_handler(event: MessageEvent, ctx: Context):
+            await asyncio.sleep(0.1)
+            return None
+
+        descriptor = HandlerDescriptor(
+            id="slow.handler",
+            trigger=CommandTrigger(command="slow"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=slow_handler,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        message = InvokeMessage(
+            id="msg_001",
+            capability="handler.invoke",
+            input={
+                "handler_id": "slow.handler",
+                "event": {
+                    "type": "message",
+                    "session_id": "s1",
+                    "user_id": "u1",
+                    "platform": "test",
+                },
+            },
+        )
+
+        cancel_token = CancelToken()
+
+        # Start invoke in background
+        task = asyncio.create_task(dispatcher.invoke(message, cancel_token))
+
+        # Give it time to start
+        await asyncio.sleep(0)
+
+        # Should have active task during execution
+        # Note: might be empty if task completes quickly
+
+        await task
+
+        # After completion, should be cleared
+        assert "msg_001" not in dispatcher._active
+
+
+class TestHandlerDispatcherCancel:
+    """Tests for HandlerDispatcher.cancel method."""
+
+    @pytest.mark.asyncio
+    async def test_cancel_stops_active_task(self):
+        """cancel should stop the active task."""
+        peer = MockPeer()
+
+        cancelled = []
+
+        async def slow_handler(event: MessageEvent, ctx: Context):
+            try:
+                await asyncio.sleep(10)  # Long sleep
+            except asyncio.CancelledError:
+                cancelled.append(True)
+                raise
+
+        descriptor = HandlerDescriptor(
+            id="slow.handler",
+            trigger=CommandTrigger(command="slow"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=slow_handler,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        message = InvokeMessage(
+            id="msg_001",
+            capability="handler.invoke",
+            input={
+                "handler_id": "slow.handler",
+                "event": {
+                    "type": "message",
+                    "session_id": "s1",
+                    "user_id": "u1",
+                    "platform": "test",
+                },
+            },
+        )
+
+        cancel_token = CancelToken()
+
+        # Start invoke
+        task = asyncio.create_task(dispatcher.invoke(message, cancel_token))
+
+        # Wait for task to be active
+        await asyncio.sleep(0.05)
+
+        # Cancel
+        await dispatcher.cancel("msg_001")
+
+        # Task should be cancelled
+        await asyncio.sleep(0.05)
+
+        assert cancelled or task.cancelled() or task.done()
+
+    @pytest.mark.asyncio
+    async def test_cancel_unknown_request_does_nothing(self):
+        """cancel should do nothing for unknown request."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        # Should not raise
+        await dispatcher.cancel("unknown_request")
+
+
+class TestHandlerDispatcherBuildArgs:
+    """Tests for HandlerDispatcher._build_args method."""
+
+    def test_build_args_event_parameter(self):
+        """_build_args should inject event parameter."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        def handler(event: MessageEvent, ctx: Context):
+            pass
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        args = dispatcher._build_args(handler, event, ctx)
+
+        assert args[0] is event
+
+    def test_build_args_ctx_parameter(self):
+        """_build_args should inject ctx/context parameter."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        def handler(event: MessageEvent, ctx: Context):
+            pass
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        args = dispatcher._build_args(handler, event, ctx)
+
+        assert args[1] is ctx
+
+    def test_build_args_legacy_args(self):
+        """_build_args should inject legacy args."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        def handler(event: MessageEvent, ctx: Context, custom_arg: str):
+            pass
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        args = dispatcher._build_args(handler, event, ctx, {"custom_arg": "value"})
+
+        assert args[2] == "value"
+
+    def test_build_args_skip_keyword_only(self):
+        """_build_args should skip keyword-only parameters."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        def handler(event: MessageEvent, *, optional: str = "default"):
+            pass
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        args = dispatcher._build_args(handler, event, ctx)
+
+        # Should only have event
+        assert len(args) == 1
+
+
+class TestHandlerDispatcherConsumeResult:
+    """Tests for HandlerDispatcher._consume_legacy_result method."""
+
+    @pytest.mark.asyncio
+    async def test_consume_plain_text_result(self):
+        """_consume_legacy_result should handle PlainTextResult."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        replies = []
+        event = create_message_event()
+        event._reply_handler = lambda text: replies.append(text)
+
+        result = PlainTextResult(text="plain text")
+        await dispatcher._consume_legacy_result(result, event)
+
+        assert "plain text" in replies
+
+    @pytest.mark.asyncio
+    async def test_consume_string(self):
+        """_consume_legacy_result should handle string."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        replies = []
+        event = create_message_event()
+        event._reply_handler = lambda text: replies.append(text)
+
+        await dispatcher._consume_legacy_result("string reply", event)
+
+        assert "string reply" in replies
+
+    @pytest.mark.asyncio
+    async def test_consume_dict_with_text(self):
+        """_consume_legacy_result should handle dict with text."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        replies = []
+        event = create_message_event()
+        event._reply_handler = lambda text: replies.append(text)
+
+        await dispatcher._consume_legacy_result({"text": "dict reply"}, event)
+
+        assert "dict reply" in replies
+
+    @pytest.mark.asyncio
+    async def test_consume_other_type_ignored(self):
+        """_consume_legacy_result should ignore other types."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        event = create_message_event()
+        event._reply_handler = MagicMock()
+
+        # Should not raise
+        await dispatcher._consume_legacy_result(123, event)
+        await dispatcher._consume_legacy_result(None, event)
+
+
+class TestHandlerDispatcherHandleError:
+    """Tests for HandlerDispatcher._handle_error method."""
+
+    @pytest.mark.asyncio
+    async def test_handle_error_with_on_error_method(self):
+        """_handle_error should call owner.on_error if available."""
+        peer = MockPeer()
+
+        errors_handled = []
+
+        class OwnerWithOnError:
+            async def on_error(self, exc: Exception, event: MessageEvent, ctx: Context):
+                errors_handled.append(exc)
+
+        owner = OwnerWithOnError()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+        exc = ValueError("test error")
+
+        await dispatcher._handle_error(owner, exc, event, ctx)
+
+        assert exc in errors_handled
+
+    @pytest.mark.asyncio
+    async def test_handle_error_without_on_error_method(self):
+        """_handle_error should use Star.on_error if owner has no on_error."""
+        peer = MockPeer()
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[],
+        )
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+        exc = ValueError("test error")
+
+        # Should not raise
+        await dispatcher._handle_error(MagicMock(), exc, event, ctx)
+
+
+class TestHandlerDispatcherRunHandler:
+    """Tests for HandlerDispatcher._run_handler method."""
+
+    @pytest.mark.asyncio
+    async def test_run_handler_sync_function(self):
+        """_run_handler should handle sync function."""
+        peer = MockPeer()
+
+        called = []
+
+        def sync_handler(event: MessageEvent, ctx: Context):
+            called.append(True)
+
+        descriptor = HandlerDescriptor(
+            id="sync.handler",
+            trigger=CommandTrigger(command="sync"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=sync_handler,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        await dispatcher._run_handler(handler, event, ctx)
+
+        assert called
+
+    @pytest.mark.asyncio
+    async def test_run_handler_async_function(self):
+        """_run_handler should handle async function."""
+        peer = MockPeer()
+
+        called = []
+
+        async def async_handler(event: MessageEvent, ctx: Context):
+            called.append(True)
+
+        descriptor = HandlerDescriptor(
+            id="async.handler",
+            trigger=CommandTrigger(command="async"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=async_handler,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        await dispatcher._run_handler(handler, event, ctx)
+
+        assert called
+
+    @pytest.mark.asyncio
+    async def test_run_handler_async_generator(self):
+        """_run_handler should handle async generator."""
+        peer = MockPeer()
+
+        replies = []
+
+        async def gen_handler(event: MessageEvent, ctx: Context):
+            yield "first"
+            yield "second"
+
+        descriptor = HandlerDescriptor(
+            id="gen.handler",
+            trigger=CommandTrigger(command="gen"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=gen_handler,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        event = create_message_event()
+        event._reply_handler = lambda text: replies.append(text)
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        await dispatcher._run_handler(handler, event, ctx)
+
+        assert "first" in replies
+        assert "second" in replies
+
+    @pytest.mark.asyncio
+    async def test_run_handler_with_exception(self):
+        """_run_handler should handle exceptions."""
+        peer = MockPeer()
+
+        async def failing_handler(event: MessageEvent, ctx: Context):
+            raise ValueError("handler error")
+
+        descriptor = HandlerDescriptor(
+            id="failing.handler",
+            trigger=CommandTrigger(command="fail"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=failing_handler,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        event = create_message_event()
+        ctx = Context(peer=peer, plugin_id="test", cancel_token=CancelToken())
+
+        with pytest.raises(ValueError, match="handler error"):
+            await dispatcher._run_handler(handler, event, ctx)
diff --git a/tests_v4/test_loader.py b/tests_v4/test_loader.py
new file mode 100644
index 000000000..e72fac1c2
--- /dev/null
+++ b/tests_v4/test_loader.py
@@ -0,0 +1,622 @@
+"""
+Tests for runtime/loader.py - Plugin loading utilities.
+"""
+from __future__ import annotations
+
+import sys
+import tempfile
+import textwrap
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import yaml
+
+from astrbot_sdk.decorators import HandlerMeta
+from astrbot_sdk.protocol.descriptors import CommandTrigger, HandlerDescriptor
+from astrbot_sdk.runtime.loader import (
+    LoadedHandler,
+    LoadedPlugin,
+    PluginDiscoveryResult,
+    PluginEnvironmentManager,
+    PluginSpec,
+    STATE_FILE_NAME,
+    _create_legacy_context,
+    _is_new_star_component,
+    _iter_handler_names,
+    _venv_python_path,
+    discover_plugins,
+    import_string,
+    load_plugin,
+    load_plugin_spec,
+)
+
+
+class TestVenvPythonPath:
+    """Tests for _venv_python_path function."""
+
+    def test_linux_path(self):
+        """_venv_python_path should return correct Linux path."""
+        with patch("os.name", "posix"):
+            path = _venv_python_path(Path("/home/user/.venv"))
+            assert path == Path("/home/user/.venv/bin/python")
+
+    def test_windows_path(self):
+        """_venv_python_path should return correct Windows path."""
+        with patch("os.name", "nt"):
+            path = _venv_python_path(Path("C:\\venv"))
+            assert path == Path("C:\\venv\\Scripts\\python.exe")
+
+
+class TestPluginSpec:
+    """Tests for PluginSpec dataclass."""
+
+    def test_init(self):
+        """PluginSpec should store all fields."""
+        plugin_dir = Path("/tmp/plugin")
+        manifest_path = plugin_dir / "plugin.yaml"
+        requirements_path = plugin_dir / "requirements.txt"
+
+        spec = PluginSpec(
+            name="test_plugin",
+            plugin_dir=plugin_dir,
+            manifest_path=manifest_path,
+            requirements_path=requirements_path,
+            python_version="3.12",
+            manifest_data={"name": "test_plugin"},
+        )
+
+        assert spec.name == "test_plugin"
+        assert spec.plugin_dir == plugin_dir
+        assert spec.manifest_path == manifest_path
+        assert spec.requirements_path == requirements_path
+        assert spec.python_version == "3.12"
+        assert spec.manifest_data == {"name": "test_plugin"}
+
+
+class TestPluginDiscoveryResult:
+    """Tests for PluginDiscoveryResult dataclass."""
+
+    def test_init(self):
+        """PluginDiscoveryResult should store plugins and skipped."""
+        spec = PluginSpec(
+            name="test",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+
+        result = PluginDiscoveryResult(
+            plugins=[spec],
+            skipped_plugins={"bad_plugin": "missing requirements.txt"},
+        )
+
+        assert len(result.plugins) == 1
+        assert result.skipped_plugins == {"bad_plugin": "missing requirements.txt"}
+
+
+class TestLoadedHandler:
+    """Tests for LoadedHandler dataclass."""
+
+    def test_init(self):
+        """LoadedHandler should store all fields."""
+        descriptor = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+
+        def handler_func():
+            pass
+
+        owner = MagicMock()
+
+        loaded = LoadedHandler(
+            descriptor=descriptor,
+            callable=handler_func,
+            owner=owner,
+            legacy_context=None,
+        )
+
+        assert loaded.descriptor == descriptor
+        assert loaded.callable == handler_func
+        assert loaded.owner == owner
+        assert loaded.legacy_context is None
+
+
+class TestLoadedPlugin:
+    """Tests for LoadedPlugin dataclass."""
+
+    def test_init(self):
+        """LoadedPlugin should store plugin and handlers."""
+        spec = PluginSpec(
+            name="test",
+            plugin_dir=Path("/tmp"),
+            manifest_path=Path("/tmp/plugin.yaml"),
+            requirements_path=Path("/tmp/requirements.txt"),
+            python_version="3.12",
+            manifest_data={},
+        )
+
+        loaded = LoadedPlugin(plugin=spec, handlers=[], instances=[])
+
+        assert loaded.plugin == spec
+        assert loaded.handlers == []
+        assert loaded.instances == []
+
+
+class TestIsNewStarComponent:
+    """Tests for _is_new_star_component function."""
+
+    def test_non_class_returns_false(self):
+        """_is_new_star_component should return False for non-class."""
+        assert _is_new_star_component("not a class") is False
+        assert _is_new_star_component(123) is False
+
+    def test_non_star_subclass_returns_false(self):
+        """_is_new_star_component should return False for non-Star class."""
+        class NotAStar:
+            pass
+
+        assert _is_new_star_component(NotAStar) is False
+
+    def test_star_without_marker_returns_true(self):
+        """_is_new_star_component should return True for Star without marker."""
+        from astrbot_sdk.star import Star
+
+        class MyStar(Star):
+            pass
+
+        assert _is_new_star_component(MyStar) is True
+
+    def test_star_with_false_marker_returns_false(self):
+        """_is_new_star_component should return False if marker returns False."""
+        from astrbot_sdk.star import Star
+
+        class LegacyStar(Star):
+            @classmethod
+            def __astrbot_is_new_star__(cls):
+                return False
+
+        assert _is_new_star_component(LegacyStar) is False
+
+
+class TestCreateLegacyContext:
+    """Tests for _create_legacy_context function."""
+
+    def test_with_factory_method(self):
+        """_create_legacy_context should use factory method if available."""
+        mock_context = MagicMock()
+
+        class ComponentWithFactory:
+            @classmethod
+            def _astrbot_create_legacy_context(cls, plugin_name):
+                return mock_context
+
+        result = _create_legacy_context(ComponentWithFactory, "test_plugin")
+        assert result == mock_context
+
+    def test_without_factory_method(self):
+        """_create_legacy_context should create default context."""
+        # Without factory, it imports LegacyContext
+        from astrbot_sdk.star import Star
+
+        class PlainStar(Star):
+            pass
+
+        result = _create_legacy_context(PlainStar, "test_plugin")
+        # Should return some context object
+        assert result is not None
+
+
+class TestIterHandlerNames:
+    """Tests for _iter_handler_names function."""
+
+    def test_with_handlers_attribute(self):
+        """_iter_handler_names should use __handlers__ if available."""
+        instance = MagicMock()
+        instance.__class__.__handlers__ = ("handler1", "handler2")
+
+        names = _iter_handler_names(instance)
+        assert names == ["handler1", "handler2"]
+
+    def test_without_handlers_attribute(self):
+        """_iter_handler_names should fall back to dir() if no __handlers__."""
+        instance = MagicMock()
+        # Remove __handlers__ attribute
+        del instance.__class__.__handlers__
+
+        # Mock dir to return specific names
+        with patch.object(instance, "__dir__", return_value=["method1", "method2"]):
+            names = _iter_handler_names(instance)
+            assert "method1" in names or "method2" in names
+
+
+class TestLoadPluginSpec:
+    """Tests for load_plugin_spec function."""
+
+    def test_loads_manifest(self):
+        """load_plugin_spec should load plugin.yaml."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.11"},
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            spec = load_plugin_spec(plugin_dir)
+
+            assert spec.name == "test_plugin"
+            assert spec.python_version == "3.11"
+            assert spec.plugin_dir.resolve() == plugin_dir.resolve()
+
+    def test_defaults_python_version(self):
+        """load_plugin_spec should default python version to current."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text(
+                yaml.dump({"name": "test_plugin"}),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            spec = load_plugin_spec(plugin_dir)
+
+            expected = f"{sys.version_info.major}.{sys.version_info.minor}"
+            assert spec.python_version == expected
+
+    def test_defaults_name_to_dir_name(self):
+        """load_plugin_spec should default name to directory name."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir) / "my_plugin"
+            plugin_dir.mkdir()
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            manifest_path.write_text("{}", encoding="utf-8")
+            requirements_path.write_text("", encoding="utf-8")
+
+            spec = load_plugin_spec(plugin_dir)
+
+            assert spec.name == "my_plugin"
+
+
+class TestDiscoverPlugins:
+    """Tests for discover_plugins function."""
+
+    def test_empty_directory(self):
+        """discover_plugins should return empty for non-existent directory."""
+        result = discover_plugins(Path("/nonexistent"))
+        assert result.plugins == []
+        assert result.skipped_plugins == {}
+
+    def test_skips_dot_directories(self):
+        """discover_plugins should skip directories starting with dot."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            # Create .hidden directory
+            hidden_dir = plugins_dir / ".hidden"
+            hidden_dir.mkdir()
+            (hidden_dir / "plugin.yaml").write_text(
+                yaml.dump({
+                    "name": "hidden",
+                    "runtime": {"python": "3.12"},
+                    "components": [{"class": "test:Test"}],
+                }),
+                encoding="utf-8",
+            )
+            (hidden_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            assert result.plugins == []
+
+    def test_skips_missing_manifest(self):
+        """discover_plugins should skip directories without plugin.yaml."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            plugin_dir = plugins_dir / "no_manifest"
+            plugin_dir.mkdir()
+            (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            assert result.plugins == []
+
+    def test_skips_missing_requirements(self):
+        """discover_plugins should skip directories without requirements.txt."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            plugin_dir = plugins_dir / "no_requirements"
+            plugin_dir.mkdir()
+            (plugin_dir / "plugin.yaml").write_text(
+                yaml.dump({"name": "test"}),
+                encoding="utf-8",
+            )
+
+            result = discover_plugins(plugins_dir)
+
+            assert "no_requirements" in result.skipped_plugins
+            assert "requirements.txt" in result.skipped_plugins["no_requirements"]
+
+    def test_validates_required_fields(self):
+        """discover_plugins should validate required manifest fields."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            # Missing name
+            plugin_dir = plugins_dir / "missing_name"
+            plugin_dir.mkdir()
+            (plugin_dir / "plugin.yaml").write_text(
+                yaml.dump({
+                    "runtime": {"python": "3.12"},
+                    "components": [{"class": "test:Test"}],
+                }),
+                encoding="utf-8",
+            )
+            (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            assert "missing_name" in result.skipped_plugins
+            assert "name" in result.skipped_plugins["missing_name"]
+
+    def test_detects_duplicate_names(self):
+        """discover_plugins should detect duplicate plugin names."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            for i, dirname in enumerate(["plugin1", "plugin2"]):
+                plugin_dir = plugins_dir / dirname
+                plugin_dir.mkdir()
+                (plugin_dir / "plugin.yaml").write_text(
+                    yaml.dump({
+                        "name": "duplicate_name",  # Same name
+                        "runtime": {"python": "3.12"},
+                        "components": [{"class": "test:Test"}],
+                    }),
+                    encoding="utf-8",
+                )
+                (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            # First one should succeed, second should be skipped
+            assert len(result.plugins) == 1
+            assert "duplicate_name" in result.skipped_plugins
+
+    def test_validates_components_list(self):
+        """discover_plugins should validate components is a non-empty list."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            plugin_dir = plugins_dir / "bad_components"
+            plugin_dir.mkdir()
+            (plugin_dir / "plugin.yaml").write_text(
+                yaml.dump({
+                    "name": "test",
+                    "runtime": {"python": "3.12"},
+                    "components": "not_a_list",
+                }),
+                encoding="utf-8",
+            )
+            (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            assert "test" in result.skipped_plugins
+            assert "components" in result.skipped_plugins["test"]
+
+    def test_discovers_valid_plugin(self):
+        """discover_plugins should discover valid plugin."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            plugin_dir = plugins_dir / "valid_plugin"
+            plugin_dir.mkdir()
+            (plugin_dir / "plugin.yaml").write_text(
+                yaml.dump({
+                    "name": "valid_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [{"class": "module:Class"}],
+                }),
+                encoding="utf-8",
+            )
+            (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            assert len(result.plugins) == 1
+            assert result.plugins[0].name == "valid_plugin"
+
+
+class TestPluginEnvironmentManager:
+    """Tests for PluginEnvironmentManager class."""
+
+    def test_init(self):
+        """PluginEnvironmentManager should initialize with repo root."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            manager = PluginEnvironmentManager(Path(temp_dir))
+            assert manager.repo_root == Path(temp_dir).resolve()
+            assert manager.cache_dir == Path(temp_dir).resolve() / ".uv-cache"
+
+    def test_uv_binary_detection(self):
+        """PluginEnvironmentManager should detect uv binary."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            with patch("shutil.which", return_value="/usr/bin/uv"):
+                manager = PluginEnvironmentManager(Path(temp_dir))
+                assert manager.uv_binary == "/usr/bin/uv"
+
+    def test_prepare_environment_without_uv_raises(self):
+        """prepare_environment should raise if uv not found."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            manager = PluginEnvironmentManager(Path(temp_dir), uv_binary=None)
+
+            spec = PluginSpec(
+                name="test",
+                plugin_dir=Path(temp_dir),
+                manifest_path=Path(temp_dir) / "plugin.yaml",
+                requirements_path=Path(temp_dir) / "requirements.txt",
+                python_version="3.12",
+                manifest_data={},
+            )
+
+            with pytest.raises(RuntimeError, match="uv"):
+                manager.prepare_environment(spec)
+
+    def test_fingerprint(self):
+        """_fingerprint should create consistent fingerprint."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            requirements = plugin_dir / "requirements.txt"
+            requirements.write_text("astrbot-sdk\n", encoding="utf-8")
+
+            spec = PluginSpec(
+                name="test",
+                plugin_dir=plugin_dir,
+                manifest_path=plugin_dir / "plugin.yaml",
+                requirements_path=requirements,
+                python_version="3.12",
+                manifest_data={},
+            )
+
+            fingerprint = PluginEnvironmentManager._fingerprint(spec)
+
+            assert "python_version" in fingerprint
+            assert "3.12" in fingerprint
+            assert "requirements" in fingerprint
+
+    def test_load_state_missing_file(self):
+        """_load_state should return empty dict for missing file."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            state = PluginEnvironmentManager._load_state(Path(temp_dir) / "missing.json")
+            assert state == {}
+
+    def test_load_state_invalid_json(self):
+        """_load_state should return empty dict for invalid JSON."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            state_path = Path(temp_dir) / "state.json"
+            state_path.write_text("not valid json", encoding="utf-8")
+
+            state = PluginEnvironmentManager._load_state(state_path)
+            assert state == {}
+
+    def test_write_state(self):
+        """_write_state should write state file."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            state_path = Path(temp_dir) / "state.json"
+            spec = PluginSpec(
+                name="test",
+                plugin_dir=Path(temp_dir),
+                manifest_path=Path(temp_dir) / "plugin.yaml",
+                requirements_path=Path(temp_dir) / "requirements.txt",
+                python_version="3.12",
+                manifest_data={},
+            )
+
+            PluginEnvironmentManager._write_state(state_path, spec, "test_fingerprint")
+
+            import json
+            state = json.loads(state_path.read_text(encoding="utf-8"))
+
+            assert state["plugin"] == "test"
+            assert state["fingerprint"] == "test_fingerprint"
+
+
+class TestImportString:
+    """Tests for import_string function."""
+
+    def test_imports_module_attribute(self):
+        """import_string should import module and get attribute."""
+        result = import_string("os:path")
+        assert result is not None
+
+    def test_raises_for_missing_module(self):
+        """import_string should raise for missing module."""
+        with pytest.raises(ImportError):
+            import_string("nonexistent_module:attr")
+
+    def test_raises_for_missing_attribute(self):
+        """import_string should raise for missing attribute."""
+        with pytest.raises(AttributeError):
+            import_string("os:nonexistent_attr")
+
+    def test_raises_for_invalid_format(self):
+        """import_string should raise for invalid format."""
+        with pytest.raises(ValueError):
+            import_string("no_colon")
+
+
+class TestLoadPlugin:
+    """Tests for load_plugin function."""
+
+    def test_loads_component_and_handlers(self):
+        """load_plugin should load component class and find handlers."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            # Create module
+            module_dir = plugin_dir / "mymodule"
+            module_dir.mkdir()
+            (module_dir / "__init__.py").write_text("", encoding="utf-8")
+            (module_dir / "component.py").write_text(
+                textwrap.dedent("""
+                    from astrbot_sdk import Star, on_command
+
+                    class MyComponent(Star):
+                        @on_command("hello")
+                        async def hello_handler(self):
+                            pass
+                """),
+                encoding="utf-8",
+            )
+
+            manifest_path.write_text(
+                yaml.dump({
+                    "name": "test_plugin",
+                    "runtime": {"python": "3.12"},
+                    "components": [{"class": "mymodule.component:MyComponent"}],
+                }),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            spec = load_plugin_spec(plugin_dir)
+
+            # Add plugin dir to sys.path for import
+            if str(plugin_dir) not in sys.path:
+                sys.path.insert(0, str(plugin_dir))
+
+            try:
+                loaded = load_plugin(spec)
+
+                assert loaded.plugin.name == "test_plugin"
+                assert len(loaded.instances) == 1
+                assert len(loaded.handlers) >= 1
+            finally:
+                if str(plugin_dir) in sys.path:
+                    sys.path.remove(str(plugin_dir))
+
+
+class TestStateFileConstant:
+    """Tests for STATE_FILE_NAME constant."""
+
+    def test_value(self):
+        """STATE_FILE_NAME should be correct."""
+        assert STATE_FILE_NAME == ".astrbot-worker-state.json"