git_service.py

import contextlib
import os
import queue
import re
import shutil
import stat
import sys
import tempfile
import threading
import time
import uuid
from collections.abc import Callable, Iterator
from dataclasses import dataclass
from enum import StrEnum
from pathlib import Path

import psutil
import yaml
from packaging import version
from pygit2 import (
    Blob,
    Commit,
    Oid,
    Remote,
    RemoteCallbacks,
    Repository,
    Walker,
    discover_repository,
    init_repository,
    settings,
)
from pygit2.enums import (
    CheckoutStrategy,
    ConfigLevel,
    ReferenceType,
    ResetMode,
    SortMode,
)

from one_dragon.base.config.config_item import ConfigItem
from one_dragon.envs.env_config import EnvConfig
from one_dragon.envs.repo_config import RepoConfig, RepositoryItem
from one_dragon.utils import os_utils
from one_dragon.utils.i18_utils import gt
from one_dragon.utils.log_utils import log

REMOTE_FETCH_INITIAL_TIMEOUT = 10.0
REMOTE_FETCH_IDLE_TIMEOUT = 30.0
REMOTE_FETCH_TIMEOUT = 120.0


_FETCH_TIMEOUT_SETTING_NAMES = ('server_connect_timeout', 'server_timeout')
_fetch_timeout_settings_lock = threading.Lock()
_fetch_timeout_settings_configured = False
_fetch_temp_cleanup_lock = threading.Lock()
_fetch_temp_cleanup_roots: set[Path] = set()


class GitSyncStatus(StrEnum):
    """Git 代码同步结果。枚举值为面向用户的中性结果文案。"""

    SUCCESS = '更新完成'
    UP_TO_DATE = '当前已是最新版本'
    RUNTIME_INCOMPATIBLE = '新版本需要更新启动器才能使用'
    BUILTIN_TAG_UNAVAILABLE = '暂时无法获取当前版本所需文件'
    REMOTE_UNAVAILABLE = '暂时无法获取更新'
    LOCAL_CHANGES = '检测到程序文件有改动，未自动更新'
    LOCAL_UPDATE_FAILED = '更新没有完成'
    FAILED = '更新失败'


class _LocalGitMetadataError(RuntimeError):
    """本地 Git 元数据损坏，不能通过切换代码源修复。"""


def _get_repository_objects_path(repo: Repository) -> Path:
    """获取仓库实际使用的 Git 对象目录，兼容 linked worktree。"""
    repo_path = Path(repo.path)
    commondir_path = repo_path / 'commondir'
    if commondir_path.is_file():
        common_dir_value = commondir_path.read_text(encoding='utf-8').strip()
        common_dir = Path(common_dir_value)
        if not common_dir.is_absolute():
            common_dir = repo_path / common_dir
        return common_dir.resolve() / 'objects'
    return repo_path / 'objects'


def _get_repository_shallow_path(repo: Repository) -> Path:
    """获取仓库级 shallow 文件路径。"""
    return _get_repository_objects_path(repo).parent / 'shallow'


def _read_shallow_snapshot(shallow_path: Path) -> bytes | None:
    """二进制读取 shallow 快照；文件不存在表示无浅边界。"""
    return shallow_path.read_bytes() if shallow_path.is_file() else None


def _parse_shallow_snapshot(snapshot: bytes | None) -> list[str]:
    """校验 shallow 快照并按原顺序返回 OID。"""
    if not snapshot:
        return []
    if b'\r' in snapshot or not snapshot.endswith(b'\n'):
        raise ValueError('shallow 文件必须使用 LF 行尾')

    oids: list[str] = []
    for raw_line in snapshot.splitlines():
        try:
            oid = Oid(hex=raw_line.decode('ascii'))
        except (UnicodeDecodeError, ValueError) as error:
            raise ValueError('shallow 文件包含无效 OID') from error
        oids.append(str(oid))
    return oids


def _write_shallow_snapshot(shallow_path: Path, snapshot: bytes | None) -> None:
    """原子写入 shallow 快照；None 表示删除文件。"""
    if snapshot is None:
        shallow_path.unlink(missing_ok=True)
        return

    temporary_path = shallow_path.with_name(
        f'{shallow_path.name}.one-dragon-{uuid.uuid4().hex}.tmp'
    )
    try:
        temporary_path.write_bytes(snapshot)
        os.replace(temporary_path, shallow_path)
    finally:
        temporary_path.unlink(missing_ok=True)


def _merge_shallow_snapshots(
    original_snapshot: bytes,
    fetched_snapshot: bytes | None,
) -> bytes:
    """完整继承失败时只追加新边界，不删除正式仓库原边界。"""
    merged_oids = _parse_shallow_snapshot(original_snapshot)
    known_oids = set(merged_oids)
    for oid in _parse_shallow_snapshot(fetched_snapshot):
        if oid not in known_oids:
            merged_oids.append(oid)
            known_oids.add(oid)
    return b''.join(f'{oid}\n'.encode('ascii') for oid in merged_oids)


def _sync_shallow_file(
    repo: Repository,
    temp_repo_dir: str,
    original_snapshot: bytes | None,
    authoritative: bool,
) -> None:
    """把临时仓库的整仓 shallow 状态同步到正式仓库。"""
    fetched_snapshot = _read_shallow_snapshot(Path(temp_repo_dir) / 'shallow')
    if authoritative or not original_snapshot:
        _parse_shallow_snapshot(fetched_snapshot)
        final_snapshot = fetched_snapshot or None
    else:
        final_snapshot = _merge_shallow_snapshots(original_snapshot, fetched_snapshot)

    _write_shallow_snapshot(_get_repository_shallow_path(repo), final_snapshot)


def _configure_alternate_objects(temp_repo: Repository, source_objects_dir: str | None) -> bool:
    """让临时仓库只读复用正式仓库的 Git 对象。"""
    if not source_objects_dir:
        return False

    source_path = Path(source_objects_dir).resolve()
    if not source_path.is_dir():
        return False

    alternates_path = Path(temp_repo.path) / 'objects' / 'info' / 'alternates'
    alternates_path.parent.mkdir(parents=True, exist_ok=True)
    alternates_path.write_text(f'{source_path}\n', encoding='utf-8')
    return True


@contextlib.contextmanager
def _temporary_fetch_timeout_context() -> Iterator[None]:
    """设置 pygit2 的连接和网络读写超时，只设置一次且不恢复。"""
    global _fetch_timeout_settings_configured
    with _fetch_timeout_settings_lock:
        if not _fetch_timeout_settings_configured:
            timeout_ms = int(REMOTE_FETCH_IDLE_TIMEOUT * 1000)
            effective_values: dict[str, int] = {}
            for setting_name in _FETCH_TIMEOUT_SETTING_NAMES:
                if hasattr(settings, setting_name):
                    setattr(settings, setting_name, timeout_ms)
                    effective_values[setting_name] = int(getattr(settings, setting_name))
            log.info(
                'Git fetch 超时设置: server_connect_timeout=%sms, server_timeout=%sms',
                effective_values.get('server_connect_timeout', '不可用'),
                effective_values.get('server_timeout', '不可用'),
            )
            _fetch_timeout_settings_configured = True
    yield


def _remove_temp_repo(temp_repo_dir: str) -> None:
    """删除 fetch 临时仓库，兼容只读文件和 Windows 文件占用。"""
    def remove_readonly(
        func: Callable[[str], object],
        path: str,
        _exc_info: object,
    ) -> None:
        Path(path).chmod(stat.S_IWRITE)
        func(path)

    try:
        shutil.rmtree(temp_repo_dir, onerror=remove_readonly)
    except FileNotFoundError:
        return
    except PermissionError as error:
        if sys.platform == 'win32' and error.winerror in (5, 32):
            log.info(f'Git fetch 临时仓库仍被占用，将在下次启动时清理: {temp_repo_dir}')
            return
        log.warning(f'清理 Git fetch 临时仓库失败: {temp_repo_dir}', exc_info=True)
    except Exception:
        log.warning(f'清理 Git fetch 临时仓库失败: {temp_repo_dir}', exc_info=True)


def _is_process_running(process_id: int) -> bool:
    """判断临时仓库所属进程是否仍在运行；无效 PID 按仍在运行处理。"""
    return process_id <= 0 or psutil.pid_exists(process_id)


def _cleanup_stale_fetch_repositories(temp_root: Path) -> None:
    """清理已退出进程遗留的新格式 fetch 临时仓库。"""
    if not temp_root.is_dir():
        return
    for temp_repo_dir in temp_root.glob('fetch_*'):
        if not temp_repo_dir.is_dir():
            continue
        name_parts = temp_repo_dir.name.split('_', 2)
        if len(name_parts) == 3 and name_parts[1].isdigit():
            if _is_process_running(int(name_parts[1])):
                continue
        _remove_temp_repo(str(temp_repo_dir))


