Merge pull request #6413 from InfinityPacer/codex/replay/g3c

2026-09-05 23:47:41 +08:00 · 2026-08-23 14:51:07 +08:00
parent b1ad309fc7 3fa8ddf63a
commit 1f64e99a6f
26 changed files with 1633 additions and 146 deletions
@@ -82,6 +82,19 @@ def _empty_installed_plugins() -> List[str]:
 _installed_plugins_provider: InstalledPluginsProvider = _empty_installed_plugins


+async def _await_thread_operation(func, *args, **kwargs):
+    """取消请求到达时先等待同步插件操作收口，避免目录写入继续进行。"""
+    task = asyncio.create_task(asyncio.to_thread(func, *args, **kwargs))
+    try:
+        return await asyncio.shield(task)
+    except asyncio.CancelledError:
+        try:
+            await asyncio.shield(task)
+        except BaseException:
+            pass
+        raise
+
+
 def configure_installed_plugins_provider(
    provider: InstalledPluginsProvider,
 ) -> None:
@@ -177,6 +190,7 @@ class PluginHelper(metaclass=WeakSingleton):
    _base_url = "https://raw.githubusercontent.com/{user}/{repo}/main/"
    # 串行化运行期依赖安装，避免多个包安装子进程和导入缓存刷新互相踩踏。
    _package_install_lock = threading.Lock()
+    PLUGIN_DEPENDENCY_INSTALL_TIMEOUT = 300
    # 同仓库的并发 Release 请求共享任务；事件循环参与键控，避免热重载或测试循环切换后复用失效任务。
    _release_task_lock = threading.Lock()
    _release_tasks: Dict[Tuple[asyncio.AbstractEventLoop, str, bool], asyncio.Task] = {}
@@ -1472,6 +1486,42 @@ class PluginHelper(metaclass=WeakSingleton):
                    temp_file.write(f"{cls.__format_package_name(package_name)}>={version}\n")
        return Path(temp_file.name)

+    @classmethod
+    async def __async_create_runtime_constraints_file(
+            cls,
+            protected_packages: Dict[str, Version],
+    ) -> Path:
+        """创建临时约束文件，取消时等待创建收口并删除已产生的文件。"""
+        create_task = asyncio.create_task(
+            asyncio.to_thread(
+                cls.__create_runtime_constraints_file,
+                protected_packages,
+            )
+        )
+        try:
+            return await asyncio.shield(create_task)
+        except asyncio.CancelledError:
+            async def cleanup_created_file() -> None:
+                try:
+                    created_file = await create_task
+                except BaseException:
+                    return
+                await asyncio.to_thread(created_file.unlink, missing_ok=True)
+
+            cleanup_task = asyncio.create_task(cleanup_created_file())
+            while not cleanup_task.done():
+                try:
+                    await asyncio.shield(cleanup_task)
+                except asyncio.CancelledError:
+                    continue
+                except Exception:
+                    break
+            try:
+                await cleanup_task
+            except Exception as err:
+                logger.warning(f"[UV] 取消后清理运行环境约束文件失败：{err}")
+            raise
+
    @staticmethod
    def __refresh_import_system():
        """
@@ -2421,18 +2471,284 @@ class PluginHelper(metaclass=WeakSingleton):

        return True, ""

-    async def __async_install_packages_with_fallback(
-            self,
-            dependency_file: Path,
-            find_links_dirs: Optional[List[Path]] = None) -> Tuple[bool, str]:
-        """
-        在线程池中执行插件依赖安装，避免同步包安装子进程阻塞事件循环。
-        """
-        return await asyncio.to_thread(
-            self.install_packages_with_fallback,
-            dependency_file,
-            find_links_dirs
+    @classmethod
+    async def __async_run_runtime_healthcheck(cls) -> Dict[str, Tuple[bool, str]]:
+        """异步执行插件安装后的运行环境检查。"""
+        health_snapshot: Dict[str, Tuple[bool, str]] = {}
+        uv_check = cls.__build_runtime_uv_command("check")
+        if uv_check:
+            checks = [("uv check", uv_check)]
+        else:
+            health_snapshot["uv check"] = (False, "未找到 uv 可执行文件")
+            checks = []
+        checks.append(("核心依赖导入检查", [
+            sys.executable,
+            "-c",
+            cls._runtime_import_probe,
+        ]))
+        for check_name, command in checks:
+            health_snapshot[check_name] = (
+                await SystemUtils.execute_with_subprocess_async(
+                    command,
+                    timeout=30,
+                )
+            )
+        return health_snapshot
+
+    @classmethod
+    async def __async_repair_main_runtime_dependencies(
+            cls,
+            snapshot_file: Optional[Path] = None,
+    ) -> Tuple[bool, str]:
+        """异步恢复主程序运行依赖，避免修复命令绕过可取消进程边界。"""
+        repair_target = snapshot_file
+        repair_desc = "主程序依赖快照"
+        if repair_target and not await _await_thread_operation(repair_target.exists):
+            repair_target = None
+        if repair_target is None:
+            repair_target = settings.ROOT_PATH / "pyproject.toml"
+            repair_desc = "主程序 uv.lock"
+        if not await _await_thread_operation(repair_target.exists):
+            return False, f"恢复依赖文件不存在：{repair_target}"
+        lock_file = settings.ROOT_PATH / "uv.lock"
+        if snapshot_file is None and not await _await_thread_operation(lock_file.exists):
+            return False, f"恢复依赖文件不存在：{settings.ROOT_PATH / 'uv.lock'}"
+
+        request = cls.__build_package_install_request(
+            repair_target,
+            purpose="runtime-repair",
        )
+        strategies = (
+            build_package_install_strategies(request)
+            if snapshot_file is not None
+            else build_project_sync_strategies(request)
+        )
+        last_error = ""
+        for strategy in strategies:
+            logger.warning(
+                f"[UV] 运行环境异常，尝试使用策略：{strategy.strategy_name} 恢复{repair_desc}"
+            )
+            success, message = await SystemUtils.execute_with_subprocess_async(
+                strategy.command,
+                env=strategy.env,
+                safe_command=strategy.safe_log_command,
+                timeout=cls.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT,
+            )
+            if success:
+                cls.__refresh_import_system()
+                return True, message
+            last_error = message
+            logger.error(
+                f"[UV] 使用策略：{strategy.strategy_name} 恢复{repair_desc}失败：{message}"
+            )
+        return False, last_error or f"恢复{repair_desc}失败"
+
+    @classmethod
+    async def __async_repair_if_runtime_broken(
+            cls,
+            snapshot_file: Optional[Path],
+            baseline_health: Dict[str, Tuple[bool, str]],
+    ) -> Tuple[bool, str]:
+        """异步检查并修复安装过程中新增的主程序环境异常。"""
+        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)
+        )
+        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,
+            dependency_files: Path | Sequence[Path],
+            find_links_dirs: Optional[List[Path]] = None,
+    ) -> Tuple[bool, str]:
+        """通过可取消子进程异步安装一组插件依赖清单。"""
+        return await self.__async_install_packages_with_fallback(
+            dependency_files,
+            find_links_dirs,
+        )
+
+    @classmethod
+    async def __async_install_packages_with_fallback(
+            cls,
+            dependency_files: Path | Sequence[Path],
+            find_links_dirs: Optional[List[Path]] = None,
+    ) -> Tuple[bool, str]:
+        """异步安装插件依赖，并让取消能够终止 uv 子进程。"""
+        if isinstance(dependency_files, Path):
+            resolved_dependency_files = (dependency_files,)
+        else:
+            resolved_dependency_files = tuple(Path(item) for item in dependency_files)
+        if not resolved_dependency_files:
+            return False, "没有传入插件依赖清单"
+
+        candidate_dirs = []
+        for dependency_file in resolved_dependency_files:
+            wheels_dir = dependency_file.parent / "wheels"
+            if await _await_thread_operation(wheels_dir.is_dir):
+                candidate_dirs.append(wheels_dir)
+        if find_links_dirs:
+            candidate_dirs.extend(find_links_dirs)
+
+        resolved_dirs = []
+        seen_dirs = set()
+        for candidate_dir in candidate_dirs:
+            candidate_path = Path(candidate_dir)
+            if not await _await_thread_operation(candidate_path.is_dir):
+                continue
+            candidate_key = str(
+                await _await_thread_operation(candidate_path.resolve)
+            )
+            if candidate_key in seen_dirs:
+                continue
+            seen_dirs.add(candidate_key)
+            resolved_dirs.append(candidate_path)
+
+        installed_packages = await _await_thread_operation(
+            cls.__get_installed_packages,
+        )
+        protected_packages = await _await_thread_operation(
+            cls.__get_protected_runtime_packages,
+            installed_packages,
+        )
+        for dependency_file in resolved_dependency_files:
+            check_ok, check_message = await _await_thread_operation(
+                cls.__validate_runtime_dependency_conflicts,
+                dependency_file,
+                protected_packages,
+            )
+            if not check_ok:
+                logger.error(f"[UV] 运行环境冲突预检失败：{check_message}")
+                return False, check_message
+
+        constraints_file = None
+        if protected_packages:
+            try:
+                constraints_file = await cls.__async_create_runtime_constraints_file(
+                    protected_packages,
+                )
+            except Exception as err:
+                logger.error(f"[UV] 创建运行环境约束文件失败：{err}")
+                return False, f"创建运行环境约束文件失败：{err}"
+
+        request = cls.__build_package_install_request(
+            resolved_dependency_files,
+            find_links_dirs=resolved_dirs,
+            constraints_file=constraints_file,
+            purpose="plugin",
+        )
+        strategies = build_package_install_strategies(request)
+        acquired = False
+        try:
+            while not cls._package_install_lock.acquire(blocking=False):
+                await asyncio.sleep(0.01)
+            acquired = True
+            baseline_health = await cls.__async_run_runtime_healthcheck()
+            baseline_health_message = cls.__runtime_health_regression_message(
+                {},
+                baseline_health,
+            )
+            if baseline_health_message:
+                logger.warning(
+                    f"[UV] 安装前运行环境已存在异常，本次安装仅拦截新增异常："
+                    f"{baseline_health_message}"
+                )
+
+            last_error = ""
+            for strategy in strategies:
+                logger.debug(
+                    f"[UV] 尝试使用策略：{strategy.strategy_name} 安装依赖，"
+                    f"命令：{' '.join(strategy.safe_log_command)}"
+                )
+                success, message = await SystemUtils.execute_with_subprocess_async(
+                    strategy.command,
+                    env=strategy.env,
+                    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
+
+                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, (
+                        f"策略 {strategy.strategy_name} 安装依赖失败：{message}；"
+                        f"{repair_message}"
+                    )
+            return False, (
+                f"[UV] 所有策略均安装依赖失败：{last_error}"
+                if last_error
+                else "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束"
+            )
+        finally:
+            if acquired:
+                cls._package_install_lock.release()
+            if constraints_file:
+                await _await_thread_operation(
+                    constraints_file.unlink,
+                    missing_ok=True,
+                )

    async def __async_backup_plugin(self, pid: str) -> str:
        """
