fix: support plugin updates through URL and file installation (#10053)

* fix: support updating installed plugins from uploaded archives

* fix: update existing plugins through URL installation

* fix: skip temporary directories during plugin discovery

* fix: stage plugin installs and backups in system temp

* test: align dashboard plugin installer mock with staging API

astrbot/core/star/star_manager.py

@@ -8,6 +8,7 @@ import json
 import keyword
 import logging
 import os
+import shutil
 import sys
 import tempfile
 import traceback
@@ -36,6 +37,7 @@ from astrbot.core.utils.astrbot_path import (
     get_astrbot_config_path,
     get_astrbot_path,
     get_astrbot_plugin_path,
+    get_astrbot_system_tmp_path,
     get_astrbot_temp_path,
 )
 from astrbot.core.utils.io import remove_dir
@@ -292,6 +294,9 @@ class PluginManager:
         dirs = os.listdir(path)
         # 遍历文件夹，找到 main.py 或者和文件夹同名的文件
         for d in dirs:
+            # Staged installs and rollback backups are not discoverable plugins.
+            if d.startswith((".plugin-install-", ".plugin-upload-", ".plugin-backup-")):
+                continue
             if os.path.isdir(os.path.join(path, d)):
                 if os.path.exists(os.path.join(path, d, "main.py")):
                     module_str = "main"
@@ -1616,117 +1621,36 @@ class PluginManager:
         ignore_version_check: bool = False,
         download_url: str = "",
     ):
-        """从仓库 URL 安装插件
-
-        从指定的仓库 URL 下载并安装插件，然后加载该插件到系统中
+        """Install or update a plugin from a repository URL.
 
         Args:
-            repo_url (str): 要安装的插件仓库 URL
-            proxy (str, optional): 用于下载的代理服务器。默认为空字符串。
-            download_url (str, optional): 插件压缩包下载地址。提供时优先从此地址下载安装包。
+            repo_url: Plugin repository URL.
+            proxy: Optional proxy prefix for repository downloads.
+            ignore_version_check: Whether to bypass AstrBot version compatibility.
+            download_url: Optional archive URL to download instead of the repository.
 
         Returns:
-            dict | None: 安装成功时返回包含插件信息的字典:
-                - repo: 插件的仓库 URL
-                - readme: README.md 文件的内容(如果存在)
-                如果找不到插件元数据则返回 None。
+            Installed plugin repository, README, and name, if registered.
 
+        Raises:
+            Exception: If downloading, validation, or loading fails.
         """
-        # this metric is for displaying plugins installation count in pages
-        asyncio.create_task(
-            Metric.upload(
-                et="install_star",
-                repo=repo_url,
-            ),
-        )
-
+        asyncio.create_task(Metric.upload(et="install_star", repo=repo_url))
         async with self._pm_lock:
-            plugin_path = ""
-            dir_name = ""
-            try:
-                if download_url:
-                    plugin_path = await self._updater.install(
-                        repo_url,
-                        proxy,
-                        download_url=download_url,
-                    )
-                else:
-                    plugin_path = await self._updater.install(repo_url, proxy)
-
-                # reload the plugin
-                dir_name = os.path.basename(plugin_path)
-                metadata_dir_name = self._get_plugin_dir_name_from_metadata(plugin_path)
-                target_plugin_path = os.path.join(
-                    self.plugin_store_path,
-                    metadata_dir_name,
+            temp_root = Path(get_astrbot_system_tmp_path())
+            temp_root.mkdir(parents=True, exist_ok=True)
+            with tempfile.TemporaryDirectory(
+                dir=temp_root, prefix=".plugin-install-"
+            ) as staging_dir:
+                plugin_path = await self._updater.install(
+                    repo_url,
+                    proxy,
+                    download_url=download_url,
+                    target_dir=Path(staging_dir) / "plugin",
                 )
-                if target_plugin_path != plugin_path and os.path.exists(
-                    target_plugin_path
-                ):
-                    raise Exception(f"安装失败：目录 {metadata_dir_name} 已存在。")
-                if target_plugin_path != plugin_path:
-                    os.rename(plugin_path, target_plugin_path)
-                    plugin_path = target_plugin_path
-                    dir_name = metadata_dir_name
-                await self._ensure_plugin_requirements(
-                    plugin_path,
-                    dir_name,
+                return await self._install_plugin_from_directory(
+                    plugin_path, ignore_version_check=ignore_version_check
                 )
-                success, error_message = await self.load(
-                    specified_dir_name=dir_name,
-                    ignore_version_check=ignore_version_check,
-                )
-                if not success:
-                    raise Exception(
-                        error_message
-                        or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
-                    )
-
-                # Get the plugin metadata to return repo info
-                plugin = self.context.get_registered_star(dir_name)
-                if not plugin:
-                    # Try to find by other name if directory name doesn't match plugin name
-                    for star in self.context.get_all_stars():
-                        if star.root_dir_name == dir_name:
-                            plugin = star
-                            break
-
-                # Extract README.md content if exists
-                readme_content = None
-                readme_path = os.path.join(plugin_path, "README.md")
-                if not os.path.exists(readme_path):
-                    readme_path = os.path.join(plugin_path, "readme.md")
-
-                if os.path.exists(readme_path):
-                    try:
-                        with open(readme_path, encoding="utf-8") as f:
-                            readme_content = f.read()
-                    except Exception as e:
-                        logger.warning(
-                            f"Failed to read README.md for plugin {dir_name}: {e!s}",
-                        )
-
-                plugin_info = None
-                if plugin:
-                    plugin_info = {
-                        "repo": plugin.repo,
-                        "readme": readme_content,
-                        "name": plugin.name,
-                    }
-
-                return plugin_info
-            except Exception as e:
-                self._track_failed_install_dir(
-                    dir_name=dir_name,
-                    plugin_path=plugin_path,
-                    error=e,
-                )
-                if dir_name and plugin_path:
-                    logger.warning(
-                        f"Failed to install plugin {dir_name}; installation "
-                        f"directory: {plugin_path}",
-                    )
-                raise
 
     async def inspect_plugin_repository(
         self,
@@ -2074,45 +1998,195 @@ class PluginManager:
     async def install_plugin_from_file(
         self, zip_file_path: str, ignore_version_check: bool = False
     ):
-        dir_name = os.path.splitext(os.path.basename(zip_file_path))[0]
-        desti_dir = tempfile.mkdtemp(
-            dir=self.plugin_store_path, prefix="plugin_upload_"
-        )
-        temp_desti_dir = desti_dir
-        skip_failed_tracking = False
+        """Install an uploaded archive or replace the same installed plugin.
 
