Refactor plugin runtime components and enhance message handling

- Removed unused core action mirror functionality from PluginRunnerSupervisor.
- Simplified action and command execution logic in send_service.py.
- Introduced ComponentQueryService for unified component querying in plugin runtime.
- Enhanced message component handling with new binary component support.
- Improved message sequence construction and detection of outbound message flags.
- Updated methods for sending messages to streamline the process and improve readability.

pytests/test_platform_io_dedupe.py

@@ -72,10 +72,10 @@ class _StubPlatformIODriver(PlatformIODriver):
 
 
 def _build_manager() -> PlatformIOManager:
-    """构造带有最小 active owner 的 Broker 管理器。
+    """构造带有最小接收路由的 Broker 管理器。
 
     Returns:
-        PlatformIOManager: 已注册测试驱动并绑定活动路由的 Broker。
+        PlatformIOManager: 已注册测试驱动并绑定接收路由的 Broker。
     """
     manager = PlatformIOManager()
     driver = _StubPlatformIODriver(
@@ -88,7 +88,7 @@ def _build_manager() -> PlatformIOManager:
         )
     )
     manager.register_driver(driver)
-    manager.bind_route(
+    manager.bind_receive_route(
         RouteBinding(
             route_key=RouteKey(platform="qq", account_id="10001", scope="main"),
             driver_id=driver.driver_id,

pytests/test_plugin_runtime_action_bridge.py

@@ -1,57 +1,109 @@
+"""核心组件查询层与插件运行时聚合测试。"""
+
 from types import SimpleNamespace
 from typing import Any
 
 import pytest
 
-from src.core.component_registry import component_registry as core_component_registry
+import src.plugin_runtime.integration as integration_module
+
+from src.core.types import ActionInfo, ToolInfo
+from src.plugin_runtime.component_query import component_query_service
 from src.plugin_runtime.host.supervisor import PluginSupervisor
-from src.plugin_runtime.protocol.envelope import ComponentDeclaration, RegisterPluginPayload
 
 
-def _build_action_payload(plugin_id: str, action_name: str) -> RegisterPluginPayload:
+class _FakeRuntimeManager:
-    """构造用于测试的 runtime Action 注册载荷。
+    """测试用插件运行时管理器。"""
+
+    def __init__(self, supervisor: PluginSupervisor, plugin_id: str, plugin_config: dict[str, Any]) -> None:
+        """初始化测试用运行时管理器。
+
+        Args:
+            supervisor: 持有测试组件的监督器。
+            plugin_id: 目标插件 ID。
+            plugin_config: 需要返回的插件配置。
+        """
+
+        self.supervisors = [supervisor]
+        self._plugin_id = plugin_id
+        self._plugin_config = plugin_config
+
+    def _get_supervisor_for_plugin(self, plugin_id: str) -> PluginSupervisor | None:
+        """按插件 ID 返回对应监督器。
+
+        Args:
+            plugin_id: 目标插件 ID。
+
+        Returns:
+            PluginSupervisor | None: 命中时返回监督器。
+        """
+
+        return self.supervisors[0] if plugin_id == self._plugin_id else None
+
+    def _load_plugin_config_for_supervisor(self, supervisor: Any, plugin_id: str) -> dict[str, Any]:
+        """返回测试配置。
+
+        Args:
+            supervisor: 监督器实例。
+            plugin_id: 目标插件 ID。
+
+        Returns:
+            dict[str, Any]: 测试配置内容。
+        """
+
+        del supervisor
+        if plugin_id != self._plugin_id:
+            return {}
+        return dict(self._plugin_config)
+
+
+def _install_runtime_manager(
+    monkeypatch: pytest.MonkeyPatch,
+    supervisor: PluginSupervisor,
+    plugin_id: str,
+    plugin_config: dict[str, Any] | None = None,
+) -> None:
+    """为测试安装假的运行时管理器。
 
     Args:
-        plugin_id: 插件 ID。
+        monkeypatch: pytest monkeypatch 对象。
-        action_name: Action 名称。
+        supervisor: 持有测试组件的监督器。
-
+        plugin_id: 测试插件 ID。
-    Returns:
+        plugin_config: 可选的测试配置内容。
-        RegisterPluginPayload: 测试用注册载荷。
     """
-    return RegisterPluginPayload(
+
-        plugin_id=plugin_id,
+    fake_manager = _FakeRuntimeManager(supervisor, plugin_id, plugin_config or {"enabled": True})
-        plugin_version="1.0.0",
+    monkeypatch.setattr(integration_module, "get_plugin_runtime_manager", lambda: fake_manager)
-        components=[
-            ComponentDeclaration(
-                name=action_name,
-                component_type="ACTION",
-                plugin_id=plugin_id,
-                metadata={
-                    "description": "发送一个测试回复",
-                    "enabled": True,
-                    "activation_type": "keyword",
-                    "activation_probability": 0.25,
-                    "activation_keywords": ["测试", "hello"],
-                    "action_parameters": {"target": "目标对象"},
-                    "action_require": ["需要发送回复时使用"],
-                    "associated_types": ["text"],
-                    "parallel_action": True,
-                },
-            )
-        ],
-    )
 
 
 @pytest.mark.asyncio