@@ -2444,14 +2760,16 @@ class PluginHelper(metaclass=WeakSingleton):
        backup_dir = AsyncPath(settings.TEMP_PATH) / "plugin_backup" / pid.lower()

        if await plugin_dir.exists():
-            # 备份时清理已有的备份目录，防止残留文件影响
-            if await backup_dir.exists():
-                await aioshutil.rmtree(backup_dir, ignore_errors=True)
-                logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}")
+            try:
+                if await backup_dir.exists():
+                    await aioshutil.rmtree(backup_dir, ignore_errors=True)
+                    logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}")

-            # 异步复制目录
-            await self._async_copytree(plugin_dir, backup_dir)
-            logger.debug(f"{pid} 插件已备份到 {backup_dir}")
+                await self._async_copytree(plugin_dir, backup_dir)
+                logger.debug(f"{pid} 插件已备份到 {backup_dir}")
+            except asyncio.CancelledError:
+                await aioshutil.rmtree(backup_dir, ignore_errors=True)
+                raise

        return str(backup_dir) if await backup_dir.exists() else None

@@ -2558,7 +2876,12 @@ class PluginHelper(metaclass=WeakSingleton):
        :return: (是否成功, 错误信息)
        """
        if self.is_local_repo_url(repo_url):
-            return await asyncio.to_thread(self.install_local, pid, repo_url, force_install)
+            return await _await_thread_operation(
+                self.install_local,
+                pid,
+                repo_url,
+                force_install,
+            )

        if SystemUtils.is_frozen():
            return False, "可执行文件模式下，只能安装本地插件"
@@ -2663,35 +2986,46 @@ class PluginHelper(metaclass=WeakSingleton):
        异步安装流程，处理插件内容准备、依赖安装和注册
        """
        backup_dir = None
-        if not force_install:
-            backup_dir = await self.__async_backup_plugin(pid)
+        try:
+            if not force_install:
+                backup_dir = await self.__async_backup_plugin(pid)

-        await self.__async_remove_old_plugin(pid)
+            await self.__async_remove_old_plugin(pid)

-        success, message = await prepare_content()
-        if not success:
-            logger.error(f"{pid} 准备插件内容失败：{message}")
+            success, message = await prepare_content()
+            if not success:
+                logger.error(f"{pid} 准备插件内容失败：{message}")
+                if backup_dir:
+                    await self.__async_restore_plugin(pid, backup_dir)
+                    logger.warning(f"{pid} 插件安装失败，已还原备份插件")
+                else:
+                    await self.__async_remove_old_plugin(pid)
+                    logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装")
+                return False, message
+
+            dependencies_exist, dep_ok, dep_msg = (
+                await self.__async_install_dependencies_if_required(pid)
+            )
+            if dependencies_exist and not dep_ok:
+                logger.error(f"{pid} 依赖安装失败：{dep_msg}")
+                if backup_dir:
+                    await self.__async_restore_plugin(pid, backup_dir)
+                    logger.warning(f"{pid} 插件安装失败，已还原备份插件")
+                else:
+                    await self.__async_remove_old_plugin(pid)
+                    logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装")
+                return False, dep_msg
+
+            await _await_thread_operation(self.refresh_persistent_plugin_backup, pid)
+            return True, ""
+        except asyncio.CancelledError:
+            logger.warning(
+                f"{pid} 插件安装被取消，Python 依赖环境可能已经改变"
+            )
+            raise
+        finally:
            if backup_dir:
-                await self.__async_restore_plugin(pid, backup_dir)
-                logger.warn(f"{pid} 插件安装失败，已还原备份插件")
-            else:
-                await self.__async_remove_old_plugin(pid)
-                logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
-            return False, message
-
-        dependencies_exist, dep_ok, dep_msg = await self.__async_install_dependencies_if_required(pid)
-        if dependencies_exist and not dep_ok:
-            logger.error(f"{pid} 依赖安装失败：{dep_msg}")
-            if backup_dir:
-                await self.__async_restore_plugin(pid, backup_dir)
-                logger.warn(f"{pid} 插件安装失败，已还原备份插件")
-            else:
-                await self.__async_remove_old_plugin(pid)
-                logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
-            return False, dep_msg
-
-        await asyncio.to_thread(self.refresh_persistent_plugin_backup, pid)
-        return True, ""
+                await aioshutil.rmtree(backup_dir, ignore_errors=True)

    def __prepare_content_via_filelist_sync(self, pid: str, user_repo: str,
                                            package_version: Optional[str]) -> Tuple[bool, str]:
@@ -1,8 +1,10 @@
+import asyncio
 import datetime
 import hashlib
 import os
 import platform
 import re
+import signal
 import shutil
 import socket
 import struct
@@ -101,6 +103,7 @@ class SystemUtils:
            command: list,
            env: Optional[dict[str, str]] = None,
            safe_command: Optional[list[str]] = None,
+            timeout: Optional[float] = 300,
    ) -> Tuple[bool, str]:
        """
        执行命令并捕获标准输出和错误输出，记录日志。
@@ -108,6 +111,7 @@ class SystemUtils:
        :param command: 要执行的命令，以列表形式提供
        :param env: 传递给子进程的环境变量
        :param safe_command: 用于错误信息展示的脱敏命令
+        :param timeout: 子进程最长运行时间，None 表示不设置超时
        :return: (命令是否成功, 输出信息或错误信息)
        """
        display_command = safe_command or command
@@ -120,6 +124,7 @@ class SystemUtils:
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=env,
+                timeout=timeout,
            )
            # 合并 stdout 和 stderr
            output = SystemUtils.redact_url_userinfo(result.stdout + result.stderr)
@@ -140,6 +145,26 @@ class SystemUtils:
                f"返回码：{e.returncode}，{'; '.join(output_parts)}"
            )
            return False, error_message
+        except subprocess.TimeoutExpired as e:
+            stdout = SystemUtils.redact_url_userinfo(
+                SystemUtils._decode_subprocess_output(e.stdout)
+            )
+            stderr = SystemUtils.redact_url_userinfo(
+                SystemUtils._decode_subprocess_output(e.stderr)
+            )
+            output = "；".join(
+                part for part in (
+                    f"标准输出：{stdout}" if stdout else "",
+                    f"错误输出：{stderr}" if stderr else "",
+                ) if part
+            )
+            if output:
+                output = f"，{output}"
+            timeout_text = "未设置" if timeout is None else f"{timeout:g} 秒"
+            return False, (
+                f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，"
+                f"执行超时（{timeout_text}）{output}"
+            )
        except Exception as e:
            error_message = (
                f"未知错误，命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，"
@@ -147,6 +172,213 @@ class SystemUtils:
            )
            return False, error_message

