fix(plugin): 插件安装改为暂存后原子换入，避免下载失败删掉已装插件 (#6678)

app/adapters/system/plugin/package.py

@@ -993,11 +993,12 @@ class PluginPackageManager:
 
         release_tag = selection.release_tag
         if release_tag and not selection.fallback_to_filelist:
-            def prepare_selected_release() -> tuple[bool, str]:
+            def prepare_selected_release(staging_dir: Path) -> tuple[bool, str]:
                 return self.__install_from_release(
                     pid,
                     selection.user_repo,
                     release_tag,
+                    staging_dir,
                 )
 
             return self.__install_flow_sync(
@@ -1010,20 +1011,24 @@ class PluginPackageManager:
 
         if release_tag:
             # 当前索引 Release 失败时回退文件列表，避免发布产物短暂滞后阻断安装。
-            def prepare_release() -> tuple[bool, str]:
+            def prepare_release(staging_dir: Path) -> tuple[bool, str]:
                 ok, msg = self.__install_from_release(
                     pid,
                     selection.user_repo,
                     release_tag,
+                    staging_dir,
                 )
                 if ok:
                     return True, msg
                 logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}")
-                self.__remove_old_plugin(pid)
+                # 回退前只清理本次暂存的半份 Release 产物；已安装的运行目录
+                # 此刻还没有被触碰，清理它会让一次可恢复的回退变成数据丢失。
+                shutil.rmtree(staging_dir, ignore_errors=True)
                 return self.__prepare_content_via_filelist_sync(
                     pid,
                     selection.user_repo,
                     selection.package_version,
+                    staging_dir,
                 )
 
             return self.__install_flow_sync(
@@ -1035,11 +1040,12 @@ class PluginPackageManager:
             )
         # 未声明 release 打包的插件继续使用文件列表方式安装。
 
-        def prepare_filelist() -> tuple[bool, str]:
+        def prepare_filelist(staging_dir: Path) -> tuple[bool, str]:
             return self.__prepare_content_via_filelist_sync(
                 pid,
                 selection.user_repo,
                 selection.package_version,
+                staging_dir,
             )
 
         return self.__install_flow_sync(
@@ -1126,11 +1132,11 @@ class PluginPackageManager:
         except Exception:
             return False, "本地插件来源路径无效"
 
-        def prepare_local() -> tuple[bool, str]:
+        def prepare_local(staging_dir: Path) -> tuple[bool, str]:
             try:
                 shutil.copytree(
                     source_dir,
-                    dest_dir,
+                    staging_dir,
                     dirs_exist_ok=True,
                     ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store", "node_modules")
                 )
@@ -1206,8 +1212,9 @@ class PluginPackageManager:
         pid: str,
         remote_path: object,
         package_version: Optional[str],
+        dest_root: Path,
     ) -> Path:
-        """把市场文件路径限定到当前插件目录，拒绝绝对路径和目录穿越。"""
+        """把市场文件路径限定到给定落盘根目录之内，拒绝绝对路径和目录穿越。"""
         if not isinstance(remote_path, str) or not remote_path or "\\" in remote_path:
             raise ValueError("插件文件路径无效")
         pure_path = PurePosixPath(remote_path)
@@ -1222,24 +1229,30 @@ class PluginPackageManager:
             or any(part in {"", ".", ".."} for part in parts[2:])
         ):
             raise ValueError("插件文件路径无效")
-        plugin_dir = (self._plugins_root() / pid.lower()).resolve()
-        file_path = (plugin_dir / Path(*parts[2:])).resolve()
-        if not file_path.is_relative_to(plugin_dir):
+        resolved_root = dest_root.resolve()
+        file_path = (resolved_root / Path(*parts[2:])).resolve()
+        if not file_path.is_relative_to(resolved_root):
             raise ValueError("插件文件路径无效")
         return file_path
 
     def __download_files(self, pid: str, file_list: list[dict[str, Any]], user_repo: str,
-                         package_version: Optional[str] = None) -> tuple[bool, str]:
+                         package_version: Optional[str] = None,
+                         dest_root: Optional[Path] = None) -> tuple[bool, str]:
         """
         下载插件文件
         :param pid: 插件 ID
         :param file_list: 要下载的文件列表，包含文件的元数据（包括下载链接）
         :param user_repo: GitHub 仓库的 user/repo 路径
+        :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录
         :return: (是否成功, 错误信息)
         """
         if not file_list:
             return False, "文件列表为空"
 
+        download_root = (
+            dest_root if dest_root is not None
+            else self._plugins_root() / pid.lower()
+        )
         # 使用栈结构来替代递归调用，避免递归深度过大问题
         stack = [(pid, file_list)]
 
@@ -1254,6 +1267,7 @@ class PluginPackageManager:
                             pid,
                             item.get("path"),
                             package_version,
+                            download_root,
                         )
                     except ValueError as error:
                         return False, str(error)
@@ -1426,56 +1440,152 @@ class PluginPackageManager:
             logger.error(f"获取插件 {pid} 元数据失败：{e}")
             return {}
 