def _cleanup_stale_fetch_repositories_once(temp_root: Path) -> None:
    """同一进程对同一临时根目录只执行一次遗留目录清理。"""
    resolved_root = temp_root.resolve()
    with _fetch_temp_cleanup_lock:
        if resolved_root in _fetch_temp_cleanup_roots:
            return
        _cleanup_stale_fetch_repositories(resolved_root)
        _fetch_temp_cleanup_roots.add(resolved_root)


def _send_fetch_worker_message(
    message_callback: Callable[[dict[str, object]], None],
    message: dict[str, object],
) -> None:
    """向 fetch worker 的线程消息队列发送消息。"""
    message_callback(message)


def _restore_temp_fetch_state(
    temp_repo: Repository,
    temp_repo_dir: str,
    shallow_snapshot: bytes | None,
    refs_to_delete: set[str],
) -> Repository:
    """恢复 fallback 前的临时 shallow 状态，并移除协商引用。"""
    with contextlib.suppress(Exception):
        temp_repo.free()
    _write_shallow_snapshot(Path(temp_repo_dir) / 'shallow', shallow_snapshot)
    restored_repo = Repository(temp_repo_dir)
    for ref_name in refs_to_delete:
        if ref_name in restored_repo.references:
            restored_repo.references.delete(ref_name)
    return restored_repo


def _fetch_remote_worker(
    temp_repo_dir: str,
    source_objects_dir: str | None,
    remote_url: str,
    branch_name: str,
    depth: int,
    proxy: str | None,
    message_callback: Callable[[dict[str, object]], None],
    abandoned: threading.Event,
    fetch_ref: str | None = None,
    base_ref: str | None = None,
    base_oid: str | None = None,
    fetch_primary_branch: bool = False,
    shallow_snapshot: bytes | None = None,
) -> None:
    """在线程中执行网络 fetch，作废后只清理自己的临时仓库。"""
    temp_repo: Repository | None = None
    try:
        GitService._ensure_config_search_path()
        temp_repo = init_repository(temp_repo_dir, bare=True)
        target_ref = fetch_ref or f'refs/heads/{branch_name}'
        target_refspec = f'+{target_ref}:{target_ref}'
        actual_depth = depth
        primary_branch_incremental = False

        alternate_ready = _configure_alternate_objects(temp_repo, source_objects_dir)
        inherited_shallow_snapshot: bytes | None = None
        shallow_authoritative = not shallow_snapshot
        if shallow_snapshot:
            _parse_shallow_snapshot(shallow_snapshot)
            if alternate_ready:
                inherited_shallow_snapshot = shallow_snapshot
                _write_shallow_snapshot(
                    Path(temp_repo_dir) / 'shallow',
                    inherited_shallow_snapshot,
                )
                shallow_authoritative = True

        if actual_depth == 0 and not alternate_ready:
            actual_depth = 1
            log.warning('正式仓库对象目录不可用，降级为 shallow fetch')
        elif actual_depth == 0 and base_ref is not None and base_oid is not None:
            try:
                temp_repo.references.create(base_ref, Oid(hex=base_oid), force=True)
                primary_branch_incremental = fetch_primary_branch
            except Exception:
                actual_depth = 1
                with contextlib.suppress(Exception):
                    temp_repo.references.delete(base_ref)
                log.warning('增量基线不可用，降级为 shallow fetch', exc_info=True)

        if alternate_ready or inherited_shallow_snapshot is not None or base_ref is not None:
            temp_repo.free()
            temp_repo = Repository(temp_repo_dir)

        refspecs = [target_refspec]
        if primary_branch_incremental:
            assert base_ref is not None
            refspecs.insert(0, f'+{base_ref}:{base_ref}')
        temp_repo.remotes.create('origin', remote_url)
        temp_repo.config['remote.origin.tagopt'] = '--no-tags'
        remote = temp_repo.remotes['origin']

        def report_progress(progress: float, message: str) -> None:
            _send_fetch_worker_message(
                message_callback,
                {'type': 'progress', 'progress': progress, 'message': message},
            )

        callbacks = _FetchProgressRemoteCallbacks(report_progress)

        try:
            log.info(
                f'worker 开始 Git fetch: branch={branch_name}, '
                f'depth={actual_depth}, proxy={bool(proxy)}'
            )
            with _temporary_fetch_timeout_context():
                remote.fetch(
                    refspecs=refspecs,
                    proxy=proxy,
                    depth=actual_depth,
                    callbacks=callbacks,
                )
            callbacks.flush_sideband_progress()
            log.info(f'worker Git fetch 已返回: branch={branch_name}, depth={actual_depth}')
        except Exception as error:
            can_fallback = primary_branch_incremental or (
                isinstance(error, KeyError)
                and 'object not found' in str(error)
                and actual_depth == 0
            )
            if not can_fallback:
                raise

            if primary_branch_incremental:
                log.warning('基于项目主分支的增量 fetch 失败，降级为 shallow fetch', exc_info=True)
            refs_to_delete = {target_ref}
            if base_ref is not None:
                refs_to_delete.add(base_ref)
            temp_repo = _restore_temp_fetch_state(
                temp_repo,
                temp_repo_dir,
                inherited_shallow_snapshot,
                refs_to_delete,
            )
            remote = temp_repo.remotes['origin']
            primary_branch_incremental = False
            callbacks = _FetchProgressRemoteCallbacks(report_progress)
            with _temporary_fetch_timeout_context():
                remote.fetch(
                    refspecs=[target_refspec],
                    proxy=proxy,
                    depth=1,
                    callbacks=callbacks,
                )
            callbacks.flush_sideband_progress()
            actual_depth = 1

        result: dict[str, object] = {
            'type': 'result',
            'success': True,
            'depth': actual_depth,
            'shallow_authoritative': shallow_authoritative,
        }
        if primary_branch_incremental:
            result['primary_branch_incremental'] = True
        _send_fetch_worker_message(message_callback, result)
    except Exception as error:
        _send_fetch_worker_message(
            message_callback,
            {'type': 'result', 'success': False, 'error': repr(error)},
        )
    finally:
        if temp_repo is not None:
            with contextlib.suppress(Exception):
                temp_repo.free()
        if abandoned.is_set():
            _remove_temp_repo(temp_repo_dir)


@dataclass
class GitLog:
    """Git 提交日志"""
    commit_id: str
    author: str
    commit_time: str
    commit_message: str


class _FetchProgressRemoteCallbacks(RemoteCallbacks):
    """转发 Git 传输进度、服务端消息和引用更新信息。"""

    def __init__(
        self,
        progress_callback: Callable[[float, str], None] | None,
        timeout: float | None = REMOTE_FETCH_TIMEOUT,
    ) -> None:
        super().__init__()
        self._progress_callback: Callable[[float, str], None] | None = progress_callback
        self._timeout: float | None = timeout
        self._started_at: float = time.monotonic()
        self._progress: float = 0.0
        self._last_transfer_messages: dict[str, str] = {}
        self._last_transfer_log_at: dict[str, float] = {}
        self._last_sideband_message: str | None = None
        self._sideband_buffer: str = ''

    def _check_timeout(self) -> None:
        if self._timeout is not None and time.monotonic() - self._started_at >= self._timeout:
            raise TimeoutError(f'Git 远程拉取超过 {self._timeout:g} 秒')

    def _emit(self, message: str, progress: float | None = None) -> None:
        if self._progress_callback is not None:
            self._progress_callback(self._progress if progress is None else progress, message)
        else:
            log.info(message)

    def _report_transfer_progress(
        self,
        stage: str,
        label: str,
        current: int,
        total: int,
    ) -> None:
        progress = min(max(current / total, 0.0), 1.0)
        is_final = current >= total
        message = f'{label} {current}/{total} ({round(progress * 100)}%)'
        if is_final:
            message = f'{message}, done.'

        self._progress = progress
        if message == self._last_transfer_messages.get(stage):
            return

        now = time.monotonic()
        last_log_at = self._last_transfer_log_at.get(stage)
        if not is_final and last_log_at is not None and now - last_log_at < 0.2:
            return

        self._last_transfer_log_at[stage] = now
        self._last_transfer_messages[stage] = message
        self._emit(message, progress)

    def _report_received_bytes(self, received_bytes: int) -> None:
        progress = 0.0
        received_mb = received_bytes / 1024 / 1024
        message = f'{gt("拉取对象")} {received_mb:.2f} MB'
        self._progress = progress
        if message == self._last_transfer_messages.get('objects'):
            return

        now = time.monotonic()
        last_log_at = self._last_transfer_log_at.get('objects')
        if last_log_at is not None and now - last_log_at < 0.2:
            return

        self._last_transfer_log_at['objects'] = now
        self._last_transfer_messages['objects'] = message
        self._emit(message, progress)

    def transfer_progress(self, stats: object) -> None:
        self._check_timeout()
        total_objects = int(getattr(stats, 'total_objects', 0) or 0)
        received_objects = int(getattr(stats, 'received_objects', 0) or 0)
        total_deltas = int(getattr(stats, 'total_deltas', 0) or 0)
        indexed_deltas = int(getattr(stats, 'indexed_deltas', 0) or 0)
        received_bytes = int(getattr(stats, 'received_bytes', 0) or 0)

        if total_objects > 0:
            self._report_transfer_progress(
                'objects',
                gt('拉取对象'),
                received_objects,
                total_objects,
            )
        else:
            self._report_received_bytes(received_bytes)

        if total_objects > 0 and received_objects >= total_objects and total_deltas > 0:
            self._report_transfer_progress(
                'deltas',
                gt('处理增量'),
                indexed_deltas,
                total_deltas,
            )

    def _emit_sideband_message(self, message: str) -> None:
        if not message.strip():
            return

        progress_prefixes = (
            ('Enumerating objects:', gt('枚举对象:')),
            ('Counting objects:', gt('统计对象:')),
            ('Compressing objects:', gt('压缩对象:')),
        )
        for original_prefix, translated_prefix in progress_prefixes:
            if message.startswith(original_prefix):
                message = f'{translated_prefix}{message[len(original_prefix):]}'
                break

        if message == self._last_sideband_message:
            return
        self._last_sideband_message = message
        self._emit(f'远程消息: {message}')

    def sideband_progress(self, string: str) -> None:
        self._check_timeout()
        buffer = f'{self._sideband_buffer}{string}'
        message_start = 0
        for index, character in enumerate(buffer):
            if character not in ('\r', '\n'):
                continue
            self._emit_sideband_message(buffer[message_start:index])
            message_start = index + 1
        self._sideband_buffer = buffer[message_start:]

    def flush_sideband_progress(self) -> None:
        """输出 fetch 退出时仍未带行尾的远端文本。"""
        message = self._sideband_buffer
        self._sideband_buffer = ''
        self._emit_sideband_message(message)

    def update_tips(self, refname: str, old: Oid, new: Oid) -> None:
        self._check_timeout()
        self.flush_sideband_progress()
        self._emit(f'更新引用: {refname}')