+    @staticmethod
+    def _decode_subprocess_output(value: object) -> str:
+        """把 subprocess 的文本或字节输出统一成可脱敏的字符串。"""
+        if value is None:
+            return ""
+        if isinstance(value, bytes):
+            return value.decode(errors="replace").strip()
+        return str(value).strip()
+
+    @staticmethod
+    async def _terminate_async_subprocess(
+            process: asyncio.subprocess.Process,
+            communication_task: "asyncio.Task[tuple[bytes, bytes]]",
+            grace_seconds: float = 5,
+    ) -> None:
+        """终止安装子进程及其同组子进程，并确保句柄已回收。"""
+        process_tree = SystemUtils._process_tree(process.pid)
+        if process.returncode is None:
+            try:
+                if os.name == "nt":
+                    SystemUtils._signal_process_tree(process_tree)
+                    process.terminate()
+                else:
+                    os.killpg(process.pid, signal.SIGTERM)
+            except (ProcessLookupError, OSError):
+                pass
+
+        try:
+            await asyncio.wait_for(
+                asyncio.shield(communication_task),
+                timeout=grace_seconds,
+            )
+        except (asyncio.TimeoutError, asyncio.CancelledError):
+            pass
+
+        # 管道可能在父进程退出时立即关闭，而后代仍在运行或忽略终止信号。
+        # 通信任务完成只代表 stdout/stderr 已收口，不能作为进程树已收口的依据。
+        try:
+            known_pids = {item.pid for item in process_tree}
+            process_tree.extend(
+                process_item
+                for process_item in SystemUtils._process_tree(process.pid)
+                if process_item.pid not in known_pids
+            )
+        except (ProcessLookupError, OSError):
+            pass
+
+        alive_processes = SystemUtils._alive_processes(process_tree)
+        if alive_processes or process.returncode is None:
+            if os.name == "nt":
+                try:
+                    SystemUtils._signal_process_tree(alive_processes, force=True)
+                    process.kill()
+                except (ProcessLookupError, OSError):
+                    pass
+            else:
+                try:
+                    os.killpg(process.pid, signal.SIGKILL)
+                except (ProcessLookupError, OSError):
+                    pass
+                try:
+                    SystemUtils._signal_process_tree(alive_processes, force=True)
+                except (ProcessLookupError, OSError):
+                    pass
+            await SystemUtils._wait_process_tree(
+                process_tree,
+                grace_seconds,
+            )
+        if not communication_task.done():
+            communication_task.cancel()
+            await asyncio.gather(communication_task, return_exceptions=True)
+        try:
+            await process.wait()
+        except (ProcessLookupError, OSError):
+            pass
+
+    @staticmethod
+    def _process_tree(pid: int) -> list[psutil.Process]:
+        """收集安装进程及其后代，避免构建进程继续写运行环境。"""
+        try:
+            parent = psutil.Process(pid)
+            return [parent, *parent.children(recursive=True)]
+        except (psutil.Error, OSError):
+            return []
+
+    @staticmethod
+    def _alive_processes(processes: list[psutil.Process]) -> list[psutil.Process]:
+        """返回仍可能执行外部副作用的进程，忽略已退出和僵尸进程。"""
+        alive = []
+        seen_pids = set()
+        for process in processes:
+            if process.pid in seen_pids:
+                continue
+            seen_pids.add(process.pid)
+            try:
+                if process.is_running() and process.status() != psutil.STATUS_ZOMBIE:
+                    alive.append(process)
+            except (psutil.Error, OSError):
+                continue
+        return alive
+
+    @staticmethod
+    async def _wait_process_tree(
+            processes: list[psutil.Process],
+            timeout: float,
+    ) -> list[psutil.Process]:
+        """在事件循环中有界等待整棵进程树退出，并返回残留进程。"""
+        deadline = asyncio.get_running_loop().time() + max(timeout, 0)
+        while True:
+            alive = SystemUtils._alive_processes(processes)
+            if not alive:
+                return []
+            remaining = deadline - asyncio.get_running_loop().time()
+            if remaining <= 0:
+                return alive
+            await asyncio.sleep(min(0.05, remaining))
+
+    @staticmethod
+    def _signal_process_tree(
+            processes: list[psutil.Process],
+            force: bool = False,
+    ) -> None:
+        """向进程树发送终止或强制结束信号。"""
+        for child in reversed(processes):
+            try:
+                (child.kill if force else child.terminate)()
+            except (psutil.Error, OSError):
+                continue
+
+    @staticmethod
+    async def execute_with_subprocess_async(
+            command: list,
+            env: Optional[dict[str, str]] = None,
+            safe_command: Optional[list[str]] = None,
+            timeout: Optional[float] = 300,
+    ) -> Tuple[bool, str]:
+        """异步执行可取消的子进程，超时或取消时回收整个进程组。"""
+        display_command = safe_command or command
+        process: Optional[asyncio.subprocess.Process] = None
+        communication_task: Optional["asyncio.Task[tuple[bytes, bytes]]"] = None
+        try:
+            creationflags = (
+                getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
+                if os.name == "nt"
+                else 0
+            )
+            process = await asyncio.create_subprocess_exec(
+                *command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                env=env,
+                start_new_session=os.name != "nt",
+                creationflags=creationflags,
+            )
+            communication_task = asyncio.create_task(process.communicate())
+            try:
+                stdout, stderr = await asyncio.wait_for(
+                    asyncio.shield(communication_task),
+                    timeout=timeout,
+                )
+            except asyncio.TimeoutError:
+                await SystemUtils._terminate_async_subprocess(
+                    process,
+                    communication_task,
+                )
+                timeout_text = "未设置" if timeout is None else f"{timeout:g} 秒"
+                return False, (
+                    f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，"
+                    f"执行超时（{timeout_text}）"
+                )
+            except asyncio.CancelledError:
+                await SystemUtils._terminate_async_subprocess(
+                    process,
+                    communication_task,
+                )
+                raise
+
+            output = SystemUtils.redact_url_userinfo(
+                SystemUtils._decode_subprocess_output(stdout)
+                + SystemUtils._decode_subprocess_output(stderr)
+            )
+            if process.returncode == 0:
+                return True, output
+            return False, (
+                f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，"
+                f"执行失败，返回码：{process.returncode}，"
+                f"{output or '无标准输出或错误输出'}"
+            )
+        except asyncio.CancelledError:
+            if process is not None and communication_task is not None:
+                await SystemUtils._terminate_async_subprocess(
+                    process,
+                    communication_task,
+                )
+            raise
+        except Exception as e:
+            if process is not None and communication_task is not None:
+                await SystemUtils._terminate_async_subprocess(
+                    process,
+                    communication_task,
+                )
+            error_message = (
+                f"未知错误，命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，"
+                f"错误：{SystemUtils.redact_url_userinfo(str(e))}"
+            )
+            return False, error_message
+
    @staticmethod
    def is_docker() -> bool:
        """
@@ -364,5 +364,17 @@ class PluginDependencyInstaller:
        return await asyncio.to_thread(self.find_missing)

    async def async_install(self, dependencies: list[str]) -> tuple[bool, str]:
-        """在线程池中安装依赖，复用同步包安装策略。"""
-        return await asyncio.to_thread(self.install, dependencies)
+        """异步安装依赖，使用可取消的包安装子进程。"""
+        if not dependencies:
+            return False, "没有传入需要安装的依赖项"
+        try:
+            manifest_paths = [manifest.path for manifest in self._plugin_manifests()]
+            if not manifest_paths:
+                return False, "没有找到已安装插件的依赖清单"
+            return await self._helper.async_install_packages_with_fallback(
+                manifest_paths,
+                self._wheels_dirs(),
+            )
+        except Exception as err:
+            logger.error(f"安装依赖项时发生错误：{err}")
+            return False, f"安装依赖项时发生错误：{err}"
@@ -19,6 +19,19 @@ from app.runtime.settings import RuntimeSettingsCompat
 settings = RuntimeSettingsCompat()


+async def _await_thread_operation(func, *args, **kwargs):
+    """取消请求到达时先等待文件操作收口，避免后台线程继续写运行目录。"""
+    task = asyncio.create_task(asyncio.to_thread(func, *args, **kwargs))
+    try:
+        return await asyncio.shield(task)
+    except asyncio.CancelledError:
+        try:
+            await asyncio.shield(task)
+        except BaseException:
+            pass
+        raise
+
+
@dataclass(frozen=True, slots=True)
 class PluginPackageCheckpoint:
    """记录一次插件包变更前可用于补偿恢复的文件快照。"""
@@ -74,7 +87,7 @@ class PluginPackageManager:

    async def async_checkpoint(self, plugin_id: str) -> PluginPackageCheckpoint:
        """在线程池中创建插件包文件快照。"""
-        return await asyncio.to_thread(self.checkpoint, plugin_id)
+        return await _await_thread_operation(self.checkpoint, plugin_id)

    @staticmethod
    def commit(checkpoint: PluginPackageCheckpoint) -> None:
@@ -83,25 +96,27 @@ class PluginPackageManager:

    async def async_commit(self, checkpoint: PluginPackageCheckpoint) -> None:
        """在线程池中清理已提交的插件包快照。"""
-        await asyncio.to_thread(self.commit, checkpoint)
+        await _await_thread_operation(self.commit, checkpoint)

    @staticmethod
    def rollback(checkpoint: PluginPackageCheckpoint) -> None:
        """删除当前包并把变更前文件快照恢复到运行目录。"""
-        if checkpoint.plugin_dir.exists():
-            shutil.rmtree(checkpoint.plugin_dir)
        snapshot_dir = checkpoint.transaction_dir / "package"
        if checkpoint.existed:
            if not snapshot_dir.is_dir():
                raise FileNotFoundError(
                    f"插件 {checkpoint.plugin_id} 的补偿快照不存在：{snapshot_dir}"
                )
+        if checkpoint.plugin_dir.exists():
+            shutil.rmtree(checkpoint.plugin_dir)
+        if checkpoint.existed:
            shutil.copytree(snapshot_dir, checkpoint.plugin_dir)
-        shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False)
+        if checkpoint.transaction_dir.exists():
+            shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False)

    async def async_rollback(self, checkpoint: PluginPackageCheckpoint) -> None:
        """在线程池中恢复插件包文件快照。"""
-        await asyncio.to_thread(self.rollback, checkpoint)
+        await _await_thread_operation(self.rollback, checkpoint)

    def install(
        self,
@@ -57,7 +57,7 @@ from app.api.dependencies.plugin import (
 from app.adapters.external.server import MoviePilotServerHelper
 from app.adapters.external.market import PluginHelper
 from app.adapters.system.plugin.package import PluginPackageManager
-from app.application.database import DatabaseWorkerOverloadedError
+from app.schemas.exception import DatabaseWorkerOverloadedError
 from app.runtime.log import logger
 from app.schemas.types import SystemConfigKey

@@ -18,14 +18,6 @@ DatabaseProbe = Callable[[], Optional[str]]
 T = TypeVar("T")


-class DatabaseWorkerClosedError(RuntimeError):
-    """数据库执行器尚未启动或已经停止。"""
-
-
-class DatabaseWorkerOverloadedError(RuntimeError):
-    """数据库执行器的运行与排队容量已经用尽。"""
-
-
 class AsyncDatabaseExecutor(Protocol):
    """让异步业务调用同步短事务而不阻塞事件循环。"""

@@ -10,10 +10,12 @@ from weakref import WeakValueDictionary

 from app.application.database import (
    AsyncDatabaseExecutor,
+)
+from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary
+from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
 )
-from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary
 from app.runtime.observability import record_metric


@@ -2,11 +2,14 @@

 from __future__ import annotations

+import asyncio
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import Any, Optional

-from app.application.database import DatabaseWorkerOverloadedError
+from app.schemas.exception import DatabaseWorkerOverloadedError
+from app.application.plugin.lifecycle import plugin_lifecycle
+from app.runtime.log import logger


 InstalledPluginsReader = Callable[[], list[str]]
@@ -58,6 +61,22 @@ class PluginInstallResult:
    rollback: PluginInstallRollback = field(default_factory=PluginInstallRollback)


+@dataclass
+class _InstallState:
+    """记录取消补偿所需的事务阶段。"""
+
+    checkpoint: Any = None
+    stage: str = "package_checkpoint"
+    package_installed: bool = False
+    installed_list_touched: bool = False
+    installed_list_persisted: bool = False
+    runtime_touched: bool = False
+    registrations_touched: bool = False
+    refresh_compensated: bool = False
+    committed: bool = False
+    original_plugins: list[str] = field(default_factory=list)
+
+
 class PluginInstallCommand:
    """协调插件检查、包事务、持久化、运行态刷新和安装上报。"""

@@ -96,14 +115,44 @@ class PluginInstallCommand:
        repo_url: Optional[str],
        release_version: Optional[str] = None,
        force: bool = False,
+    ) -> PluginInstallResult:
+        """串行执行同一插件的完整安装生命周期，并保证取消后的补偿。"""
+        state = _InstallState()
+        async with plugin_lifecycle.hold(plugin_id):
+            try:
+                return await self._execute_locked(
+                    plugin_id=plugin_id,
+                    repo_url=repo_url,
+                    release_version=release_version,
+                    force=force,
+                    state=state,
+                )
+            except asyncio.CancelledError:
+                await self._rollback_cancelled(
+                    plugin_id=plugin_id,
+                    original_plugins=state.original_plugins,
+                    state=state,
+                )
+                raise
+
+    async def _execute_locked(
+        self,
+        *,
+        plugin_id: str,
+        repo_url: Optional[str],
+        release_version: Optional[str],
+        force: bool,
+        state: _InstallState,
    ) -> PluginInstallResult:
        """执行插件安装，并在关键阶段失败时恢复可补偿状态。"""
        installed_plugins = list(self._installed_plugins_reader() or [])
+        state.original_plugins = installed_plugins
        refreshed_only = not force and plugin_id in self._plugin_ids_provider()
        if refreshed_only:
            return await self._refresh_existing(
                plugin_id=plugin_id,
                repo_url=repo_url,
+                state=state,
            )
        if not repo_url:
            return PluginInstallResult(
@@ -112,8 +161,16 @@ class PluginInstallCommand:
                failure_stage="validation",
            )

+        checkpoint_task = asyncio.create_task(self._package_checkpointer(plugin_id))
        try:
-            checkpoint = await self._package_checkpointer(plugin_id)
+            checkpoint = await asyncio.shield(checkpoint_task)
+            state.checkpoint = checkpoint
+        except asyncio.CancelledError:
+            try:
+                state.checkpoint = await asyncio.shield(checkpoint_task)
+            except BaseException:
+                pass
+            raise
        except Exception as err:
            return PluginInstallResult(
                success=False,
@@ -121,13 +178,15 @@ class PluginInstallCommand:
                failure_stage="package_checkpoint",
            )

+        state.stage = "package_install"
        try:
-            state, message = await self._package_installer(
+            package_installed, message = await self._package_installer(
                plugin_id,
                repo_url,
                release_version,
                force,
            )
+            state.package_installed = package_installed
        except Exception as err:
            result = await self._failure(
                plugin_id=plugin_id,
@@ -140,7 +199,7 @@ class PluginInstallCommand:
            if isinstance(err, DatabaseWorkerOverloadedError):
                raise
            return result
-        if not state:
+        if not package_installed:
            return await self._failure(
                plugin_id=plugin_id,
                original_plugins=installed_plugins,
@@ -154,8 +213,11 @@ class PluginInstallCommand:
        if plugin_id not in installed_plugins:
            updated_plugins = [*installed_plugins, plugin_id]
            try:
+                # 写入方可能在返回前已经提交；取消时按已触碰处理，恢复原清单是幂等的。
+                state.installed_list_touched = True
                await self._installed_plugins_writer(updated_plugins)
                installed_list_persisted = True
+                state.installed_list_persisted = True
            except Exception as err:
                result = await self._failure(
                    plugin_id=plugin_id,
@@ -164,11 +226,14 @@ class PluginInstallCommand:
                    stage="installed_list_persistence",
                    message=str(err),
                    package_installed=True,
+                    installed_list_persisted=state.installed_list_touched,
                )
                if isinstance(err, DatabaseWorkerOverloadedError):
                    raise
                return result

+        state.stage = "runtime_reload"
+        state.runtime_touched = True
        try:
            await self._plugin_reloader(plugin_id)
        except Exception as err:
@@ -186,6 +251,8 @@ class PluginInstallCommand:
                raise
            return result

+        state.stage = "registration_refresh"
+        state.registrations_touched = True
        try:
            await self._registration_refresher(plugin_id)
        except Exception as err:
@@ -205,6 +272,9 @@ class PluginInstallCommand:
            return result

        checkpoint_cleanup_error = ""
+        state.stage = "checkpoint_commit"
+        # 运行态和注册已完成，后续只清理临时快照，不再把取消当作未提交安装回滚。
+        state.committed = True
        try:
            await self._package_committer(checkpoint)
        except Exception as err:
@@ -212,6 +282,7 @@ class PluginInstallCommand:

        reported = False
        report_error = ""
+        state.stage = "report"
        try:
            report_result = await self._install_reporter(plugin_id, repo_url)
            reported = report_result is not False
@@ -237,11 +308,65 @@ class PluginInstallCommand:
            checkpoint_cleanup_error=checkpoint_cleanup_error,
        )

+    async def _rollback_cancelled(
+        self,
+        *,
+        plugin_id: str,
+        original_plugins: list[str],
+        state: _InstallState,
+    ) -> None:
+        """在保留取消语义的同时完成文件、清单和运行态补偿。"""
+        if state.committed:
+            logger.warning(
+                f"插件 {plugin_id} 在安装提交后被取消，Python 依赖环境可能已经改变"
+            )
+            return
+        if state.refresh_compensated:
+            return
+        if state.checkpoint is None:
+            logger.warning(
+                f"插件 {plugin_id} 在创建安装快照前被取消，无法执行文件补偿"
+            )
+            return
+
+        rollback_task = asyncio.create_task(
+            self._failure(
+                plugin_id=plugin_id,
+                original_plugins=original_plugins,
+                checkpoint=state.checkpoint,
+                stage=state.stage,
+                message="插件安装已取消",
+                package_installed=state.package_installed,
+                installed_list_persisted=state.installed_list_touched,
+                runtime_touched=state.runtime_touched,
+                registrations_touched=state.registrations_touched,
+            )
+        )
+        try:
+            result = await asyncio.shield(rollback_task)
+        except asyncio.CancelledError:
+            try:
+                result = await asyncio.shield(rollback_task)
+            except BaseException as err:
+                logger.error(f"插件 {plugin_id} 取消后的补偿未完成：{err}")
+                return
+        except Exception as err:
+            logger.error(f"插件 {plugin_id} 取消后的补偿失败：{err}")
+            return
+        if result.rollback.errors:
+            logger.error(
+                f"插件 {plugin_id} 取消后的补偿存在错误：{'；'.join(result.rollback.errors)}"
+            )
+        logger.warning(
+            f"插件 {plugin_id} 安装已取消，插件文件已尝试恢复，Python 依赖环境可能已经改变"
+        )
+
    async def _refresh_existing(
        self,
        *,
        plugin_id: str,
        repo_url: Optional[str],
+        state: _InstallState,
    ) -> PluginInstallResult:
        """刷新已存在插件，不触碰包文件和已安装列表。"""
        if repo_url:
@@ -261,33 +386,31 @@ class PluginInstallCommand:
            await self._plugin_reloader(plugin_id)
            failure_stage = "registration_refresh"
            await self._registration_refresher(plugin_id)
-        except Exception as err:
-            rollback_errors = []
-            runtime_restored = False
-            registrations_restored = False
-            try:
-                await self._plugin_reloader(plugin_id)
-                runtime_restored = True
-            except Exception as rollback_err:
-                rollback_errors.append(f"运行态恢复失败：{rollback_err}")
-            if runtime_restored:
+        except asyncio.CancelledError:
+            cleanup_task = asyncio.create_task(
+                self._restore_refreshed_runtime(plugin_id)
+            )
+            while not cleanup_task.done():
                try:
-                    await self._registration_refresher(plugin_id)
-                    registrations_restored = True
-                except Exception as rollback_err:
-                    rollback_errors.append(f"路由和服务注册恢复失败：{rollback_err}")
+                    await asyncio.shield(cleanup_task)
+                except asyncio.CancelledError:
+                    continue
+            rollback = await cleanup_task
+            state.refresh_compensated = True
+            if rollback.errors:
+                logger.error(
+                    f"插件 {plugin_id} 取消刷新后的运行态补偿存在错误："
+                    f"{'；'.join(rollback.errors)}"
+                )
+            raise
+        except Exception as err:
+            rollback = await self._restore_refreshed_runtime(plugin_id)
            result = PluginInstallResult(
                success=False,
                message=f"刷新插件运行态失败：{err}",
                refreshed_only=True,
                failure_stage=failure_stage,
-                rollback=PluginInstallRollback(
-                    runtime_attempted=True,
-                    runtime_restored=runtime_restored,
-                    registrations_attempted=True,
-                    registrations_restored=registrations_restored,
-                    errors=tuple(rollback_errors),
-                ),
+                rollback=rollback,
            )
            if isinstance(err, DatabaseWorkerOverloadedError):
                raise
@@ -316,6 +439,33 @@ class PluginInstallCommand:
            report_error=report_error,
        )

+    async def _restore_refreshed_runtime(
+        self,
+        plugin_id: str,
+    ) -> PluginInstallRollback:
+        """重新加载插件并刷新注册，使中断的运行态切换恢复到完整状态。"""
+        errors = []
+        runtime_restored = False
+        registrations_restored = False
+        try:
+            await self._plugin_reloader(plugin_id)
+            runtime_restored = True
+        except Exception as err:
+            errors.append(f"运行态恢复失败：{err}")
+        if runtime_restored:
+            try:
+                await self._registration_refresher(plugin_id)
+                registrations_restored = True
+            except Exception as err:
+                errors.append(f"路由和服务注册恢复失败：{err}")
+        return PluginInstallRollback(
+            runtime_attempted=True,
+            runtime_restored=runtime_restored,
+            registrations_attempted=True,
+            registrations_restored=registrations_restored,
+            errors=tuple(errors),
+        )
+
    async def _failure(
        self,
        *,
@@ -0,0 +1,71 @@
+"""插件安装与启动同步之间共享的生命周期互斥。"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from contextlib import asynccontextmanager
+
+
+class PluginLifecycleCoordinator:
+    """在事件循环和同步启动线程之间协调插件生命周期操作。"""
+
+    def __init__(self) -> None:
+        self._condition = threading.Condition()
+        self._active_plugins: set[str] = set()
+        self._startup_active = False
+
+    @staticmethod
+    def _normalize(plugin_id: str) -> str:
+        return (plugin_id or "").strip().lower()
+
+    def _try_acquire_plugin(self, plugin_id: str) -> bool:
+        normalized_id = self._normalize(plugin_id)
+        if not normalized_id:
+            raise ValueError("插件ID不能为空")
+        with self._condition:
+            if self._startup_active or normalized_id in self._active_plugins:
+                return False
+            self._active_plugins.add(normalized_id)
+            return True
+
+    def _release_plugin(self, plugin_id: str) -> None:
+        normalized_id = self._normalize(plugin_id)
+        with self._condition:
+            self._active_plugins.discard(normalized_id)
+            self._condition.notify_all()
+
+    def _try_acquire_startup(self) -> bool:
+        with self._condition:
+            if self._startup_active or self._active_plugins:
+                return False
+            self._startup_active = True
+            return True
+
+    def _release_startup(self) -> None:
+        with self._condition:
+            self._startup_active = False
+            self._condition.notify_all()
+
+    @asynccontextmanager
+    async def hold(self, plugin_id: str):
+        """异步持有单个插件的生命周期资格，不在线程池中等待锁。"""
+        while not self._try_acquire_plugin(plugin_id):
+            await asyncio.sleep(0.01)
+        try:
+            yield
+        finally:
+            self._release_plugin(plugin_id)
+
+    @asynccontextmanager
+    async def hold_startup(self):
+        """异步持有启动同步的全局资格，阻止安装请求穿过启动收口。"""
+        while not self._try_acquire_startup():
+            await asyncio.sleep(0.01)
+        try:
+            yield
+        finally:
+            self._release_startup()
+
+
+plugin_lifecycle = PluginLifecycleCoordinator()
@@ -10,7 +10,7 @@ from contextvars import copy_context
 from dataclasses import dataclass
 from typing import Callable, TypeVar