+    def __new_install_staging_dir(self, pid: str) -> Path:
+        """分配一个全新的安装暂存目录，让新内容在触碰运行目录前先完整就位。"""
+        return (
+            Path(get_runtime_setting('TEMP_PATH'))
+            / "plugin_install_staging"
+            / f"{pid.lower()}-{uuid.uuid4().hex}"
+        )
+
+    @staticmethod
+    def __swap_staged_plugin_content(staging_dir: Path, final_dir: Path) -> None:
+        """把已就位的暂存内容换入插件运行目录，失败时运行目录保持换入前的内容。
+
+        手法与运行目录补偿替换一致（见 __restore_tree）：先把旧目录改名挪到同级
+        隐藏位置留作回滚材料，再把暂存目录改名落位，确认落位后才删除旧目录。同一
+        文件系统内两次改名都是原子的，因此进程在任意时刻被杀，磁盘上要么是旧内容
+        要么是新内容，唯一的中间态是"运行目录短暂缺失、旧内容完整躺在 previous
+        目录里"，仍可人工恢复，不会出现任何一份内容被删到无处可取的时点。
+
+        两个改名都可能因跨文件系统失败：overlayfs 会拒绝把镜像层的插件目录直接改
+        名到可写层，暂存目录又常落在独立的临时分区。两处都退化为复制，代价是复制
+        期间可能中途失败留下半份目录，因此复制失败一律先删掉半成品再把旧目录换
+        回；换回本身也失败时只记录旧目录的保留位置、不吞掉原始异常，避免出现
+        "看起来只是安装失败"实则运行目录已空的假象。
+
+        :param staging_dir: 已就位的待安装内容目录
+        :param final_dir: 插件运行目录，可能已存在旧内容
+        :raise OSError: 换入失败，且已尽力把运行目录恢复或保留为换入前的内容
+        """
+        final_dir.parent.mkdir(parents=True, exist_ok=True)
+        previous = final_dir.parent / f".{final_dir.name}.previous-{uuid.uuid4().hex}"
+        previous_available = False
+        cleared = False
+        published = False
+        try:
+            if final_dir.exists():
+                try:
+                    final_dir.replace(previous)
+                except OSError as error:
+                    if error.errno != errno.EXDEV:
+                        raise
+                    shutil.copytree(final_dir, previous, symlinks=True)
+                    previous_available = True
+                    PluginPackageManager.__remove_snapshot_path(final_dir)
+                else:
+                    previous_available = True
+            # 旧内容已挪开或本来就不存在，此后才允许重建运行目录
+            cleared = True
+            try:
+                staging_dir.replace(final_dir)
+            except OSError as error:
+                if error.errno != errno.EXDEV:
+                    raise
+                logger.debug(
+                    f"插件安装内容跨文件系统无法改名换入，退化为复制：{staging_dir}"
+                )
+                shutil.copytree(staging_dir, final_dir, symlinks=True)
+            published = True
+        except Exception:
+            if cleared and not published:
+                try:
+                    PluginPackageManager.__remove_snapshot_path(final_dir)
+                    if previous_available:
+                        previous.replace(final_dir)
+                        previous_available = False
+                except Exception as rollback_error:
+                    logger.error(
+                        f"插件安装换入失败后恢复旧目录失败，已保留恢复材料 {previous}: "
+                        f"{rollback_error}"
+                    )
+            raise
+        finally:
+            if previous.exists() and (published or not previous_available):
+                shutil.rmtree(previous, ignore_errors=True)
+
+    async def __async_swap_staged_plugin_content(
+        self, staging_dir: Path, final_dir: Path
+    ) -> None:
+        """在线程中执行换入，避免目录改名与复制阻塞事件循环，语义与同步换入一致。"""
+        await _await_thread_operation(
+            self.__swap_staged_plugin_content, staging_dir, final_dir
+        )
+
     def __install_flow_sync(
         self,
         pid: str,
         force_install: bool,
-        prepare_content: Callable[[], tuple[bool, str]],
+        prepare_content: Callable[[Path], tuple[bool, str]],
         repo_url: Optional[str] = None,
         before_dependency_install: Optional[Callable[[], None]] = None,
     ) -> tuple[bool, str]:
         """
-        同步安装统一流程：备份→清理→准备内容→安装依赖→上报
-        prepare_content 负责把插件文件放到 app/plugins/{pid}
+        同步安装统一流程：暂存内容→备份→换入→安装依赖→上报
+        prepare_content 负责把插件文件放到调用时给定的暂存目录；只有新内容在暂存
+        目录里完整就位后才会触碰运行目录，因此下载或解压失败时已装插件原样保留。
         """
-        backup_dir = None
-        if not force_install:
-            backup_dir = self.__backup_plugin(pid)
+        plugin_dir = self.__plugin_dir(pid)
+        staging_dir = self.__new_install_staging_dir(pid)
+        try:
+            success, message = prepare_content(staging_dir)
+            if not success:
+                # 运行目录此刻尚未被触碰，已装插件天然完整，无需还原也无需清理
+                logger.error(f"{pid} 准备插件内容失败：{message}")
+                return False, message
 
-        self.__remove_old_plugin(pid)
+            backup_dir = None
+            if not force_install:
+                backup_dir = self.__backup_plugin(pid)
 
-        success, message = prepare_content()
-        if not success:
-            logger.error(f"{pid} 准备插件内容失败：{message}")
-            if backup_dir:
-                self.__restore_plugin(pid, backup_dir)
-                logger.warn(f"{pid} 插件安装失败，已还原备份插件")
-            else:
-                self.__remove_old_plugin(pid)
-                logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
-            return False, message
+            try:
+                self.__swap_staged_plugin_content(staging_dir, plugin_dir)
+            except OSError as error:
+                message = f"写入插件内容失败：{error}"
+                logger.error(f"{pid} {message}")
+                # 换入函数已保证失败时运行目录保持换入前的内容；只有连恢复都失败、
+                # 运行目录确实缺失时，才动用本次的临时备份兜底。
+                if backup_dir and not plugin_dir.exists():
+                    self.__restore_plugin(pid, backup_dir)
+                    logger.warn(f"{pid} 插件安装失败，已还原备份插件")
+                elif backup_dir:
+                    shutil.rmtree(backup_dir, ignore_errors=True)
+                return False, message
 
-        dependencies_exist, dep_ok, dep_msg = (
-            self.__install_dependencies_if_required(
-                pid,
-                before_dependency_install,
+            dependencies_exist, dep_ok, dep_msg = (
+                self.__install_dependencies_if_required(
+                    pid,
+                    before_dependency_install,
+                )
+                if before_dependency_install is not None
+                else self.__install_dependencies_if_required(pid)
             )
-            if before_dependency_install is not None
-            else self.__install_dependencies_if_required(pid)
-        )
-        if dependencies_exist and not dep_ok:
-            logger.error(f"{pid} 依赖安装失败：{dep_msg}")
-            if backup_dir:
-                self.__restore_plugin(pid, backup_dir)
-                logger.warn(f"{pid} 插件安装失败，已还原备份插件")
-            else:
-                self.__remove_old_plugin(pid)
-                logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
-            return False, dep_msg
+            if dependencies_exist and not dep_ok:
+                logger.error(f"{pid} 依赖安装失败：{dep_msg}")
+                if backup_dir:
+                    self.__restore_plugin(pid, backup_dir)
+                    logger.warn(f"{pid} 插件安装失败，已还原备份插件")
+                else:
+                    self.__remove_old_plugin(pid)
+                    logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
+                return False, dep_msg
 
-        if backup_dir:
-            shutil.rmtree(backup_dir, ignore_errors=True)
-        return True, ""
+            if backup_dir:
+                shutil.rmtree(backup_dir, ignore_errors=True)
+            return True, ""
+        finally:
+            if staging_dir.exists():
+                shutil.rmtree(staging_dir, ignore_errors=True)
 
     @staticmethod
     def __validate_release_zip_name(name: str) -> None:
@@ -1566,11 +1676,12 @@ class PluginPackageManager:
             targets.append((info, dest_path, info.is_dir()))
         return targets
 
-    def __install_from_release(self, pid: str, user_repo: str, release_tag: str) -> tuple[bool, str]:
+    def __install_from_release(self, pid: str, user_repo: str, release_tag: str,
+                               dest_root: Optional[Path] = None) -> tuple[bool, str]:
         """
         通过 GitHub Release 资产文件安装插件。
         规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件；
-        将其全部解压到 app/plugins/{pid}
+        将其全部解压到 dest_root，未指定时直接解压到插件运行目录。
         """
         # 拼接资产文件名
         asset_name = f"{release_tag.lower()}.zip"
@@ -1616,7 +1727,10 @@ class PluginPackageManager:
                 infos = zf.infolist()
                 if not infos:
                     return False, "压缩包内容为空"
-                dest_base = self._plugins_root() / pid.lower()
+                dest_base = (
+                    dest_root if dest_root is not None
+                    else self._plugins_root() / pid.lower()
+                )
                 targets = self.__iter_release_zip_targets(zf, dest_base)
                 wrote_any = False
                 for info, dest_path, is_dir in targets:
@@ -1671,17 +1785,23 @@ class PluginPackageManager:
             return None, "插件数据解析失败"
 
     async def __async_download_files(self, pid: str, file_list: list[dict[str, Any]], user_repo: str,
-                                     package_version: Optional[str] = None) -> tuple[bool, str]:
+                                     package_version: Optional[str] = None,
+                                     dest_root: Optional[Path] = None) -> tuple[bool, str]:
         """
         异步下载插件文件
         :param pid: 插件 ID
         :param file_list: 要下载的文件列表，包含文件的元数据（包括下载链接）
         :param user_repo: GitHub 仓库的 user/repo 路径
+        :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录
         :return: (是否成功, 错误信息)
         """
         if not file_list:
             return False, "文件列表为空"
 
