feat: add plugin dependency pipeline and HTML rendering service

- Implemented a new dependency pipeline for plugins to manage Python package dependencies, including conflict detection and automatic installation of missing dependencies.
- Introduced an HTML rendering service that utilizes existing browsers to render HTML content as PNG images, with support for various configurations and error handling.

src/plugin_runtime/__init__.py

@@ -20,5 +20,8 @@ ENV_HOST_VERSION = "MAIBOT_HOST_VERSION"
 ENV_EXTERNAL_PLUGIN_IDS = "MAIBOT_EXTERNAL_PLUGIN_IDS"
 """Runner 启动时可视为已满足的外部插件依赖版本映射（JSON 对象）"""
 
+ENV_BLOCKED_PLUGIN_REASONS = "MAIBOT_BLOCKED_PLUGIN_REASONS"
+"""Runner 启动时收到的拒绝加载插件原因映射（JSON 对象）"""
+
 ENV_GLOBAL_CONFIG_SNAPSHOT = "MAIBOT_GLOBAL_CONFIG_SNAPSHOT"
 """Runner 启动时注入的全局配置快照（JSON 对象）"""

src/plugin_runtime/capabilities/__init__.py

@@ -1,9 +1,11 @@
 from .components import RuntimeComponentCapabilityMixin
 from .core import RuntimeCoreCapabilityMixin
 from .data import RuntimeDataCapabilityMixin
+from .render import RuntimeRenderCapabilityMixin
 
 __all__ = [
     "RuntimeComponentCapabilityMixin",
     "RuntimeCoreCapabilityMixin",
     "RuntimeDataCapabilityMixin",
+    "RuntimeRenderCapabilityMixin",
 ]

src/plugin_runtime/capabilities/registry.py