-from app.application.database import (
+from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
 )
@@ -14,7 +14,7 @@ from app.adapters.observability.otel import build_observation_port
 from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry
 from app.adapters.web.health import install_health_routes
 from app.application.plugin.routes import configure_plugin_routes
-from app.application.database import (
+from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
 )
@@ -63,6 +63,27 @@ class PluginDependencyService:
        """安装当前环境缺失的插件依赖并保持历史列表返回合同。"""
        return self.install_missing_with_status().missing

+    async def async_install_missing_with_status(self) -> PluginDependencyInstallResult:
+        """在异步启动链中恢复缺失依赖，确保安装子进程可取消。"""
+        installer = self._system().dependency
+        missing = await installer.async_find_missing()
+        if not missing:
+            return PluginDependencyInstallResult(missing=[], success=True)
+        self._logger.debug(f"检测到缺失的依赖项: {missing}")
+        self._logger.info(f"开始安装缺失的依赖项，共 {len(missing)} 个...")
+        started = time.time()
+        success, _message = await installer.async_install(missing)
+        elapsed = time.time() - started
+        if success:
+            self._logger.info(
+                f"已完成 {len(missing)} 个依赖项安装，总耗时：{elapsed:.2f} 秒"
+            )
+        else:
+            self._logger.warning(
+                f"存在缺失依赖项安装失败，请尝试手动安装，总耗时：{elapsed:.2f} 秒"
+            )
+        return PluginDependencyInstallResult(missing=missing, success=success)
+
    def classify_plugins(self) -> PluginDependencyClassification:
        """返回启动编排使用的轻量插件分类。"""
        ready, missing_dependencies, missing_source = (
@@ -603,6 +603,14 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton):
            log=logger,
        ).install_missing_with_status()

