feat: Enhance plugin runtime configuration and hook management

- Added `inactive_plugins` field to `RunnerReadyPayload` and `ReloadPluginResultPayload` to track plugins that are not activated due to being disabled or unmet dependencies.
- Introduced `InspectPluginConfigPayload` and `InspectPluginConfigResultPayload` for inspecting plugin configuration metadata.
- Implemented `PluginActivationStatus` enum to better represent plugin activation states.
- Updated `_activate_plugin` method to return activation status and handle inactive plugins accordingly.
- Added hooks for send service to allow modification of messages before and after sending.
- Created new runtime routes for listing hook specifications in the WebUI.
- Refactored plugin configuration handling to utilize runtime inspection for better accuracy and flexibility.
- Enhanced error handling and logging for plugin configuration operations.

src/plugin_runtime/component_query.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional, Tuple, cast
 
 from src.common.logger import get_logger
@@ -908,5 +909,27 @@ class ComponentQueryService:
             return None
         return dict(registration.config_schema)
 
+    def list_hook_specs(self) -> list[dict[str, Any]]:
+        """返回当前运行时公开的 Hook 规格清单。
+
+        Returns:
+            list[dict[str, Any]]: 可直接序列化给 WebUI 的 Hook 规格列表。
+        """
+
+        runtime_manager = self._get_runtime_manager()
+        return [
+            {
+                "name": spec.name,
+                "description": spec.description,
+                "parameters_schema": deepcopy(spec.parameters_schema),
+                "default_timeout_ms": spec.default_timeout_ms,
+                "allow_blocking": spec.allow_blocking,
+                "allow_observe": spec.allow_observe,
+                "allow_abort": spec.allow_abort,
+                "allow_kwargs_mutation": spec.allow_kwargs_mutation,
+            }
+            for spec in runtime_manager.list_hook_specs()
+        ]
+
 
 component_query_service = ComponentQueryService()

src/plugin_runtime/hook_catalog.py

@@ -0,0 +1,46 @@
+"""内置命名 Hook 目录注册器。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import List
+
+from src.plugin_runtime.host.hook_spec_registry import HookSpec, HookSpecRegistry
+
+
+HookSpecRegistrar = Callable[[HookSpecRegistry], List[HookSpec]]
+"""单个业务模块向注册中心写入 Hook 规格的注册器签名。"""
+
+
+def _get_builtin_hook_spec_registrars() -> List[HookSpecRegistrar]:
+    """返回当前内置 Hook 规格注册器列表。
+
+    Returns:
+        List[HookSpecRegistrar]: 已启用的内置 Hook 注册器列表。
+    """
+
+    from src.chat.message_receive.bot import register_chat_hook_specs
+    from src.maisaka.chat_loop_service import register_maisaka_hook_specs
+    from src.services.send_service import register_send_service_hook_specs
+
+    return [
+        register_chat_hook_specs,
+        register_send_service_hook_specs,
+        register_maisaka_hook_specs,
+    ]
+
+
+def register_builtin_hook_specs(registry: HookSpecRegistry) -> List[HookSpec]:
+    """向注册中心写入全部内置 Hook 规格。
+
+    Args:
+        registry: 目标 Hook 规格注册中心。
+
+    Returns:
+        List[HookSpec]: 本次完成注册后的全部内置 Hook 规格。
+    """
+
+    registered_specs: List[HookSpec] = []
+    for registrar in _get_builtin_hook_spec_registrars():
+        registered_specs.extend(registrar(registry))
+    return registered_specs

src/plugin_runtime/hook_payloads.py

@@ -0,0 +1,178 @@
+"""运行时 Hook 载荷序列化辅助。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Sequence
+
+from src.chat.message_receive.message import SessionMessage
+from src.common.data_models.llm_service_data_models import PromptMessage
+from src.llm_models.payload_content.message import Message
+from src.llm_models.payload_content.tool_option import ToolCall, ToolDefinitionInput, normalize_tool_options
+from src.plugin_runtime.host.message_utils import PluginMessageUtils
+
+
+def serialize_session_message(message: SessionMessage) -> Dict[str, Any]:
+    """将会话消息序列化为 Hook 可传输载荷。
+
+    Args:
+        message: 待序列化的会话消息。
+
+    Returns:
+        Dict[str, Any]: 可通过插件运行时传输的消息字典。
+    """
+
+    return dict(PluginMessageUtils._session_message_to_dict(message))
+
+
+def deserialize_session_message(raw_message: Any) -> SessionMessage:
+    """从 Hook 载荷恢复会话消息。
+
+    Args:
+        raw_message: Hook 返回的消息字典。
+
+    Returns:
+        SessionMessage: 恢复后的会话消息对象。
+
+    Raises:
+        ValueError: 消息结构不合法时抛出。
+    """
+
+    if not isinstance(raw_message, dict):
+        raise ValueError("Hook 返回的 `message` 必须是字典")
+    return PluginMessageUtils._build_session_message_from_dict(raw_message)
+
+
+def serialize_tool_calls(tool_calls: Sequence[ToolCall] | None) -> List[Dict[str, Any]]:
+    """将工具调用列表序列化为 Hook 可传输载荷。
+
+    Args:
+        tool_calls: 原始工具调用列表。
+
+    Returns:
+        List[Dict[str, Any]]: 序列化后的工具调用列表。
+    """
+
+    if not tool_calls:
+        return []
+
+    return [
+        {
+            "id": tool_call.call_id,
+            "function": {
+                "name": tool_call.func_name,
+                "arguments": dict(tool_call.args or {}),
+            },
+        }
+        for tool_call in tool_calls
+    ]
+
+
+def deserialize_tool_calls(raw_tool_calls: Any) -> List[ToolCall]:
+    """从 Hook 载荷恢复工具调用列表。
+
+    Args:
+        raw_tool_calls: Hook 返回的工具调用列表。
+
+    Returns:
+        List[ToolCall]: 恢复后的工具调用列表。
+
+    Raises:
+        ValueError: 结构不合法时抛出。
+    """
+
+    if raw_tool_calls in (None, []):
+        return []
+    if not isinstance(raw_tool_calls, list):
+        raise ValueError("Hook 返回的 `tool_calls` 必须是列表")
+
+    normalized_tool_calls: List[ToolCall] = []
+    for raw_tool_call in raw_tool_calls:
+        if not isinstance(raw_tool_call, dict):
+            raise ValueError("Hook 返回的工具调用项必须是字典")
+
+        function_info = raw_tool_call.get("function", {})
+        if isinstance(function_info, dict):
+            function_name = function_info.get("name")
+            function_arguments = function_info.get("arguments")
+        else:
+            function_name = raw_tool_call.get("name")
+            function_arguments = raw_tool_call.get("arguments")
+
+        call_id = raw_tool_call.get("id") or raw_tool_call.get("call_id")
+        if not isinstance(call_id, str) or not isinstance(function_name, str):
+            raise ValueError("Hook 返回的工具调用缺少 `id` 或函数名称")
+
+        normalized_tool_calls.append(
+            ToolCall(
+                call_id=call_id,
+                func_name=function_name,
+                args=function_arguments if isinstance(function_arguments, dict) else {},
+            )
+        )
+    return normalized_tool_calls
+
+
+def serialize_prompt_messages(messages: Sequence[Message]) -> List[PromptMessage]:
+    """将 LLM 消息列表序列化为 Hook 可传输载荷。
+
+    Args:
+        messages: 原始 LLM 消息列表。
+
+    Returns:
+        List[PromptMessage]: 序列化后的消息字典列表。
+    """
+
+    serialized_messages: List[PromptMessage] = []
+    for message in messages:
+        serialized_message: PromptMessage = {
+            "role": message.role.value,
+            "content": message.content,
+        }
+        if message.tool_call_id:
+            serialized_message["tool_call_id"] = message.tool_call_id
+        if message.tool_calls:
+            serialized_message["tool_calls"] = serialize_tool_calls(message.tool_calls)
+        serialized_messages.append(serialized_message)
+    return serialized_messages
+
+
+def deserialize_prompt_messages(raw_messages: Any) -> List[Message]:
+    """从 Hook 载荷恢复 LLM 消息列表。
+
+    Args:
+        raw_messages: Hook 返回的消息列表。
+
+    Returns:
+        List[Message]: 恢复后的 LLM 消息列表。
+
+    Raises:
+        ValueError: 结构不合法时抛出。
+    """
+
+    if not isinstance(raw_messages, list):
+        raise ValueError("Hook 返回的 `messages` 必须是列表")
+
+    from src.services.llm_service import _build_message_from_dict
+
+    normalized_messages: List[Message] = []
+    for raw_message in raw_messages:
+        if not isinstance(raw_message, dict):
+            raise ValueError("Hook 返回的消息项必须是字典")
+        normalized_messages.append(_build_message_from_dict(raw_message))
+    return normalized_messages
+
+
+def serialize_tool_definitions(tool_definitions: Sequence[ToolDefinitionInput]) -> List[Dict[str, Any]]:
+    """将工具定义列表序列化为 Hook 可传输载荷。
+
+    Args:
+        tool_definitions: 原始工具定义列表。
+
+    Returns:
+        List[Dict[str, Any]]: 序列化后的工具定义列表。
+    """
+
+    normalized_tool_options = normalize_tool_options(list(tool_definitions))
+    if not normalized_tool_options:
+        return []
+    return [tool_option.to_openai_function_schema() for tool_option in normalized_tool_options]

src/plugin_runtime/hook_schema_utils.py

