MoviePilot/app/adapters/system/plugin/package.py

"""
插件包文件安装和快照恢复适配器。
"""

from __future__ import annotations

import asyncio
import errno
import hashlib
import io
import shutil
import stat
import uuid
import zipfile
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any, Optional, Protocol, Sequence, cast
from urllib.parse import quote

import aiofiles
import aioshutil
from anyio import Path as AsyncPath

from app.adapters.system.host import SystemUtils
from app.adapters.system.plugin.health import PluginRuntimeHealth
from app.adapters.system.plugin.manifest import (
    PluginDependencyManifestError,
    dependency_manifest_declares_installation,
    load_dependency_manifest,
)
from app.domain.plugin import (
    PluginReleaseInstallPlan,
    build_plugin_release_install_plan,
)
from app.runtime.dependencies.native import (
    LoadedNativeDependencySnapshot,
    NativeDependencyChange,
    capture_loaded_native_dependencies,
    detect_changed_native_dependencies,
)
from app.runtime.execution import (
    plugin_package_lock,
)
from app.runtime.execution import (
    run_in_threadpool_to_completion as _await_thread_operation,
)
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
from app.runtime.version import get_app_version


class PluginContentSwapError(OSError):
    """
    插件内容换入失败，并随异常携带运行目录的实际恢复结论。

    上层能观察到的只有运行目录还在不在，而被删到一半的运行目录同样"存在"；
    只凭存在与否判断就会把残骸当成完好的旧插件，进而把本次的临时备份当作多余
    材料删掉，结果是安装报失败、运行目录却只剩半份旧文件且无处可取。因此把
    "运行目录是否仍是换入前那一份"由换入方作为事实上报，不让上层去猜。
    """

    def __init__(self, cause: BaseException, *, runtime_intact: bool) -> None:
        """
        按原始失败构造换入异常，并记录运行目录的恢复结论。

        :param cause: 触发换入失败的原始异常，消息原样透传给上层展示
        :param runtime_intact: 运行目录是否仍是换入前那一份（未触碰或已回滚到位）
        """
        super().__init__(str(cause))
        self.runtime_intact = runtime_intact


class PluginPackageSourcePort(Protocol):
    """
    插件包安装器读取本地来源、市场元数据和远端制品的接口。
    """

    def is_local_repo_url(self, repo_url: Optional[str]) -> bool:
        """
        判断来源是否为本地仓库。

        :param repo_url: 待检查的来源标识，允许为空
        :return: 符合本地仓库标识格式时返回 True，否则返回 False
        """

    def parse_local_repo_url(self, repo_url: str) -> Optional[str]:
        """
        解析本地来源中的插件标识。

        :param repo_url: 本地仓库来源标识
        :return: 其中的插件 ID；来源不符合约定时返回 None
        """

    def parse_local_repo_path(self, repo_url: str) -> Optional[Path]:
        """
        解析本地来源中的仓库目录。

        :param repo_url: 可能附带仓库路径的本地来源标识
        :return: 解析后的仓库路径；未指定或无法解析时返回 None
        """

    def parse_local_repo_package_version(self, repo_url: str) -> Optional[str]:
        """
        解析本地来源中的包代际。

        :param repo_url: 本地仓库来源标识
        :return: 显式指定的包代际；未指定时返回 None
        """

    def make_local_repo_url(
        self,
        plugin_id: str,
        repo_path: Optional[Path] = None,
        package_version: Optional[str] = None,
    ) -> str:
        """
        生成本地仓库来源标识。

        :param plugin_id: 插件标识
        :param repo_path: 显式本地仓库目录；None 时不写入路径参数
        :param package_version: 插件包代际；None 时不写入代际参数
        :return: 可供后续解析的本地仓库来源标识
        """

    def get_local_plugin_candidate(
        self,
        pid: str,
        package_version: Optional[str] = None,
        repo_path: Optional[Path] = None,
        strict_compat: bool = True,
        strict_system_version: bool = True,
    ) -> Optional[dict[str, Any]]:
        """
        读取一个本地插件候选。

        :param pid: 要查找的插件 ID
        :param package_version: 限定的包代际；None 时按已配置代际顺序查找
        :param repo_path: 限定的本地仓库；None 时扫描配置中的本地仓库
        :param strict_compat: 是否跳过与当前包代际不兼容的候选
        :param strict_system_version: 是否把宿主版本不匹配投影为不可安装候选
        :return: 含路径和兼容标志的候选信息；未找到时返回 None，返回候选仍需检查兼容标志
        """

    def check_plugin_system_version(
        self, plugin_info: dict[str, Any]
    ) -> tuple[bool, str]:
        """
        校验插件声明的宿主版本约束。

        :param plugin_info: 含插件宿主版本要求的元数据
        :return: 是否允许当前宿主安装及不兼容原因
        """

    def check_plugin_install_compatibility(
        self, plugin_info: dict[str, Any]
    ) -> tuple[bool, str]:
        """
        校验插件与当前运行时和宿主版本是否都允许安装。

        :param plugin_info: 含运行时和宿主版本要求的插件元数据
        :return: 是否满足安装兼容要求及失败说明；先检查运行时，再检查宿主版本
        """

    def get_plugin_package_version(
        self, plugin_id: str, repo_url: str, package_version: Optional[str]
    ) -> Optional[str]:
        """
        选择适用于当前宿主的远端索引代际。

        :param plugin_id: 要在索引中查找的插件 ID
        :param repo_url: 远端插件仓库地址
        :param package_version: 首选索引代际；None 使用宿主当前代际并依次尝试兼容代际
        :return: 可用代际；空字符串表示旧版 package.json，None 表示没有可安装候选
        """

    async def async_get_plugin_package_version(
        self, plugin_id: str, repo_url: str, package_version: Optional[str]
    ) -> Optional[str]:
        """
        异步选择适用于当前宿主的远端索引代际。

        :param plugin_id: 要在索引中查找的插件 ID
        :param repo_url: 远端插件仓库地址
        :param package_version: 首选索引代际；None 使用宿主当前代际并依次尝试兼容代际
        :return: 可用代际；空字符串表示旧版 package.json，None 表示没有可安装候选
        """

    def get_plugins(
        self, repo_url: str, package_version: Optional[str] = None
    ) -> Optional[dict[str, dict[str, Any]]]:
        """
        读取远端插件索引。

        :param repo_url: 远端插件仓库地址
        :param package_version: 要读取的索引代际；None 表示旧版 package.json
        :return: 插件 ID 到元数据的映射；索引不存在时为空字典，请求或解析失败时为 None
        """

    async def async_get_plugins(
        self, repo_url: str, package_version: Optional[str] = None
    ) -> Optional[dict[str, dict[str, Any]]]:
        """
        异步读取远端插件索引。

        :param repo_url: 远端插件仓库地址
        :param package_version: 要读取的索引代际；None 表示旧版 package.json
        :return: 插件 ID 到元数据的映射；索引不存在时为空字典，请求或解析失败时为 None
        """

    def get_plugin_release_versions(
        self, plugin_id: str, repo_url: str
    ) -> list[dict[str, Any]]:
        """
        读取插件可安装 Release。

        :param plugin_id: 用于筛选发布标签和资产名的插件 ID
        :param repo_url: GitHub 插件仓库地址
        :return: 符合标签和 ZIP 资产命名约定的发布列表；没有可用结果时为空列表
        """

    async def async_get_plugin_release_versions(
        self, plugin_id: str, repo_url: str
    ) -> list[dict[str, Any]]:
        """
        异步读取插件可安装 Release。

        :param plugin_id: 用于筛选发布标签和资产名的插件 ID
        :param repo_url: GitHub 插件仓库地址
        :return: 符合标签和 ZIP 资产命名约定的发布列表；没有可用结果时为空列表
        """

    def get_repo_info(self, repo_url: str) -> tuple[Optional[str], Optional[str]]:
        """
        解析 GitHub 仓库所有者和仓库名。

        :param repo_url: GitHub 仓库地址
        :return: 所有者与仓库名二元组；空值或结构不足时两项均为 None
        """

    def request(
        self,
        url: str,
        *,
        headers: Optional[dict[str, str]] = None,
        timeout: Optional[int] = 60,
        is_api: bool = False,
        mirror_url: Optional[str] = None,
    ) -> Any:
        """
        同步读取 GitHub 元数据或制品。

        :param url: GitHub 元数据或制品地址
        :param headers: 附加请求头；None 使用请求适配器默认值
        :param timeout: 每次请求超时，单位为秒，默认 60；None 沿用传输实现约定
        :param is_api: 是否为 API 请求；True 且未提供公开资产地址时跳过镜像
        :param mirror_url: 可选的公开 Release 资产地址，镜像匿名下载失败后回退原始地址
        :return: 已读取内容的响应；所有传输策略均失败时返回 None
        """

    async def async_request(
        self,
        url: str,
        *,
        headers: Optional[dict[str, str]] = None,
        timeout: Optional[int] = 60,
        is_api: bool = False,
        mirror_url: Optional[str] = None,
    ) -> Any:
        """
        异步读取 GitHub 元数据或制品。

        :param url: GitHub 元数据或制品地址
        :param headers: 附加请求头；None 使用请求适配器默认值
        :param timeout: 每次请求超时，单位为秒，默认 60；None 沿用传输实现约定
        :param is_api: 是否为 API 请求；True 且未提供公开资产地址时跳过镜像
        :param mirror_url: 可选的公开 Release 资产地址，镜像匿名下载失败后回退原始地址
        :return: 已读取内容的响应；所有传输策略均失败时返回 None
        """


@dataclass(slots=True)
class PluginPackageCheckpoint:
    """
    记录插件运行目录的快照和待提升的容器恢复备份。

    :param plugin_id: 需要恢复的插件标识
    :param plugin_dir: 安装前后的插件运行目录
    :param persistent_backup_dir: 用于容器重建恢复的持久备份目录
    :param backup_staging_dir: 等待提交的新备份暂存目录；未创建时为 None
    :param backup_previous_dir: 提升新备份时保留的旧备份目录；未创建时为 None
    :param transaction_dir: 本次安装事务的检查点目录，保存运行文件快照与补偿标记
    :param plugin_existed: 创建检查点时运行目录是否存在
    :param persistent_backup_existed: 创建检查点时持久备份是否存在
    :param native_dependencies: 安装前已载入原生依赖的快照，供安装后检测变动；未采集时为 None
    """

    plugin_id: str
    plugin_dir: Path
    persistent_backup_dir: Path
    backup_staging_dir: Path | None
    backup_previous_dir: Path | None
    transaction_dir: Path
    plugin_existed: bool
    persistent_backup_existed: bool
    native_dependencies: LoadedNativeDependencySnapshot | None = None

    @property
    def existed(self) -> bool:
        """
        保留旧调用方读取运行目录存在状态的兼容属性。
        """
        return self.plugin_existed

    @property
    def rollback_marker(self) -> Path:
        """
        返回文件补偿完成标记，供 PREPARED 重放保持幂等。
        """
        return self.transaction_dir / ".rollback-complete"