-        try:
-            self._updater._extract_plugin_archive(zip_file_path, desti_dir)
-            metadata_dir_name = self._get_plugin_dir_name_from_metadata(desti_dir)
-            target_plugin_path = os.path.join(
-                self.plugin_store_path,
-                metadata_dir_name,
-            )
-            if target_plugin_path != desti_dir and os.path.exists(target_plugin_path):
-                skip_failed_tracking = True
-                raise Exception(f"安装失败：目录 {metadata_dir_name} 已存在。")
-            if target_plugin_path != desti_dir:
-                os.rename(desti_dir, target_plugin_path)
-                dir_name = metadata_dir_name
-                desti_dir = target_plugin_path
+        Args:
+            zip_file_path: Path to the uploaded plugin archive.
+            ignore_version_check: Whether to bypass AstrBot version compatibility.
 
-            # remove the zip
-            try:
-                os.remove(zip_file_path)
-            except BaseException as e:
-                logger.warning(f"Failed to delete the plugin archive: {e!s}")
-            await self._ensure_plugin_requirements(desti_dir, dir_name)
-            # await self.reload()
-            success, error_message = await self.load(
-                specified_dir_name=dir_name,
-                ignore_version_check=ignore_version_check,
-            )
-            if not success:
-                raise Exception(
-                    error_message
-                    or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
+        Returns:
+            Installed plugin repository, README, and name, if registered.
+
+        Raises:
+            Exception: If validation, dependency installation, or loading fails.
+        """
+        async with self._pm_lock:
+            temp_root = Path(get_astrbot_system_tmp_path())
+            temp_root.mkdir(parents=True, exist_ok=True)
+            with tempfile.TemporaryDirectory(
+                dir=temp_root, prefix=".plugin-upload-"
+            ) as staging_dir:
+                plugin_path = Path(staging_dir) / "plugin"
+                self._updater._extract_plugin_archive(zip_file_path, str(plugin_path))
+                plugin_info = await self._install_plugin_from_directory(
+                    str(plugin_path), ignore_version_check=ignore_version_check
                 )
+            try:
+                Path(zip_file_path).unlink(missing_ok=True)
+            except OSError as exc:
+                logger.warning("Failed to delete the plugin archive: %s", exc)
+            if plugin_info and plugin_info.get("repo"):
+                asyncio.create_task(
+                    Metric.upload(et="install_star_f", repo=plugin_info["repo"])
+                )
+            return plugin_info
 
+    async def _install_plugin_from_directory(
+        self, plugin_path: str, ignore_version_check: bool = False
+    ):
+        """Install a staged plugin directory, restoring old code on update failure.
+
+        The caller holds the plugin manager lock and owns the staging directory.
+        Only failed first installs moved to a permanent directory are tracked.
+
+        Args:
+            plugin_path: Extracted or cloned plugin in a temporary directory.
+            ignore_version_check: Whether to bypass AstrBot version compatibility.
+
+        Returns:
+            Installed plugin repository, README, and name, if registered.
+
+        Raises:
+            Exception: If validation, dependency installation, or loading fails.
+        """
+        desti_dir = plugin_path
+        dir_name = Path(plugin_path).name
+        track_failed_install = False
+        try:
+            metadata_dir_name = self._get_plugin_dir_name_from_metadata(desti_dir)
+            plugin = next(
+                (
+                    star
+                    for star in self.context.get_all_stars()
+                    if star.name == metadata_dir_name
+                ),
+                None,
+            )
+            target_plugin_path = Path(self.plugin_store_path) / (
+                plugin.root_dir_name
+                if plugin and plugin.root_dir_name
+                else metadata_dir_name
+            )
+            if plugin and plugin.reserved:
+                raise Exception("该插件是 AstrBot 保留插件，无法更新。")
+            if target_plugin_path.exists():
+                # Only replace a directory whose metadata identifies the same plugin.
+                if (
+                    target_plugin_path.is_symlink()
+                    or not target_plugin_path.is_dir()
+                    or self._get_plugin_dir_name_from_metadata(str(target_plugin_path))
+                    != metadata_dir_name
+                ):
+                    raise Exception(
+                        f"安装失败：目录 {target_plugin_path.name} 已存在。"
+                    )
+
+                metadata = self._load_plugin_metadata(desti_dir)
+                if metadata and not ignore_version_check:
+                    is_valid, error_message = self._validate_astrbot_version_specifier(
+                        metadata.astrbot_version
+                    )
+                    if not is_valid:
+                        raise PluginVersionUnsupportedError(error_message)
+                dir_name = target_plugin_path.name
+                await self._ensure_plugin_requirements(desti_dir, dir_name)
+
+                # Finish copying the backup before removing any installed files.
+                # The system temporary directory may be on a different filesystem.
+                backup_dir = Path(
+                    tempfile.mkdtemp(
+                        dir=get_astrbot_system_tmp_path(), prefix=".plugin-backup-"
+                    )
+                )
+                backup_path = backup_dir / dir_name
+                backup_complete = False
+                keep_backup = False
+                try:
+                    if plugin:
+                        try:
+                            await self._terminate_plugin(plugin)
+                        except Exception:
+                            logger.warning(
+                                "Plugin %s did not terminate cleanly",
+                                dir_name,
+                                exc_info=True,
+                            )
+                    self._cleanup_plugin_state(dir_name)
+                    shutil.copytree(target_plugin_path, backup_path, symlinks=True)
+                    backup_complete = True
+                    keep_backup = True
+                    remove_dir(str(target_plugin_path))
+                    shutil.move(desti_dir, str(target_plugin_path))
+                    desti_dir = str(target_plugin_path)
+                    success, error_message = await self.load(
+                        specified_dir_name=dir_name,
+                        ignore_version_check=ignore_version_check,
+                    )
+                    if not success:
+                        raise Exception(error_message or f"更新插件 {dir_name} 失败。")
+                    keep_backup = False
+                except BaseException:
+                    try:
+                        self._cleanup_plugin_state(dir_name)
+                        if backup_complete:
+                            if target_plugin_path.exists():
+                                remove_dir(str(target_plugin_path))
+                            shutil.copytree(
+                                backup_path, target_plugin_path, symlinks=True
+                            )
+                        restored, restore_error = await self.load(
+                            specified_dir_name=dir_name,
+                            ignore_version_check=True,
+                        )
+                        if not restored:
+                            logger.error(
+                                "Failed to reload restored plugin %s: %s",
+                                dir_name,
+                                restore_error,
+                            )
+                        else:
+                            keep_backup = False
+                            self.failed_plugin_dict.pop(dir_name, None)
+                            self._rebuild_failed_plugin_info()
+                    except Exception:
+                        logger.exception(
+                            "Failed to restore plugin %s; backup: %s",
+                            dir_name,
+                            backup_path,
+                        )
+                    raise
+                finally:
+                    if not keep_backup:
+                        try:
+                            remove_dir(str(backup_dir))
+                        except Exception:
+                            logger.warning(
+                                "Failed to remove plugin backup %s",
+                                backup_dir,
+                                exc_info=True,
+                            )
+                    else:
+                        logger.warning(
+                            "Retained plugin backup for recovery: %s", backup_path
+                        )
+            else:
+                track_failed_install = True
+                dir_name = target_plugin_path.name
+                desti_dir = str(target_plugin_path)
+                shutil.move(plugin_path, desti_dir)
+                await self._ensure_plugin_requirements(desti_dir, dir_name)
+                success, error_message = await self.load(
+                    specified_dir_name=dir_name,
+                    ignore_version_check=ignore_version_check,
+                )
+                if not success:
+                    raise Exception(
+                        error_message
+                        or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
+                    )
+
+            self.failed_plugin_dict.pop(dir_name, None)
+            self._rebuild_failed_plugin_info()
             # Get the plugin metadata to return repo info
             plugin = self.context.get_registered_star(dir_name)
             if not plugin:
@@ -2145,17 +2219,9 @@ class PluginManager:
                     "name": plugin.name,
                 }
 
-                if plugin.repo:
-                    asyncio.create_task(
-                        Metric.upload(
-                            et="install_star_f",  # install star
-                            repo=plugin.repo,
-                        ),
-                    )
-
             return plugin_info
         except Exception as e:
-            if not skip_failed_tracking:
+            if track_failed_install:
                 self._track_failed_install_dir(
                     dir_name=dir_name,
                     plugin_path=desti_dir,
@@ -2166,14 +2232,3 @@ class PluginManager:
                 f"{desti_dir}",
             )
             raise
-        finally:
-            if (skip_failed_tracking or temp_desti_dir != desti_dir) and os.path.isdir(
-                temp_desti_dir
-            ):
-                try:
-                    remove_dir(temp_desti_dir)
-                except Exception as e:
-                    logger.warning(
-                        "Failed to remove the temporary plugin extraction directory "
-                        f"{temp_desti_dir}: {e!s}",
-                    )

astrbot/core/star/updater.py

@@ -197,11 +197,33 @@ class _PluginUpdater(_RepoZipUpdater):
             "repo": str(normalized_metadata.get("repo") or normalized_url),
         }
 
-    async def install(self, repo_url: str, proxy="", download_url: str = "") -> str:
+    async def install(
+        self,
+        repo_url: str,
+        proxy="",
+        download_url: str = "",
+        *,
+        target_dir: Path | None = None,
+    ) -> str:
+        """Download or clone a plugin into a new directory.
+
+        Args:
+            repo_url: Plugin repository URL.
+            proxy: Optional proxy prefix for repository downloads.
+            download_url: Optional archive URL to use instead of the repository.
+            target_dir: Staging destination supplied by the plugin manager. Defaults
+                to the repository directory under the plugin store.
+
+        Returns:
+            Path to the extracted or cloned plugin.
+
+        Raises:
+            Exception: If the destination exists or downloading or validation fails.
+        """
         normalized_url = normalize_repository_url(repo_url)
         repository = parse_repository_url(normalized_url)
         repo_name = self._format_name(repository.name)
-        plugin_path = os.path.join(self.plugin_store_path, repo_name)
+        plugin_path = str(target_dir or Path(self.plugin_store_path) / repo_name)
         if os.path.exists(plugin_path):
             raise Exception(f"安装失败：目录 {repo_name} 已存在。")
         if download_url:

tests/fixtures/helpers.py

@@ -434,12 +434,15 @@ class MockPluginBuilder:
     def create(
         self,
         plugin_config: str | MockPluginConfig | None = None,
+        *,
+        target_dir: Path | None = None,
         **kwargs,
     ) -> Path:
         """创建模拟插件。
 
         Args:
             plugin_config: 插件名称字符串、MockPluginConfig 对象或 None
+            target_dir: Optional staging directory instead of the installed path.
             **kwargs: 如果 plugin_config 是字符串或 None，这些参数用于构建 MockPluginConfig
 
         Returns:
@@ -456,7 +459,11 @@ class MockPluginBuilder:
             raise TypeError(f"Invalid plugin_config type: {type(plugin_config)}")
 
         # 创建插件目录
-        plugin_dir = self.plugin_store_path / config.name
+        plugin_dir = (
+            target_dir
+            if target_dir is not None
+            else self.plugin_store_path / config.name
+        )
         plugin_dir.mkdir(parents=True, exist_ok=True)
 
         # 创建 metadata.yaml
@@ -548,8 +555,25 @@ def create_mock_updater_install(
         Callable: 异步函数，可用于 monkeypatch.setattr
     """
 
-    async def mock_install(repo_url: str, proxy: str = "") -> str:
-        """Mock updater.install 方法。"""
+    async def mock_install(
+        repo_url: str,
+        proxy: str = "",
+        download_url: str = "",
+        *,
+        target_dir: Path | None = None,
+    ) -> str:
+        """Create a plugin at the updater's requested destination.
+
+        Args:
+            repo_url: Repository URL used for plugin identity and metadata.
+            proxy: Unused download proxy.
+            download_url: Unused archive URL.
+            target_dir: Optional staging directory supplied by the manager.
+
+        Returns:
+            Path to the prepared plugin directory.
+        """
+        del proxy, download_url
         # 查找插件名称
         plugin_name = None
         if repo_to_plugin:
@@ -563,7 +587,7 @@ def create_mock_updater_install(
 
         # 创建插件目录
         config = MockPluginConfig(name=plugin_name, repo=repo_url)
-        plugin_dir = plugin_builder.create(config)
+        plugin_dir = plugin_builder.create(config, target_dir=target_dir)
         return str(plugin_dir)
 
     return mock_install

tests/test_plugin_manager.py

@@ -1,10 +1,14 @@
 import asyncio
+import errno
 import functools
 import json
 import os
+import sys
+import zipfile
 from pathlib import Path
 from types import ModuleType
 from typing import Any, cast
+from unittest.mock import AsyncMock
 
 import pytest
 import yaml
@@ -384,6 +388,11 @@ def plugin_manager_pm(tmp_path, monkeypatch):
         "astrbot.core.star.star_manager.get_astrbot_plugin_path",
         lambda: str(plugin_dir),
     )
+    monkeypatch.setattr(
+        star_manager_module,
+        "get_astrbot_system_tmp_path",
+        lambda: str(tmp_path / "system_temp"),
+    )
 
     return pm
 
@@ -401,18 +410,26 @@ def local_updater(plugin_manager_pm):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("dependency_install_fails", [False, True])
+@pytest.mark.parametrize("cross_filesystem", [False, True])
 async def test_install_plugin_dependency_install_flow(
-    plugin_manager_pm: PluginManager, monkeypatch, dependency_install_fails: bool
+    plugin_manager_pm: PluginManager, monkeypatch, dependency_install_fails: bool,
+    cross_filesystem: bool,
 ):
     plugin_path = Path(plugin_manager_pm.plugin_store_path) / TEST_PLUGIN_DIR
     events = []
     _mock_missing_requirements(monkeypatch, {"networkx"})
+    if cross_filesystem:
+        def cross_device_rename(*args, **kwargs):
+            raise OSError(errno.EXDEV, "Cross-device move")
 
-    async def mock_install(repo_url: str, proxy=""):
+        monkeypatch.setattr(star_manager_module.os, "rename", cross_device_rename)
+
+    async def mock_install(repo_url: str, proxy="", *, download_url="", target_dir):
         assert repo_url == TEST_PLUGIN_REPO
-        _write_local_test_plugin(plugin_path, repo_url)
-        _write_requirements(plugin_path)
-        return str(plugin_path)
+        staged_path = Path(target_dir)
+        _write_local_test_plugin(staged_path, repo_url)
+        _write_requirements(staged_path)
+        return str(staged_path)
 
     monkeypatch.setattr(plugin_manager_pm._updater, "install", mock_install)
     monkeypatch.setattr(
@@ -435,6 +452,8 @@ async def test_install_plugin_dependency_install_flow(
             expected_original_path=plugin_path / "requirements.txt",
             expected_content="networkx\n",
         )
+        assert set(plugin_manager_pm.failed_plugin_dict) == {TEST_PLUGIN_DIR}
+        assert set(Path(plugin_manager_pm.plugin_store_path).iterdir()) == {plugin_path}
     else:
         await plugin_manager_pm.install_plugin(TEST_PLUGIN_REPO)
         assert len(events) == 2
@@ -501,7 +520,7 @@ async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean(
     zip_file_path = tmp_path / "plugin_upload_helloworld_v2.zip"
     zip_file_path.write_text("placeholder", encoding="utf-8")
     plugin_store_path = Path(plugin_manager_pm.plugin_store_path)
-    existing_upload_dirs = set(plugin_store_path.glob("plugin_upload_*"))
+    existing_dirs = set(plugin_store_path.iterdir())
 
     def mock_unzip_file(zip_path: str, target_dir: str) -> None:
         assert zip_path == str(zip_file_path)
@@ -512,6 +531,10 @@ async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean(
         )
 
     assert local_updater.is_dir()
+    metadata_path = local_updater / "metadata.yaml"
+    metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
+    metadata["name"] = "another_plugin"
+    metadata_path.write_text(yaml.safe_dump(metadata), encoding="utf-8")
     monkeypatch.setattr(
         plugin_manager_pm._updater,
         "_extract_plugin_archive",
@@ -521,13 +544,453 @@ async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean(
     with pytest.raises(Exception, match=f"安装失败：目录 {TEST_PLUGIN_DIR} 已存在。"):
         await plugin_manager_pm.install_plugin_from_file(str(zip_file_path))
 
-    new_upload_dirs = [
-        upload_dir
-        for upload_dir in plugin_store_path.glob("plugin_upload_*")
-        if upload_dir not in existing_upload_dirs
-    ]
     assert plugin_manager_pm.failed_plugin_dict == {}
-    assert new_upload_dirs == []
+    assert set(plugin_store_path.iterdir()) == existing_dirs
+    assert yaml.safe_load(metadata_path.read_text(encoding="utf-8")) == metadata
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("install_source", ["upload", "github", "url", "git"])
+@pytest.mark.parametrize("legacy_directory", [False, True])
+@pytest.mark.parametrize("failure", [None, "dependencies", "load", "cancel", "version"])
+async def test_install_updates_existing_plugin_and_restores_on_failure(
+    plugin_manager_pm: PluginManager,
+    local_updater: Path,
+    monkeypatch,
+    tmp_path: Path,
+    legacy_directory: bool,
+    failure: str | None,
+    install_source: str,
+):
+    """Every install source replaces old code or restores it after a failed load."""
+    _clear_star_runtime_state()
+    if legacy_directory:
+        local_updater = local_updater.rename(local_updater.with_name("legacy_plugin"))
+    system_temp = Path(star_manager_module.get_astrbot_system_tmp_path())
+    original_rename = os.rename
+
+    def cross_device_rename(source, destination, *args, **kwargs):
+        if Path(source).is_relative_to(system_temp) != Path(destination).is_relative_to(system_temp):
+            raise OSError(errno.EXDEV, "Cross-device move")
+        return original_rename(source, destination, *args, **kwargs)
+
+    monkeypatch.setattr(star_manager_module.os, "rename", cross_device_rename)
+    dir_name = local_updater.name
+    module_path = f"data.plugins.{dir_name}.main"
+    old_plugin = star_manager_module.StarMetadata(
+        name=TEST_PLUGIN_NAME,
+        root_dir_name=dir_name,
+        module_path=module_path,
+        version="1.0.0",
+        reserved=False,
+    )
+    star_manager_module.star_registry.append(old_plugin)
+    star_manager_module.star_map[module_path] = old_plugin
+    sys.modules[module_path] = ModuleType(module_path)
+    monkeypatch.setattr(
+        plugin_manager_pm.context, "stars", star_manager_module.star_registry
+    )
+    (local_updater / "obsolete.py").write_text("old code", encoding="utf-8")
+    config_path = tmp_path / "config" / f"{dir_name}_config.json"
+    config_path.parent.mkdir()
+    config_path.write_text('{"custom": true}', encoding="utf-8")
+    monkeypatch.setattr(
+        plugin_manager_pm, "plugin_config_path", str(config_path.parent)
+    )
+    data_path = tmp_path / "plugin_data" / dir_name / "digest.db"
+    data_path.parent.mkdir(parents=True)
+    data_path.write_bytes(b"saved digest")
+
+    source_path = tmp_path / "new_version"
+    _write_local_test_plugin(source_path, TEST_PLUGIN_REPO, version="2.0.0")
+    metadata_path = source_path / "metadata.yaml"
+    metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
+    metadata.pop("repo")
+    if failure == "version":
+        metadata["astrbot_version"] = ">=999.0"
+    metadata_path.write_text(yaml.safe_dump(metadata), encoding="utf-8")
+    (source_path / "README.md").write_text("Updated README", encoding="utf-8")
+    zip_path = tmp_path / "plugin-v2.zip"
+    with zipfile.ZipFile(zip_path, "w") as archive:
+        for path in source_path.iterdir():
+            archive.write(path, f"release-v2/{path.name}")
+
+    repo_url = f"https://github.com/AstrBotDevs/{TEST_PLUGIN_DIR}"
+    if install_source == "git":
+        repo_url = f"https://gitee.com/AstrBotDevs/{TEST_PLUGIN_DIR}.git"
+    proxy_url = "https://proxy.example"
+    download_url = "https://cdn.example/plugin-v2.zip"
+
+    async def download_repository(plugin_path, url, proxy):
+        assert url == repo_url
+        assert proxy == proxy_url
+        Path(plugin_path + ".zip").write_bytes(zip_path.read_bytes())
+
+    async def download_file(url, path):
+        assert url == download_url
+        Path(path).write_bytes(zip_path.read_bytes())
+
+    async def clone_repository(url, target_path):
+        assert url == repo_url
+        target_path = Path(target_path)
+        assert not target_path.exists()
+        target_path.mkdir()
+        for path in source_path.iterdir():
+            (target_path / path.name).write_bytes(path.read_bytes())
+
+    monkeypatch.setattr(
+        plugin_manager_pm._updater, "_download_repository", download_repository
+    )
+    monkeypatch.setattr(plugin_manager_pm._updater, "_download_file", download_file)
+    monkeypatch.setattr(
+        plugin_manager_pm._updater, "_clone_repository", clone_repository
+    )
+    monkeypatch.setattr(star_manager_module.Metric, "upload", AsyncMock())
+    events = []
+
+    async def ensure_requirements(plugin_dir_path, plugin_label):
+        assert plugin_label == dir_name
+        assert (local_updater / "obsolete.py").exists()
+        assert Path(plugin_dir_path) != local_updater
+        assert Path(plugin_dir_path).is_relative_to(system_temp)
+        events.append("dependencies")
+        if failure == "dependencies":
+            raise RuntimeError("dependency failure")
+
+    async def terminate(plugin):
+        assert plugin is old_plugin
+        assert (local_updater / "obsolete.py").exists()
+        events.append("terminate")
+
+    async def load(specified_dir_name=None, ignore_version_check=False):
+        assert specified_dir_name == dir_name
+        assert module_path not in sys.modules
+        assert module_path not in star_manager_module.star_map
+        assert old_plugin not in star_manager_module.star_registry
+        current = plugin_manager_pm._load_plugin_metadata(str(local_updater))
+        assert current is not None
+        events.append(current.version)
+        if current.version == "2.0.0":
+            assert list(system_temp.glob(".plugin-backup-*"))
+            assert not (local_updater / "obsolete.py").exists()
+            if failure == "load":
+                sys.modules[module_path] = ModuleType(module_path)
+                return False, "new plugin failed to load"
+            if failure == "cancel":
+                raise asyncio.CancelledError
+        else:
+            assert ignore_version_check is True
+            assert (local_updater / "obsolete.py").read_text() == "old code"
+        current.root_dir_name = dir_name
+        current.module_path = module_path
+        star_manager_module.star_registry.append(current)
+        star_manager_module.star_map[module_path] = current
+        return True, None
+
+    monkeypatch.setattr(
+        plugin_manager_pm, "_ensure_plugin_requirements", ensure_requirements
+    )
+    monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", terminate)
+    monkeypatch.setattr(plugin_manager_pm, "load", load)
+    try:
+        if install_source == "upload":
+            operation = plugin_manager_pm.install_plugin_from_file(str(zip_path))
+        else:
+            operation = plugin_manager_pm.install_plugin(
+                repo_url,
+                proxy=proxy_url,
+                download_url=download_url if install_source == "url" else "",
+            )
+        if failure:
+            exception_type = (
+                asyncio.CancelledError if failure == "cancel" else Exception
+            )
+            with pytest.raises(exception_type):
+                await operation
+            assert (
+                plugin_manager_pm._load_plugin_metadata(str(local_updater)).version
+                == "1.0.0"
+            )
+            if failure in {"load", "cancel"}:
+                assert events == ["dependencies", "terminate", "2.0.0", "1.0.0"]
+            else:
+                assert "terminate" not in events
+                assert star_manager_module.star_map[module_path] is old_plugin
+        else:
+            result = await operation
+            assert result == {
+                "name": TEST_PLUGIN_NAME,
+                "repo": None,
+                "readme": "Updated README",
+            }
+            assert events == ["dependencies", "terminate", "2.0.0"]
+            assert zip_path.exists() is (install_source != "upload")
+        assert config_path.read_text() == '{"custom": true}'
+        assert data_path.read_bytes() == b"saved digest"
+        assert set(Path(plugin_manager_pm.plugin_store_path).iterdir()) == {
+            local_updater
+        }
+        assert plugin_manager_pm.failed_plugin_dict == {}
+        assert list(system_temp.iterdir()) == []
+    finally:
+        sys.modules.pop(module_path, None)
+        _clear_star_runtime_state()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure_stage", ["backup", "install", "restore"])
+async def test_install_copy_failure_preserves_complete_old_code(
+    plugin_manager_pm: PluginManager,
+    local_updater: Path,
+    monkeypatch,
+    tmp_path: Path,
+    failure_stage: str,
+):
+    """A partial copy must never replace a complete recovery copy."""
+    old_files = {path.name: path.read_bytes() for path in local_updater.iterdir()}
+    metadata = yaml.safe_load(old_files["metadata.yaml"])
+    metadata["version"] = "2.0.0"
+    zip_path = tmp_path / "update.zip"
+    with zipfile.ZipFile(zip_path, "w") as archive:
+        archive.writestr("metadata.yaml", yaml.safe_dump(metadata))
+        archive.writestr("main.py", "pass\n")
+
+    original_copytree = star_manager_module.shutil.copytree
+
+    def copytree(source, destination, *args, **kwargs):
+        source, destination = Path(source), Path(destination)
+        is_backup = destination.parent.name.startswith(".plugin-backup-")
+        is_restore = source.parent.name.startswith(".plugin-backup-")
+        if (failure_stage == "backup" and is_backup) or (
+            failure_stage == "restore" and is_restore
+        ):
+            destination.mkdir()
+            (destination / "partial.py").write_text("partial", encoding="utf-8")
+            raise OSError("copy failed")
+        return original_copytree(source, destination, *args, **kwargs)
+
+    original_move = star_manager_module.shutil.move
+
+    def move(source, destination):
+        if failure_stage == "install":
+            destination = Path(destination)
+            destination.mkdir()
+            (destination / "partial.py").write_text("partial", encoding="utf-8")
+            raise OSError("copy failed")
+        return original_move(source, destination)
+
+    versions_loaded = []
+
+    async def load(specified_dir_name=None, ignore_version_check=False):
+        version = plugin_manager_pm._load_plugin_metadata(str(local_updater)).version
+        versions_loaded.append(version)
+        return version == "1.0.0", "new plugin failed to load"
+
+    monkeypatch.setattr(star_manager_module.shutil, "copytree", copytree)
+    monkeypatch.setattr(star_manager_module.shutil, "move", move)
+    monkeypatch.setattr(plugin_manager_pm, "load", load)
+    with pytest.raises(Exception, match="copy failed|new plugin failed to load"):
+        await plugin_manager_pm.install_plugin_from_file(str(zip_path))
+
+    system_temp = Path(star_manager_module.get_astrbot_system_tmp_path())
+    if failure_stage == "restore":
+        backup_dirs = list(system_temp.iterdir())
+        assert len(backup_dirs) == 1
+        assert backup_dirs[0].name.startswith(".plugin-backup-")
+        backup = backup_dirs[0] / TEST_PLUGIN_DIR
+        assert {path.name: path.read_bytes() for path in backup.iterdir()} == old_files
+        assert versions_loaded == ["2.0.0"]
+    else:
+        assert {path.name: path.read_bytes() for path in local_updater.iterdir()} == old_files
+        assert versions_loaded == ["1.0.0"]
+        assert list(system_temp.iterdir()) == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("install_source", ["github", "url", "git"])
+@pytest.mark.parametrize("failure", ["download", "cancel", "metadata"])
+async def test_url_install_preparation_failure_preserves_existing_plugin(
+    plugin_manager_pm: PluginManager,
+    local_updater: Path,
+    monkeypatch,
+    install_source: str,
+    failure: str,
+):
+    """Failed downloads and invalid packages leave the existing plugin untouched."""
+    original_files = {path.name: path.read_bytes() for path in local_updater.iterdir()}
+    load = AsyncMock()
+    terminate = AsyncMock()
+    monkeypatch.setattr(plugin_manager_pm, "load", load)
+    monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", terminate)
+    monkeypatch.setattr(star_manager_module.Metric, "upload", AsyncMock())
+
+    async def prepare(*args):
+        if install_source == "git":
+            target = Path(args[1])
+            target.mkdir()
+            (target / "main.py").write_text("pass\n", encoding="utf-8")
+        else:
+            target = Path(args[0] + ".zip" if install_source == "github" else args[1])
+            with zipfile.ZipFile(target, "w") as archive:
+                archive.writestr("main.py", "pass\n")
+        if failure == "download":
+            raise RuntimeError("download interrupted")
+        if failure == "cancel":
+            raise asyncio.CancelledError
+
+    monkeypatch.setattr(plugin_manager_pm._updater, "_download_repository", prepare)
+    monkeypatch.setattr(plugin_manager_pm._updater, "_download_file", prepare)
+    monkeypatch.setattr(plugin_manager_pm._updater, "_clone_repository", prepare)
+    repo_url = (
+        f"https://gitee.com/AstrBotDevs/{TEST_PLUGIN_DIR}.git"
+        if install_source == "git"
+        else f"https://github.com/AstrBotDevs/{TEST_PLUGIN_DIR}"
+    )
+    with pytest.raises(asyncio.CancelledError if failure == "cancel" else Exception):
+        await plugin_manager_pm.install_plugin(
+            repo_url,
+            download_url="https://cdn.example/update.zip"
+            if install_source == "url"
+            else "",
+        )
+    assert {
+        path.name: path.read_bytes() for path in local_updater.iterdir()
+    } == original_files
+    assert set(Path(plugin_manager_pm.plugin_store_path).iterdir()) == {local_updater}
+    assert plugin_manager_pm.failed_plugin_dict == {}
+    load.assert_not_awaited()
+    terminate.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("install_source", ["upload", "url"])
+@pytest.mark.parametrize("disabled", [False, True])
+@pytest.mark.parametrize("initialization_fails", [False, True])
+async def test_upload_update_reloads_runtime_and_preserves_activation(
+    plugin_manager_pm: PluginManager,
+    local_updater: Path,
+    monkeypatch,
+    tmp_path: Path,
+    disabled: bool,
+    initialization_fails: bool,
+    install_source: str,
+):
+    """Exercise real loading, registration, disabled state, and rollback."""
+    _clear_star_runtime_state()
+    module_path = f"data.plugins.{TEST_PLUGIN_DIR}.main"
+    metadata_path = local_updater / "metadata.yaml"
+    metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
+    metadata.pop("repo")
+    metadata_path.write_text(yaml.safe_dump(metadata), encoding="utf-8")
+    source = (
+        "from astrbot.api.star import Star\n"
+        "from astrbot.api.event import filter\n"
+        "class Main(Star):\n"
+        "    marker = 'old'\n"
+        "    async def initialize(self):\n"
+        "        pass\n"
+        "    @filter.command('upload_update_test')\n"
+        "    async def command(self, event):\n"
+        "        pass\n"
+    )
+    (local_updater / "main.py").write_text(source, encoding="utf-8")
+    monkeypatch.setattr(
+        plugin_manager_pm.context, "stars", star_manager_module.star_registry
+    )
+    monkeypatch.setattr(
+        plugin_manager_pm, "reserved_plugin_path", str(tmp_path / "reserved")
+    )
+    monkeypatch.setattr(
+        plugin_manager_pm, "plugin_config_path", str(tmp_path / "config")
+    )
+
+    async def global_get(key, default=None):
+        if key == "inactivated_plugins":
+            return [module_path] if disabled else []
+        return default
+
+    async def import_plugin(*, path, **kwargs):
+        if path not in sys.modules:
+            module = ModuleType(path)
+            sys.modules[path] = module
+            main_path = local_updater / "main.py"
+            exec(
+                compile(main_path.read_text(encoding="utf-8"), str(main_path), "exec"),
+                module.__dict__,
+            )
+        return sys.modules[path]
+
+    async def sync_command_configs():
+        return None
+
+    monkeypatch.setattr(star_manager_module.sp, "global_get", global_get)
+    monkeypatch.setattr(
+        star_manager_module, "sync_command_configs", sync_command_configs
+    )
+    monkeypatch.setattr(
+        plugin_manager_pm, "_import_plugin_with_dependency_recovery", import_plugin
+    )
+    try:
+        success, error = await plugin_manager_pm.load(
+            specified_dir_name=TEST_PLUGIN_DIR
+        )
+        assert success, error
+        old_plugin = star_manager_module.star_map[module_path]
+        assert old_plugin.star_cls_type.marker == "old"
+
+        metadata["version"] = "2.0.0"
+        metadata["astrbot_version"] = ">=999.0"
+        updated_source = source.replace("marker = 'old'", "marker = 'new'")
+        if initialization_fails:
+            updated_source = updated_source.replace(
+                "        pass", "        raise RuntimeError('init failed')", 1
+            )
+        zip_path = tmp_path / "update.zip"
+        with zipfile.ZipFile(zip_path, "w") as archive:
+            archive.writestr("metadata.yaml", yaml.safe_dump(metadata))
+            archive.writestr("main.py", updated_source)
+
+        async def download_file(url, path):
+            Path(path).write_bytes(zip_path.read_bytes())
+
+        monkeypatch.setattr(plugin_manager_pm._updater, "_download_file", download_file)
+        monkeypatch.setattr(star_manager_module.Metric, "upload", AsyncMock())
+        if install_source == "upload":
+            operation = plugin_manager_pm.install_plugin_from_file(
+                str(zip_path), ignore_version_check=True
+            )
+        else:
+            operation = plugin_manager_pm.install_plugin(
+                TEST_PLUGIN_REPO,
+                download_url="https://cdn.example/update.zip",
+                ignore_version_check=True,
+            )
+        if initialization_fails and not disabled:
+            with pytest.raises(Exception, match="init failed"):
+                await operation
+        else:
+            await operation
+        current_plugin = star_manager_module.star_map[module_path]
+        assert current_plugin is not old_plugin
+        assert current_plugin.activated is (not disabled)
+        assert (current_plugin.star_cls is None) is disabled
+        assert current_plugin.star_cls_type.marker == (
+            "old" if initialization_fails and not disabled else "new"
+        )
+        assert len(star_manager_module.star_registry) == 1
+        assert (
+            len(
+                star_manager_module.star_handlers_registry.get_handlers_by_module_name(
+                    module_path
+                )
+            )
+            == 1
+        )
+        assert plugin_manager_pm.failed_plugin_dict == {}
+        assert plugin_manager_pm.failed_plugin_info == ""
+    finally:
+        sys.modules.pop(module_path, None)
+        _clear_star_runtime_state()
 
 
 @pytest.mark.asyncio
@@ -1561,10 +2024,11 @@ async def test_install_plugin_skips_dependency_install_when_no_requirements_miss
     events = []
     _mock_missing_requirements(monkeypatch, set())
 
-    async def mock_install(repo_url: str, proxy=""):
-        _write_local_test_plugin(plugin_path, repo_url)
-        _write_requirements(plugin_path)
-        return str(plugin_path)
+    async def mock_install(repo_url: str, proxy="", *, download_url="", target_dir):
+        staged_path = Path(target_dir)
+        _write_local_test_plugin(staged_path, repo_url)
+        _write_requirements(staged_path)
+        return str(staged_path)
 
     monkeypatch.setattr(plugin_manager_pm._updater, "install", mock_install)
     monkeypatch.setattr(
@@ -1591,10 +2055,11 @@ async def test_install_plugin_runs_dependency_install_when_precheck_fails(
     plugin_path = Path(plugin_manager_pm.plugin_store_path) / TEST_PLUGIN_DIR
     events = []
 
-    async def mock_install(repo_url: str, proxy=""):
-        _write_local_test_plugin(plugin_path, repo_url)
-        _write_requirements(plugin_path)
-        return str(plugin_path)
+    async def mock_install(repo_url: str, proxy="", *, download_url="", target_dir):
+        staged_path = Path(target_dir)
+        _write_local_test_plugin(staged_path, repo_url)
+        _write_requirements(staged_path)
+        return str(staged_path)
 
     _mock_precheck_fails(monkeypatch)
     monkeypatch.setattr(plugin_manager_pm._updater, "install", mock_install)