Merge remote-tracking branch 'origin/v3' into v3

# Conflicts: # tests/fixtures/architecture/dependency-baseline.json
2026-09-04 23:17:20 +08:00 · 2026-08-23 14:57:03 +08:00
parent d814c24169 1f64e99a6f
commit 0459b6a662
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
 from app.api.context import get_background_task_registry, resolve_background_task_registry
@@ -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
@@ -77,13 +78,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