-async def test_runtime_actions_are_mirrored_into_core_registry_and_invoked(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_core_component_registry_reads_runtime_action_and_executor(
-    """运行时 Action 应镜像到旧核心注册表，并可由旧 Planner 执行。"""
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """核心查询层应直接读取运行时 Action，并返回 RPC 执行闭包。"""
+
     plugin_id = "runtime_action_bridge_plugin"
     action_name = "runtime_action_bridge_test"
-    payload = _build_action_payload(plugin_id=plugin_id, action_name=action_name)
     supervisor = PluginSupervisor(plugin_dirs=[])
     captured: dict[str, Any] = {}
 
-    core_component_registry.remove_action(action_name)
+    supervisor.component_registry.register_component(
+        name=action_name,
+        component_type="ACTION",
+        plugin_id=plugin_id,
+        metadata={
+            "description": "发送一个测试回复",
+            "enabled": True,
+            "activation_type": "keyword",
+            "activation_probability": 0.25,
+            "activation_keywords": ["测试", "hello"],
+            "action_parameters": {"target": "目标对象"},
+            "action_require": ["需要发送回复时使用"],
+            "associated_types": ["text"],
+            "parallel_action": True,
+        },
+    )
+    _install_runtime_manager(monkeypatch, supervisor, plugin_id, {"enabled": True, "mode": "test"})
 
     async def fake_invoke_plugin(
         method: str,
@@ -60,18 +112,8 @@ async def test_runtime_actions_are_mirrored_into_core_registry_and_invoked(monke
         args: dict[str, Any] | None = None,
         timeout_ms: int = 30000,
     ) -> Any:
-        """模拟 plugin runtime Action 调用。
+        """模拟动作 RPC 调用。"""
 
-        Args:
-            method: RPC 方法名。
-            plugin_id: 插件 ID。
-            component_name: 组件名称。
-            args: 调用参数。
-            timeout_ms: RPC 超时时间。
-
-        Returns:
-            Any: 伪造的 RPC 响应对象。
-        """
         captured["method"] = method
         captured["plugin_id"] = plugin_id
         captured["component_name"] = component_name
@@ -81,58 +123,162 @@ async def test_runtime_actions_are_mirrored_into_core_registry_and_invoked(monke
 
     monkeypatch.setattr(supervisor, "invoke_plugin", fake_invoke_plugin)
 
-    try:
+    action_info = component_query_service.get_action_info(action_name)
-        supervisor._mirror_runtime_actions_to_core_registry(payload)
+    assert isinstance(action_info, ActionInfo)
+    assert action_info.plugin_name == plugin_id
+    assert action_info.description == "发送一个测试回复"
+    assert action_info.activation_keywords == ["测试", "hello"]
+    assert action_info.random_activation_probability == 0.25
+    assert action_info.parallel_action is True
+    assert action_name in component_query_service.get_default_actions()
+    assert component_query_service.get_plugin_config(plugin_id) == {"enabled": True, "mode": "test"}
 
-        action_info = core_component_registry.get_action_info(action_name)
+    executor = component_query_service.get_action_executor(action_name)
-        assert action_info is not None
+    assert executor is not None
-        assert action_info.plugin_name == plugin_id
-        assert action_info.description == "发送一个测试回复"
-        assert action_info.activation_keywords == ["测试", "hello"]
-        assert action_info.random_activation_probability == 0.25
-        assert action_info.parallel_action is True
 
-        executor = core_component_registry.get_action_executor(action_name)
+    success, reason = await executor(
-        assert executor is not None
+        action_data={"target": "MaiBot"},
+        action_reasoning="当前适合使用这个动作",
+        cycle_timers={"planner": 0.1},
+        thinking_id="tid-1",
+        chat_stream=SimpleNamespace(session_id="stream-1"),
+        log_prefix="[test]",
+        shutting_down=False,
+        plugin_config={"enabled": True},
+    )
 
-        success, reason = await executor(
+    assert success is True
-            action_data={"target": "MaiBot"},
+    assert reason == "runtime action executed"
-            action_reasoning="当前适合使用这个动作",
+    assert captured["method"] == "plugin.invoke_action"
-            cycle_timers={"planner": 0.1},
+    assert captured["plugin_id"] == plugin_id
-            thinking_id="tid-1",
+    assert captured["component_name"] == action_name
-            chat_stream=SimpleNamespace(session_id="stream-1"),
+    assert captured["args"]["stream_id"] == "stream-1"
-            log_prefix="[test]",
+    assert captured["args"]["chat_id"] == "stream-1"
-            shutting_down=False,
+    assert captured["args"]["reasoning"] == "当前适合使用这个动作"
-            plugin_config={"enabled": True},
+    assert captured["args"]["target"] == "MaiBot"
-        )
+    assert captured["args"]["action_data"] == {"target": "MaiBot"}
-
-        assert success is True
-        assert reason == "runtime action executed"
-        assert captured["method"] == "plugin.invoke_action"
-        assert captured["plugin_id"] == plugin_id
-        assert captured["component_name"] == action_name
-        assert captured["args"]["stream_id"] == "stream-1"
-        assert captured["args"]["chat_id"] == "stream-1"
-        assert captured["args"]["reasoning"] == "当前适合使用这个动作"
-        assert captured["args"]["target"] == "MaiBot"
-        assert captured["args"]["action_data"] == {"target": "MaiBot"}
-    finally:
-        supervisor._remove_core_action_mirrors(plugin_id)
-        core_component_registry.remove_action(action_name)
 
 
-def test_clear_runner_state_removes_mirrored_runtime_actions() -> None:
+@pytest.mark.asyncio
-    """清理 Runner 状态时应同步移除旧核心注册表中的镜像 Action。"""
+async def test_core_component_registry_reads_runtime_command_and_executor(
-    plugin_id = "runtime_action_bridge_cleanup_plugin"
+    monkeypatch: pytest.MonkeyPatch,
-    action_name = "runtime_action_bridge_cleanup_test"
+) -> None:
-    payload = _build_action_payload(plugin_id=plugin_id, action_name=action_name)
+    """核心查询层应直接使用运行时命令匹配与执行闭包。"""
+
+    plugin_id = "runtime_command_bridge_plugin"
+    command_name = "runtime_command_bridge_test"
+    supervisor = PluginSupervisor(plugin_dirs=[])
+    captured: dict[str, Any] = {}
+
+    supervisor.component_registry.register_component(
+        name=command_name,
+        component_type="COMMAND",
+        plugin_id=plugin_id,
+        metadata={
+            "description": "测试命令",
+            "enabled": True,
+            "command_pattern": r"^/test(?:\s+.+)?$",
+            "aliases": ["/hello"],
+            "intercept_message_level": 1,
+        },
+    )
+    _install_runtime_manager(monkeypatch, supervisor, plugin_id, {"mode": "command"})
+
+    async def fake_invoke_plugin(
+        method: str,
+        plugin_id: str,
+        component_name: str,
+        args: dict[str, Any] | None = None,
+        timeout_ms: int = 30000,
+    ) -> Any:
+        """模拟命令 RPC 调用。"""
+
+        captured["method"] = method
+        captured["plugin_id"] = plugin_id
+        captured["component_name"] = component_name
+        captured["args"] = args or {}
+        captured["timeout_ms"] = timeout_ms
+        return SimpleNamespace(payload={"success": True, "result": (True, "command ok", True)})
+
+    monkeypatch.setattr(supervisor, "invoke_plugin", fake_invoke_plugin)
+
+    matched = component_query_service.find_command_by_text("/test hello")
+    assert matched is not None
+    command_executor, matched_groups, command_info = matched
+
+    assert matched_groups == {}
+    assert command_info.plugin_name == plugin_id
+    assert command_info.command_pattern == r"^/test(?:\s+.+)?$"
+
+    success, response_text, intercept = await command_executor(
+        message=SimpleNamespace(processed_plain_text="/test hello", session_id="stream-2"),
+        plugin_config={"mode": "command"},
+        matched_groups=matched_groups,
+    )
+
+    assert success is True
+    assert response_text == "command ok"
+    assert intercept is True
+    assert captured["method"] == "plugin.invoke_command"
+    assert captured["plugin_id"] == plugin_id
+    assert captured["component_name"] == command_name
+    assert captured["args"]["text"] == "/test hello"
+    assert captured["args"]["stream_id"] == "stream-2"
+    assert captured["args"]["plugin_config"] == {"mode": "command"}
+
+
+@pytest.mark.asyncio
+async def test_core_component_registry_reads_runtime_tools_and_executor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """核心查询层应直接读取运行时 Tool，并返回 RPC 执行闭包。"""
+
+    plugin_id = "runtime_tool_bridge_plugin"
+    tool_name = "runtime_tool_bridge_test"
     supervisor = PluginSupervisor(plugin_dirs=[])
 
-    core_component_registry.remove_action(action_name)
+    supervisor.component_registry.register_component(
+        name=tool_name,
+        component_type="TOOL",
+        plugin_id=plugin_id,
+        metadata={
+            "description": "测试工具",
+            "enabled": True,
+            "parameters": [
+                {
+                    "name": "query",
+                    "param_type": "string",
+                    "description": "查询词",
+                    "required": True,
+                }
+            ],
+        },
+    )
+    _install_runtime_manager(monkeypatch, supervisor, plugin_id)
 
-    supervisor._mirror_runtime_actions_to_core_registry(payload)
+    async def fake_invoke_plugin(
-    assert core_component_registry.get_action_info(action_name) is not None
+        method: str,
+        plugin_id: str,
+        component_name: str,
+        args: dict[str, Any] | None = None,
+        timeout_ms: int = 30000,
+    ) -> Any:
+        """模拟工具 RPC 调用。"""
 
-    supervisor._clear_runner_state()
+        del timeout_ms
+        assert method == "plugin.invoke_tool"
+        assert plugin_id == "runtime_tool_bridge_plugin"
+        assert component_name == "runtime_tool_bridge_test"
+        assert args == {"query": "MaiBot"}
+        return SimpleNamespace(payload={"success": True, "result": {"content": "tool ok"}})
 
-    assert core_component_registry.get_action_info(action_name) is None
+    monkeypatch.setattr(supervisor, "invoke_plugin", fake_invoke_plugin)
+
+    tool_info = component_query_service.get_tool_info(tool_name)
+    assert isinstance(tool_info, ToolInfo)
+    assert tool_info.tool_description == "测试工具"
+    assert tool_name in component_query_service.get_llm_available_tools()
+
+    executor = component_query_service.get_tool_executor(tool_name)
+    assert executor is not None
+    assert await executor({"query": "MaiBot"}) == {"content": "tool ok"}

src/chat/brain_chat/brain_planner.py

@@ -1,30 +1,32 @@
+from datetime import datetime
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+
 import json
-import time
-import traceback
 import random
 import re
-from typing import Dict, Optional, Tuple, List, TYPE_CHECKING
+import time
-from rich.traceback import install
+import traceback
-from datetime import datetime
+
-from json_repair import repair_json
+from json_repair import repair_json
+from rich.traceback import install
 
-from src.llm_models.utils_model import LLMRequest
-from src.config.config import global_config, model_config
-from src.common.logger import get_logger
 from src.chat.logger.plan_reply_logger import PlanReplyLogger
+from src.chat.message_receive.chat_manager import chat_manager as _chat_manager
+from src.chat.planner_actions.action_manager import ActionManager
+from src.chat.utils.utils import get_chat_type_and_target_info
 from src.common.data_models.info_data_model import ActionPlannerInfo
+from src.common.logger import get_logger
 from src.common.utils.utils_action import ActionUtils
+from src.config.config import global_config, model_config
+from src.core.types import ActionActivationType, ActionInfo, ComponentType
+from src.llm_models.utils_model import LLMRequest
+from src.plugin_runtime.component_query import component_query_service
 from src.prompt.prompt_manager import prompt_manager
 from src.services.message_service import (
     build_readable_messages_with_id,
     get_actions_by_timestamp_with_chat,
     get_messages_before_time_in_chat,
 )
-from src.chat.utils.utils import get_chat_type_and_target_info
-from src.chat.planner_actions.action_manager import ActionManager
-from src.chat.message_receive.chat_manager import chat_manager as _chat_manager
-from src.core.types import ActionActivationType, ActionInfo, ComponentType
-from src.core.component_registry import component_registry
 
 if TYPE_CHECKING:
     from src.common.data_models.info_data_model import TargetPersonInfo
@@ -320,7 +322,7 @@ class BrainPlanner:
         current_available_actions_dict = self.action_manager.get_using_actions()
 
         # 获取完整的动作信息
-        all_registered_actions: Dict[str, ActionInfo] = component_registry.get_components_by_type(  # type: ignore
+        all_registered_actions: Dict[str, ActionInfo] = component_query_service.get_components_by_type(  # type: ignore
             ComponentType.ACTION
         )
         current_available_actions = {}

src/chat/message_receive/bot.py

@@ -1,19 +1,19 @@
 from contextlib import suppress
-import traceback
-import os
-
-from maim_message import MessageBase
 from typing import Any, Dict, Optional
 
+import os
+import traceback
 
+from maim_message import MessageBase
+
+from src.chat.heart_flow.heartflow_message_processor import HeartFCMessageReceiver
 from src.common.logger import get_logger
 from src.common.utils.utils_message import MessageUtils
 from src.common.utils.utils_session import SessionUtils
-from src.chat.heart_flow.heartflow_message_processor import HeartFCMessageReceiver
 
 # from src.chat.brain_chat.PFC.pfc_manager import PFCManager
 from src.core.announcement_manager import global_announcement_manager
-from src.core.component_registry import component_registry
+from src.plugin_runtime.component_query import component_query_service
 
 from .message import SessionMessage
 from .chat_manager import chat_manager
@@ -58,16 +58,22 @@ class ChatBot:
             logger.error(f"创建PFC聊天失败: {e}")
             logger.error(traceback.format_exc())
 
-    async def _process_commands(self, message: SessionMessage):
+    async def _process_commands(self, message: SessionMessage) -> tuple[bool, Optional[str], bool]:
-        # sourcery skip: use-named-expression
+        """使用统一组件注册表处理命令。
-        """使用新插件系统处理命令"""
+
+        Args:
+            message: 当前待处理的会话消息。
+
+        Returns:
+            tuple[bool, Optional[str], bool]: ``(是否命中命令, 命令响应文本, 是否继续后续处理)``。
+        """
         if not message.processed_plain_text:
             return False, None, True  # 没有文本内容，继续处理消息
         try:
             text = message.processed_plain_text
 
-            # 使用核心组件注册表查找命令
+            # 使用插件运行时统一查询服务查找命令
-            command_result = component_registry.find_command_by_text(text)
+            command_result = component_query_service.find_command_by_text(text)
             if command_result:
                 command_executor, matched_groups, command_info = command_result
                 plugin_name = command_info.plugin_name
@@ -81,7 +87,7 @@ class ChatBot:
                 message.is_command = True
 
                 # 获取插件配置
-                plugin_config = component_registry.get_plugin_config(plugin_name)
+                plugin_config = component_query_service.get_plugin_config(plugin_name)
 
                 try:
                     # 调用命令执行器
@@ -112,88 +118,32 @@ class ChatBot:
                     # 命令出错时，根据命令的拦截设置决定是否继续处理消息
                     return True, str(e), False  # 出错时继续处理消息
 
-            # 没有找到旧系统命令，尝试新版本插件运行时
+            return False, None, True
-            new_cmd_result = await self._process_new_runtime_command(message)
-            return new_cmd_result if new_cmd_result is not None else (False, None, True)
 
         except Exception as e:
             logger.error(f"处理命令时出错: {e}")
             return False, None, True  # 出错时继续处理消息
 
-    async def _process_new_runtime_command(self, message: SessionMessage):
-        """尝试在新版本插件运行时中查找并执行命令
-
-        Returns:
-            (found, response, continue_processing) 三元组，
-            或 None 表示新运行时中也未找到匹配命令。
-        """
-        from src.plugin_runtime.integration import get_plugin_runtime_manager
-
-        prm = get_plugin_runtime_manager()
-        if not prm.is_running:
-            return None
-
-        matched = prm.find_command_by_text(message.processed_plain_text)
-        if matched is None:
-            return None
-
-        command_name = matched["name"]
-        if message.session_id and command_name in global_announcement_manager.get_disabled_chat_commands(
-            message.session_id
-        ):
-            logger.info(f"[新运行时] 用户禁用的命令，跳过处理: {matched['full_name']}")
-            return False, None, True
-
-        message.is_command = True
-        logger.info(f"[新运行时] 匹配命令: {matched['full_name']}")
-
-        try:
-            resp = await prm.invoke_plugin(
-                method="plugin.invoke_command",
-                plugin_id=matched["plugin_id"],
-                component_name=matched["name"],
-                args={
-                    "text": message.processed_plain_text,
-                    "stream_id": message.session_id or "",
-                    "matched_groups": matched.get("matched_groups") or {},
-                },
-                timeout_ms=30000,
-            )
-
-            payload = resp.payload
-            success = payload.get("success", False)
-            cmd_result = payload.get("result")
-
-            # 拦截位优先从命令返回值中获取（支持运行时动态决定），
-            # 回退到组件 metadata 中的静态声明
-            if isinstance(cmd_result, (list, tuple)) and len(cmd_result) >= 3:
-                # 命令返回 (found, response_text, intercept_bool) 三元组
-                response_text = cmd_result[1] if cmd_result[1] is not None else ""
-                intercept = bool(cmd_result[2])
-            else:
-                response_text = cmd_result if cmd_result is not None else ""
-                intercept = bool(matched["metadata"].get("intercept_message_level", 0))
-
-            self._mark_command_message(message, int(intercept))
-
-            if success:
-                logger.info(f"[新运行时] 命令执行成功: {matched['full_name']}")
-            else:
-                logger.warning(f"[新运行时] 命令执行失败: {matched['full_name']} - {response_text}")
-
-            return True, response_text, not intercept
-
-        except Exception as e:
-            logger.error(f"[新运行时] 执行命令 {matched['full_name']} 异常: {e}", exc_info=True)
-            return True, str(e), True
-
     @staticmethod
     def _mark_command_message(message: SessionMessage, intercept_message_level: int) -> None:
+        """标记消息已经被命令链消费。
+
+        Args:
+            message: 待标记的会话消息。
+            intercept_message_level: 命令设置的拦截级别。
+        """
+
         message.is_command = True
         message.message_info.additional_config["intercept_message_level"] = intercept_message_level
 
     @staticmethod
     def _store_intercepted_command_message(message: SessionMessage) -> None:
+        """将被命令链拦截的消息写入数据库。
+
+        Args:
+            message: 已完成命令处理的会话消息。
+        """
+
         MessageUtils.store_message_to_db(message)
 
     async def _handle_command_processing_result(

src/chat/planner_actions/action_manager.py

@@ -3,8 +3,8 @@ from typing import Dict, Optional, Tuple
 from src.chat.message_receive.chat_manager import BotChatSession
 from src.chat.message_receive.message import SessionMessage
 from src.common.logger import get_logger
-from src.core.component_registry import component_registry, ActionExecutor
 from src.core.types import ActionInfo
+from src.plugin_runtime.component_query import ActionExecutor, component_query_service
 
 logger = get_logger("action_manager")
 
@@ -28,7 +28,7 @@ class ActionManager:
     """
     动作管理器，用于管理各种类型的动作
 
-    使用核心组件注册表的 executor-based 模式。
+    使用插件运行时统一查询服务的 executor-based 模式。
     """
 
     def __init__(self):
@@ -38,7 +38,7 @@ class ActionManager:
         self._using_actions: Dict[str, ActionInfo] = {}
 
         # 初始化时将默认动作加载到使用中的动作
-        self._using_actions = component_registry.get_default_actions()
+        self._using_actions = component_query_service.get_default_actions()
 
     # === 执行Action方法 ===
 
@@ -72,17 +72,17 @@ class ActionManager:
             Optional[ActionHandle]: 执行句柄，如果动作未注册则返回 None
         """
         try:
-            executor = component_registry.get_action_executor(action_name)
+            executor = component_query_service.get_action_executor(action_name)
             if not executor:
                 logger.warning(f"{log_prefix} 未找到Action组件: {action_name}")
                 return None
 
-            info = component_registry.get_action_info(action_name)
+            info = component_query_service.get_action_info(action_name)
             if not info:
                 logger.warning(f"{log_prefix} 未找到Action组件信息: {action_name}")
                 return None
 
-            plugin_config = component_registry.get_plugin_config(info.plugin_name) or {}
+            plugin_config = component_query_service.get_plugin_config(info.plugin_name) or {}
 
             handle = ActionHandle(
                 executor,
@@ -133,5 +133,5 @@ class ActionManager:
     def restore_actions(self) -> None:
         """恢复到默认动作集"""
         actions_to_restore = list(self._using_actions.keys())
-        self._using_actions = component_registry.get_default_actions()
+        self._using_actions = component_query_service.get_default_actions()
         logger.debug(f"恢复动作集: 从 {actions_to_restore} 恢复到默认动作集 {list(self._using_actions.keys())}")

src/chat/planner_actions/planner.py

@@ -1,33 +1,36 @@
+from collections import OrderedDict
+from datetime import datetime
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
+
+import contextlib
 import json
-import time
-import traceback
 import random
 import re
-import contextlib
+import time
-from typing import Dict, Optional, Tuple, List, TYPE_CHECKING, Union
+import traceback
-from collections import OrderedDict
+
-from rich.traceback import install
-from datetime import datetime
 from json_repair import repair_json
-from src.llm_models.utils_model import LLMRequest
+from rich.traceback import install
-from src.config.config import global_config, model_config
+
-from src.common.logger import get_logger
 from src.chat.logger.plan_reply_logger import PlanReplyLogger
+from src.chat.message_receive.chat_manager import chat_manager as _chat_manager
+from src.chat.message_receive.message import SessionMessage
+from src.chat.planner_actions.action_manager import ActionManager
+from src.chat.utils.utils import get_chat_type_and_target_info, is_bot_self
 from src.common.data_models.info_data_model import ActionPlannerInfo
+from src.common.logger import get_logger
+from src.config.config import global_config, model_config
+from src.core.types import ActionActivationType, ActionInfo, ComponentType
+from src.llm_models.utils_model import LLMRequest
+from src.person_info.person_info import Person
+from src.plugin_runtime.component_query import component_query_service
 from src.prompt.prompt_manager import prompt_manager
 from src.services.message_service import (
     build_readable_messages_with_id,
-    replace_user_references,
     get_messages_before_time_in_chat,
+    replace_user_references,
     translate_pid_to_description,
 )
-from src.chat.utils.utils import get_chat_type_and_target_info, is_bot_self
-from src.chat.planner_actions.action_manager import ActionManager
-from src.chat.message_receive.chat_manager import chat_manager as _chat_manager
-from src.chat.message_receive.message import SessionMessage
-from src.core.types import ActionActivationType, ActionInfo, ComponentType
-from src.core.component_registry import component_registry
-from src.person_info.person_info import Person
 
 if TYPE_CHECKING:
     from src.common.data_models.info_data_model import TargetPersonInfo
@@ -634,7 +637,7 @@ class ActionPlanner:
         current_available_actions_dict = self.action_manager.get_using_actions()
 
         # 获取完整的动作信息
-        all_registered_actions: Dict[str, ActionInfo] = component_registry.get_components_by_type(  # type: ignore
+        all_registered_actions: Dict[str, ActionInfo] = component_query_service.get_components_by_type(  # type: ignore
             ComponentType.ACTION
         )
         current_available_actions = {}

src/chat/tool_executor.py

@@ -1,22 +1,20 @@
-"""
+"""工具执行器。
-工具执行器
 
 独立的工具执行组件，可以直接输入聊天消息内容，
 自动判断并执行相应的工具，返回结构化的工具执行结果。
-
-从 src.plugin_system.core.tool_use 迁移，使用新的核心组件注册表。
 """
 
+from typing import Any, Dict, List, Optional, Tuple
+
 import hashlib
 import time
-from typing import Any, Dict, List, Optional, Tuple
 
 from src.common.logger import get_logger
 from src.config.config import global_config, model_config
 from src.core.announcement_manager import global_announcement_manager
-from src.core.component_registry import component_registry
 from src.llm_models.payload_content import ToolCall
 from src.llm_models.utils_model import LLMRequest
+from src.plugin_runtime.component_query import component_query_service
 from src.prompt.prompt_manager import prompt_manager
 
 logger = get_logger("tool_use")
@@ -89,7 +87,7 @@ class ToolExecutor:
 
     def _get_tool_definitions(self) -> List[Dict[str, Any]]:
         """获取 LLM 可用的工具定义列表"""
-        all_tools = component_registry.get_llm_available_tools()
+        all_tools = component_query_service.get_llm_available_tools()
         user_disabled_tools = global_announcement_manager.get_disabled_chat_tools(self.chat_id)
         return [info.get_llm_definition() for name, info in all_tools.items() if name not in user_disabled_tools]
 
@@ -152,7 +150,7 @@ class ToolExecutor:
         function_args = tool_call.args or {}
         function_args["llm_called"] = True
 
-        executor = component_registry.get_tool_executor(function_name)
+        executor = component_query_service.get_tool_executor(function_name)
         if not executor:
             logger.warning(f"未知工具名称: {function_name}")
             return None

src/core/component_registry.py

@@ -1,239 +0,0 @@
-"""
-核心组件注册表
-
-面向最终架构的组件管理：
-- Action：注册 ActionInfo + 执行器（本地 callable 或 IPC 路由）
-- Command：注册正则模式 + 执行器
-- Tool：注册工具定义 + 执行器
-
-不依赖任何插件基类，组件执行器是纯 async callable。
-"""
-
-import re
-from typing import Any, Awaitable, Callable, Dict, Optional, Pattern, Tuple
-
-from src.common.logger import get_logger
-from src.core.types import (
-    ActionInfo,
-    CommandInfo,
-    ComponentInfo,
-    ComponentType,
-    ToolInfo,
-)
-
-logger = get_logger("component_registry")
-
-# 执行器类型
-ActionExecutor = Callable[..., Awaitable[Any]]
-CommandExecutor = Callable[..., Awaitable[Tuple[bool, Optional[str], bool]]]
-ToolExecutor = Callable[..., Awaitable[Any]]
-
-
-class ComponentRegistry:
-    """核心组件注册表
-
-    管理 action、command、tool 三类组件。
-    每个组件由「元信息 + 执行器」构成，执行器是 async callable，
-    不需要继承任何基类。
-    """
-
-    def __init__(self):
-        # Action 注册
-        self._actions: Dict[str, ActionInfo] = {}
-        self._action_executors: Dict[str, ActionExecutor] = {}
-        self._default_actions: Dict[str, ActionInfo] = {}
-
-        # Command 注册
-        self._commands: Dict[str, CommandInfo] = {}
-        self._command_executors: Dict[str, CommandExecutor] = {}
-        self._command_patterns: Dict[Pattern, str] = {}
-
-        # Tool 注册
-        self._tools: Dict[str, ToolInfo] = {}
-        self._tool_executors: Dict[str, ToolExecutor] = {}
-        self._llm_available_tools: Dict[str, ToolInfo] = {}
-
-        # 插件配置（plugin_name -> config dict）
-        self._plugin_configs: Dict[str, dict] = {}
-
-        logger.info("核心组件注册表初始化完成")
-
-    # ========== Action ==========
-
-    def register_action(
-        self,
-        info: ActionInfo,
-        executor: ActionExecutor,
-    ) -> bool:
-        """注册 action
-
-        Args:
-            info: action 元信息
-            executor: 执行器，async callable
-        """
-        name = info.name
-        if name in self._actions:
-            logger.warning(f"Action {name} 已存在，跳过注册")
-            return False
-
-        self._actions[name] = info
-        self._action_executors[name] = executor
-
-        if info.enabled:
-            self._default_actions[name] = info
-
-        logger.debug(f"注册 Action: {name}")
-        return True
-
-    def get_action_info(self, name: str) -> Optional[ActionInfo]:
-        return self._actions.get(name)
-
-    def get_action_executor(self, name: str) -> Optional[ActionExecutor]:
-        return self._action_executors.get(name)
-
-    def get_default_actions(self) -> Dict[str, ActionInfo]:
-        return self._default_actions.copy()
-
-    def get_all_actions(self) -> Dict[str, ActionInfo]:
-        return self._actions.copy()
-
-    def remove_action(self, name: str) -> bool:
-        if name not in self._actions:
-            return False
-        del self._actions[name]
-        self._action_executors.pop(name, None)
-        self._default_actions.pop(name, None)
-        logger.debug(f"移除 Action: {name}")
-        return True
-
-    # ========== Command ==========
-
-    def register_command(
-        self,
-        info: CommandInfo,
-        executor: CommandExecutor,
-    ) -> bool:
-        """注册 command"""
-        name = info.name
-        if name in self._commands:
-            logger.warning(f"Command {name} 已存在，跳过注册")
-            return False
-
-        self._commands[name] = info
-        self._command_executors[name] = executor
-
-        if info.enabled and info.command_pattern:
-            pattern = re.compile(info.command_pattern, re.IGNORECASE | re.DOTALL)
-            self._command_patterns[pattern] = name
-
-        logger.debug(f"注册 Command: {name}")
-        return True
-
-    def find_command_by_text(self, text: str) -> Optional[Tuple[CommandExecutor, dict, CommandInfo]]:
-        """根据文本查找匹配的命令
-
-        Returns:
-            (executor, matched_groups, command_info) 或 None
-        """
-        candidates = [p for p in self._command_patterns if p.match(text)]
-        if not candidates:
-            return None
-        if len(candidates) > 1:
-            logger.warning(f"文本 '{text[:50]}' 匹配到多个命令模式，使用第一个")
-        pattern = candidates[0]
-        name = self._command_patterns[pattern]
-        return (
-            self._command_executors[name],
-            pattern.match(text).groupdict(),  # type: ignore
-            self._commands[name],
-        )
-
-    def remove_command(self, name: str) -> bool:
-        if name not in self._commands:
-            return False
-        del self._commands[name]
-        self._command_executors.pop(name, None)
-        self._command_patterns = {k: v for k, v in self._command_patterns.items() if v != name}
-        logger.debug(f"移除 Command: {name}")
-        return True
-
-    # ========== Tool ==========
-
-    def register_tool(
-        self,
-        info: ToolInfo,
-        executor: ToolExecutor,
-    ) -> bool:
-        """注册 tool"""
-        name = info.name
-        if name in self._tools:
-            logger.warning(f"Tool {name} 已存在，跳过注册")
-            return False
-
-        self._tools[name] = info
-        self._tool_executors[name] = executor
-
-        if info.enabled:
-            self._llm_available_tools[name] = info
-
-        logger.debug(f"注册 Tool: {name}")
-        return True
-
-    def get_tool_info(self, name: str) -> Optional[ToolInfo]:
-        return self._tools.get(name)
-
-    def get_tool_executor(self, name: str) -> Optional[ToolExecutor]:
-        return self._tool_executors.get(name)
-
-    def get_llm_available_tools(self) -> Dict[str, ToolInfo]:
-        return self._llm_available_tools.copy()
-
-    def get_all_tools(self) -> Dict[str, ToolInfo]:
-        return self._tools.copy()
-
-    def remove_tool(self, name: str) -> bool:
-        if name not in self._tools:
-            return False
-        del self._tools[name]
-        self._tool_executors.pop(name, None)
-        self._llm_available_tools.pop(name, None)
-        logger.debug(f"移除 Tool: {name}")
-        return True
-
-    # ========== 通用查询 ==========
-
-    def get_component_info(self, name: str, component_type: ComponentType) -> Optional[ComponentInfo]:
-        """获取组件元信息"""
-        match component_type:
-            case ComponentType.ACTION:
-                return self._actions.get(name)
-            case ComponentType.COMMAND:
-                return self._commands.get(name)
-            case ComponentType.TOOL:
-                return self._tools.get(name)
-            case _:
-                return None
-
-    def get_components_by_type(self, component_type: ComponentType) -> Dict[str, ComponentInfo]:
-        """获取某类型的所有组件"""
-        match component_type:
-            case ComponentType.ACTION:
-                return dict(self._actions)
-            case ComponentType.COMMAND:
-                return dict(self._commands)
-            case ComponentType.TOOL:
-                return dict(self._tools)
-            case _:
-                return {}
-
-    # ========== 插件配置 ==========
-
-    def set_plugin_config(self, plugin_name: str, config: dict) -> None:
-        self._plugin_configs[plugin_name] = config
-
-    def get_plugin_config(self, plugin_name: str) -> Optional[dict]:
-        return self._plugin_configs.get(plugin_name)
-
-
-# 全局单例
-component_registry = ComponentRegistry()

src/platform_io/manager.py

@@ -178,12 +178,6 @@ class PlatformIOManager:
 
         return self._receive_route_table
 
-    @property
-    def route_table(self) -> RouteTable:
-        """兼容旧接口，返回发送路由表。"""
-
-        return self._send_route_table
-
     @property
     def deduplicator(self) -> MessageDeduplicator:
         """返回管理器持有的入站去重器。
@@ -369,12 +363,6 @@ class PlatformIOManager:
         self._validate_binding_against_driver(binding, driver)
         self._receive_route_table.bind(binding)
 
-    def bind_route(self, binding: RouteBinding) -> None:
-        """兼容旧接口，默认同时绑定发送表和接收表。"""
-
-        self.bind_send_route(binding)
-        self.bind_receive_route(binding)
-
     def unbind_send_route(self, route_key: RouteKey, driver_id: Optional[str] = None) -> None:
         """移除发送路由绑定。
 
@@ -395,12 +383,6 @@ class PlatformIOManager:
 
         self._receive_route_table.unbind(route_key, driver_id)
 
-    def unbind_route(self, route_key: RouteKey, driver_id: Optional[str] = None) -> None:
-        """兼容旧接口，默认同时从发送表和接收表解绑。"""
-
-        self.unbind_send_route(route_key, driver_id)
-        self.unbind_receive_route(route_key, driver_id)
-
     def resolve_drivers(self, route_key: RouteKey) -> List[PlatformIODriver]:
         """解析某个路由键当前命中的全部发送驱动。
 
@@ -430,12 +412,6 @@ class PlatformIOManager:
             return []
         return [fallback_driver]
 
-    def resolve_driver(self, route_key: RouteKey) -> Optional[PlatformIODriver]:
-        """兼容旧接口，返回首个命中的发送驱动。"""
-
-        drivers = self.resolve_drivers(route_key)
-        return drivers[0] if drivers else None
-
     @staticmethod
     def build_route_key_from_message(message: "SessionMessage") -> RouteKey:
         """根据 ``SessionMessage`` 构造路由键。

src/plugin_runtime/capabilities/core.py

@@ -238,14 +238,14 @@ class RuntimeCoreCapabilityMixin:
             return {"success": False, "value": None, "error": str(e)}
 
     async def _cap_config_get_plugin(self, plugin_id: str, capability: str, args: Dict[str, Any]) -> Any:
-        from src.core.component_registry import component_registry as core_registry
+        from src.plugin_runtime.component_query import component_query_service
 
         plugin_name: str = args.get("plugin_name", plugin_id)
         key: str = args.get("key", "")
         default = args.get("default")
 
         try:
-            config = core_registry.get_plugin_config(plugin_name)
+            config = component_query_service.get_plugin_config(plugin_name)
             if config is None:
                 return {"success": False, "value": default, "error": f"未找到插件 {plugin_name} 的配置"}
 
@@ -258,11 +258,11 @@ class RuntimeCoreCapabilityMixin:
             return {"success": False, "value": default, "error": str(e)}
 
     async def _cap_config_get_all(self, plugin_id: str, capability: str, args: Dict[str, Any]) -> Any:
-        from src.core.component_registry import component_registry as core_registry
+        from src.plugin_runtime.component_query import component_query_service
 
         plugin_name: str = args.get("plugin_name", plugin_id)
         try:
-            config = core_registry.get_plugin_config(plugin_name)
+            config = component_query_service.get_plugin_config(plugin_name)
             if config is None:
                 return {"success": True, "value": {}}
             return {"success": True, "value": config}

src/plugin_runtime/capabilities/data.py

@@ -648,10 +648,10 @@ class RuntimeDataCapabilityMixin:
             return {"success": False, "error": str(e)}
 
     async def _cap_tool_get_definitions(self, plugin_id: str, capability: str, args: Dict[str, Any]) -> Any:
-        from src.core.component_registry import component_registry as core_registry
+        from src.plugin_runtime.component_query import component_query_service
 
         try:
-            tools = core_registry.get_llm_available_tools()
+            tools = component_query_service.get_llm_available_tools()
             return {
                 "success": True,
                 "tools": [{"name": name, "definition": info.get_llm_definition()} for name, info in tools.items()],

src/plugin_runtime/component_query.py

@@ -0,0 +1,709 @@
+"""插件运行时统一组件查询服务。
+
+该模块统一从插件运行时的 Host ComponentRegistry 中聚合只读视图，
+供 HFC/PFC、Planner、ToolExecutor 和运行时能力层查询与调用。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional, Tuple
+
+from src.common.logger import get_logger
+from src.core.types import ActionActivationType, ActionInfo, CommandInfo, ComponentInfo, ComponentType, ToolInfo
+from src.llm_models.payload_content.tool_option import ToolParamType
+
+if TYPE_CHECKING:
+    from src.plugin_runtime.host.component_registry import ActionEntry, CommandEntry, ComponentEntry, ToolEntry
+    from src.plugin_runtime.host.supervisor import PluginSupervisor
+    from src.plugin_runtime.integration import PluginRuntimeManager
+
+logger = get_logger("plugin_runtime.component_query")
+
+ActionExecutor = Callable[..., Awaitable[Any]]
+CommandExecutor = Callable[..., Awaitable[Tuple[bool, Optional[str], bool]]]
+ToolExecutor = Callable[..., Awaitable[Any]]
+
+_HOST_COMPONENT_TYPE_MAP: Dict[ComponentType, str] = {
+    ComponentType.ACTION: "ACTION",
+    ComponentType.COMMAND: "COMMAND",
+    ComponentType.TOOL: "TOOL",
+}
+_TOOL_PARAM_TYPE_MAP: Dict[str, ToolParamType] = {
+    "string": ToolParamType.STRING,
+    "integer": ToolParamType.INTEGER,
+    "float": ToolParamType.FLOAT,
+    "boolean": ToolParamType.BOOLEAN,
+    "bool": ToolParamType.BOOLEAN,
+}
+
+
+class ComponentQueryService:
+    """插件运行时统一组件查询服务。
+
+    该对象不维护独立状态，只读取插件系统中的注册结果。
+    所有注册、删除、配置写入等写操作都被显式禁用。
+    """
+
+    @staticmethod
+    def _get_runtime_manager() -> "PluginRuntimeManager":
+        """获取插件运行时管理器单例。
+
+        Returns:
+            PluginRuntimeManager: 当前全局插件运行时管理器。
+        """
+
+        from src.plugin_runtime.integration import get_plugin_runtime_manager
+
+        return get_plugin_runtime_manager()
+
+    def _iter_supervisors(self) -> list["PluginSupervisor"]:
+        """获取当前所有活跃的插件运行时监督器。
+
+        Returns:
+            list[PluginSupervisor]: 当前运行中的监督器列表。
+        """
+
+        runtime_manager = self._get_runtime_manager()
+        return list(runtime_manager.supervisors)
+
+    def _iter_component_entries(
+        self,
+        component_type: ComponentType,
+        *,
+        enabled_only: bool = True,
+    ) -> list[tuple["PluginSupervisor", "ComponentEntry"]]:
+        """遍历指定类型的全部组件条目。
+
+        Args:
+            component_type: 目标组件类型。
+            enabled_only: 是否仅返回启用状态的组件。
+
+        Returns:
+            list[tuple[PluginSupervisor, ComponentEntry]]: ``(监督器, 组件条目)`` 列表。
+        """
+
+        host_component_type = _HOST_COMPONENT_TYPE_MAP.get(component_type)
+        if host_component_type is None:
+            return []
+
+        collected_entries: list[tuple["PluginSupervisor", "ComponentEntry"]] = []
+        for supervisor in self._iter_supervisors():
+            for component in supervisor.component_registry.get_components_by_type(
+                host_component_type,
+                enabled_only=enabled_only,
+            ):
+                collected_entries.append((supervisor, component))
+        return collected_entries
+
+    @staticmethod
+    def _coerce_action_activation_type(raw_value: Any) -> ActionActivationType:
+        """规范化动作激活类型。
+
+        Args:
+            raw_value: 原始激活类型值。
+
+        Returns:
+            ActionActivationType: 规范化后的激活类型枚举。
+        """
+
+        normalized_value = str(raw_value or "").strip().lower()
+        if normalized_value == ActionActivationType.NEVER.value:
+            return ActionActivationType.NEVER
+        if normalized_value == ActionActivationType.RANDOM.value:
+            return ActionActivationType.RANDOM
+        if normalized_value == ActionActivationType.KEYWORD.value:
+            return ActionActivationType.KEYWORD
+        return ActionActivationType.ALWAYS
+
+    @staticmethod
+    def _coerce_float(value: Any, default: float = 0.0) -> float:
+        """将任意值安全转换为浮点数。
+
+        Args:
+            value: 待转换的输入值。
+            default: 转换失败时返回的默认值。
+
+        Returns:
+            float: 转换后的浮点结果。
+        """
+
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return default
+
+    @staticmethod
+    def _build_action_info(entry: "ActionEntry") -> ActionInfo:
+        """将运行时 Action 条目转换为核心动作信息。
+
+        Args:
+            entry: 插件运行时中的 Action 条目。
+
+        Returns:
+            ActionInfo: 供核心 Planner 使用的动作信息。
+        """
+
+        metadata = dict(entry.metadata)
+        raw_action_parameters = metadata.get("action_parameters")
+        action_parameters = (
+            {
+                str(param_name): str(param_description)
+                for param_name, param_description in raw_action_parameters.items()
+            }
+            if isinstance(raw_action_parameters, dict)
+            else {}
+        )
+        action_require = [
+            str(item)
+            for item in (metadata.get("action_require") or [])
+            if item is not None and str(item).strip()
+        ]
+        associated_types = [
+            str(item)
+            for item in (metadata.get("associated_types") or [])
+            if item is not None and str(item).strip()
+        ]
+        activation_keywords = [
+            str(item)
+            for item in (metadata.get("activation_keywords") or [])
+            if item is not None and str(item).strip()
+        ]
+
+        return ActionInfo(
+            name=entry.name,
+            component_type=ComponentType.ACTION,
+            description=str(metadata.get("description", "") or ""),
+            enabled=bool(entry.enabled),
+            plugin_name=entry.plugin_id,
+            metadata=metadata,
+            action_parameters=action_parameters,
+            action_require=action_require,
+            associated_types=associated_types,
+            activation_type=ComponentQueryService._coerce_action_activation_type(metadata.get("activation_type")),
+            random_activation_probability=ComponentQueryService._coerce_float(
+                metadata.get("activation_probability"),
+                0.0,
+            ),
+            activation_keywords=activation_keywords,
+            parallel_action=bool(metadata.get("parallel_action", False)),
+        )
+
+    @staticmethod
+    def _build_command_info(entry: "CommandEntry") -> CommandInfo:
+        """将运行时 Command 条目转换为核心命令信息。
+
+        Args:
+            entry: 插件运行时中的 Command 条目。
+
+        Returns:
+            CommandInfo: 供核心命令链使用的命令信息。
+        """
+
+        metadata = dict(entry.metadata)
+        return CommandInfo(
+            name=entry.name,
+            component_type=ComponentType.COMMAND,
+            description=str(metadata.get("description", "") or ""),
+            enabled=bool(entry.enabled),
+            plugin_name=entry.plugin_id,
+            metadata=metadata,
+            command_pattern=str(metadata.get("command_pattern", "") or ""),
+        )
+
+    @staticmethod
+    def _coerce_tool_param_type(raw_value: Any) -> ToolParamType:
+        """规范化工具参数类型。
+
+        Args:
+            raw_value: 原始工具参数类型值。
+
+        Returns:
+            ToolParamType: 规范化后的工具参数类型。
+        """
+
+        normalized_value = str(raw_value or "").strip().lower()
+        return _TOOL_PARAM_TYPE_MAP.get(normalized_value, ToolParamType.STRING)
+
+    @staticmethod
+    def _build_tool_parameters(entry: "ToolEntry") -> list[tuple[str, ToolParamType, str, bool, list[str] | None]]:
+        """将运行时工具参数元数据转换为核心 ToolInfo 参数列表。
+
+        Args:
+            entry: 插件运行时中的 Tool 条目。
+
+        Returns:
+            list[tuple[str, ToolParamType, str, bool, list[str] | None]]: 转换后的参数列表。
+        """
+
+        structured_parameters = entry.parameters if isinstance(entry.parameters, list) else []
+        if not structured_parameters and isinstance(entry.parameters_raw, dict):
+            structured_parameters = [
+                {"name": key, **value}
+                for key, value in entry.parameters_raw.items()
+                if isinstance(value, dict)
+            ]
+
+        normalized_parameters: list[tuple[str, ToolParamType, str, bool, list[str] | None]] = []
+        for parameter in structured_parameters:
+            if not isinstance(parameter, dict):
+                continue
+
+            parameter_name = str(parameter.get("name", "") or "").strip()
+            if not parameter_name:
+                continue
+
+            enum_values = parameter.get("enum")
+            normalized_enum_values = (
+                [str(item) for item in enum_values if item is not None]
+                if isinstance(enum_values, list)
+                else None
+            )
+            normalized_parameters.append(
+                (
+                    parameter_name,
+                    ComponentQueryService._coerce_tool_param_type(parameter.get("param_type") or parameter.get("type")),
+                    str(parameter.get("description", "") or ""),
+                    bool(parameter.get("required", True)),
+                    normalized_enum_values,
+                )
+            )
+        return normalized_parameters
+
+    @staticmethod
+    def _build_tool_info(entry: "ToolEntry") -> ToolInfo:
+        """将运行时 Tool 条目转换为核心工具信息。
+
+        Args:
+            entry: 插件运行时中的 Tool 条目。
+
+        Returns:
+            ToolInfo: 供 ToolExecutor 与能力层使用的工具信息。
+        """
+
+        return ToolInfo(
+            name=entry.name,
+            component_type=ComponentType.TOOL,
+            description=entry.description,
+            enabled=bool(entry.enabled),
+            plugin_name=entry.plugin_id,
+            metadata=dict(entry.metadata),
+            tool_parameters=ComponentQueryService._build_tool_parameters(entry),
+            tool_description=entry.description,
+        )
+
+    @staticmethod
+    def _log_duplicate_component(component_type: ComponentType, component_name: str) -> None:
+        """记录重复组件名称冲突。
+
+        Args:
+            component_type: 组件类型。
+            component_name: 发生冲突的组件名称。
+        """
+
+        logger.warning(f"检测到重复{component_type.value}名称 {component_name}，将只保留首个匹配项")
+
+    def _get_unique_component_entry(
+        self,
+        component_type: ComponentType,
+        name: str,
+    ) -> Optional[tuple["PluginSupervisor", "ComponentEntry"]]:
+        """按组件短名解析唯一条目。
+
+        Args:
+            component_type: 目标组件类型。
+            name: 组件短名。
+
+        Returns:
+            Optional[tuple[PluginSupervisor, ComponentEntry]]: 唯一命中的组件条目。
+        """
+
+        matched_entries = [
+            (supervisor, entry)
+            for supervisor, entry in self._iter_component_entries(component_type)
+            if entry.name == name
+        ]
+        if not matched_entries:
+            return None
+        if len(matched_entries) > 1:
+            self._log_duplicate_component(component_type, name)
+        return matched_entries[0]
+
+    def _collect_unique_component_infos(
+        self,
+        component_type: ComponentType,
+    ) -> Dict[str, ComponentInfo]:
+        """收集某类组件的唯一信息视图。
+
+        Args:
+            component_type: 目标组件类型。
+
+        Returns:
+            Dict[str, ComponentInfo]: 组件名到核心组件信息的映射。
+        """
+
+        collected_components: Dict[str, ComponentInfo] = {}
+        for _supervisor, entry in self._iter_component_entries(component_type):
+            if entry.name in collected_components:
+                self._log_duplicate_component(component_type, entry.name)
+                continue
+
+            if component_type == ComponentType.ACTION:
+                collected_components[entry.name] = self._build_action_info(entry)  # type: ignore[arg-type]
+            elif component_type == ComponentType.COMMAND:
+                collected_components[entry.name] = self._build_command_info(entry)  # type: ignore[arg-type]
+            elif component_type == ComponentType.TOOL:
+                collected_components[entry.name] = self._build_tool_info(entry)  # type: ignore[arg-type]
+        return collected_components
+
+    @staticmethod
+    def _extract_stream_id_from_action_kwargs(kwargs: Dict[str, Any]) -> str:
+        """从旧 ActionManager 参数中提取聊天流 ID。
+
+        Args:
+            kwargs: 旧动作执行器收到的关键字参数。
+
+        Returns:
+            str: 提取出的 ``stream_id``。
+        """
+
+        chat_stream = kwargs.get("chat_stream")
+        if chat_stream is not None:
+            try:
+                return str(chat_stream.session_id)
+            except AttributeError:
+                pass
+
+        return str(kwargs.get("stream_id", "") or "")
+
+    @staticmethod
+    def _build_action_executor(supervisor: "PluginSupervisor", plugin_id: str, component_name: str) -> ActionExecutor:
+        """构造动作执行 RPC 闭包。
+
+        Args:
+            supervisor: 负责该组件的监督器。
+            plugin_id: 插件 ID。
+            component_name: 组件名称。
+
+        Returns:
+            ActionExecutor: 兼容旧 Planner 的异步执行器。
+        """
+
+        async def _executor(**kwargs: Any) -> tuple[bool, str]:
+            """将核心动作调用桥接到插件运行时。
+
+            Args:
+                **kwargs: 旧 ActionManager 传入的上下文参数。
+
+            Returns:
+                tuple[bool, str]: ``(是否成功, 结果说明)``。
+            """
+
+            invoke_args: Dict[str, Any] = {}
+            action_data = kwargs.get("action_data")
+            if isinstance(action_data, dict):
+                invoke_args.update(action_data)
+
+            stream_id = ComponentQueryService._extract_stream_id_from_action_kwargs(kwargs)
+            invoke_args["action_data"] = action_data if isinstance(action_data, dict) else {}
+            invoke_args["stream_id"] = stream_id
+            invoke_args["chat_id"] = stream_id
+            invoke_args["reasoning"] = str(kwargs.get("action_reasoning", "") or "")
+
+            if (thinking_id := kwargs.get("thinking_id")) is not None:
+                invoke_args["thinking_id"] = str(thinking_id)
+            if isinstance(kwargs.get("cycle_timers"), dict):
+                invoke_args["cycle_timers"] = kwargs["cycle_timers"]
+            if isinstance(kwargs.get("plugin_config"), dict):
+                invoke_args["plugin_config"] = kwargs["plugin_config"]
+            if isinstance(kwargs.get("log_prefix"), str):
+                invoke_args["log_prefix"] = kwargs["log_prefix"]
+            if isinstance(kwargs.get("shutting_down"), bool):
+                invoke_args["shutting_down"] = kwargs["shutting_down"]
+
+            try:
+                response = await supervisor.invoke_plugin(
+                    method="plugin.invoke_action",
+                    plugin_id=plugin_id,
+                    component_name=component_name,
+                    args=invoke_args,
+                    timeout_ms=30000,
+                )
+            except Exception as exc:
+                logger.error(f"运行时 Action {plugin_id}.{component_name} 执行失败: {exc}", exc_info=True)
+                return False, str(exc)
+
+            payload = response.payload if isinstance(response.payload, dict) else {}
+            success = bool(payload.get("success", False))
+            result = payload.get("result")
+            if isinstance(result, (list, tuple)):
+                if len(result) >= 2:
+                    return bool(result[0]), "" if result[1] is None else str(result[1])
+                if len(result) == 1:
+                    return bool(result[0]), ""
+            if success:
+                return True, "" if result is None else str(result)
+            return False, "" if result is None else str(result)
+
+        return _executor
+
+    @staticmethod
+    def _build_command_executor(
+        supervisor: "PluginSupervisor",
+        plugin_id: str,
+        component_name: str,
+        metadata: Dict[str, Any],
+    ) -> CommandExecutor:
+        """构造命令执行 RPC 闭包。
+
+        Args:
+            supervisor: 负责该组件的监督器。
+            plugin_id: 插件 ID。
+            component_name: 组件名称。
+            metadata: 命令组件元数据。
+
+        Returns:
+            CommandExecutor: 兼容旧消息命令链的执行器。
+        """
+
+        async def _executor(**kwargs: Any) -> tuple[bool, Optional[str], bool]:
+            """将核心命令调用桥接到插件运行时。
+
+            Args:
+                **kwargs: 命令执行上下文参数。
+
+            Returns:
+                tuple[bool, Optional[str], bool]: ``(是否成功, 返回文本, 是否拦截后续消息)``。
+            """
+
+            message = kwargs.get("message")
+            matched_groups = kwargs.get("matched_groups")
+            plugin_config = kwargs.get("plugin_config")
+            invoke_args: Dict[str, Any] = {
+                "text": str(getattr(message, "processed_plain_text", "") or ""),
+                "stream_id": str(getattr(message, "session_id", "") or ""),
+                "matched_groups": matched_groups if isinstance(matched_groups, dict) else {},
+            }
+            if isinstance(plugin_config, dict):
+                invoke_args["plugin_config"] = plugin_config
+
+            try:
+                response = await supervisor.invoke_plugin(
+                    method="plugin.invoke_command",
+                    plugin_id=plugin_id,
+                    component_name=component_name,
+                    args=invoke_args,
+                    timeout_ms=30000,
+                )
+            except Exception as exc:
+                logger.error(f"运行时 Command {plugin_id}.{component_name} 执行失败: {exc}", exc_info=True)
+                return False, str(exc), True
+
+            payload = response.payload if isinstance(response.payload, dict) else {}
+            success = bool(payload.get("success", False))
+            result = payload.get("result")
+            intercept = bool(metadata.get("intercept_message_level", 0))
+            response_text: Optional[str]
+
+            if isinstance(result, (list, tuple)) and len(result) >= 3:
+                response_text = None if result[1] is None else str(result[1])
+                intercept = bool(result[2])
+            else:
+                response_text = None if result is None else str(result)
+
+            return success, response_text, intercept
+
+        return _executor
+
+    @staticmethod
+    def _build_tool_executor(supervisor: "PluginSupervisor", plugin_id: str, component_name: str) -> ToolExecutor:
+        """构造工具执行 RPC 闭包。
+
+        Args:
+            supervisor: 负责该组件的监督器。
+            plugin_id: 插件 ID。
+            component_name: 组件名称。
+
+        Returns:
+            ToolExecutor: 兼容旧 ToolExecutor 的异步执行器。
+        """
+
+        async def _executor(function_args: Dict[str, Any]) -> Any:
+            """将核心工具调用桥接到插件运行时。
+
+            Args:
+                function_args: 工具调用参数。
+
+            Returns:
+                Any: 插件工具返回结果；若结果不是字典，则会包装为 ``{"content": ...}``。
+            """
+
+            try:
+                response = await supervisor.invoke_plugin(
+                    method="plugin.invoke_tool",
+                    plugin_id=plugin_id,
+                    component_name=component_name,
+                    args=function_args,
+                    timeout_ms=30000,
+                )
+            except Exception as exc:
+                logger.error(f"运行时 Tool {plugin_id}.{component_name} 执行失败: {exc}", exc_info=True)
+                return {"content": f"工具 {component_name} 执行失败: {exc}"}
+
+            payload = response.payload if isinstance(response.payload, dict) else {}
+            result = payload.get("result")
+            if isinstance(result, dict):
+                return result
+            return {"content": "" if result is None else str(result)}
+
+        return _executor
+
+    def get_action_info(self, name: str) -> Optional[ActionInfo]:
+        """获取指定动作的信息。
+
+        Args:
+            name: 动作名称。
+
+        Returns:
+            Optional[ActionInfo]: 匹配到的动作信息。
+        """
+
+        matched_entry = self._get_unique_component_entry(ComponentType.ACTION, name)
+        if matched_entry is None:
+            return None
+        _supervisor, entry = matched_entry
+        return self._build_action_info(entry)  # type: ignore[arg-type]
+
+    def get_action_executor(self, name: str) -> Optional[ActionExecutor]:
+        """获取指定动作的执行器。
+
+        Args:
+            name: 动作名称。
+
+        Returns:
+            Optional[ActionExecutor]: 运行时 RPC 执行闭包。
+        """
+
+        matched_entry = self._get_unique_component_entry(ComponentType.ACTION, name)
+        if matched_entry is None:
+            return None
+        supervisor, entry = matched_entry
+        return self._build_action_executor(supervisor, entry.plugin_id, entry.name)
+
+    def get_default_actions(self) -> Dict[str, ActionInfo]:
+        """获取当前默认启用的动作集合。
+
+        Returns:
+            Dict[str, ActionInfo]: 动作名到动作信息的映射。
+        """
+
+        action_infos = self._collect_unique_component_infos(ComponentType.ACTION)
+        return {name: info for name, info in action_infos.items() if isinstance(info, ActionInfo) and info.enabled}
+
+    def find_command_by_text(self, text: str) -> Optional[Tuple[CommandExecutor, dict, CommandInfo]]:
+        """根据文本查找匹配的命令。
+
+        Args:
+            text: 待匹配的文本内容。
+
+        Returns:
+            Optional[Tuple[CommandExecutor, dict, CommandInfo]]: 匹配结果。
+        """
+
+        for supervisor in self._iter_supervisors():
+            match_result = supervisor.component_registry.find_command_by_text(text)
+            if match_result is None:
+                continue
+
+            entry, matched_groups = match_result
+            command_info = self._build_command_info(entry)  # type: ignore[arg-type]
+            command_executor = self._build_command_executor(
+                supervisor,
+                entry.plugin_id,
+                entry.name,
+                dict(entry.metadata),
+            )
+            return command_executor, matched_groups, command_info
+        return None
+
+    def get_tool_info(self, name: str) -> Optional[ToolInfo]:
+        """获取指定工具的信息。
+
+        Args:
+            name: 工具名称。
+
+        Returns:
+            Optional[ToolInfo]: 匹配到的工具信息。
+        """
+
+        matched_entry = self._get_unique_component_entry(ComponentType.TOOL, name)
+        if matched_entry is None:
+            return None
+        _supervisor, entry = matched_entry
+        return self._build_tool_info(entry)  # type: ignore[arg-type]
+
+    def get_tool_executor(self, name: str) -> Optional[ToolExecutor]:
+        """获取指定工具的执行器。
+
+        Args:
+            name: 工具名称。
+
+        Returns:
+            Optional[ToolExecutor]: 运行时 RPC 执行闭包。
+        """
+
+        matched_entry = self._get_unique_component_entry(ComponentType.TOOL, name)
+        if matched_entry is None:
+            return None
+        supervisor, entry = matched_entry
+        return self._build_tool_executor(supervisor, entry.plugin_id, entry.name)
+
+    def get_llm_available_tools(self) -> Dict[str, ToolInfo]:
+        """获取当前可供 LLM 选择的工具集合。
+
+        Returns:
+            Dict[str, ToolInfo]: 工具名到工具信息的映射。
+        """
+
+        tool_infos = self._collect_unique_component_infos(ComponentType.TOOL)
+        return {name: info for name, info in tool_infos.items() if isinstance(info, ToolInfo) and info.enabled}
+
+    def get_components_by_type(self, component_type: ComponentType) -> Dict[str, ComponentInfo]:
+        """获取某类组件的全部信息。
+
+        Args:
+            component_type: 组件类型。
+
+        Returns:
+            Dict[str, ComponentInfo]: 组件名到组件信息的映射。
+        """
+
+        return self._collect_unique_component_infos(component_type)
+
+    def get_plugin_config(self, plugin_name: str) -> Optional[dict]:
+        """读取指定插件的配置文件内容。
+
+        Args:
+            plugin_name: 插件名称。
+
+        Returns:
+            Optional[dict]: 读取成功时返回配置字典；未找到时返回 ``None``。
+        """
+
+        runtime_manager = self._get_runtime_manager()
+        try:
+            supervisor = runtime_manager._get_supervisor_for_plugin(plugin_name)
+        except RuntimeError as exc:
+            logger.error(f"读取插件配置失败: {exc}")
+            return None
+
+        if supervisor is None:
+            return None
+
+        try:
+            return runtime_manager._load_plugin_config_for_supervisor(supervisor, plugin_name)
+        except Exception as exc:
+            logger.error(f"读取插件 {plugin_name} 配置失败: {exc}", exc_info=True)
+            return None
+
+
+component_query_service = ComponentQueryService()

src/plugin_runtime/host/component_registry.py

@@ -31,12 +31,12 @@ class ComponentTypes(str, Enum):
 
 class StatusDict(TypedDict):
     total: int
-    ACTION: int
+    action: int
-    COMMAND: int
+    command: int
-    TOOL: int
+    tool: int
-    EVENT_HANDLER: int
+    event_handler: int
-    HOOK_HANDLER: int
+    hook_handler: int
-    MESSAGE_GATEWAY: int
+    message_gateway: int
     plugins: int
 
 
@@ -185,6 +185,23 @@ class ComponentRegistry:
         # 按插件索引
         self._by_plugin: Dict[str, List[ComponentEntry]] = {}
 
+    @staticmethod
+    def _normalize_component_type(component_type: str) -> ComponentTypes:
+        """规范化组件类型输入。
+
+        Args:
+            component_type: 原始组件类型字符串。
+
+        Returns:
+            ComponentTypes: 规范化后的组件类型枚举。
+
+        Raises:
+            ValueError: 当组件类型不受支持时抛出。
+        """
+
+        normalized_value = str(component_type or "").strip().upper()
+        return ComponentTypes(normalized_value)
+
     def clear(self) -> None:
         """清空全部组件注册状态。"""
         self._components.clear()
@@ -205,18 +222,19 @@ class ComponentRegistry:
             success (bool): 是否成功注册（失败原因通常是组件类型无效）
         """
         try:
-            if component_type == ComponentTypes.ACTION:
+            normalized_type = self._normalize_component_type(component_type)
-                comp = ActionEntry(name, component_type, plugin_id, metadata)
+            if normalized_type == ComponentTypes.ACTION:
-            elif component_type == ComponentTypes.COMMAND:
+                comp = ActionEntry(name, normalized_type.value, plugin_id, metadata)
-                comp = CommandEntry(name, component_type, plugin_id, metadata)
+            elif normalized_type == ComponentTypes.COMMAND:
-            elif component_type == ComponentTypes.TOOL:
+                comp = CommandEntry(name, normalized_type.value, plugin_id, metadata)
-                comp = ToolEntry(name, component_type, plugin_id, metadata)
+            elif normalized_type == ComponentTypes.TOOL:
-            elif component_type == ComponentTypes.EVENT_HANDLER:
+                comp = ToolEntry(name, normalized_type.value, plugin_id, metadata)
-                comp = EventHandlerEntry(name, component_type, plugin_id, metadata)
+            elif normalized_type == ComponentTypes.EVENT_HANDLER:
-            elif component_type == ComponentTypes.HOOK_HANDLER:
+                comp = EventHandlerEntry(name, normalized_type.value, plugin_id, metadata)
-                comp = HookHandlerEntry(name, component_type, plugin_id, metadata)
+            elif normalized_type == ComponentTypes.HOOK_HANDLER:
-            elif component_type == ComponentTypes.MESSAGE_GATEWAY:
+                comp = HookHandlerEntry(name, normalized_type.value, plugin_id, metadata)
-                comp = MessageGatewayEntry(name, component_type, plugin_id, metadata)
+            elif normalized_type == ComponentTypes.MESSAGE_GATEWAY:
+                comp = MessageGatewayEntry(name, normalized_type.value, plugin_id, metadata)
             else:
                 raise ValueError(f"组件类型 {component_type} 不存在")
         except ValueError:
@@ -304,6 +322,20 @@ class ComponentRegistry:
             comp.enabled = enabled
         return True
 
+    def set_component_enabled(self, full_name: str, enabled: bool, session_id: Optional[str] = None) -> bool:
+        """设置指定组件的启用状态。
+
+        Args:
+            full_name: 组件全名。
+            enabled: 目标启用状态。
+            session_id: 可选的会话 ID，仅对该会话生效。
+
+        Returns:
+            bool: 是否设置成功。
+        """
+
+        return self.toggle_component_status(full_name, enabled, session_id=session_id)
+
     def toggle_plugin_status(self, plugin_id: str, enabled: bool, session_id: Optional[str] = None) -> int:
         """批量启用或禁用某插件的所有组件。
 
@@ -348,7 +380,7 @@ class ComponentRegistry:
             components (List[ComponentEntry]): 组件条目列表
         """
         try:
-            comp_type = ComponentTypes(component_type)
+            comp_type = self._normalize_component_type(component_type)
         except ValueError:
             logger.error(f"组件类型 {component_type} 不存在")
             raise
@@ -536,6 +568,6 @@ class ComponentRegistry:
         """
         stats: StatusDict = {"total": len(self._components)}  # type: ignore
         for comp_type, type_dict in self._by_type.items():
-            stats[comp_type.value] = len(type_dict)
+            stats[comp_type.value.lower()] = len(type_dict)
         stats["plugins"] = len(self._by_plugin)
         return stats

src/plugin_runtime/host/supervisor.py

@@ -9,8 +9,6 @@ import sys
 
 from src.common.logger import get_logger
 from src.config.config import global_config
-from src.core.component_registry import component_registry as core_component_registry
-from src.core.types import ActionActivationType, ActionInfo, ComponentType as CoreComponentType
 from src.platform_io import DriverKind, InboundMessageEnvelope, RouteBinding, RouteKey, get_platform_io_manager
 from src.platform_io.drivers import PluginPlatformDriver
 from src.platform_io.route_key_factory import RouteKeyFactory
@@ -107,7 +105,6 @@ class PluginRunnerSupervisor:
         self._runner_process: Optional[asyncio.subprocess.Process] = None
         self._registered_plugins: Dict[str, RegisterPluginPayload] = {}
         self._message_gateway_states: Dict[str, Dict[str, _MessageGatewayRuntimeState]] = {}
-        self._mirrored_core_actions: Dict[str, List[str]] = {}
         self._runner_ready_events: asyncio.Event = asyncio.Event()
         self._runner_ready_payloads: RunnerReadyPayload = RunnerReadyPayload()
         self._health_task: Optional[asyncio.Task[None]] = None
@@ -510,7 +507,6 @@ class PluginRunnerSupervisor:
         except Exception as exc:
             return envelope.make_error_response(ErrorCode.E_BAD_PAYLOAD.value, str(exc))
 
-        self._remove_core_action_mirrors(payload.plugin_id)
         self._component_registry.remove_components_by_plugin(payload.plugin_id)
         await self._unregister_all_message_gateway_drivers_for_plugin(payload.plugin_id)
 
@@ -520,7 +516,6 @@ class PluginRunnerSupervisor:
         )
         self._registered_plugins[payload.plugin_id] = payload
         self._message_gateway_states[payload.plugin_id] = {}
-        self._mirror_runtime_actions_to_core_registry(payload)
 
         return envelope.make_response(
             payload={
@@ -550,7 +545,6 @@ class PluginRunnerSupervisor:
         removed_components = self._component_registry.remove_components_by_plugin(payload.plugin_id)
         self._authorization.revoke_permission_token(payload.plugin_id)
         removed_registration = self._registered_plugins.pop(payload.plugin_id, None) is not None
-        self._remove_core_action_mirrors(payload.plugin_id)
         await self._unregister_all_message_gateway_drivers_for_plugin(payload.plugin_id)
         self._message_gateway_states.pop(payload.plugin_id, None)
 
@@ -564,236 +558,6 @@ class PluginRunnerSupervisor:
             }
         )
 
-    @staticmethod
-    def _coerce_action_activation_type(raw_value: Any) -> ActionActivationType:
-        """将运行时 Action 激活类型转换为旧核心枚举。
-
-        Args:
-            raw_value: 插件运行时声明中的激活类型值。
-
-        Returns:
-            ActionActivationType: 可供旧 Planner 使用的激活类型枚举。
-        """
-        normalized_value = str(raw_value or ActionActivationType.ALWAYS.value).strip().lower()
-        try:
-            return ActionActivationType(normalized_value)
-        except ValueError:
-            return ActionActivationType.ALWAYS
-
-    @staticmethod
-    def _coerce_float(value: Any, default: float = 0.0) -> float:
-        """将任意输入尽量转换为浮点数。
-
-        Args:
-            value: 待转换的值。
-            default: 转换失败时使用的默认值。
-
-        Returns:
-            float: 转换结果。
-        """
-        try:
-            return float(value)
-        except (TypeError, ValueError):
-            return default
-
-    @staticmethod
-    def _build_core_action_info(plugin_id: str, component_name: str, metadata: Dict[str, Any]) -> ActionInfo:
-        """将运行时 Action 元数据映射为旧核心 ActionInfo。
-
-        Args:
-            plugin_id: 插件 ID。
-            component_name: 组件名称。
-            metadata: 运行时组件元数据。
-
-        Returns:
-            ActionInfo: 兼容旧 Planner 的动作定义。
-        """
-        activation_keywords = [
-            str(item)
-            for item in (metadata.get("activation_keywords") or [])
-            if item is not None and str(item).strip()
-        ]
-        action_require = [
-            str(item)
-            for item in (metadata.get("action_require") or [])
-            if item is not None and str(item).strip()
-        ]
-        associated_types = [
-            str(item)
-            for item in (metadata.get("associated_types") or [])
-            if item is not None and str(item).strip()
-        ]
-        raw_action_parameters = metadata.get("action_parameters") or {}
-        action_parameters = {
-            str(param_name): str(param_description)
-            for param_name, param_description in raw_action_parameters.items()
-        } if isinstance(raw_action_parameters, dict) else {}
-
-        return ActionInfo(
-            name=component_name,
-            component_type=CoreComponentType.ACTION,
-            description=str(metadata.get("description", "") or ""),
-            enabled=bool(metadata.get("enabled", True)),
-            plugin_name=plugin_id,
-            metadata=dict(metadata),
-            action_parameters=action_parameters,
-            action_require=action_require,
-            associated_types=associated_types,
-            activation_type=PluginRunnerSupervisor._coerce_action_activation_type(metadata.get("activation_type")),
-            random_activation_probability=PluginRunnerSupervisor._coerce_float(
-                metadata.get("activation_probability"),
-                0.0,
-            ),
-            activation_keywords=activation_keywords,
-            parallel_action=bool(metadata.get("parallel_action", False)),
-        )
-
-    @staticmethod
-    def _extract_stream_id_from_action_kwargs(kwargs: Dict[str, Any]) -> str:
-        """从旧 ActionManager 传入参数中提取聊天流 ID。
-
-        Args:
-            kwargs: 旧动作执行器收到的关键字参数。
-
-        Returns:
-            str: 可用于新运行时 Action 的 ``stream_id``。
-        """
-        chat_stream = kwargs.get("chat_stream")
-        if chat_stream is not None:
-            try:
-                return str(chat_stream.session_id)
-            except AttributeError:
-                pass
-
-        raw_stream_id = kwargs.get("stream_id", "")
-        return str(raw_stream_id or "")
-
-    def _build_runtime_action_executor(
-        self,
-        plugin_id: str,
-        component_name: str,
-    ) -> Any:
-        """构造一个转发到 plugin runtime 的旧核心 Action 执行器。
-
-        Args:
-            plugin_id: 目标插件 ID。
-            component_name: 目标 Action 组件名称。
-
-        Returns:
-            Callable[..., Coroutine[Any, Any, tuple[bool, str]]]: 兼容旧 ActionManager 的执行器。
-        """
-
-        async def _executor(**kwargs: Any) -> tuple[bool, str]:
-            """将旧 Planner 的动作调用桥接到 plugin runtime。
-
-            Args:
-                **kwargs: 旧 ActionManager 传入的运行时上下文参数。
-
-            Returns:
-                tuple[bool, str]: ``(是否成功, 动作说明)``。
-            """
-            invoke_args: Dict[str, Any] = {}
-            action_data = kwargs.get("action_data")
-            if isinstance(action_data, dict):
-                invoke_args.update(action_data)
-
-            stream_id = self._extract_stream_id_from_action_kwargs(kwargs)
-            invoke_args["action_data"] = action_data if isinstance(action_data, dict) else {}
-            invoke_args["stream_id"] = stream_id
-            invoke_args["chat_id"] = stream_id
-            invoke_args["reasoning"] = str(kwargs.get("action_reasoning", "") or "")
-
-            thinking_id = kwargs.get("thinking_id")
-            if thinking_id is not None:
-                invoke_args["thinking_id"] = str(thinking_id)
-
-            cycle_timers = kwargs.get("cycle_timers")
-            if isinstance(cycle_timers, dict):
-                invoke_args["cycle_timers"] = cycle_timers
-
-            plugin_config = kwargs.get("plugin_config")
-            if isinstance(plugin_config, dict):
-                invoke_args["plugin_config"] = plugin_config
-
-            log_prefix = kwargs.get("log_prefix")
-            if isinstance(log_prefix, str):
-                invoke_args["log_prefix"] = log_prefix
-
-            shutting_down = kwargs.get("shutting_down")
-            if isinstance(shutting_down, bool):
-                invoke_args["shutting_down"] = shutting_down
-
-            try:
-                response = await self.invoke_plugin(
-                    method="plugin.invoke_action",
-                    plugin_id=plugin_id,
-                    component_name=component_name,
-                    args=invoke_args,
-                    timeout_ms=30000,
-                )
-            except Exception as exc:
-                logger.error(f"运行时 Action {plugin_id}.{component_name} 执行失败: {exc}", exc_info=True)
-                return False, str(exc)
-
-            payload = response.payload if isinstance(response.payload, dict) else {}
-            success = bool(payload.get("success", False))
-            result = payload.get("result")
-
-            if isinstance(result, (list, tuple)):
-                if len(result) >= 2:
-                    return bool(result[0]), "" if result[1] is None else str(result[1])
-                if len(result) == 1:
-                    return bool(result[0]), ""
-
-            if success:
-                return True, "" if result is None else str(result)
-            return False, "" if result is None else str(result)
-
-        return _executor
-
-    def _mirror_runtime_actions_to_core_registry(self, payload: RegisterPluginPayload) -> None:
-        """将 plugin runtime 中声明的 Action 镜像到旧核心注册表。
-
-        Args:
-            payload: 当前插件的注册载荷。
-        """
-        mirrored_action_names: List[str] = []
-
-        for component in payload.components:
-            if str(component.component_type).upper() != CoreComponentType.ACTION.name:
-                continue
-
-            action_info = self._build_core_action_info(
-                plugin_id=payload.plugin_id,
-                component_name=component.name,
-                metadata=component.metadata,
-            )
-            action_executor = self._build_runtime_action_executor(
-                plugin_id=payload.plugin_id,
-                component_name=component.name,
-            )
-            registered = core_component_registry.register_action(action_info, action_executor)
-            if not registered:
-                logger.warning(
-                    f"运行时 Action {payload.plugin_id}.{component.name} 无法镜像到旧核心注册表，"
-                    "可能与现有 Action 重名"
-                )
-                continue
-            mirrored_action_names.append(component.name)
-
-        if mirrored_action_names:
-            self._mirrored_core_actions[payload.plugin_id] = mirrored_action_names
-
-    def _remove_core_action_mirrors(self, plugin_id: str) -> None:
-        """移除某个插件镜像到旧核心注册表的所有 Action。
-
-        Args:
-            plugin_id: 目标插件 ID。
-        """
-        mirrored_action_names = self._mirrored_core_actions.pop(plugin_id, [])
-        for action_name in mirrored_action_names:
-            core_component_registry.remove_action(action_name)
-
     @staticmethod
     def _build_message_gateway_driver_id(plugin_id: str, gateway_name: str) -> str:
         """构造消息网关驱动 ID。
@@ -1407,8 +1171,6 @@ class PluginRunnerSupervisor:
 
     def _clear_runner_state(self) -> None:
         """清理当前 Runner 对应的 Host 侧注册状态。"""
-        for plugin_id in list(self._mirrored_core_actions.keys()):
-            self._remove_core_action_mirrors(plugin_id)
         self._authorization.clear()
         self._component_registry.clear()
         self._registered_plugins.clear()

src/services/send_service.py

@@ -8,10 +8,9 @@
 3. 具体走插件链还是 legacy 旧链，由 Platform IO 内部统一决策。
 """
 
+from copy import deepcopy
 from typing import Any, Dict, List, Optional
 
-from maim_message import Seg
-
 import asyncio
 import base64
 import hashlib
@@ -28,6 +27,7 @@ from src.common.data_models.message_component_data_model import (
     AtComponent,
     DictComponent,
     EmojiComponent,
+    ForwardNodeComponent,
     ImageComponent,
     MessageSequence,
     ReplyComponent,
@@ -72,88 +72,163 @@ def _inherit_platform_io_route_metadata(target_stream: BotChatSession) -> Dict[s
         if normalized_value:
             inherited_metadata[key] = value
 
-    if target_stream.group_id:
+    if target_stream.group_id and (normalized_group_id := str(target_stream.group_id).strip()):
-        normalized_group_id = str(target_stream.group_id).strip()
+        inherited_metadata["platform_io_target_group_id"] = normalized_group_id
-        if normalized_group_id:
-            inherited_metadata["platform_io_target_group_id"] = normalized_group_id
 
-    if target_stream.user_id:
+    if target_stream.user_id and (normalized_user_id := str(target_stream.user_id).strip()):
-        normalized_user_id = str(target_stream.user_id).strip()
+        inherited_metadata["platform_io_target_user_id"] = normalized_user_id
-        if normalized_user_id:
-            inherited_metadata["platform_io_target_user_id"] = normalized_user_id
 
     return inherited_metadata
 
 
-def _build_component_from_seg(message_segment: Seg) -> StandardMessageComponents:
+def _build_binary_component_from_base64(component_type: str, raw_data: str) -> StandardMessageComponents:
-    """将单个消息段转换为内部消息组件。
+    """根据 Base64 数据构造二进制消息组件。
 
     Args:
-        message_segment: 待转换的消息段。
+        component_type: 组件类型名称。
+        raw_data: Base64 编码后的二进制数据。
 
     Returns:
         StandardMessageComponents: 转换后的内部消息组件。
+
+    Raises:
+        ValueError: 当组件类型不受支持时抛出。
     """
-    segment_type = str(message_segment.type or "").strip().lower()
+    binary_data = base64.b64decode(raw_data)
-    segment_data = message_segment.data
+    binary_hash = hashlib.sha256(binary_data).hexdigest()
 
-    if segment_type == "text":
+    if component_type == "image":
-        return TextComponent(text=str(segment_data or ""))
+        return ImageComponent(binary_hash=binary_hash, binary_data=binary_data)
-
+    if component_type == "emoji":
-    if segment_type == "image":
+        return EmojiComponent(binary_hash=binary_hash, binary_data=binary_data)
-        image_binary = base64.b64decode(str(segment_data or ""))
+    if component_type == "voice":
-        return ImageComponent(
+        return VoiceComponent(binary_hash=binary_hash, binary_data=binary_data)
-            binary_hash=hashlib.sha256(image_binary).hexdigest(),
+    raise ValueError(f"不支持的二进制组件类型: {component_type}")
-            binary_data=image_binary,
-        )
-
-    if segment_type == "emoji":
-        emoji_binary = base64.b64decode(str(segment_data or ""))
-        return EmojiComponent(
-            binary_hash=hashlib.sha256(emoji_binary).hexdigest(),
-            binary_data=emoji_binary,
-        )
-
-    if segment_type == "voice":
-        voice_binary = base64.b64decode(str(segment_data or ""))
-        return VoiceComponent(
-            binary_hash=hashlib.sha256(voice_binary).hexdigest(),
-            binary_data=voice_binary,
-        )
-
-    if segment_type == "at":
-        return AtComponent(target_user_id=str(segment_data or ""))
-
-    if segment_type == "reply":
-        return ReplyComponent(target_message_id=str(segment_data or ""))
-
-    if segment_type == "dict" and isinstance(segment_data, dict):
-        return DictComponent(data=segment_data)
-
-    return DictComponent(data={"type": segment_type, "data": segment_data})
 
 
-def _build_message_sequence_from_seg(message_segment: Seg) -> MessageSequence:
+def _build_message_sequence_from_custom_message(
-    """将消息段转换为内部消息组件序列。
+    message_type: str,
+    content: str | Dict[str, Any],
+) -> MessageSequence:
+    """根据自定义消息类型构造内部消息组件序列。
 
     Args:
-        message_segment: 待转换的消息段。
+        message_type: 自定义消息类型。
+        content: 自定义消息内容。
 
     Returns:
         MessageSequence: 转换后的消息组件序列。
     """
-    if str(message_segment.type or "").strip().lower() == "seglist":
+    normalized_type = message_type.strip().lower()
-        raw_segments = message_segment.data
-        if not isinstance(raw_segments, list):
-            raise ValueError("seglist 类型的消息段数据必须是列表")
-        components = [
-            _build_component_from_seg(item)
-            for item in raw_segments
-            if isinstance(item, Seg)
-        ]
-        return MessageSequence(components=components)
 
-    return MessageSequence(components=[_build_component_from_seg(message_segment)])
+    if normalized_type == "text":
+        return MessageSequence(components=[TextComponent(text=str(content))])
+
+    if normalized_type in {"image", "emoji", "voice"}:
+        return MessageSequence(
+            components=[_build_binary_component_from_base64(normalized_type, str(content))]
+        )
+
+    if normalized_type == "at":
+        return MessageSequence(components=[AtComponent(target_user_id=str(content))])
+
+    if normalized_type == "reply":
+        return MessageSequence(components=[ReplyComponent(target_message_id=str(content))])
+
+    if normalized_type == "dict" and isinstance(content, dict):
+        return MessageSequence(components=[DictComponent(data=deepcopy(content))])
+
+    return MessageSequence(
+        components=[
+            DictComponent(
+                data={
+                    "type": normalized_type,
+                    "data": deepcopy(content),
+                }
+            )
+        ]
+    )
+
+
+def _clone_message_sequence(message_sequence: MessageSequence) -> MessageSequence:
+    """复制消息组件序列，避免原对象被发送流程修改。
+
+    Args:
+        message_sequence: 原始消息组件序列。
+
+    Returns:
+        MessageSequence: 深拷贝后的消息组件序列。
+    """
+    return deepcopy(message_sequence)
+
+
+def _detect_outbound_message_flags(message_sequence: MessageSequence) -> Dict[str, bool]:
+    """根据消息组件序列推断出站消息标记。
+
+    Args:
+        message_sequence: 待发送的消息组件序列。
+
+    Returns:
+        Dict[str, bool]: 包含 ``is_emoji``、``is_picture``、``is_command`` 的标记字典。
+    """
+    if len(message_sequence.components) != 1:
+        return {
+            "is_emoji": False,
+            "is_picture": False,
+            "is_command": False,
+        }
+
+    component = message_sequence.components[0]
+    is_command = False
+    if isinstance(component, DictComponent) and isinstance(component.data, dict):
+        is_command = str(component.data.get("type") or "").strip().lower() == "command"
+
+    return {
+        "is_emoji": isinstance(component, EmojiComponent),
+        "is_picture": isinstance(component, ImageComponent),
+        "is_command": is_command,
+    }
+
+
+def _describe_message_sequence(message_sequence: MessageSequence) -> str:
+    """生成消息组件序列的简短描述文本。
+
+    Args:
+        message_sequence: 待描述的消息组件序列。
+
+    Returns:
+        str: 适用于日志的简短类型描述。
+    """
+    if len(message_sequence.components) != 1:
+        return "message_sequence"
+
+    component = message_sequence.components[0]
+    if isinstance(component, DictComponent) and isinstance(component.data, dict):
+        custom_type = str(component.data.get("type") or "").strip()
+        return custom_type or "dict"
+
+    if isinstance(component, TextComponent):
+        return component.format_name
+
+    if isinstance(component, ImageComponent):
+        return component.format_name
+
+    if isinstance(component, EmojiComponent):
+        return component.format_name
+
+    if isinstance(component, VoiceComponent):
+        return component.format_name
+
+    if isinstance(component, AtComponent):
+        return component.format_name
+
+    if isinstance(component, ReplyComponent):
+        return component.format_name
+
+    if isinstance(component, ForwardNodeComponent):
+        return component.format_name
+
+    return "unknown"
 
 
 def _build_processed_plain_text(message: SessionMessage) -> str:
@@ -204,7 +279,7 @@ def _build_processed_plain_text(message: SessionMessage) -> str:
 
 
 def _build_outbound_session_message(
-    message_segment: Seg,
+    message_sequence: MessageSequence,
     stream_id: str,
     display_message: str = "",
     reply_message: Optional[MaiMessage] = None,
@@ -213,7 +288,7 @@ def _build_outbound_session_message(
     """根据目标会话构建待发送的内部消息对象。
 
     Args:
-        message_segment: 待发送的消息段。
+        message_sequence: 待发送的消息组件序列。
         stream_id: 目标会话 ID。
         display_message: 用于界面展示的文本内容。
         reply_message: 被回复的锚点消息。
@@ -268,13 +343,14 @@ def _build_outbound_session_message(
         group_info=group_info,
         additional_config=additional_config,
     )
-    outbound_message.raw_message = _build_message_sequence_from_seg(message_segment)
+    outbound_message.raw_message = _clone_message_sequence(message_sequence)
     outbound_message.session_id = target_stream.session_id
     outbound_message.display_message = display_message
     outbound_message.reply_to = anchor_message.message_id if anchor_message is not None else None
-    outbound_message.is_emoji = message_segment.type == "emoji"
+    message_flags = _detect_outbound_message_flags(outbound_message.raw_message)
-    outbound_message.is_picture = message_segment.type == "image"
+    outbound_message.is_emoji = message_flags["is_emoji"]
-    outbound_message.is_command = message_segment.type == "command"
+    outbound_message.is_picture = message_flags["is_picture"]
+    outbound_message.is_command = message_flags["is_command"]
     outbound_message.initialized = True
     return outbound_message
 
@@ -467,7 +543,7 @@ async def send_session_message(
 
 
 async def _send_to_target(
-    message_segment: Seg,
+    message_sequence: MessageSequence,
     stream_id: str,
     display_message: str = "",
     typing: bool = False,
@@ -480,7 +556,7 @@ async def _send_to_target(
     """向指定目标构建并发送消息。
 
     Args:
-        message_segment: 待发送的消息段。
+        message_sequence: 待发送的消息组件序列。
         stream_id: 目标会话 ID。
         display_message: 用于界面展示的文本内容。
         typing: 是否显示输入中状态。
@@ -499,10 +575,10 @@ async def _send_to_target(
             return False
 
         if show_log:
-            logger.debug(f"[SendService] 发送{message_segment.type}消息到 {stream_id}")
+            logger.debug(f"[SendService] 发送{_describe_message_sequence(message_sequence)}消息到 {stream_id}")
 
         outbound_message = _build_outbound_session_message(
-            message_segment=message_segment,
+            message_sequence=message_sequence,
             stream_id=stream_id,
             display_message=display_message,
             reply_message=reply_message,
@@ -555,7 +631,7 @@ async def text_to_stream(
         bool: 发送成功时返回 ``True``。
     """
     return await _send_to_target(
-        message_segment=Seg(type="text", data=text),
+        message_sequence=MessageSequence(components=[TextComponent(text=text)]),
         stream_id=stream_id,
         display_message="",
         typing=typing,
@@ -586,7 +662,7 @@ async def emoji_to_stream(
         bool: 发送成功时返回 ``True``。
     """
     return await _send_to_target(
-        message_segment=Seg(type="emoji", data=emoji_base64),
+        message_sequence=_build_message_sequence_from_custom_message("emoji", emoji_base64),
         stream_id=stream_id,
         display_message="",
         typing=False,
@@ -616,7 +692,7 @@ async def image_to_stream(
         bool: 发送成功时返回 ``True``。
     """
     return await _send_to_target(
-        message_segment=Seg(type="image", data=image_base64),
+        message_sequence=_build_message_sequence_from_custom_message("image", image_base64),
         stream_id=stream_id,
         display_message="",
         typing=False,
@@ -654,7 +730,7 @@ async def custom_to_stream(
         bool: 发送成功时返回 ``True``。
     """
     return await _send_to_target(
-        message_segment=Seg(type=message_type, data=content),  # type: ignore[arg-type]
+        message_sequence=_build_message_sequence_from_custom_message(message_type, content),
         stream_id=stream_id,
         display_message=display_message,
         typing=typing,
@@ -688,28 +764,15 @@ async def custom_reply_set_to_stream(
         show_log: 是否输出发送日志。
 
     Returns:
-        bool: 全部组件发送成功时返回 ``True``。
+        bool: 发送成功时返回 ``True``。
     """
-    success = True
+    return await _send_to_target(
-    for component in reply_set.components:
+        message_sequence=reply_set,
-        if isinstance(component, DictComponent):
+        stream_id=stream_id,
-            message_seg = Seg(type="dict", data=component.data)  # type: ignore[arg-type]
+        display_message=display_message,
-        else:
+        typing=typing,
-            message_seg = await component.to_seg()
+        reply_message=reply_message,
-
+        set_reply=set_reply,
-        status = await _send_to_target(
+        storage_message=storage_message,
-            message_segment=message_seg,
+        show_log=show_log,
-            stream_id=stream_id,
+    )
-            display_message=display_message,
-            typing=typing,
-            reply_message=reply_message,
-            set_reply=set_reply,
-            storage_message=storage_message,
-            show_log=show_log,
-        )
-        if not status:
-            success = False
-            logger.error(f"[SendService] 发送消息组件失败，组件类型：{type(component).__name__}")
-        set_reply = False
-
-    return success