+        download_root = (
+            dest_root if dest_root is not None
+            else self._plugins_root() / pid.lower()
+        )
         # 使用栈结构来替代递归调用，避免递归深度过大问题
         stack = [(pid, file_list)]
 
@@ -1696,6 +1816,7 @@ class PluginPackageManager:
                             pid,
                             item.get("path"),
                             package_version,
+                            download_root,
                         )
                     except ValueError as error:
                         return False, str(error)
@@ -1894,11 +2015,12 @@ class PluginPackageManager:
 
         release_tag = selection.release_tag
         if release_tag and not selection.fallback_to_filelist:
-            async def prepare_selected_release() -> tuple[bool, str]:
+            async def prepare_selected_release(staging_dir: Path) -> tuple[bool, str]:
                 return await self.__async_install_from_release(
                     pid,
                     selection.user_repo,
                     release_tag,
+                    staging_dir,
                 )
 
             return await self.__install_flow_async(
@@ -1911,20 +2033,24 @@ class PluginPackageManager:
 
         if release_tag:
             # 当前索引 Release 失败时回退文件列表，保持同步与异步安装一致。
-            async def prepare_release() -> tuple[bool, str]:
+            async def prepare_release(staging_dir: Path) -> tuple[bool, str]:
                 ok, msg = await self.__async_install_from_release(
                     pid,
                     selection.user_repo,
                     release_tag,
+                    staging_dir,
                 )
                 if ok:
                     return True, msg
                 logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}")
-                await self.__async_remove_old_plugin(pid)
+                # 回退前只清理本次暂存的半份 Release 产物；已安装的运行目录
+                # 此刻还没有被触碰，清理它会让一次可恢复的回退变成数据丢失。
+                await aioshutil.rmtree(staging_dir, ignore_errors=True)
                 return await self.__prepare_content_via_filelist_async(
                     pid,
                     selection.user_repo,
                     selection.package_version,
+                    staging_dir,
                 )
 
             return await self.__install_flow_async(
@@ -1936,11 +2062,12 @@ class PluginPackageManager:
             )
         # 未声明 release 打包的插件继续使用文件列表方式安装。
 
-        async def prepare_filelist() -> tuple[bool, str]:
+        async def prepare_filelist(staging_dir: Path) -> tuple[bool, str]:
             return await self.__prepare_content_via_filelist_async(
                 pid,
                 selection.user_repo,
                 selection.package_version,
+                staging_dir,
             )
 
         return await self.__install_flow_async(
@@ -2006,29 +2133,38 @@ class PluginPackageManager:
         self,
         pid: str,
         force_install: bool,
-        prepare_content: Callable[[], Awaitable[tuple[bool, str]]],
+        prepare_content: Callable[[Path], Awaitable[tuple[bool, str]]],
         repo_url: Optional[str] = None,
         before_dependency_install: Optional[Callable[[], None]] = None,
     ) -> tuple[bool, str]:
         """
-        异步安装流程，处理插件内容准备、依赖安装和注册
+        异步安装统一流程：暂存内容→备份→换入→安装依赖→上报
+        prepare_content 负责把插件文件放到调用时给定的暂存目录；只有新内容在暂存
+        目录里完整就位后才会触碰运行目录，中断语义与同步流程一致。
         """
+        plugin_dir = self.__plugin_dir(pid)
+        staging_dir = self.__new_install_staging_dir(pid)
         backup_dir = None
         try:
+            success, message = await prepare_content(staging_dir)
+            if not success:
+                # 运行目录此刻尚未被触碰，已装插件天然完整，无需还原也无需清理
+                logger.error(f"{pid} 准备插件内容失败：{message}")
+                return False, message
+
             if not force_install:
                 backup_dir = await self.__async_backup_plugin(pid)
 
-            await self.__async_remove_old_plugin(pid)
-
-            success, message = await prepare_content()
-            if not success:
-                logger.error(f"{pid} 准备插件内容失败：{message}")
-                if backup_dir:
+            try:
+                await self.__async_swap_staged_plugin_content(staging_dir, plugin_dir)
+            except OSError as error:
+                message = f"写入插件内容失败：{error}"
+                logger.error(f"{pid} {message}")
+                # 换入函数已保证失败时运行目录保持换入前的内容；只有连恢复都失败、
+                # 运行目录确实缺失时，才动用本次的临时备份兜底。
+                if backup_dir and not plugin_dir.exists():
                     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 = (
@@ -2058,11 +2194,15 @@ class PluginPackageManager:
         finally:
             if backup_dir:
                 await aioshutil.rmtree(backup_dir, ignore_errors=True)
+            if await AsyncPath(staging_dir).exists():
+                await aioshutil.rmtree(staging_dir, ignore_errors=True)
 
     def __prepare_content_via_filelist_sync(self, pid: str, user_repo: str,
-                                            package_version: Optional[str]) -> tuple[bool, str]:
+                                            package_version: Optional[str],
+                                            dest_root: Optional[Path] = None) -> tuple[bool, str]:
         """
         同步准备插件内容，通过文件列表获取插件文件和依赖
+        :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录
         """
         runtime_pid = pid.lower()
         file_list, msg = self.__get_file_list(runtime_pid, user_repo, package_version)
@@ -2070,15 +2210,19 @@ class PluginPackageManager:
             if msg == "插件源码目录不存在":
                 return False, f"{pid} {msg}"
             return False, msg or "插件文件列表读取失败"
-        ok, m = self.__download_files(runtime_pid, file_list, user_repo, package_version)
+        ok, m = self.__download_files(
+            runtime_pid, file_list, user_repo, package_version, dest_root
+        )
         if not ok:
             return False, m
         return True, ""
 
     async def __prepare_content_via_filelist_async(self, pid: str, user_repo: str,
-                                                   package_version: Optional[str]) -> tuple[bool, str]:
+                                                   package_version: Optional[str],
+                                                   dest_root: Optional[Path] = None) -> tuple[bool, str]:
         """
         异步准备插件内容，通过文件列表获取插件文件和依赖
+        :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录
         """
         runtime_pid = pid.lower()
         file_list, msg = await self.__async_get_file_list(
@@ -2095,16 +2239,18 @@ class PluginPackageManager:
             file_list,
             user_repo,
             package_version,
+            dest_root,
         )
         if not ok:
             return False, m
         return True, ""
 
