Merge remote-tracking branch 'origin/v3' into codex/pr-6941-agent-media-type

app/adapters/system/plugin/dependency.py

@@ -19,8 +19,10 @@ from packaging.version import InvalidVersion, Version
 from app.adapters.system.plugin.health import PluginRuntimeHealth
 from app.adapters.system.plugin.manifest import (
     PluginDependencyManifestError,
+    load_dependency_file,
     load_dependency_manifest,
 )
+from app.runtime.execution import run_in_threadpool_to_completion
 from app.runtime.log import logger
 from app.runtime.settings import get_runtime_setting
 
@@ -69,13 +71,15 @@ class PluginDependencyInstaller:
         packages: Optional[PluginDependencyPackagePort] = None,
         *,
         installed_plugins_provider: Optional[Callable[[], list[str]]] = None,
+        enabled_plugins_provider: Optional[Callable[[], set[str]]] = None,
         plugin_dir: Optional[Path] = None,
     ) -> None:
-        """保存包安装端口和启动层提供的已安装插件读取器。"""
+        """保存安装清单和启用源码读取器；旧调用方未提供启用位时保留安装集合。"""
         if packages is None:
             packages = PluginRuntimeHealth()
         self._packages = packages
         self._installed_plugins_provider = installed_plugins_provider or (lambda: [])
+        self._enabled_plugins_provider = enabled_plugins_provider
         self._plugin_dir = plugin_dir or (
             Path(get_runtime_setting('ROOT_PATH')) / "app" / "plugins"
         )
@@ -273,13 +277,22 @@ class PluginDependencyInstaller:
             merged.append(Requirement(target))
         return merged
 
