From 54ba4f435b46159b10ae92580b9ad54ee4a79dee Mon Sep 17 00:00:00 2001 From: DDSRem <73049927+DDSRem@users.noreply.github.com> Date: Sun, 18 Oct 2026 15:29:31 +0800 Subject: [PATCH] refactor(plugin): share dependency recovery decisions (#7060) --- app/adapters/system/plugin/health.py | 330 +++++++------- .../architecture/complexity-baseline.json | 6 +- tests/test_plugin_runtime_recovery.py | 410 ++++++++++++++++++ 3 files changed, 587 insertions(+), 159 deletions(-) create mode 100644 tests/test_plugin_runtime_recovery.py diff --git a/app/adapters/system/plugin/health.py b/app/adapters/system/plugin/health.py index fe24a0c55..3541bb0df 100644 --- a/app/adapters/system/plugin/health.py +++ b/app/adapters/system/plugin/health.py @@ -12,6 +12,8 @@ import sys import tempfile import threading from collections import deque +from dataclasses import dataclass +from enum import Enum from importlib.metadata import distributions from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Set, Tuple @@ -45,6 +47,36 @@ from app.runtime.log import logger from app.runtime.settings import get_runtime_setting +class _RecoveryState(Enum): + """ + 插件安装后相对原健康基线的四种恢复结果。 + """ + + # 没有新增异常，无需恢复。 + UNCHANGED = "unchanged" + # 修复命令成功，复检已回到安装前的健康基线。 + RESTORED = "restored" + # 修复命令失败，尚未复检。 + REPAIR_FAILED = "repair_failed" + # 修复命令成功，但复检仍有新增异常。 + STILL_BROKEN = "still_broken" + + +@dataclass(frozen=True) +class _RuntimeRecovery: + """ + 同步和异步 I/O 驱动共享的恢复事实，不持有进程、锁或临时文件。 + + :param state: 恢复结果，决定是否终止策略降级及刷新导入缓存 + :param regression: 安装引入的原始健康异常；未回退时为空 + :param detail: 修复命令失败输出或复检残留异常；无失败详情时为空 + """ + + state: _RecoveryState + regression: str = "" + detail: str = "" + + class PluginRuntimeHealth: """ 负责插件依赖写入的运行环境保护、检查和故障恢复。 @@ -482,32 +514,119 @@ class PluginRuntimeHealth: ) @classmethod - def __repair_if_runtime_broken( + def __repair_install_runtime( cls, - snapshot_file: Optional[Path] = None, - baseline_health: Optional[Dict[str, Tuple[bool, str]]] = None - ) -> Tuple[bool, str]: + health_message: str, + snapshot_file: Optional[Path], + baseline_health: Dict[str, Tuple[bool, str]], + ) -> _RuntimeRecovery: """ - 安装失败后检查主运行环境；若相对安装前新增异常，先恢复主程序依赖再返回。 + 同步执行恢复所需的 I/O，恢复结果由共享规则判定。 + + :param health_message: 本次安装新增异常；空串表示无需恢复 + :param snapshot_file: 安装前的依赖约束快照；None 时使用项目锁文件 + :param baseline_health: 安装前健康基线 + :return: 未发生回退、已恢复、修复失败或复检失败的事实 """ - current_health = cls.__run_runtime_healthcheck() - health_message = cls.__runtime_health_regression_message( - baseline_health or {}, - current_health - ) if not health_message: - return True, "" - repair_ok, repair_message = cls.__repair_main_runtime_dependencies(snapshot_file) - if not repair_ok: - return False, f"插件依赖安装失败后主运行环境异常，且恢复失败：{health_message}; {repair_message}" - restored_health = cls.__run_runtime_healthcheck() - restored_message = cls.__runtime_health_regression_message( - baseline_health or {}, - restored_health + return _RuntimeRecovery(_RecoveryState.UNCHANGED) + repair_result = cls.__repair_main_runtime_dependencies(snapshot_file) + restored_health = cls.__run_runtime_healthcheck() if repair_result[0] else {} + return cls.__runtime_recovery_outcome( + health_message, repair_result, baseline_health, restored_health, ) - if restored_message: - return False, f"插件依赖安装失败后主运行环境异常，恢复后仍异常：{restored_message}" - return True, "主运行环境已恢复" + + @classmethod + def __runtime_recovery_outcome( + cls, + health_message: str, + repair_result: Tuple[bool, str], + baseline_health: Dict[str, Tuple[bool, str]], + restored_health: Dict[str, Tuple[bool, str]], + ) -> _RuntimeRecovery: + """ + 统一判定修复命令和复检结果，不把命令成功当作环境已恢复。 + + :param health_message: 安装导致的原始健康回退 + :param repair_result: 修复命令的成功标志和输出 + :param baseline_health: 安装前健康基线 + :param restored_health: 修复成功后的复检快照；命令失败时为空 + :return: 带原始回退和失败详情的恢复事实 + """ + repair_ok, repair_message = repair_result + if not repair_ok: + return _RuntimeRecovery(_RecoveryState.REPAIR_FAILED, health_message, repair_message) + remaining = cls.__runtime_health_regression_message(baseline_health, restored_health) + if remaining: + return _RuntimeRecovery(_RecoveryState.STILL_BROKEN, health_message, remaining) + return _RuntimeRecovery(_RecoveryState.RESTORED, health_message) + + @classmethod + def __installation_regression( + cls, + success: bool, + baseline_health: Dict[str, Tuple[bool, str]], + current_health: Dict[str, Tuple[bool, str]], + ) -> str: + """ + 检测安装新增的环境异常，并保留安装成功后回退的原有错误日志。 + """ + message = cls.__runtime_health_regression_message(baseline_health, current_health) + if success and message: + logger.error(f"[UV] 依赖安装后运行环境自检失败：{message}") + return message + + @classmethod + def __installation_result( + cls, + success: bool, + message: str, + strategy_name: str, + recovery: _RuntimeRecovery, + ) -> Optional[Tuple[bool, str]]: + """ + 统一决定策略结果、导入刷新及是否允许尝试下一策略。 + + :param success: 安装子进程是否成功 + :param message: 安装子进程的原始输出 + :param strategy_name: 当前安装策略名称 + :param recovery: 安装后的环境恢复事实 + :return: 终态结果；仅安装失败且环境无新增异常时返回 None 继续降级 + """ + if success: + if recovery.state == _RecoveryState.UNCHANGED: + cls.__refresh_import_system() + return True, message + if recovery.state == _RecoveryState.RESTORED: + cls.__refresh_import_system() + return False, f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：{recovery.regression}" + if recovery.state == _RecoveryState.STILL_BROKEN: + return False, f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常：{recovery.detail}" + return False, f"依赖安装后运行环境自检失败，且自动恢复主程序依赖失败：{recovery.detail}" + + logger.error(f"[UV] 策略：{strategy_name} 安装依赖失败，错误信息：{message}") + if recovery.state == _RecoveryState.UNCHANGED: + return None + if recovery.state == _RecoveryState.RESTORED: + repair_message = "主运行环境已恢复" + elif recovery.state == _RecoveryState.STILL_BROKEN: + repair_message = f"插件依赖安装失败后主运行环境异常，恢复后仍异常：{recovery.detail}" + else: + repair_message = ( + f"插件依赖安装失败后主运行环境异常，且恢复失败：{recovery.regression}; {recovery.detail}" + ) + return False, cls.__with_free_threaded_reason( + f"策略 {strategy_name} 安装依赖失败：{message}；{repair_message}" + ) + + @classmethod + def __exhausted_install_result(cls, last_error: str) -> Tuple[bool, str]: + """ + 所有策略失败后保留最后一个错误及既有 free-threaded 诊断。 + """ + if last_error: + return False, cls.__with_free_threaded_reason(f"[UV] 所有策略均安装依赖失败：{last_error}") + return False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束" @classmethod def __run_runtime_healthcheck(cls) -> Dict[str, Tuple[bool, str]]: @@ -819,70 +938,31 @@ class PluginRuntimeHealth: ) if success: logger.debug(f"[UV] 策略：{strategy.strategy_name} 安装依赖成功，输出：{message}") - current_health = cls.__run_runtime_healthcheck() - health_message = cls.__runtime_health_regression_message( - baseline_health, - current_health - ) - if health_message: - logger.error(f"[UV] 依赖安装后运行环境自检失败：{health_message}") - repair_ok, repair_message = cls.__repair_main_runtime_dependencies( - constraints_file if protected_packages else None - ) - if repair_ok: - restored_health = cls.__run_runtime_healthcheck() - restored_message = cls.__runtime_health_regression_message( - baseline_health, - restored_health - ) - if not restored_message: - cls.__refresh_import_system() - return False, ( - f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：{health_message}" - ) - logger.error( - f"[UV] 主程序依赖恢复后仍未通过健康检查：{restored_message}" - ) - return False, ( - f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常：" - f"{restored_message}" - ) - return False, ( - f"依赖安装后运行环境自检失败，且自动恢复主程序依赖失败：" - f"{repair_message}" - ) - + current_health = cls.__run_runtime_healthcheck() + health_message = cls.__installation_regression(success, baseline_health, current_health) + recovery = cls.__repair_install_runtime(health_message, constraints_file, baseline_health) + if success and recovery.state == _RecoveryState.UNCHANGED: remaining_health_message = cls.__runtime_health_regression_message({}, current_health) if remaining_health_message: logger.warning( f"[UV] 依赖安装成功，安装前已有的运行环境异常仍然存在：" f"{remaining_health_message}" ) - - cls.__refresh_import_system() - loaded_modules_after_install = set(sys.modules.keys()) - loaded_modules_during_install = loaded_modules_after_install - loaded_modules_before_install - logger.debug(f"[UV] 已刷新导入系统，新加载的模块: {loaded_modules_during_install}") - return True, message - + elif success and recovery.state == _RecoveryState.STILL_BROKEN: + logger.error(f"[UV] 主程序依赖恢复后仍未通过健康检查：{recovery.detail}") + result = cls.__installation_result(success, message, strategy.strategy_name, recovery) + if result is not None: + if result[0]: + loaded_modules_after_install = set(sys.modules.keys()) + loaded_modules_during_install = loaded_modules_after_install - loaded_modules_before_install + logger.debug(f"[UV] 已刷新导入系统，新加载的模块: {loaded_modules_during_install}") + return result last_error = message - repair_ok, repair_message = cls.__repair_if_runtime_broken( - constraints_file if protected_packages else None, - baseline_health - ) - logger.error(f"[UV] 策略：{strategy.strategy_name} 安装依赖失败，错误信息：{message}") - if not repair_ok or repair_message: - return False, cls.__with_free_threaded_reason( - f"策略 {strategy.strategy_name} 安装依赖失败：{message}；" - f"{repair_message}" - ) finally: if constraints_file: constraints_file.unlink(missing_ok=True) - if last_error: - return False, cls.__with_free_threaded_reason(f"[UV] 所有策略均安装依赖失败：{last_error}") - return False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束" + return cls.__exhausted_install_result(last_error) @classmethod async def __async_run_runtime_healthcheck(cls) -> Dict[str, Tuple[bool, str]]: @@ -962,40 +1042,27 @@ class PluginRuntimeHealth: return False, last_error or f"恢复{repair_desc}失败" @classmethod - async def __async_repair_if_runtime_broken( + async def __async_repair_install_runtime( cls, + health_message: str, snapshot_file: Optional[Path], baseline_health: Dict[str, Tuple[bool, str]], - ) -> Tuple[bool, str]: + ) -> _RuntimeRecovery: """ - 异步检查并修复安装过程中新增的主程序环境异常。 + 通过可取消子进程执行修复与复检，使用同步入口相同的恢复判定。 + + :param health_message: 本次安装新增异常；空串表示无需恢复 + :param snapshot_file: 安装前的依赖约束快照；None 时使用项目锁文件 + :param baseline_health: 安装前健康基线 + :return: 未发生回退、已恢复、修复失败或复检失败的事实 """ - current_health = await cls.__async_run_runtime_healthcheck() - health_message = cls.__runtime_health_regression_message( - baseline_health, - current_health, - ) if not health_message: - return True, "" - repair_ok, repair_message = ( - await cls.__async_repair_main_runtime_dependencies(snapshot_file) + return _RuntimeRecovery(_RecoveryState.UNCHANGED) + repair_result = await cls.__async_repair_main_runtime_dependencies(snapshot_file) + restored_health = await cls.__async_run_runtime_healthcheck() if repair_result[0] else {} + return cls.__runtime_recovery_outcome( + health_message, repair_result, baseline_health, restored_health, ) - if not repair_ok: - return False, ( - f"插件依赖安装失败后主运行环境异常，且恢复失败：" - f"{health_message}; {repair_message}" - ) - restored_health = await cls.__async_run_runtime_healthcheck() - restored_message = cls.__runtime_health_regression_message( - baseline_health, - restored_health, - ) - if restored_message: - return False, ( - f"插件依赖安装失败后主运行环境异常，恢复后仍异常：" - f"{restored_message}" - ) - return True, "主运行环境已恢复" async def async_install_packages_with_fallback( self, @@ -1116,61 +1183,14 @@ class PluginRuntimeHealth: safe_command=strategy.safe_log_command, timeout=cls.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT, ) - if success: - current_health = await cls.__async_run_runtime_healthcheck() - health_message = cls.__runtime_health_regression_message( - baseline_health, - current_health, - ) - if health_message: - logger.error(f"[UV] 依赖安装后运行环境自检失败：{health_message}") - repair_ok, repair_message = ( - await cls.__async_repair_main_runtime_dependencies( - constraints_file if protected_packages else None - ) - ) - if repair_ok: - restored_health = await cls.__async_run_runtime_healthcheck() - restored_message = cls.__runtime_health_regression_message( - baseline_health, - restored_health, - ) - if not restored_message: - cls.__refresh_import_system() - return False, ( - f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：" - f"{health_message}" - ) - return False, ( - f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常：" - f"{restored_message}" - ) - return False, ( - f"依赖安装后运行环境自检失败，且自动恢复主程序依赖失败：" - f"{repair_message}" - ) - - cls.__refresh_import_system() - return True, message - + current_health = await cls.__async_run_runtime_healthcheck() + health_message = cls.__installation_regression(success, baseline_health, current_health) + recovery = await cls.__async_repair_install_runtime(health_message, constraints_file, baseline_health) + result = cls.__installation_result(success, message, strategy.strategy_name, recovery) + if result is not None: + return result last_error = message - repair_ok, repair_message = await cls.__async_repair_if_runtime_broken( - constraints_file if protected_packages else None, - baseline_health, - ) - logger.error( - f"[UV] 策略：{strategy.strategy_name} 安装依赖失败，错误信息：{message}" - ) - if not repair_ok or repair_message: - return False, cls.__with_free_threaded_reason( - f"策略 {strategy.strategy_name} 安装依赖失败：{message}；" - f"{repair_message}" - ) - return False, ( - cls.__with_free_threaded_reason(f"[UV] 所有策略均安装依赖失败：{last_error}") - if last_error - else "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束" - ) + return cls.__exhausted_install_result(last_error) finally: if acquired: cls._package_install_lock.release() diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index f59215a45..9e7675561 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -6,8 +6,8 @@ "app/adapters/external/plugin/client.py:PluginMarketTransport.get_local_plugin_candidate": 23, "app/adapters/network/http.py:AsyncRequestUtils.get_stream": 20, "app/adapters/network/http.py:RequestUtils.get_decoded_xml_content": 18, - "app/adapters/system/plugin/health.py:PluginRuntimeHealth.__async_install_packages_with_fallback": 24, - "app/adapters/system/plugin/health.py:PluginRuntimeHealth.install_packages_with_fallback": 27, + "app/adapters/system/plugin/health.py:PluginRuntimeHealth.__async_install_packages_with_fallback": 20, + "app/adapters/system/plugin/health.py:PluginRuntimeHealth.install_packages_with_fallback": 26, "app/agent/callback/__init__.py:StreamingHandler._classify_tool_call": 27, "app/agent/lifecycle.py:AgentLifecycleOwner.close": 21, "app/agent/llm/helper.py:_patch_gemini_thought_signature": 16, @@ -133,8 +133,6 @@ "app/adapters/network/http.py:get_caller": 6, "app/adapters/system/host.py:SystemUtils.exits_files._search_files": 6, "app/adapters/system/host.py:SystemUtils.list_files._scan_directory": 6, - "app/adapters/system/plugin/health.py:PluginRuntimeHealth.__async_install_packages_with_fallback": 6, - "app/adapters/system/plugin/health.py:PluginRuntimeHealth.install_packages_with_fallback": 7, "app/agent/callback/__init__.py:StreamingHandler._flush": 6, "app/agent/llm/helper.py:_patch_gemini_thought_signature": 6, "app/agent/policy/sanitizer.py:sanitize_for_host": 6, diff --git a/tests/test_plugin_runtime_recovery.py b/tests/test_plugin_runtime_recovery.py new file mode 100644 index 000000000..7c0a2aa51 --- /dev/null +++ b/tests/test_plugin_runtime_recovery.py @@ -0,0 +1,410 @@ +""" +插件依赖安装同步、异步入口的恢复决策和资源清理回归。 +""" + +import asyncio +import threading +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from app.adapters.system.package import PackageInstallRequest, PackageInstallStrategy +from app.adapters.system.plugin import health as health_module +from app.adapters.system.plugin.health import PluginRuntimeHealth + +_HEALTHY = {"核心依赖导入检查": (True, "ok")} +_BROKEN = {"核心依赖导入检查": (False, "new failure")} +_OLD_FAILURE = {"核心依赖导入检查": (False, "old failure")} +_REGRESSION = "核心依赖导入检查失败：new failure" + + +@dataclass(frozen=True) +class _RecoveryCase: + """ + 一次安装在外部边界的固定应答与预期结果。 + + :param installs: 各安装策略返回的成功标志和输出 + :param health: 安装前、安装后及修复后的有序健康快照 + :param repairs: 各修复策略返回的成功标志和输出 + :param result: 对外返回的完整成功标志和说明 + :param events: 必须保持的安装、探测、修复和导入刷新顺序 + """ + + installs: tuple[tuple[bool, str], ...] + health: tuple[dict[str, tuple[bool, str]], ...] + repairs: tuple[tuple[bool, str], ...] + result: tuple[bool, str] + events: tuple[str, ...] + + +_CASES = [ + pytest.param(_RecoveryCase( + ((True, "installed"),), (_HEALTHY, _HEALTHY), (), (True, "installed"), + ("health", "install", "health", "refresh"), + ), id="healthy"), + pytest.param(_RecoveryCase( + ((True, "installed"),), (_OLD_FAILURE, _OLD_FAILURE), (), (True, "installed"), + ("health", "install", "health", "refresh"), + ), id="existing-failure"), + pytest.param(_RecoveryCase( + ((False, "offline"), (True, "installed")), (_HEALTHY, _HEALTHY, _HEALTHY), (), + (True, "installed"), ("health", "install", "health", "install", "health", "refresh"), + ), id="fallback"), + pytest.param(_RecoveryCase( + ((False, "offline"), (False, "still offline")), (_HEALTHY, _HEALTHY, _HEALTHY), (), + (False, "[UV] 所有策略均安装依赖失败：still offline"), + ("health", "install", "health", "install", "health"), + ), id="exhausted"), + pytest.param(_RecoveryCase( + ((False, ""), (False, "")), (_HEALTHY, _HEALTHY, _HEALTHY), (), + (False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束"), + ("health", "install", "health", "install", "health"), + ), id="empty-errors"), + pytest.param(_RecoveryCase( + ((True, "installed"),), (_HEALTHY, _BROKEN, _HEALTHY), ((True, "repaired"),), + (False, f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：{_REGRESSION}"), + ("health", "install", "health", "repair", "refresh", "health", "refresh"), + ), id="successful-install-restored"), + pytest.param(_RecoveryCase( + ((True, "installed"),), (_HEALTHY, _BROKEN), ((False, "repair failed"),), + (False, "依赖安装后运行环境自检失败，且自动恢复主程序依赖失败：repair failed"), + ("health", "install", "health", "repair"), + ), id="successful-install-repair-failed"), + pytest.param(_RecoveryCase( + ((True, "installed"),), (_HEALTHY, _BROKEN, _BROKEN), ((True, "repaired"),), + (False, f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常：{_REGRESSION}"), + ("health", "install", "health", "repair", "refresh", "health"), + ), id="successful-install-still-broken"), + pytest.param(_RecoveryCase( + ((False, "install failed"),), (_HEALTHY, _BROKEN, _HEALTHY), ((True, "repaired"),), + (False, "策略 install-0 安装依赖失败：install failed；主运行环境已恢复"), + ("health", "install", "health", "repair", "refresh", "health"), + ), id="failed-install-restored-stops-fallback"), + pytest.param(_RecoveryCase( + ((False, "install failed"),), (_HEALTHY, _BROKEN), ((False, "repair failed"),), + (False, f"策略 install-0 安装依赖失败：install failed；插件依赖安装失败后主运行环境异常，且恢复失败：{_REGRESSION}; repair failed"), + ("health", "install", "health", "repair"), + ), id="failed-install-repair-failed"), + pytest.param(_RecoveryCase( + ((False, "install failed"),), (_HEALTHY, _BROKEN, _BROKEN), ((True, "repaired"),), + (False, f"策略 install-0 安装依赖失败：install failed；插件依赖安装失败后主运行环境异常，恢复后仍异常：{_REGRESSION}"), + ("health", "install", "health", "repair", "refresh", "health"), + ), id="failed-install-still-broken"), +] + + +@pytest.fixture +def recovery_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """ + 隔离依赖清单、约束文件、安装锁及子进程，保留真实安装与恢复编排。 + + :param tmp_path: 本用例独占的临时目录 + :param monkeypatch: 在用例结束时撤销边界替换的夹具 + :return: 清单路径、约束路径、独占安装锁以及捕获的日志 + """ + manifest = tmp_path / "requirements.txt" + manifest.write_text("demo\n", encoding="utf-8") + constraints = tmp_path / "constraints.txt" + lock = threading.Lock() + logs: list[tuple[str, str]] = [] + monkeypatch.setattr(PluginRuntimeHealth, "_package_install_lock", lock) + monkeypatch.setattr(health_module, "is_free_threaded_runtime", lambda: False) + monkeypatch.setattr(health_module, "get_runtime_setting", lambda key: { + "ROOT_PATH": tmp_path, "CONFIG_PATH": tmp_path, "TEMP_PATH": tmp_path, + "PACKAGE_CACHE_PATH": tmp_path, "PIP_PROXY": "", "PROXY_HOST": "", + }[key]) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__get_installed_packages", lambda: {}) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__get_protected_runtime_packages", lambda _: {"fastapi": "1"}) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__validate_runtime_dependency_conflicts", lambda *_: (True, "")) + + def create_constraints(_packages): + """ + 在临时目录生成约束，检查后续路径是否承担清理责任。 + """ + constraints.write_text("fastapi==1\n", encoding="utf-8") + return constraints + + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__create_runtime_constraints_file", create_constraints) + monkeypatch.setattr(health_module.logger, "warning", lambda message: logs.append(("warning", message))) + monkeypatch.setattr(health_module.logger, "error", lambda message: logs.append(("error", message))) + return manifest, constraints, lock, logs + + +def _patch_runtime(monkeypatch, environment, case: _RecoveryCase) -> list[str]: + """ + 在子进程和健康探测边界重放固定应答，并记录持锁期间的操作顺序。 + + :param monkeypatch: 自动恢复边界方法的夹具 + :param environment: 隔离的清单、约束、锁与日志 + :param case: 完整应答序列及预期结果 + :return: 按真实发生顺序追加的操作列表 + """ + _, constraints, lock, _ = environment + events: list[str] = [] + installs, repairs, health = iter(case.installs), iter(case.repairs), iter(case.health) + + def strategies(request: PackageInstallRequest, dry_run: bool = False) -> list[PackageInstallStrategy]: + """ + 返回离线命令，保持真实修复目标和安装策略切换可观察。 + """ + assert not dry_run + kind = "repair" if request.purpose == "runtime-repair" else "install" + if kind == "repair": + assert request.dependency_files == (constraints,) + else: + assert request.constraints_file == constraints + return [PackageInstallStrategy(f"{kind}-{index}", [kind], {}, [kind]) + for index in range(1 if kind == "repair" else 2)] + + def execute(command, **kwargs): + """ + 检查安装与恢复持有同一锁，禁止启动真实子进程。 + """ + assert lock.locked() and constraints.exists() + assert kwargs.get("timeout") in (None, 300) + kind = command[0] + events.append(kind) + return next(repairs if kind == "repair" else installs) + + async def execute_async(command, **kwargs): + """ + 复用边界应答，同时要求异步执行保留原有超时。 + """ + assert kwargs["timeout"] == 300 + return execute(command, **kwargs) + + def probe(): + """ + 重放健康快照，若多探测或少探测则由序列与事件断言识别。 + """ + assert lock.locked() + events.append("health") + return next(health) + + async def probe_async(): + """ + 通过相同快照验证异步恢复决策。 + """ + return probe() + + monkeypatch.setattr(health_module, "build_package_install_strategies", strategies) + monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess", execute) + monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess_async", execute_async) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__run_runtime_healthcheck", probe) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_run_runtime_healthcheck", probe_async) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__refresh_import_system", lambda: events.append("refresh")) + return events + + +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("case", _CASES) +def test_install_recovery_contract(recovery_environment, monkeypatch, case, use_async) -> None: + """ + 两种入口保持既有结果、恢复顺序、停止条件及诊断差异。 + + :param recovery_environment: 隔离的安装环境 + :param monkeypatch: 自动恢复子进程和健康检查的夹具 + :param case: 参数化的外部应答及预期合同 + :param use_async: 是否通过异步公开入口执行 + """ + manifest, constraints, lock, logs = recovery_environment + events = _patch_runtime(monkeypatch, recovery_environment, case) + if use_async: + result = asyncio.run(PluginRuntimeHealth().async_install_packages_with_fallback(manifest)) + else: + result = PluginRuntimeHealth.install_packages_with_fallback(manifest) + assert result == case.result + assert tuple(events) == case.events + assert not constraints.exists() + assert not lock.locked() + remaining = [message for _, message in logs if "安装前已有的运行环境异常仍然存在" in message] + assert bool(remaining) == (not use_async and case.health == (_OLD_FAILURE, _OLD_FAILURE)) + restored_errors = [message for _, message in logs if "主程序依赖恢复后仍未通过健康检查" in message] + assert bool(restored_errors) == (not use_async and case.installs[0][0] and case.health[-1] == _BROKEN and bool(case.repairs) and case.repairs[0][0]) + + +@pytest.mark.parametrize("stage", ["baseline", "install", "post-install", "repair", "post-repair"]) +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +def test_install_exception_releases_resources(recovery_environment, monkeypatch, stage, use_async) -> None: + """ + 探测、安装和恢复异常按原样传播，始终释放约束文件与安装锁。 + + :param recovery_environment: 隔离的安装环境 + :param monkeypatch: 自动恢复边界方法的夹具 + :param stage: 注入异常的外部操作阶段 + :param use_async: 是否执行异步公开入口 + """ + manifest, constraints, lock, _ = recovery_environment + case = _RecoveryCase(((True, "ok"),), (_HEALTHY, _BROKEN, _HEALTHY), + ((True, "ok"),), (False, ""), ()) + _patch_runtime(monkeypatch, recovery_environment, case) + failure = RuntimeError(f"{stage} failed") + probe = PluginRuntimeHealth._PluginRuntimeHealth__run_runtime_healthcheck + execute = health_module.SystemUtils.execute_with_subprocess + phases = iter(("baseline", "post-install", "post-repair")) + + def failing_probe(): + """ + 在指定的健康探测阶段注入同一异常对象。 + """ + if next(phases) == stage: + raise failure + return probe() + + def failing_execute(command, **kwargs): + """ + 在安装或修复子进程边界注入异常。 + """ + if command[0] == stage: + raise failure + return execute(command, **kwargs) + + async def failing_probe_async(): + """ + 保持异步探测入口的异常传播行为。 + """ + return failing_probe() + + async def failing_execute_async(command, **kwargs): + """ + 保持异步子进程入口的异常传播行为。 + """ + return failing_execute(command, **kwargs) + + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__run_runtime_healthcheck", failing_probe) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_run_runtime_healthcheck", failing_probe_async) + monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess", failing_execute) + monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess_async", failing_execute_async) + with pytest.raises(RuntimeError) as raised: + if use_async: + asyncio.run(PluginRuntimeHealth().async_install_packages_with_fallback(manifest)) + else: + PluginRuntimeHealth.install_packages_with_fallback(manifest) + assert raised.value is failure + assert not constraints.exists() + assert not lock.locked() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stage", ["baseline", "install", "post-install", "repair", "post-repair"]) +async def test_cancelled_recovery_releases_resources(recovery_environment, monkeypatch, stage) -> None: + """ + 在真实等待点取消安装或恢复，取消不得转换成失败结果或继续下一策略。 + + :param recovery_environment: 隔离的安装环境 + :param monkeypatch: 自动恢复子进程和探测的夹具 + :param stage: 等待取消的外部操作阶段 + """ + manifest, constraints, lock, _ = recovery_environment + case = _RecoveryCase(((True, "ok"),), (_HEALTHY, _BROKEN, _HEALTHY), + ((True, "ok"),), (False, ""), ()) + events = _patch_runtime(monkeypatch, recovery_environment, case) + probe = PluginRuntimeHealth._PluginRuntimeHealth__run_runtime_healthcheck + execute = health_module.SystemUtils.execute_with_subprocess + phases = iter(("baseline", "post-install", "post-repair")) + entered = asyncio.Event() + exited = asyncio.Event() + + async def wait_for_cancel(phase: str) -> None: + """ + 使用屏障精确定位取消时点，收尾标记用于验证子操作先退出。 + """ + if phase == stage: + entered.set() + try: + await asyncio.Future() + finally: + exited.set() + + async def blocked_probe(): + """ + 在目标健康探测之前暂停。 + """ + await wait_for_cancel(next(phases)) + return probe() + + async def blocked_execute(command, **kwargs): + """ + 在目标子进程边界暂停，不启动真实安装器。 + """ + await wait_for_cancel(command[0]) + return execute(command, **kwargs) + + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_run_runtime_healthcheck", blocked_probe) + monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess_async", blocked_execute) + task = asyncio.create_task(PluginRuntimeHealth().async_install_packages_with_fallback(manifest)) + try: + await asyncio.wait_for(entered.wait(), timeout=5) + before_cancel = list(events) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=5) + assert exited.is_set() + assert events == before_cancel + assert not constraints.exists() + assert not lock.locked() + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_cancelled_install_waiting_for_lock_preserves_owner(recovery_environment, monkeypatch) -> None: + """ + 排队时取消只清理自己的约束，不能释放其他安装者持有的锁。 + + :param recovery_environment: 隔离的安装环境 + :param monkeypatch: 自动恢复预检边界的夹具 + """ + manifest, constraints, lock, _ = recovery_environment + ready = asyncio.Event() + + async def preflight(_request): + """ + 在创建约束之后标记就绪，下一步将等待已被占用的锁。 + """ + ready.set() + return None + + monkeypatch.setattr(health_module, "build_package_install_strategies", lambda _request: []) + monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_free_threaded_preflight", preflight) + lock.acquire() + task = asyncio.create_task(PluginRuntimeHealth().async_install_packages_with_fallback(manifest)) + try: + await asyncio.wait_for(ready.wait(), timeout=5) + assert constraints.exists() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=5) + assert lock.locked() + assert not constraints.exists() + finally: + lock.release() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +def test_empty_install_strategies_preserve_failure(recovery_environment, monkeypatch, use_async) -> None: + """ + 无可用策略时保留原有失败说明，不启动安装或恢复。 + + :param recovery_environment: 隔离的安装环境 + :param monkeypatch: 自动恢复策略工厂和边界方法的夹具 + :param use_async: 是否执行异步公开入口 + """ + manifest, constraints, lock, _ = recovery_environment + case = _RecoveryCase((), (_HEALTHY,), (), (False, ""), ()) + events = _patch_runtime(monkeypatch, recovery_environment, case) + monkeypatch.setattr(health_module, "build_package_install_strategies", lambda _request: []) + if use_async: + result = asyncio.run(PluginRuntimeHealth().async_install_packages_with_fallback(manifest)) + else: + result = PluginRuntimeHealth.install_packages_with_fallback(manifest) + assert result == (False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束") + assert events == ["health"] + assert not constraints.exists() + assert not lock.locked()