@@ -0,0 +1,31 @@
+"""Hook 参数模型构造辅助。"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any, Dict, Sequence
+
+
+def build_object_schema(
+    properties: Dict[str, Dict[str, Any]],
+    *,
+    required: Sequence[str] | None = None,
+) -> Dict[str, Any]:
+    """构造对象级 JSON Schema。
+
+    Args:
+        properties: 字段定义映射。
+        required: 必填字段名列表。
+
+    Returns:
+        Dict[str, Any]: 标准化后的对象级 Schema。
+    """
+
+    schema: Dict[str, Any] = {
+        "type": "object",
+        "properties": deepcopy(properties),
+    }
+    normalized_required = [str(item).strip() for item in (required or []) if str(item).strip()]
+    if normalized_required:
+        schema["required"] = normalized_required
+    return schema

src/plugin_runtime/host/component_registry.py

@@ -18,9 +18,37 @@ import re
 from src.common.logger import get_logger
 from src.core.tooling import build_tool_detailed_description
 
+from .hook_spec_registry import HookSpecRegistry
+
 logger = get_logger("plugin_runtime.host.component_registry")
 
 
+class ComponentRegistrationError(ValueError):
+    """组件注册失败异常。"""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        component_name: str = "",
+        component_type: str = "",
+        plugin_id: str = "",
+    ) -> None:
+        """初始化组件注册失败异常。
+
+        Args:
+            message: 原始错误信息。
+            component_name: 组件名称。
+            component_type: 组件类型。
+            plugin_id: 插件 ID。
+        """
+
+        self.component_name = str(component_name or "").strip()
+        self.component_type = str(component_type or "").strip()
+        self.plugin_id = str(plugin_id or "").strip()
+        super().__init__(message)
+
+
 class ComponentTypes(str, Enum):
     ACTION = "ACTION"
     COMMAND = "COMMAND"
@@ -359,7 +387,14 @@ class ComponentRegistry:
     供业务层查询可用组件、匹配命令、调度 action/event 等。
     """
 
-    def __init__(self) -> None:
+    def __init__(self, hook_spec_registry: Optional[HookSpecRegistry] = None) -> None:
+        """初始化组件注册表。
+
+        Args:
+            hook_spec_registry: 可选的 Hook 规格注册中心；提供后会在注册
+                HookHandler 时执行规格校验。
+        """
+
         # 全量索引
         self._components: Dict[str, ComponentEntry] = {}  # full_name -> comp
 
@@ -370,6 +405,7 @@ class ComponentRegistry:
 
         # 按插件索引
         self._by_plugin: Dict[str, List[ComponentEntry]] = {}
