fix(plugin): make installation lifecycle cancellable

app/adapters/external/market.py

@@ -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] = {}
@@ -2421,18 +2435,279 @@ 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 repair_target.exists():
+            repair_target = None
+        if repair_target is None:
+            repair_target = settings.ROOT_PATH / "pyproject.toml"
+            repair_desc = "主程序 uv.lock"
+        if not repair_target.exists():
+            return False, f"恢复依赖文件不存在：{repair_target}"
+        if snapshot_file is None and not (settings.ROOT_PATH / "uv.lock").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 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 candidate_path.is_dir():
+                continue
+            candidate_key = str(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 _await_thread_operation(
+                    cls.__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:
+                constraints_file.unlink(missing_ok=True)
 
     async def __async_backup_plugin(self, pid: str) -> str:
         """
@@ -2444,14 +2719,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 +2835,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 +2945,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]:

app/adapters/system/host.py

@@ -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,136 @@ 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:
+        """终止安装子进程及其同组子进程，并确保句柄已回收。"""
+        if process.returncode is None:
+            try:
+                if os.name == "nt":
+                    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):
+            try:
+                if os.name == "nt":
+                    process.kill()
+                else:
+                    os.killpg(process.pid, signal.SIGKILL)
+            except (ProcessLookupError, OSError):
+                pass
+            try:
+                await asyncio.wait_for(
+                    asyncio.shield(communication_task),
+                    timeout=grace_seconds,
+                )
+            except (asyncio.TimeoutError, asyncio.CancelledError):
+                communication_task.cancel()
+                await asyncio.gather(communication_task, return_exceptions=True)
+        finally:
+            try:
+                await process.wait()
+            except (ProcessLookupError, OSError):
+                pass
+
+    @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:
         """

app/adapters/system/plugin/dependency.py

@@ -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}"

app/adapters/system/plugin/package.py

@@ -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,7 +96,7 @@ 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:
@@ -101,7 +114,7 @@ class PluginPackageManager:
 
     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,

app/application/plugin/install.py

@@ -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.application.plugin.lifecycle import plugin_lifecycle
+from app.runtime.log import logger
 
 
 InstalledPluginsReader = Callable[[], list[str]]
@@ -58,6 +61,20 @@ class PluginInstallResult:
     rollback: PluginInstallRollback = field(default_factory=PluginInstallRollback)
 
 
+@dataclass
+class _InstallState:
+    """记录取消补偿所需的事务阶段。"""
+
+    checkpoint: Any = None
+    stage: str = "package_checkpoint"
+    package_installed: bool = False
+    installed_list_persisted: bool = False
+    runtime_touched: bool = False
+    registrations_touched: bool = False
+    committed: bool = False
+    original_plugins: list[str] = field(default_factory=list)
+
+
 class PluginInstallCommand:
     """协调插件检查、包事务、持久化、运行态刷新和安装上报。"""
 
@@ -96,9 +113,38 @@ 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(
@@ -112,8 +158,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 +175,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 +196,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,
@@ -156,6 +212,7 @@ class PluginInstallCommand:
             try:
                 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,
@@ -169,6 +226,8 @@ class PluginInstallCommand:
                     raise
                 return result
 
+        state.stage = "runtime_reload"
+        state.runtime_touched = True
         try:
             await self._plugin_reloader(plugin_id)
         except Exception as err:
@@ -186,6 +245,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,13 +266,16 @@ class PluginInstallCommand:
             return result
 
         checkpoint_cleanup_error = ""
+        state.stage = "checkpoint_commit"
         try:
             await self._package_committer(checkpoint)
+            state.committed = True
         except Exception as err:
             checkpoint_cleanup_error = str(err)
 
         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,6 +301,57 @@ 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.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_persisted,
+                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,
         *,

app/application/plugin/lifecycle.py

@@ -0,0 +1,88 @@
+"""插件安装与启动同步之间共享的生命周期互斥。"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from contextlib import asynccontextmanager, contextmanager
+from typing import Iterator
+
+
+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)
+
+    @contextmanager
+    def hold_sync(self, plugin_id: str) -> Iterator[None]:
+        """同步持有单个插件的生命周期资格。"""
+        normalized_id = self._normalize(plugin_id)
+        if not normalized_id:
+            raise ValueError("插件ID不能为空")
+        with self._condition:
+            while self._startup_active or normalized_id in self._active_plugins:
+                self._condition.wait()
+            self._active_plugins.add(normalized_id)
+        try:
+            yield
+        finally:
+            self._release_plugin(normalized_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()
+

app/startup/lifecycle/__init__.py

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