-    async def __async_install_from_release(self, pid: str, user_repo: str, release_tag: str) -> tuple[bool, str]:
+    async def __async_install_from_release(self, pid: str, user_repo: str, release_tag: str,
+                                           dest_root: Optional[Path] = None) -> tuple[bool, str]:
         """
         通过 GitHub Release 资产文件安装插件（异步）。
         规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件；
-        将其全部解压到 app/plugins/{pid}
+        将其全部解压到 dest_root，未指定时直接解压到插件运行目录。
         """
         # 拼接资产文件名
         asset_name = f"{release_tag.lower()}.zip"
@@ -2152,7 +2298,10 @@ class PluginPackageManager:
                 infos = zf.infolist()
                 if not infos:
                     return False, "压缩包内容为空"
-                dest_base = self._plugins_root() / pid.lower()
+                dest_base = (
+                    dest_root if dest_root is not None
+                    else self._plugins_root() / pid.lower()
+                )
                 targets = self.__iter_release_zip_targets(zf, dest_base)
                 wrote_any = False
                 for info, dest_path, is_dir in targets:

tests/fixtures/architecture/concurrency-baseline.json

@@ -65,6 +65,7 @@
   "app/adapters/system/plugin/package.py:app.runtime.execution.run_in_threadpool_to_completion": {
     "owners": {
       "PluginPackageManager.__async_install_dependencies_if_required": 1,
+      "PluginPackageManager.__async_swap_staged_plugin_content": 1,
       "PluginPackageManager.async_activate_persistent_backup": 1,
       "PluginPackageManager.async_checkpoint": 1,
       "PluginPackageManager.async_cleanup": 1,

tests/test_plugin_adapter_parity.py

@@ -127,7 +127,7 @@ async def test_plugin_package_sync_async_preflight_failures_are_identical(
             {"version": "1.2.3", "release": True},
             None,
             [],
-            ["release", "remove", "filelist"],
+            ["release", "filelist"],
         ),
         (
             {"version": "1.2.3", "release": True},
@@ -139,6 +139,7 @@ async def test_plugin_package_sync_async_preflight_failures_are_identical(
 )
 async def test_plugin_package_sync_async_execute_same_selected_strategy(
     monkeypatch,
+    tmp_path: Path,
     metadata: dict,
     release_version: str | None,
     release_items: list[dict],
@@ -198,10 +199,10 @@ async def test_plugin_package_sync_async_execute_same_selected_strategy(
         async_calls.append("remove")
 
     def install_flow(_pid, _force, prepare, _repo_url, _before):
-        return prepare()
+        return prepare(tmp_path / "staging")
 
     async def async_install_flow(_pid, _force, prepare, _repo_url, _before):
-        return await prepare()
+        return await prepare(tmp_path / "staging")
 
     monkeypatch.setattr(manager, "async_get_plugin_package_version", selected_version)
     monkeypatch.setattr(manager, "async_get_plugin_release_versions", releases)

tests/test_plugin_helper.py

@@ -205,12 +205,18 @@ def _patch_sync_remote_install(helper, monkeypatch, meta: dict,
     monkeypatch.setattr(helper, "_PluginPackageManager__backup_plugin", lambda _pid: None)
     monkeypatch.setattr(helper, "_PluginPackageManager__remove_old_plugin", lambda _pid: calls.append("remove"))
     monkeypatch.setattr(helper, "_PluginPackageManager__install_dependencies_if_required", lambda _pid: (False, True, ""))
+    # 内容准备被替身接管、不会真的产出暂存内容，换入同样替身化，避免测试写入真实运行目录
+    monkeypatch.setattr(
+        helper,
+        "_PluginPackageManager__swap_staged_plugin_content",
+        lambda _staging_dir, _final_dir: None,
+    )
 
-    def fake_release(_pid, _user_repo, _release_tag):
+    def fake_release(_pid, _user_repo, _release_tag, _dest_root=None):
         calls.append("release")
         return release_result
 
-    def fake_filelist(_pid, _user_repo, _package_version):
+    def fake_filelist(_pid, _user_repo, _package_version, _dest_root=None):
         calls.append("filelist")
         return filelist_result
 
@@ -240,14 +246,17 @@ def _patch_async_remote_install(helper, monkeypatch, meta: dict,
     async def fake_dependencies(_pid):
         return False, True, ""
 
-    async def fake_release(_pid, _user_repo, _release_tag):
+    async def fake_release(_pid, _user_repo, _release_tag, _dest_root=None):
         calls.append("release")
         return release_result
 
-    async def fake_filelist(_pid, _user_repo, _package_version):
+    async def fake_filelist(_pid, _user_repo, _package_version, _dest_root=None):
         calls.append("filelist")
         return filelist_result
 
+    async def fake_swap(_staging_dir, _final_dir):
+        """内容准备被替身接管，换入同样替身化，避免测试写入真实运行目录。"""
+
     async def fake_to_thread(func, *args, **kwargs):
         calls.append(("to_thread", func, args, kwargs))
         return None
@@ -256,6 +265,9 @@ def _patch_async_remote_install(helper, monkeypatch, meta: dict,
     monkeypatch.setattr(helper, "_PluginPackageManager__async_get_plugin_meta", fake_meta)
     monkeypatch.setattr(helper, "_PluginPackageManager__async_backup_plugin", fake_backup)
     monkeypatch.setattr(helper, "_PluginPackageManager__async_remove_old_plugin", fake_remove)
+    monkeypatch.setattr(
+        helper, "_PluginPackageManager__async_swap_staged_plugin_content", fake_swap
+    )
     monkeypatch.setattr(helper, "_PluginPackageManager__async_install_dependencies_if_required", fake_dependencies)
     monkeypatch.setattr(helper, "_PluginPackageManager__async_install_from_release", fake_release)
     monkeypatch.setattr(helper, "_PluginPackageManager__prepare_content_via_filelist_async", fake_filelist)
@@ -2232,11 +2244,11 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert ["remove", "release"] == calls
+        assert ["release"] == calls
 
     def test_install_falls_back_to_filelist_when_release_is_missing(self, monkeypatch):
         """
-        release 标记存在但 tag 或 zip 尚未生成时，清理可能残留的安装目录后回退文件列表安装。
+        release 标记存在但 tag 或 zip 尚未生成时，清理本次暂存的半份产物后回退文件列表安装。
         """
         try:
             from app.adapters.external.market import PluginHelper
@@ -2256,11 +2268,11 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert ["remove", "release", "remove", "filelist"] == calls
+        assert ["release", "filelist"] == calls
 
     def test_install_reports_filelist_error_after_release_fallback_fails(self, monkeypatch):
         """
-        release 和源码目录都不存在时返回稳定业务错误，并在每次写入前后保持目录可回滚。
+        release 和源码目录都不存在时返回稳定业务错误，全程不触碰已安装的运行目录。
         """
         try:
             from app.adapters.external.market import PluginHelper
@@ -2280,7 +2292,7 @@ demo = { index = "private" }
 
         assert not success
         assert "DemoPlugin 插件源码目录不存在" == message
-        assert ["remove", "release", "remove", "filelist", "remove"] == calls
+        assert ["release", "filelist"] == calls
 
     def test_install_uses_filelist_when_release_flag_is_disabled(self, monkeypatch):
         """
@@ -2304,7 +2316,7 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert ["remove", "filelist"] == calls
+        assert ["filelist"] == calls
 
     def test_install_rejects_release_without_version(self, monkeypatch):
         """
@@ -2414,7 +2426,7 @@ demo = { index = "private" }
 
         assert not success
         assert "未找到资产文件：demoplugin_v1.2.0.zip" == message
-        assert ["remove", "release", "remove"] == calls
+        assert ["release"] == calls
 
     def test_install_rejects_release_version_missing_from_release_list(self, monkeypatch):
         """
@@ -2508,6 +2520,11 @@ demo = { index = "private" }
         monkeypatch.setattr(helper, "_PluginPackageManager__remove_old_plugin", lambda _pid: None)
         monkeypatch.setattr(helper, "_PluginPackageManager__install_dependencies_if_required", lambda _pid: (False, True, ""))
         monkeypatch.setattr(helper, "_PluginPackageManager__prepare_content_via_filelist_sync", lambda *_args: (True, ""))
+        monkeypatch.setattr(
+            helper,
+            "_PluginPackageManager__swap_staged_plugin_content",
+            lambda _staging_dir, _final_dir: None,
+        )
 
         success, message = helper.install_raw(PLUGIN_ID, REPO_URL, force_install=True)
 
@@ -2561,7 +2578,7 @@ demo = { index = "private" }
 
     def test_install_release_download_failure_falls_back_to_filelist(self, monkeypatch):
         """
-        release tag 存在但 zip 下载失败时清理可能残留的目录，再回退文件列表安装。
+        release tag 存在但 zip 下载失败时清理本次暂存的半份产物，再回退文件列表安装。
         """
         try:
             from app.adapters.external.market import PluginHelper
@@ -2581,7 +2598,7 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert ["remove", "release", "remove", "filelist"] == calls
+        assert ["release", "filelist"] == calls
 
     def test_async_install_uses_release_package_when_asset_is_available(self, monkeypatch):
         """
@@ -2606,11 +2623,11 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert calls == ["remove", "release"]
+        assert calls == ["release"]
 
     def test_async_install_falls_back_to_filelist_when_release_is_missing(self, monkeypatch):
         """
-        异步安装路径在 release tag 或 zip 未生成时，清理可能残留的安装目录后回退文件列表安装。
+        异步安装路径在 release tag 或 zip 未生成时，清理本次暂存的半份产物后回退文件列表安装。
         """
         try:
             from app.adapters.external.market import PluginHelper
@@ -2632,7 +2649,7 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert calls == ["remove", "release", "remove", "filelist"]
+        assert calls == ["release", "filelist"]
 
     def test_async_install_old_release_version_uses_release_asset_without_filelist_fallback(self, monkeypatch):
         """
@@ -2666,7 +2683,7 @@ demo = { index = "private" }
 
         assert not success
         assert "未找到资产文件：demoplugin_v1.2.0.zip" == message
-        assert calls[:3] == ["remove", "release", "remove"]
+        assert calls[:1] == ["release"]
 
     def test_async_install_rejects_release_version_missing_from_release_list(self, monkeypatch):
         """
@@ -2724,7 +2741,7 @@ demo = { index = "private" }
 
         assert not success
         assert "DemoPlugin 插件源码目录不存在" == message
-        assert calls == ["remove", "release", "remove", "filelist", "remove"]
+        assert calls == ["release", "filelist"]
 
     def test_async_install_release_fallback_preserves_plugin_id(self, monkeypatch):
         """
@@ -2745,7 +2762,7 @@ demo = { index = "private" }
             (True, ""),
         )
 
-        async def fake_filelist(pid, _user_repo, _package_version):
+        async def fake_filelist(pid, _user_repo, _package_version, _dest_root=None):
             filelist_pids.append(pid)
             return True, ""
 
@@ -2778,7 +2795,7 @@ demo = { index = "private" }
             (True, ""),
         )
 
-        async def fake_filelist(pid, _user_repo, _package_version):
+        async def fake_filelist(pid, _user_repo, _package_version, _dest_root=None):
             filelist_pids.append(pid)
             return True, ""
 
@@ -3093,9 +3110,9 @@ demo = { index = "private" }
         assert not success
         assert "解压 Release 压缩包失败" in message
 
-    def test_install_flow_sync_restores_backup_when_prepare_fails(self, monkeypatch):
+    def test_install_flow_sync_keeps_installed_plugin_when_prepare_fails(self, monkeypatch):
         """
-        内容准备失败时恢复备份，避免安装失败后留下半成品目录。
+        内容准备失败时运行目录尚未被触碰，既不需要还原也不该被清理。
         """
         try:
             from app.adapters.external.market import PluginHelper
@@ -3109,12 +3126,12 @@ demo = { index = "private" }
         monkeypatch.setattr(helper, "_PluginPackageManager__restore_plugin", lambda _pid, _backup: calls.append("restore"))
 
         success, message = helper._PluginPackageManager__install_flow_sync(
-            PLUGIN_ID, False, lambda: (False, "prepare failed")
+            PLUGIN_ID, False, lambda _staging_dir: (False, "prepare failed")
         )
 
         assert not success
         assert "prepare failed" == message
-        assert ["remove", "restore"] == calls
+        assert [] == calls
 
     def test_install_flow_sync_restores_backup_when_dependency_install_fails(self, monkeypatch):
         """
@@ -3130,6 +3147,11 @@ demo = { index = "private" }
         monkeypatch.setattr(helper, "_PluginPackageManager__backup_plugin", lambda _pid: "/backup")
         monkeypatch.setattr(helper, "_PluginPackageManager__remove_old_plugin", lambda _pid: calls.append("remove"))
         monkeypatch.setattr(helper, "_PluginPackageManager__restore_plugin", lambda _pid, _backup: calls.append("restore"))
+        monkeypatch.setattr(
+            helper,
+            "_PluginPackageManager__swap_staged_plugin_content",
+            lambda _staging_dir, _final_dir: calls.append("swap"),
+        )
         monkeypatch.setattr(
             helper,
             "_PluginPackageManager__install_dependencies_if_required",
@@ -3137,12 +3159,12 @@ demo = { index = "private" }
         )
 
         success, message = helper._PluginPackageManager__install_flow_sync(
-            PLUGIN_ID, False, lambda: (True, "")
+            PLUGIN_ID, False, lambda _staging_dir: (True, "")
         )
 
         assert not success
         assert "dependency failed" == message
-        assert ["remove", "restore"] == calls
+        assert ["swap", "restore"] == calls
 
     def test_install_flow_sync_restores_backup_for_invalid_modern_manifest(self, tmp_path, monkeypatch):
         """现代清单无效时恢复旧插件目录。"""
@@ -3155,9 +3177,9 @@ demo = { index = "private" }
         monkeypatch.setattr(market_module, "PLUGIN_DIR", plugin_root)
         monkeypatch.setattr(market_module.settings, "CONFIG_DIR", str(tmp_path))
 
-        def prepare_content():
-            plugin_dir.mkdir(parents=True)
-            (plugin_dir / "pyproject.toml").write_text(
+        def prepare_content(staging_dir):
+            staging_dir.mkdir(parents=True)
+            (staging_dir / "pyproject.toml").write_text(
                 "[project]\nname = 'demo'\n",
                 encoding="utf-8",
             )
@@ -3271,7 +3293,7 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert calls == [("demoplugin", file_list, "demo/repo", "v2")]
+        assert calls == [("demoplugin", file_list, "demo/repo", "v2", None)]
 
     def test_prepare_content_via_filelist_sync_reports_missing_file_list(self, monkeypatch):
         """
@@ -3369,7 +3391,7 @@ demo = { index = "private" }
 
         assert success
         assert "" == message
-        assert calls == [("demoplugin", file_list, "demo/repo", "v2")]
+        assert calls == [("demoplugin", file_list, "demo/repo", "v2", None)]
 
     def test_async_prepare_content_via_filelist_reports_missing_file_list(self, monkeypatch):
         """
@@ -3453,9 +3475,9 @@ demo = { index = "private" }
         assert not success
         assert "download failed" == message
 
-    def test_install_flow_async_restores_backup_when_prepare_fails(self, monkeypatch):
+    def test_install_flow_async_keeps_installed_plugin_when_prepare_fails(self, monkeypatch):
         """
-        异步内容准备失败时恢复备份。
+        异步内容准备失败时运行目录尚未被触碰，既不需要还原也不该被清理。
         """
         try:
             from app.adapters.external.market import PluginHelper
@@ -3474,7 +3496,7 @@ demo = { index = "private" }
         async def restore(_pid, _backup):
             calls.append("restore")
 
-        async def prepare():
+        async def prepare(_staging_dir):
             return False, "prepare failed"
 
         monkeypatch.setattr(helper, "_PluginPackageManager__async_backup_plugin", backup)
@@ -3485,7 +3507,7 @@ demo = { index = "private" }
 
         assert not success
         assert "prepare failed" == message
-        assert ["remove", "restore"] == calls
+        assert [] == calls
 
     def test_install_flow_async_restores_backup_when_dependency_install_fails(self, monkeypatch):
         """
@@ -3508,22 +3530,26 @@ demo = { index = "private" }
         async def restore(_pid, _backup):
             calls.append("restore")
 
-        async def prepare():
+        async def prepare(_staging_dir):
             return True, ""
 
+        async def swap(_staging_dir, _final_dir):
+            calls.append("swap")
+
         async def dependencies(_pid):
             return True, False, "dependency failed"
 
         monkeypatch.setattr(helper, "_PluginPackageManager__async_backup_plugin", backup)
         monkeypatch.setattr(helper, "_PluginPackageManager__async_remove_old_plugin", remove)
         monkeypatch.setattr(helper, "_PluginPackageManager__async_restore_plugin", restore)
+        monkeypatch.setattr(helper, "_PluginPackageManager__async_swap_staged_plugin_content", swap)
         monkeypatch.setattr(helper, "_PluginPackageManager__async_install_dependencies_if_required", dependencies)
 
         success, message = asyncio.run(helper._PluginPackageManager__install_flow_async(PLUGIN_ID, False, prepare))
 
         assert not success
         assert "dependency failed" == message
-        assert ["remove", "restore"] == calls
+        assert ["swap", "restore"] == calls
 
     def test_async_install_from_release_reports_missing_asset(self, monkeypatch):
         """

tests/test_plugin_install_atomic_swap.py

@@ -0,0 +1,356 @@
+"""插件安装的换入原子性：准备失败不得删掉已装插件，换入失败必须还原运行目录。"""
+
+import errno
+import shutil
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from app.adapters.system.plugin.package import PluginPackageManager
+
+PLUGIN_ID = "DemoPlugin"
+REPO_URL = "https://github.com/demo/MoviePilot-Plugins"
+INSTALLED_MARK = "installed"
+DOWNLOAD_ERROR = "文件 plugins.v2/demoplugin/__init__.py 下载失败！"
+
+
+def _swap(staging_dir: Path, final_dir: Path) -> None:
+    """调用被测的换入实现，保持用例只表达行为约定。"""
+    PluginPackageManager._PluginPackageManager__swap_staged_plugin_content(  # type: ignore[attr-defined]
+        staging_dir, final_dir
+    )
+
+
+def _installed_manager(monkeypatch, tmp_path: Path) -> tuple[PluginPackageManager, Path]:
+    """构造隔离目录的包 owner，并预置一个已安装且可用的插件运行目录。"""
+    plugin_root = tmp_path / "plugins"
+    settings = SimpleNamespace(
+        ROOT_PATH=tmp_path,
+        TEMP_PATH=tmp_path / "temp",
+        CONFIG_PATH=tmp_path / "config",
+    )
+    monkeypatch.setattr(
+        "app.adapters.system.plugin.package.get_runtime_setting",
+        lambda key: getattr(settings, key),
+    )
+    plugin_dir = plugin_root / PLUGIN_ID.lower()
+    plugin_dir.mkdir(parents=True)
+    (plugin_dir / "__init__.py").write_text(INSTALLED_MARK, encoding="utf-8")
+    manager = PluginPackageManager(source=Mock(), plugin_root=plugin_root)
+    return manager, plugin_dir
+
+
+def _stub_market_lookup(manager: PluginPackageManager, monkeypatch) -> None:
+    """把远端市场查询固定下来，让内容准备的成败成为唯一变量。"""
+    monkeypatch.setattr(
+        "app.adapters.system.plugin.package.SystemUtils.is_frozen", lambda: False
+    )
+    monkeypatch.setattr(manager, "is_local_repo_url", lambda _repo_url: False)
+    monkeypatch.setattr(manager, "get_repo_info", lambda _repo_url: ("demo", "repo"))
+    monkeypatch.setattr(manager, "get_plugin_package_version", lambda *_args: "v2")
+    monkeypatch.setattr(
+        manager,
+        "_PluginPackageManager__get_plugin_meta",
+        lambda *_args: {"release": False, "version": "1.0.0"},
+    )
+
+    async def async_package_version(*_args) -> str:
+        """异步入口复用同一索引代际。"""
+        return "v2"
+
+    async def async_meta(*_args) -> dict:
+        """异步入口复用同一插件元数据。"""
+        return {"release": False, "version": "1.0.0"}
+
+    monkeypatch.setattr(
+        manager, "async_get_plugin_package_version", async_package_version
+    )
+    monkeypatch.setattr(
+        manager, "_PluginPackageManager__async_get_plugin_meta", async_meta
+    )
+
+
+def _staging_root(tmp_path: Path) -> Path:
+    """返回安装暂存根目录。"""
+    return tmp_path / "temp" / "plugin_install_staging"
+
+
+def _assert_no_staging_residue(tmp_path: Path) -> None:
+    """安装结束后不得残留暂存目录。"""
+    staging_root = _staging_root(tmp_path)
+    assert not staging_root.exists() or not list(staging_root.iterdir())
+
+
+def _assert_no_swap_residue(plugin_dir: Path) -> None:
+    """换入结束后不得在插件根目录残留换入用的回滚材料。"""
+    assert not list(plugin_dir.parent.glob(f".{plugin_dir.name}.previous-*"))
+
+
+@pytest.mark.parametrize("force_install", [True, False])
+def test_sync_install_keeps_installed_plugin_when_content_preparation_fails(
+    monkeypatch, tmp_path: Path, force_install: bool
+) -> None:
+    """同步安装在下载失败时必须原样保留已装插件，无论本次是否留了备份。"""
+    manager, plugin_dir = _installed_manager(monkeypatch, tmp_path)
+    _stub_market_lookup(manager, monkeypatch)
+    monkeypatch.setattr(
+        manager,
+        "_PluginPackageManager__prepare_content_via_filelist_sync",
+        lambda *_args: (False, DOWNLOAD_ERROR),
+    )
+
+    success, message = manager.install_raw(
+        PLUGIN_ID, REPO_URL, package_version="v2", force_install=force_install
+    )
+
+    assert not success
+    assert message == DOWNLOAD_ERROR
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK
+    _assert_no_staging_residue(tmp_path)
+    _assert_no_swap_residue(plugin_dir)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("force_install", [True, False])
+async def test_async_install_keeps_installed_plugin_when_content_preparation_fails(
+    monkeypatch, tmp_path: Path, force_install: bool
+) -> None:
+    """异步安装与同步保持一致的失败语义，下载失败不得删掉已装插件。"""
+    manager, plugin_dir = _installed_manager(monkeypatch, tmp_path)
+    _stub_market_lookup(manager, monkeypatch)
+
+    async def failing_filelist(*_args) -> tuple[bool, str]:
+        """模拟资产瞬时不可用导致的内容准备失败。"""
+        return False, DOWNLOAD_ERROR
+
+    monkeypatch.setattr(
+        manager,
+        "_PluginPackageManager__prepare_content_via_filelist_async",
+        failing_filelist,
+    )
+
+    success, message = await manager.async_install_raw(
+        PLUGIN_ID, REPO_URL, package_version="v2", force_install=force_install
+    )
+
+    assert not success
+    assert message == DOWNLOAD_ERROR
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK
+    _assert_no_staging_residue(tmp_path)
+    _assert_no_swap_residue(plugin_dir)
+
+
+def test_sync_install_replaces_installed_plugin_when_content_is_ready(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """同步安装成功时用新内容整体替换运行目录，不与旧文件混合。"""
+    manager, plugin_dir = _installed_manager(monkeypatch, tmp_path)
+    (plugin_dir / "stale.py").write_text("stale", encoding="utf-8")
+    _stub_market_lookup(manager, monkeypatch)
+
+    def prepare(_pid, _user_repo, _package_version, dest_root: Path) -> tuple[bool, str]:
+        """把新版本内容写进本次分配的暂存目录。"""
+        dest_root.mkdir(parents=True, exist_ok=True)
+        (dest_root / "__init__.py").write_text("upgraded", encoding="utf-8")
+        return True, ""
+
+    monkeypatch.setattr(
+        manager, "_PluginPackageManager__prepare_content_via_filelist_sync", prepare
+    )
+
+    success, message = manager.install_raw(
+        PLUGIN_ID, REPO_URL, package_version="v2", force_install=True
+    )
+
+    assert (success, message) == (True, "")
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded"
+    assert not (plugin_dir / "stale.py").exists()
+    _assert_no_staging_residue(tmp_path)
+    _assert_no_swap_residue(plugin_dir)
+
+
+@pytest.mark.asyncio
+async def test_async_install_replaces_installed_plugin_when_content_is_ready(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """异步安装成功时同样整体替换运行目录。"""
+    manager, plugin_dir = _installed_manager(monkeypatch, tmp_path)
+    (plugin_dir / "stale.py").write_text("stale", encoding="utf-8")
+    _stub_market_lookup(manager, monkeypatch)
+
+    async def prepare(
+        _pid, _user_repo, _package_version, dest_root: Path
+    ) -> tuple[bool, str]:
+        """把新版本内容写进本次分配的暂存目录。"""
+        dest_root.mkdir(parents=True, exist_ok=True)
+        (dest_root / "__init__.py").write_text("upgraded", encoding="utf-8")
+        return True, ""
+
+    monkeypatch.setattr(
+        manager, "_PluginPackageManager__prepare_content_via_filelist_async", prepare
+    )
+
+    success, message = await manager.async_install_raw(
+        PLUGIN_ID, REPO_URL, package_version="v2", force_install=True
+    )
+
+    assert (success, message) == (True, "")
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded"
+    assert not (plugin_dir / "stale.py").exists()
+    _assert_no_staging_residue(tmp_path)
+    _assert_no_swap_residue(plugin_dir)
+
+
+def test_sync_install_keeps_installed_plugin_when_swap_fails(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """换入自身失败时运行目录必须仍是换入前那一份，并报出失败原因。"""
+    manager, plugin_dir = _installed_manager(monkeypatch, tmp_path)
+    _stub_market_lookup(manager, monkeypatch)
+
+    def prepare(_pid, _user_repo, _package_version, dest_root: Path) -> tuple[bool, str]:
+        """内容准备成功，但随后的换入会被模拟为失败。"""
+        dest_root.mkdir(parents=True, exist_ok=True)
+        (dest_root / "__init__.py").write_text("upgraded", encoding="utf-8")
+        return True, ""
+
+    def failing_swap(_staging_dir: Path, _final_dir: Path) -> None:
+        """模拟运行目录不可写导致的换入失败。"""
+        raise OSError(errno.EACCES, "permission denied")
+
+    monkeypatch.setattr(
+        manager, "_PluginPackageManager__prepare_content_via_filelist_sync", prepare
+    )
+    monkeypatch.setattr(
+        manager, "_PluginPackageManager__swap_staged_plugin_content", failing_swap
+    )
+
+    success, message = manager.install_raw(
+        PLUGIN_ID, REPO_URL, package_version="v2", force_install=True
+    )
+
+    assert not success
+    assert "写入插件内容失败" in message
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK
+    _assert_no_staging_residue(tmp_path)
+
+
+def _staged_pair(tmp_path: Path) -> tuple[Path, Path]:
+    """准备一份待换入内容和一个已有旧内容的目标目录。"""
+    staging_dir = tmp_path / "staging"
+    staging_dir.mkdir()
+    (staging_dir / "__init__.py").write_text("upgraded", encoding="utf-8")
+    final_dir = tmp_path / "plugins" / "demoplugin"
+    final_dir.mkdir(parents=True)
+    (final_dir / "__init__.py").write_text(INSTALLED_MARK, encoding="utf-8")
+    (final_dir / "stale.py").write_text("stale", encoding="utf-8")
+    return staging_dir, final_dir
+
+
+def _break_replace(monkeypatch, *, source: Path, code: int) -> None:
+    """让指定源路径的原子改名失败，模拟跨文件系统或权限受限。"""
+    original = Path.replace
+
+    def guarded(self: Path, target):
+        """只拦截被指定的源路径，其它改名仍走真实实现。"""
+        if self == source:
+            raise OSError(code, "simulated")
+        return original(self, target)
+
+    monkeypatch.setattr(Path, "replace", guarded)
+
+
+def test_swap_publishes_new_content_and_drops_old_content(tmp_path: Path) -> None:
+    """换入后运行目录只剩新内容，回滚材料被清理。"""
+    staging_dir, final_dir = _staged_pair(tmp_path)
+
+    _swap(staging_dir, final_dir)
+
+    assert (final_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded"
+    assert not (final_dir / "stale.py").exists()
+    _assert_no_swap_residue(final_dir)
+
+
+def test_swap_falls_back_to_copy_when_staging_crosses_filesystems(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """暂存目录与插件根目录跨文件系统时退化为复制，结果与改名一致。"""
+    staging_dir, final_dir = _staged_pair(tmp_path)
+    _break_replace(monkeypatch, source=staging_dir, code=errno.EXDEV)
+
+    _swap(staging_dir, final_dir)
+
+    assert (final_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded"
+    assert not (final_dir / "stale.py").exists()
+    _assert_no_swap_residue(final_dir)
+
+
+def test_swap_falls_back_to_copy_when_old_directory_cannot_be_renamed(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """overlayfs 拒绝改名旧目录时先复制出回滚材料，再继续换入新内容。"""
+    staging_dir, final_dir = _staged_pair(tmp_path)
+    _break_replace(monkeypatch, source=final_dir, code=errno.EXDEV)
+
+    _swap(staging_dir, final_dir)
+
+    assert (final_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded"
+    assert not (final_dir / "stale.py").exists()
+    _assert_no_swap_residue(final_dir)
+
+
+def test_swap_keeps_old_content_when_staging_is_missing(tmp_path: Path) -> None:
+    """待换入内容不存在时运行目录必须原样退回，而不是被清空。"""
+    staging_dir, final_dir = _staged_pair(tmp_path)
+    shutil.rmtree(staging_dir)
+
+    with pytest.raises(OSError):
+        _swap(staging_dir, final_dir)
+
+    assert (final_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK
+    assert (final_dir / "stale.py").exists()
+    _assert_no_swap_residue(final_dir)
+
+
+def test_swap_keeps_old_content_when_old_directory_cannot_be_moved_aside(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """旧目录挪不开时必须整体放弃换入，绝不先删后写。"""
+    staging_dir, final_dir = _staged_pair(tmp_path)
+    _break_replace(monkeypatch, source=final_dir, code=errno.EACCES)
+
+    with pytest.raises(OSError):
+        _swap(staging_dir, final_dir)
+
+    assert (final_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK
+    assert (final_dir / "stale.py").exists()
+    assert (staging_dir / "__init__.py").exists()
+    _assert_no_swap_residue(final_dir)
+
+
+def test_swap_rolls_back_when_cross_filesystem_copy_fails(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """跨文件系统复制中途失败时删掉半成品并把旧目录换回。"""
+    staging_dir, final_dir = _staged_pair(tmp_path)
+    _break_replace(monkeypatch, source=staging_dir, code=errno.EXDEV)
+
+    def failing_copytree(src, dst, *_args, **_kwargs):
+        """模拟复制到一半磁盘写入失败，留下半份目标目录。"""
+        Path(dst).mkdir(parents=True, exist_ok=True)
+        (Path(dst) / "partial.py").write_text("partial", encoding="utf-8")
+        raise OSError(errno.ENOSPC, "no space left on device")
+
+    monkeypatch.setattr(
+        "app.adapters.system.plugin.package.shutil.copytree", failing_copytree
+    )
+
+    with pytest.raises(OSError):
+        _swap(staging_dir, final_dir)
+
+    assert (final_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK
+    assert (final_dir / "stale.py").exists()
+    assert not (final_dir / "partial.py").exists()
+    _assert_no_swap_residue(final_dir)

tests/test_plugin_install_backup_cleanup.py

@@ -33,6 +33,11 @@ async def test_successful_install_flows_remove_transient_backups(
         "_PluginPackageManager__remove_old_plugin",
         lambda _pid: None,
     )
+    monkeypatch.setattr(
+        package,
+        "_PluginPackageManager__swap_staged_plugin_content",
+        lambda _staging_dir, _final_dir: None,
+    )
     monkeypatch.setattr(
         package,
         "_PluginPackageManager__install_dependencies_if_required",
@@ -47,7 +52,7 @@ async def test_successful_install_flows_remove_transient_backups(
     sync_result = package._PluginPackageManager__install_flow_sync(
         "DemoPlugin",
         False,
-        lambda: (True, ""),
+        lambda _staging_dir: (True, ""),
     )
 
     async def backup_plugin(_pid: str) -> str:
@@ -61,10 +66,13 @@ async def test_successful_install_flows_remove_transient_backups(
         """表示测试插件没有额外依赖。"""
         return False, False, "不存在依赖"
 
-    async def prepare_content() -> tuple[bool, str]:
+    async def prepare_content(_staging_dir: Path) -> tuple[bool, str]:
         """表示异步内容准备成功。"""
         return True, ""
 
+    async def swap_content(_staging_dir: Path, _final_dir: Path) -> None:
+        """隔离测试中的真实运行目录换入。"""
+
     monkeypatch.setattr(
         package,
         "_PluginPackageManager__async_backup_plugin",
@@ -75,6 +83,11 @@ async def test_successful_install_flows_remove_transient_backups(
         "_PluginPackageManager__async_remove_old_plugin",
         remove_plugin,
     )
+    monkeypatch.setattr(
+        package,
+        "_PluginPackageManager__async_swap_staged_plugin_content",
+        swap_content,
+    )
     monkeypatch.setattr(
         package,
         "_PluginPackageManager__async_install_dependencies_if_required",