class GitService:

    def __init__(
        self,
        env_config: EnvConfig,
        repo_config: RepoConfig,
        repo_dir: str | None = None,
    ) -> None:
        self.env_config: EnvConfig = env_config
        self.repo_config: RepoConfig = repo_config

        if repo_dir:
            if not Path(repo_dir).is_absolute():
                repo_dir = str(Path(os_utils.get_work_dir()) / repo_dir)
        else:
            repo_dir = os_utils.get_work_dir()
        self.repo_dir: str = repo_dir

        self._repo: Repository | None = None
        self._rebuilding_repository: bool = False
        self._ensure_config_search_path()

    # ================== 私有辅助方法 ==================

    @staticmethod
    def _ensure_config_search_path() -> None:
        """
        通过设置配置搜索路径为空字符串，忽略用户的系统级和全局级 git 配置。
        这可以避免用户的全局配置（如 http.proxy、user.name、SSL 证书路径等）影响程序的 git 操作。
        同时忽略用户可能残留的无效 SSL 证书配置，让 libgit2 使用系统默认的证书验证机制，避免 SSL 证书问题。
        """
        settings.search_path[ConfigLevel.PROGRAMDATA] = ''  # 机器范围 (C:\ProgramData\Git\config)
        settings.search_path[ConfigLevel.SYSTEM] = ''       # 系统级 (如 C:\Program Files\Git\mingw64\etc\gitconfig)
        settings.search_path[ConfigLevel.GLOBAL] = ''       # 用户全局 (%USERPROFILE%\.gitconfig)
        settings.search_path[ConfigLevel.XDG] = ''          # XDG 配置 (%USERPROFILE%\.config\git\config)
        settings.owner_validation = False                   # 禁用仓库所有权验证

    def _open_repo(self, refresh: bool = False) -> Repository:
        """打开仓库（带缓存）"""
        if refresh:
            self._repo = None

        if self._repo is None:
            # 检查是否是有效的 git 仓库
            git_dir = discover_repository(self.repo_dir)
            if not git_dir:
                raise ValueError(f'目录 {self.repo_dir} 不是有效的 Git 仓库')
            try:
                self._repo = Repository(git_dir)
            except Exception:
                try:
                    _parse_shallow_snapshot(
                        _read_shallow_snapshot(Path(git_dir) / 'shallow')
                    )
                except ValueError as shallow_error:
                    raise _LocalGitMetadataError('shallow metadata corrupted') from shallow_error
                raise

        return self._repo

    def _ensure_remote(self, remote_url: str | None = None) -> Remote:
        """确保指定远程仓库地址配置到当前本地 remote。"""
        if remote_url is None:
            remote_url = self._get_git_repository()
        if not remote_url:
            raise ValueError('未能获取有效的远程仓库地址')

        repo = self._open_repo()
        remote_name = self.env_config.git_remote

        if remote_name in repo.remotes.names():
            remote = repo.remotes[remote_name]
            if remote.url == remote_url:
                return remote

            log.info(f'更新远程仓库地址: {remote.url} -> {remote_url}')
            repo.remotes.set_url(remote_name, remote_url)
            return repo.remotes[remote_name]

        log.info(f'创建远程仓库: {remote_name} -> {remote_url}')
        repo.remotes.create(remote_name, remote_url)
        return repo.remotes[remote_name]

    def _get_repository_item(self, repository: RepositoryItem) -> ConfigItem:
        """获取代码源配置项。"""
        return repository.config_item

    def _find_repository(self, value: str) -> RepositoryItem | None:
        """按仓库 ID、显示标题或 URL 查找代码源。"""
        return self.repo_config.find_repository(value)

    def _get_repository_url(self, repository: RepositoryItem, use_gh_proxy: bool = True) -> str:
        """获取指定代码源的 HTTPS 地址。"""
        repository_url = repository.url
        if use_gh_proxy and repository.use_proxy and self.env_config.is_gh_proxy:
            return f'{self.env_config.gh_proxy_url.rstrip("/")}/{repository_url}'
        return repository_url

    def _get_repository_candidates(self) -> list[tuple[RepositoryItem, str]]:
        """按用户选择、上次成功源和 YAML 声明顺序生成候选列表。"""
        repository_url = self.env_config.repository_url
        preferred_repository = self._find_repository(repository_url)
        if preferred_repository is None and repository_url != RepoConfig.AUTO_REPOSITORY_VALUE:
            self.env_config.repository_url = RepoConfig.AUTO_REPOSITORY_VALUE
        if preferred_repository is None:
            preferred_repository = self._find_repository(self.env_config.last_repository_url)

        candidates: list[tuple[RepositoryItem, str]] = []
        for repository in [preferred_repository, *self.repo_config.repositories]:
            if repository is None or any(candidate[0] is repository for candidate in candidates):
                continue
            repository_url = self._get_repository_url(repository)
            if repository_url:
                candidates.append((repository, repository_url))
        return candidates

    def _get_git_repository(self) -> str:
        """获取当前选择模式下首个候选代码源地址。"""
        candidates = self._get_repository_candidates()
        if not candidates:
            raise ValueError('未能获取有效的远程仓库地址')
        return candidates[0][1]

    def _restore_origin(self) -> bool:
        """将当前本地 remote 恢复为项目主仓库 HTTPS 地址。"""
        primary_url = self._get_repository_url(self.repo_config.primary_repository, use_gh_proxy=False)
        if not primary_url:
            return False
        try:
            self._ensure_remote(primary_url)
            return True
        except Exception:
            log.error('恢复主仓库远程地址失败', exc_info=True)
            return False

    @contextlib.contextmanager
    def _temporary_fetch_timeout(self) -> Iterator[None]:
        """设置 pygit2 的连接和网络读写超时，只设置一次且不恢复。"""
        with _temporary_fetch_timeout_context():
            yield

    def _get_proxy_address(self) -> str | None:
        """获取代理地址"""
        if not self.env_config.is_personal_proxy:
            return None

        proxy = self.env_config.personal_proxy.strip()
        if not proxy:
            return None

        if proxy.startswith(('http://', 'https://', 'socks5://')):
            return proxy

        return f'http://{proxy}'

    def _create_fetch_callbacks(
        self,
        progress_callback: Callable[[float, str], None] | None,
        stage_start: float,
        stage_end: float,
    ) -> RemoteCallbacks:
        """创建远程拉取回调。"""
        def stage_progress_callback(progress: float, message: str) -> None:
            mapped_progress = stage_start + (stage_end - stage_start) * progress
            if progress_callback is not None:
                progress_callback(mapped_progress, message)

        return _FetchProgressRemoteCallbacks(stage_progress_callback, REMOTE_FETCH_TIMEOUT)

    def _validate_history(self, repo: Repository, start_oid: Oid) -> None:
        """确认提交历史可遍历到完整根或合法 shallow 边界。"""
        try:
            for _ in repo.walk(start_oid, SortMode.TOPOLOGICAL):
                pass
        except KeyError as error:
            if self._is_missing_object_error(error):
                raise _LocalGitMetadataError('commit history has undeclared gap') from error
            raise

    def _derive_single_shallow_boundary(self, repo: Repository, start_oid: Oid) -> Oid:
        """沿项目主分支第一父链推导唯一 shallow 边界。"""
        current_oid = start_oid
        child_oid: Oid | None = None
        latest_merge_oid: Oid | None = None
        latest_merge_child_oid: Oid | None = None
        while True:
            try:
                commit = repo[current_oid]
            except KeyError as error:
                if self._is_missing_object_error(error):
                    raise _LocalGitMetadataError('commit history has undeclared gap') from error
                raise
            if len(commit.parent_ids) > 1:
                latest_merge_oid = current_oid
                latest_merge_child_oid = child_oid
            if not commit.parent_ids:
                break
            try:
                parent = commit.parents[0]
            except KeyError as error:
                if self._is_missing_object_error(error):
                    break
                raise
            child_oid = current_oid
            current_oid = parent.id
        if latest_merge_oid is not None:
            if latest_merge_child_oid is not None:
                return latest_merge_child_oid
            return latest_merge_oid
        return current_oid

    def _repair_primary_branch_shallow(
        self,
        repo: Repository,
        start_oid: Oid,
        shallow_snapshot: bytes | None,
        primary_branch: str,
    ) -> tuple[Repository, bytes | None]:
        """把主分支缺口前的单个可达提交追加为 shallow 边界。"""
        boundary_oid = self._derive_single_shallow_boundary(repo, start_oid)
        shallow_path = _get_repository_shallow_path(repo)
        shallow_oids = _parse_shallow_snapshot(shallow_snapshot)
        if str(boundary_oid) not in shallow_oids:
            shallow_oids.append(str(boundary_oid))
        repaired_snapshot = b''.join(
            f'{oid}\n'.encode('ascii')
            for oid in shallow_oids
        )
        _write_shallow_snapshot(shallow_path, repaired_snapshot)

        self._repo = None
        with contextlib.suppress(Exception):
            repo.free()
        repaired_repo = self._open_repo()
        try:
            self._validate_history(repaired_repo, start_oid)
        except BaseException as error:
            with contextlib.suppress(Exception):
                _write_shallow_snapshot(shallow_path, shallow_snapshot)
            self._repo = None
            with contextlib.suppress(Exception):
                repaired_repo.free()
            with contextlib.suppress(Exception):
                self._open_repo()
            raise _LocalGitMetadataError('commit history has undeclared gap') from error

        log.warning(
            f'检测到项目主分支 {primary_branch} 存在未声明历史缺口，'
            f'已将 {boundary_oid} 记录为 shallow 边界；如需完整历史，请执行 '
            f'git fetch --unshallow {self.env_config.git_remote} {primary_branch}'
        )
        return repaired_repo, repaired_snapshot

    def _get_primary_branch_oid(
        self,
        repo: Repository,
        primary_branch: str,
    ) -> Oid | None:
        """获取可用于其他分支增量 fetch 的项目主分支 tip。"""
        remote_ref = f'refs/remotes/{self.env_config.git_remote}/{primary_branch}'
        local_ref = f'refs/heads/{primary_branch}'
        for ref_name in (remote_ref, local_ref):
            if ref_name not in repo.references:
                continue
            primary_branch_oid = repo.references[ref_name].target
            if primary_branch_oid is not None:
                return primary_branch_oid
        return None

    def _import_fetch_result(
        self,
        temp_repo_dir: str,
        progress_callback: Callable[[float, str], None] | None,
        stage_start: float,
        stage_end: float,
        tag_name: str | None = None,
        import_primary_branch: bool = False,
        original_shallow_snapshot: bytes | None = None,
        shallow_authoritative: bool = True,
    ) -> None:
        """将临时仓库导入正式仓库，并同步整仓 shallow 状态。"""
        repo = self._open_repo()
        active_repo = repo
        branch_name = self.env_config.git_branch
        remote_name = f'one-dragon-fetch-{uuid.uuid4().hex}'
        # 直接使用本地原生路径作为 remote，不能转成 file:// URI：
        # UNC 路径（\\server\share\...）转出的 file://server/share/... 带远程主机部分，
        # libgit2 的本地 transport 只接受 file:/// 绝对路径或 file://localhost/，
        # 会把整个 URI 当成文件系统路径导致 "failed to resolve path"。
        remote_path = str(Path(temp_repo_dir).resolve())
        remote_branch_ref = f'refs/remotes/{self.env_config.git_remote}/{branch_name}'
        if tag_name is None:
            source_ref = f'refs/heads/{branch_name}'
            target_ref = remote_branch_ref
        else:
            source_ref = f'refs/tags/{tag_name}'
            target_ref = source_ref
        refspecs = [f'+{source_ref}:{target_ref}']
        affected_refs = {target_ref}
        if tag_name is not None:
            affected_refs.add(remote_branch_ref)
        if import_primary_branch and tag_name is None:
            primary_branch = self.repo_config.primary_branch
            if branch_name != primary_branch:
                primary_source_ref = f'refs/heads/{primary_branch}'
                primary_target_ref = (
                    f'refs/remotes/{self.env_config.git_remote}/{primary_branch}'
                )
                refspecs.insert(0, f'+{primary_source_ref}:{primary_target_ref}')
                affected_refs.add(primary_target_ref)

        original_ref_targets: dict[str, Oid | None] = {}
        for ref_name in affected_refs:
            original_ref_targets[ref_name] = (
                repo.references[ref_name].target
                if ref_name in repo.references
                else None
            )
        shallow_path = _get_repository_shallow_path(repo)

        def report_progress(progress: float, message: str) -> None:
            if progress_callback is not None:
                progress_callback(
                    stage_start + (stage_end - stage_start) * progress,
                    message,
                )

        callbacks = _FetchProgressRemoteCallbacks(report_progress, timeout=None)

        hidden_refs: list[tuple[str, Oid]] = []
        try:
            log.info(f'开始导入临时 Git 仓库: {remote_path}')
            hidden_refs = self._hide_non_commit_refs(active_repo)
            active_repo.remotes.create(remote_name, remote_path)
            active_repo.config[f'remote.{remote_name}.tagopt'] = '--no-tags'
            remote = active_repo.remotes[remote_name]
            remote.fetch(refspecs=refspecs, depth=0, callbacks=callbacks)
            callbacks.flush_sideband_progress()
            _sync_shallow_file(
                active_repo,
                temp_repo_dir,
                original_shallow_snapshot,
                shallow_authoritative,
            )
            active_repo.remotes.delete(remote_name)
            self._repo = None
            active_repo.free()
            active_repo = self._open_repo()

            if tag_name is not None:
                tag_object = active_repo.revparse_single(target_ref)
                tag_commit = tag_object.peel(Commit)
                active_repo.references.create(remote_branch_ref, tag_commit.id, force=True)
            log.info(f'临时 Git 仓库导入完成: branch={branch_name}')
        except BaseException:
            with contextlib.suppress(Exception):
                if remote_name in active_repo.remotes.names():
                    active_repo.remotes.delete(remote_name)
            with contextlib.suppress(Exception):
                active_repo.free()
            self._repo = None
            try:
                _write_shallow_snapshot(shallow_path, original_shallow_snapshot)
                rollback_repo = self._open_repo()
                if remote_name in rollback_repo.remotes.names():
                    rollback_repo.remotes.delete(remote_name)
                for ref_name, original_target in original_ref_targets.items():
                    if original_target is None:
                        if ref_name in rollback_repo.references:
                            rollback_repo.references.delete(ref_name)
                    else:
                        rollback_repo.references.create(
                            ref_name,
                            original_target,
                            force=True,
                        )
            except Exception:
                log.error('恢复 Git 导入前状态失败', exc_info=True)
            raise
        finally:
            if hidden_refs:
                try:
                    restore_repo = self._open_repo()
                    self._restore_hidden_refs(restore_repo, hidden_refs)
                    restore_repo.free()
                except Exception:
                    log.error(
                        '恢复导入前隐藏的非 commit 引用失败，对象仍保留在对象库，'
                        '可用 git fsck --lost-found 找回',
                        exc_info=True,
                    )

    def _fetch_remote_once(
        self,
        remote_url: str,
        progress_callback: Callable[[float, str], None] | None,
        stage_start: float,
        stage_end: float,
        tag_name: str | None = None,
    ) -> None:
        """在线程中拉取单个代码源，超时后作废本次尝试。"""
        repo = self._open_repo()
        branch_name = self.env_config.git_branch
        primary_branch = self.repo_config.primary_branch
        local_ref = f'refs/heads/{branch_name}'
        try:
            shallow_snapshot = _read_shallow_snapshot(_get_repository_shallow_path(repo))
            _parse_shallow_snapshot(shallow_snapshot)
        except ValueError as error:
            raise _LocalGitMetadataError('shallow metadata corrupted') from error
        base_ref: str | None = None
        base_oid: Oid | None = None
        fetch_primary_branch = False

        local_oid = (
            repo.references[local_ref].target
            if local_ref in repo.references
            else None
        )
        if tag_name is not None:
            depth = 1
        elif local_oid is not None:
            self._validate_history(repo, local_oid)
            depth = 0
            base_ref = local_ref
            base_oid = local_oid
        elif branch_name != primary_branch:
            base_oid = self._get_primary_branch_oid(repo, primary_branch)
            if base_oid is None:
                depth = 1
            else:
                try:
                    self._validate_history(repo, base_oid)
                except _LocalGitMetadataError:
                    repo, shallow_snapshot = self._repair_primary_branch_shallow(
                        repo,
                        base_oid,
                        shallow_snapshot,
                        primary_branch,
                    )
                depth = 0
                base_ref = f'refs/heads/{primary_branch}'
                fetch_primary_branch = True
        else:
            depth = 1

        proxy = self._get_proxy_address()
        temp_root = Path(os_utils.get_path_under_work_dir('.install', 'git_fetch_tmp'))
        temp_root.mkdir(parents=True, exist_ok=True)
        _cleanup_stale_fetch_repositories_once(temp_root)
        temp_repo_dir = tempfile.mkdtemp(prefix=f'fetch_{os.getpid()}_', dir=temp_root)
        source_objects_dir = str(_get_repository_objects_path(repo))
        messages: queue.Queue[dict[str, object]] = queue.Queue()
        abandoned = threading.Event()
        worker = threading.Thread(
            target=_fetch_remote_worker,
            args=(
                temp_repo_dir,
                source_objects_dir,
                remote_url,
                branch_name,
                depth,
                proxy,
                messages.put,
                abandoned,
                f'refs/tags/{tag_name}' if tag_name is not None else None,
                base_ref,
                str(base_oid) if base_oid is not None else None,
                fetch_primary_branch,
                shallow_snapshot,
            ),
            daemon=True,
            name='git-fetch-worker',
        )
        result: dict[str, object] | None = None
        has_message = False
        last_message_at: float | None = None

        def handle_message(message: dict[str, object]) -> None:
            nonlocal has_message, last_message_at, result
            has_message = True
            last_message_at = time.monotonic()
            message_type = message.get('type')
            if message_type == 'progress':
                progress = float(message.get('progress', 0.0))
                mapped_progress = stage_start + (stage_end - stage_start) * progress
                if progress_callback is not None:
                    progress_callback(mapped_progress, str(message.get('message', '')))
            elif message_type == 'result':
                result = message

        try:
            log.info(
                f'启动 Git fetch worker: branch={branch_name}, depth={depth}, '
                f'proxy={bool(proxy)}, timeout={REMOTE_FETCH_TIMEOUT:g}s'
            )
            worker.start()
            started_at = time.monotonic()

            while result is None:
                if not worker.is_alive() and messages.empty():
                    break

                now = time.monotonic()
                total_remaining = REMOTE_FETCH_TIMEOUT - (now - started_at)
                if has_message:
                    assert last_message_at is not None
                    activity_remaining = REMOTE_FETCH_IDLE_TIMEOUT - (now - last_message_at)
                    activity_timeout_message = (
                        f'Git 远程拉取消息空闲超过 {REMOTE_FETCH_IDLE_TIMEOUT:g} 秒'
                    )
                else:
                    activity_remaining = REMOTE_FETCH_INITIAL_TIMEOUT - (now - started_at)
                    activity_timeout_message = (
                        f'Git 远程拉取首条消息超过 {REMOTE_FETCH_INITIAL_TIMEOUT:g} 秒'
                    )

                if total_remaining <= 0:
                    timeout_message = f'Git 远程拉取超过 {REMOTE_FETCH_TIMEOUT:g} 秒'
                elif activity_remaining <= 0:
                    timeout_message = activity_timeout_message
                else:
                    timeout_message = ''

                if timeout_message:
                    abandoned.set()
                    log.warning(
                        f'Git fetch worker 超时，作废本次尝试: url={remote_url}, {timeout_message}'
                    )
                    raise TimeoutError(timeout_message)

                wait_timeout = min(total_remaining, activity_remaining, 0.1)
                with contextlib.suppress(queue.Empty):
                    handle_message(messages.get(timeout=max(wait_timeout, 0.001)))

            while not messages.empty():
                handle_message(messages.get_nowait())

            if result is None:
                raise RuntimeError('Git fetch worker 异常退出，未返回结果')
            if not bool(result.get('success')):
                raise RuntimeError(str(result.get('error', '未知错误')))

            worker.join(timeout=2)
            if worker.is_alive():
                abandoned.set()
                raise RuntimeError('Git fetch worker 返回结果后未能退出')

            self._import_fetch_result(
                temp_repo_dir,
                progress_callback,
                stage_start,
                stage_end,
                tag_name,
                bool(result.get('primary_branch_incremental')),
                shallow_snapshot,
                bool(result.get('shallow_authoritative', True)),
            )
        except BaseException:
            if worker.is_alive():
                abandoned.set()
            raise
        finally:
            if not worker.is_alive():
                _remove_temp_repo(temp_repo_dir)

    @staticmethod
    def _is_missing_object_error(error: BaseException) -> bool:
        """判断异常是否明确表示本地 Git 对象缺失。"""
        if not isinstance(error, KeyError):
            return False
        message = str(error)
        if 'object not found' in message:
            return True
        return re.fullmatch(r"'{0,1}[0-9a-f]{40}'{0,1}", message) is not None

    @staticmethod
    def _hide_non_commit_refs(repo: Repository) -> list[tuple[str, Oid]]:
        """删除指向非 commit 对象的引用并记录原名与目标，规避 libgit2 导入报错。

        libgit2 本地 transport 导入 fetch 时会枚举正式仓库的全部引用做 revwalk
        hide，若某个引用指向 tree/blob 等非 commit 对象（如 Codex CLI 的
        checkpoint 引用），会报 "object is not a committish" 导致代码同步失败。
        导入前临时删除这些引用，导入完成后按记录重建；对象本身仍留在对象库，
        重建不丢数据。
        """
        hidden: list[tuple[str, Oid]] = []
        for ref_name in list(repo.references):
            ref = repo.references[ref_name]
            if ref.type != ReferenceType.DIRECT:
                continue
            try:
                repo[ref.target].peel(Commit)
            except Exception:
                pass
            else:
                continue
            hidden.append((ref.name, ref.target))
            repo.references.delete(ref.name)
        if hidden:
            log.info(
                '导入前临时隐藏 %d 个非 commit 引用: %s',
                len(hidden),
                ', '.join(name for name, _ in hidden),
            )
        return hidden

    @staticmethod
    def _restore_hidden_refs(repo: Repository, hidden: list[tuple[str, Oid]]) -> None:
        """按记录重建导入前删除的非 commit 引用。"""
        for ref_name, target in hidden:
            repo.references.create(ref_name, target, force=True)
        if hidden:
            log.info(f'已恢复 {len(hidden)} 个非 commit 引用')

    def _rebuild_repository(
        self,
        progress_callback: Callable[[float, str], None] | None,
        initial_tag: str | None = None,
    ) -> tuple[GitSyncStatus, str]:
        """备份损坏的 Git 目录并重新建立浅仓库。"""
        failure_message = gt('本地代码更新失败')
        if self._rebuilding_repository:
            log.error('本地 Git 仓库重建期间再次检测到损坏，停止重复重建')
            return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message

        opened_repo: Repository | None = None
        try:
            discovered_git_dir = discover_repository(self.repo_dir)
            if not discovered_git_dir:
                log.error(f'目录 {self.repo_dir} 不是有效的 Git 仓库，无法备份')
                return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message

            git_dir = Path(discovered_git_dir).resolve()
            if not git_dir.is_dir():
                log.error(f'本地 Git 目录不存在，无法备份: {git_dir}')
                return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message
            if (git_dir / 'commondir').is_file() or (git_dir / 'worktrees').is_dir():
                log.error('检测到关联 worktree，暂不自动重建本地 Git 仓库')
                return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message

            try:
                opened_repo = Repository(str(git_dir))
            except Exception:
                log.warning('旧 Git 仓库无法打开，跳过 remote 检查并直接备份重建')
            else:
                # one-dragon-fetch-* 是导入临时仓库时创建的内部 remote，
                # 正常用完即删；仅进程被强杀等场景可能残留空配置，不算用户额外 remote。
                extra_remotes = sorted(
                    name
                    for name in opened_repo.remotes.names()
                    if name != 'origin' and not name.startswith('one-dragon-fetch-')
                )
                if extra_remotes:
                    log.error(f'检测到 origin 以外的远程仓库，暂不自动重建本地 Git 仓库: {extra_remotes}')
                    return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message

            timestamp = time.strftime('%Y%m%d_%H%M%S')
            backup_dir = git_dir.with_name(f'{git_dir.name}.corrupted.{timestamp}')
            if backup_dir.exists():
                backup_dir = git_dir.with_name(f'{git_dir.name}.corrupted.{timestamp}.{uuid.uuid4().hex[:8]}')

            log.warning(f'检测到本地 Git 仓库损坏，备份旧 Git 目录: {git_dir} -> {backup_dir}')
            cached_repo = self._repo
            self._repo = None
            if cached_repo is not None:
                with contextlib.suppress(Exception):
                    cached_repo.free()
            if opened_repo is not None and opened_repo is not cached_repo:
                with contextlib.suppress(Exception):
                    opened_repo.free()
                opened_repo = None
            git_dir.rename(backup_dir)

            self._rebuilding_repository = True
            try:
                status, message = self._clone_repository(progress_callback, initial_tag)
            finally:
                self._rebuilding_repository = False

            if status is not GitSyncStatus.SUCCESS:
                log.error(f'本地 Git 仓库重建失败，旧目录已保留: {message}')
                return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message
            log.info(f'本地 Git 仓库重建完成，旧目录备份于: {backup_dir}')
            return GitSyncStatus.SUCCESS, gt('本地代码更新完成')
        except Exception:
            log.error('备份或重建本地 Git 仓库失败，保留现场', exc_info=True)
            return GitSyncStatus.LOCAL_UPDATE_FAILED, failure_message
        finally:
            if opened_repo is not None:
                with contextlib.suppress(Exception):
                    opened_repo.free()

    def _fetch_remote(
        self,
        progress_callback: Callable[[float, str], None] | None = None,
        stage_start: float = 0.0,
        stage_end: float = 1.0,
        tag_name: str | None = None,
    ) -> tuple[GitSyncStatus, str]:
        """按候选代码源顺序拉取远程代码，失败后自动回退。"""
        log.info(gt('拉取远程代码...'))
        success = False
        used_repository: RepositoryItem | None = None
        local_repository_damaged = False

        try:
            for repository, repository_url in self._get_repository_candidates():
                repository_name = self._get_repository_item(repository).ui_text
                log.info(f'尝试代码源: {repository_name}')
                if progress_callback is not None:
                    progress_callback(stage_start, f'尝试代码源: {repository_name}')
                try:
                    self._fetch_remote_once(
                        repository_url,
                        progress_callback,
                        stage_start,
                        stage_end,
                        tag_name,
                    )
                    success = True
                    used_repository = repository
                    break
                except _LocalGitMetadataError:
                    local_repository_damaged = True
                    log.error('本地 Git 元数据损坏，停止代码源回退', exc_info=True)
                    break
                except Exception as error:
                    if self._is_missing_object_error(error):
                        local_repository_damaged = True
                        log.error('导入远程对象时发现本地 Git 对象缺失，停止代码源回退', exc_info=True)
                        break
                    log.warning(f'代码源 {repository_name} 拉取失败，尝试下一个代码源', exc_info=True)
        finally:
            restored = True if local_repository_damaged else self._restore_origin()

        if not success and local_repository_damaged:
            if self._rebuilding_repository:
                return GitSyncStatus.LOCAL_UPDATE_FAILED, gt('本地代码更新失败')
            log.warning('检测到本地 Git 仓库损坏，开始自动备份并重建本地 Git 仓库')
            return self._rebuild_repository(progress_callback, tag_name)
        if not restored:
            log.error('拉取结束后无法恢复主仓库远程地址')
            return GitSyncStatus.LOCAL_UPDATE_FAILED, gt('本地代码更新失败')
        if not success:
            log.error('所有代码源均拉取失败')
            return GitSyncStatus.REMOTE_UNAVAILABLE, gt('暂时无法获取更新')

        if used_repository is not None:
            try:
                self.env_config.last_repository_url = used_repository.url
            except Exception:
                log.warning('记录上次成功代码源失败', exc_info=True)

        used_repository_name = self._get_repository_item(used_repository).ui_text if used_repository else ''
        log.info(f'远程代码拉取成功，实际代码源: {used_repository_name}')
        if progress_callback is not None:
            progress_callback(stage_end, gt('拉取远程代码成功'))
        return GitSyncStatus.SUCCESS, gt('拉取远程代码成功')

    def _reset_hard(self, target_id: str | Oid) -> bool:
        """硬重置仓库到指定提交
        会丢弃工作区和暂存区的所有修改

        Args:
            target_id: 目标提交ID，支持以下格式:
                - OID 对象: pygit2.Oid 实例
                - 提交哈希: 完整或短格式的 commit hash (如 'abc123' 或 'abc123def456...')
                - 引用名称: 分支名、标签名等 (如 'main', 'v1.0.0')
                - 相对引用: HEAD~1, HEAD^, origin/main 等

        Returns:
            是否成功
        """
        try:
            repo = self._open_repo()
            # 如果是字符串，需要先解析为OID对象
            if isinstance(target_id, str):
                obj = repo.revparse_single(target_id)
                target_oid = obj.id
            else:
                target_oid = target_id

            repo.reset(target_oid, ResetMode.HARD)
            return True
        except Exception:
            log.error(f'重置到提交 {target_id} 失败', exc_info=True)
            return False

    def _get_local_and_remote_oid(self) -> tuple[Oid | None, Oid | None, str]:
        """获取本地HEAD和远程分支的提交ID

        Returns:
            (本地提交ID, 远程提交ID, 错误消息) - 远程提交ID为None时表示失败
        """
        try:
            repo = self._open_repo()
            local_oid = repo.head.target
        except Exception:
            local_oid = None
            msg = gt('获取本地提交信息失败')
            log.error(msg, exc_info=True)
            return local_oid, None, msg

        # 检查远程分支是否存在
        remote_branch_name = f'{self.env_config.git_remote}/{self.env_config.git_branch}'
        remote_ref = f'refs/remotes/{remote_branch_name}'
        if remote_ref not in repo.references:
            msg = f'{gt("远程分支不存在")}: {remote_branch_name}'
            log.error(msg)
            return local_oid, None, msg

        try:
            remote_oid = repo.references[remote_ref].target
        except Exception:
            msg = gt('获取远程提交信息失败')
            log.error(msg, exc_info=True)
            return local_oid, None, msg

        return local_oid, remote_oid, ''

    def _validate_working_directory(self) -> tuple[GitSyncStatus, str]:
        """验证工作区状态。"""
        log.info(gt('检测当前代码是否有修改'))
        try:
            repo = self._open_repo()
            is_clean = len(repo.status()) == 0
        except Exception:
            log.error('检测当前代码是否有修改失败', exc_info=True)
            return GitSyncStatus.LOCAL_UPDATE_FAILED, gt('检测当前代码状态失败')

        if not is_clean and not self.env_config.force_update:
            return GitSyncStatus.LOCAL_CHANGES, gt('检测到程序文件有改动')

        return GitSyncStatus.SUCCESS, ''

    def _get_commit_walker(self, sort_mode: SortMode = SortMode.TOPOLOGICAL) -> Walker | None:
        """获取commit遍历器

        Args:
            sort_mode: 排序模式

        Returns:
            commit遍历器，失败时返回None
        """
        try:
            repo = self._open_repo()
            head_target = repo.head.target
            return repo.walk(head_target, sort_mode)
        except Exception:
            log.error('获取commit遍历器失败', exc_info=True)
            return None

    def _get_file_at_commit(self, commit_oid: Oid, file_path: str) -> bytes | None:
        """获取指定 commit 中某文件的内容

        Args:
            commit_oid: 提交 OID
            file_path: 相对于仓库根目录的文件路径（如 'deploy/module_manifest.py'）

        Returns:
            文件内容的字节，文件不存在时返回 None
        """
        try:
            repo = self._open_repo()
            obj = repo.revparse_single(f'{commit_oid}:{file_path}')
            if isinstance(obj, Blob):
                return obj.data
            return None
        except (KeyError, ValueError):
            return None

    # ================== 模块清单检查 ==================

    def _check_manifest_compatible(self, target_oid: Oid) -> tuple[bool, str]:
        """检查模块清单是否与当前运行环境兼容

        本地清单从 .runtime/module_manifest.py 读取（打包时写入），
        远程清单路径从目标 commit 的 project.yml 的 manifest_path 字段获取。
        仅在 frozen 环境（PyInstaller 打包后）下执行检查。

        Args:
            target_oid: 目标提交 OID

        Returns:
            (是否兼容, 提示消息)
        """
        if not getattr(sys, 'frozen', False):
            return True, ''

        # 读取本地 manifest（打包进 .runtime/ 的文件）
        runtime_dir = Path(getattr(sys, '_MEIPASS', ''))
        local_manifest_path = runtime_dir / 'module_manifest.py'
        if not local_manifest_path.is_file():
            return True, ''

        try:
            local_manifest = local_manifest_path.read_bytes()
        except Exception:
            log.warning('读取本地模块清单失败，跳过检查', exc_info=True)
            return True, ''

        # 从目标 commit 的 project.yml 获取清单路径
        manifest_git_path = self._get_manifest_path_from_commit(target_oid)
        if not manifest_git_path:
            return True, ''

        # 读取目标 commit 中的 manifest
        remote_manifest = self._get_file_at_commit(target_oid, manifest_git_path)
        if remote_manifest is None:
            return True, ''

        if local_manifest == remote_manifest:
            return True, ''

        # 兼容旧打包产物的 CRLF；先走 raw bytes 快路径，仅不一致时再归一化。
        if b'\r' in local_manifest or b'\r' in remote_manifest:
            normalized_local_manifest = local_manifest.replace(b'\r\n', b'\n').replace(b'\r', b'\n')
            normalized_remote_manifest = remote_manifest.replace(b'\r\n', b'\n').replace(b'\r', b'\n')
            if normalized_local_manifest == normalized_remote_manifest:
                return True, ''

        msg = gt('目标版本的运行环境与当前不兼容')
        log.warning(f'模块清单已变更，阻止代码更新。目标: {str(target_oid)[:7]}')
        return False, msg

    def _get_manifest_path_from_commit(self, commit_oid: Oid) -> str | None:
        """从指定 commit 的 project.yml 中读取 manifest_path

        Args:
            commit_oid: 目标提交 OID

        Returns:
            清单文件的仓库路径，读取失败时返回 None
        """
        raw = self._get_file_at_commit(commit_oid, 'config/project.yml')
        if raw is None:
            return None
        try:
            data = yaml.safe_load(raw)
            path = data.get('manifest_path') if isinstance(data, dict) else None
            return path if isinstance(path, str) and path else None
        except Exception:
            return None

    def _check_remote_manifest_compatible(self) -> tuple[bool, str]:
        """检查远程分支的模块清单是否与当前运行环境兼容

        封装远程 OID 解析 + 清单比对，异常时跳过检查。

        Returns:
            (是否兼容, 提示消息)
        """
        remote_ref = f'refs/remotes/{self.env_config.git_remote}/{self.env_config.git_branch}'
        try:
            repo = self._open_repo()
            if remote_ref not in repo.references:
                return True, ''
            remote_oid = repo.references[remote_ref].target
            return self._check_manifest_compatible(remote_oid)
        except Exception:
            log.warning('检查模块清单时出错，跳过检查', exc_info=True)
            return True, ''

    def _checkout_branch(self) -> tuple[GitSyncStatus, bool, str]:
        """切换到目标分支，并返回是否发生了分支切换。"""
        try:
            repo = self._open_repo()
        except Exception:
            message = gt('切换到目标版本失败')
            log.error('打开本地仓库失败', exc_info=True)
            return GitSyncStatus.LOCAL_UPDATE_FAILED, False, message

        remote_name = self.env_config.git_remote
        branch_name = self.env_config.git_branch
        remote_branch_name = f'{remote_name}/{branch_name}'
        local_ref = f'refs/heads/{branch_name}'
        remote_ref = f'refs/remotes/{remote_branch_name}'
        try:
            branch_changed = repo.head.name != local_ref
        except Exception:
            branch_changed = True

        if local_ref not in repo.references:
            if remote_ref in repo.references:
                try:
                    remote_commit = repo.get(repo.references[remote_ref].target)
                    repo.create_branch(branch_name, remote_commit)
                    log.debug(f'从远程分支创建本地分支: {branch_name}')
                except Exception:
                    message = gt('切换到目标版本失败')
                    log.error(f'创建本地分支 {branch_name} 失败', exc_info=True)
                    return GitSyncStatus.LOCAL_UPDATE_FAILED, False, message
            else:
                message = gt('切换到目标版本失败')
                log.error(f'本地和远程都不存在分支 {branch_name}')
                return GitSyncStatus.LOCAL_UPDATE_FAILED, False, message

        try:
            repo.checkout(local_ref, strategy=CheckoutStrategy.FORCE)
            repo.set_head(local_ref)
            log.info(f'成功切换到分支 {branch_name}')
            return GitSyncStatus.SUCCESS, branch_changed, ''
        except Exception:
            message = gt('切换到目标版本失败')
            log.error(f'切换到分支 {branch_name} 失败', exc_info=True)
            return GitSyncStatus.LOCAL_UPDATE_FAILED, False, message

    def _sync_with_remote(self, force: bool) -> tuple[GitSyncStatus, str]:
        """同步本地代码到远程分支状态。"""
        local_oid, remote_oid, message = self._get_local_and_remote_oid()
        if remote_oid is None:
            return GitSyncStatus.LOCAL_UPDATE_FAILED, message

        if local_oid is None:
            if force:
                if self._reset_hard(remote_oid):
                    log.debug(f'重置到远程提交成功: {str(remote_oid)[:7]}')
                    return GitSyncStatus.SUCCESS, gt('更新完成')

                message = gt('重置到远程提交失败')
                log.error(f'{message}: {str(remote_oid)[:7]}')
                return GitSyncStatus.LOCAL_UPDATE_FAILED, message

            message = gt('HEAD 不存在且未开启强制更新')
            log.error(message)
            return GitSyncStatus.LOCAL_UPDATE_FAILED, message

        if local_oid == remote_oid:
            log.info(f'本地代码已是最新: {str(local_oid)[:7]}')
            return GitSyncStatus.UP_TO_DATE, gt('当前已是最新版本')

        can_fast_forward = False
        with contextlib.suppress(Exception):
            repo = self._open_repo()
            can_fast_forward = repo.descendant_of(remote_oid, local_oid) and len(repo.status()) == 0

        if can_fast_forward:
            if self._reset_hard(remote_oid):
                log.debug(f'快进更新成功: {str(local_oid)[:7]} -> {str(remote_oid)[:7]}')
                return GitSyncStatus.SUCCESS, gt('更新完成')

            message = gt('快进更新失败')
            log.error(f'{message}: {str(local_oid)[:7]} -> {str(remote_oid)[:7]}')
            return GitSyncStatus.LOCAL_UPDATE_FAILED, message

        if force:
            if self._reset_hard(remote_oid):
                log.debug(f'强制更新成功: {str(local_oid)[:7]} -> {str(remote_oid)[:7]}')
                return GitSyncStatus.SUCCESS, gt('更新完成')

            message = gt('强制更新失败')
            log.error(f'{message}: {str(local_oid)[:7]} -> {str(remote_oid)[:7]}')
            return GitSyncStatus.LOCAL_UPDATE_FAILED, message

        message = gt('检测到程序文件有改动')
        log.error(f'本地代码无法快进且未开启强制更新: {str(local_oid)[:7]} -> {str(remote_oid)[:7]}')
        return GitSyncStatus.LOCAL_CHANGES, message

    def _clone_repository(
        self,
        progress_callback: Callable[[float, str], None] | None = None,
        initial_tag: str | None = None,
    ) -> tuple[GitSyncStatus, str]:
        """初始化本地仓库并同步远程目标分支。"""
        if progress_callback:
            progress_callback(1 / 6, gt('初始化本地 Git 仓库'))

        try:
            init_repository(self.repo_dir)
        except Exception:
            message = gt('初始化本地 Git 仓库失败')
            log.error(message, exc_info=True)
            return GitSyncStatus.LOCAL_UPDATE_FAILED, message

        if progress_callback:
            progress_callback(2 / 6, gt('拉取远程代码'))

        fetch_status, message = self._fetch_remote(progress_callback, 2 / 6, 3 / 6, initial_tag)
        if fetch_status is not GitSyncStatus.SUCCESS:
            if initial_tag is not None and fetch_status is GitSyncStatus.REMOTE_UNAVAILABLE:
                return GitSyncStatus.BUILTIN_TAG_UNAVAILABLE, gt('暂时无法获取当前版本所需文件')
            return fetch_status, message

        if initial_tag is None:
            if progress_callback:
                progress_callback(3 / 6, gt('检查运行环境兼容性'))

            compatible, message = self._check_remote_manifest_compatible()
            if not compatible:
                return GitSyncStatus.RUNTIME_INCOMPATIBLE, message

        if progress_callback:
            progress_callback(4 / 6, gt('切换到目标分支'))

        checkout_status, _, message = self._checkout_branch()
        if checkout_status is not GitSyncStatus.SUCCESS:
            return checkout_status, message

        if progress_callback:
            progress_callback(5 / 6, gt('同步本地代码'))

        sync_status, message = self._sync_with_remote(force=True)
        if sync_status not in (GitSyncStatus.SUCCESS, GitSyncStatus.UP_TO_DATE):
            return sync_status, message

        if progress_callback:
            progress_callback(6 / 6, gt('克隆仓库成功'))

        return GitSyncStatus.SUCCESS, gt('更新完成')

    def _fetch_and_checkout_latest_branch(
        self,
        progress_callback: Callable[[float, str], None] | None = None,
    ) -> tuple[GitSyncStatus, str]:
        """切换到最新的目标分支并更新代码。"""
        log.info(gt('核对当前仓库'))

        if progress_callback:
            progress_callback(1 / 6, gt('拉取远程代码'))

        fetch_status, message = self._fetch_remote(progress_callback, 1 / 6, 2 / 6)
        if fetch_status is not GitSyncStatus.SUCCESS:
            return fetch_status, message

        if progress_callback:
            progress_callback(2 / 6, gt('检查运行环境兼容性'))

        compatible, message = self._check_remote_manifest_compatible()
        if not compatible:
            return GitSyncStatus.RUNTIME_INCOMPATIBLE, message

        if progress_callback:
            progress_callback(3 / 6, gt('检查工作区状态'))

        validate_status, message = self._validate_working_directory()
        if validate_status is not GitSyncStatus.SUCCESS:
            return validate_status, message

        if progress_callback:
            progress_callback(4 / 6, gt('切换到目标分支'))

        checkout_status, branch_changed, message = self._checkout_branch()
        if checkout_status is not GitSyncStatus.SUCCESS:
            return checkout_status, message

        if progress_callback:
            progress_callback(5 / 6, gt('同步本地代码'))

        sync_status, message = self._sync_with_remote(self.env_config.force_update)
        if sync_status not in (GitSyncStatus.SUCCESS, GitSyncStatus.UP_TO_DATE):
            return sync_status, message
        if branch_changed and sync_status is GitSyncStatus.UP_TO_DATE:
            sync_status = GitSyncStatus.SUCCESS
            message = gt('更新完成')

        if progress_callback:
            progress_callback(6 / 6, message)

        return sync_status, message

    # ================== 公共 API ==================

    def check_repo_exists(self) -> bool:
        """检查本地仓库是否存在。"""
        return discover_repository(self.repo_dir) is not None

    def is_initial_checkout_pending(self) -> bool:
        """检查目标本地分支是否尚未建立。"""
        if not self.check_repo_exists():
            return True

        local_ref = f'refs/heads/{self.env_config.git_branch}'
        try:
            return local_ref not in self._open_repo().references
        except Exception:
            log.warning('无法判断首次 checkout 是否完成，按已有仓库更新处理', exc_info=True)
            return False

    def fetch_latest_code(
        self,
        progress_callback: Callable[[float, str], None] | None = None,
        initial_tag: str | None = None,
    ) -> tuple[GitSyncStatus, str]:
        """更新最新代码，并返回明确的同步状态和中性结果文案。"""
        try:
            if self.is_initial_checkout_pending():
                status, _ = self._clone_repository(progress_callback, initial_tag)
            else:
                status, _ = self._fetch_and_checkout_latest_branch(progress_callback)
        except Exception:
            log.error('更新代码时发生未处理异常', exc_info=True)
            status = GitSyncStatus.FAILED
        return status, gt(status.value)

    def get_current_branch(self) -> str | None:
        """
        获取当前分支名称
        """
        log.info(gt('检测当前代码分支'))
        try:
            repo = self._open_repo()
            head = repo.head
            return head.shorthand if head else None
        except Exception:
            log.error('获取当前分支失败', exc_info=True)
            return None

    def get_head_commit_id(self, short: bool = False) -> str | None:
        """获取当前 HEAD 的 commit hash"""
        try:
            repo = self._open_repo()
            oid = str(repo.head.target)
            return oid[:8] if short else oid
        except Exception:
            return None

    def is_current_branch_latest(self) -> tuple[bool, str]:
        """
        当前分支是否已经最新 与远程分支一致
        """
        log.info(gt('检测当前代码是否最新'))

        fetch_status, message = self._fetch_remote()
        if fetch_status is not GitSyncStatus.SUCCESS:
            return False, message

        # 获取本地和远程的提交ID
        local_oid, remote_oid, msg = self._get_local_and_remote_oid()
        if local_oid is None or remote_oid is None:
            log.error(msg)
            return False, msg

        # 比较提交是否相同
        if local_oid == remote_oid:
            return True, ''

        return False, gt('与远程分支不一致')

    def fetch_total_commit(self) -> int:
        """
        获取commit的总数。获取失败时返回0
        """
        log.info(gt('获取commit总数'))
        walker = self._get_commit_walker()
        return sum(1 for _ in walker) if walker else 0

    def fetch_page_commit(self, page_num: int, page_size: int) -> list[GitLog]:
        """获取分页commit

        Args:
            page_num: 页码（从0开始）
            page_size: 每页数量

        Returns:
            GitLog列表
        """
        log.info(f"{gt('获取commit')} 第{page_num + 1}页")
        walker = self._get_commit_walker()
        if not walker:
            return []

        logs: list[GitLog] = []
        for idx, commit in enumerate(walker):
            if idx < page_num * page_size:
                continue
            if len(logs) >= page_size:
                break

            short_id = str(commit.id)[:7]
            author = commit.author.name if commit.author and commit.author.name else ''
            commit_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(commit.commit_time))
            message = commit.message.splitlines()[0] if commit.message else ''

            logs.append(GitLog(short_id, author, commit_time, message))

        return logs

    def update_remote(self) -> None:
        """
        更新remote
        """
        if not self.check_repo_exists():
            return

        try:
            self._ensure_remote()
        except Exception:
            log.error('更新远程仓库地址失败', exc_info=True)

    def reset_to_commit(self, commit_id: str) -> tuple[bool, str]:
        """
        回滚到特定commit，会先检查模块清单兼容性

        Returns:
            (是否成功, 提示消息)
        """
        try:
            repo = self._open_repo()
            obj = repo.revparse_single(commit_id)
            target_oid = obj.id
        except Exception:
            log.error(f'解析提交ID失败: {commit_id}', exc_info=True)
            return False, gt('解析提交ID失败')

        compatible, msg = self._check_manifest_compatible(target_oid)
        if not compatible:
            return False, msg

        if self._reset_hard(target_oid):
            return True, ''
        return False, gt('回滚失败')

    def get_current_version(self) -> str | None:
        """
        获取当前代码版本
        """
        logs = self.fetch_page_commit(0, 1)
        return logs[0].commit_id if logs else None

    def get_latest_tag(self) -> tuple[str, str]:
        """获取最新tag，未找到时返回空字符串

        Returns:
            (最新稳定版, 最新测试版)
        """
        # 如果不存在本地仓库，返回空
        if not self.check_repo_exists():
            return '', ''

        heads = None
        try:
            for repository, repository_url in self._get_repository_candidates():
                repository_name = self._get_repository_item(repository).ui_text
                try:
                    remote = self._ensure_remote(repository_url)
                    callbacks = self._create_fetch_callbacks(None, 0.0, 1.0)
                    with self._temporary_fetch_timeout():
                        heads = remote.list_heads(callbacks=callbacks, proxy=self._get_proxy_address())
                    log.info(f'标签查询使用代码源: {repository_name}')
                    break
                except Exception:
                    log.warning(f'代码源 {repository_name} 获取标签失败，尝试下一个代码源', exc_info=True)
        finally:
            restored = self._restore_origin()

        if heads is None or not restored:
            log.error('获取最新标签失败')
            return '', ''

        # 提取标签名称并解析为 Version 对象
        tags: dict[str, version.Version] = {}
        for h in heads:
            if h.name.startswith("refs/tags/"):
                tag = h.name[len("refs/tags/"):]
                # 验证是否为有效版本
                with contextlib.suppress(version.InvalidVersion):
                    parsed = version.parse(tag)
                    tags[tag] = parsed

        # 按 Version 对象排序
        versions = sorted(tags.items(), key=lambda x: x[1], reverse=True)

        # 找出最新的稳定版和测试版
        latest_stable = ''
        latest_beta = ''

        for tag, ver in versions:
            if ver.is_prerelease:
                if not latest_beta:
                    latest_beta = tag
            else:
                if not latest_stable:
                    latest_stable = tag
                    break

        return latest_stable, latest_beta