-    def _plugin_manifests(self) -> list[Any]:
-        """返回已安装插件当前生效的依赖清单。"""
-        manifests = []
-        installed_plugins = {
+    def _dependency_plugin_ids(self) -> set[str]:
+        """只为已安装且至少有一个启用实例的物理插件恢复依赖。"""
+        installed = {
             plugin_id.lower()
             for plugin_id in self._installed_plugins_provider() or []
         }
+        if self._enabled_plugins_provider is None:
+            return installed
+        return installed & {
+            plugin_id.lower() for plugin_id in self._enabled_plugins_provider()
+        }
+
+    def _plugin_manifests(self) -> list[Any]:
+        """返回需要装载的已安装插件的原始依赖清单。"""
+        manifests = []
+        installed_plugins = self._dependency_plugin_ids()
         try:
             plugin_dirs = list(self._plugin_dir.iterdir())
         except (FileNotFoundError, OSError):
@@ -297,7 +310,7 @@ class PluginDependencyInstaller:
         return manifests
 
     def _plugin_dependencies(self) -> list[Requirement]:
-        """扫描已安装插件的生效依赖清单并合并版本约束。"""
+        """扫描启用插件的生效依赖清单并合并版本约束。"""
         dependencies: list[Requirement] = []
         for manifest in self._plugin_manifests():
             for requirement in manifest.dependencies:
@@ -356,13 +369,9 @@ class PluginDependencyInstaller:
         return ready, missing_dependencies, missing_source
 
     def _wheels_dirs(self) -> list[Path]:
-        """收集已安装插件附带的本地 wheels 目录。"""
+        """收集启用插件附带的本地 wheels 目录。"""
         result = []
-        installed_plugins = {
-            plugin_id.lower()
-            for plugin_id in self._installed_plugins_provider() or []
-        }
-        for plugin_id in installed_plugins:
+        for plugin_id in sorted(self._dependency_plugin_ids()):
             wheels_dir = self._plugin_dir / plugin_id / "wheels"
             if wheels_dir.is_dir():
                 result.append(wheels_dir)
@@ -395,17 +404,77 @@ class PluginDependencyInstaller:
         logger.error(message)
         return False, message
 
+    def _fallback_plan(
+        self, request: _DependencyInstallRequest
+    ) -> tuple[list[Path], list[Path]]:
+        """批量失败后保留已满足清单的约束，逐个恢复其余插件。
+
+        成功恢复的清单也会加入后续请求；互斥版本因而不能通过轮流覆盖环境假装成功。
+        始终传递原始清单，保留 uv sources、索引、direct URL 和 extras 的语义。
+        """
+        installed = self._installed_packages()
+        accepted: list[Path] = []
+        pending: list[Path] = []
+        for path in request.manifest_paths:
+            manifest = load_dependency_file(path)
+            satisfied = all(
+                self._requirement_satisfied(requirement, installed)
+                for requirement in manifest.dependencies
+                if not requirement.marker or requirement.marker.evaluate()
+            )
+            (accepted if satisfied else pending).append(path)
+        return accepted, pending
+
+    @staticmethod
+    def _record_group_result(
+        path: Path,
+        result: tuple[bool, str],
+        accepted: list[Path],
+        failures: list[str],
+    ) -> None:
+        """让成功清单保护后续安装，并记录失败插件而不中断其它组。"""
+        success, message = result
+        if success:
+            accepted.append(path)
+            return
+        failure = f"插件 {path.parent.name} 依赖恢复失败：{message}"
+        logger.warning(failure)
+        failures.append(failure)
+
+    @staticmethod
+    def _fallback_result(failures: list[str]) -> tuple[bool, str]:
+        """部分恢复仍返回失败，使启动继续按实际环境分类和激活就绪插件。"""
+        if failures:
+            return False, "\n".join(failures)
+        return True, "插件依赖已分组恢复"
+
     def install(self, dependencies: list[str]) -> tuple[bool, str]:
-        """把已安装插件的原始清单交给一次统一包安装。"""
+        """先统一安装；失败后逐组恢复，并保护已就绪插件的约束。"""
         request, error_result = self._prepare_install_request(dependencies)
         if error_result is not None:
             return error_result
         assert request is not None
         try:
-            return self._packages.install_packages_with_fallback(
+            result = self._packages.install_packages_with_fallback(
                 request.manifest_paths,
                 request.wheels_dirs,
             )
+            if result[0] or len(request.manifest_paths) < 2:
+                return result
+            logger.warning("插件依赖批量安装失败，开始逐组恢复并保留已就绪插件约束")
+            accepted, pending = self._fallback_plan(request)
+            if not pending:
+                return result
+            failures: list[str] = []
+            for path in pending:
+                try:
+                    group_result = self._packages.install_packages_with_fallback(
+                        [*accepted, path], request.wheels_dirs
+                    )
+                except Exception as error:  # noqa: BLE001 - 单组失败不得中断其余组
+                    group_result = self._dependency_install_failure(error)
+                self._record_group_result(path, group_result, accepted, failures)
+            return self._fallback_result(failures)
         except Exception as error:  # noqa: BLE001 - 统一映射为公开安装结果
             return self._dependency_install_failure(error)
 
@@ -414,15 +483,39 @@ class PluginDependencyInstaller:
         return await asyncio.to_thread(self.find_missing)
 
     async def async_install(self, dependencies: list[str]) -> tuple[bool, str]:
-        """异步安装依赖，使用可取消的包安装子进程。"""
-        request, error_result = self._prepare_install_request(dependencies)
+        """使用可取消子进程执行统一安装和受约束的逐组恢复。"""
+        prepared: tuple[
+            Optional[_DependencyInstallRequest], Optional[tuple[bool, str]]
+        ] = await run_in_threadpool_to_completion(
+            self._prepare_install_request, dependencies
+        )
+        request, error_result = prepared
         if error_result is not None:
             return error_result
         assert request is not None
         try:
-            return await self._packages.async_install_packages_with_fallback(
+            result = await self._packages.async_install_packages_with_fallback(
                 request.manifest_paths,
                 request.wheels_dirs,
             )
+            if result[0] or len(request.manifest_paths) < 2:
+                return result
+            logger.warning("插件依赖批量安装失败，开始逐组恢复并保留已就绪插件约束")
+            plan: tuple[list[Path], list[Path]] = await run_in_threadpool_to_completion(
+                self._fallback_plan, request
+            )
+            accepted, pending = plan
+            if not pending:
+                return result
+            failures: list[str] = []
+            for path in pending:
+                try:
+                    group_result = await self._packages.async_install_packages_with_fallback(
+                        [*accepted, path], request.wheels_dirs
+                    )
+                except Exception as error:  # noqa: BLE001 - CancelledError 必须向上传播
+                    group_result = self._dependency_install_failure(error)
+                self._record_group_result(path, group_result, accepted, failures)
+            return self._fallback_result(failures)
         except Exception as error:  # noqa: BLE001 - 统一映射为公开安装结果
             return self._dependency_install_failure(error)

app/runtime/extensions/plugin/dependency.py

@@ -102,11 +102,15 @@ class PluginDependencyService:
         物理插件那一层按安装清单划分，而安装清单只回答「包在不在磁盘上」；分类结果
         随后会被逐个 ``start()``，因此这里必须先按启用位过滤掉停用的本体，否则停用
         的插件会在开机与配置热重载时被重新拉起来，启用位形同虚设。三个桶一起过滤：
-        停用的插件既不该被装载，也不该为它安装缺失依赖。
+        停用的本体不应装载；启用分身仍按过滤前的共享源码结论分类。依赖安装只覆盖
+        存在启用本体或分身的源码插件，由启动层向安装器注入该集合。
         """
         ready, missing_dependencies, missing_source = (
             self._system().dependency.classify_plugins()
         )
+        # 分身只共享源码和依赖，本体的停用位不能让启用分身丢失源码结论。
+        source_ready = set(ready)
+        source_pending = set(missing_dependencies)
         loadable = self._loadable_hosts() if self._loadable_hosts is not None else None
         if loadable is not None:
             ready = [plugin_id for plugin_id in ready if plugin_id in loadable]
@@ -119,8 +123,6 @@ class PluginDependencyService:
         ready = list(ready)
         missing_dependencies = list(missing_dependencies)
         missing_source = list(missing_source)
-        source_ready = set(ready)
-        source_pending = set(missing_dependencies)
         for instance in self._instances().values():
             if instance.source_plugin_id in source_ready:
                 ready.append(instance.instance_id)

app/startup/composition/plugin.py

@@ -34,6 +34,7 @@ _market_client: Optional[PluginMarketClient] = None
 def compose_plugin_market(
     *,
     installed_plugins_provider: Callable[[], list[str]],
+    enabled_plugins_provider: Optional[Callable[[], set[str]]] = None,
 ) -> PluginMarketComposition:
     """一次性构造插件市场技术依赖，供同一 lifespan 内所有用例复用。"""
     global _market_client
@@ -52,6 +53,7 @@ def compose_plugin_market(
         dependency=PluginDependencyInstaller(
             health,
             installed_plugins_provider=installed_plugins_provider,
+            enabled_plugins_provider=enabled_plugins_provider,
             plugin_dir=plugin_root,
         ),
         health=health,

app/startup/initializers/plugins.py

@@ -492,7 +492,11 @@ def configure_plugin_services() -> None:
         installed_plugins_provider=lambda: get_configured_system_config().get(
             SystemConfigKey.UserInstalledPlugins
         )
-        or []
+        or [],
+        enabled_plugins_provider=lambda: {
+            instance.source_plugin_id
+            for instance in get_plugin_instance_directory().list_enabled()
+        },
     )
     market_transport = market_composition.transport
     market_client = market_composition.client

docs/architecture-overview.md

@@ -763,7 +763,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1114 |
-| 内部导入边 | 9,422 |
+| 内部导入边 | 9,423 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 49（债务已清零，49 条精确 containment） |
@@ -773,7 +773,7 @@ flowchart LR
 | Model/Oper 自动事务与自建 Session | 0 |
 | 组合根外 `SystemConfigOper()` | 0 |
 
-整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,422；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 582、`app.factory` 为 595、`app.main` 为 597。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
+整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,423；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 582、`app.factory` 为 595、`app.main` 为 597。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
 
 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次

docs/development-setup.md

@@ -151,6 +151,13 @@ dependencies = ["example-package>=1,<2"]
 - 仅有 `requirements.txt` 的历史插件继续按原方式安装；
 - 宿主不消费插件自己的 `uv.lock`，因为多个插件共享同一主程序环境，不能分别同步独立锁文件。
 
+启动恢复依赖时，只收集已安装且存在启用本体或分身的插件清单；全部实例都停用的插件
+不会参与依赖解析，也不会提供本地 wheels。先尝试统一安装，失败后逐个恢复插件，并在
+后续请求中保留已就绪和已恢复插件的原始清单约束，避免互斥版本轮流覆盖共享环境。
+仍不可满足的插件保持 `dependency_pending`，其它就绪插件继续激活。两个互斥版本无法
+在同一环境中同时满足，应停用所有冲突实例、卸载冲突插件或由插件作者统一依赖版本。
+安装结束后宿主重新分类并激活就绪插件；容器外手工修改依赖后，需重载插件或重启以更新运行状态。
+
 ### 3.2 异步 HTTP 客户端边界
 
 主程序自建的 `AsyncRequestUtils` 使用 HTTPX2，`app.sdk.network.AsyncRequestUtils` 与旧插件

docs/refactor/optimization-checklist.md

@@ -98,7 +98,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 1114 / 9,422 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 |
+| 宿主 Python 模块 / 内部依赖边 | 1114 / 9,423 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

docs/rules/05-architecture.md

@@ -1234,7 +1234,7 @@ driven workflow registration.
 | `app/domain/plugin.py` | Pure local-source, generation-compatibility and Release-install-plan rules |
 | `app/adapters/external/plugin/client.py` | `PluginMarketTransport`, the unique market transport, index/Release and local-candidate owner |
 | `app/adapters/system/plugin/package.py` | `PluginPackageManager`, the unique install, checkpoint, backup, restore and physical-delete owner |
-| `app/adapters/system/plugin/dependency.py` | `PluginDependencyInstaller`, the dependency manifest, missing-package inspection and installation owner |
+| `app/adapters/system/plugin/dependency.py` | `PluginDependencyInstaller`, enabled-source dependency inspection and constrained per-plugin recovery owner; `async_install` has two completion-owned worker boundaries for request preparation and fallback planning, while package subprocess cancellation remains with `PluginRuntimeHealth` |
 | `app/adapters/system/plugin/health.py` | `PluginRuntimeHealth`, the runtime-environment guard, pre/post-install check and recovery owner |
 | `app/startup/composition/plugin.py` | Constructs one market transport/client, package manager, runtime-health owner and dependency installer per lifespan; `PluginHelper` compatibility calls consume those same published owners |
 | `app/runtime/extensions/resource.py` | Data-only managed-resource registry and sync/async lifecycle adapters |

tests/fixtures/architecture/concurrency-baseline.json

@@ -36,6 +36,12 @@
     },
     "target": "asyncio.create_task"
   },
+  "app/adapters/system/plugin/dependency.py:app.runtime.execution.run_in_threadpool_to_completion": {
+    "owners": {
+      "PluginDependencyInstaller.async_install": 2
+    },
+    "target": "app.runtime.execution.run_in_threadpool_to_completion"
+  },
   "app/adapters/system/plugin/dependency.py:asyncio.to_thread": {
     "owners": {
       "PluginDependencyInstaller.async_find_missing": 1

tests/fixtures/architecture/dependency-baseline.json

@@ -941,8 +941,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 9422,
-  "edge_sha256": "db087968510ab776ed59d5c8a06b673795620b5d15e00752731f1c96af59947d",
+  "edge_count": 9423,
+  "edge_sha256": "d7c280e5230a61a296bfa27ce8d98bb51f7ae7418fb792dfac2c0d4b53ad3ce6",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -1103,6 +1103,7 @@
     "app.adapters.system.plugin.dependency -> app.adapters.system.plugin.health",
     "app.adapters.system.plugin.dependency -> app.adapters.system.plugin.manifest",
     "app.adapters.system.plugin.dependency -> app.runtime",
+    "app.adapters.system.plugin.dependency -> app.runtime.execution",
     "app.adapters.system.plugin.dependency -> app.runtime.log",
     "app.adapters.system.plugin.dependency -> app.runtime.settings",
     "app.adapters.system.plugin.health -> app.adapters",

tests/test_plugin_dependency_installer.py

@@ -1,6 +1,8 @@
+import asyncio
+import threading
 from pathlib import Path
 from types import SimpleNamespace
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 from packaging.requirements import Requirement
@@ -485,3 +487,265 @@ demo = { index = "private" }
     ]
     assert "[tool.uv.sources]" in manifest_paths[0].read_text(encoding="utf-8")
     assert "--extra-index-url" in manifest_paths[1].read_text(encoding="utf-8")
+
+
+def test_disabled_plugins_do_not_contribute_dependencies_or_wheels(
+    tmp_path, monkeypatch
+):
+    """停用插件的版本钉子、无效清单和 wheels 都不得进入恢复输入。"""
+    _write_requirements(tmp_path, "Enabled", "demo==2\n")
+    _write_requirements(tmp_path, "Disabled", "demo==1\n")
+    _write_pyproject(tmp_path, "Invalid", "[project\n")
+    enabled_wheels = tmp_path / "enabled" / "wheels"
+    enabled_wheels.mkdir()
+    (tmp_path / "disabled" / "wheels").mkdir()
+    enabled = {"Enabled"}
+    packages = Mock()
+    packages.install_packages_with_fallback.return_value = (True, "installed")
+    installer = PluginDependencyInstaller(
+        packages,
+        installed_plugins_provider=lambda: ["Enabled", "Disabled", "Invalid"],
+        enabled_plugins_provider=lambda: enabled,
+        plugin_dir=tmp_path,
+    )
+    monkeypatch.setattr(installer, "_installed_packages", lambda: {})
+
+    assert installer.find_missing() == ["demo==2"]
+    assert installer.install(installer.find_missing()) == (True, "installed")
+    packages.install_packages_with_fallback.assert_called_once_with(
+        [tmp_path / "enabled" / "requirements.txt"], [enabled_wheels]
+    )
+
+    enabled.clear()
+    assert installer.find_missing() == []
+    assert installer._wheels_dirs() == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("already_ready", [False, True])
+async def test_conflicting_plugin_does_not_block_others_or_replace_ready_pin(
+    tmp_path, monkeypatch, use_async, already_ready
+):
+    """冲突降级时保护已就绪和刚恢复的插件，并继续恢复无关插件。"""
+    _write_requirements(tmp_path, "Alpha", "demo==1\n")
+    _write_requirements(tmp_path, "Beta", "demo==2\n")
+    _write_requirements(tmp_path, "Other", "unrelated>=1\n")
+    installed = {"demo": Version("2")} if already_ready else {}
+    calls = []
+
+    def install(paths, _wheels):
+        """模拟解析器拒绝互斥约束，不访问真实 Python 环境或网络。"""
+        calls.append([path.parent.name for path in paths])
+        requirements = [
+            requirement
+            for path in paths
+            for requirement in load_dependency_file(path).dependencies
+        ]
+        pins = {str(item.specifier) for item in requirements if item.name == "demo"}
+        if len(pins) > 1:
+            return False, "No solution found when resolving dependencies"
+        for item in requirements:
+            installed[item.name] = Version("2" if str(item.specifier) == "==2" else "1")
+        return True, "installed"
+
+    packages = SimpleNamespace(
+        install_packages_with_fallback=Mock(side_effect=install),
+        async_install_packages_with_fallback=AsyncMock(side_effect=install),
+    )
+    installer = PluginDependencyInstaller(
+        packages,
+        installed_plugins_provider=lambda: ["Alpha", "Beta", "Other"],
+        plugin_dir=tmp_path,
+    )
+    monkeypatch.setattr(installer, "_installed_packages", lambda: dict(installed))
+    monkeypatch.setattr(installer, "_installed_distribution", lambda _name: None)
+    missing = installer.find_missing()
+
+    result = (
+        await installer.async_install(missing)
+        if use_async else installer.install(missing)
+    )
+
+    assert result[0] is False
+    assert ("alpha" if already_ready else "beta") in result[1]
+    assert installed["demo"] == Version("2" if already_ready else "1")
+    assert installed["unrelated"] == Version("1")
+    assert calls == (
+        [["alpha", "beta", "other"], ["beta", "alpha"], ["beta", "other"]]
+        if already_ready else
+        [
+            ["alpha", "beta", "other"], ["alpha"],
+            ["alpha", "beta"], ["alpha", "other"],
+        ]
+    )
+    ready, pending, _source_missing = installer.classify_plugins()
+    assert ready == (["Beta", "Other"] if already_ready else ["Alpha", "Other"])
+    assert pending == (["Alpha"] if already_ready else ["Beta"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_group_failure_or_exception_keeps_other_manifests_recoverable(
+    tmp_path, monkeypatch, use_async
+):
+    """单组异常也不阻断后续组，原始 modern/legacy 来源声明保持完整。"""
+    _write_requirements(tmp_path, "Broken", "broken>=1\n")
+    _write_requirements(
+        tmp_path, "Legacy", "--extra-index-url https://example.com\nlegacy\n"
+    )
+    modern = _write_pyproject(
+        tmp_path, "Modern",
+        '[project]\nname="modern"\nversion="1"\ndependencies=["modern"]\n'
+        '[tool.uv.sources]\nmodern={url="https://example.com/modern.whl"}\n',
+    )
+    calls = []
+
+    def install(paths, _wheels):
+        """拒绝包含坏插件的请求，记录后续原始清单输入。"""
+        calls.append(paths)
+        if len(calls) == 1:
+            return False, "batch failed"
+        if paths[0].parent.name == "broken":
+            raise OSError("group failed")
+        return True, "installed"
+
+    packages = SimpleNamespace(
+        install_packages_with_fallback=Mock(side_effect=install),
+        async_install_packages_with_fallback=AsyncMock(side_effect=install),
+    )
+    installer = PluginDependencyInstaller(
+        packages,
+        installed_plugins_provider=lambda: ["Broken", "Legacy", "Modern"],
+        plugin_dir=tmp_path,
+    )
+    monkeypatch.setattr(installer, "_installed_packages", lambda: {})
+
+    dependencies = ["broken", "legacy", "modern"]
+    result = (
+        await installer.async_install(dependencies)
+        if use_async else installer.install(dependencies)
+    )
+
+    assert result[0] is False
+    assert "broken" in result[1]
+    assert "group failed" in result[1]
+    assert calls[-1] == [
+        tmp_path / "legacy" / "requirements.txt", modern / "pyproject.toml"
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_fallback_reports_success_when_every_group_recovers(
+    tmp_path, monkeypatch, use_async
+):
+    """批量失败但所有分组均恢复后，同步和异步结果都报告成功。"""
+    for plugin_id in ("Alpha", "Beta"):
+        _write_requirements(tmp_path, plugin_id, f"{plugin_id.lower()}>=1\n")
+    results = [(False, "batch failed"), (True, "alpha"), (True, "beta")]
+    packages = SimpleNamespace(
+        install_packages_with_fallback=Mock(side_effect=results),
+        async_install_packages_with_fallback=AsyncMock(side_effect=results),
+    )
+    installer = PluginDependencyInstaller(
+        packages,
+        installed_plugins_provider=lambda: ["Alpha", "Beta"],
+        plugin_dir=tmp_path,
+    )
+    monkeypatch.setattr(installer, "_installed_packages", lambda: {})
+    result = (
+        await installer.async_install(["alpha", "beta"])
+        if use_async else installer.install(["alpha", "beta"])
+    )
+    assert result == (True, "插件依赖已分组恢复")
+
+
+@pytest.mark.asyncio
+async def test_async_fallback_cancellation_stops_remaining_groups(tmp_path, monkeypatch):
+    """取消分组安装必须传播到启动调用方，后续插件不得继续写环境。"""
+    for plugin_id in ("Alpha", "Beta"):
+        _write_requirements(tmp_path, plugin_id, f"{plugin_id.lower()}>=1\n")
+    packages = SimpleNamespace(
+        async_install_packages_with_fallback=AsyncMock(
+            side_effect=[(False, "batch failed"), asyncio.CancelledError()]
+        ),
+    )
+    installer = PluginDependencyInstaller(
+        packages,
+        installed_plugins_provider=lambda: ["Alpha", "Beta"],
+        plugin_dir=tmp_path,
+    )
+    monkeypatch.setattr(installer, "_installed_packages", lambda: {})
+
+    with pytest.raises(asyncio.CancelledError):
+        await installer.async_install(["alpha", "beta"])
+
+    assert packages.async_install_packages_with_fallback.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_async_install_inspection_runs_outside_event_loop(tmp_path, monkeypatch):
+    """请求准备和失败分组扫描都由托管 worker 执行，不阻塞宿主事件循环。"""
+    for plugin_id in ("Alpha", "Beta"):
+        _write_requirements(tmp_path, plugin_id, f"{plugin_id.lower()}>=1\n")
+    packages = SimpleNamespace(async_install_packages_with_fallback=AsyncMock(
+        side_effect=[(False, "batch failed"), (True, "alpha"), (True, "beta")]
+    ))
+    installer = PluginDependencyInstaller(
+        packages,
+        installed_plugins_provider=lambda: ["Alpha", "Beta"],
+        plugin_dir=tmp_path,
+    )
+    monkeypatch.setattr(installer, "_installed_packages", lambda: {})
+    inspections = []
+
+    def record(operation):
+        """保留真实扫描行为并记录执行线程。"""
+        def inspect(*args):
+            """记录当前检查操作的线程身份。"""
+            inspections.append((operation.__name__, threading.get_ident()))
+            return operation(*args)
+        return inspect
+
+    monkeypatch.setattr(installer, "_prepare_install_request", record(
+        installer._prepare_install_request
+    ))
+    monkeypatch.setattr(installer, "_fallback_plan", record(installer._fallback_plan))
+    assert (await installer.async_install(["alpha", "beta"]))[0] is True
+    assert [name for name, _thread in inspections] == [
+        "_prepare_install_request", "_fallback_plan"
+    ]
+    assert all(thread != threading.get_ident() for _name, thread in inspections)
+
+
+@pytest.mark.asyncio
+async def test_async_install_waits_for_inspection_before_propagating_cancel(
+    tmp_path, monkeypatch
+):
+    """取消请求等待在途扫描收敛，再向上抛出，且不启动任何包写入。"""
+    started = asyncio.Event()
+    release = threading.Event()
+    loop = asyncio.get_running_loop()
+    packages = SimpleNamespace(async_install_packages_with_fallback=AsyncMock())
+    installer = PluginDependencyInstaller(packages, plugin_dir=tmp_path)
+
+    def prepare(_dependencies):
+        """用受控只读扫描模拟取消发生时仍在途的 worker。"""
+        loop.call_soon_threadsafe(started.set)
+        assert release.wait(timeout=5)
+        return None, (False, "no manifest")
+
+    monkeypatch.setattr(installer, "_prepare_install_request", prepare)
+    task = asyncio.create_task(installer.async_install(["demo"]))
+    try:
+        await asyncio.wait_for(started.wait(), timeout=5)
+        task.cancel()
+        await asyncio.sleep(0)
+        assert not task.done()
+        packages.async_install_packages_with_fallback.assert_not_awaited()
+    finally:
+        release.set()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    packages.async_install_packages_with_fallback.assert_not_awaited()

tests/test_plugin_external_install_boundary.py

@@ -15,6 +15,7 @@ from app.api.endpoints import plugin as plugin_endpoint
 from app.application.plugin import management as plugin_management
 from app.runtime.config import global_vars
 from app.schemas.plugin import (
+    PluginInstance,
     PluginSourceChangeRequest,
     PluginSourceIdentity,
     PluginSourceInstallRequest,
@@ -714,6 +715,13 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) ->
     package_manager = Mock()
     health = Mock()
     dependency = Mock()
+    market_composition = Mock(return_value=SimpleNamespace(
+        transport=transport,
+        client=market_client,
+        package=package_manager,
+        health=health,
+        dependency=dependency,
+    ))
     gateway_calls = []
     application_calls = []
     gateway = Mock()
@@ -723,13 +731,7 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) ->
     monkeypatch.setattr(
         plugins_initializer,
         "compose_plugin_market",
-        lambda **_kwargs: SimpleNamespace(
-            transport=transport,
-            client=market_client,
-            package=package_manager,
-            health=health,
-            dependency=dependency,
-        ),
+        market_composition,
     )
     monkeypatch.setattr(
         plugins_initializer,
@@ -810,6 +812,18 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) ->
 
     plugins_initializer.configure_plugin_services()
 
+    monkeypatch.setattr(
+        plugins_initializer,
+        "get_plugin_instance_directory",
+        lambda: SimpleNamespace(list_enabled=lambda: [
+            PluginInstance(instance_id="Enabled", source_plugin_id="Enabled"),
+            PluginInstance(instance_id="Clone", source_plugin_id="DisabledHost"),
+        ]),
+    )
+    assert market_composition.call_args.kwargs["enabled_plugins_provider"]() == {
+        "Enabled", "DisabledHost"
+    }
+
     assert application_calls == [gateway]
     assert len(gateway_calls) == 1
     assert callable(gateway_calls[0]["install"])

tests/test_plugin_instance_enabled_endpoint.py

@@ -135,3 +135,26 @@ def test_classification_without_a_loadable_port_keeps_every_plugin():
     assert classification.ready == ("PluginA",)
     assert classification.missing_dependencies == ("PluginC",)
     assert classification.missing_source == ("PluginD",)
+
+
+def test_enabled_clone_keeps_source_classification_when_host_is_disabled():
+    """停用本体只阻止本体启动，启用分身仍按共享源码和实际依赖分类。"""
+    service = PluginDependencyService(
+        system=lambda: SimpleNamespace(dependency=SimpleNamespace(
+            classify_plugins=lambda: (["Ready"], ["Pending"], ["Missing"])
+        )),
+        instances=lambda: {
+            source + "Clone": PluginInstance(
+                instance_id=source + "Clone", source_plugin_id=source
+            )
+            for source in ("Ready", "Pending", "Missing")
+        },
+        loadable_hosts=set,
+        log=SimpleNamespace(),
+    )
+
+    classification = service.classify_plugins()
+
+    assert classification.ready == ("ReadyClone",)
+    assert classification.missing_dependencies == ("PendingClone",)
+    assert classification.missing_source == ("MissingClone",)