@dataclass(frozen=True, slots=True)
class _RemotePluginInstallSelection:
    """
    冻结远端插件通过准入检查后的来源和安装方式。

    :param user_repo: 已确认的远端仓库地址
    :param package_version: 已选定的插件索引代际
    :param release_tag: 要安装的发布标签；None 表示不使用发布包
    :param fallback_to_filelist: 发布包不可用时是否允许回退到索引文件清单
    """

    user_repo: str
    package_version: str
    release_tag: Optional[str]
    fallback_to_filelist: bool


class PluginPackageManager:
    """
    隔离插件包安装、本地同步、分身改写和文件补偿能力。
    """

    _COPY_IGNORE = ("__pycache__", "*.pyc", ".DS_Store", "node_modules")

    def __init__(
        self,
        source: Optional[PluginPackageSourcePort] = None,
        *,
        health: Optional[PluginRuntimeHealth] = None,
        plugin_root: Optional[Path] = None,
    ) -> None:
        """
        保存外部来源端口和依赖健康 owner。
        """
        self._source = source
        self._health = health or PluginRuntimeHealth()
        self._plugin_root = plugin_root.resolve() if plugin_root else None

    def _require_source(self) -> PluginPackageSourcePort:
        """
        返回已装配来源端口，未完成组合时拒绝执行包写入。
        """
        if self._source is None:
            raise RuntimeError("插件包来源客户端尚未完成装配")
        return self._source

    def _plugins_root(self) -> Path:
        """
        返回组合根注入目录；旧直接构造调用按当前运行配置解析。
        """
        return self._plugin_root or (
            Path(get_runtime_setting('ROOT_PATH')) / "app" / "plugins"
        ).resolve()

    def __plugin_dir(self, plugin_id: str) -> Path:
        """
        解析插件运行目录并拒绝越出宿主插件根目录的标识。
        """
        plugins_root = self._plugins_root()
        plugin_dir = (plugins_root / plugin_id.lower()).resolve()
        if plugin_dir == plugins_root or not plugin_dir.is_relative_to(plugins_root):
            raise ValueError(f"非法插件ID：{plugin_id}")
        return plugin_dir

    def checkpoint(
        self,
        plugin_id: str,
        transaction_id: Optional[str] = None,
    ) -> PluginPackageCheckpoint:
        """
        在包变更前保存运行目录；持久事务使用配置目录承载恢复材料。
        """
        plugin_dir = self.__plugin_dir(plugin_id)
        durable = transaction_id is not None
        persistent_backup_dir = (
            Path(get_runtime_setting('CONFIG_PATH'))
            / "plugins_backup"
            / plugin_id.lower()
        ).resolve()
        backup_staging_dir = (
            persistent_backup_dir.parent
            / f".{plugin_id.lower()}.staging-{transaction_id}"
            if durable and SystemUtils.is_docker()
            else None
        )
        backup_previous_dir = (
            persistent_backup_dir.parent
            / f".{plugin_id.lower()}.previous-{transaction_id}"
            if durable and SystemUtils.is_docker()
            else None
        )
        transaction_root = (
            Path(get_runtime_setting('CONFIG_PATH'))
            if durable
            else Path(get_runtime_setting('TEMP_PATH'))
        )
        transaction_dir = (
            transaction_root
            / "plugin_transactions"
            / (transaction_id or f"{plugin_id.lower()}-{uuid.uuid4().hex}")
        )
        plugin_existed = plugin_dir.exists()
        persistent_backup_existed = persistent_backup_dir.exists()
        try:
            transaction_dir.mkdir(parents=True, exist_ok=False)
            if plugin_existed:
                shutil.copytree(plugin_dir, transaction_dir / "package")
        except Exception:
            shutil.rmtree(transaction_dir, ignore_errors=True)
            raise
        return PluginPackageCheckpoint(
            plugin_id=plugin_id,
            plugin_dir=plugin_dir,
            persistent_backup_dir=persistent_backup_dir,
            backup_staging_dir=backup_staging_dir,
            backup_previous_dir=backup_previous_dir,
            transaction_dir=transaction_dir,
            plugin_existed=plugin_existed,
            persistent_backup_existed=persistent_backup_existed,
        )

    def restore_checkpoint(
        self,
        *,
        plugin_id: str,
        transaction_id: str,
        plugin_existed: bool,
        persistent_backup_existed: bool,
    ) -> PluginPackageCheckpoint:
        """
        按受控根目录和事务 ID 重建崩溃回放所需的文件引用。
        """
        plugin_dir = self.__plugin_dir(plugin_id)
        persistent_backup_dir = (
            Path(get_runtime_setting('CONFIG_PATH'))
            / "plugins_backup"
            / plugin_id.lower()
        ).resolve()
        durable_backup = SystemUtils.is_docker()
        return PluginPackageCheckpoint(
            plugin_id=plugin_id,
            plugin_dir=plugin_dir,
            persistent_backup_dir=persistent_backup_dir,
            backup_staging_dir=(
                persistent_backup_dir.parent
                / f".{plugin_id.lower()}.staging-{transaction_id}"
                if durable_backup
                else None
            ),
            backup_previous_dir=(
                persistent_backup_dir.parent
                / f".{plugin_id.lower()}.previous-{transaction_id}"
                if durable_backup
                else None
            ),
            transaction_dir=(
                Path(get_runtime_setting('CONFIG_PATH'))
                / "plugin_transactions"
                / transaction_id
            ),
            plugin_existed=plugin_existed,
            persistent_backup_existed=persistent_backup_existed,
        )

    async def async_checkpoint(
        self,
        plugin_id: str,
        transaction_id: Optional[str] = None,
    ) -> PluginPackageCheckpoint:
        """
        在线程池中创建插件包文件快照。
        """
        return cast(
            PluginPackageCheckpoint,
            await _await_thread_operation(
                self.checkpoint,
                plugin_id,
                transaction_id,
            ),
        )

    @staticmethod
    def commit(checkpoint: PluginPackageCheckpoint) -> None:
        """
        清理已完成事务的运行目录快照和残余替换材料。
        """
        if checkpoint.backup_staging_dir and checkpoint.backup_staging_dir.exists():
            raise RuntimeError("持久备份尚未提升，不能清理插件安装事务")
        if checkpoint.backup_previous_dir and checkpoint.backup_previous_dir.exists():
            raise RuntimeError("旧持久备份尚未清理，不能结束插件安装事务")
        if checkpoint.transaction_dir.exists():
            shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False)

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

    @staticmethod
    def native_dependency_changes(
        checkpoint: PluginPackageCheckpoint,
    ) -> tuple[NativeDependencyChange, ...]:
        """
        返回安装中被替换、但当前进程仍持有旧代码的原生发行包。
        """
        if checkpoint.native_dependencies is None:
            return ()
        try:
            return detect_changed_native_dependencies(checkpoint.native_dependencies)
        # 诊断失败不能改写安装结果
        except Exception as error:  # noqa: BLE001
            logger.warning("检测插件原生依赖变更失败：%s", error)
            return ()

    async def async_native_dependency_changes(
        self,
        checkpoint: PluginPackageCheckpoint,
    ) -> tuple[NativeDependencyChange, ...]:
        """
        在线程池中比较原生发行包，避免文件枚举阻塞事件循环。
        """
        return cast(
            tuple[NativeDependencyChange, ...],
            await _await_thread_operation(
                self.native_dependency_changes,
                checkpoint,
            ),
        )

    @staticmethod
    def rollback(checkpoint: PluginPackageCheckpoint) -> None:
        """
        兼容旧调用方，恢复运行目录和持久备份后清理恢复材料。
        """
        PluginPackageManager.restore(checkpoint)
        PluginPackageManager.cleanup(checkpoint)

    @staticmethod
    def restore(checkpoint: PluginPackageCheckpoint) -> None:
        """
        恢复运行目录和提交前持久备份，并保留快照直到 journal 删除。
        """
        if checkpoint.rollback_marker.is_file():
            return
        PluginPackageManager.__restore_tree(
            target=checkpoint.plugin_dir,
            snapshot=checkpoint.transaction_dir / "package",
            existed=checkpoint.plugin_existed,
            label=f"插件 {checkpoint.plugin_id} 运行目录",
        )
        PluginPackageManager.__rollback_persistent_backup(checkpoint)
        if checkpoint.backup_staging_dir and checkpoint.backup_staging_dir.exists():
            shutil.rmtree(checkpoint.backup_staging_dir, ignore_errors=False)
        checkpoint.transaction_dir.mkdir(parents=True, exist_ok=True)
        checkpoint.rollback_marker.touch(exist_ok=True)

    @staticmethod
    def cleanup(checkpoint: PluginPackageCheckpoint) -> None:
        """
        在 journal 已删除后清理恢复材料；重复调用保持幂等。
        """
        if checkpoint.transaction_dir.exists():
            shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False)

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

    async def async_restore(self, checkpoint: PluginPackageCheckpoint) -> None:
        """
        在线程池恢复插件状态，并保留 journal 仍需引用的材料。
        """
        await _await_thread_operation(self.restore, checkpoint)

    async def async_cleanup(self, checkpoint: PluginPackageCheckpoint) -> None:
        """
        在线程池清理已失去 journal 所有权的恢复材料。
        """
        await _await_thread_operation(self.cleanup, checkpoint)

    @staticmethod
    def __rollback_persistent_backup(
        checkpoint: PluginPackageCheckpoint,
    ) -> None:
        """
        把已激活但尚未提交的持久备份恢复到事务前状态。
        """
        previous = checkpoint.backup_previous_dir
        staging = checkpoint.backup_staging_dir
        if previous is None or staging is None:
            return

        target = checkpoint.persistent_backup_dir
        if previous.exists():
            discarded = target.parent / f".{target.name}.discard-{uuid.uuid4().hex}"
            try:
                if target.exists():
                    target.replace(discarded)
                previous.replace(target)
                if discarded.exists():
                    shutil.rmtree(discarded, ignore_errors=False)
            except Exception:
                if not target.exists() and discarded.exists():
                    discarded.replace(target)
                raise
            finally:
                if target.exists() and discarded.exists():
                    shutil.rmtree(discarded, ignore_errors=True)
            return

        if staging.exists():
            return
        if checkpoint.persistent_backup_existed:
            if target.exists():
                return
            raise FileNotFoundError(
                f"插件 {checkpoint.plugin_id} 的旧持久备份恢复材料不存在"
            )
        if target.exists():
            shutil.rmtree(target, ignore_errors=False)

    @staticmethod
    def __restore_tree(
        *,
        target: Path,
        snapshot: Path,
        existed: bool,
        label: str,
    ) -> None:
        """
        用同级 staging 替换目录，并兼容 overlayfs 的跨设备替换。
        """
        with plugin_package_lock(target.name):
            PluginPackageManager.__restore_tree_locked(
                target=target,
                snapshot=snapshot,
                existed=existed,
                label=label,
            )

    @staticmethod
    def __restore_tree_locked(
        *,
        target: Path,
        snapshot: Path,
        existed: bool,
        label: str,
    ) -> None:
        """
        在持有插件包锁时用同级 staging 替换目录。
        """
        if existed and not snapshot.is_dir():
            raise FileNotFoundError(f"{label}补偿快照不存在：{snapshot}")

        target.parent.mkdir(parents=True, exist_ok=True)
        staging = target.parent / f".{target.name}.restore-{uuid.uuid4().hex}"
        previous = target.parent / f".{target.name}.previous-{uuid.uuid4().hex}"
        previous_available = False
        published = False
        try:
            if existed:
                shutil.copytree(snapshot, staging)
            if target.exists():
                try:
                    target.replace(previous)
                except OSError as error:
                    if error.errno != errno.EXDEV:
                        raise
                    # overlayfs 可能拒绝把镜像层目录直接 rename 到可写层，
                    # 先复制旧目标保留回滚材料，再删除旧目录继续发布快照。
                    if target.is_dir():
                        shutil.copytree(target, previous, symlinks=True)
                    else:
                        shutil.copy2(target, previous, follow_symlinks=False)
                    previous_available = True
                    PluginPackageManager.__remove_snapshot_path(target)
                else:
                    previous_available = True
            if existed:
                staging.replace(target)
            published = True
        except Exception:
            if previous_available and not published:
                try:
                    PluginPackageManager.__remove_snapshot_path(target)
                    previous.replace(target)
                    previous_available = False
                except Exception as rollback_error:
                    logger.error(
                        f"恢复{label}旧目录失败，已保留恢复材料 {previous}: "
                        f"{rollback_error}"
                    )
            raise
        finally:
            if staging.exists():
                shutil.rmtree(staging, ignore_errors=True)
            if published and previous.exists():
                shutil.rmtree(previous, ignore_errors=True)

    @staticmethod
    def __remove_snapshot_path(path: Path) -> None:
        """
        删除待替换的当前路径，保留快照材料供失败回滚。
        """
        if path.is_dir() and not path.is_symlink():
            shutil.rmtree(path)
        elif path.exists() or path.is_symlink():
            path.unlink()

    @classmethod
    def stage_persistent_backup(cls, checkpoint: PluginPackageCheckpoint) -> None:
        """
        把新载荷复制到持久配置目录的独立 staging，不覆盖现有备份。
        """
        staging = checkpoint.backup_staging_dir
        if staging is None:
            return
        if not checkpoint.plugin_dir.is_dir():
            raise FileNotFoundError(
                f"插件 {checkpoint.plugin_id} 运行目录不存在"
            )
        staging.parent.mkdir(parents=True, exist_ok=True)
        if staging.exists():
            shutil.rmtree(staging, ignore_errors=False)
        shutil.copytree(
            checkpoint.plugin_dir,
            staging,
            ignore=shutil.ignore_patterns(*cls._COPY_IGNORE),
        )

    async def async_stage_persistent_backup(
        self,
        checkpoint: PluginPackageCheckpoint,
    ) -> None:
        """
        在线程池准备新载荷的容器恢复备份。
        """
        await _await_thread_operation(self.stage_persistent_backup, checkpoint)

    @staticmethod
    def activate_persistent_backup(checkpoint: PluginPackageCheckpoint) -> None:
        """
        在数据库提交前激活新备份，并保留上一份备份供失败补偿。
        """
        staging = checkpoint.backup_staging_dir
        previous = checkpoint.backup_previous_dir
        if staging is None or previous is None:
            return

        target = checkpoint.persistent_backup_dir
        target.parent.mkdir(parents=True, exist_ok=True)
        if staging.exists():
            if target.exists() and not previous.exists():
                target.replace(previous)
            if not target.exists():
                staging.replace(target)
        elif not target.exists():
            raise FileNotFoundError(
                f"插件 {checkpoint.plugin_id} 的持久备份 staging 不存在"
            )

    async def async_activate_persistent_backup(
        self,
        checkpoint: PluginPackageCheckpoint,
    ) -> None:
        """
        在线程池激活新持久备份，同时保留失败补偿材料。
        """
        await _await_thread_operation(self.activate_persistent_backup, checkpoint)

    @staticmethod
    def finalize_persistent_backup(checkpoint: PluginPackageCheckpoint) -> None:
        """
        数据库提交后清理上一份持久备份；重复调用保持幂等。
        """
        staging = checkpoint.backup_staging_dir
        previous = checkpoint.backup_previous_dir
        if staging is None or previous is None:
            return
        if staging.exists():
            raise RuntimeError("新持久备份尚未激活")
        if not checkpoint.persistent_backup_dir.is_dir():
            raise FileNotFoundError(
                f"插件 {checkpoint.plugin_id} 的已提交持久备份不存在"
            )
        if previous.exists():
            shutil.rmtree(previous, ignore_errors=False)

    async def async_finalize_persistent_backup(
        self,
        checkpoint: PluginPackageCheckpoint,
    ) -> None:
        """
        在线程池清理数据库提交后的旧持久备份。
        """
        await _await_thread_operation(self.finalize_persistent_backup, checkpoint)

    def payload_receipt(self, plugin_id: str) -> str:
        """
        按稳定相对路径和文件内容计算已安装载荷收据。
        """
        plugin_dir = self.__plugin_dir(plugin_id)
        if not plugin_dir.is_dir():
            raise FileNotFoundError(f"插件 {plugin_id} 运行目录不存在")
        return self.__tree_receipt(plugin_dir)

    @classmethod
    def persistent_backup_receipt(
        cls,
        checkpoint: PluginPackageCheckpoint,
    ) -> str:
        """
        计算已提升持久备份的内容收据，供崩溃回放确认终态。
        """
        if not checkpoint.persistent_backup_dir.is_dir():
            raise FileNotFoundError(
                f"插件 {checkpoint.plugin_id} 持久备份不存在"
            )
        return cls.__tree_receipt(checkpoint.persistent_backup_dir)

    @classmethod
    def __tree_receipt(cls, root: Path) -> str:
        """
        对插件目录使用稳定路径和文件内容生成审计收据。
        """

        digest = hashlib.sha256()
        for path in sorted(
            root.rglob("*"),
            key=lambda item: item.relative_to(root).as_posix(),
        ):
            relative = path.relative_to(root).as_posix()
            if cls.__ignored_receipt_path(path, root):
                continue
            encoded_path = relative.encode("utf-8")
            digest.update(len(encoded_path).to_bytes(4, "big"))
            digest.update(encoded_path)
            if path.is_symlink():
                digest.update(b"L")
                target = path.readlink().as_posix().encode("utf-8")
                digest.update(len(target).to_bytes(4, "big"))
                digest.update(target)
            elif path.is_dir():
                digest.update(b"D")
            elif path.is_file():
                digest.update(b"F")
                with path.open("rb") as file_handle:
                    for chunk in iter(lambda: file_handle.read(1024 * 1024), b""):
                        digest.update(chunk)
        return f"sha256:{digest.hexdigest()}"

    async def async_payload_receipt(self, plugin_id: str) -> str:
        """
        在线程池计算插件载荷收据。
        """
        return cast(
            str,
            await _await_thread_operation(self.payload_receipt, plugin_id),
        )

    async def async_committed_payload_receipt(
        self,
        checkpoint: PluginPackageCheckpoint,
    ) -> str:
        """
        读取数据库已提交载荷在当前部署模式下的恢复事实。
        """
        if checkpoint.backup_staging_dir is not None:
            return cast(
                str,
                await _await_thread_operation(
                    self.persistent_backup_receipt,
                    checkpoint,
                ),
            )
        return await self.async_payload_receipt(checkpoint.plugin_id)

    @classmethod
    def __ignored_receipt_path(cls, path: Path, root: Path) -> bool:
        """
        排除不会进入运行载荷和持久备份的派生文件。
        """
        relative_parts = path.relative_to(root).parts
        return any(
            part in {"__pycache__", "node_modules", ".DS_Store"}
            or part.endswith(".pyc")
            for part in relative_parts
        )

    def is_local_repo_url(self, repo_url: Optional[str]) -> bool:
        """
        委托外部来源客户端判断本地仓库标识。
        """
        return self._require_source().is_local_repo_url(repo_url)

    def parse_local_repo_url(self, repo_url: str) -> Optional[str]:
        """
        委托外部来源客户端解析本地插件标识。
        """
        return self._require_source().parse_local_repo_url(repo_url)

    def parse_local_repo_path(self, repo_url: str) -> Optional[Path]:
        """
        委托外部来源客户端解析本地仓库路径。
        """
        return self._require_source().parse_local_repo_path(repo_url)

    def parse_local_repo_package_version(self, repo_url: str) -> Optional[str]:
        """
        委托外部来源客户端解析本地包代际。
        """
        return self._require_source().parse_local_repo_package_version(repo_url)

    def make_local_repo_url(
        self,
        plugin_id: str,
        repo_path: Optional[Path] = None,
        package_version: Optional[str] = None,
    ) -> str:
        """
        委托外部来源客户端生成本地来源标识。
        """
        return self._require_source().make_local_repo_url(
            plugin_id, repo_path, package_version
        )

    def get_local_plugin_candidate(
        self,
        pid: str,
        package_version: Optional[str] = None,
        repo_path: Optional[Path] = None,
        strict_compat: bool = True,
        strict_system_version: bool = True,
    ) -> Optional[dict[str, Any]]:
        """
        委托外部来源客户端读取本地插件候选。
        """
        return self._require_source().get_local_plugin_candidate(
            pid=pid,
            package_version=package_version,
            repo_path=repo_path,
            strict_compat=strict_compat,
            strict_system_version=strict_system_version,
        )

    def check_plugin_system_version(
        self, plugin_info: dict[str, Any]
    ) -> tuple[bool, str]:
        """
        委托外部来源客户端校验宿主版本约束。
        """
        return self._require_source().check_plugin_system_version(plugin_info)

    def check_plugin_install_compatibility(
        self, plugin_info: dict[str, Any]
    ) -> tuple[bool, str]:
        """
        委托外部来源客户端校验运行时与宿主版本约束。
        """
        return self._require_source().check_plugin_install_compatibility(plugin_info)

    def get_plugin_package_version(
        self, plugin_id: str, repo_url: str, package_version: Optional[str]
    ) -> Optional[str]:
        """
        委托外部来源客户端选择远端索引代际。
        """
        return self._require_source().get_plugin_package_version(
            plugin_id, repo_url, package_version
        )

    async def async_get_plugin_package_version(
        self, plugin_id: str, repo_url: str, package_version: Optional[str]
    ) -> Optional[str]:
        """
        异步委托外部来源客户端选择远端索引代际。
        """
        return await self._require_source().async_get_plugin_package_version(
            plugin_id, repo_url, package_version
        )

    def get_plugins(
        self, repo_url: str, package_version: Optional[str] = None
    ) -> Optional[dict[str, dict[str, Any]]]:
        """
        委托外部来源客户端读取远端插件索引。
        """
        return self._require_source().get_plugins(repo_url, package_version)

    async def async_get_plugins(
        self, repo_url: str, package_version: Optional[str] = None
    ) -> Optional[dict[str, dict[str, Any]]]:
        """
        异步委托外部来源客户端读取远端插件索引。
        """
        return await self._require_source().async_get_plugins(
            repo_url, package_version
        )

    def get_plugin_release_versions(
        self, plugin_id: str, repo_url: str
    ) -> list[dict[str, Any]]:
        """
        委托外部来源客户端读取可安装 Release。
        """
        return self._require_source().get_plugin_release_versions(
            plugin_id, repo_url
        )

    async def async_get_plugin_release_versions(
        self, plugin_id: str, repo_url: str
    ) -> list[dict[str, Any]]:
        """
        异步委托外部来源客户端读取可安装 Release。
        """
        return await self._require_source().async_get_plugin_release_versions(
            plugin_id, repo_url
        )

    def get_repo_info(self, repo_url: str) -> tuple[Optional[str], Optional[str]]:
        """
        委托外部来源客户端解析 GitHub 仓库信息。
        """
        return self._require_source().get_repo_info(repo_url)

    def __request_with_fallback(
        self,
        url: str,
        headers: Optional[dict[str, str]] = None,
        timeout: Optional[int] = 60,
        is_api: bool = False,
        mirror_url: Optional[str] = None,
    ) -> Any:
        """
        通过外部来源客户端同步读取远端内容。

        :param mirror_url: 可选的公开 Release 资产地址，由来源客户端隔离镜像凭据
        """
        return self._require_source().request(
            url, headers=headers, timeout=timeout, is_api=is_api, mirror_url=mirror_url
        )

    async def __async_request_with_fallback(
        self,
        url: str,
        headers: Optional[dict[str, str]] = None,
        timeout: Optional[int] = 60,
        is_api: bool = False,
        mirror_url: Optional[str] = None,
    ) -> Any:
        """
        通过外部来源客户端异步读取远端内容。

        :param mirror_url: 可选的公开 Release 资产地址，由来源客户端隔离镜像凭据
        """
        return await self._require_source().async_request(
            url, headers=headers, timeout=timeout, is_api=is_api, mirror_url=mirror_url
        )

    @staticmethod
    def _build_remote_plugin_install_plan(
        *,
        pid: str,
        meta: dict[str, Any],
        release_version: Optional[str] = None,
        release_items: Sequence[dict[str, Any]] = (),
    ) -> tuple[Optional[PluginReleaseInstallPlan], str]:
        """
        使用纯领域规则选择 Release 或文件列表安装计划。

        文件列表偏好只影响默认安装，显式选择的 Release 仍按指定版本安装。
        """
        return build_plugin_release_install_plan(
            plugin_id=pid,
            metadata=(
                meta
                if release_version or not meta.get("release") or get_runtime_setting('PLUGIN_INSTALL_FROM_RELEASE')
                else {**meta, "release": False}
            ),
            release_version=release_version,
            release_items=release_items,
            current_version=get_app_version(),
        )

    def _resolve_remote_plugin_repository(
        self,
        pid: str,
        repo_url: str,
    ) -> tuple[Optional[str], Optional[tuple[bool, str]]]:
        """
        统一校验远端安装前置条件并返回规范仓库标识。
        """
        if SystemUtils.is_frozen():
            return None, (False, "可执行文件模式下，只能安装本地插件")
        if not pid or not repo_url:
            return None, (False, "参数错误")
        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return None, (False, "不支持的插件仓库地址格式")
        return f"{user}/{repo}", None

    @staticmethod
    def _accept_remote_package_version(
        pid: str,
        package_version: Optional[str],
    ) -> Optional[tuple[bool, str]]:
        """
        统一记录索引代际选择，并映射找不到兼容插件的失败结果。
        """
        if package_version is None:
            message = f"{pid} 没有找到适用于当前版本的插件"
            logger.debug(message)
            return False, message
        if package_version:
            logger.debug(
                f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件"
            )
        else:
            logger.debug(f"{pid} 从 package.json 中找到适用于当前版本的插件")
        return None

    @classmethod
    def _select_remote_install(
        cls,
        *,
        pid: str,
        user_repo: str,
        package_version: str,
        meta: dict[str, Any],
        release_version: Optional[str],
        release_items: Sequence[dict[str, Any]],
    ) -> tuple[Optional[_RemotePluginInstallSelection], str]:
        """
        把元数据和 Release 列表统一映射为可执行的远端安装选择。
        """
        plan, message = cls._build_remote_plugin_install_plan(
            pid=pid,
            meta=meta,
            release_version=release_version,
            release_items=release_items,
        )
        if plan is None:
            return None, message
        return (
            _RemotePluginInstallSelection(
                user_repo=user_repo,
                package_version=package_version,
                release_tag=plan.release_tag,
                fallback_to_filelist=plan.fallback_to_filelist,
            ),
            "",
        )

    def install_packages_with_fallback(
        self,
        dependency_files: Path | Sequence[Path],
        find_links_dirs: Optional[list[Path]] = None,
    ) -> tuple[bool, str]:
        """
        把插件依赖清单交给运行环境健康 owner 安装。
        """
        return self._health.install_packages_with_fallback(
            dependency_files, find_links_dirs
        )

    async def async_install_packages_with_fallback(
        self,
        dependency_files: Path | Sequence[Path],
        find_links_dirs: Optional[list[Path]] = None,
    ) -> tuple[bool, str]:
        """
        异步把插件依赖清单交给运行环境健康 owner 安装。
        """
        return await self._health.async_install_packages_with_fallback(
            dependency_files, find_links_dirs
        )

    def install(
        self,
        plugin_id: str,
        repo_url: str,
        package_version: Optional[str] = None,
        release_version: Optional[str] = None,
        force_install: bool = False,
        checkpoint: PluginPackageCheckpoint | None = None,
    ) -> tuple[bool, str]:
        """
        同步安装插件包；已有事务快照时禁止兼容层再创建第二份备份。
        """
        return self.install_raw(
            pid=plugin_id,
            repo_url=repo_url,
            package_version=package_version,
            release_version=release_version,
            force_install=force_install or checkpoint is not None,
            before_dependency_install=(
                (lambda: self.__capture_native_dependencies(checkpoint))
                if checkpoint is not None
                else None
            ),
        )

    async def async_install(
        self,
        plugin_id: str,
        repo_url: str,
        package_version: Optional[str] = None,
        release_version: Optional[str] = None,
        force_install: bool = False,
        checkpoint: PluginPackageCheckpoint | None = None,
    ) -> tuple[bool, str]:
        """
        异步安装插件包；已有事务快照时禁止兼容层再创建第二份备份。
        """
        return await self.async_install_raw(
            pid=plugin_id,
            repo_url=repo_url,
            package_version=package_version,
            release_version=release_version,
            force_install=force_install or checkpoint is not None,
            before_dependency_install=(
                (lambda: self.__capture_native_dependencies(checkpoint))
                if checkpoint is not None
                else None
            ),
        )

    @staticmethod
    def __capture_native_dependencies(
        checkpoint: PluginPackageCheckpoint,
    ) -> None:
        """
        仅在插件依赖即将安装时记录当前进程的原生载荷。
        """
        if checkpoint.native_dependencies is not None:
            return
        try:
            checkpoint.native_dependencies = capture_loaded_native_dependencies()
        # 诊断失败不能阻断插件安装
        except Exception as error:  # noqa: BLE001
            logger.warning("记录插件原生依赖安装前状态失败：%s", error)

    def install_raw(self, pid: str, repo_url: str, package_version: Optional[str] = None,
                    release_version: Optional[str] = None, force_install: bool = False,
                    before_dependency_install: Optional[Callable[[], None]] = None) \
            -> tuple[bool, str]:
        """
        执行已通过来源准入的同步包安装，不负责身份或运行态提交。
        """
        if self.is_local_repo_url(repo_url):
            return self.__install_local_package(
                pid=pid,
                repo_url=repo_url,
                force_install=force_install,
                before_dependency_install=before_dependency_install,
            )

        user_repo, error_result = self._resolve_remote_plugin_repository(
            pid, repo_url
        )
        if error_result is not None:
            return error_result
        assert user_repo is not None

        preferred_version = package_version or get_runtime_setting('VERSION_FLAG')
        selected_version = self.get_plugin_package_version(
            pid, repo_url, preferred_version
        )
        if error_result := self._accept_remote_package_version(
            pid, selected_version
        ):
            return error_result
        assert selected_version is not None

        # 2. 决定安装方式（release 或文件列表）并执行统一安装流程。
        meta = self.__get_plugin_meta(pid, repo_url, selected_version)
        release_items = (
            self.get_plugin_release_versions(pid, repo_url)
            if release_version
            else []
        )
        selection, message = self._select_remote_install(
            pid=pid,
            user_repo=user_repo,
            package_version=selected_version,
            meta=meta,
            release_version=release_version,
            release_items=release_items,
        )
        if selection is None:
            return False, message

        release_tag = selection.release_tag
        if release_tag and not selection.fallback_to_filelist:
            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(
                pid,
                force_install,
                prepare_selected_release,
                repo_url,
                before_dependency_install,
            )

        if release_tag:
            # 当前索引 Release 失败时回退文件列表，避免发布产物短暂滞后阻断安装。
            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}")
                # 回退前只清理本次暂存的半份 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(
                pid,
                force_install,
                prepare_release,
                repo_url,
                before_dependency_install,
            )
        # 未声明 release 打包的插件继续使用文件列表方式安装。

        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(
            pid,
            force_install,
            prepare_filelist,
            repo_url,
            before_dependency_install,
        )

    def install_local_raw(
        self,
        plugin_id: str,
        repo_url: str = "",
        force_install: bool = False,
        before_dependency_install: Optional[Callable[[], None]] = None,
    ) -> tuple[bool, str]:
        """
        提供给兼容 Facade 的本地载荷安装入口。
        """
        return self.__install_local_package(
            pid=plugin_id,
            repo_url=repo_url,
            force_install=force_install,
            before_dependency_install=before_dependency_install,
        )

    def backup_plugin(self, plugin_id: str) -> Optional[str]:
        """
        提供给兼容 Facade 的临时备份入口。
        """
        return self.__backup_plugin(plugin_id)

    def restore_plugin(self, plugin_id: str, backup_dir: str) -> None:
        """
        提供给兼容 Facade 的临时备份恢复入口。
        """
        self.__restore_plugin(plugin_id, backup_dir)

    def remove_plugin(self, plugin_id: str) -> bool:
        """
        删除受控插件目录，返回删除前是否存在。
        """
        plugin_dir = self.__plugin_dir(plugin_id)
        if not plugin_dir.exists():
            return False
        shutil.rmtree(plugin_dir, ignore_errors=False)
        return True

    def install_from_release(
        self, plugin_id: str, user_repo: str, release_tag: str
    ) -> tuple[bool, str]:
        """
        提供给兼容 Facade 的 Release 制品安装入口。
        """
        return self.__install_from_release(plugin_id, user_repo, release_tag)

    def __install_local_package(
        self,
        pid: str,
        repo_url: str = "",
        force_install: bool = False,
        before_dependency_install: Optional[Callable[[], None]] = None,
    ) -> tuple[bool, str]:
        """
        执行已通过来源准入的本地插件包安装。
        """
        local_pid = self.parse_local_repo_url(repo_url) if repo_url else pid
        if not local_pid or local_pid.lower() != pid.lower():
            return False, "本地插件来源与插件ID不匹配"

        repo_path = self.parse_local_repo_path(repo_url) if repo_url else None
        package_version = self.parse_local_repo_package_version(repo_url) if repo_url else None
        candidate = self.get_local_plugin_candidate(
            pid,
            package_version=package_version,
            repo_path=repo_path
        )
        if not candidate:
            return False, f"未找到本地插件：{pid}"
        # 本地插件包走同一条准入判据：运行时不兼容要如实报，不能落到版本提示上
        compatible, message = self.check_plugin_install_compatibility(candidate)
        if not compatible:
            logger.debug(f"{pid} 本地插件安装兼容性检查失败：{message}")
            return False, message

        raw_source_path = candidate.get("path")
        if not isinstance(raw_source_path, (str, Path)):
            return False, "本地插件来源路径无效"
        source_dir = Path(raw_source_path)
        dest_dir = self._plugins_root() / pid.lower()
        try:
            if source_dir.resolve() == dest_dir.resolve():
                return False, "本地插件来源不能与运行目录相同"
        except Exception:
            return False, "本地插件来源路径无效"

        def prepare_local(staging_dir: Path) -> tuple[bool, str]:
            try:
                shutil.copytree(
                    source_dir,
                    staging_dir,
                    dirs_exist_ok=True,
                    ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store", "node_modules")
                )
                return True, ""
            except Exception as e:
                logger.error(f"复制本地插件 {pid} 失败：{e}")
                return False, f"复制本地插件失败：{e}"

        return self.__install_flow_sync(
            pid=pid,
            force_install=force_install,
            prepare_content=prepare_local,
            repo_url=repo_url or self.make_local_repo_url(
                pid,
                (
                    Path(repo_path_value)
                    if isinstance(
                        (repo_path_value := candidate.get("repo_path")),
                        (str, Path),
                    )
                    else None
                ),
                (
                    package_generation
                    if isinstance(
                        (package_generation := candidate.get("package_version")),
                        str,
                    )
                    else None
                ),
            ),
            before_dependency_install=before_dependency_install,
        )

    def __get_file_list(self, pid: str, user_repo: str, package_version: Optional[str] = None) -> \
            tuple[Optional[list[dict[str, Any]]], Optional[str]]:
        """
        获取插件的文件列表

        :param pid: 插件 ID
        :param user_repo: GitHub 仓库的 user/repo 路径
        :return: (文件列表, 错误信息)
        """
        file_api = f"https://api.github.com/repos/{user_repo}/contents/plugins"
        # 如果 package_version 存在（如 "v2"），则加上版本号
        if package_version:
            file_api += f".{package_version}"
        file_api += f"/{pid.lower()}"

        res = self.__request_with_fallback(file_api,
                                           headers=get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo),
                                           is_api=True,
                                           timeout=30)
        if res is None:
            return None, "连接仓库失败"
        elif res.status_code == 404:
            return None, "插件源码目录不存在"
        elif res.status_code != 200:
            return None, f"连接仓库失败：{res.status_code} - " \
                f"{'超出速率限制，请设置Github Token或稍后重试' if res.status_code == 403 else res.reason}"

        try:
            ret = res.json()
            if isinstance(ret, list) and len(ret) > 0 and "message" not in ret[0]:
                return ret, ""
            else:
                return None, "插件在仓库中不存在或返回数据格式不正确"
        except Exception as e:
            logger.error(f"插件数据解析失败：{e}")
            return None, "插件数据解析失败"

    def __resolve_download_file(
        self,
        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)
        if pure_path.is_absolute() or PureWindowsPath(remote_path).is_absolute():
            raise ValueError("插件文件路径无效")
        parts = pure_path.parts
        expected_root = f"plugins.{package_version}" if package_version else "plugins"
        if (
            len(parts) < 3
            or parts[0] != expected_root
            or parts[1].casefold() != pid.casefold()
            or any(part in {"", ".", ".."} for part in parts[2:])
        ):
            raise ValueError("插件文件路径无效")
        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,
                         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)]

        while stack:
            current_pid, current_file_list = stack.pop()

            for item in current_file_list:
                download_url = item.get("download_url")
                if isinstance(download_url, str) and download_url:
                    try:
                        file_path = self.__resolve_download_file(
                            pid,
                            item.get("path"),
                            package_version,
                            download_root,
                        )
                    except ValueError as error:
                        return False, str(error)
                    logger.debug(f"正在下载文件：{item.get('path')}")
                    res = self.__request_with_fallback(download_url,
                                                       headers=get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo))
                    if not res:
                        return False, f"文件 {item.get('path')} 下载失败！"
                    elif res.status_code != 200:
                        return False, f"下载文件 {item.get('path')} 失败：{res.status_code}"

                    # GitHub 文件列表同时承载 Python 源码和 wheel 等二进制制品，统一按原始字节写入
                    file_path.parent.mkdir(parents=True, exist_ok=True)
                    with open(file_path, "wb") as f:
                        f.write(res.content)
                    logger.debug(f"文件 {item.get('path')} 下载成功，保存路径：{file_path}")
                else:
                    # 如果是子目录，则将子目录内容加入栈中继续处理
                    directory_name = item.get("name")
                    if (
                        not isinstance(directory_name, str)
                        or directory_name in {"", ".", ".."}
                        or PurePosixPath(directory_name).name != directory_name
                        or PureWindowsPath(directory_name).name != directory_name
                    ):
                        return False, "插件目录路径无效"
                    sub_list, msg = self.__get_file_list(f"{current_pid}/{directory_name}", user_repo,
                                                         package_version)
                    if not sub_list:
                        return False, msg or "插件子目录读取失败"
                    stack.append((f"{current_pid}/{directory_name}", sub_list))

        return True, ""

    def __install_dependencies_if_required(
        self,
        pid: str,
        before_dependency_install: Optional[Callable[[], None]] = None,
    ) -> tuple[bool, bool, str]:
        """
        安装插件依赖。

        :param pid: 插件 ID
        :return: (是否存在依赖，安装是否成功, 错误信息)
        """
        plugin_dir = self._plugins_root() / pid.lower()
        try:
            manifest = load_dependency_manifest(plugin_dir)
        except PluginDependencyManifestError as error:
            logger.error(f"{pid} 依赖清单无效：{error}")
            return True, False, str(error)
        if manifest is not None:
            if (
                before_dependency_install is not None
                and dependency_manifest_declares_installation(manifest)
            ):
                try:
                    before_dependency_install()
                # 观察失败不能阻断安装
                except Exception as error:  # noqa: BLE001
                    logger.warning(f"{pid} 依赖安装前状态记录失败：{error}")
            logger.info(f"{pid} 存在依赖，开始尝试安装依赖")
            success, error_message = self.install_packages_with_fallback(manifest.path)
            return True, success, "" if success else error_message

        return False, False, "不存在依赖"

    def __backup_plugin(self, pid: str) -> Optional[str]:
        """
        备份旧插件目录

        :param pid: 插件 ID
        :return: 备份目录路径
        """
        plugin_dir = self._plugins_root() / pid.lower()
        backup_dir = Path(get_runtime_setting('TEMP_PATH')) / "plugin_backup" / pid.lower()

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

            shutil.copytree(plugin_dir, backup_dir, dirs_exist_ok=True)
            logger.debug(f"{pid} 插件已备份到 {backup_dir}")

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

    def __restore_plugin(self, pid: str, backup_dir: str) -> None:
        """
        还原旧插件目录

        :param pid: 插件 ID
        :param backup_dir: 备份目录路径
        """
        plugin_dir = self._plugins_root() / pid.lower()
        if plugin_dir.exists():
            shutil.rmtree(plugin_dir, ignore_errors=True)
            logger.debug(f"{pid} 已清理插件目录 {plugin_dir}")

        if Path(backup_dir).exists():
            shutil.copytree(backup_dir, plugin_dir, dirs_exist_ok=True)
            logger.debug(f"{pid} 已还原插件目录 {plugin_dir}")
            shutil.rmtree(backup_dir, ignore_errors=True)
            logger.debug(f"{pid} 已删除备份目录 {backup_dir}")

    def __remove_old_plugin(self, pid: str) -> None:
        """
        删除旧插件

        :param pid: 插件 ID
        """
        plugin_dir = self._plugins_root() / pid.lower()
        if plugin_dir.exists():
            shutil.rmtree(plugin_dir, ignore_errors=True)

    def refresh_persistent_backup(self, pid: str) -> bool:
        """
        刷新插件持久化备份目录，供 docker 重置后恢复使用
        """
        if not SystemUtils.is_docker():
            return True

        plugin_dir = self._plugins_root() / pid.lower()
        if not plugin_dir.exists():
            logger.warn(f"{pid} 插件目录不存在，跳过刷新插件备份")
            return False

        backup_root = get_runtime_setting('CONFIG_PATH') / "plugins_backup"
        backup_dir = backup_root / pid.lower()
        staging_dir = backup_root / f".{pid.lower()}.tmp-{uuid.uuid4().hex}"
        previous_dir = backup_root / f".{pid.lower()}.old-{uuid.uuid4().hex}"
        try:
            backup_root.mkdir(parents=True, exist_ok=True)
            shutil.copytree(
                plugin_dir,
                staging_dir,
                ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store")
            )
            if backup_dir.exists():
                backup_dir.replace(previous_dir)
            staging_dir.replace(backup_dir)
            if previous_dir.exists():
                shutil.rmtree(previous_dir, ignore_errors=True)
            logger.info(f"已刷新插件备份: {pid}")
            return True
        except Exception as e:
            if not backup_dir.exists() and previous_dir.exists():
                try:
                    previous_dir.replace(backup_dir)
                except Exception as rollback_error:
                    logger.error(
                        f"恢复插件旧备份失败，已保留恢复材料 {previous_dir}: "
                        f"{rollback_error}"
                    )
            logger.error(f"刷新插件备份失败: {pid} - {e}")
            return False
        finally:
            if staging_dir.exists():
                shutil.rmtree(staging_dir, ignore_errors=True)
            if backup_dir.exists() and previous_dir.exists():
                shutil.rmtree(previous_dir, ignore_errors=True)

    def __get_plugin_meta(self, pid: str, repo_url: str,
                          package_version: Optional[str]) -> dict[str, Any]:
        """
        读取远端插件元数据，并把异常收敛为空映射。
        """
        try:
            plugins = (
                self.get_plugins(repo_url) if not package_version
                else self.get_plugins(repo_url, package_version)
            ) or {}
            meta = plugins.get(pid)
            return meta if isinstance(meta, dict) else {}
        except Exception as e:
            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 PluginContentSwapError: 换入失败，异常携带运行目录的实际恢复结论
        """
        with plugin_package_lock(final_dir.name):
            PluginPackageManager.__swap_staged_plugin_content_locked(
                staging_dir=staging_dir,
                final_dir=final_dir,
            )

    @staticmethod
    def __swap_staged_plugin_content_locked(
        *,
        staging_dir: Path,
        final_dir: Path,
    ) -> None:
        """
        在持有插件包锁时发布暂存内容并保留失败回滚能力。
        """
        final_dir.parent.mkdir(parents=True, exist_ok=True)
        previous = final_dir.parent / f".{final_dir.name}.previous-{uuid.uuid4().hex}"
        publish_staging: Optional[Path] = None
        previous_available = False
        runtime_dirty = False
        published = False
        try:
            publish_source, publish_staging = (
                PluginPackageManager.__prepare_publish_source(
                    staging_dir=staging_dir,
                    final_dir=final_dir,
                )
            )
            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
                    # 删除从这一刻起就可能只删一半，先认账再动手
                    runtime_dirty = True
                    PluginPackageManager.__remove_snapshot_path(final_dir)
                else:
                    previous_available = True
            # 旧内容已挪开或本来就不存在，此后才允许重建运行目录
            runtime_dirty = True
            try:
                publish_source.replace(final_dir)
            except OSError as error:
                if error.errno != errno.EXDEV:
                    raise
                if publish_source != staging_dir:
                    raise
                logger.debug(
                    f"插件安装内容跨文件系统无法改名换入，先复制到运行目录同级暂存："
                    f"{staging_dir}"
                )
                publish_staging = (
                    final_dir.parent
                    / f".{final_dir.name}.staging-{uuid.uuid4().hex}"
                )
                shutil.copytree(staging_dir, publish_staging, symlinks=True)
                publish_staging.replace(final_dir)
            published = True
        except Exception as error:
            # 运行目录从未被触碰时无需回滚，它本身就还是换入前那一份
            runtime_intact = not runtime_dirty
            if runtime_dirty and not published:
                try:
                    PluginPackageManager.__remove_snapshot_path(final_dir)
                    if previous_available:
                        previous.replace(final_dir)
                        previous_available = False
                    runtime_intact = True
                except Exception as rollback_error:
                    logger.error(
                        f"插件安装换入失败后恢复旧目录失败，已保留恢复材料 {previous}: "
                        f"{rollback_error}"
                    )
            raise PluginContentSwapError(
                error, runtime_intact=runtime_intact
            ) from error
        finally:
            if publish_staging and publish_staging.exists():
                shutil.rmtree(publish_staging, ignore_errors=True)
            if previous.exists() and (published or not previous_available):
                shutil.rmtree(previous, ignore_errors=True)

    @staticmethod
    def __prepare_publish_source(
        *,
        staging_dir: Path,
        final_dir: Path,
    ) -> tuple[Path, Optional[Path]]:
        """
        将跨文件系统的安装载荷复制到运行目录同级，供后续原子改名发布。
        """
        if (
            not staging_dir.exists()
            or staging_dir.stat().st_dev == final_dir.parent.stat().st_dev
        ):
            return staging_dir, None

        publish_staging = (
            final_dir.parent / f".{final_dir.name}.staging-{uuid.uuid4().hex}"
        )
        shutil.copytree(staging_dir, publish_staging, symlinks=True)
        return publish_staging, publish_staging

    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
        )

    @staticmethod
    def __swap_left_runtime_intact(error: OSError) -> bool:
        """
        判断换入失败后运行目录是否确定仍是换入前那一份。

        只认换入方随异常带出的回滚结论：运行目录存在与否区分不了"完整旧内容"和
        "删到一半的残骸"。拿不到结论的异常按最坏情况处理，宁可多走一次备份兜底。

        :param error: 换入阶段抛出的异常
        :return: 运行目录是否确定保持换入前内容
        """
        return isinstance(error, PluginContentSwapError) and error.runtime_intact

    def __recover_after_swap_failure(
        self, pid: str, error: OSError, backup_dir: Optional[str]
    ) -> None:
        """
        换入失败后按实际回滚结论处置本次临时备份。

        :param pid: 插件 ID
        :param error: 换入阶段抛出的异常
        :param backup_dir: 本次安装前留下的临时备份目录，强制安装时为空
        """
        if self.__swap_left_runtime_intact(error):
            if backup_dir:
                shutil.rmtree(backup_dir, ignore_errors=True)
            return
        if backup_dir:
            self.__restore_plugin(pid, backup_dir)
            logger.warn(f"{pid} 插件安装失败，已还原备份插件")
            return
        logger.error(
            f"{pid} 换入失败且运行目录未能恢复，本次未留临时备份，"
            f"请按上一条日志保留的恢复材料人工处理"
        )

    async def __async_recover_after_swap_failure(
        self, pid: str, error: OSError, backup_dir: Optional[str]
    ) -> None:
        """
        异步流程的备份处置，判据与同步一致；备份清理由统一 finally 负责。

        :param pid: 插件 ID
        :param error: 换入阶段抛出的异常
        :param backup_dir: 本次安装前留下的临时备份目录，强制安装时为空
        """
        if self.__swap_left_runtime_intact(error):
            return
        if backup_dir:
            await self.__async_restore_plugin(pid, backup_dir)
            logger.warning(f"{pid} 插件安装失败，已还原备份插件")
            return
        logger.error(
            f"{pid} 换入失败且运行目录未能恢复，本次未留临时备份，"
            f"请按上一条日志保留的恢复材料人工处理"
        )

    def __install_flow_sync(
        self,
        pid: str,
        force_install: bool,
        prepare_content: Callable[[Path], 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)
        try:
            success, message = prepare_content(staging_dir)
            if not success:
                # 运行目录此刻尚未被触碰，已装插件天然完整，无需还原也无需清理
                logger.error(f"{pid} 准备插件内容失败：{message}")
                return False, message

            backup_dir = None
            if not force_install:
                backup_dir = self.__backup_plugin(pid)

            try:
                self.__swap_staged_plugin_content(staging_dir, plugin_dir)
            except OSError as error:
                message = f"写入插件内容失败：{error}"
                logger.error(f"{pid} {message}")
                self.__recover_after_swap_failure(pid, error, backup_dir)
                return False, message

            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 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, ""
        finally:
            if staging_dir.exists():
                shutil.rmtree(staging_dir, ignore_errors=True)

    @staticmethod
    def __validate_release_zip_name(name: str) -> None:
        """
        校验 release zip 成员名在 POSIX 与 Windows 语义下都只能表示相对路径。
        """
        if not name:
            raise ValueError("非法 Release 压缩包成员：成员名为空")
        if "\x00" in name:
            raise ValueError(f"非法 Release 压缩包成员：{name}")
        if "\\" in name:
            raise ValueError(f"非法 Release 压缩包成员：{name}")

        posix_path = PurePosixPath(name)
        windows_path = PureWindowsPath(name)
        if (
            name.startswith("//")
            or posix_path.is_absolute()
            or windows_path.is_absolute()
            or windows_path.drive
        ):
            raise ValueError(f"非法 Release 压缩包成员：{name}")

        parts = [part for part in posix_path.parts if part not in ("", ".")]
        if not parts:
            raise ValueError(f"非法 Release 压缩包成员：{name}")
        if ".." in parts:
            raise ValueError(f"非法 Release 压缩包成员：{name}")

    @staticmethod
    def __validate_release_zip_type(info: zipfile.ZipInfo) -> None:
        """
        release zip 只接受普通文件和目录，避免归档内的符号链接或设备文件影响安装边界。
        """
        mode = info.external_attr >> 16
        file_type = stat.S_IFMT(mode)
        if not file_type:
            return
        if stat.S_ISREG(mode) or stat.S_ISDIR(mode):
            return
        raise ValueError(f"非法 Release 压缩包成员：{info.filename}")

    @staticmethod
    def __get_release_zip_base_prefix(infos: list[zipfile.ZipInfo]) -> str:
        """
        识别 release zip 的单一顶层目录，用于保持插件包根目录剥离行为。
        """
        names = [info.filename for info in infos]
        names_with_slash = [name for name in names if "/" in name]
        if names_with_slash and len(names_with_slash) == len(names):
            first_seg = names_with_slash[0].split("/", 1)[0]
            if first_seg and all(name.startswith(first_seg + "/") for name in names):
                return first_seg + "/"
        return ""

    @classmethod
    def __iter_release_zip_targets(
        cls, zf: zipfile.ZipFile, dest_base: Path
    ) -> list[tuple[zipfile.ZipInfo, Path, bool]]:
        """
        将 release zip 成员解析为安装目标路径，并保证目标路径不会逃逸插件目录。
        """
        infos = zf.infolist()
        for info in infos:
            cls.__validate_release_zip_type(info)
            cls.__validate_release_zip_name(info.filename)

        base_prefix = cls.__get_release_zip_base_prefix(infos)
        dest_root = dest_base.resolve()
        targets = []
        for info in infos:
            raw_name = info.filename
            rel_name = raw_name[len(base_prefix):] if base_prefix else raw_name
            if not rel_name:
                if base_prefix and raw_name == base_prefix:
                    continue
                raise ValueError(f"非法 Release 压缩包成员：{raw_name}")

            cls.__validate_release_zip_name(rel_name)
            rel_parts = [part for part in PurePosixPath(rel_name).parts if part not in ("", ".")]
            if not rel_parts:
                raise ValueError(f"非法 Release 压缩包成员：{raw_name}")
            dest_path = (dest_root / Path(*rel_parts)).resolve()
            try:
                dest_path.relative_to(dest_root)
            except ValueError as exc:
                raise ValueError(f"非法 Release 压缩包成员：{raw_name}") from exc
            targets.append((info, dest_path, info.is_dir()))
        return targets

    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 根即插件文件；
        将其全部解压到 dest_root，未指定时直接解压到插件运行目录。

        :param pid: 插件标识
        :param user_repo: GitHub 仓库所有者及仓库名
        :param release_tag: 要安装的发布标签，同时用于确定规范资产名
        :param dest_root: 本次安装的暂存目录；None 使用插件运行目录
        :return: 安装是否成功及失败原因；镜像不可用时保留 API 下载
        """
        # 拼接资产文件名
        asset_name = f"{release_tag.lower()}.zip"
        timeout = get_runtime_setting('PLUGIN_RELEASE_TIMEOUT')

        release_api = f"https://api.github.com/repos/{user_repo}/releases/tags/{release_tag}"
        rel_res = self.__request_with_fallback(
            release_api,
            headers=get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo),
            timeout=timeout,
            is_api=True,
        )
        if rel_res is None:
            return False, "获取 Release 信息失败：连接失败"
        if rel_res.status_code == 404:
            return False, f"{release_tag} 插件发布包不存在"
        if rel_res.status_code != 200:
            return False, f"获取 Release 信息失败：{rel_res.status_code}"

        try:
            rel_json = rel_res.json()
            assets = rel_json.get("assets") or []
            asset = next((a for a in assets if a.get("name") == asset_name), None)
            if not asset:
                return False, f"未找到资产文件：{asset_name}"
            asset_id = asset.get("id")
            if not asset_id:
                return False, "资产缺少ID信息"
            # 构建资产的API下载URL
            download_url = f"https://api.github.com/repos/{user_repo}/releases/assets/{asset_id}"
        except Exception as e:
            logger.error(f"解析 Release 信息失败：{e}")
            return False, f"解析 Release 信息失败：{e}"

        # 使用资产的API端点下载，需要设置Accept头为application/octet-stream
        headers = get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo).copy()
        headers["Accept"] = "application/octet-stream"
        mirror_url = (
            f"https://github.com/{user_repo}/releases/download/"
            f"{quote(release_tag, safe='')}/{quote(asset_name, safe='')}"
        )
        res = self.__request_with_fallback(
            download_url, headers=headers, timeout=timeout, is_api=True, mirror_url=mirror_url,
        )
        if res is None or res.status_code != 200:
            return False, f"下载资产失败：{res.status_code if res else '连接失败'}"

        try:
            with zipfile.ZipFile(io.BytesIO(res.content)) as zf:
                infos = zf.infolist()
                if not infos:
                    return False, "压缩包内容为空"
                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:
                    if is_dir:
                        dest_path.mkdir(parents=True, exist_ok=True)
                        continue
                    dest_path.parent.mkdir(parents=True, exist_ok=True)
                    with zf.open(info, 'r') as src, open(dest_path, 'wb') as dst:
                        dst.write(src.read())
                    wrote_any = True
                if not wrote_any:
                    return False, "压缩包中无可写入文件"
            return True, ""
        except Exception as e:
            logger.error(f"解压 Release 压缩包失败：{e}")
            return False, f"解压 Release 压缩包失败：{e}"

    async def __async_get_file_list(self, pid: str, user_repo: str, package_version: Optional[str] = None) -> \
            tuple[Optional[list[dict[str, Any]]], Optional[str]]:
        """
        异步获取插件的文件列表

        :param pid: 插件 ID
        :param user_repo: GitHub 仓库的 user/repo 路径
        :return: (文件列表, 错误信息)
        """
        file_api = f"https://api.github.com/repos/{user_repo}/contents/plugins"
        # 如果 package_version 存在（如 "v2"），则加上版本号
        if package_version:
            file_api += f".{package_version}"
        file_api += f"/{pid.lower()}"

        res = await self.__async_request_with_fallback(file_api,
                                                       headers=get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo),
                                                       is_api=True,
                                                       timeout=30)
        if res is None:
            return None, "连接仓库失败"
        elif res.status_code == 404:
            return None, "插件源码目录不存在"
        elif res.status_code != 200:
            return None, f"连接仓库失败：{res.status_code} - " \
                f"{'超出速率限制，请设置Github Token或稍后重试' if res.status_code == 403 else res.text}"

        try:
            ret = res.json()
            if isinstance(ret, list) and len(ret) > 0 and "message" not in ret[0]:
                return ret, ""
            else:
                return None, "插件在仓库中不存在或返回数据格式不正确"
        except Exception as e:
            logger.error(f"插件数据解析失败：{e}")
            return None, "插件数据解析失败"

    async def __async_download_files(self, pid: str, file_list: list[dict[str, Any]], user_repo: 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)]

        while stack:
            current_pid, current_file_list = stack.pop()

            for item in current_file_list:
                download_url = item.get("download_url")
                if isinstance(download_url, str) and download_url:
                    try:
                        resolved_path = self.__resolve_download_file(
                            pid,
                            item.get("path"),
                            package_version,
                            download_root,
                        )
                    except ValueError as error:
                        return False, str(error)
                    logger.debug(f"正在下载文件：{item.get('path')}")
                    res = await self.__async_request_with_fallback(download_url,
                                                                   headers=get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo))
                    if not res:
                        return False, f"文件 {item.get('path')} 下载失败！"
                    elif res.status_code != 200:
                        return False, f"下载文件 {item.get('path')} 失败：{res.status_code}"

                    # GitHub 文件列表同时承载 Python 源码和 wheel 等二进制制品，统一按原始字节写入
                    file_path = AsyncPath(resolved_path)
                    await file_path.parent.mkdir(parents=True, exist_ok=True)
                    async with aiofiles.open(file_path, "wb") as f:
                        await f.write(res.content)
                    logger.debug(f"文件 {item.get('path')} 下载成功，保存路径：{file_path}")
                else:
                    # 如果是子目录，则将子目录内容加入栈中继续处理
                    directory_name = item.get("name")
                    if (
                        not isinstance(directory_name, str)
                        or directory_name in {"", ".", ".."}
                        or PurePosixPath(directory_name).name != directory_name
                        or PureWindowsPath(directory_name).name != directory_name
                    ):
                        return False, "插件目录路径无效"
                    sub_list, msg = await self.__async_get_file_list(f"{current_pid}/{directory_name}", user_repo,
                                                                     package_version)
                    if not sub_list:
                        return False, msg or "插件子目录读取失败"
                    stack.append((f"{current_pid}/{directory_name}", sub_list))

        return True, ""

    async def __async_backup_plugin(self, pid: str) -> Optional[str]:
        """
        异步备份旧插件目录

        :param pid: 插件 ID
        :return: 备份目录路径
        """
        plugin_dir = AsyncPath(self._plugins_root()) / pid.lower()
        backup_dir = AsyncPath(get_runtime_setting('TEMP_PATH')) / "plugin_backup" / pid.lower()

        if await plugin_dir.exists():
            try:
                if await backup_dir.exists():
                    await aioshutil.rmtree(backup_dir, ignore_errors=True)
                    logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}")

                await self._async_copytree(plugin_dir, backup_dir)
                logger.debug(f"{pid} 插件已备份到 {backup_dir}")
            except asyncio.CancelledError:
                await aioshutil.rmtree(backup_dir, ignore_errors=True)
                raise

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

    async def __async_restore_plugin(self, pid: str, backup_dir: str) -> None:
        """
        异步还原旧插件目录

        :param pid: 插件 ID
        :param backup_dir: 备份目录路径
        """
        plugin_dir = AsyncPath(self._plugins_root()) / pid.lower()
        if await plugin_dir.exists():
            await aioshutil.rmtree(plugin_dir, ignore_errors=True)
            logger.debug(f"{pid} 已清理插件目录 {plugin_dir}")

        backup_path = AsyncPath(backup_dir)
        if await backup_path.exists():
            await self._async_copytree(src=backup_path, dst=plugin_dir)
            logger.debug(f"{pid} 已还原插件目录 {plugin_dir}")
            await aioshutil.rmtree(backup_path, ignore_errors=True)
            logger.debug(f"{pid} 已删除备份目录 {backup_dir}")

    async def __async_remove_old_plugin(self, pid: str) -> None:
        """
        异步删除旧插件

        :param pid: 插件 ID
        """
        plugin_dir = AsyncPath(self._plugins_root()) / pid.lower()
        if await plugin_dir.exists():
            await aioshutil.rmtree(plugin_dir, ignore_errors=True)

    async def _async_copytree(self, src: AsyncPath, dst: AsyncPath) -> None:
        """
        异步递归复制目录

        :param src: 源目录
        :param dst: 目标目录
        """
        if not await src.exists():
            return

        await dst.mkdir(parents=True, exist_ok=True)

        async for item in src.iterdir():
            dst_item = dst / item.name
            if await item.is_dir():
                await self._async_copytree(item, dst_item)
            else:
                async with aiofiles.open(item, 'rb') as src_file:
                    content = await src_file.read()
                async with aiofiles.open(dst_item, 'wb') as dst_file:
                    await dst_file.write(content)

    async def __async_install_dependencies_if_required(
        self,
        pid: str,
        before_dependency_install: Optional[Callable[[], None]] = None,
    ) -> tuple[bool, bool, str]:
        """
        异步安装插件依赖。

        :param pid: 插件 ID
        :return: (是否存在依赖，安装是否成功, 错误信息)
        """
        plugin_dir = self._plugins_root() / pid.lower()
        try:
            manifest = load_dependency_manifest(plugin_dir)
        except PluginDependencyManifestError as error:
            logger.error(f"{pid} 依赖清单无效：{error}")
            return True, False, str(error)
        if manifest is not None:
            if (
                before_dependency_install is not None
                and dependency_manifest_declares_installation(manifest)
            ):
                try:
                    await _await_thread_operation(before_dependency_install)
                # 观察失败不能阻断安装
                except Exception as error:  # noqa: BLE001
                    logger.warning(f"{pid} 依赖安装前状态记录失败：{error}")
            logger.info(f"{pid} 存在依赖，开始尝试安装依赖")
            success, error_message = await self.async_install_packages_with_fallback(
                manifest.path
            )
            return True, success, "" if success else error_message

        return False, False, "不存在依赖"

    async def async_install_raw(
            self,
            pid: str,
            repo_url: str,
            package_version: Optional[str] = None,
            release_version: Optional[str] = None,
            force_install: bool = False,
            before_dependency_install: Optional[Callable[[], None]] = None,
    ) -> tuple[bool, str]:
        """
        执行已通过来源准入的异步包安装，不负责身份或运行态提交。
        """
        if self.is_local_repo_url(repo_url):
            return cast(
                tuple[bool, str],
                await _await_thread_operation(
                    self.__install_local_package,
                    pid,
                    repo_url,
                    force_install,
                    before_dependency_install,
                ),
            )

        user_repo, error_result = self._resolve_remote_plugin_repository(
            pid, repo_url
        )
        if error_result is not None:
            return error_result
        assert user_repo is not None

        preferred_version = package_version or get_runtime_setting('VERSION_FLAG')
        selected_version = await self.async_get_plugin_package_version(
            pid, repo_url, preferred_version
        )
        if error_result := self._accept_remote_package_version(
            pid, selected_version
        ):
            return error_result
        assert selected_version is not None

        # 2. 统一异步安装流程（release 或文件列表）。
        meta = await self.__async_get_plugin_meta(pid, repo_url, selected_version)
        release_items = (
            await self.async_get_plugin_release_versions(pid, repo_url)
            if release_version
            else []
        )
        selection, message = self._select_remote_install(
            pid=pid,
            user_repo=user_repo,
            package_version=selected_version,
            meta=meta,
            release_version=release_version,
            release_items=release_items,
        )
        if selection is None:
            return False, message

        release_tag = selection.release_tag
        if release_tag and not selection.fallback_to_filelist:
            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(
                pid,
                force_install,
                prepare_selected_release,
                repo_url,
                before_dependency_install,
            )

        if release_tag:
            # 当前索引 Release 失败时回退文件列表，保持同步与异步安装一致。
            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}")
                # 回退前只清理本次暂存的半份 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(
                pid,
                force_install,
                prepare_release,
                repo_url,
                before_dependency_install,
            )
        # 未声明 release 打包的插件继续使用文件列表方式安装。

        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(
            pid,
            force_install,
            prepare_filelist,
            repo_url,
            before_dependency_install,
        )

    async def async_install_local_raw(
        self,
        plugin_id: str,
        repo_url: str = "",
        force_install: bool = False,
        before_dependency_install: Optional[Callable[[], None]] = None,
    ) -> tuple[bool, str]:
        """
        提供给兼容 Facade 的异步本地载荷安装入口。
        """
        return await self.async_install_raw(
            pid=plugin_id,
            repo_url=repo_url or self.make_local_repo_url(plugin_id),
            force_install=force_install,
            before_dependency_install=before_dependency_install,
        )

    async def async_backup_plugin(self, plugin_id: str) -> Optional[str]:
        """
        提供给兼容 Facade 的异步临时备份入口。
        """
        return await self.__async_backup_plugin(plugin_id)

    async def async_restore_plugin(
        self, plugin_id: str, backup_dir: str
    ) -> None:
        """
        提供给兼容 Facade 的异步临时备份恢复入口。
        """
        await self.__async_restore_plugin(plugin_id, backup_dir)

    async def async_remove_plugin(self, plugin_id: str) -> None:
        """
        提供给兼容 Facade 的异步旧载荷清理入口。
        """
        await self.__async_remove_old_plugin(plugin_id)

    async def async_install_from_release(
        self, plugin_id: str, user_repo: str, release_tag: str
    ) -> tuple[bool, str]:
        """
        提供给兼容 Facade 的异步 Release 制品安装入口。
        """
        return await self.__async_install_from_release(
            plugin_id, user_repo, release_tag
        )

    async def __async_get_plugin_meta(self, pid: str, repo_url: str,
                                      package_version: Optional[str]) -> dict[str, Any]:
        """
        异步读取远端插件元数据，并把异常收敛为空映射。
        """
        try:
            plugins = (
                await self.async_get_plugins(repo_url) if not package_version
                else await self.async_get_plugins(repo_url, package_version)
            ) or {}
            meta = plugins.get(pid)
            return meta if isinstance(meta, dict) else {}
        except Exception as e:
            logger.warn(f"获取插件 {pid} 元数据失败：{e}")
            return {}

    async def __install_flow_async(
        self,
        pid: str,
        force_install: bool,
        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)

            try:
                await self.__async_swap_staged_plugin_content(staging_dir, plugin_dir)
            except OSError as error:
                message = f"写入插件内容失败：{error}"
                logger.error(f"{pid} {message}")
                await self.__async_recover_after_swap_failure(pid, error, backup_dir)
                return False, message

            dependencies_exist, dep_ok, dep_msg = (
                await self.__async_install_dependencies_if_required(
                    pid,
                    before_dependency_install,
                )
                if before_dependency_install is not None
                else await self.__async_install_dependencies_if_required(pid)
            )
            if dependencies_exist and not dep_ok:
                logger.error(f"{pid} 依赖安装失败：{dep_msg}")
                if backup_dir:
                    await self.__async_restore_plugin(pid, backup_dir)
                    logger.warning(f"{pid} 插件安装失败，已还原备份插件")
                else:
                    await self.__async_remove_old_plugin(pid)
                    logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装")
                return False, dep_msg

            return True, ""
        except asyncio.CancelledError:
            logger.warning(
                f"{pid} 插件安装被取消，Python 依赖环境可能已经改变"
            )
            raise
        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],
                                            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)
        if not file_list:
            if msg == "插件源码目录不存在":
                return False, f"{pid} {msg}"
            return False, msg or "插件文件列表读取失败"
        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],
                                                   dest_root: Optional[Path] = None) -> tuple[bool, str]:
        """
        异步准备插件内容，通过文件列表获取插件文件和依赖

        :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录
        """
        runtime_pid = pid.lower()
        file_list, msg = await self.__async_get_file_list(
            runtime_pid,
            user_repo,
            package_version,
        )
        if not file_list:
            if msg == "插件源码目录不存在":
                return False, f"{pid} {msg}"
            return False, msg or "插件文件列表读取失败"
        ok, m = await self.__async_download_files(
            runtime_pid,
            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,
                                           dest_root: Optional[Path] = None) -> tuple[bool, str]:
        """
        通过 GitHub Release 资产文件安装插件（异步）。
        规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件；
        将其全部解压到 dest_root，未指定时直接解压到插件运行目录。

        :param pid: 插件标识
        :param user_repo: GitHub 仓库所有者及仓库名
        :param release_tag: 要安装的发布标签，同时用于确定规范资产名
        :param dest_root: 本次安装的暂存目录；None 使用插件运行目录
        :return: 安装是否成功及失败原因；镜像不可用时保留 API 下载
        """
        # 拼接资产文件名
        asset_name = f"{release_tag.lower()}.zip"
        timeout = get_runtime_setting('PLUGIN_RELEASE_TIMEOUT')

        release_api = f"https://api.github.com/repos/{user_repo}/releases/tags/{release_tag}"
        rel_res = await self.__async_request_with_fallback(
            release_api,
            headers=get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo),
            timeout=timeout,
            is_api=True,
        )
        if rel_res is None:
            return False, "获取 Release 信息失败：连接失败"
        if rel_res.status_code == 404:
            return False, f"{release_tag} 插件发布包不存在"
        if rel_res.status_code != 200:
            return False, f"获取 Release 信息失败：{rel_res.status_code}"

        try:
            rel_json = rel_res.json()
            assets = rel_json.get("assets") or []
            asset = next((a for a in assets if a.get("name") == asset_name), None)
            if not asset:
                return False, f"未找到资产文件：{asset_name}"
            asset_id = asset.get("id")
            if not asset_id:
                return False, "资产缺少ID信息"
            # 构建资产的API下载URL
            download_url = f"https://api.github.com/repos/{user_repo}/releases/assets/{asset_id}"
        except Exception as e:
            logger.error(f"解析 Release 信息失败：{e}")
            return False, f"解析 Release 信息失败：{e}"

        # 使用资产的API端点下载，需要设置Accept头为application/octet-stream
        headers = get_runtime_setting('REPO_GITHUB_HEADERS')(repo=user_repo).copy()
        headers["Accept"] = "application/octet-stream"
        mirror_url = (
            f"https://github.com/{user_repo}/releases/download/"
            f"{quote(release_tag, safe='')}/{quote(asset_name, safe='')}"
        )
        res = await self.__async_request_with_fallback(
            download_url, headers=headers, timeout=timeout, is_api=True, mirror_url=mirror_url,
        )
        if res is None or res.status_code != 200:
            return False, f"下载资产失败：{res.status_code if res else '连接失败'}"

        try:
            with zipfile.ZipFile(io.BytesIO(res.content)) as zf:
                infos = zf.infolist()
                if not infos:
                    return False, "压缩包内容为空"
                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:
                    async_dest_path = AsyncPath(dest_path)
                    if is_dir:
                        await async_dest_path.mkdir(parents=True, exist_ok=True)
                        continue
                    await async_dest_path.parent.mkdir(parents=True, exist_ok=True)
                    data = await asyncio.to_thread(
                        self.__read_release_zip_member,
                        zf,
                        info,
                    )
                    async with aiofiles.open(dest_path, 'wb') as dst:
                        await dst.write(data)
                    wrote_any = True
                if not wrote_any:
                    return False, "压缩包中无可写入文件"
            return True, ""
        except Exception as e:
            logger.error(f"解压 Release 压缩包失败：{e}")
            return False, f"解压 Release 压缩包失败：{e}"

    @staticmethod
    def __read_release_zip_member(zf: zipfile.ZipFile, info: zipfile.ZipInfo) -> bytes:
        """
        在线程池读取并解压单个 Release 文件，避免阻塞事件循环。
        """
        with zf.open(info, "r") as source:
            return source.read()

    def sync_local(self, plugin_id: str, source_dir: Path) -> bool:
        """
        用本地仓库内容原子替换运行副本，失败时恢复原目录。
        """
        source_dir = source_dir.resolve()
        plugin_dir = self.__plugin_dir(plugin_id)
        if source_dir == plugin_dir:
            return True
        checkpoint = self.checkpoint(plugin_id)
        try:
            if plugin_dir.exists():
                shutil.rmtree(plugin_dir)
            shutil.copytree(
                source_dir,
                plugin_dir,
                ignore=shutil.ignore_patterns(*self._COPY_IGNORE),
            )
            self.commit(checkpoint)
            return True
        except Exception as err:
            logger.error(f"同步本地插件 {plugin_id} 失败：{err}")
            try:
                self.rollback(checkpoint)
            except Exception as rollback_err:
                logger.error(
                    f"恢复本地插件 {plugin_id} 原目录失败：{rollback_err}",
                    exc_info=True,
                )
            return False