From b663825fe704408f77747c41693eb902e386c440 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 10 Oct 2026 12:19:39 +0800 Subject: [PATCH] fix(plugin): recover dependencies without disabled constraints Keep enabled host and clone sources in dependency recovery. Fall back to per-plugin requests while retaining ready and recovered manifest constraints, preserving original sources and cancellation ownership. Refs #6943 --- app/adapters/system/plugin/dependency.py | 127 +++++++-- app/runtime/extensions/plugin/dependency.py | 8 +- app/startup/composition/plugin.py | 2 + app/startup/initializers/plugins.py | 6 +- docs/architecture-overview.md | 4 +- docs/development-setup.md | 7 + docs/refactor/optimization-checklist.md | 2 +- docs/rules/05-architecture.md | 2 +- .../architecture/concurrency-baseline.json | 6 + .../architecture/dependency-baseline.json | 5 +- tests/test_plugin_dependency_installer.py | 266 +++++++++++++++++- .../test_plugin_external_install_boundary.py | 28 +- .../test_plugin_instance_enabled_endpoint.py | 23 ++ 13 files changed, 451 insertions(+), 35 deletions(-) diff --git a/app/adapters/system/plugin/dependency.py b/app/adapters/system/plugin/dependency.py index 593d0bda0..01b7fc89a 100644 --- a/app/adapters/system/plugin/dependency.py +++ b/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) diff --git a/app/runtime/extensions/plugin/dependency.py b/app/runtime/extensions/plugin/dependency.py index a600ab95d..709800f1a 100644 --- a/app/runtime/extensions/plugin/dependency.py +++ b/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) diff --git a/app/startup/composition/plugin.py b/app/startup/composition/plugin.py index a34f4d11a..b18da6ad6 100644 --- a/app/startup/composition/plugin.py +++ b/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, diff --git a/app/startup/initializers/plugins.py b/app/startup/initializers/plugins.py index af95a596c..b4b71debb 100644 --- a/app/startup/initializers/plugins.py +++ b/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 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 1fb16c68f..8ba151f78 100644 --- a/docs/architecture-overview.md +++ b/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` 再执行快照测试及一次 diff --git a/docs/development-setup.md b/docs/development-setup.md index 38f47c10f..c72602bdc 100644 --- a/docs/development-setup.md +++ b/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` 与旧插件 diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index a8356bd08..a29b7b500 100644 --- a/docs/refactor/optimization-checklist.md +++ b/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 等基线均为零 | diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 1905e74d1..c043d00ef 100644 --- a/docs/rules/05-architecture.md +++ b/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 | diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index 4aa8adf31..3be86493a 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/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 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index aaac0ca4c..7def2944a 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/test_plugin_dependency_installer.py b/tests/test_plugin_dependency_installer.py index 1c9b2b76b..be9e54434 100644 --- a/tests/test_plugin_dependency_installer.py +++ b/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() diff --git a/tests/test_plugin_external_install_boundary.py b/tests/test_plugin_external_install_boundary.py index 17bb9b714..a696008f4 100644 --- a/tests/test_plugin_external_install_boundary.py +++ b/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"]) diff --git a/tests/test_plugin_instance_enabled_endpoint.py b/tests/test_plugin_instance_enabled_endpoint.py index 53267c361..d7970b24a 100644 --- a/tests/test_plugin_instance_enabled_endpoint.py +++ b/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",)