+        self._hook_spec_registry = hook_spec_registry
 
     @staticmethod
     def _convert_action_metadata_to_tool_metadata(
@@ -475,77 +511,211 @@ class ComponentRegistry:
             type_dict.clear()
         self._by_plugin.clear()
 
-    # ====== 注册 / 注销 ======
-    def register_component(self, name: str, component_type: str, plugin_id: str, metadata: Dict[str, Any]) -> bool:
-        """注册单个组件
+    @staticmethod
+    def _is_legacy_action_component(component: ComponentEntry) -> bool:
+        """判断组件是否为兼容旧 Action 的 Tool 条目。
 
         Args:
-            name: 组件名称（不含插件id前缀）
-            component_type: 组件类型（如 `ACTION`、`COMMAND` 等）
-            plugin_id: 插件id
-            metadata: 组件元数据
+            component: 待判断的组件条目。
+
         Returns:
-            success (bool): 是否成功注册（失败原因通常是组件类型无效）
+            bool: 是否为兼容旧 Action 组件。
         """
+
+        if not isinstance(component, ToolEntry):
+            return False
+        return str(component.metadata.get("legacy_component_type", "") or "").strip().upper() == "ACTION"
+
+    def _validate_hook_handler_entry(self, component: HookHandlerEntry) -> None:
+        """校验 HookHandler 是否满足已注册的 Hook 规格。
+
+        Args:
+            component: 待校验的 HookHandler 条目。
+
+        Raises:
+            ComponentRegistrationError: HookHandler 声明不合法时抛出。
+        """
+
+        if self._hook_spec_registry is None:
+            return
+
+        hook_spec = self._hook_spec_registry.get_hook_spec(component.hook)
+        if hook_spec is None:
+            raise ComponentRegistrationError(
+                f"HookHandler {component.full_name} 声明了未注册的 Hook: {component.hook}",
+                component_name=component.name,
+                component_type=component.component_type.value,
+                plugin_id=component.plugin_id,
+            )
+
+        if component.is_blocking and not hook_spec.allow_blocking:
+            raise ComponentRegistrationError(
+                f"HookHandler {component.full_name} 不能注册为 blocking：Hook {component.hook} 不允许 blocking 处理器",
+                component_name=component.name,
+                component_type=component.component_type.value,
+                plugin_id=component.plugin_id,
+            )
+
+        if component.is_observe and not hook_spec.allow_observe:
+            raise ComponentRegistrationError(
+                f"HookHandler {component.full_name} 不能注册为 observe：Hook {component.hook} 不允许 observe 处理器",
+                component_name=component.name,
+                component_type=component.component_type.value,
+                plugin_id=component.plugin_id,
+            )
+
+        if component.error_policy == "abort" and not hook_spec.allow_abort:
+            raise ComponentRegistrationError(
+                f"HookHandler {component.full_name} 不能使用 error_policy=abort：Hook {component.hook} 不允许 abort",
+                component_name=component.name,
+                component_type=component.component_type.value,
+                plugin_id=component.plugin_id,
+            )
+
+    def _build_component_entry(
+        self,
+        name: str,
+        component_type: str,
+        plugin_id: str,
+        metadata: Dict[str, Any],
+    ) -> ComponentEntry:
+        """根据声明构造组件条目。
+
+        Args:
+            name: 组件名称。
+            component_type: 组件类型。
+            plugin_id: 插件 ID。
+            metadata: 组件元数据。
+
+        Returns:
+            ComponentEntry: 已构造并完成校验的组件条目。
+
+        Raises:
+            ComponentRegistrationError: 组件声明不合法时抛出。
+        """
+
         try:
             normalized_type = self._normalize_component_type(component_type)
             normalized_metadata = dict(metadata)
             if normalized_type == ComponentTypes.ACTION:
                 normalized_metadata = self._convert_action_metadata_to_tool_metadata(name, normalized_metadata)
-                comp = ToolEntry(name, ComponentTypes.TOOL.value, plugin_id, normalized_metadata)
+                component = ToolEntry(name, ComponentTypes.TOOL.value, plugin_id, normalized_metadata)
             elif normalized_type == ComponentTypes.COMMAND:
-                comp = CommandEntry(name, normalized_type.value, plugin_id, normalized_metadata)
+                component = CommandEntry(name, normalized_type.value, plugin_id, normalized_metadata)
             elif normalized_type == ComponentTypes.TOOL:
-                comp = ToolEntry(name, normalized_type.value, plugin_id, normalized_metadata)
+                component = ToolEntry(name, normalized_type.value, plugin_id, normalized_metadata)
             elif normalized_type == ComponentTypes.EVENT_HANDLER:
-                comp = EventHandlerEntry(name, normalized_type.value, plugin_id, normalized_metadata)
+                component = EventHandlerEntry(name, normalized_type.value, plugin_id, normalized_metadata)
             elif normalized_type == ComponentTypes.HOOK_HANDLER:
-                comp = HookHandlerEntry(name, normalized_type.value, plugin_id, normalized_metadata)
+                component = HookHandlerEntry(name, normalized_type.value, plugin_id, normalized_metadata)
+                self._validate_hook_handler_entry(component)
             elif normalized_type == ComponentTypes.MESSAGE_GATEWAY:
-                comp = MessageGatewayEntry(name, normalized_type.value, plugin_id, normalized_metadata)
+                component = MessageGatewayEntry(name, normalized_type.value, plugin_id, normalized_metadata)
             else:
-                raise ValueError(f"组件类型 {component_type} 不存在")
-        except ValueError:
-            logger.error(f"组件类型 {component_type} 不存在")
-            return False
+                raise ComponentRegistrationError(
+                    f"组件类型 {component_type} 不存在",
+                    component_name=name,
+                    component_type=component_type,
+                    plugin_id=plugin_id,
+                )
+        except ComponentRegistrationError:
+            raise
+        except Exception as exc:
+            raise ComponentRegistrationError(
+                str(exc),
+                component_name=name,
+                component_type=component_type,
+                plugin_id=plugin_id,
+            ) from exc
 
-        if comp.full_name in self._components:
-            logger.warning(f"组件 {comp.full_name} 已存在，覆盖")
-            old_comp = self._components[comp.full_name]
-            # 从 _by_plugin 列表中移除旧条目，防止幽灵组件堆积
-            old_list = self._by_plugin.get(old_comp.plugin_id)
-            if old_list is not None:
-                with contextlib.suppress(ValueError):
-                    old_list.remove(old_comp)
-            # 从旧类型索引中移除，防止类型变更时幽灵残留
-            if old_type_dict := self._by_type.get(old_comp.component_type):
-                old_type_dict.pop(comp.full_name, None)
+        return component
 
-        self._components[comp.full_name] = comp
-        self._by_type[comp.component_type][comp.full_name] = comp
-        self._by_plugin.setdefault(plugin_id, []).append(comp)
+    def _remove_existing_component_entry(self, component: ComponentEntry) -> None:
+        """移除同名旧组件条目。
 
+        Args:
+            component: 即将写入的新组件条目。
+        """
+
+        if component.full_name not in self._components:
+            return
+
+        logger.warning(f"组件 {component.full_name} 已存在，覆盖")
+        old_component = self._components[component.full_name]
+        old_list = self._by_plugin.get(old_component.plugin_id)
+        if old_list is not None:
+            with contextlib.suppress(ValueError):
+                old_list.remove(old_component)
+        if old_type_dict := self._by_type.get(old_component.component_type):
+            old_type_dict.pop(component.full_name, None)
+
+    def _add_component_entry(self, component: ComponentEntry) -> None:
+        """写入单个组件条目到全部索引。
+
+        Args:
+            component: 待写入的组件条目。
+        """
+
+        self._remove_existing_component_entry(component)
+        self._components[component.full_name] = component
+        self._by_type[component.component_type][component.full_name] = component
+        self._by_plugin.setdefault(component.plugin_id, []).append(component)
+
+    # ====== 注册 / 注销 ======
+    def register_component(self, name: str, component_type: str, plugin_id: str, metadata: Dict[str, Any]) -> bool:
+        """注册单个组件。
+
+        Args:
+            name: 组件名称（不含插件 ID 前缀）。
+            component_type: 组件类型（如 ``ACTION``、``COMMAND`` 等）。
+            plugin_id: 插件 ID。
+            metadata: 组件元数据。
+
+        Returns:
+            bool: 注册成功时恒为 ``True``。
+
+        Raises:
+            ComponentRegistrationError: 组件声明不合法时抛出。
+        """
+
+        component = self._build_component_entry(name, component_type, plugin_id, metadata)
+        self._add_component_entry(component)
         return True
 
     def register_plugin_components(self, plugin_id: str, components: List[Dict[str, Any]]) -> int:
-        """批量注册一个插件的所有组件，返回成功注册数。
+        """批量替换一个插件的组件集合。
+
+        该方法会先完整校验所有组件声明，只有全部通过后才会替换旧组件，
+        从而避免插件进入半注册状态。
+
         Args:
-            plugin_id (str): 插件id
-            components (List[Dict[str, Any]]): 组件字典列表，每个组件包含 name, component_type, metadata 等字段
+            plugin_id: 插件 ID。
+            components: 组件声明字典列表。
+
         Returns:
-            count (int): 成功注册的组件数量
+            int: 实际注册的组件数量。
+
+        Raises:
+            ComponentRegistrationError: 任一组件声明不合法时抛出。
         """
-        count = 0
-        for comp_data in components:
-            ok = self.register_component(
-                name=comp_data.get("name", ""),
-                component_type=comp_data.get("component_type", ""),
-                plugin_id=plugin_id,
-                metadata=comp_data.get("metadata", {}),
+
+        prepared_components: List[ComponentEntry] = []
+        for component_data in components:
+            prepared_components.append(
+                self._build_component_entry(
+                    name=str(component_data.get("name", "") or ""),
+                    component_type=str(component_data.get("component_type", "") or ""),
+                    plugin_id=plugin_id,
+                    metadata=component_data.get("metadata", {})
+                    if isinstance(component_data.get("metadata"), dict)
+                    else {},
+                )
             )
-            if ok:
-                count += 1
-        return count
+
+        self.remove_components_by_plugin(plugin_id)
+        for component in prepared_components:
+            self._add_component_entry(component)
+        return len(prepared_components)
 
     def remove_components_by_plugin(self, plugin_id: str) -> int:
         """移除某个插件的所有组件，返回移除数量。
@@ -652,6 +822,17 @@ class ComponentRegistry:
         except ValueError:
             logger.error(f"组件类型 {component_type} 不存在")
             raise
+
+        if comp_type == ComponentTypes.ACTION:
+            action_components = [
+                component
+                for component in self._by_type.get(ComponentTypes.TOOL, {}).values()
+                if self._is_legacy_action_component(component)
+            ]
+            if enabled_only:
+                return [component for component in action_components if self.check_component_enabled(component, session_id)]
+            return action_components
+
         type_dict = self._by_type.get(comp_type, {})
         if enabled_only:
             return [c for c in type_dict.values() if self.check_component_enabled(c, session_id)]
@@ -854,6 +1035,34 @@ class ComponentRegistry:
                 tools.append(comp)
         return tools
 
+    def get_tools_for_llm(self, *, enabled_only: bool = True, session_id: Optional[str] = None) -> List[Dict[str, Any]]:
+        """兼容旧接口，返回可供 LLM 使用的工具条目列表。
+
+        Args:
+            enabled_only: 是否仅返回启用的组件。
+            session_id: 可选的会话 ID，若提供则考虑会话禁用状态。
+
+        Returns:
+            List[Dict[str, Any]]: 兼容旧结构的工具组件字典列表。
+        """
+
+        return [
+            {
+                "name": tool.full_name,
+                "description": tool.description,
+                "parameters": (
+                    dict(tool.parameters_raw)
+                    if isinstance(tool.parameters_raw, dict) and tool.parameters_raw
+                    else tool._get_parameters_schema() or {}
+                ),
+                "parameters_raw": tool.parameters_raw,
+                "enabled": tool.enabled,
+                "plugin_id": tool.plugin_id,
+            }
+            for tool in self.get_tools(enabled_only=enabled_only, session_id=session_id)
+            if not self._is_legacy_action_component(tool)
+        ]
+
     # ====== 统计信息 ======
     def get_stats(self) -> StatusDict:
         """获取注册统计。
@@ -863,9 +1072,21 @@ class ComponentRegistry:
         """
         return StatusDict(
             total=len(self._components),
-            action=len(self._by_type[ComponentTypes.ACTION]),
+            action=len(
+                [
+                    component
+                    for component in self._by_type.get(ComponentTypes.TOOL, {}).values()
+                    if self._is_legacy_action_component(component)
+                ]
+            ),
             command=len(self._by_type[ComponentTypes.COMMAND]),
-            tool=len(self._by_type[ComponentTypes.TOOL]),
+            tool=len(
+                [
+                    component
+                    for component in self._by_type.get(ComponentTypes.TOOL, {}).values()
+                    if not self._is_legacy_action_component(component)
+                ]
+            ),
             event_handler=len(self._by_type[ComponentTypes.EVENT_HANDLER]),
             hook_handler=len(self._by_type[ComponentTypes.HOOK_HANDLER]),
             message_gateway=len(self._by_type[ComponentTypes.MESSAGE_GATEWAY]),

src/plugin_runtime/host/hook_dispatcher.py

@@ -26,6 +26,8 @@ import contextlib
 from src.common.logger import get_logger
 from src.config.config import global_config
 
+from .hook_spec_registry import HookSpec, HookSpecRegistry
+
 if TYPE_CHECKING:
     from .component_registry import HookHandlerEntry
     from .supervisor import PluginRunnerSupervisor
@@ -33,29 +35,6 @@ if TYPE_CHECKING:
 logger = get_logger("plugin_runtime.host.hook_dispatcher")
 
 
-@dataclass(slots=True)
-class HookSpec:
-    """命名 Hook 的静态规格定义。
-
-    Attributes:
-        name: Hook 的唯一名称。
-        description: Hook 描述。
-        default_timeout_ms: 默认超时毫秒数；为 `0` 时退回系统默认值。
-        allow_blocking: 是否允许注册阻塞处理器。
-        allow_observe: 是否允许注册观察处理器。
-        allow_abort: 是否允许处理器中止当前 Hook 调用。
-        allow_kwargs_mutation: 是否允许阻塞处理器修改 `kwargs`。
-    """
-
-    name: str
-    description: str = ""
-    default_timeout_ms: int = 0
-    allow_blocking: bool = True
-    allow_observe: bool = True
-    allow_abort: bool = True
-    allow_kwargs_mutation: bool = True
-
-
 @dataclass(slots=True)
 class HookHandlerExecutionResult:
     """单个 HookHandler 的执行结果。
@@ -121,17 +100,19 @@ class HookDispatcher:
     def __init__(
         self,
         supervisors_provider: Optional[Callable[[], Sequence["PluginRunnerSupervisor"]]] = None,
+        hook_spec_registry: Optional[HookSpecRegistry] = None,
     ) -> None:
         """初始化 Hook 分发器。
 
         Args:
             supervisors_provider: 可选的 Supervisor 提供器。若调用 `invoke_hook()`
                 时未显式传入 `supervisors`，则使用该回调获取目标 Supervisor 列表。
+            hook_spec_registry: 可选的 Hook 规格注册中心；留空时使用独立注册中心。
         """
 
         self._background_tasks: Set[asyncio.Task[Any]] = set()
-        self._hook_specs: Dict[str, HookSpec] = {}
         self._supervisors_provider = supervisors_provider
+        self._hook_spec_registry = hook_spec_registry or HookSpecRegistry()
 
     async def stop(self) -> None:
         """停止分发器并取消所有未完成的观察任务。"""
@@ -148,16 +129,7 @@ class HookDispatcher:
             spec: 需要注册的 Hook 规格。
         """
 
-        normalized_name = self._normalize_hook_name(spec.name)
-        self._hook_specs[normalized_name] = HookSpec(
-            name=normalized_name,
-            description=spec.description,
-            default_timeout_ms=max(int(spec.default_timeout_ms), 0),
-            allow_blocking=bool(spec.allow_blocking),
-            allow_observe=bool(spec.allow_observe),
-            allow_abort=bool(spec.allow_abort),
-            allow_kwargs_mutation=bool(spec.allow_kwargs_mutation),
-        )
+        self._hook_spec_registry.register_hook_spec(spec)
 
     def register_hook_specs(self, specs: Sequence[HookSpec]) -> None:
         """批量注册命名 Hook 规格。
@@ -180,14 +152,37 @@ class HookDispatcher:
         """
 
         normalized_name = self._normalize_hook_name(hook_name)
-        if normalized_name in self._hook_specs:
-            return self._hook_specs[normalized_name]
+        registered_spec = self._hook_spec_registry.get_hook_spec(normalized_name)
+        if registered_spec is not None:
+            return registered_spec
 
         return HookSpec(
             name=normalized_name,
+            parameters_schema={},
             default_timeout_ms=self._get_default_timeout_ms(),
         )
 
+    def unregister_hook_spec(self, hook_name: str) -> bool:
+        """注销指定命名 Hook 规格。
+
+        Args:
+            hook_name: 目标 Hook 名称。
+
+        Returns:
+            bool: 是否成功注销。
+        """
+
+        return self._hook_spec_registry.unregister_hook_spec(hook_name)
+
+    def list_hook_specs(self) -> List[HookSpec]:
+        """返回当前全部显式注册的 Hook 规格。
+
+        Returns:
+            List[HookSpec]: 已注册 Hook 规格列表。
+        """
+
+        return self._hook_spec_registry.list_hook_specs()
+
     async def invoke_hook(
         self,
         hook_name: str,

src/plugin_runtime/host/hook_spec_registry.py

@@ -0,0 +1,190 @@
+"""命名 Hook 规格注册中心。"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Sequence
+
+
+@dataclass(slots=True)
+class HookSpec:
+    """命名 Hook 的静态规格定义。
+
+    Attributes:
+        name: Hook 的唯一名称。
+        description: Hook 描述。
+        parameters_schema: Hook 参数模型，使用对象级 JSON Schema 表示。
+        default_timeout_ms: 默认超时毫秒数；为 ``0`` 时退回系统默认值。
+        allow_blocking: 是否允许注册阻塞处理器。
+        allow_observe: 是否允许注册观察处理器。
+        allow_abort: 是否允许处理器中止当前 Hook 调用。
+        allow_kwargs_mutation: 是否允许阻塞处理器修改 ``kwargs``。
+    """
+
+    name: str
+    description: str = ""
+    parameters_schema: Dict[str, Any] = field(default_factory=dict)
+    default_timeout_ms: int = 0
+    allow_blocking: bool = True
+    allow_observe: bool = True
+    allow_abort: bool = True
+    allow_kwargs_mutation: bool = True
+
+
+class HookSpecRegistry:
+    """命名 Hook 规格注册中心。"""
+
+    def __init__(self) -> None:
+        """初始化 Hook 规格注册中心。"""
+
+        self._hook_specs: Dict[str, HookSpec] = {}
+
+    @staticmethod
+    def _normalize_hook_name(hook_name: str) -> str:
+        """规范化 Hook 名称。
+
+        Args:
+            hook_name: 原始 Hook 名称。
+
+        Returns:
+            str: 规范化后的 Hook 名称。
+
+        Raises:
+            ValueError: Hook 名称为空时抛出。
+        """
+
+        normalized_name = str(hook_name or "").strip()
+        if not normalized_name:
+            raise ValueError("Hook 名称不能为空")
+        return normalized_name
+
+    @staticmethod
+    def _normalize_parameters_schema(raw_schema: Any) -> Dict[str, Any]:
+        """规范化 Hook 参数模型。
+
+        Args:
+            raw_schema: 原始参数模型。
+
+        Returns:
+            Dict[str, Any]: 规范化后的对象级 JSON Schema。
+
+        Raises:
+            ValueError: 参数模型不是合法对象级 Schema 时抛出。
+        """
+
+        if raw_schema is None:
+            return {}
+        if not isinstance(raw_schema, dict):
+            raise ValueError("Hook 参数模型必须是字典")
+        if not raw_schema:
+            return {}
+
+        normalized_schema = deepcopy(raw_schema)
+        schema_type = normalized_schema.get("type")
+        properties = normalized_schema.get("properties")
+        if schema_type not in {"", None, "object"} and properties is None:
+            raise ValueError("Hook 参数模型必须是 object 类型或属性映射")
+        if schema_type in {"", None} and properties is None:
+            normalized_schema = {
+                "type": "object",
+                "properties": normalized_schema,
+            }
+        elif schema_type in {"", None}:
+            normalized_schema["type"] = "object"
+
+        if normalized_schema.get("type") != "object":
+            raise ValueError("Hook 参数模型必须是 object 类型")
+        return normalized_schema
+
+    @classmethod
+    def _normalize_spec(cls, spec: HookSpec) -> HookSpec:
+        """规范化 Hook 规格对象。
+
+        Args:
+            spec: 原始 Hook 规格。
+
+        Returns:
+            HookSpec: 规范化后的 Hook 规格副本。
+        """
+
+        return HookSpec(
+            name=cls._normalize_hook_name(spec.name),
+            description=str(spec.description or "").strip(),
+            parameters_schema=cls._normalize_parameters_schema(spec.parameters_schema),
+            default_timeout_ms=max(int(spec.default_timeout_ms), 0),
+            allow_blocking=bool(spec.allow_blocking),
+            allow_observe=bool(spec.allow_observe),
+            allow_abort=bool(spec.allow_abort),
+            allow_kwargs_mutation=bool(spec.allow_kwargs_mutation),
+        )
+
+    def clear(self) -> None:
+        """清空全部 Hook 规格。"""
+
+        self._hook_specs.clear()
+
+    def register_hook_spec(self, spec: HookSpec) -> HookSpec:
+        """注册单个 Hook 规格。
+
+        Args:
+            spec: 需要注册的 Hook 规格。
+
+        Returns:
+            HookSpec: 规范化后实际注册的 Hook 规格。
+        """
+
+        normalized_spec = self._normalize_spec(spec)
+        self._hook_specs[normalized_spec.name] = normalized_spec
+        return normalized_spec
+
+    def register_hook_specs(self, specs: Sequence[HookSpec]) -> List[HookSpec]:
+        """批量注册 Hook 规格。
+
+        Args:
+            specs: 需要注册的 Hook 规格列表。
+
+        Returns:
+            List[HookSpec]: 规范化后实际注册的 Hook 规格列表。
+        """
+
+        return [self.register_hook_spec(spec) for spec in specs]
+
+    def unregister_hook_spec(self, hook_name: str) -> bool:
+        """注销指定 Hook 规格。
+
+        Args:
+            hook_name: 目标 Hook 名称。
+
+        Returns:
+            bool: 是否成功删除。
+        """
+
+        normalized_name = self._normalize_hook_name(hook_name)
+        return self._hook_specs.pop(normalized_name, None) is not None
+
+    def get_hook_spec(self, hook_name: str) -> Optional[HookSpec]:
+        """获取指定 Hook 的显式规格。
+
+        Args:
+            hook_name: 目标 Hook 名称。
+
+        Returns:
+            Optional[HookSpec]: 已注册时返回规格副本，否则返回 ``None``。
+        """
+
+        normalized_name = self._normalize_hook_name(hook_name)
+        spec = self._hook_specs.get(normalized_name)
+        return None if spec is None else self._normalize_spec(spec)
+
+    def list_hook_specs(self) -> List[HookSpec]:
+        """返回当前全部 Hook 规格。
+
+        Returns:
+            List[HookSpec]: 按 Hook 名称升序排列的规格副本列表。
+        """
+
+        return [
+            self._normalize_spec(spec)
+            for _, spec in sorted(self._hook_specs.items(), key=lambda item: item[0])
+        ]

src/plugin_runtime/host/supervisor.py

@@ -27,6 +27,8 @@ from src.plugin_runtime.protocol.envelope import (
     ConfigUpdatedPayload,
     Envelope,
     HealthPayload,
+    InspectPluginConfigPayload,
+    InspectPluginConfigResultPayload,
     MessageGatewayStateUpdatePayload,
     MessageGatewayStateUpdateResultPayload,
     PROTOCOL_VERSION,
@@ -52,6 +54,7 @@ from .capability_service import CapabilityService
 from .component_registry import ComponentRegistry
 from .event_dispatcher import EventDispatcher
 from .hook_dispatcher import HookDispatchResult, HookDispatcher
+from .hook_spec_registry import HookSpecRegistry
 from .logger_bridge import RunnerLogBridge
 from .message_gateway import MessageGateway
 from .rpc_server import RPCServer
@@ -84,6 +87,7 @@ class PluginRunnerSupervisor:
         self,
         plugin_dirs: Optional[List[Path]] = None,
         group_name: str = "third_party",
+        hook_spec_registry: Optional[HookSpecRegistry] = None,
         socket_path: Optional[str] = None,
         health_check_interval_sec: Optional[float] = None,
         max_restart_attempts: Optional[int] = None,
@@ -94,6 +98,7 @@ class PluginRunnerSupervisor:
         Args:
             plugin_dirs: 由当前 Runner 负责加载的插件目录列表。
             group_name: 当前 Supervisor 所属运行时分组名称。
+            hook_spec_registry: 可选的共享 Hook 规格注册中心。
             socket_path: 自定义 IPC 地址；留空时由传输层自动生成。
             health_check_interval_sec: 健康检查间隔，单位秒。
             max_restart_attempts: 自动重启 Runner 的最大次数。
@@ -110,9 +115,12 @@ class PluginRunnerSupervisor:
         self._authorization = AuthorizationManager()
         self._capability_service = CapabilityService(self._authorization)
         self._api_registry = APIRegistry()
-        self._component_registry = ComponentRegistry()
+        self._component_registry = ComponentRegistry(hook_spec_registry=hook_spec_registry)
         self._event_dispatcher = EventDispatcher(self._component_registry)
-        self._hook_dispatcher = HookDispatcher(lambda: [self])
+        self._hook_dispatcher = HookDispatcher(
+            lambda: [self],
+            hook_spec_registry=hook_spec_registry,
+        )
         self._message_gateway = MessageGateway(self._component_registry)
         self._log_bridge = RunnerLogBridge()
 
@@ -581,6 +589,49 @@ class PluginRunnerSupervisor:
             raise ValueError("插件配置校验失败")
         return dict(result.normalized_config)
 
+    async def inspect_plugin_config(
+        self,
+        plugin_id: str,
+        config_data: Optional[Dict[str, Any]] = None,
+        *,
+        use_provided_config: bool = False,
+    ) -> InspectPluginConfigResultPayload:
+        """请求 Runner 解析插件配置元数据。
+
+        Args:
+            plugin_id: 目标插件 ID。
+            config_data: 可选的配置内容。
+            use_provided_config: 是否优先使用传入配置而不是磁盘配置。
+
+        Returns:
+            InspectPluginConfigResultPayload: 插件配置解析结果。
+
+        Raises:
+            ValueError: Runner 无法解析插件或返回了错误响应时抛出。
+        """
+
+        payload = InspectPluginConfigPayload(
+            config_data=config_data or {},
+            use_provided_config=use_provided_config,
+        )
+        try:
+            response = await self._rpc_server.send_request(
+                "plugin.inspect_config",
+                plugin_id=plugin_id,
+                payload=payload.model_dump(),
+                timeout_ms=10000,
+            )
+        except Exception as exc:
+            raise ValueError(f"插件配置解析请求失败: {exc}") from exc
+
+        if response.error:
+            raise ValueError(str(response.error.get("message", "插件配置解析失败")))
+
+        result = InspectPluginConfigResultPayload.model_validate(response.payload)
+        if not result.success:
+            raise ValueError("插件配置解析失败")
+        return result
+
     def get_config_reload_subscribers(self, scope: str) -> List[str]:
         """返回订阅指定全局配置广播的插件列表。
 
@@ -713,15 +764,25 @@ class PluginRunnerSupervisor:
 
         component_declarations = [component.model_dump() for component in payload.components]
         runtime_components, api_components = self._split_component_declarations(component_declarations)
-        self._component_registry.remove_components_by_plugin(payload.plugin_id)
-        self._api_registry.remove_apis_by_plugin(payload.plugin_id)
-        await self._unregister_all_message_gateway_drivers_for_plugin(payload.plugin_id)
+        try:
+            registered_count = self._component_registry.register_plugin_components(
+                payload.plugin_id,
+                runtime_components,
+            )
+        except Exception as exc:
+            logger.error(f"插件 {payload.plugin_id} 组件注册失败: {exc}")
+            return envelope.make_error_response(
+                ErrorCode.E_BAD_PAYLOAD.value,
+                str(exc),
+                details={
+                    "plugin_id": payload.plugin_id,
+                    "component_count": len(runtime_components),
+                },
+            )
 
-        registered_count = self._component_registry.register_plugin_components(
-            payload.plugin_id,
-            runtime_components,
-        )
+        self._api_registry.remove_apis_by_plugin(payload.plugin_id)
         registered_api_count = self._api_registry.register_plugin_apis(payload.plugin_id, api_components)
+        await self._unregister_all_message_gateway_drivers_for_plugin(payload.plugin_id)
         self._registered_plugins[payload.plugin_id] = payload
         self._message_gateway_states[payload.plugin_id] = {}
 

src/plugin_runtime/integration.py

@@ -25,6 +25,7 @@ from typing import (
 )
 
 import asyncio
+import inspect
 
 import tomlkit
 
@@ -32,14 +33,17 @@ from src.common.logger import get_logger
 from src.config.config import config_manager
 from src.config.file_watcher import FileChange, FileWatcher
 from src.platform_io import DeliveryBatch, InboundMessageEnvelope, get_platform_io_manager
+from src.plugin_runtime.hook_catalog import register_builtin_hook_specs
 from src.plugin_runtime.capabilities import (
     RuntimeComponentCapabilityMixin,
     RuntimeCoreCapabilityMixin,
     RuntimeDataCapabilityMixin,
 )
 from src.plugin_runtime.capabilities.registry import register_capability_impls
-from src.plugin_runtime.host.hook_dispatcher import HookDispatchResult, HookDispatcher, HookSpec
+from src.plugin_runtime.host.hook_dispatcher import HookDispatchResult, HookDispatcher
+from src.plugin_runtime.host.hook_spec_registry import HookSpec, HookSpecRegistry
 from src.plugin_runtime.host.message_utils import MessageDict, PluginMessageUtils
+from src.plugin_runtime.protocol.envelope import InspectPluginConfigResultPayload
 from src.plugin_runtime.runner.manifest_validator import ManifestValidator
 
 if TYPE_CHECKING:
@@ -87,7 +91,12 @@ class PluginRuntimeManager(
         self._manifest_validator: ManifestValidator = ManifestValidator()
         self._config_reload_callback: Callable[[Sequence[str]], Awaitable[None]] = self._handle_main_config_reload
         self._config_reload_callback_registered: bool = False
-        self._hook_dispatcher: HookDispatcher = HookDispatcher(lambda: self.supervisors)
+        self._hook_spec_registry: HookSpecRegistry = HookSpecRegistry()
+        self._builtin_hook_specs_registered: bool = False
+        self._hook_dispatcher: HookDispatcher = HookDispatcher(
+            lambda: self.supervisors,
+            hook_spec_registry=self._hook_spec_registry,
+        )
 
     async def _dispatch_platform_inbound(self, envelope: InboundMessageEnvelope) -> None:
         """接收 Platform IO 审核后的入站消息并送入主消息链。
@@ -155,6 +164,33 @@ class PluginRuntimeManager(
             return ["third_party", "builtin"]
         return ["builtin", "third_party"]
 
+    @staticmethod
+    def _instantiate_supervisor(supervisor_cls: Any, **kwargs: Any) -> Any:
+        """兼容不同构造签名地实例化 Supervisor。
+
+        Args:
+            supervisor_cls: 目标 Supervisor 类。
+            **kwargs: 期望传入的构造参数。
+
+        Returns:
+            Any: 实例化后的 Supervisor。
+        """
+
+        signature = inspect.signature(supervisor_cls)
+        accepts_var_keyword = any(
+            parameter.kind == inspect.Parameter.VAR_KEYWORD
+            for parameter in signature.parameters.values()
+        )
+        if accepts_var_keyword:
+            return supervisor_cls(**kwargs)
+
+        supported_kwargs = {
+            key: value
+            for key, value in kwargs.items()
+            if key in signature.parameters
+        }
+        return supervisor_cls(**supported_kwargs)
+
     # ─── 生命周期 ─────────────────────────────────────────────
 
     async def start(self) -> None:
@@ -185,6 +221,7 @@ class PluginRuntimeManager(
             logger.info("未找到任何插件目录，跳过插件运行时启动")
             return
 
+        self.ensure_builtin_hook_specs_registered()
         platform_io_manager = get_platform_io_manager()
 
         # 从配置读取自定义 IPC socket 路径（留空则自动生成）
@@ -196,17 +233,21 @@ class PluginRuntimeManager(
 
         # 创建两个 Supervisor，各自拥有独立的 socket / Runner 子进程
         if builtin_dirs:
-            self._builtin_supervisor = PluginSupervisor(
+            self._builtin_supervisor = self._instantiate_supervisor(
+                PluginSupervisor,
                 plugin_dirs=builtin_dirs,
                 group_name="builtin",
+                hook_spec_registry=self._hook_spec_registry,
                 socket_path=builtin_socket,
             )
             self._register_capability_impls(self._builtin_supervisor)
 
         if third_party_dirs:
-            self._third_party_supervisor = PluginSupervisor(
+            self._third_party_supervisor = self._instantiate_supervisor(
+                PluginSupervisor,
                 plugin_dirs=third_party_dirs,
                 group_name="third_party",
+                hook_spec_registry=self._hook_spec_registry,
                 socket_path=third_party_socket,
             )
             self._register_capability_impls(self._third_party_supervisor)
@@ -328,6 +369,7 @@ class PluginRuntimeManager(
             spec: 需要注册的 Hook 规格。
         """
 
+        self.ensure_builtin_hook_specs_registered()
         self._hook_dispatcher.register_hook_spec(spec)
 
     def register_hook_specs(self, specs: Sequence[HookSpec]) -> None:
@@ -337,8 +379,41 @@ class PluginRuntimeManager(
             specs: 需要注册的 Hook 规格序列。
         """
 
+        self.ensure_builtin_hook_specs_registered()
         self._hook_dispatcher.register_hook_specs(specs)
 
+    def unregister_hook_spec(self, hook_name: str) -> bool:
+        """注销指定命名 Hook 规格。
+
+        Args:
+            hook_name: 目标 Hook 名称。
+
+        Returns:
+            bool: 是否成功注销。
+        """
+
+        self.ensure_builtin_hook_specs_registered()
+        return self._hook_dispatcher.unregister_hook_spec(hook_name)
+
+    def list_hook_specs(self) -> List[HookSpec]:
+        """返回当前全部命名 Hook 规格。
+
+        Returns:
+            List[HookSpec]: 当前已注册的 Hook 规格列表。
+        """
+
+        self.ensure_builtin_hook_specs_registered()
+        return self._hook_dispatcher.list_hook_specs()
+
+    def ensure_builtin_hook_specs_registered(self) -> None:
+        """确保内置 Hook 规格已经注册到共享中心表。"""
+
+        if self._builtin_hook_specs_registered:
+            return
+
+        register_builtin_hook_specs(self._hook_spec_registry)
+        self._builtin_hook_specs_registered = True
+
     def _build_registered_dependency_map(self) -> Dict[str, Set[str]]:
         """根据当前已注册插件构建全局依赖图。"""
 
@@ -542,8 +617,8 @@ class PluginRuntimeManager(
             config_data: 待校验的配置内容。
 
         Returns:
-            Dict[str, Any] | None: 校验成功时返回规范化后的配置；若插件当前未加载
-            或运行时不可用，则返回 ``None`` 以便调用方回退到静态 Schema 方案。
+            Dict[str, Any] | None: 校验成功时返回规范化后的配置；若插件不存在、
+            当前不可路由或运行时不可用，则返回 ``None`` 以便调用方回退到弱推断方案。
 
         Raises:
             ValueError: 插件已加载，但配置校验失败时抛出。
@@ -558,6 +633,8 @@ class PluginRuntimeManager(
             logger.warning(f"插件 {plugin_id} 配置校验路由失败，将回退到静态 Schema: {exc}")
             return None
 
+        if supervisor is None:
+            supervisor = self._find_supervisor_by_plugin_directory(plugin_id)
         if supervisor is None:
             return None
 
@@ -569,6 +646,54 @@ class PluginRuntimeManager(
             logger.warning(f"插件 {plugin_id} 运行时配置校验不可用，将回退到静态 Schema: {exc}")
             return None
 
+    async def inspect_plugin_config(
+        self,
+        plugin_id: str,
+        config_data: Optional[Dict[str, Any]] = None,
+        *,
+        use_provided_config: bool = False,
+    ) -> InspectPluginConfigResultPayload | None:
+        """请求运行时解析插件配置元数据。
+
+        Args:
+            plugin_id: 目标插件 ID。
+            config_data: 可选的配置内容。
+            use_provided_config: 是否优先使用传入的配置内容而不是磁盘配置。
+
+        Returns:
+            InspectPluginConfigResultPayload | None: 解析成功时返回结构化结果；若插件
+            当前不可路由或运行时不可用，则返回 ``None``。
+
+        Raises:
+            ValueError: 插件存在，但运行时明确拒绝解析请求时抛出。
+        """
+
+        if not self._started:
+            return None
+
+        try:
+            supervisor = self._get_supervisor_for_plugin(plugin_id)
+        except RuntimeError as exc:
+            logger.warning(f"插件 {plugin_id} 配置解析路由失败: {exc}")
+            return None
+
+        if supervisor is None:
+            supervisor = self._find_supervisor_by_plugin_directory(plugin_id)
+        if supervisor is None:
+            return None
+
+        try:
+            return await supervisor.inspect_plugin_config(
+                plugin_id=plugin_id,
+                config_data=config_data,
+                use_provided_config=use_provided_config,
+            )
+        except ValueError:
+            raise
+        except Exception as exc:
+            logger.warning(f"插件 {plugin_id} 配置解析不可用: {exc}")
+            return None
+
     @staticmethod
     def _normalize_config_reload_scopes(changed_scopes: Sequence[str]) -> tuple[str, ...]:
         """规范化配置热重载范围列表。
@@ -771,7 +896,15 @@ class PluginRuntimeManager(
         return matches[0] if matches else None
 
     async def load_plugin_globally(self, plugin_id: str, reason: str = "manual") -> bool:
-        """加载或重载单个插件，并为其补齐跨 Supervisor 外部依赖。"""
+        """加载或重载单个插件，并为其补齐跨 Supervisor 外部依赖。
+
+        Args:
+            plugin_id: 目标插件 ID。
+            reason: 加载或重载原因。
+
+        Returns:
+            bool: 插件最终是否处于已加载状态。
+        """
 
         normalized_plugin_id = str(plugin_id or "").strip()
         if not normalized_plugin_id:
@@ -789,11 +922,12 @@ class PluginRuntimeManager(
         if supervisor is None:
             return False
 
-        return await supervisor.reload_plugins(
+        reloaded = await supervisor.reload_plugins(
             plugin_ids=[normalized_plugin_id],
             reason=reason,
             external_available_plugins=self._build_external_available_plugins_for_supervisor(supervisor),
         )
+        return reloaded and normalized_plugin_id in supervisor.get_loaded_plugin_ids()
 
     @classmethod
     def _find_duplicate_plugin_ids(cls, plugin_dirs: List[Path]) -> Dict[str, List[Path]]:
@@ -920,15 +1054,16 @@ class PluginRuntimeManager(
         return None
 
     def _refresh_plugin_config_watch_subscriptions(self) -> None:
-        """按当前已注册插件集合刷新 config.toml 的单插件订阅。
+        """按当前可识别插件集合刷新 config.toml 的单插件订阅。
 
         当插件热重载后，插件集合或目录位置可能发生变化，因此需要重新对齐
         watcher 的订阅，确保每个插件配置变更只触发对应 plugin_id。
+        这里不仅覆盖当前已注册插件，也覆盖已存在但暂未激活的合法插件。
         """
         if self._plugin_file_watcher is None:
             return
 
-        desired_plugin_paths = dict(self._iter_registered_plugin_paths())
+        desired_plugin_paths = dict(self._iter_watchable_plugin_paths())
         self._plugin_path_cache = desired_plugin_paths.copy()
         desired_config_paths = {
             plugin_id: plugin_path / "config.toml" for plugin_id, plugin_path in desired_plugin_paths.items()
@@ -970,6 +1105,18 @@ class PluginRuntimeManager(
                 if plugin_path := self._get_plugin_path_for_supervisor(supervisor, plugin_id):
                     yield plugin_id, plugin_path
 
+    def _iter_watchable_plugin_paths(self) -> Iterable[Tuple[str, Path]]:
+        """迭代应被配置监听器追踪的插件目录。
+
+        Returns:
+            Iterable[Tuple[str, Path]]: ``(plugin_id, plugin_path)`` 迭代器。
+        """
+
+        watchable_plugin_paths = dict(self._iter_discovered_plugin_paths(self._iter_plugin_dirs()))
+        for plugin_id, plugin_path in self._iter_registered_plugin_paths():
+            watchable_plugin_paths.setdefault(plugin_id, plugin_path)
+        yield from watchable_plugin_paths.items()
+
     def _get_plugin_config_path_for_supervisor(self, supervisor: Any, plugin_id: str) -> Optional[Path]:
         """从指定 Supervisor 的插件目录中定位某个插件的 config.toml。"""
         plugin_path = self._get_plugin_path_for_supervisor(supervisor, plugin_id)
@@ -993,18 +1140,43 @@ class PluginRuntimeManager(
             return
 
         if supervisor is None:
+            supervisor = self._find_supervisor_by_plugin_directory(plugin_id)
+        if supervisor is None:
+            return
+
+        plugin_is_loaded = plugin_id in getattr(supervisor, "_registered_plugins", {})
+
+        try:
+            snapshot = await supervisor.inspect_plugin_config(plugin_id)
+        except Exception as exc:
+            logger.warning(f"插件 {plugin_id} 配置文件变更解析失败: {exc}")
             return
 
         try:
-            config_payload = self._load_plugin_config_for_supervisor(supervisor, plugin_id)
-            delivered = await supervisor.notify_plugin_config_updated(
-                plugin_id=plugin_id,
-                config_data=config_payload,
-                config_version="",
-                config_scope="self",
-            )
-            if not delivered:
-                logger.warning(f"插件 {plugin_id} 配置文件变更后通知失败")
+            if plugin_is_loaded and snapshot.enabled:
+                delivered = await supervisor.notify_plugin_config_updated(
+                    plugin_id=plugin_id,
+                    config_data=dict(snapshot.normalized_config),
+                    config_version="",
+                    config_scope="self",
+                )
+                if not delivered:
+                    logger.warning(f"插件 {plugin_id} 配置文件变更后通知失败")
+                return
+
+            if plugin_is_loaded and not snapshot.enabled:
+                reloaded = await self.reload_plugins_globally([plugin_id], reason="config_disabled")
+                if not reloaded:
+                    logger.warning(f"插件 {plugin_id} 禁用配置已写入，但运行时卸载失败")
+                return
+
+            if not snapshot.enabled:
+                logger.info(f"插件 {plugin_id} 当前处于禁用状态，跳过自动加载")
+                return
+
+            loaded = await self.load_plugin_globally(plugin_id, reason="config_enabled")
+            if not loaded:
+                logger.warning(f"插件 {plugin_id} 配置文件变更后自动加载失败")
         except Exception as exc:
             logger.warning(f"插件 {plugin_id} 配置文件变更处理失败: {exc}")
 

src/plugin_runtime/protocol/envelope.py

@@ -288,6 +288,8 @@ class RunnerReadyPayload(BaseModel):
     """已完成初始化的插件列表"""
     failed_plugins: List[str] = Field(default_factory=list, description="初始化失败的插件列表")
     """初始化失败的插件列表"""
+    inactive_plugins: List[str] = Field(default_factory=list, description="当前因禁用或依赖不可用而未激活的插件列表")
+    """当前因禁用或依赖不可用而未激活的插件列表"""
 
 
 # ====== 配置更新 ======
@@ -311,6 +313,32 @@ class ValidatePluginConfigPayload(BaseModel):
     """待校验的配置内容"""
 
 
+class InspectPluginConfigPayload(BaseModel):
+    """plugin.inspect_config 请求 payload。"""
+
+    config_data: Dict[str, Any] = Field(default_factory=dict, description="可选的配置内容")
+    """可选的配置内容"""
+    use_provided_config: bool = Field(default=False, description="是否优先使用请求中携带的配置内容")
+    """是否优先使用请求中携带的配置内容"""
+
+
+class InspectPluginConfigResultPayload(BaseModel):
+    """plugin.inspect_config 响应 payload。"""
+
+    success: bool = Field(description="是否解析成功")
+    """是否解析成功"""
+    default_config: Dict[str, Any] = Field(default_factory=dict, description="插件默认配置")
+    """插件默认配置"""
+    config_schema: Dict[str, Any] = Field(default_factory=dict, description="插件配置 Schema")
+    """插件配置 Schema"""
+    normalized_config: Dict[str, Any] = Field(default_factory=dict, description="归一化后的配置内容")
+    """归一化后的配置内容"""
+    changed: bool = Field(default=False, description="是否在归一化过程中自动补齐或修正了配置")
+    """是否在归一化过程中自动补齐或修正了配置"""
+    enabled: bool = Field(default=True, description="插件在当前配置下是否应被视为启用")
+    """插件在当前配置下是否应被视为启用"""
+
+
 class ValidatePluginConfigResultPayload(BaseModel):
     """plugin.validate_config 响应 payload。"""
 
@@ -380,6 +408,8 @@ class ReloadPluginResultPayload(BaseModel):
     """成功完成重载的插件列表"""
     unloaded_plugins: List[str] = Field(default_factory=list, description="本次已卸载的插件列表")
     """本次已卸载的插件列表"""
+    inactive_plugins: List[str] = Field(default_factory=list, description="本次处于未激活状态的插件列表")
+    """本次处于未激活状态的插件列表"""
     failed_plugins: Dict[str, str] = Field(default_factory=dict, description="重载失败的插件及原因")
     """重载失败的插件及原因"""
 
@@ -395,6 +425,8 @@ class ReloadPluginsResultPayload(BaseModel):
     """成功完成重载的插件列表"""
     unloaded_plugins: List[str] = Field(default_factory=list, description="本次已卸载的插件列表")
     """本次已卸载的插件列表"""
+    inactive_plugins: List[str] = Field(default_factory=list, description="本次处于未激活状态的插件列表")
+    """本次处于未激活状态的插件列表"""
     failed_plugins: Dict[str, str] = Field(default_factory=dict, description="重载失败的插件及原因")
     """重载失败的插件及原因"""
 

src/plugin_runtime/runner/runner_main.py

@@ -9,8 +9,10 @@
 6. 转发插件的能力调用到 Host
 """
 
+from collections.abc import Mapping
+from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Mapping, Optional, Protocol, Set, Tuple, cast
+from typing import Any, Callable, Dict, List, Optional, Protocol, Set, Tuple, cast
 
 import asyncio
 import contextlib
@@ -39,6 +41,8 @@ from src.plugin_runtime.protocol.envelope import (
     ConfigUpdatedPayload,
     Envelope,
     HealthPayload,
+    InspectPluginConfigPayload,
+    InspectPluginConfigResultPayload,
     InvokePayload,
     InvokeResultPayload,
     RegisterPluginPayload,
@@ -141,6 +145,14 @@ class _ConfigAwarePlugin(Protocol):
         ...
 
 
+class PluginActivationStatus(str, Enum):
+    """描述插件激活结果。"""
+
+    LOADED = "loaded"
+    INACTIVE = "inactive"
+    FAILED = "failed"
+
+
 def _install_shutdown_signal_handlers(
     mark_runner_shutting_down: Callable[[], None],
     loop: Optional[asyncio.AbstractEventLoop] = None,
@@ -236,13 +248,43 @@ class PluginRunner:
 
         # 4. 注入 PluginContext + 调用 on_load 生命周期钩子
         failed_plugins: Set[str] = set(self._loader.failed_plugins.keys())
+        inactive_plugins: Set[str] = set()
+        available_plugin_versions: Dict[str, str] = dict(self._external_available_plugins)
         for meta in plugins:
-            ok = await self._activate_plugin(meta)
-            if not ok:
+            unsatisfied_dependencies = [
+                dependency.id
+                for dependency in meta.manifest.plugin_dependencies
+                if dependency.id not in available_plugin_versions
+                or not self._loader.manifest_validator.is_plugin_dependency_satisfied(
+                    dependency,
+                    available_plugin_versions[dependency.id],
+                )
+            ]
+            if unsatisfied_dependencies:
+                if any(dependency_id in inactive_plugins for dependency_id in unsatisfied_dependencies):
+                    logger.info(
+                        f"插件 {meta.plugin_id} 依赖的插件当前未激活，跳过本次启动: {', '.join(unsatisfied_dependencies)}"
+                    )
+                    inactive_plugins.add(meta.plugin_id)
+                    continue
                 failed_plugins.add(meta.plugin_id)
+                continue
 
-        successful_plugins = [meta.plugin_id for meta in plugins if meta.plugin_id not in failed_plugins]
-        await self._notify_ready(successful_plugins, sorted(failed_plugins))
+            activation_status = await self._activate_plugin(meta)
+            if activation_status == PluginActivationStatus.LOADED:
+                available_plugin_versions[meta.plugin_id] = meta.version
+                continue
+            if activation_status == PluginActivationStatus.INACTIVE:
+                inactive_plugins.add(meta.plugin_id)
+                continue
+            failed_plugins.add(meta.plugin_id)
+
+        successful_plugins = [
+            meta.plugin_id
+            for meta in plugins
+            if meta.plugin_id not in failed_plugins and meta.plugin_id not in inactive_plugins
+        ]
+        await self._notify_ready(successful_plugins, sorted(failed_plugins), sorted(inactive_plugins))
 
         # 5. 等待直到收到关停信号
         with contextlib.suppress(asyncio.CancelledError):
@@ -352,17 +394,17 @@ class PluginRunner:
         cast(_ContextAwarePlugin, instance)._set_context(ctx)
         logger.debug(f"已为插件 {plugin_id} 注入 PluginContext")
 
-    def _apply_plugin_config(self, meta: PluginMeta, config_data: Optional[Dict[str, Any]] = None) -> None:
+    def _apply_plugin_config(self, meta: PluginMeta, config_data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
         """在 Runner 侧为插件实例注入当前插件配置。
 
         Args:
             meta: 插件元数据。
             config_data: 可选的配置数据；留空时自动从插件目录读取。
+
+        Returns:
+            Dict[str, Any]: 归一化后的当前插件配置。
         """
         instance = meta.instance
-        if not hasattr(instance, "set_plugin_config"):
-            return
-
         raw_config = config_data if config_data is not None else self._load_plugin_config(meta.plugin_dir)
         plugin_config, should_persist = self._normalize_plugin_config(instance, raw_config)
         config_path = Path(meta.plugin_dir) / "config.toml"
@@ -370,10 +412,12 @@ class PluginRunner:
         should_initialize_file = not config_path.exists() and bool(default_config)
         if should_persist or should_initialize_file:
             self._save_plugin_config(meta.plugin_dir, plugin_config)
-        try:
-            cast(_ConfigAwarePlugin, instance).set_plugin_config(plugin_config)
-        except Exception as exc:
-            logger.warning(f"插件 {meta.plugin_id} 配置注入失败: {exc}")
+        if hasattr(instance, "set_plugin_config"):
+            try:
+                cast(_ConfigAwarePlugin, instance).set_plugin_config(plugin_config)
+            except Exception as exc:
+                logger.warning(f"插件 {meta.plugin_id} 配置注入失败: {exc}")
+        return plugin_config
 
     def _normalize_plugin_config(
         self,
@@ -405,6 +449,33 @@ class PluginRunner:
             logger.warning(f"插件配置归一化失败，将回退为原始配置: {exc}")
             return normalized_config, False
 
+    @staticmethod
+    def _is_plugin_enabled(config_data: Optional[Mapping[str, Any]]) -> bool:
+        """根据配置内容判断插件是否应被视为启用。
+
+        Args:
+            config_data: 当前插件配置。
+
+        Returns:
+            bool: 插件是否启用。
+        """
+
+        if not isinstance(config_data, Mapping):
+            return True
+
+        plugin_section = config_data.get("plugin")
+        if not isinstance(plugin_section, Mapping):
+            return True
+
+        enabled_value = plugin_section.get("enabled", True)
+        if isinstance(enabled_value, str):
+            normalized_value = enabled_value.strip().lower()
+            if normalized_value in {"0", "false", "no", "off"}:
+                return False
+            if normalized_value in {"1", "true", "yes", "on"}:
+                return True
+        return bool(enabled_value)
+
     @staticmethod
     def _save_plugin_config(plugin_dir: str, config_data: Dict[str, Any]) -> None:
         """将插件配置写回到 ``config.toml``。
@@ -435,6 +506,99 @@ class PluginRunner:
 
         return loaded if isinstance(loaded, dict) else {}
 
+    def _resolve_plugin_candidate(self, plugin_id: str) -> Tuple[Optional[PluginCandidate], Optional[str]]:
+        """解析指定插件的候选目录。
+
+        Args:
+            plugin_id: 目标插件 ID。
+
+        Returns:
+            Tuple[Optional[PluginCandidate], Optional[str]]: 候选插件与错误信息。
+        """
+
+        candidates, duplicate_candidates = self._loader.discover_candidates(self._plugin_dirs)
+        if plugin_id in duplicate_candidates:
+            conflict_paths = ", ".join(str(path) for path in duplicate_candidates[plugin_id])
+            return None, f"检测到重复插件 ID: {conflict_paths}"
+
+        candidate = candidates.get(plugin_id)
+        if candidate is None:
+            return None, f"未找到插件: {plugin_id}"
+        return candidate, None
+
+    def _resolve_plugin_meta_for_config_request(
+        self,
+        plugin_id: str,
+    ) -> Tuple[Optional[PluginMeta], bool, Optional[str]]:
+        """为配置相关请求解析插件元数据。
+
+        Args:
+            plugin_id: 目标插件 ID。
+
+        Returns:
+            Tuple[Optional[PluginMeta], bool, Optional[str]]: 依次为插件元数据、
+            是否为临时冷加载实例、以及错误信息。
+        """
+
+        loaded_meta = self._loader.get_plugin(plugin_id)
+        if loaded_meta is not None:
+            return loaded_meta, False, None
+
+        candidate, error_message = self._resolve_plugin_candidate(plugin_id)
+        if candidate is None:
+            return None, False, error_message
+
+        try:
+            meta = self._loader.load_candidate(plugin_id, candidate)
+        except Exception as exc:
+            return None, False, str(exc)
+        if meta is None:
+            return None, False, "插件模块加载失败"
+        return meta, True, None
+
+    def _inspect_plugin_config(
+        self,
+        meta: PluginMeta,
+        *,
+        config_data: Optional[Dict[str, Any]] = None,
+        use_provided_config: bool = False,
+        suppress_errors: bool = True,
+    ) -> InspectPluginConfigResultPayload:
+        """解析插件代码定义的配置元数据。
+
+        Args:
+            meta: 插件元数据。
+            config_data: 可选的配置内容。
+            use_provided_config: 是否优先使用传入的配置内容。
+            suppress_errors: 是否在归一化失败时回退原始配置。
+
+        Returns:
+            InspectPluginConfigResultPayload: 结构化解析结果。
+        """
+
+        raw_config = config_data if use_provided_config else self._load_plugin_config(meta.plugin_dir)
+        if use_provided_config and config_data is None:
+            raw_config = {}
+
+        normalized_config, changed = self._normalize_plugin_config(
+            meta.instance,
+            raw_config,
+            suppress_errors=suppress_errors,
+        )
+        default_config = self._get_plugin_default_config(meta.instance)
+        if not normalized_config and not raw_config and default_config:
+            normalized_config = dict(default_config)
+            changed = True
+
+        return InspectPluginConfigResultPayload(
+            success=True,
+            default_config=default_config,
+            config_schema=self._get_plugin_config_schema(meta),
+            normalized_config=normalized_config,
+            changed=changed,
+            enabled=self._is_plugin_enabled(normalized_config),
+        )
+
     def _register_handlers(self) -> None:
         """注册 Host -> Runner 的方法处理器。"""
         self._rpc_client.register_method("plugin.invoke_command", self._handle_invoke)
@@ -448,6 +612,7 @@ class PluginRunner:
         self._rpc_client.register_method("plugin.prepare_shutdown", self._handle_prepare_shutdown)
         self._rpc_client.register_method("plugin.shutdown", self._handle_shutdown)
         self._rpc_client.register_method("plugin.config_updated", self._handle_config_updated)
+        self._rpc_client.register_method("plugin.inspect_config", self._handle_inspect_plugin_config)
         self._rpc_client.register_method("plugin.validate_config", self._handle_validate_plugin_config)
         self._rpc_client.register_method("plugin.reload", self._handle_reload_plugin)
         self._rpc_client.register_method("plugin.reload_batch", self._handle_reload_plugins)
@@ -579,6 +744,9 @@ class PluginRunner:
             )
             if response.error:
                 raise RuntimeError(response.error.get("message", "插件注册失败"))
+            response_payload = response.payload if isinstance(response.payload, dict) else {}
+            if not bool(response_payload.get("accepted", True)):
+                raise RuntimeError(str(response_payload.get("reason", "插件注册失败")))
             logger.info(f"插件 {meta.plugin_id} 注册完成")
             return True
         except Exception as e:
@@ -689,36 +857,40 @@ class PluginRunner:
         except Exception as exc:
             logger.error(f"插件 {meta.plugin_id} on_unload 失败: {exc}", exc_info=True)
 
-    async def _activate_plugin(self, meta: PluginMeta) -> bool:
+    async def _activate_plugin(self, meta: PluginMeta) -> PluginActivationStatus:
         """完成插件注入、授权、生命周期和组件注册。
 
         Args:
             meta: 待激活的插件元数据。
 
         Returns:
-            bool: 是否激活成功。
+            PluginActivationStatus: 插件激活结果。
         """
         self._inject_context(meta.plugin_id, meta.instance)
-        self._apply_plugin_config(meta)
+        plugin_config = self._apply_plugin_config(meta)
+        if not self._is_plugin_enabled(plugin_config):
+            logger.info(f"插件 {meta.plugin_id} 已在配置中禁用，跳过激活")
+            self._loader.purge_plugin_modules(meta.plugin_id, meta.plugin_dir)
+            return PluginActivationStatus.INACTIVE
 
         if not await self._bootstrap_plugin(meta):
             self._loader.purge_plugin_modules(meta.plugin_id, meta.plugin_dir)
-            return False
+            return PluginActivationStatus.FAILED
 
         if not await self._register_plugin(meta):
             await self._invoke_plugin_on_unload(meta)
             await self._deactivate_plugin(meta)
             self._loader.purge_plugin_modules(meta.plugin_id, meta.plugin_dir)
-            return False
+            return PluginActivationStatus.FAILED
 
         if not await self._invoke_plugin_on_load(meta):
             await self._unregister_plugin(meta.plugin_id, reason="on_load_failed")
             await self._deactivate_plugin(meta)
             self._loader.purge_plugin_modules(meta.plugin_id, meta.plugin_dir)
-            return False
+            return PluginActivationStatus.FAILED
 
         self._loader.set_loaded_plugin(meta)
-        return True
+        return PluginActivationStatus.LOADED
 
     async def _unload_plugin(self, meta: PluginMeta, reason: str, *, purge_modules: bool = True) -> None:
         """卸载单个插件并清理 Host/Runner 两侧状态。
@@ -879,6 +1051,7 @@ class PluginRunner:
             requested_plugin_id=plugin_id,
             reloaded_plugins=batch_result.reloaded_plugins,
             unloaded_plugins=batch_result.unloaded_plugins,
+            inactive_plugins=batch_result.inactive_plugins,
             failed_plugins=batch_result.failed_plugins,
         )
 
@@ -973,6 +1146,8 @@ class PluginRunner:
             },
         }
         reloaded_plugins: List[str] = []
+        inactive_plugins: List[str] = []
+        inactive_plugin_ids: Set[str] = set()
 
         for load_plugin_id in load_order:
             if load_plugin_id in failed_plugins:
@@ -983,10 +1158,28 @@ class PluginRunner:
                 continue
 
             _, manifest, _ = candidate
+            unsatisfied_dependency_ids = [
+                dependency.id
+                for dependency in manifest.plugin_dependencies
+                if dependency.id not in available_plugins
+                or not self._loader.manifest_validator.is_plugin_dependency_satisfied(
+                    dependency,
+                    available_plugins[dependency.id],
+                )
+            ]
             if unsatisfied_dependencies := self._loader.manifest_validator.get_unsatisfied_plugin_dependencies(
                 manifest,
                 available_plugin_versions=available_plugins,
             ):
+                if load_plugin_id not in reload_root_ids and any(
+                    dependency_id in inactive_plugin_ids for dependency_id in unsatisfied_dependency_ids
+                ):
+                    logger.info(
+                        f"插件 {load_plugin_id} 的依赖当前未激活，保留为未激活状态: {', '.join(unsatisfied_dependencies)}"
+                    )
+                    inactive_plugin_ids.add(load_plugin_id)
+                    inactive_plugins.append(load_plugin_id)
+                    continue
                 failed_plugins[load_plugin_id] = f"依赖未满足: {', '.join(unsatisfied_dependencies)}"
                 continue
 
@@ -996,9 +1189,13 @@ class PluginRunner:
                 continue
 
             activated = await self._activate_plugin(meta)
-            if not activated:
+            if activated == PluginActivationStatus.FAILED:
                 failed_plugins[load_plugin_id] = "插件初始化失败"
                 continue
+            if activated == PluginActivationStatus.INACTIVE:
+                inactive_plugin_ids.add(load_plugin_id)
+                inactive_plugins.append(load_plugin_id)
+                continue
 
             available_plugins[load_plugin_id] = meta.version
             reloaded_plugins.append(load_plugin_id)
@@ -1033,7 +1230,7 @@ class PluginRunner:
                     rollback_failures[rollback_plugin_id] = str(exc)
                     continue
 
-                if not restored:
+                if restored != PluginActivationStatus.LOADED:
                     rollback_failures[rollback_plugin_id] = "无法重新激活旧版本"
 
             return ReloadPluginsResultPayload(
@@ -1041,29 +1238,40 @@ class PluginRunner:
                 requested_plugin_ids=normalized_plugin_ids,
                 reloaded_plugins=[],
                 unloaded_plugins=unloaded_plugins,
+                inactive_plugins=[],
                 failed_plugins=self._finalize_failed_reload_messages(failed_plugins, rollback_failures),
             )
 
-        requested_plugin_success = all(plugin_id in reloaded_plugins for plugin_id in reload_root_ids)
+        requested_plugin_success = all(
+            plugin_id in reloaded_plugins or plugin_id in inactive_plugins for plugin_id in reload_root_ids
+        )
 
         return ReloadPluginsResultPayload(
             success=requested_plugin_success and not failed_plugins,
             requested_plugin_ids=normalized_plugin_ids,
             reloaded_plugins=reloaded_plugins,
             unloaded_plugins=unloaded_plugins,
+            inactive_plugins=inactive_plugins,
             failed_plugins=failed_plugins,
         )
 
-    async def _notify_ready(self, loaded_plugins: List[str], failed_plugins: List[str]) -> None:
+    async def _notify_ready(
+        self,
+        loaded_plugins: List[str],
+        failed_plugins: List[str],
+        inactive_plugins: List[str],
+    ) -> None:
         """通知 Host 当前 Runner 已完成插件初始化。
 
         Args:
             loaded_plugins: 成功初始化的插件列表。
             failed_plugins: 初始化失败的插件列表。
+            inactive_plugins: 因禁用或依赖不可用而未激活的插件列表。
         """
         payload = RunnerReadyPayload(
             loaded_plugins=loaded_plugins,
             failed_plugins=failed_plugins,
+            inactive_plugins=inactive_plugins,
         )
         await self._rpc_client.send_request(
             "runner.ready",
@@ -1289,6 +1497,44 @@ class PluginRunner:
                 return envelope.make_error_response(ErrorCode.E_UNKNOWN.value, str(e))
         return envelope.make_response(payload={"acknowledged": True})
 
+    async def _handle_inspect_plugin_config(self, envelope: Envelope) -> Envelope:
+        """处理插件配置元数据解析请求。
+
+        Args:
+            envelope: RPC 请求信封。
+
+        Returns:
+            Envelope: RPC 响应信封。
+        """
+
+        try:
+            payload = InspectPluginConfigPayload.model_validate(envelope.payload)
+        except Exception as exc:
+            return envelope.make_error_response(ErrorCode.E_BAD_PAYLOAD.value, str(exc))
+
+        plugin_id = envelope.plugin_id
+        meta, is_temporary_meta, error_message = self._resolve_plugin_meta_for_config_request(plugin_id)
+        if meta is None:
+            return envelope.make_error_response(
+                ErrorCode.E_PLUGIN_NOT_FOUND.value,
+                error_message or f"未找到插件: {plugin_id}",
+            )
+
+        try:
+            result = self._inspect_plugin_config(
+                meta,
+                config_data=payload.config_data,
+                use_provided_config=payload.use_provided_config,
+                suppress_errors=True,
+            )
+        except Exception as exc:
+            return envelope.make_error_response(ErrorCode.E_BAD_PAYLOAD.value, str(exc))
+        finally:
+            if is_temporary_meta:
+                self._loader.purge_plugin_modules(plugin_id, meta.plugin_dir)
+
+        return envelope.make_response(payload=result.model_dump())
+
     async def _handle_validate_plugin_config(self, envelope: Envelope) -> Envelope:
         """处理插件配置校验请求。
 
@@ -1305,23 +1551,30 @@ class PluginRunner:
             return envelope.make_error_response(ErrorCode.E_BAD_PAYLOAD.value, str(exc))
 
         plugin_id = envelope.plugin_id
-        meta = self._loader.get_plugin(plugin_id)
+        meta, is_temporary_meta, error_message = self._resolve_plugin_meta_for_config_request(plugin_id)
         if meta is None:
-            return envelope.make_error_response(ErrorCode.E_PLUGIN_NOT_FOUND.value, f"未找到插件: {plugin_id}")
+            return envelope.make_error_response(
+                ErrorCode.E_PLUGIN_NOT_FOUND.value,
+                error_message or f"未找到插件: {plugin_id}",
+            )
 
         try:
-            normalized_config, changed = self._normalize_plugin_config(
-                meta.instance,
-                payload.config_data,
+            inspection_result = self._inspect_plugin_config(
+                meta,
+                config_data=payload.config_data,
+                use_provided_config=True,
                 suppress_errors=False,
             )
         except Exception as exc:
             return envelope.make_error_response(ErrorCode.E_BAD_PAYLOAD.value, str(exc))
+        finally:
+            if is_temporary_meta:
+                self._loader.purge_plugin_modules(plugin_id, meta.plugin_dir)
 
         result = ValidatePluginConfigResultPayload(
             success=True,
-            normalized_config=normalized_config,
-            changed=changed,
+            normalized_config=inspection_result.normalized_config,
+            changed=inspection_result.changed,
         )
         return envelope.make_response(payload=result.model_dump())