diff --git a/AGENTS.md b/AGENTS.md
index 002f550fb..add07dc4e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -4,6 +4,7 @@
 - 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
 - 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
 - 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
+- 2026-03-13: Transport-pair startup tests for `SupervisorRuntime` must start a real peer on the opposite transport and provide an `initialize` response. Wiring only the supervisor side drops or captures the outgoing initialize message without replying, and `Peer.initialize()` then waits forever.
 
 # 开发命令
 
@@ -19,10 +20,12 @@ ruff check . --fix # 使用 ruff 检查并自动修复问题
 
 ## 测试
 
+如果修改了内容可能影响现有功能，请运行测试以确保没有引入错误：
+如果修改了bug或者更改了功能需要添加新的测试
+
 ```bash
 python run_tests.py            # 运行所有测试
 python run_tests.py -v         # 详细输出
 python run_tests.py -k "test_peer"  # 运行匹配模式的测试
 python run_tests.py --cov      # 运行测试并生成覆盖率报告
 ```
-
diff --git a/CLAUDE.md b/CLAUDE.md
index 002f550fb..add07dc4e 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -4,6 +4,7 @@
 - 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
 - 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
 - 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
+- 2026-03-13: Transport-pair startup tests for `SupervisorRuntime` must start a real peer on the opposite transport and provide an `initialize` response. Wiring only the supervisor side drops or captures the outgoing initialize message without replying, and `Peer.initialize()` then waits forever.
 
 # 开发命令
 
@@ -19,10 +20,12 @@ ruff check . --fix # 使用 ruff 检查并自动修复问题
 
 ## 测试
 
+如果修改了内容可能影响现有功能，请运行测试以确保没有引入错误：
+如果修改了bug或者更改了功能需要添加新的测试
+
 ```bash
 python run_tests.py            # 运行所有测试
 python run_tests.py -v         # 详细输出
 python run_tests.py -k "test_peer"  # 运行匹配模式的测试
 python run_tests.py --cov      # 运行测试并生成覆盖率报告
 ```
-
diff --git a/src-new/astrbot_sdk/_legacy_api.py b/src-new/astrbot_sdk/_legacy_api.py
index 6fbfafe25..89e20fea4 100644
--- a/src-new/astrbot_sdk/_legacy_api.py
+++ b/src-new/astrbot_sdk/_legacy_api.py
@@ -9,6 +9,7 @@ from .clients.llm import LLMResponse
 from .context import Context as NewContext
 from .star import Star
 
+# TODO-迁移文档要写，我好烦烦烦你烦烦烦你
 MIGRATION_DOC_URL = "https://docs.astrbot.app/migration/v3"
 _warned_methods: set[str] = set()
 
@@ -26,13 +27,33 @@ def _warn_once(old_name: str, replacement: str) -> None:
 
 
 class LegacyConversationManager:
+    """旧版会话管理器的兼容实现。
+
+    使用 db 存储会话数据，key 为 `__compat_conversations__`。
+
+    注意：此实现不提供持久化保证，会话数据仅在当前运行时有效。
+    """
+
     def __init__(self, parent: "LegacyContext") -> None:
         self._parent = parent
         self._counters: defaultdict[str, int] = defaultdict(int)
+        # 记录每个 unified_msg_origin 的当前会话 ID
+        self._current_conversations: dict[str, str] = {}
 
     def _ctx(self) -> NewContext:
         return self._parent.require_runtime_context()
 
+    async def _get_stored(self) -> dict[str, dict[str, Any]]:
+        """获取存储的所有会话数据。"""
+        ctx = self._ctx()
+        stored = await ctx.db.get("__compat_conversations__")
+        return stored if isinstance(stored, dict) else {}
+
+    async def _set_stored(self, stored: dict[str, dict[str, Any]]) -> None:
+        """保存会话数据。"""
+        ctx = self._ctx()
+        await ctx.db.set("__compat_conversations__", stored)
+
     async def new_conversation(
         self,
         unified_msg_origin: str,
@@ -41,10 +62,11 @@ class LegacyConversationManager:
         title: str | None = None,
         persona_id: str | None = None,
     ) -> str:
+        """创建新会话并返回会话 ID。"""
         ctx = self._ctx()
         self._counters[unified_msg_origin] += 1
         conversation_id = f"{ctx.plugin_id}-conv-{self._counters[unified_msg_origin]}"
-        stored = await ctx.db.get("__compat_conversations__") or {}
+        stored = await self._get_stored()
         stored[conversation_id] = {
             "unified_msg_origin": unified_msg_origin,
             "platform_id": platform_id,
@@ -52,9 +74,277 @@ class LegacyConversationManager:
             "title": title,
             "persona_id": persona_id,
         }
-        await ctx.db.set("__compat_conversations__", stored)
+        await self._set_stored(stored)
+        # 设置为当前会话
+        self._current_conversations[unified_msg_origin] = conversation_id
         return conversation_id
 
+    async def switch_conversation(
+        self, unified_msg_origin: str, conversation_id: str
+    ) -> None:
+        """切换到指定会话。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+            conversation_id: 要切换到的会话 ID
+        """
+        stored = await self._get_stored()
+        if conversation_id not in stored:
+            return
+        # 验证会话属于该 unified_msg_origin
+        conv_data = stored[conversation_id]
+        if conv_data.get("unified_msg_origin") != unified_msg_origin:
+            return
+        self._current_conversations[unified_msg_origin] = conversation_id
+
+    async def delete_conversation(
+        self,
+        unified_msg_origin: str,
+        conversation_id: str | None = None,
+    ) -> None:
+        """删除指定会话。
+
+        当 conversation_id 为 None 时，删除当前会话。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+            conversation_id: 要删除的会话 ID，为 None 时删除当前会话
+        """
+        # 如果 conversation_id 为 None，使用当前会话
+        if conversation_id is None:
+            conversation_id = self._current_conversations.get(unified_msg_origin)
+            if conversation_id is None:
+                return
+
+        stored = await self._get_stored()
+        if conversation_id not in stored:
+            return
+        conv_data = stored[conversation_id]
+        if conv_data.get("unified_msg_origin") != unified_msg_origin:
+            return
+        del stored[conversation_id]
+        await self._set_stored(stored)
+        # 如果删除的是当前会话，清除当前会话记录
+        if self._current_conversations.get(unified_msg_origin) == conversation_id:
+            del self._current_conversations[unified_msg_origin]
+
+    async def get_curr_conversation_id(self, unified_msg_origin: str) -> str | None:
+        """获取当前会话 ID。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+
+        Returns:
+            当前会话 ID，若无则返回 None
+        """
+        return self._current_conversations.get(unified_msg_origin)
+
+    async def get_conversation(
+        self,
+        unified_msg_origin: str,
+        conversation_id: str,
+        create_if_not_exists: bool = False,
+    ) -> dict[str, Any] | None:
+        """获取指定会话的数据。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+            conversation_id: 会话 ID
+            create_if_not_exists: 如果会话不存在，是否创建新会话
+
+        Returns:
+            会话数据字典，不存在则返回 None
+        """
+        stored = await self._get_stored()
+        conv = stored.get(conversation_id)
+        if conv is None and create_if_not_exists:
+            # 创建新会话
+            conv = {
+                "unified_msg_origin": unified_msg_origin,
+                "platform_id": None,
+                "content": [],
+                "title": None,
+                "persona_id": None,
+            }
+            stored[conversation_id] = conv
+            await self._set_stored(stored)
+            self._current_conversations[unified_msg_origin] = conversation_id
+        return conv
+
+    async def get_conversations(
+        self,
+        unified_msg_origin: str | None = None,
+        platform_id: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """获取会话列表。
+
+        Args:
+            unified_msg_origin: 统一消息来源，可选
+            platform_id: 平台 ID，可选
+
+        Returns:
+            会话列表，每个元素包含 conversation_id 和会话数据
+        """
+        stored = await self._get_stored()
+        result = []
+        for conv_id, conv_data in stored.items():
+            # 按 unified_msg_origin 过滤
+            if unified_msg_origin is not None:
+                if conv_data.get("unified_msg_origin") != unified_msg_origin:
+                    continue
+            # 按 platform_id 过滤
+            if platform_id is not None:
+                if conv_data.get("platform_id") != platform_id:
+                    continue
+            result.append({"conversation_id": conv_id, **conv_data})
+        return result
+
+    async def update_conversation(
+        self,
+        unified_msg_origin: str,
+        conversation_id: str | None = None,
+        history: list[dict] | None = None,
+        title: str | None = None,
+        persona_id: str | None = None,
+    ) -> None:
+        """更新会话数据。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+            conversation_id: 会话 ID，为 None 时更新当前会话
+            history: 对话历史记录
+            title: 会话标题
+            persona_id: Persona ID
+        """
+        # 如果 conversation_id 为 None，使用当前会话
+        if conversation_id is None:
+            conversation_id = self._current_conversations.get(unified_msg_origin)
+            if conversation_id is None:
+                return
+
+        stored = await self._get_stored()
+        if conversation_id not in stored:
+            return
+
+        updates: dict[str, Any] = {}
+        if history is not None:
+            updates["content"] = history
+        if title is not None:
+            updates["title"] = title
+        if persona_id is not None:
+            updates["persona_id"] = persona_id
+
+        stored[conversation_id].update(updates)
+        await self._set_stored(stored)
+
+    async def delete_conversations_by_user_id(self, unified_msg_origin: str) -> None:
+        """删除指定用户的所有会话。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+        """
+        stored = await self._get_stored()
+        to_delete = [
+            conv_id
+            for conv_id, conv_data in stored.items()
+            if conv_data.get("unified_msg_origin") == unified_msg_origin
+        ]
+        for conv_id in to_delete:
+            del stored[conv_id]
+        await self._set_stored(stored)
+        # 清除当前会话记录
+        if unified_msg_origin in self._current_conversations:
+            del self._current_conversations[unified_msg_origin]
+
+    async def add_message_pair(
+        self,
+        cid: str,
+        user_message: str | dict,
+        assistant_message: str | dict,
+    ) -> None:
+        """向会话添加消息对。
+
+        Args:
+            cid: 会话 ID
+            user_message: 用户消息
+            assistant_message: 助手消息
+        """
+        stored = await self._get_stored()
+        if cid not in stored:
+            return
+        content = stored[cid].get("content", [])
+        # 处理消息格式
+        user_msg = (
+            user_message
+            if isinstance(user_message, dict)
+            else {"role": "user", "content": user_message}
+        )
+        assistant_msg = (
+            assistant_message
+            if isinstance(assistant_message, dict)
+            else {"role": "assistant", "content": assistant_message}
+        )
+        content.append(user_msg)
+        content.append(assistant_msg)
+        stored[cid]["content"] = content
+        await self._set_stored(stored)
+
+    async def update_conversation_title(
+        self,
+        unified_msg_origin: str,
+        title: str,
+        conversation_id: str | None = None,
+    ) -> None:
+        """更新会话标题。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+            title: 会话标题
+            conversation_id: 会话 ID，为 None 时更新当前会话
+
+        Deprecated:
+            请使用 update_conversation() 的 title 参数。
+        """
+        await self.update_conversation(
+            unified_msg_origin, conversation_id, title=title
+        )
+
+    async def update_conversation_persona_id(
+        self,
+        unified_msg_origin: str,
+        persona_id: str,
+        conversation_id: str | None = None,
+    ) -> None:
+        """更新会话 Persona ID。
+
+        Args:
+            unified_msg_origin: 统一消息来源
+            persona_id: Persona ID
+            conversation_id: 会话 ID，为 None 时更新当前会话
+
+        Deprecated:
+            请使用 update_conversation() 的 persona_id 参数。
+        """
+        await self.update_conversation(
+            unified_msg_origin, conversation_id, persona_id=persona_id
+        )
+
+    async def get_filtered_conversations(self, *args: Any, **kwargs: Any) -> Any:
+        """已弃用：v4 不支持此方法。"""
+        raise NotImplementedError(
+            "get_filtered_conversations() 在 v4 中不再支持。\n"
+            f"请使用 ctx.db.query(...) 自行实现过滤逻辑。\n"
+            f"迁移文档：{MIGRATION_DOC_URL}"
+        )
+
+    async def get_human_readable_context(self, *args: Any, **kwargs: Any) -> Any:
+        """已弃用：v4 不支持此方法。"""
+        raise NotImplementedError(
+            "get_human_readable_context() 在 v4 中不再支持。\n"
+            f"请自行遍历会话 content 字段格式化输出。\n"
+            f"迁移文档：{MIGRATION_DOC_URL}"
+        )
+
 
 class LegacyContext:
     def __init__(self, plugin_id: str) -> None:
@@ -129,9 +419,13 @@ class LegacyContext:
             text = str(message_chain)
         await ctx.platform.send(session, text)
 
+    # TODO:迁移文档中说明已废弃 add_llm_tools()，但仍保留接口以避免核心依赖问题。后续版本将移除此接口。
     async def add_llm_tools(self, *tools: Any) -> None:
-        _warn_once("context.add_llm_tools()", "ctx.llm.chat_raw(..., tools=...)")
-        return None
+        raise NotImplementedError(
+            "context.add_llm_tools() 在 v4 中不再支持。\n"
+            "请使用 ctx.llm.chat_raw(..., tools=[...]) 直接传递工具。\n"
+            f"迁移文档：{MIGRATION_DOC_URL}"
+        )
 
     async def put_kv_data(self, key: str, value: dict[str, Any]) -> None:
         _warn_once("context.put_kv_data()", "ctx.db.set(key, value)")
diff --git a/src-new/astrbot_sdk/clients/memory.py b/src-new/astrbot_sdk/clients/memory.py
index e88c7dbe5..8c6665da2 100644
--- a/src-new/astrbot_sdk/clients/memory.py
+++ b/src-new/astrbot_sdk/clients/memory.py
@@ -26,5 +26,18 @@ class MemoryClient:
             payload.update(extra)
         await self._proxy.call("memory.save", {"key": key, "value": payload})
 
+    async def get(self, key: str) -> dict[str, Any] | None:
+        """精确获取：通过唯一键获取单个记忆项。
+
+        Args:
+            key: 记忆项的唯一键
+
+        Returns:
+            记忆项内容，若不存在则返回 None
+        """
+        output = await self._proxy.call("memory.get", {"key": key})
+        value = output.get("value")
+        return value if isinstance(value, dict) else None
+
     async def delete(self, key: str) -> None:
         await self._proxy.call("memory.delete", {"key": key})
diff --git a/src-new/astrbot_sdk/protocol/legacy_adapter.py b/src-new/astrbot_sdk/protocol/legacy_adapter.py
index 7a99cfd36..63edafbb5 100644
--- a/src-new/astrbot_sdk/protocol/legacy_adapter.py
+++ b/src-new/astrbot_sdk/protocol/legacy_adapter.py
@@ -13,7 +13,6 @@ from .messages import (
     InitializeMessage,
     InvokeMessage,
     PeerInfo,
-    ProtocolMessage,
     ResultMessage,
 )
 
diff --git a/src-new/astrbot_sdk/protocol/messages.py b/src-new/astrbot_sdk/protocol/messages.py
index 1f25775cd..7fa3f9952 100644
--- a/src-new/astrbot_sdk/protocol/messages.py
+++ b/src-new/astrbot_sdk/protocol/messages.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from .descriptors import CapabilityDescriptor, HandlerDescriptor
 
@@ -65,6 +65,36 @@ class EventMessage(_MessageBase):
     output: dict[str, Any] = Field(default_factory=dict)
     error: ErrorPayload | None = None
 
+    @model_validator(mode="after")
+    def validate_phase_constraints(self) -> "EventMessage":
+        """验证各 phase 的字段约束。
+
+        - started: 所有字段必须为空
+        - delta: 必须有 data，output/error 必须为空
+        - completed: 必须有 output，data/error 必须为空
+        - failed: 必须有 error，data/output 必须为空
+        """
+        phase = self.phase
+        if phase == "started":
+            if self.data or self.output or self.error:
+                raise ValueError("started phase 必须所有字段为空")
+        elif phase == "delta":
+            if not self.data:
+                raise ValueError("delta phase 需要 data")
+            if self.output or self.error:
+                raise ValueError("delta phase 的 output/error 必须为空")
+        elif phase == "completed":
+            if not self.output:
+                raise ValueError("completed phase 需要 output")
+            if self.data or self.error:
+                raise ValueError("completed phase 的 data/error 必须为空")
+        elif phase == "failed":
+            if self.error is None:
+                raise ValueError("failed phase 需要 error")
+            if self.data or self.output:
+                raise ValueError("failed phase 的 data/output 必须为空")
+        return self
+
 
 class CancelMessage(_MessageBase):
     type: Literal["cancel"] = "cancel"
diff --git a/src-new/astrbot_sdk/runtime/bootstrap.py b/src-new/astrbot_sdk/runtime/bootstrap.py
index 6c747c0c0..dd0572cd8 100644
--- a/src-new/astrbot_sdk/runtime/bootstrap.py
+++ b/src-new/astrbot_sdk/runtime/bootstrap.py
@@ -220,6 +220,7 @@ class SupervisorRuntime:
         self.peer.set_cancel_handler(self._handle_upstream_cancel)
         self.worker_sessions: dict[str, WorkerSession] = {}
         self.handler_to_worker: dict[str, WorkerSession] = {}
+        self._handler_sources: dict[str, str] = {}  # handler_id -> plugin_name
         self.active_requests: dict[str, WorkerSession] = {}
         self.loaded_plugins: list[str] = []
         self.skipped_plugins: dict[str, str] = {}
@@ -249,6 +250,28 @@ class SupervisorRuntime:
             exposed=False,
         )
 
