refactor(plugin): share dependency recovery decisions (#7060)

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()

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,

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()