+    @staticmethod
+    async def async_install_plugin_missing_dependencies_with_status() -> PluginDependencyInstallResult:
+        """在异步启动链中恢复插件依赖并保留取消语义。"""
+        return await PluginDependencyService(
+            system=get_plugin_system,
+            log=logger,
+        ).async_install_missing_with_status()
+
    def classify_plugins(self) -> PluginDependencyClassification:
        """按源码依赖状态分类物理插件，并把结果映射到虚拟实例。"""
        source_classification = PluginDependencyService(
@@ -1,3 +1,14 @@
+__all__ = (
+    "ImmediateException",
+    "LimitException",
+    "APIRateLimitException",
+    "RateLimitExceededException",
+    "OperationInterrupted",
+    "StorageQueryError",
+    "TMDbException",
+)
+
+
 class ImmediateException(Exception):
    """
    用于立即抛出异常而不重试的特殊异常类。
@@ -47,6 +58,14 @@ class StorageQueryError(Exception):
    pass


+class DatabaseWorkerClosedError(RuntimeError):
+    """数据库执行器尚未启动或已经停止。"""
+
+
+class DatabaseWorkerOverloadedError(RuntimeError):
+    """数据库执行器的运行与排队容量已经用尽。"""
+
+
 class TMDbException(Exception):
    """
    用于表示TheMovieDB数据源请求失败的跨层异常契约。
@@ -25,6 +25,7 @@ except Exception:
    pass

 from app.chain.system import SystemChain
+from app.application.plugin.lifecycle import plugin_lifecycle
 from app.application.plugin.runtime import get_plugin_manager
 from app.runtime.config import global_vars
 from app.runtime.settings import RuntimeSettingsCompat
@@ -76,13 +77,14 @@ async def init_extra():
        return
    plugin_manager = get_plugin_manager()
    try:
-        if await sync_plugins():
-            await execute_task(
-                global_vars.loop,
-                init_plugin_scheduler,
-                "插件定时服务刷新",
-            )
-            await asyncio.wrap_future(restart_command())
+        async with plugin_lifecycle.hold_startup():
+            if await sync_plugins():
+                await execute_task(
+                    global_vars.loop,
+                    init_plugin_scheduler,
+                    "插件定时服务刷新",
+                )
+                await asyncio.wrap_future(restart_command())
    finally:
        plugin_manager.set_plugin_settling(False)
        plugin_manager.start_monitor()
@@ -148,10 +148,8 @@ async def sync_plugins() -> bool:
        plugin_manager.set_plugin_settling(True)

        sync_result = await execute_task(loop, plugin_manager.sync, "插件同步到本地")
-        dependency_result = await execute_task(
-            loop,
-            plugin_manager.install_plugin_missing_dependencies_with_status,
-            "缺失依赖项安装",
+        dependency_result = await (
+            plugin_manager.async_install_plugin_missing_dependencies_with_status()
        )
        if dependency_result is None:
            return False
@@ -13,8 +13,8 @@
    "runtime_to_db": [],
    "workflow_to_db": []
  },
-  "edge_count": 6444,
-  "edge_sha256": "b5db7b31c7ea4dd7311fcd9e11a49eb32feb6939f896ed85703b3573408752a8",
+  "edge_count": 6451,
+  "edge_sha256": "9075d8717e384580cd6a41bc36685438770db4a8d8f18c57e6c494f32937113a",
  "edges": [
    "app -> app.runtime",
    "app -> app.runtime.compat",
@@ -2056,7 +2056,6 @@
    "app.api.endpoints.plugin -> app.application",
    "app.api.endpoints.plugin -> app.application.commands",
    "app.api.endpoints.plugin -> app.application.configuration",
-    "app.api.endpoints.plugin -> app.application.database",
    "app.api.endpoints.plugin -> app.application.plugin",
    "app.api.endpoints.plugin -> app.application.plugin.config",
    "app.api.endpoints.plugin -> app.application.plugin.folders",
@@ -2072,6 +2071,7 @@
    "app.api.endpoints.plugin -> app.runtime.log",
    "app.api.endpoints.plugin -> app.schemas",
    "app.api.endpoints.plugin -> app.schemas.common",
+    "app.api.endpoints.plugin -> app.schemas.exception",
    "app.api.endpoints.plugin -> app.schemas.plugin",
    "app.api.endpoints.plugin -> app.schemas.response",
    "app.api.endpoints.plugin -> app.schemas.token",
@@ -2563,6 +2563,7 @@
    "app.application.messaging.chat -> app.runtime.observability",
    "app.application.messaging.chat -> app.schemas",
    "app.application.messaging.chat -> app.schemas.agent",
+    "app.application.messaging.chat -> app.schemas.exception",
    "app.application.messaging.interaction -> app.schemas",
    "app.application.messaging.interaction -> app.schemas.message",
    "app.application.messaging.interaction -> app.schemas.notification",
@@ -2649,7 +2650,12 @@
    "app.application.plugin.folders -> app.schemas",
    "app.application.plugin.folders -> app.schemas.types",
    "app.application.plugin.install -> app.application",
-    "app.application.plugin.install -> app.application.database",
+    "app.application.plugin.install -> app.application.plugin",
+    "app.application.plugin.install -> app.application.plugin.lifecycle",
+    "app.application.plugin.install -> app.runtime",
+    "app.application.plugin.install -> app.runtime.log",
+    "app.application.plugin.install -> app.schemas",
+    "app.application.plugin.install -> app.schemas.exception",
    "app.application.recognition -> app.application",
    "app.application.recognition -> app.application.configuration",
    "app.application.recognition -> app.schemas",
@@ -3679,10 +3685,10 @@
    "app.db.session -> app.runtime.config",
    "app.db.session -> app.runtime.log",
    "app.db.session -> app.runtime.observability",
-    "app.db.worker -> app.application",
-    "app.db.worker -> app.application.database",
    "app.db.worker -> app.runtime",
    "app.db.worker -> app.runtime.observability",
+    "app.db.worker -> app.schemas",
+    "app.db.worker -> app.schemas.exception",
    "app.doctor.checks -> app.adapters",
    "app.doctor.checks -> app.adapters.system",
    "app.doctor.checks -> app.adapters.system.backup",
@@ -3806,7 +3812,6 @@
    "app.factory -> app.api",
    "app.factory -> app.api.response",
    "app.factory -> app.application",
-    "app.factory -> app.application.database",
    "app.factory -> app.application.plugin",
    "app.factory -> app.application.plugin.routes",
    "app.factory -> app.application.security",
@@ -3820,6 +3825,7 @@
    "app.factory -> app.runtime.observability",
    "app.factory -> app.runtime.settings",
    "app.factory -> app.schemas",
+    "app.factory -> app.schemas.exception",
    "app.factory -> app.schemas.mcp",
    "app.factory -> app.schemas.openai",
    "app.factory -> app.schemas.response",
@@ -6039,6 +6045,7 @@
    "app.startup.lifecycle -> app.adapters.network.http",
    "app.startup.lifecycle -> app.application",
    "app.startup.lifecycle -> app.application.plugin",
+    "app.startup.lifecycle -> app.application.plugin.lifecycle",
    "app.startup.lifecycle -> app.application.plugin.runtime",
    "app.startup.lifecycle -> app.chain",
    "app.startup.lifecycle -> app.chain.system",
@@ -6461,7 +6468,7 @@
    "app.workflow.actions.transfer_file -> app.workflow",
    "app.workflow.actions.transfer_file -> app.workflow.actions"
  ],
-  "module_count": 798,
+  "module_count": 799,
  "modules": [
    "app",
    "app.adapters",
@@ -6753,6 +6760,7 @@
    "app.application.plugin.data",
    "app.application.plugin.folders",
    "app.application.plugin.install",
+    "app.application.plugin.lifecycle",
    "app.application.plugin.routes",
    "app.application.plugin.runtime",
    "app.application.recognition",
@@ -11,7 +11,7 @@ from uuid import uuid4
 import pytest
 from sqlalchemy import delete, select

-from app.application.database import (
+from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
 )
@@ -23,7 +23,7 @@ from app.factory import (
    localized_unhandled_exception_handler,
    localized_validation_exception_handler,
 )
-from app.application.database import (
+from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
 )
@@ -6,8 +6,8 @@ from unittest.mock import patch

 import pytest

-from app.db.worker import (
-    DatabaseWorker,
+from app.db.worker import DatabaseWorker
+from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
 )
@@ -1,7 +1,13 @@
+from unittest.mock import AsyncMock
 from types import SimpleNamespace
 from unittest.mock import MagicMock

-from app.runtime.extensions.plugin.dependency import PluginDependencyService
+import pytest
+
+from app.runtime.extensions.plugin.dependency import (
+    PluginDependencyInstallResult,
+    PluginDependencyService,
+)


 def test_install_missing_skips_installer_when_environment_is_satisfied() -> None:
@@ -35,3 +41,25 @@ def test_install_missing_preserves_list_return_contract() -> None:

    assert service.install_missing() == ["demo>=1"]
    installer.install.assert_called_once_with(["demo>=1"])
+
+
+@pytest.mark.asyncio
+async def test_async_install_missing_uses_async_installer() -> None:
+    """异步启动恢复必须调用可取消的依赖安装入口。"""
+    installer = SimpleNamespace(
+        async_find_missing=AsyncMock(return_value=["demo>=1"]),
+        async_install=AsyncMock(return_value=(True, "")),
+    )
+    service = PluginDependencyService(
+        system=lambda: SimpleNamespace(dependency=installer),
+        log=MagicMock(),
+    )
+
+    result = await service.async_install_missing_with_status()
+
+    assert result == PluginDependencyInstallResult(
+        missing=["demo>=1"],
+        success=True,
+    )
+    installer.async_find_missing.assert_awaited_once()
+    installer.async_install.assert_awaited_once_with(["demo>=1"])
@@ -9,7 +9,7 @@ import time
 import zipfile
 from pathlib import Path
 from types import ModuleType, SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import AsyncMock, Mock, patch

 import pytest

@@ -1615,39 +1615,251 @@ demo = { index = "private" }
        assert env["HTTPS_PROXY"] == "http://proxy.example:7890"
        assert "user:pass" not in " ".join(safe_command)

-    def test_async_package_install_runs_in_threadpool(self):
-        """
-        验证异步安装路径会把同步包安装派发到线程池，避免阻塞事件循环。
-        """
+    def test_async_package_install_uses_cancellable_subprocess(self):
+        """异步依赖安装应直接使用可取消的子进程执行器。"""
        try:
            from app.adapters.external.market import PluginHelper
        except ModuleNotFoundError as exc:
            pytest.skip(f"missing dependency: {exc}")

        helper = PluginHelper()
-        requirements_file = Path("/tmp/demo-requirements.txt")
-        find_links_dirs = [Path("/tmp/demo-wheels")]
-        calls = []
+        with tempfile.TemporaryDirectory() as temp_dir:
+            requirements_file = Path(temp_dir) / "demo-requirements.txt"
+            requirements_file.write_text("demo-package\n", encoding="utf-8")
+            find_links_dirs = [Path(temp_dir) / "wheels"]

-        async def run_install():
-            return await helper._PluginHelper__async_install_packages_with_fallback(
-                requirements_file,
-                find_links_dirs
+            async def run_install():
+                return await helper._PluginHelper__async_install_packages_with_fallback(
+                    requirements_file,
+                    find_links_dirs,
+                )
+
+            health = {
+                "uv check": (True, "ok"),
+                "核心依赖导入检查": (True, "ok"),
+            }
+            strategy = Mock(
+                strategy_name="uv:test",
+                command=["uv", "pip", "install"],
+                env={},
+                safe_log_command=["uv", "pip", "install"],
            )

-        async def fake_to_thread(func, *args, **kwargs):
-            calls.append((func, args, kwargs))
-            return True, "ok"
-
-        with patch("app.adapters.external.market.asyncio.to_thread", side_effect=fake_to_thread):
-            success, message = asyncio.run(run_install())
+            with patch.object(
+                    PluginHelper,
+                    "_PluginHelper__get_installed_packages",
+                    return_value={},
+            ), patch.object(
+                    PluginHelper,
+                    "_PluginHelper__get_protected_runtime_packages",
+                    return_value={},
+            ), patch.object(
+                    PluginHelper,
+                    "_PluginHelper__validate_runtime_dependency_conflicts",
+                    return_value=(True, ""),
+            ), patch(
+                    "app.adapters.external.market.build_package_install_strategies",
+                    return_value=[strategy],
+            ), patch.object(
+                    PluginHelper,
+                    "_PluginHelper__async_run_runtime_healthcheck",
+                    side_effect=[health, health],
+            ), patch.object(
+                    PluginHelper,
+                    "_PluginHelper__refresh_import_system",
+            ), patch(
+                    "app.adapters.external.market.SystemUtils.execute_with_subprocess_async",
+                    new=AsyncMock(return_value=(True, "ok")),
+            ) as execute_mock:
+                success, message = asyncio.run(run_install())

        assert success
        assert "ok" == message
-        assert 1 == len(calls)
-        assert helper.install_packages_with_fallback == calls[0][0]
-        assert (requirements_file, find_links_dirs) == calls[0][1]
-        assert {} == calls[0][2]
+        execute_mock.assert_awaited_once()
+        assert execute_mock.await_args.kwargs["timeout"] == (
+            PluginHelper.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT
+        )
+
+    def test_async_package_install_cancellation_closes_full_lifecycle(self, tmp_path):
+        """取消真实安装进程后必须回收进程树、临时约束和安装锁。"""
+        import psutil
+
+        from app.adapters.external.market import PluginHelper
+
+        helper = PluginHelper()
+        requirements_file = tmp_path / "requirements.txt"
+        requirements_file.write_text("demo-package\n", encoding="utf-8")
+        constraints_file = tmp_path / "runtime-constraints.txt"
+        marker = tmp_path / "install-pids"
+        child_code = "import time; time.sleep(60)"
+        install_code = (
+            "from pathlib import Path; import os, subprocess, time; "
+            f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}]); "
+            f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); "
+            "time.sleep(60)"
+        )
+        strategy = Mock(
+            strategy_name="uv:test",
+            command=[sys.executable, "-c", install_code],
+            env=os.environ.copy(),
+            safe_log_command=[sys.executable, "-c", "<install>"],
+        )
+        health = {
+            "uv check": (True, "ok"),
+            "核心依赖导入检查": (True, "ok"),
+        }
+
+        def create_constraints(_protected_packages):
+            constraints_file.write_text("fastapi==0\n", encoding="utf-8")
+            return constraints_file
+
+        async def run_install():
+            task = asyncio.create_task(
+                helper.async_install_packages_with_fallback(requirements_file)
+            )
+            deadline = time.monotonic() + 2
+            while not marker.exists() and time.monotonic() < deadline:
+                await asyncio.sleep(0.01)
+            assert marker.exists()
+
+            pids = [int(value) for value in marker.read_text().split(":")]
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+            assert not constraints_file.exists()
+            assert PluginHelper._package_install_lock.acquire(blocking=False)
+            PluginHelper._package_install_lock.release()
+            for _ in range(100):
+                alive = []
+                for pid in pids:
+                    try:
+                        process = psutil.Process(pid)
+                        if (
+                            process.is_running()
+                            and process.status() != psutil.STATUS_ZOMBIE
+                        ):
+                            alive.append(pid)
+                    except (psutil.Error, OSError):
+                        continue
+                if not alive:
+                    break
+                await asyncio.sleep(0.01)
+            else:
+                pytest.fail(f"安装进程树仍在运行：{alive}")
+
+        with patch.object(
+                PluginHelper,
+                "_PluginHelper__get_installed_packages",
+                return_value={},
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__get_protected_runtime_packages",
+                return_value={"fastapi": "0"},
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__validate_runtime_dependency_conflicts",
+                return_value=(True, ""),
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__create_runtime_constraints_file",
+                side_effect=create_constraints,
+        ), patch(
+                "app.adapters.external.market.build_package_install_strategies",
+                return_value=[strategy],
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__async_run_runtime_healthcheck",
+                new=AsyncMock(return_value=health),
+        ):
+            asyncio.run(run_install())
+
+    def test_constraints_created_during_cancellation_are_removed(self, tmp_path):
+        """约束文件创建线程收口后仍须响应取消并删除临时文件。"""
+        from app.adapters.external.market import PluginHelper
+
+        helper = PluginHelper()
+        requirements_file = tmp_path / "requirements.txt"
+        requirements_file.write_text("demo-package\n", encoding="utf-8")
+        constraints_file = tmp_path / "runtime-constraints.txt"
+        created = threading.Event()
+        release = threading.Event()
+
+        def create_constraints(_protected_packages):
+            constraints_file.write_text("fastapi==0\n", encoding="utf-8")
+            created.set()
+            release.wait(timeout=2)
+            return constraints_file
+
+        async def run_install():
+            task = asyncio.create_task(
+                helper.async_install_packages_with_fallback(requirements_file)
+            )
+            assert await asyncio.to_thread(created.wait, 2)
+            task.cancel()
+            release.set()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        with patch.object(
+                PluginHelper,
+                "_PluginHelper__get_installed_packages",
+                return_value={},
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__get_protected_runtime_packages",
+                return_value={"fastapi": "0"},
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__validate_runtime_dependency_conflicts",
+                return_value=(True, ""),
+        ), patch.object(
+                PluginHelper,
+                "_PluginHelper__create_runtime_constraints_file",
+                side_effect=create_constraints,
+        ):
+            asyncio.run(run_install())
+
+        assert not constraints_file.exists()
+
+    def test_constraints_cleanup_failure_preserves_cancellation(self, tmp_path):
+        """临时文件删除失败只记录日志，不得替换调用方的取消异常。"""
+        from app.adapters.external.market import PluginHelper
+
+        constraints_file = tmp_path / "runtime-constraints.txt"
+        created = threading.Event()
+        release = threading.Event()
+
+        def create_constraints(_protected_packages):
+            constraints_file.write_text("fastapi==0\n", encoding="utf-8")
+            created.set()
+            release.wait(timeout=2)
+            return constraints_file
+
+        async def run_create():
+            task = asyncio.create_task(
+                PluginHelper._PluginHelper__async_create_runtime_constraints_file(
+                    {"fastapi": Version("0")}
+                )
+            )
+            assert await asyncio.to_thread(created.wait, 2)
+            task.cancel()
+            release.set()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        with patch.object(
+                PluginHelper,
+                "_PluginHelper__create_runtime_constraints_file",
+                side_effect=create_constraints,
+        ), patch.object(
+                Path,
+                "unlink",
+                side_effect=PermissionError("locked"),
+        ), patch("app.adapters.external.market.logger.warning") as warning:
+            asyncio.run(run_create())
+
+        warning.assert_called_once()

    def test_install_uses_release_package_when_asset_is_available(self, monkeypatch):
        """
@@ -1,8 +1,9 @@
-from unittest.mock import AsyncMock, Mock
+import asyncio
+from unittest.mock import AsyncMock, Mock, patch

 import pytest

-from app.application.database import DatabaseWorkerOverloadedError
+from app.schemas.exception import DatabaseWorkerOverloadedError
 from app.application.plugin.install import PluginInstallCommand


@@ -155,6 +156,42 @@ async def test_existing_plugin_checks_compatibility_without_reinstalling_package
    checkpointer.assert_not_awaited()


+@pytest.mark.asyncio
+async def test_cancelled_existing_plugin_refresh_restores_runtime_and_registrations():
+    """已存在插件刷新被取消时，必须重新收敛运行态和注册。"""
+    registration_started = asyncio.Event()
+    calls: list[str] = []
+
+    async def reload_plugin(_plugin_id: str) -> None:
+        calls.append("reload")
+
+    async def refresh_registrations(_plugin_id: str) -> None:
+        calls.append("registrations")
+        if calls.count("registrations") == 1:
+            registration_started.set()
+            await asyncio.Event().wait()
+
+    with patch("app.application.plugin.install.logger.warning") as warning:
+        task = asyncio.create_task(
+            _command(
+                installed=["DemoPlugin"],
+                plugin_ids=["DemoPlugin"],
+                reloader=reload_plugin,
+                refresher=refresh_registrations,
+            ).execute(
+                plugin_id="DemoPlugin",
+                repo_url="https://github.com/demo/plugins",
+            )
+        )
+        await registration_started.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    assert calls == ["reload", "registrations", "reload", "registrations"]
+    warning.assert_not_called()
+
+
@pytest.mark.asyncio
 async def test_persistence_failure_restores_package_without_touching_runtime():
    """已安装列表保存失败时恢复文件，且运行态尚未开始切换。"""
@@ -176,12 +213,41 @@ async def test_persistence_failure_restores_package_without_touching_runtime():
    assert result.success is False
    assert result.failure_stage == "installed_list_persistence"
    assert result.rollback.file_restored is True
-    assert result.rollback.installed_list_attempted is False
+    assert result.rollback.installed_list_attempted is True
    assert result.rollback.runtime_attempted is False
    rollback.assert_awaited_once_with(checkpoint)
    reloader.assert_not_awaited()


+@pytest.mark.asyncio
+async def test_persistence_exception_after_write_restores_installed_list():
+    """清单写入已提交后抛异常时，文件和清单必须一起恢复。"""
+    persisted: list[list[str]] = []
+    checkpoint = object()
+    rollback = AsyncMock()
+
+    async def write(plugin_ids: list[str]) -> None:
+        persisted.append(list(plugin_ids))
+        if len(persisted) == 1:
+            raise RuntimeError("write acknowledgement lost")
+
+    result = await _command(
+        checkpointer=AsyncMock(return_value=checkpoint),
+        writer=write,
+        rollback=rollback,
+    ).execute(
+        plugin_id="DemoPlugin",
+        repo_url="https://github.com/demo/plugins",
+    )
+
+    assert result.success is False
+    assert result.failure_stage == "installed_list_persistence"
+    assert result.rollback.installed_list_attempted is True
+    assert result.rollback.installed_list_restored is True
+    assert persisted == [["DemoPlugin"], []]
+    rollback.assert_awaited_once_with(checkpoint)
+
+
@pytest.mark.asyncio
 async def test_database_worker_overload_rolls_back_and_reaches_api_boundary():
    """配置 worker 背压完成补偿后继续抛出，交由 API 映射为 503。"""
@@ -306,6 +372,148 @@ async def test_registration_failure_restores_instance_files_and_routes() -> None
    ]


+@pytest.mark.asyncio
+async def test_same_plugin_install_lifecycle_is_serialized() -> None:
+    """同一插件的两个安装调用不得同时修改包、运行态和注册信息。"""
+    first_started = asyncio.Event()
+    release_first = asyncio.Event()
+    calls: list[str] = []
+
+    async def install(plugin_id, *_args):
+        calls.append(plugin_id)
+        if len(calls) == 1:
+            first_started.set()
+            await release_first.wait()
+        return True, "ok"
+
+    command = _command(installer=install)
+    first = asyncio.create_task(
+        command.execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await first_started.wait()
+    second = asyncio.create_task(
+        command.execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await asyncio.sleep(0.02)
+    assert calls == ["DemoPlugin"]
+
+    release_first.set()
+    results = await asyncio.gather(first, second)
+    assert all(result.success for result in results)
+    assert calls == ["DemoPlugin", "DemoPlugin"]
+
+
+@pytest.mark.asyncio
+async def test_cancelled_install_waits_for_rollback_before_releasing_lifecycle() -> None:
+    """取消安装后先完成包快照补偿，再允许同一插件的新调用进入。"""
+    install_started = asyncio.Event()
+    release_install = asyncio.Event()
+    rollback = AsyncMock()
+
+    async def install(*_args):
+        install_started.set()
+        await release_install.wait()
+        return True, "ok"
+
+    command = _command(installer=install, rollback=rollback)
+    task = asyncio.create_task(
+        command.execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await install_started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    rollback.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_cancelled_persisted_list_is_restored_conservatively() -> None:
+    """清单写入已产生副作用但尚未返回时取消，也必须恢复原清单。"""
+    persisted: list[list[str]] = []
+    writer_started = asyncio.Event()
+    rollback = AsyncMock()
+
+    async def writer(plugin_ids: list[str]) -> None:
+        persisted.append(list(plugin_ids))
+        if len(persisted) == 1:
+            writer_started.set()
+            await asyncio.Event().wait()
+
+    task = asyncio.create_task(
+        _command(writer=writer, rollback=rollback).execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await writer_started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    assert persisted == [["DemoPlugin"], []]
+    rollback.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_cancelled_snapshot_cleanup_does_not_rollback_committed_plugin() -> None:
+    """运行态提交后清理快照期间取消，不得删除已生效插件。"""
+    cleanup_started = asyncio.Event()
+    rollback = AsyncMock()
+
+    async def committer(_checkpoint) -> None:
+        cleanup_started.set()
+        await asyncio.Event().wait()
+
+    task = asyncio.create_task(
+        _command(committer=committer, rollback=rollback).execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await cleanup_started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    rollback.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_startup_lifecycle_lock_blocks_plugin_install_until_settlement() -> None:
+    """启动同步持有全局资格时，插件安装不得穿过启动收口。"""
+    from app.application.plugin.lifecycle import plugin_lifecycle
+
+    entered = asyncio.Event()
+    release = asyncio.Event()
+
+    async def startup_scope():
+        async with plugin_lifecycle.hold_startup():
+            entered.set()
+            await release.wait()
+
+    startup = asyncio.create_task(startup_scope())
+    await entered.wait()
+    plugin_context = plugin_lifecycle.hold("DemoPlugin")
+    plugin_scope = asyncio.create_task(plugin_context.__aenter__())
+    await asyncio.sleep(0.02)
+    assert plugin_scope.done() is False
+
+    release.set()
+    await plugin_scope
+    await plugin_context.__aexit__(None, None, None)
+    await startup
+
+
@pytest.mark.asyncio
 async def test_report_failure_does_not_rollback_completed_local_install():
    """统计上报失败属于非关键副作用，不得撤销已成功的本地安装。"""
@@ -2,7 +2,7 @@ import asyncio
 import threading
 import time
 from types import SimpleNamespace
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock

 import pytest

@@ -139,6 +139,12 @@ def _patch_sync_plugins(monkeypatch, manager: MagicMock) -> MagicMock:
    monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager)
    monkeypatch.setattr(plugins_initializer, "execute_task", execute)
    monkeypatch.setattr(plugins_initializer, "register_plugin_api", register)
+    dependency_result = (
+        manager.async_install_plugin_missing_dependencies_with_status.return_value
+    )
+    manager.async_install_plugin_missing_dependencies_with_status = AsyncMock(
+        return_value=dependency_result,
+    )
    manager.get_plugin_runtime_statuses.return_value = {}
    return register

@@ -150,7 +156,7 @@ async def test_sync_plugins_activates_ready_plugins_when_dependencies_fail(
    """依赖恢复失败时仍激活无关的已就绪插件。"""
    manager = MagicMock()
    manager.sync.return_value = ["demo"]
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
        PluginDependencyInstallResult(missing=["demo>=1"], success=False)
    )
    manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -175,7 +181,7 @@ async def test_sync_plugins_loads_only_plugins_that_become_ready(
    """后台依赖恢复后只启动尚未运行且当前已就绪的插件。"""
    manager = MagicMock()
    manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
        PluginDependencyInstallResult(missing=["demo>=1"], success=True)
    )
    manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -204,7 +210,7 @@ async def test_sync_plugins_reloads_only_updated_running_plugins(monkeypatch) ->
    """源码同步只重载对应运行实例，不重启其他插件。"""
    manager = MagicMock()
    manager.sync.return_value = ["UpdatedPlugin"]
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
        PluginDependencyInstallResult(missing=[], success=True)
    )
    manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -232,7 +238,7 @@ async def test_sync_plugins_reloads_running_plugin_after_dependency_recovery(
    """依赖恢复后，已运行的旧实例必须切换到新源码。"""
    manager = MagicMock()
    manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
        PluginDependencyInstallResult(missing=["demo>=1"], success=True)
    )
    manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -258,7 +264,7 @@ async def test_sync_plugins_keeps_runtime_when_nothing_changed(monkeypatch) -> N
    """源码和依赖均无变化时保留首次初始化结果。"""
    manager = MagicMock()
    manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
        PluginDependencyInstallResult(missing=[], success=True)
    )
    manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -283,7 +289,7 @@ async def test_sync_plugins_keeps_event_loop_responsive_during_activation(
    """插件初始化运行在线程池时，Web 事件循环仍可继续调度。"""
    manager = MagicMock()
    manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
        PluginDependencyInstallResult(missing=[], success=True)
    )
    manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -299,6 +305,9 @@ async def test_sync_plugins_keeps_event_loop_responsive_during_activation(
        time.sleep(0.1)

    manager.start.side_effect = slow_start
+    manager.async_install_plugin_missing_dependencies_with_status = AsyncMock(
+        return_value=PluginDependencyInstallResult(missing=[], success=True),
+    )
    monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None)
    monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager)
    monkeypatch.setattr(plugins_initializer, "register_plugin_api", MagicMock())
@@ -1,7 +1,10 @@
+import shutil
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import Mock

+import pytest
+
 from app.adapters.system.plugin.package import PluginPackageManager


@@ -46,6 +49,23 @@ def test_checkpoint_rollback_removes_new_package(monkeypatch, tmp_path):
    assert not checkpoint.transaction_dir.exists()


+def test_rollback_does_not_delete_package_when_snapshot_is_missing(monkeypatch, tmp_path):
+    """补偿快照损坏时先失败，不能先删除当前可用插件。"""
+    manager = _manager(monkeypatch, tmp_path)
+    plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
+    plugin_dir.mkdir(parents=True)
+    (plugin_dir / "__init__.py").write_text("old", encoding="utf-8")
+
+    checkpoint = manager.checkpoint("DemoPlugin")
+    shutil.rmtree(checkpoint.transaction_dir / "package")
+    (plugin_dir / "__init__.py").write_text("new", encoding="utf-8")
+
+    with pytest.raises(FileNotFoundError):
+        manager.rollback(checkpoint)
+
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "new"
+
+
 def test_local_sync_failure_restores_previous_runtime_copy(monkeypatch, tmp_path):
    """本地来源不可复制时不得丢失已经运行的插件副本。"""
    manager = _manager(monkeypatch, tmp_path)
@@ -1,13 +1,17 @@
+import asyncio
 import errno
 import itertools
 import os
 import struct
 import subprocess
+import sys
 import tempfile
+import time
 from pathlib import Path
 from unittest import TestCase
 from unittest.mock import MagicMock, call, patch

+import psutil
 import pytest

 from app.runtime.state import SystemHelper
@@ -155,6 +159,148 @@ def test_execute_with_subprocess_uses_safe_command_in_failure_message():
    assert run_mock.call_args.args[0] == command


+@pytest.mark.asyncio
+async def test_async_subprocess_timeout_reaps_process():
+    """异步安装命令超时后应终止并回收子进程。"""
+    success, message = await SystemUtils.execute_with_subprocess_async(
+        [sys.executable, "-c", "import time; time.sleep(60)"],
+        timeout=0.05,
+    )
+
+    assert success is False
+    assert "执行超时" in message
+
+
+@pytest.mark.asyncio
+async def test_async_subprocess_cancellation_reaps_process(tmp_path):
+    """调用方取消安装任务时，底层子进程不得继续运行。"""
+    marker = tmp_path / "pid"
+    command = [
+        sys.executable,
+        "-c",
+        (
+            "from pathlib import Path; import os, time; "
+            f"Path({str(marker)!r}).write_text(str(os.getpid())); time.sleep(60)"
+        ),
+    ]
+    task = asyncio.create_task(
+        SystemUtils.execute_with_subprocess_async(command, timeout=30)
+    )
+    deadline = time.monotonic() + 2
+    while not marker.exists() and time.monotonic() < deadline:
+        await asyncio.sleep(0.01)
+    assert marker.exists()
+
+    pid = int(marker.read_text())
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    for _ in range(100):
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            break
+        await asyncio.sleep(0.01)
+    else:
+        pytest.fail(f"子进程仍在运行：{pid}")
+
+
+@pytest.mark.asyncio
+async def test_async_subprocess_cancellation_reaps_process_tree(tmp_path):
+    """取消安装命令时，子进程派生的构建进程也不得继续运行。"""
+    marker = tmp_path / "pids"
+    child_code = "import time; time.sleep(60)"
+    command = [
+        sys.executable,
+        "-c",
+        (
+            "from pathlib import Path; import subprocess, os, time; "
+            f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}]); "
+            f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); "
+            "time.sleep(60)"
+        ),
+    ]
+    task = asyncio.create_task(
+        SystemUtils.execute_with_subprocess_async(command, timeout=30)
+    )
+    deadline = time.monotonic() + 2
+    while not marker.exists() and time.monotonic() < deadline:
+        await asyncio.sleep(0.01)
+    assert marker.exists()
+
+    pids = [int(value) for value in marker.read_text().split(":")]
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    for _ in range(100):
+        alive = []
+        for pid in pids:
+            try:
+                process = psutil.Process(pid)
+                if not process.is_running() or process.status() == psutil.STATUS_ZOMBIE:
+                    continue
+            except (psutil.Error, OSError):
+                continue
+            alive.append(pid)
+        if not alive:
+            break
+        await asyncio.sleep(0.01)
+    else:
+        pytest.fail(f"进程树仍在运行：{alive}")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Windows 没有 POSIX 进程组信号语义")
+@pytest.mark.asyncio
+async def test_async_subprocess_reaps_descendant_after_early_pipe_close(tmp_path):
+    """父进程关闭管道后，忽略终止信号的后代也必须被强制回收。"""
+    marker = tmp_path / "pids"
+    child_code = (
+        "import os, signal, time; os.close(1); os.close(2); "
+        "signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)"
+    )
+    command = [
+        sys.executable,
+        "-c",
+        (
+            "from pathlib import Path; import os, signal, subprocess, time; "
+            f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}], "
+            "start_new_session=True); "
+            f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); "
+            "signal.signal(signal.SIGTERM, lambda *_: os._exit(0)); time.sleep(60)"
+        ),
+    ]
+    task = asyncio.create_task(
+        SystemUtils.execute_with_subprocess_async(command, timeout=30)
+    )
+    deadline = time.monotonic() + 2
+    while not marker.exists() and time.monotonic() < deadline:
+        await asyncio.sleep(0.01)
+    assert marker.exists()
+
+    pids = [int(value) for value in marker.read_text().split(":")]
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    for _ in range(100):
+        alive = []
+        for pid in pids:
+            try:
+                process = psutil.Process(pid)
+                if not process.is_running() or process.status() == psutil.STATUS_ZOMBIE:
+                    continue
+            except (psutil.Error, OSError):
+                continue
+            alive.append(pid)
+        if not alive:
+            break
+        await asyncio.sleep(0.01)
+    else:
+        pytest.fail(f"通信已结束但进程树仍在运行：{alive}")
+
+
 def test_execute_with_subprocess_redacts_userinfo_from_stdout_and_stderr():
    error = subprocess.CalledProcessError(
        returncode=1,