+    def _register_handler(
+        self, handler, session: WorkerSession, plugin_name: str
+    ) -> None:
+        """注册 handler，处理冲突时输出警告。
+
+        Args:
+            handler: Handler 描述符
+            session: Worker 会话
+            plugin_name: 插件名称
+        """
+        handler_id = handler.id
+        existing_plugin = self._handler_sources.get(handler_id)
+
+        if existing_plugin is not None:
+            logger.warning(
+                f"Handler ID 冲突：'{handler_id}' 已被插件 '{existing_plugin}' 注册，"
+                f"现在被插件 '{plugin_name}' 覆盖。"
+            )
+
+        self.handler_to_worker[handler_id] = session
+        self._handler_sources[handler_id] = plugin_name
+
     async def start(self) -> None:
         discovery = discover_plugins(self.plugins_dir)
         self.skipped_plugins = dict(discovery.skipped_plugins)
@@ -270,7 +293,7 @@ class SupervisorRuntime:
                 self.worker_sessions[plugin.name] = session
                 self.loaded_plugins.append(plugin.name)
                 for handler in session.handlers:
-                    self.handler_to_worker[handler.id] = session
+                    self._register_handler(handler, session, plugin.name)
 
             aggregated_handlers = list(self.handler_to_worker.keys())
             logger.info(
@@ -299,9 +322,12 @@ class SupervisorRuntime:
         session = self.worker_sessions.pop(plugin_name, None)
         if session is None:
             return
-        # 从 handler_to_worker 中移除该 worker 的所有 handlers
+        # 从 handler_to_worker 中移除该插件注册的 handlers（仅当来源仍为此插件时）
         for handler in session.handlers:
-            self.handler_to_worker.pop(handler.id, None)
+            source_plugin = self._handler_sources.get(handler.id)
+            if source_plugin == plugin_name:
+                self.handler_to_worker.pop(handler.id, None)
+                self._handler_sources.pop(handler.id, None)
         # 从 loaded_plugins 中移除
         if plugin_name in self.loaded_plugins:
             self.loaded_plugins.remove(plugin_name)
diff --git a/src-new/astrbot_sdk/runtime/handler_dispatcher.py b/src-new/astrbot_sdk/runtime/handler_dispatcher.py
index df0baae57..91e6aa0aa 100644
--- a/src-new/astrbot_sdk/runtime/handler_dispatcher.py
+++ b/src-new/astrbot_sdk/runtime/handler_dispatcher.py
@@ -2,7 +2,8 @@ from __future__ import annotations
 
 import asyncio
 import inspect
-from typing import Any
+import typing
+from typing import Any, get_type_hints
 
 from ..context import CancelToken, Context
 from ..events import MessageEvent, PlainTextResult
@@ -80,24 +81,111 @@ class HandlerDispatcher:
         ctx: Context,
         legacy_args: dict[str, Any] | None = None,
     ) -> list[Any]:
+        """构建 handler 参数列表。
+
+        注入优先级：
+        1. 按类型注解注入（支持 Optional[Type]）
+        2. 按参数名注入（兼容无类型注解的情况）
+        3. 从 legacy_args 注入（命令参数、regex 捕获组等）
+
+        Args:
+            handler: Handler 可调用对象
+            event: 消息事件
+            ctx: 运行时上下文
+            legacy_args: 旧版参数字典
+
+        Returns:
+            参数列表
+        """
+        from loguru import logger
+
         signature = inspect.signature(handler)
         args: list[Any] = []
         legacy_args = legacy_args or {}
+
+        # 尝试获取类型注解
+        type_hints: dict[str, Any] = {}
+        try:
+            type_hints = get_type_hints(handler)
+        except Exception:
+            pass
+
         for parameter in signature.parameters.values():
             if parameter.kind not in (
                 inspect.Parameter.POSITIONAL_ONLY,
                 inspect.Parameter.POSITIONAL_OR_KEYWORD,
             ):
                 continue
-            if parameter.name == "event":
-                args.append(event)
-            elif parameter.name in {"ctx", "context"}:
-                args.append(ctx)
-            elif parameter.name in legacy_args:
-                # 支持从 legacy args 中注入参数（如命令参数、regex 捕获组等）
-                args.append(legacy_args[parameter.name])
+
+            injected = None
+
+            # 1. 优先按类型注解注入
+            param_type = type_hints.get(parameter.name)
+            if param_type is not None:
+                injected = self._inject_by_type(param_type, event, ctx)
+
+            # 2. Fallback 按名字注入
+            if injected is None:
+                if parameter.name == "event":
+                    injected = event
+                elif parameter.name in {"ctx", "context"}:
+                    injected = ctx
+                elif parameter.name in legacy_args:
+                    injected = legacy_args[parameter.name]
+
+            # 3. 检查是否有默认值
+            if injected is None:
+                if parameter.default is not parameter.empty:
+                    # 有默认值，跳过注入
+                    continue
+                # 无默认值且无法注入，警告并传 None
+                logger.warning(
+                    f"Handler '{handler.__name__}': 参数 '{parameter.name}' "
+                    f"无法注入（类型: {param_type or '未知'}），将传入 None"
+                )
+                args.append(None)
+            else:
+                args.append(injected)
+
         return args
 
+    def _inject_by_type(
+        self, param_type: Any, event: MessageEvent, ctx: Context
+    ) -> Any:
+        """根据类型注解注入参数。
+
+        支持 Optional[Type] 类型。
+
+        Args:
+            param_type: 参数类型注解
+            event: 消息事件
+            ctx: 运行时上下文
+
+        Returns:
+            注入的值，若无法注入则返回 None
+        """
+        # 处理 Optional[Type] 情况
+        origin = typing.get_origin(param_type)
+        if origin is typing.Union:
+            args = typing.get_args(param_type)
+            non_none_types = [a for a in args if a is not type(None)]
+            if len(non_none_types) == 1:
+                param_type = non_none_types[0]
+
+        # 注入 MessageEvent 及其子类
+        if param_type is MessageEvent or (
+            isinstance(param_type, type) and issubclass(param_type, MessageEvent)
+        ):
+            return event
+
+        # 注入 Context 及其子类
+        if param_type is Context or (
+            isinstance(param_type, type) and issubclass(param_type, Context)
+        ):
+            return ctx
+
+        return None
+
     async def _consume_legacy_result(self, item: Any, event: MessageEvent) -> None:
         if isinstance(item, PlainTextResult):
             await event.reply(item.text)