@@ -91,4 +91,5 @@ def register_capability_impls(manager: "PluginRuntimeManager", supervisor: Plugi
     _register("component.reload_plugin", manager._cap_component_reload_plugin)
 
     _register("knowledge.search", manager._cap_knowledge_search)
+    _register("render.html2png", manager._cap_render_html2png)
     logger.debug("已注册全部主程序能力实现")

src/plugin_runtime/capabilities/render.py

@@ -0,0 +1,121 @@
+"""插件运行时的浏览器渲染能力。"""
+
+from typing import Any, Dict
+
+from src.common.logger import get_logger
+from src.services.html_render_service import HtmlRenderRequest, get_html_render_service
+
+logger = get_logger("plugin_runtime.integration")
+
+
+class RuntimeRenderCapabilityMixin:
+    """插件运行时的浏览器渲染能力混入。"""
+
+    @staticmethod
+    def _coerce_int(value: Any, default: int) -> int:
+        """将任意值尽量转换为整数。
+
+        Args:
+            value: 原始输入值。
+            default: 转换失败时返回的默认值。
+
+        Returns:
+            int: 规范化后的整数结果。
+        """
+
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return default
+
+    @staticmethod
+    def _coerce_float(value: Any, default: float) -> float:
+        """将任意值尽量转换为浮点数。
+
+        Args:
+            value: 原始输入值。
+            default: 转换失败时返回的默认值。
+
+        Returns:
+            float: 规范化后的浮点结果。
+        """
+
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return default
+
+    @staticmethod
+    def _coerce_bool(value: Any, default: bool = False) -> bool:
+        """将任意值转换为布尔值。
+
+        Args:
+            value: 原始输入值。
+            default: 输入为空时返回的默认值。
+
+        Returns:
+            bool: 规范化后的布尔结果。
+        """
+
+        if value is None:
+            return default
+        if isinstance(value, str):
+            normalized_value = value.strip().lower()
+            if normalized_value in {"0", "false", "no", "off"}:
+                return False
+            if normalized_value in {"1", "true", "yes", "on"}:
+                return True
+        return bool(value)
+
+    def _build_html_render_request(self, args: Dict[str, Any]) -> HtmlRenderRequest:
+        """根据 capability 调用参数构造渲染请求。
+
+        Args:
+            args: capability 调用参数。
+
+        Returns:
+            HtmlRenderRequest: 结构化后的渲染请求。
+        """
+
+        viewport = args.get("viewport", {})
+        viewport_width = 900
+        viewport_height = 500
+        if isinstance(viewport, dict):
+            viewport_width = self._coerce_int(viewport.get("width"), viewport_width)
+            viewport_height = self._coerce_int(viewport.get("height"), viewport_height)
+
+        return HtmlRenderRequest(
+            html=str(args.get("html", "") or ""),
+            selector=str(args.get("selector", "body") or "body"),
+            viewport_width=viewport_width,
+            viewport_height=viewport_height,
+            device_scale_factor=self._coerce_float(args.get("device_scale_factor"), 2.0),
+            full_page=self._coerce_bool(args.get("full_page"), False),
+            omit_background=self._coerce_bool(args.get("omit_background"), False),
+            wait_until=str(args.get("wait_until", "load") or "load"),
+            wait_for_selector=str(args.get("wait_for_selector", "") or ""),
+            wait_for_timeout_ms=self._coerce_int(args.get("wait_for_timeout_ms"), 0),
+            timeout_ms=self._coerce_int(args.get("timeout_ms"), 0),
+            allow_network=self._coerce_bool(args.get("allow_network"), False),
+        )
+
+    async def _cap_render_html2png(self, plugin_id: str, capability: str, args: Dict[str, Any]) -> Any:
+        """将 HTML 内容渲染为 PNG 图片。
+
+        Args:
+            plugin_id: 调用该能力的插件 ID。
+            capability: 当前能力名称。
+            args: 能力调用参数。
+
+        Returns:
+            Any: 标准化后的能力返回结构。
+        """
+
+        del plugin_id, capability
+        try:
+            request = self._build_html_render_request(args)
+            result = await get_html_render_service().render_html_to_png(request)
+            return {"success": True, "result": result.to_payload()}
+        except Exception as exc:
+            logger.error(f"[cap.render.html2png] 执行失败: {exc}", exc_info=True)
+            return {"success": False, "error": str(exc)}

src/plugin_runtime/dependency_pipeline.py

@@ -0,0 +1,441 @@
+"""插件 Python 依赖流水线。
+
+负责在 Host 侧统一完成以下工作：
+1. 扫描插件 Manifest；
+2. 检测插件与主程序、插件与插件之间的 Python 依赖冲突；
+3. 为可加载插件自动安装缺失的 Python 依赖；
+4. 产出最终的拒绝加载列表，供运行时使用。
+"""
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Iterable, List, Optional, Sequence, Tuple
+
+import asyncio
+import shutil
+import subprocess
+import sys
+
+from packaging.utils import canonicalize_name
+
+from src.common.logger import get_logger
+from src.plugin_runtime.runner.manifest_validator import ManifestValidator, PluginManifest
+
+
+logger = get_logger("plugin_runtime.dependency_pipeline")
+
+
+@dataclass(frozen=True)
+class PackageDependencyUsage:
+    """记录单个插件对某个 Python 包的依赖声明。"""
+
+    package_name: str
+    plugin_id: str
+    version_spec: str
+
+
+@dataclass(frozen=True)
+class CombinedPackageRequirement:
+    """表示一个已经合并后的 Python 包安装需求。"""
+
+    package_name: str
+    plugin_ids: Tuple[str, ...]
+    requirement_text: str
+    version_spec: str
+
+
+@dataclass(frozen=True)
+class DependencyPipelinePlan:
+    """表示一次依赖分析后得到的计划。"""
+
+    blocked_plugin_reasons: Dict[str, str]
+    install_requirements: Tuple[CombinedPackageRequirement, ...]
+
+
+@dataclass(frozen=True)
+class DependencyPipelineResult:
+    """表示一次依赖流水线执行后的结果。"""
+
+    blocked_plugin_reasons: Dict[str, str]
+    environment_changed: bool
+    install_requirements: Tuple[CombinedPackageRequirement, ...]
+
+
+class PluginDependencyPipeline:
+    """插件依赖流水线。
+
+    该类不负责插件启停，只负责对插件目录进行依赖分析，并在必要时
+    使用 ``uv`` 为可加载插件补齐缺失的 Python 依赖。
+    """
+
+    def __init__(self, project_root: Optional[Path] = None) -> None:
+        """初始化依赖流水线。
+
+        Args:
+            project_root: 项目根目录；留空时自动推断。
+        """
+
+        self._project_root: Path = project_root or Path(__file__).resolve().parents[2]
+        self._manifest_validator: ManifestValidator = ManifestValidator(
+            project_root=self._project_root,
+            validate_python_package_dependencies=False,
+        )
+
+    async def execute(self, plugin_dirs: Iterable[Path]) -> DependencyPipelineResult:
+        """执行完整的依赖分析与自动安装流程。
+
+        Args:
+            plugin_dirs: 需要扫描的插件根目录集合。
+
+        Returns:
+            DependencyPipelineResult: 最终的阻止加载结果与环境变更状态。
+        """
+
+        plan = self.build_plan(plugin_dirs)
+        if not plan.install_requirements:
+            return DependencyPipelineResult(
+                blocked_plugin_reasons=dict(plan.blocked_plugin_reasons),
+                environment_changed=False,
+                install_requirements=plan.install_requirements,
+            )
+
+        install_succeeded, error_message = await self._install_requirements(plan.install_requirements)
+        if install_succeeded:
+            return DependencyPipelineResult(
+                blocked_plugin_reasons=dict(plan.blocked_plugin_reasons),
+                environment_changed=True,
+                install_requirements=plan.install_requirements,
+            )
+
+        blocked_plugin_reasons = dict(plan.blocked_plugin_reasons)
+        affected_plugin_ids = sorted(
+            {
+                plugin_id
+                for requirement in plan.install_requirements
+                for plugin_id in requirement.plugin_ids
+            }
+        )
+        for plugin_id in affected_plugin_ids:
+            self._append_block_reason(
+                blocked_plugin_reasons,
+                plugin_id,
+                f"自动安装 Python 依赖失败: {error_message}",
+            )
+
+        return DependencyPipelineResult(
+            blocked_plugin_reasons=blocked_plugin_reasons,
+            environment_changed=False,
+            install_requirements=plan.install_requirements,
+        )
+
+    def build_plan(self, plugin_dirs: Iterable[Path]) -> DependencyPipelinePlan:
+        """构建依赖分析计划。
+
+        Args:
+            plugin_dirs: 需要扫描的插件根目录集合。
+
+        Returns:
+            DependencyPipelinePlan: 分析后的阻止加载列表与安装计划。
+        """
+
+        manifests = self._collect_manifests(plugin_dirs)
+        blocked_plugin_reasons = self._detect_host_conflicts(manifests)
+        plugin_conflict_reasons = self._detect_plugin_conflicts(manifests, blocked_plugin_reasons)
+        for plugin_id, reason in plugin_conflict_reasons.items():
+            self._append_block_reason(blocked_plugin_reasons, plugin_id, reason)
+
+        install_requirements = self._build_install_requirements(manifests, blocked_plugin_reasons)
+        return DependencyPipelinePlan(
+            blocked_plugin_reasons=blocked_plugin_reasons,
+            install_requirements=install_requirements,
+        )
+
+    def _collect_manifests(self, plugin_dirs: Iterable[Path]) -> Dict[str, PluginManifest]:
+        """收集所有可成功解析的插件 Manifest。
+
+        Args:
+            plugin_dirs: 需要扫描的插件根目录集合。
+
+        Returns:
+            Dict[str, PluginManifest]: 以插件 ID 为键的 Manifest 映射。
+        """
+
+        manifests: Dict[str, PluginManifest] = {}
+        for _plugin_path, manifest in self._manifest_validator.iter_plugin_manifests(plugin_dirs):
+            manifests[manifest.id] = manifest
+        return manifests
+
+    def _detect_host_conflicts(self, manifests: Dict[str, PluginManifest]) -> Dict[str, str]:
+        """检测插件与主程序依赖之间的冲突。
+
+        Args:
+            manifests: 当前已解析到的插件 Manifest 映射。
+
+        Returns:
+            Dict[str, str]: 需要被阻止加载的插件及原因。
+        """
+
+        host_requirements = self._manifest_validator.load_host_dependency_requirements()
+        blocked_plugin_reasons: Dict[str, str] = {}
+
+        for manifest in manifests.values():
+            for dependency in manifest.python_package_dependencies:
+                package_specifier = self._manifest_validator.build_specifier_set(dependency.version_spec)
+                if package_specifier is None:
+                    self._append_block_reason(
+                        blocked_plugin_reasons,
+                        manifest.id,
+                        f"Python 包依赖声明无效: {dependency.name}{dependency.version_spec}",
+                    )
+                    continue
+
+                normalized_package_name = canonicalize_name(dependency.name)
+                host_requirement = host_requirements.get(normalized_package_name)
+                if host_requirement is None:
+                    continue
+
+                if self._manifest_validator.requirements_may_overlap(
+                    host_requirement.specifier,
+                    package_specifier,
+                ):
+                    continue
+
+                host_specifier_text = str(host_requirement.specifier or "") or "任意版本"
+                self._append_block_reason(
+                    blocked_plugin_reasons,
+                    manifest.id,
+                    (
+                        f"Python 包依赖与主程序冲突: {dependency.name} 需要 "
+                        f"{dependency.version_spec}，主程序约束为 {host_specifier_text}"
+                    ),
+                )
+
+        return blocked_plugin_reasons
+
+    def _detect_plugin_conflicts(
+        self,
+        manifests: Dict[str, PluginManifest],
+        blocked_plugin_reasons: Dict[str, str],
+    ) -> Dict[str, str]:
+        """检测插件之间的 Python 依赖冲突。
+
+        Args:
+            manifests: 当前已解析到的插件 Manifest 映射。
+            blocked_plugin_reasons: 已经因为其他原因被阻止加载的插件。
+
+        Returns:
+            Dict[str, str]: 新增的插件冲突原因映射。
+        """
+
+        blocked_by_plugin_conflicts: Dict[str, str] = {}
+        dependency_usages = self._collect_package_usages(manifests, blocked_plugin_reasons)
+
+        for _package_name, usages in dependency_usages.items():
+            display_package_name = usages[0].package_name
+            for index, left_usage in enumerate(usages):
+                for right_usage in usages[index + 1 :]:
+                    left_specifier = self._manifest_validator.build_specifier_set(left_usage.version_spec)
+                    right_specifier = self._manifest_validator.build_specifier_set(right_usage.version_spec)
+                    if left_specifier is None or right_specifier is None:
+                        continue
+
+                    if self._manifest_validator.requirements_may_overlap(left_specifier, right_specifier):
+                        continue
+
+                    left_reason = (
+                        f"Python 包依赖冲突: 与插件 {right_usage.plugin_id} 在 {display_package_name} 上的约束不兼容 "
+                        f"({left_usage.version_spec} vs {right_usage.version_spec})"
+                    )
+                    right_reason = (
+                        f"Python 包依赖冲突: 与插件 {left_usage.plugin_id} 在 {display_package_name} 上的约束不兼容 "
+                        f"({right_usage.version_spec} vs {left_usage.version_spec})"
+                    )
+                    self._append_block_reason(blocked_by_plugin_conflicts, left_usage.plugin_id, left_reason)
+                    self._append_block_reason(blocked_by_plugin_conflicts, right_usage.plugin_id, right_reason)
+
+        return blocked_by_plugin_conflicts
+
+    def _collect_package_usages(
+        self,
+        manifests: Dict[str, PluginManifest],
+        blocked_plugin_reasons: Dict[str, str],
+    ) -> Dict[str, List[PackageDependencyUsage]]:
+        """收集所有未被阻止加载插件的包依赖声明。
+
+        Args:
+            manifests: 当前已解析到的插件 Manifest 映射。
+            blocked_plugin_reasons: 已经被阻止加载的插件及原因。
+
+        Returns:
+            Dict[str, List[PackageDependencyUsage]]: 按规范化包名分组后的依赖声明。
+        """
+
+        dependency_usages: Dict[str, List[PackageDependencyUsage]] = {}
+        for manifest in manifests.values():
+            if manifest.id in blocked_plugin_reasons:
+                continue
+
+            for dependency in manifest.python_package_dependencies:
+                normalized_package_name = canonicalize_name(dependency.name)
+                dependency_usages.setdefault(normalized_package_name, []).append(
+                    PackageDependencyUsage(
+                        package_name=dependency.name,
+                        plugin_id=manifest.id,
+                        version_spec=dependency.version_spec,
+                    )
+                )
+
+        return dependency_usages
+
+    def _build_install_requirements(
+        self,
+        manifests: Dict[str, PluginManifest],
+        blocked_plugin_reasons: Dict[str, str],
+    ) -> Tuple[CombinedPackageRequirement, ...]:
+        """构建需要安装到当前环境的 Python 包需求列表。
+
+        Args:
+            manifests: 当前已解析到的插件 Manifest 映射。
+            blocked_plugin_reasons: 已经被阻止加载的插件及原因。
+
+        Returns:
+            Tuple[CombinedPackageRequirement, ...]: 需要安装或调整版本的依赖列表。
+        """
+
+        combined_requirements: List[CombinedPackageRequirement] = []
+        dependency_usages = self._collect_package_usages(manifests, blocked_plugin_reasons)
+
+        for usages in dependency_usages.values():
+            merged_specifier_text = self._merge_specifier_texts([usage.version_spec for usage in usages])
+            package_name = usages[0].package_name
+            requirement_text = f"{package_name}{merged_specifier_text}"
+            installed_version = self._manifest_validator.get_installed_package_version(package_name)
+            if installed_version is not None and self._manifest_validator.version_matches_specifier(
+                installed_version,
+                merged_specifier_text,
+            ):
+                continue
+
+            combined_requirements.append(
+                CombinedPackageRequirement(
+                    package_name=package_name,
+                    plugin_ids=tuple(sorted({usage.plugin_id for usage in usages})),
+                    requirement_text=requirement_text,
+                    version_spec=merged_specifier_text,
+                )
+            )
+
+        return tuple(sorted(combined_requirements, key=lambda requirement: canonicalize_name(requirement.package_name)))
+
+    @staticmethod
+    def _merge_specifier_texts(specifier_texts: Sequence[str]) -> str:
+        """合并多个版本约束文本。
+
+        Args:
+            specifier_texts: 需要合并的版本约束文本序列。
+
+        Returns:
+            str: 合并后的版本约束文本。
+        """
+
+        merged_parts: List[str] = []
+        for specifier_text in specifier_texts:
+            for part in str(specifier_text or "").split(","):
+                normalized_part = part.strip()
+                if not normalized_part or normalized_part in merged_parts:
+                    continue
+                merged_parts.append(normalized_part)
+        return f"{','.join(merged_parts)}" if merged_parts else ""
+
+    async def _install_requirements(self, requirements: Sequence[CombinedPackageRequirement]) -> Tuple[bool, str]:
+        """安装指定的 Python 包需求列表。
+
+        Args:
+            requirements: 需要安装的依赖列表。
+
+        Returns:
+            Tuple[bool, str]: 安装是否成功，以及错误摘要。
+        """
+
+        requirement_texts = [requirement.requirement_text for requirement in requirements]
+        if not requirement_texts:
+            return True, ""
+
+        logger.info(f"开始自动安装插件 Python 依赖: {', '.join(requirement_texts)}")
+        command = self._build_install_command(requirement_texts)
+
+        try:
+            completed_process = await asyncio.to_thread(
+                subprocess.run,
+                command,
+                capture_output=True,
+                check=False,
+                cwd=self._project_root,
+                text=True,
+            )
+        except Exception as exc:
+            return False, str(exc)
+
+        if completed_process.returncode == 0:
+            logger.info("插件 Python 依赖自动安装完成")
+            return True, ""
+
+        output = self._summarize_install_error(completed_process.stdout, completed_process.stderr)
+        return False, output or f"命令执行失败，退出码 {completed_process.returncode}"
+
+    @staticmethod
+    def _build_install_command(requirement_texts: Sequence[str]) -> List[str]:
+        """构造依赖安装命令。
+
+        Args:
+            requirement_texts: 待安装的依赖文本序列。
+
+        Returns:
+            List[str]: 适用于 ``subprocess.run`` 的命令参数列表。
+        """
+
+        if shutil.which("uv"):
+            return ["uv", "pip", "install", "--python", sys.executable, *requirement_texts]
+        return [sys.executable, "-m", "pip", "install", *requirement_texts]
+
+    @staticmethod
+    def _summarize_install_error(stdout: str, stderr: str) -> str:
+        """提炼安装失败输出。
+
+        Args:
+            stdout: 标准输出内容。
+            stderr: 标准错误内容。
+
+        Returns:
+            str: 简短的错误摘要。
+        """
+
+        merged_output = "\n".join(part.strip() for part in (stderr, stdout) if part and part.strip()).strip()
+        if not merged_output:
+            return ""
+        lines = [line.strip() for line in merged_output.splitlines() if line.strip()]
+        return " | ".join(lines[-5:])
+
+    @staticmethod
+    def _append_block_reason(
+        blocked_plugin_reasons: Dict[str, str],
+        plugin_id: str,
+        reason: str,
+    ) -> None:
+        """向阻止加载映射中追加原因。
+
+        Args:
+            blocked_plugin_reasons: 待更新的阻止加载映射。
+            plugin_id: 目标插件 ID。
+            reason: 需要追加的原因文本。
+        """
+
+        existing_reason = blocked_plugin_reasons.get(plugin_id)
+        if existing_reason is None:
+            blocked_plugin_reasons[plugin_id] = reason
+            return
+
+        existing_parts = [part.strip() for part in existing_reason.split("；") if part.strip()]
+        if reason in existing_parts:
+            return
+        blocked_plugin_reasons[plugin_id] = f"{existing_reason}；{reason}"

src/plugin_runtime/host/supervisor.py

@@ -14,6 +14,7 @@ from src.platform_io import DriverKind, InboundMessageEnvelope, RouteBinding, Ro
 from src.platform_io.drivers import PluginPlatformDriver
 from src.platform_io.route_key_factory import RouteKeyFactory
 from src.plugin_runtime import (
+    ENV_BLOCKED_PLUGIN_REASONS,
     ENV_EXTERNAL_PLUGIN_IDS,
     ENV_GLOBAL_CONFIG_SNAPSHOT,
     ENV_HOST_VERSION,
@@ -131,6 +132,7 @@ class PluginRunnerSupervisor:
         self._registered_plugins: Dict[str, RegisterPluginPayload] = {}
         self._message_gateway_states: Dict[str, Dict[str, _MessageGatewayRuntimeState]] = {}
         self._external_available_plugins: Dict[str, str] = {}
+        self._blocked_plugin_reasons: Dict[str, str] = {}
         self._runner_ready_events: asyncio.Event = asyncio.Event()
         self._runner_ready_payloads: RunnerReadyPayload = RunnerReadyPayload()
         self._health_task: Optional[asyncio.Task[None]] = None
@@ -211,6 +213,19 @@ class PluginRunnerSupervisor:
         """
         return {plugin_id: registration.plugin_version for plugin_id, registration in self._registered_plugins.items()}
 
+    def set_blocked_plugin_reasons(self, blocked_plugin_reasons: Dict[str, str]) -> None:
+        """设置当前 Runner 启动时应拒绝加载的插件列表。
+
+        Args:
+            blocked_plugin_reasons: 需要拒绝加载的插件及原因映射。
+        """
+
+        self._blocked_plugin_reasons = {
+            str(plugin_id or "").strip(): str(reason or "").strip()
+            for plugin_id, reason in blocked_plugin_reasons.items()
+            if str(plugin_id or "").strip() and str(reason or "").strip()
+        }
+
     @staticmethod
     def _normalize_reload_plugin_ids(plugin_ids: Optional[List[str] | str]) -> List[str]:
         """规范化批量重载入参。
@@ -1303,6 +1318,7 @@ class PluginRunnerSupervisor:
         global_config_snapshot = config_manager.get_global_config().model_dump(mode="json")
         global_config_snapshot["model"] = config_manager.get_model_config().model_dump(mode="json")
         return {
+            ENV_BLOCKED_PLUGIN_REASONS: json.dumps(self._blocked_plugin_reasons, ensure_ascii=False),
             ENV_EXTERNAL_PLUGIN_IDS: json.dumps(self._external_available_plugins, ensure_ascii=False),
             ENV_GLOBAL_CONFIG_SNAPSHOT: json.dumps(global_config_snapshot, ensure_ascii=False),
             ENV_HOST_VERSION: PROTOCOL_VERSION,

src/plugin_runtime/integration.py

@@ -8,6 +8,7 @@
 5. 提供统一的能力实现注册接口，使插件可以调用主程序功能
 """
 
+from dataclasses import dataclass
 from pathlib import Path
 from typing import (
     TYPE_CHECKING,
@@ -33,13 +34,15 @@ from src.common.logger import get_logger
 from src.config.config import config_manager
 from src.config.file_watcher import FileChange, FileWatcher
 from src.platform_io import DeliveryBatch, InboundMessageEnvelope, get_platform_io_manager
-from src.plugin_runtime.hook_catalog import register_builtin_hook_specs
 from src.plugin_runtime.capabilities import (
     RuntimeComponentCapabilityMixin,
     RuntimeCoreCapabilityMixin,
     RuntimeDataCapabilityMixin,
+    RuntimeRenderCapabilityMixin,
 )
 from src.plugin_runtime.capabilities.registry import register_capability_impls
+from src.plugin_runtime.dependency_pipeline import PluginDependencyPipeline
+from src.plugin_runtime.hook_catalog import register_builtin_hook_specs
 from src.plugin_runtime.host.hook_dispatcher import HookDispatchResult, HookDispatcher
 from src.plugin_runtime.host.hook_spec_registry import HookSpec, HookSpecRegistry
 from src.plugin_runtime.host.message_utils import MessageDict, PluginMessageUtils
@@ -67,10 +70,19 @@ _EVENT_TYPE_MAP: Dict[str, str] = {
 }
 
 
+@dataclass(frozen=True)
+class DependencySyncState:
+    """表示一次插件依赖同步后的状态。"""
+
+    blocked_changed_plugin_ids: Set[str]
+    environment_changed: bool
+
+
 class PluginRuntimeManager(
     RuntimeCoreCapabilityMixin,
     RuntimeDataCapabilityMixin,
     RuntimeComponentCapabilityMixin,
+    RuntimeRenderCapabilityMixin,
 ):
     """插件运行时管理器（单例）
 
@@ -88,7 +100,9 @@ class PluginRuntimeManager(
         self._plugin_source_watcher_subscription_id: Optional[str] = None
         self._plugin_config_watcher_subscriptions: Dict[str, Tuple[Path, str]] = {}
         self._plugin_path_cache: Dict[str, Path] = {}
-        self._manifest_validator: ManifestValidator = ManifestValidator()
+        self._manifest_validator: ManifestValidator = ManifestValidator(validate_python_package_dependencies=False)
+        self._plugin_dependency_pipeline: PluginDependencyPipeline = PluginDependencyPipeline()
+        self._blocked_plugin_reasons: Dict[str, str] = {}
         self._config_reload_callback: Callable[[Sequence[str]], Awaitable[None]] = self._handle_main_config_reload
         self._config_reload_callback_registered: bool = False
         self._hook_spec_registry: HookSpecRegistry = HookSpecRegistry()
@@ -131,7 +145,7 @@ class PluginRuntimeManager(
     @classmethod
     def _discover_plugin_dependency_map(cls, plugin_dirs: Iterable[Path]) -> Dict[str, List[str]]:
         """扫描指定插件目录集合，返回 ``plugin_id -> dependencies`` 映射。"""
-        validator = ManifestValidator()
+        validator = ManifestValidator(validate_python_package_dependencies=False)
         return validator.build_plugin_dependency_map(plugin_dirs)
 
     @classmethod
@@ -191,6 +205,206 @@ class PluginRuntimeManager(
         }
         return supervisor_cls(**supported_kwargs)
 
+    def _resolve_runtime_plugin_dirs(self) -> Tuple[List[Path], List[Path]]:
+        """解析当前运行时应管理的插件根目录。
+
+        Returns:
+            Tuple[List[Path], List[Path]]: 内置插件目录列表与第三方插件目录列表。
+        """
+
+        return self._get_builtin_plugin_dirs(), self._get_third_party_plugin_dirs()
+
+    @staticmethod
+    def _resolve_supervisor_socket_paths() -> Tuple[Optional[str], Optional[str]]:
+        """解析内置与第三方 Supervisor 的 IPC 地址。
+
+        Returns:
+            Tuple[Optional[str], Optional[str]]: 内置 Runner 与第三方 Runner 的 socket 地址。
+        """
+
+        runtime_config = config_manager.get_global_config().plugin_runtime
+        socket_path_base = runtime_config.ipc_socket_path or None
+        builtin_socket = f"{socket_path_base}-builtin" if socket_path_base else None
+        third_party_socket = f"{socket_path_base}-third_party" if socket_path_base else None
+        return builtin_socket, third_party_socket
+
+    def _apply_blocked_plugin_reasons_to_supervisors(self) -> None:
+        """将当前阻止加载插件列表同步到全部 Supervisor。"""
+
+        for supervisor in self.supervisors:
+            set_blocked_plugin_reasons = getattr(supervisor, "set_blocked_plugin_reasons", None)
+            if callable(set_blocked_plugin_reasons):
+                set_blocked_plugin_reasons(self._blocked_plugin_reasons)
+
+    def _set_blocked_plugin_reasons(self, blocked_plugin_reasons: Dict[str, str]) -> Set[str]:
+        """更新 Host 侧维护的阻止加载插件列表。
+
+        Args:
+            blocked_plugin_reasons: 最新的阻止加载插件及原因映射。
+
+        Returns:
+            Set[str]: 本次发生状态变化的插件 ID 集合。
+        """
+
+        normalized_reasons = {
+            str(plugin_id or "").strip(): str(reason or "").strip()
+            for plugin_id, reason in blocked_plugin_reasons.items()
+            if str(plugin_id or "").strip() and str(reason or "").strip()
+        }
+        changed_plugin_ids = {
+            plugin_id
+            for plugin_id in set(self._blocked_plugin_reasons) | set(normalized_reasons)
+            if self._blocked_plugin_reasons.get(plugin_id) != normalized_reasons.get(plugin_id)
+        }
+        self._blocked_plugin_reasons = normalized_reasons
+        self._apply_blocked_plugin_reasons_to_supervisors()
+        return changed_plugin_ids
+
+    async def _sync_plugin_dependencies(self, plugin_dirs: Sequence[Path]) -> DependencySyncState:
+        """执行插件依赖同步，并刷新阻止加载插件列表。
+
+        Args:
+            plugin_dirs: 当前需要参与分析的插件根目录列表。
+
+        Returns:
+            DependencySyncState: 同步后的环境变更状态与阻止列表变化集合。
+        """
+
+        result = await self._plugin_dependency_pipeline.execute(plugin_dirs)
+        changed_plugin_ids = self._set_blocked_plugin_reasons(result.blocked_plugin_reasons)
+        return DependencySyncState(
+            blocked_changed_plugin_ids=changed_plugin_ids,
+            environment_changed=result.environment_changed,
+        )
+
+    def _build_supervisors(self, builtin_dirs: Sequence[Path], third_party_dirs: Sequence[Path]) -> None:
+        """根据目录列表创建当前运行时所需的 Supervisor。
+
+        Args:
+            builtin_dirs: 内置插件目录列表。
+            third_party_dirs: 第三方插件目录列表。
+        """
+
+        from src.plugin_runtime.host.supervisor import PluginSupervisor
+
+        builtin_socket, third_party_socket = self._resolve_supervisor_socket_paths()
+        self._builtin_supervisor = None
+        self._third_party_supervisor = None
+
+        if builtin_dirs:
+            builtin_supervisor = self._instantiate_supervisor(
+                PluginSupervisor,
+                plugin_dirs=list(builtin_dirs),
+                group_name="builtin",
+                hook_spec_registry=self._hook_spec_registry,
+                socket_path=builtin_socket,
+            )
+            self._builtin_supervisor = builtin_supervisor
+            self._register_capability_impls(builtin_supervisor)
+
+        if third_party_dirs:
+            third_party_supervisor = self._instantiate_supervisor(
+                PluginSupervisor,
+                plugin_dirs=list(third_party_dirs),
+                group_name="third_party",
+                hook_spec_registry=self._hook_spec_registry,
+                socket_path=third_party_socket,
+            )
+            self._third_party_supervisor = third_party_supervisor
+            self._register_capability_impls(third_party_supervisor)
+
+        self._apply_blocked_plugin_reasons_to_supervisors()
+
+    async def _start_supervisors(
+        self,
+        builtin_dirs: Sequence[Path],
+        third_party_dirs: Sequence[Path],
+    ) -> List["PluginSupervisor"]:
+        """按依赖顺序启动当前已创建的 Supervisor。
+
+        Args:
+            builtin_dirs: 内置插件目录列表。
+            third_party_dirs: 第三方插件目录列表。
+
+        Returns:
+            List[PluginSupervisor]: 成功启动的 Supervisor 列表。
+        """
+
+        started_supervisors: List["PluginSupervisor"] = []
+        supervisor_groups: Dict[str, Optional["PluginSupervisor"]] = {
+            "builtin": self._builtin_supervisor,
+            "third_party": self._third_party_supervisor,
+        }
+        start_order = self._build_group_start_order(builtin_dirs, third_party_dirs)
+
+        try:
+            for group_name in start_order:
+                supervisor = supervisor_groups.get(group_name)
+                if supervisor is None:
+                    continue
+
+                external_plugin_versions = {
+                    plugin_id: plugin_version
+                    for started_supervisor in started_supervisors
+                    for plugin_id, plugin_version in started_supervisor.get_loaded_plugin_versions().items()
+                }
+                supervisor.set_external_available_plugins(external_plugin_versions)
+                set_blocked_plugin_reasons = getattr(supervisor, "set_blocked_plugin_reasons", None)
+                if callable(set_blocked_plugin_reasons):
+                    set_blocked_plugin_reasons(self._blocked_plugin_reasons)
+                await supervisor.start()
+                started_supervisors.append(supervisor)
+        except Exception:
+            await asyncio.gather(*(supervisor.stop() for supervisor in started_supervisors), return_exceptions=True)
+            raise
+
+        return started_supervisors
+
+    async def _stop_supervisors(self) -> None:
+        """停止当前全部 Supervisor。"""
+
+        supervisors = self.supervisors
+        if not supervisors:
+            return
+
+        await asyncio.gather(*(supervisor.stop() for supervisor in supervisors), return_exceptions=True)
+        self._builtin_supervisor = None
+        self._third_party_supervisor = None
+
+    async def _restart_supervisors(self, reason: str) -> bool:
+        """重启当前全部 Supervisor。
+
+        Args:
+            reason: 本次重启的原因。
+
+        Returns:
+            bool: 是否重启成功。
+        """
+
+        builtin_dirs, third_party_dirs = self._resolve_runtime_plugin_dirs()
+        if duplicate_plugin_ids := self._find_duplicate_plugin_ids(builtin_dirs + third_party_dirs):
+            details = "; ".join(
+                f"{plugin_id}: {', '.join(str(path) for path in paths)}"
+                for plugin_id, paths in sorted(duplicate_plugin_ids.items())
+            )
+            logger.error(f"检测到重复插件 ID，拒绝执行 Supervisor 重启: {details}")
+            return False
+
+        logger.info(f"开始重启插件运行时 Supervisor: {reason}")
+        await self._stop_supervisors()
+        self._build_supervisors(builtin_dirs, third_party_dirs)
+
+        try:
+            await self._start_supervisors(builtin_dirs, third_party_dirs)
+        except Exception as exc:
+            logger.error(f"重启插件运行时 Supervisor 失败: {exc}", exc_info=True)
+            await self._stop_supervisors()
+            return False
+
+        self._refresh_plugin_config_watch_subscriptions()
+        logger.info(f"插件运行时 Supervisor 已重启完成: {reason}")
+        return True
+
     # ─── 生命周期 ─────────────────────────────────────────────
 
     async def start(self) -> None:
@@ -204,10 +418,7 @@ class PluginRuntimeManager(
             logger.info("插件运行时已在配置中禁用，跳过启动")
             return
 
-        from src.plugin_runtime.host.supervisor import PluginSupervisor
-
-        builtin_dirs = self._get_builtin_plugin_dirs()
-        third_party_dirs = self._get_third_party_plugin_dirs()
+        builtin_dirs, third_party_dirs = self._resolve_runtime_plugin_dirs()
 
         if duplicate_plugin_ids := self._find_duplicate_plugin_ids(builtin_dirs + third_party_dirs):
             details = "; ".join(
@@ -221,61 +432,19 @@ class PluginRuntimeManager(
             logger.info("未找到任何插件目录，跳过插件运行时启动")
             return
 
+        dependency_sync_state = await self._sync_plugin_dependencies(builtin_dirs + third_party_dirs)
+        if dependency_sync_state.environment_changed:
+            logger.info("插件依赖流水线已更新当前 Python 环境，启动时将直接加载最新环境")
+
         self.ensure_builtin_hook_specs_registered()
         platform_io_manager = get_platform_io_manager()
+        self._build_supervisors(builtin_dirs, third_party_dirs)
 
-        # 从配置读取自定义 IPC socket 路径（留空则自动生成）
-        socket_path_base = _cfg.ipc_socket_path or None
-
-        # 当用户指定了自定义路径时，为两个 Supervisor 添加后缀以避免 UDS 冲突
-        builtin_socket = f"{socket_path_base}-builtin" if socket_path_base else None
-        third_party_socket = f"{socket_path_base}-third_party" if socket_path_base else None
-
-        # 创建两个 Supervisor，各自拥有独立的 socket / Runner 子进程
-        if builtin_dirs:
-            self._builtin_supervisor = self._instantiate_supervisor(
-                PluginSupervisor,
-                plugin_dirs=builtin_dirs,
-                group_name="builtin",
-                hook_spec_registry=self._hook_spec_registry,
-                socket_path=builtin_socket,
-            )
-            self._register_capability_impls(self._builtin_supervisor)
-
-        if third_party_dirs:
-            self._third_party_supervisor = self._instantiate_supervisor(
-                PluginSupervisor,
-                plugin_dirs=third_party_dirs,
-                group_name="third_party",
-                hook_spec_registry=self._hook_spec_registry,
-                socket_path=third_party_socket,
-            )
-            self._register_capability_impls(self._third_party_supervisor)
-
-        started_supervisors: List[PluginSupervisor] = []
+        started_supervisors: List["PluginSupervisor"] = []
         try:
             platform_io_manager.set_inbound_dispatcher(self._dispatch_platform_inbound)
             await platform_io_manager.ensure_send_pipeline_ready()
-
-            supervisor_groups: Dict[str, Optional[PluginSupervisor]] = {
-                "builtin": self._builtin_supervisor,
-                "third_party": self._third_party_supervisor,
-            }
-            start_order = self._build_group_start_order(builtin_dirs, third_party_dirs)
-
-            for group_name in start_order:
-                supervisor = supervisor_groups.get(group_name)
-                if supervisor is None:
-                    continue
-
-                external_plugin_versions = {
-                    plugin_id: plugin_version
-                    for started_supervisor in started_supervisors
-                    for plugin_id, plugin_version in started_supervisor.get_loaded_plugin_versions().items()
-                }
-                supervisor.set_external_available_plugins(external_plugin_versions)
-                await supervisor.start()
-                started_supervisors.append(supervisor)
+            started_supervisors = await self._start_supervisors(builtin_dirs, third_party_dirs)
 
             await self._start_plugin_file_watcher()
             config_manager.register_reload_callback(self._config_reload_callback)
@@ -529,6 +698,21 @@ class PluginRuntimeManager(
         if not normalized_plugin_ids:
             return True
 
+        blocked_plugin_ids = [plugin_id for plugin_id in normalized_plugin_ids if plugin_id in self._blocked_plugin_reasons]
+        if blocked_plugin_ids:
+            logger.warning(
+                "以下插件当前被依赖流水线阻止加载，已拒绝重载请求: "
+                + ", ".join(
+                    f"{plugin_id} ({self._blocked_plugin_reasons[plugin_id]})"
+                    for plugin_id in sorted(blocked_plugin_ids)
+                )
+            )
+            normalized_plugin_ids = [
+                plugin_id for plugin_id in normalized_plugin_ids if plugin_id not in self._blocked_plugin_reasons
+            ]
+            if not normalized_plugin_ids:
+                return False
+
         dependency_map = self._build_registered_dependency_map()
         supervisor_by_plugin = self._build_registered_supervisor_map()
         supervisor_roots: Dict["PluginSupervisor", List[str]] = {}
@@ -909,6 +1093,12 @@ class PluginRuntimeManager(
         normalized_plugin_id = str(plugin_id or "").strip()
         if not normalized_plugin_id:
             return False
+        if normalized_plugin_id in self._blocked_plugin_reasons:
+            logger.warning(
+                f"插件 {normalized_plugin_id} 当前被依赖流水线阻止加载: "
+                f"{self._blocked_plugin_reasons[normalized_plugin_id]}"
+            )
+            return False
 
         try:
             registered_supervisor = self._get_supervisor_for_plugin(normalized_plugin_id)
@@ -933,7 +1123,7 @@ class PluginRuntimeManager(
     def _find_duplicate_plugin_ids(cls, plugin_dirs: List[Path]) -> Dict[str, List[Path]]:
         """扫描插件目录，找出被多个目录重复声明的插件 ID。"""
         plugin_locations: Dict[str, List[Path]] = {}
-        validator = ManifestValidator()
+        validator = ManifestValidator(validate_python_package_dependencies=False)
         for plugin_path, manifest in validator.iter_plugin_manifests(plugin_dirs):
             plugin_locations.setdefault(manifest.id, []).append(plugin_path)
 
@@ -1190,7 +1380,8 @@ class PluginRuntimeManager(
         if not self._started or not changes:
             return
 
-        if duplicate_plugin_ids := self._find_duplicate_plugin_ids(list(self._iter_plugin_dirs())):
+        plugin_dirs = list(self._iter_plugin_dirs())
+        if duplicate_plugin_ids := self._find_duplicate_plugin_ids(plugin_dirs):
             details = "; ".join(
                 f"{plugin_id}: {', '.join(str(path) for path in paths)}"
                 for plugin_id, paths in sorted(duplicate_plugin_ids.items())
@@ -1198,21 +1389,24 @@ class PluginRuntimeManager(
             logger.error(f"检测到重复插件 ID，跳过本次插件热重载: {details}")
             return
 
-        changed_plugin_ids: List[str] = []
-        changed_paths = [change.path.resolve() for change in changes]
+        relevant_source_changes = [
+            change.path.resolve()
+            for change in changes
+            if change.path.name in {"plugin.py", "_manifest.json"} or change.path.suffix == ".py"
+        ]
+        if not relevant_source_changes:
+            return
 
-        for supervisor in self.supervisors:
-            for path in changed_paths:
-                plugin_id = self._match_plugin_id_for_supervisor(supervisor, path)
-                if plugin_id is None:
-                    continue
-                if path.name in {"plugin.py", "_manifest.json"} or path.suffix == ".py":
-                    if plugin_id not in changed_plugin_ids:
-                        changed_plugin_ids.append(plugin_id)
+        dependency_sync_state = await self._sync_plugin_dependencies(plugin_dirs)
+        restart_reason = "file_watcher"
+        if dependency_sync_state.environment_changed:
+            restart_reason = "file_watcher_dependency_install"
+        elif dependency_sync_state.blocked_changed_plugin_ids:
+            restart_reason = "file_watcher_blocklist_changed"
 
-        if changed_plugin_ids:
-            await self.reload_plugins_globally(changed_plugin_ids, reason="file_watcher")
-            self._refresh_plugin_config_watch_subscriptions()
+        restarted = await self._restart_supervisors(restart_reason)
+        if not restarted:
+            logger.warning(f"插件源码变更后重启 Supervisor 失败: {restart_reason}")
 
     @staticmethod
     def _plugin_dir_matches(path: Path, plugin_dir: Path) -> bool:

src/plugin_runtime/runner/manifest_validator.py

@@ -609,6 +609,7 @@ class ManifestValidator:
         host_version: str = "",
         sdk_version: str = "",
         project_root: Optional[Path] = None,
+        validate_python_package_dependencies: bool = True,
     ) -> None:
         """初始化 Manifest 校验器。
 
@@ -616,10 +617,12 @@ class ManifestValidator:
             host_version: 当前 Host 版本号；留空时自动从主程序 ``pyproject.toml`` 读取。
             sdk_version: 当前 SDK 版本号；留空时自动从运行环境中探测。
             project_root: 项目根目录；留空时自动推断。
+            validate_python_package_dependencies: 是否校验 Python 包依赖与当前环境的关系。
         """
         self._project_root: Path = project_root or self._resolve_project_root()
         self._host_version: str = host_version or self._detect_default_host_version(self._project_root)
         self._sdk_version: str = sdk_version or self._detect_default_sdk_version(self._project_root)
+        self._validate_python_package_dependencies: bool = validate_python_package_dependencies
         self.errors: List[str] = []
         self.warnings: List[str] = []
 
@@ -823,9 +826,10 @@ class ManifestValidator:
         if not sdk_ok:
             self.errors.append(f"SDK 版本不兼容: {sdk_message} (当前 SDK: {self._sdk_version})")
 
-        self._validate_python_package_dependencies(manifest)
+        if self._validate_python_package_dependencies:
+            self._validate_python_package_dependencies_against_runtime(manifest)
 
-    def _validate_python_package_dependencies(self, manifest: PluginManifest) -> None:
+    def _validate_python_package_dependencies_against_runtime(self, manifest: PluginManifest) -> None:
         """校验 Python 包依赖与主程序运行环境是否冲突。
 
         Args:
@@ -865,6 +869,68 @@ class ManifestValidator:
                     f"主程序依赖约束为 {host_specifier or '任意版本'}"
                 )
 
+    def load_host_dependency_requirements(self) -> Dict[str, Requirement]:
+        """读取主程序在 ``pyproject.toml`` 中声明的依赖约束。
+
+        Returns:
+            Dict[str, Requirement]: 以规范化包名为键的依赖约束映射。
+        """
+
+        return self._load_host_dependency_requirements(self._project_root)
+
+    def get_installed_package_version(self, package_name: str) -> Optional[str]:
+        """查询当前运行环境中指定包的安装版本。
+
+        Args:
+            package_name: 需要查询的包名。
+
+        Returns:
+            Optional[str]: 已安装版本号；未安装时返回 ``None``。
+        """
+
+        return self._get_installed_package_version(package_name)
+
+    @staticmethod
+    def build_specifier_set(version_spec: str) -> Optional[SpecifierSet]:
+        """将版本约束文本转换为 ``SpecifierSet``。
+
+        Args:
+            version_spec: 原始版本约束文本。
+
+        Returns:
+            Optional[SpecifierSet]: 转换成功时返回约束对象，否则返回 ``None``。
+        """
+
+        return ManifestValidator._build_specifier_set(version_spec)
+
+    @staticmethod
+    def version_matches_specifier(version: str, version_spec: str) -> bool:
+        """判断版本号是否满足给定约束。
+
+        Args:
+            version: 待判断的版本号。
+            version_spec: 版本约束表达式。
+
+        Returns:
+            bool: 是否满足约束。
+        """
+
+        return ManifestValidator._version_matches_specifier(version, version_spec)
+
+    @classmethod
+    def requirements_may_overlap(cls, left: SpecifierSet, right: SpecifierSet) -> bool:
+        """判断两个版本约束是否可能存在交集。
+
+        Args:
+            left: 左侧版本约束。
+            right: 右侧版本约束。
+
+        Returns:
+            bool: 若两者可能同时满足则返回 ``True``。
+        """
+
+        return cls._requirements_may_overlap(left, right)
+
     def _log_errors(self) -> None:
         """输出当前累计的 Manifest 校验错误。"""
         for error_message in self.errors:

src/plugin_runtime/runner/plugin_loader.py

@@ -75,6 +75,35 @@ class PluginLoader:
         self._failed_plugins: Dict[str, str] = {}
         self._manifest_validator = ManifestValidator(host_version=host_version)
         self._compat_hook_installed = False
+        self._blocked_plugin_reasons: Dict[str, str] = {}
+
+    def set_blocked_plugin_reasons(self, blocked_plugin_reasons: Optional[Dict[str, str]] = None) -> None:
+        """更新当前加载器持有的拒绝加载插件列表。
+
+        Args:
+            blocked_plugin_reasons: 需要拒绝加载的插件及原因映射。
+        """
+
+        self._blocked_plugin_reasons = {
+            str(plugin_id or "").strip(): str(reason or "").strip()
+            for plugin_id, reason in (blocked_plugin_reasons or {}).items()
+            if str(plugin_id or "").strip() and str(reason or "").strip()
+        }
+
+    def get_blocked_plugin_reason(self, plugin_id: str) -> Optional[str]:
+        """返回指定插件当前的拒绝加载原因。
+
+        Args:
+            plugin_id: 目标插件 ID。
+
+        Returns:
+            Optional[str]: 若插件被阻止加载则返回原因，否则返回 ``None``。
+        """
+
+        normalized_plugin_id = str(plugin_id or "").strip()
+        if not normalized_plugin_id:
+            return None
+        return self._blocked_plugin_reasons.get(normalized_plugin_id)
 
     def discover_and_load(
         self,
@@ -156,6 +185,11 @@ class PluginLoader:
             return None
 
         plugin_id = manifest.id
+        if blocked_reason := self.get_blocked_plugin_reason(plugin_id):
+            self._failed_plugins[plugin_id] = blocked_reason
+            logger.warning(f"插件 {plugin_id} 已被 Host 依赖流水线阻止加载: {blocked_reason}")
+            return None
+
         return plugin_id, (plugin_dir, manifest, plugin_path)
 
     def _record_duplicate_candidates(self, duplicate_candidates: Dict[str, List[Path]]) -> None:

src/plugin_runtime/runner/runner_main.py

@@ -29,6 +29,7 @@ import tomlkit
 
 from src.common.logger import get_console_handler, get_logger, initialize_logging
 from src.plugin_runtime import (
+    ENV_BLOCKED_PLUGIN_REASONS,
     ENV_EXTERNAL_PLUGIN_IDS,
     ENV_HOST_VERSION,
     ENV_IPC_ADDRESS,
@@ -196,6 +197,7 @@ class PluginRunner:
         session_token: str,
         plugin_dirs: List[str],
         external_available_plugins: Optional[Dict[str, str]] = None,
+        blocked_plugin_reasons: Optional[Dict[str, str]] = None,
     ) -> None:
         """初始化 Runner。
 
@@ -204,6 +206,7 @@ class PluginRunner:
             session_token: 握手用会话令牌。
             plugin_dirs: 当前 Runner 负责扫描的插件目录列表。
             external_available_plugins: 视为已满足的外部依赖插件版本映射。
+            blocked_plugin_reasons: 需要拒绝加载的插件及原因映射。
         """
         self._host_address: str = host_address
         self._session_token: str = session_token
@@ -213,9 +216,15 @@ class PluginRunner:
             for plugin_id, plugin_version in (external_available_plugins or {}).items()
             if str(plugin_id or "").strip() and str(plugin_version or "").strip()
         }
+        self._blocked_plugin_reasons: Dict[str, str] = {
+            str(plugin_id or "").strip(): str(reason or "").strip()
+            for plugin_id, reason in (blocked_plugin_reasons or {}).items()
+            if str(plugin_id or "").strip() and str(reason or "").strip()
+        }
 
         self._rpc_client: RPCClient = RPCClient(host_address, session_token)
         self._loader: PluginLoader = PluginLoader(host_version=os.getenv(ENV_HOST_VERSION, ""))
+        self._loader.set_blocked_plugin_reasons(self._blocked_plugin_reasons)
         self._start_time: float = time.monotonic()
         self._shutting_down: bool = False
         self._reload_lock: asyncio.Lock = asyncio.Lock()
@@ -1639,6 +1648,7 @@ class PluginRunner:
 
 async def _async_main() -> None:
     """异步主入口"""
+    blocked_plugin_reasons_raw = os.environ.get(ENV_BLOCKED_PLUGIN_REASONS, "")
     host_address = os.environ.pop(ENV_IPC_ADDRESS, "")
     external_plugin_ids_raw = os.environ.get(ENV_EXTERNAL_PLUGIN_IDS, "")
     session_token = os.environ.pop(ENV_SESSION_TOKEN, "")
@@ -1658,13 +1668,30 @@ async def _async_main() -> None:
         logger.warning("外部依赖插件版本映射格式非法，已回退为空映射")
         external_plugin_ids = {}
 
+    try:
+        blocked_plugin_reasons = json.loads(blocked_plugin_reasons_raw) if blocked_plugin_reasons_raw else {}
+    except json.JSONDecodeError:
+        logger.warning("解析阻止加载插件原因映射失败，已回退为空映射")
+        blocked_plugin_reasons = {}
+    if not isinstance(blocked_plugin_reasons, dict):
+        logger.warning("阻止加载插件原因映射格式非法，已回退为空映射")
+        blocked_plugin_reasons = {}
+
+    runner_kwargs: Dict[str, Any] = {
+        "external_available_plugins": {
+            str(plugin_id): str(plugin_version) for plugin_id, plugin_version in external_plugin_ids.items()
+        }
+    }
+    if blocked_plugin_reasons:
+        runner_kwargs["blocked_plugin_reasons"] = {
+            str(plugin_id): str(reason) for plugin_id, reason in blocked_plugin_reasons.items()
+        }
+
     runner = PluginRunner(
         host_address,
         session_token,
         plugin_dirs,
-        external_available_plugins={
-            str(plugin_id): str(plugin_version) for plugin_id, plugin_version in external_plugin_ids.items()
-        },
+        **runner_kwargs,
     )
 
     # 注册信号处理