# Copyright (C) 2008 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import datetime
import errno
import filecmp
import functools
import glob
import os
import platform
import random
import re
import shutil
import stat
import string
import subprocess
import sys
import tarfile
import tempfile
import time
from typing import Dict, List, NamedTuple, Optional
import urllib.parse

from color import Coloring
from error import DownloadError
from error import GitAuthError
from error import GitError
from error import ManifestInvalidPathError
from error import ManifestInvalidRevisionError
from error import ManifestParseError
from error import NoManifestException
from error import RepoError
from error import UploadError
import fetch
from git_command import git_require
from git_command import GitCommand
from git_command import GitCommandError
from git_config import GetSchemeFromUrl
from git_config import GetUrlCookieFile
from git_config import GitConfig
from git_config import IsId
from git_refs import GitRefs
from git_refs import HEAD
from git_refs import R_HEADS
from git_refs import R_M
from git_refs import R_PUB
from git_refs import R_TAGS
from git_refs import R_WORKTREE_M
import git_superproject
from git_trace2_event_log import EventLog
import platform_utils
import progress
from repo_logging import RepoLogger
from repo_trace import Trace


logger = RepoLogger(__file__)


class SyncNetworkHalfResult(NamedTuple):
    """Sync_NetworkHalf return value."""

    # Did we query the remote? False when optimized_fetch is True and we have
    # the commit already present.
    remote_fetched: bool
    # Error from SyncNetworkHalf
    error: Exception = None

    @property
    def success(self) -> bool:
        return not self.error


class SyncNetworkHalfError(RepoError):
    """Failure trying to sync."""


class DeleteWorktreeError(RepoError):
    """Failure to delete worktree."""

    def __init__(
        self, *args, aggregate_errors: List[Exception] = None, **kwargs
    ) -> None:
        super().__init__(*args, **kwargs)
        self.aggregate_errors = aggregate_errors or []


class DeleteDirtyWorktreeError(DeleteWorktreeError):
    """Failure to delete worktree due to uncommitted changes."""


# Maximum sleep time allowed during retries.
MAXIMUM_RETRY_SLEEP_SEC = 3600.0
# +-10% random jitter is added to each Fetches retry sleep duration.
RETRY_JITTER_PERCENT = 0.1

# Whether to use alternates.  Switching back and forth is *NOT* supported.
# TODO(vapier): Remove knob once behavior is verified.
_ALTERNATES = os.environ.get("REPO_USE_ALTERNATES") == "1"


def _lwrite(path, content):
    lock = "%s.lock" % path

    # Maintain Unix line endings on all OS's to match git behavior.
    with open(lock, "w", newline="\n") as fd:
        fd.write(content)

    try:
        platform_utils.rename(lock, path)
    except OSError:
        platform_utils.remove(lock)
        raise


def not_rev(r):
    return "^" + r


def sq(r):
    return "'" + r.replace("'", "'''") + "'"


_project_hook_list = None


def _ProjectHooks():
    """List the hooks present in the 'hooks' directory.

    These hooks are project hooks and are copied to the '.git/hooks' directory
    of all subprojects.

    This function caches the list of hooks (based on the contents of the
    'repo/hooks' directory) on the first call.

    Returns:
        A list of absolute paths to all of the files in the hooks directory.
    """
    global _project_hook_list
    if _project_hook_list is None:
        d = os.path.realpath(os.path.abspath(os.path.dirname(__file__)))
        d = os.path.join(d, "hooks")
        _project_hook_list = [
            os.path.join(d, x) for x in platform_utils.listdir(d)
        ]
    return _project_hook_list


class DownloadedChange:
    _commit_cache = None

    def __init__(self, project, base, change_id, ps_id, commit):
        self.project = project
        self.base = base
        self.change_id = change_id
        self.ps_id = ps_id
        self.commit = commit

    @property
    def commits(self):
        if self._commit_cache is None:
            self._commit_cache = self.project.bare_git.rev_list(
                "--abbrev=8",
                "--abbrev-commit",
                "--pretty=oneline",
                "--reverse",
                "--date-order",
                not_rev(self.base),
                self.commit,
                "--",
            )
        return self._commit_cache


class ReviewableBranch:
    _commit_cache = None
    _base_exists = None

    def __init__(self, project, branch, base):
        self.project = project
        self.branch = branch
        self.base = base

    @property
    def name(self):
        return self.branch.name

    @property
    def current(self) -> bool:
        return getattr(self.branch, "current", False)

    @property
    def commits(self):
        if self._commit_cache is None:
            args = (
                "--abbrev=8",
                "--abbrev-commit",
                "--pretty=oneline",
                "--reverse",
                "--date-order",
                not_rev(self.base),
                R_HEADS + self.name,
                "--",
            )
            try:
                self._commit_cache = self.project.bare_git.rev_list(
                    *args, log_as_error=self.base_exists
                )
            except GitError:
                # We weren't able to probe the commits for this branch.  Was it
                # tracking a branch that no longer exists?  If so, return no
                # commits.  Otherwise, rethrow the error as we don't know what's
                # going on.
                if self.base_exists:
                    raise

                self._commit_cache = []

        return self._commit_cache

    @property
    def unabbrev_commits(self):
        r = {}
        for commit in self.project.bare_git.rev_list(
            not_rev(self.base), R_HEADS + self.name, "--"
        ):
            r[commit[0:8]] = commit
        return r

    @property
    def date(self):
        return self.project.bare_git.log(
            "--pretty=format:%cd", "-n", "1", R_HEADS + self.name, "--"
        )

    @property
    def modified_files(self) -> List[str]:
        return self.project.bare_git.diff(
            "--name-only", f"{self.base}...{R_HEADS}{self.name}"
        ).splitlines()

    @property
    def base_exists(self):
        """Whether the branch we're tracking exists.

        Normally it should, but sometimes branches we track can get deleted.
        """
        if self._base_exists is None:
            try:
                self.project.bare_git.rev_parse("--verify", not_rev(self.base))
                # If we're still here, the base branch exists.
                self._base_exists = True
            except GitError:
                # If we failed to verify, the base branch doesn't exist.
                self._base_exists = False

        return self._base_exists

    def UploadForReview(
        self,
        people,
        dryrun=False,
        topic=None,
        hashtags=(),
        labels=(),
        private=False,
        notify=None,
        wip=False,
        ready=False,
        dest_branch=None,
        validate_certs=True,
        push_options=None,
        patchset_description=None,
        git_event_log: Optional[EventLog] = None,
    ) -> None:
        self.project.UploadForReview(
            branch=self.name,
            people=people,
            dryrun=dryrun,
            topic=topic,
            hashtags=hashtags,
            labels=labels,
            private=private,
            notify=notify,
            wip=wip,
            ready=ready,
            dest_branch=dest_branch,
            validate_certs=validate_certs,
            push_options=push_options,
            patchset_description=patchset_description,
            git_event_log=git_event_log,
        )

    def GetPublishedRefs(self):
        refs = {}
        output = self.project.bare_git.ls_remote(
            self.branch.remote.SshReviewUrl(self.project.UserEmail),
            "refs/changes/*",
        )
        for line in output.split("\n"):
            try:
                (sha, ref) = line.split()
                refs[sha] = ref
            except ValueError:
                pass

        return refs


class StatusColoring(Coloring):
    def __init__(self, config):
        super().__init__(config, "status")
        self.project = self.printer("header", attr="bold")
        self.branch = self.printer("header", attr="bold")
        self.nobranch = self.printer("nobranch", fg="red")
        self.important = self.printer("important", fg="red")

        self.added = self.printer("added", fg="green")
        self.changed = self.printer("changed", fg="red")
        self.untracked = self.printer("untracked", fg="red")


class DiffColoring(Coloring):
    def __init__(self, config):
        super().__init__(config, "diff")
        self.project = self.printer("header", attr="bold")
        self.fail = self.printer("fail", fg="red")


class Annotation:
    def __init__(self, name, value, keep):
        self.name = name
        self.value = value
        self.keep = keep

    def __eq__(self, other):
        if not isinstance(other, Annotation):
            return False
        return self.__dict__ == other.__dict__

    def __lt__(self, other):
        # This exists just so that lists of Annotation objects can be sorted,
        # for use in comparisons.
        if not isinstance(other, Annotation):
            raise ValueError("comparison is not between two Annotation objects")
        if self.name == other.name:
            if self.value == other.value:
                return self.keep < other.keep
            return self.value < other.value
        return self.name < other.name


def _SafeExpandPath(base, subpath, skipfinal=False):
    """Make sure |subpath| is completely safe under |base|.

    We make sure no intermediate symlinks are traversed, and that the final path
    is not a special file (e.g. not a socket or fifo).

    NB: We rely on a number of paths already being filtered out while parsing
    the manifest.  See the validation logic in manifest_xml.py for more details.
    """
    # Split up the path by its components.  We can't use os.path.sep exclusively
    # as some platforms (like Windows) will convert / to \ and that bypasses all
    # our constructed logic here.  Especially since manifest authors only use
    # / in their paths.
    resep = re.compile(r"[/%s]" % re.escape(os.path.sep))
    components = resep.split(subpath)
    if skipfinal:
        # Whether the caller handles the final component itself.
        finalpart = components.pop()

    path = base
    for part in components:
        if part in {".", ".."}:
            raise ManifestInvalidPathError(
                f'{subpath}: "{part}" not allowed in paths'
            )

        path = os.path.join(path, part)
        if platform_utils.islink(path):
            raise ManifestInvalidPathError(
                f"{path}: traversing symlinks not allow"
            )

        if os.path.exists(path):
            if not os.path.isfile(path) and not platform_utils.isdir(path):
                raise ManifestInvalidPathError(
                    f"{path}: only regular files & directories allowed"
                )

    if skipfinal:
        path = os.path.join(path, finalpart)

    return path


class _CopyFile(NamedTuple):
    """Container for <copyfile> manifest element."""

    # Absolute path to the git project checkout.
    git_worktree: str
    # Relative path under |git_worktree| of file to read.
    src: str
    # Absolute path to the top of the repo client checkout.
    topdir: str
    # Relative path under |topdir| of file to write.
    dest: str

    def _Copy(self):
        src = _SafeExpandPath(self.git_worktree, self.src)
        dest = _SafeExpandPath(self.topdir, self.dest)

        if platform_utils.isdir(src):
            raise ManifestInvalidPathError(
                f"{self.src}: copying from directory not supported"
            )
        if platform_utils.isdir(dest):
            raise ManifestInvalidPathError(
                f"{self.dest}: copying to directory not allowed"
            )

        # Copy file if it does not exist or is out of date.
        if not os.path.exists(dest) or not filecmp.cmp(src, dest):
            try:
                # Remove existing file first, since it might be read-only.
                if os.path.exists(dest):
                    platform_utils.remove(dest)
                else:
                    dest_dir = os.path.dirname(dest)
                    if not platform_utils.isdir(dest_dir):
                        os.makedirs(dest_dir)
                shutil.copy(src, dest)
                # Make the file read-only.
                mode = os.stat(dest)[stat.ST_MODE]
                mode = mode & ~(stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
                os.chmod(dest, mode)
            except OSError:
                logger.error("error: Cannot copy file %s to %s", src, dest)


class _LinkFile(NamedTuple):
    """Container for <linkfile> manifest element."""

    # Absolute path to the git project checkout.
    git_worktree: str
    # Target of symlink relative to path under |git_worktree|.
    src: str
    # Absolute path to the top of the repo client checkout.
    topdir: str
    # Relative path under |topdir| of symlink to create.
    dest: str

    def __linkIt(self, relSrc, absDest):
        # Link file if it does not exist or is out of date.
        if not platform_utils.islink(absDest) or (
            platform_utils.readlink(absDest) != relSrc
        ):
            try:
                # Remove existing path first, since it might be read-only.
                if os.path.lexists(absDest):
                    # removedirs handles symlinks, empty dirs, and nested
                    # trees of stale linkfile dests without deleting user
                    # content.  Falls through to symlink() if the path was
                    # fully removed, or raises OSError if not.
                    platform_utils.removedirs(absDest)
                else:
                    dest_dir = os.path.dirname(absDest)
                    if not platform_utils.isdir(dest_dir):
                        os.makedirs(dest_dir)
                platform_utils.symlink(relSrc, absDest)
            except OSError:
                logger.error("error: Cannot symlink %s to %s", absDest, relSrc)

    def _Link(self):
        """Link the self.src & self.dest paths.

        Handles wild cards on the src linking all of the files in the source in
        to the destination directory.
        """
        # Some people use src="." to create stable links to projects.  Let's
        # allow that but reject all other uses of "." to keep things simple.
        if self.src == ".":
            src = self.git_worktree
        else:
            src = _SafeExpandPath(self.git_worktree, self.src)

        if not glob.has_magic(src):
            # Entity does not contain a wild card so just a simple one to one
            # link operation.
            dest = _SafeExpandPath(self.topdir, self.dest, skipfinal=True)
            # dest & src are absolute paths at this point.  Make sure the target
            # of the symlink is relative in the context of the repo client
            # checkout.
            relpath = os.path.relpath(src, os.path.dirname(dest))
            self.__linkIt(relpath, dest)
        else:
            dest = _SafeExpandPath(self.topdir, self.dest)
            # Entity contains a wild card.
            if os.path.exists(dest) and not platform_utils.isdir(dest):
                logger.error(
                    "Link error: src with wildcard, %s must be a directory",
                    dest,
                )
            else:
                for absSrcFile in glob.glob(src):
                    # Create a releative path from source dir to destination
                    # dir.
                    absSrcDir = os.path.dirname(absSrcFile)
                    relSrcDir = os.path.relpath(absSrcDir, dest)

                    # Get the source file name.
                    srcFile = os.path.basename(absSrcFile)

                    # Now form the final full paths to srcFile. They will be
                    # absolute for the desintaiton and relative for the source.
                    absDest = os.path.join(dest, srcFile)
                    relSrc = os.path.join(relSrcDir, srcFile)
                    self.__linkIt(relSrc, absDest)


class RemoteSpec:
    def __init__(
        self,
        name,
        url=None,
        pushUrl=None,
        review=None,
        revision=None,
        orig_name=None,
        fetchUrl=None,
    ):
        self.name = name
        self.url = url
        self.pushUrl = pushUrl
        self.review = review
        self.revision = revision
        self.orig_name = orig_name
        self.fetchUrl = fetchUrl


class Project:
    # These objects can be shared between several working trees.
    @property
    def shareable_dirs(self):
        """Return the shareable directories"""
        if self.UseAlternates:
            return ["hooks", "rr-cache"]
        else:
            return ["hooks", "objects", "rr-cache"]

    def __init__(
        self,
        manifest,
        name,
        remote,
        gitdir,
        objdir,
        worktree,
        relpath,
        revisionExpr,
        revisionId,
        rebase=True,
        groups=None,
        sync_c=False,
        sync_s=False,
        sync_tags=True,
        clone_depth=None,
        sync_strategy=None,
        upstream=None,
        parent=None,
        use_git_worktrees=False,
        is_derived=False,
        gitlink_path: Optional[str] = None,
        dest_branch=None,
        optimized_fetch=False,
        retry_fetches=0,
    ):
        """Init a Project object.

        Args:
            manifest: The XmlManifest object.
            name: The `name` attribute of manifest.xml's project element.
            remote: RemoteSpec object specifying its remote's properties.
            gitdir: Absolute path of git directory.
            objdir: Absolute path of directory to store git objects.
            worktree: Absolute path of git working tree.
            relpath: Relative path of git working tree to repo's top directory.
            revisionExpr: The `revision` attribute of manifest.xml's project
                element.
            revisionId: git commit id for checking out.
            rebase: The `rebase` attribute of manifest.xml's project element.
            groups: The `groups` attribute of manifest.xml's project element.
            sync_c: The `sync-c` attribute of manifest.xml's project element.
            sync_s: The `sync-s` attribute of manifest.xml's project element.
            sync_tags: The `sync-tags` attribute of manifest.xml's project
                element.
            upstream: The `upstream` attribute of manifest.xml's project
                element.
            parent: The parent Project object.
            use_git_worktrees: Whether to use `git worktree` for this project.
            is_derived: False if the project was explicitly defined in the
                manifest; True if the project is a discovered submodule.
            gitlink_path: For a discovered submodule, its path inside the
                parent project.
            dest_branch: The branch to which to push changes for review by
                default.
            optimized_fetch: If True, when a project is set to a sha1 revision,
                only fetch from the remote if the sha1 is not present locally.
            retry_fetches: Retry remote fetches n times upon receiving transient
                error with exponential backoff and jitter.
        """
        self.client = self.manifest = manifest
        self.name = name
        self.remote = remote
        self.UpdatePaths(relpath, worktree, gitdir, objdir)
        self.SetRevision(revisionExpr, revisionId=revisionId)

        self.rebase = rebase
        self.groups = groups if groups is not None else set()
        self.sync_c = sync_c
        self.sync_s = sync_s
        self.sync_tags = sync_tags
        self.clone_depth = clone_depth
        self.sync_strategy = sync_strategy
        self.upstream = upstream
        self.parent = parent
        # NB: Do not use this setting in __init__ to change behavior so that the
        # manifest.git checkout can inspect & change it after instantiating.
        # See the XmlManifest init code for more info.
        self.use_git_worktrees = use_git_worktrees
        self.is_derived = is_derived
        self.gitlink_path = gitlink_path
        self.optimized_fetch = optimized_fetch
        self.retry_fetches = max(0, retry_fetches)
        self.subprojects = []

        self.snapshots = {}
        # Use dicts to dedupe while maintaining declared order.
        self.copyfiles = {}
        self.linkfiles = {}
        self.annotations = []
        self.dest_branch = dest_branch
        self.stateless_prune_needed = False

        # This will be filled in if a project is later identified to be the
        # project containing repo hooks.
        self.enabled_repo_hooks = []

    def RelPath(self, local=True):
        """Return the path for the project relative to a manifest.

        Args:
            local: a boolean, if True, the path is relative to the local
                (sub)manifest.  If false, the path is relative to the outermost
                manifest.
        """
        if local:
            return self.relpath
        return os.path.join(self.manifest.path_prefix, self.relpath)

    def GetEnvVars(self, local: bool = True) -> Dict[str, str]:
        """Get project-context environment variables.

        Args:
            local: If True, REPO_PATH is relative to the local (sub)manifest.
                If False, it is relative to the outermost manifest.

        Returns:
            A dictionary mapping environment variable names to their values.

        Environment Variables:
            See the Environment section in `repo help forall` or
            `subcmds/forall.py` for details on the available variables.
            Note that `forall.py` also documents some extra variables that are
            specific to how the `repo forall` command iterates over projects
            (e.g., `REPO_COUNT` and `REPO_I`).
        """
        env = {}

        def setenv(name, val):
            if val is None:
                val = ""
            env[name] = val

        setenv("REPO_PROJECT", self.name)
        setenv("REPO_OUTERPATH", self.manifest.path_prefix)
        setenv("REPO_INNERPATH", self.relpath)
        setenv("REPO_PATH", self.RelPath(local=local))
        setenv("REPO_REMOTE", self.remote.name)
        setenv("REPO_PROJECT_FETCH_URL", self.remote.url)

        try:
            lrev = "" if self.manifest.IsMirror else self.GetRevisionId()
        except ManifestInvalidRevisionError:
            lrev = ""
        setenv("REPO_LREV", lrev)
        setenv("REPO_RREV", self.revisionExpr)
        setenv("REPO_UPSTREAM", self.upstream)
        setenv("REPO_DEST_BRANCH", self.dest_branch)

        for annotation in self.annotations:
            setenv(f"REPO__{annotation.name}", annotation.value)

        return env

    def SetRevision(self, revisionExpr, revisionId=None):
        """Set revisionId based on revision expression and id"""
        self.revisionExpr = revisionExpr
        if revisionId is None and revisionExpr and IsId(revisionExpr):
            self.revisionId = self.revisionExpr
        else:
            self.revisionId = revisionId

    def UpdatePaths(self, relpath, worktree, gitdir, objdir):
        """Update paths used by this project"""
        self.gitdir = gitdir.replace("\\", "/")
        self.objdir = objdir.replace("\\", "/")
        if worktree:
            self.worktree = os.path.normpath(worktree).replace("\\", "/")
        else:
            self.worktree = None
        self.relpath = relpath

        self.config = GitConfig.ForRepository(
            gitdir=self.gitdir, defaults=self.manifest.globalConfig
        )

        if self.worktree:
            self.work_git = self._GitGetByExec(
                self, bare=False, gitdir=self.gitdir
            )
        else:
            self.work_git = None
        self.bare_git = self._GitGetByExec(self, bare=True, gitdir=self.gitdir)
        self.bare_ref = GitRefs(self.gitdir)
        self.bare_objdir = self._GitGetByExec(
            self, bare=True, gitdir=self.objdir
        )

    @property
    def UseAlternates(self):
        """Whether git alternates are in use.

        This will be removed once migration to alternates is complete.
        """
        return _ALTERNATES or self.manifest.is_multimanifest

    @property
    def Derived(self):
        return self.is_derived

    @property
    def Exists(self):
        return platform_utils.isdir(self.gitdir) and platform_utils.isdir(
            self.objdir
        )

    @property
    def CurrentBranch(self):
        """Obtain the name of the currently checked out branch.

        The branch name omits the 'refs/heads/' prefix.
        None is returned if the project is on a detached HEAD, or if the
        work_git is otheriwse inaccessible (e.g. an incomplete sync).
        """
        try:
            b = self._GetHead()
        except NoManifestException:
            # If the local checkout is in a bad state, don't barf.  Let the
            # callers process this like the head is unreadable.
            return None
        if b and b.startswith(R_HEADS):
            return b[len(R_HEADS) :]
        return None

    def _GetHead(self) -> Optional[str]:
        """Return worktree HEAD, reusing a compatible loaded ref snapshot."""
        if not self.work_git:
            return None
        # Git worktrees keep the checkout's HEAD in the worktree admin dir,
        # while bare_ref reads the shared repository.  Its HEAD is not the
        # checked-out worktree's HEAD and must not be reused here.
        if not self.use_git_worktrees and self.bare_ref.is_loaded:
            head = self.bare_ref.head
            if head:
                return head
        return self.work_git.GetHead()

    def IsRebaseInProgress(self):
        """Returns true if a rebase or "am" is in progress"""
        # "rebase-apply" is used for "git rebase".
        # "rebase-merge" is used for "git am".
        return (
            os.path.exists(self.work_git.GetDotgitPath("rebase-apply"))
            or os.path.exists(self.work_git.GetDotgitPath("rebase-merge"))
            or os.path.exists(os.path.join(self.worktree, ".dotest"))
        )

    def IsCherryPickInProgress(self):
        """Returns True if a cherry-pick is in progress."""
        return os.path.exists(self.work_git.GetDotgitPath("CHERRY_PICK_HEAD"))

    def _AbortRebase(self):
        """Abort ongoing rebase, cherry-pick or patch apply (am).

        If no rebase, cherry-pick or patch apply was in progress, this method
        ignores the status and continues.
        """

        def _git(*args):
            # Ignore return code, in case there was no rebase in progress.
            GitCommand(self, args, log_as_error=False).Wait()

        _git("cherry-pick", "--abort")
        _git("rebase", "--abort")
        _git("am", "--abort")

    def _RefreshIndexStatCache(self) -> None:
        """Refresh the index's cached stat information."""
        args = ["--unmerged", "--ignore-missing", "--refresh"]

        # Run twice because -q is needed and unhelpful in equal measure. It
        # keeps git quiet about uncommitted changes, which would otherwise
        # exit 1 and report a failure to telemetry for every modified project,
        # but it also suppresses the reason a refresh failed, leaving a bare
        # exit 128. So refresh with it, and repeat without it only to get
        # the reason.
        try:
            self.work_git.update_index("-q", *args, log_as_error=False)
            return
        except GitError:
            pass

        # Exit code 1 means there are modified files, which is okay.
        try:
            self.work_git.update_index(*args)
        except GitCommandError as e:
            if e.git_rc != 1:
                raise

    def IsDirty(self, consider_untracked=True):
        """Is the working directory modified in some way?"""
        self._RefreshIndexStatCache()
        if self.work_git.DiffZ("diff-index", "-M", "--cached", HEAD):
            return True
        if self.work_git.DiffZ("diff-files"):
            return True
        if consider_untracked and self.UntrackedFiles():
            return True
        return False

    def HasStash(self) -> bool:
        """Returns True if there is a stash in the repository."""
        p = GitCommand(
            self,
            ["rev-parse", "--verify", "refs/stash"],
            bare=True,
            capture_stdout=True,
            capture_stderr=True,
            log_as_error=False,
        )
        return p.Wait() == 0

    _userident_name = None
    _userident_email = None

    @property
    def UserName(self):
        """Obtain the user's personal name."""
        if self._userident_name is None:
            self._LoadUserIdentity()
        return self._userident_name

    @property
    def UserEmail(self):
        """Obtain the user's email address.  This is very likely
        to be their Gerrit login.
        """
        if self._userident_email is None:
            self._LoadUserIdentity()
        return self._userident_email

    def _LoadUserIdentity(self):
        u = self.bare_git.var("GIT_COMMITTER_IDENT")
        m = re.compile("^(.*) <([^>]*)> ").match(u)
        if m:
            self._userident_name = m.group(1)
            self._userident_email = m.group(2)
        else:
            self._userident_name = ""
            self._userident_email = ""

    def GetRemote(self, name=None):
        """Get the configuration for a single remote.

        Defaults to the current project's remote.
        """
        if name is None:
            name = self.remote.name
        return self.config.GetRemote(name)

    def GetBranch(self, name):
        """Get the configuration for a single branch."""
        return self.config.GetBranch(name)

    def GetBranches(self):
        """Get all existing local branches."""
        all_refs = self._allrefs
        current = self.CurrentBranch
        heads = {}

        for name, ref_id in all_refs.items():
            if name.startswith(R_HEADS):
                name = name[len(R_HEADS) :]
                b = self.GetBranch(name)
                b.current = name == current
                b.published = None
                b.revision = ref_id
                heads[name] = b

        for name, ref_id in all_refs.items():
            if name.startswith(R_PUB):
                name = name[len(R_PUB) :]
                b = heads.get(name)
                if b:
                    b.published = ref_id

        return heads

    def MatchesGroups(self, manifest_groups):
        """Returns true if the manifest groups specified at init should cause
        this project to be synced.
        Prefixing a manifest group with "-" inverts the meaning of a group.
        All projects are implicitly labelled with "all".

        labels are resolved in order.  In the example case of
        project_groups: "all,group1,group2"
        manifest_groups: "-group1,group2"
        the project will be matched.

        The special manifest group "default" will match any project that
        does not have the special project group "notdefault"
        """
        default_groups = self.manifest.default_groups or ["default"]
        expanded_manifest_groups = manifest_groups or default_groups
        expanded_project_groups = {"all"} | self.groups
        if "notdefault" not in expanded_project_groups:
            expanded_project_groups |= {"default"}

        matched = False
        for group in expanded_manifest_groups:
            if group.startswith("-") and group[1:] in expanded_project_groups:
                matched = False
            elif group in expanded_project_groups:
                matched = True

        return matched

    def UncommitedFiles(self, get_all=True):
        """Returns a list of strings, uncommitted files in the git tree.

        Args:
            get_all: a boolean, if True - get information about all different
                uncommitted files. If False - return as soon as any kind of
                uncommitted files is detected.
        """
        details = []
        self._RefreshIndexStatCache()
        if self.IsRebaseInProgress():
            details.append("rebase in progress")
            if not get_all:
                return details

        changes = self.work_git.DiffZ("diff-index", "--cached", HEAD).keys()
        if changes:
            details.extend(changes)
            if not get_all:
                return details

        changes = self.work_git.DiffZ("diff-files").keys()
        if changes:
            details.extend(changes)
            if not get_all:
                return details

        changes = self.UntrackedFiles()
        if changes:
            details.extend(changes)

        return details

    def UntrackedFiles(self):
        """Returns a list of strings, untracked files in the git tree."""
        return self.work_git.LsOthers()

    def HasChanges(self):
        """Returns true if there are uncommitted changes."""
        return bool(self.UncommitedFiles(get_all=False))

    def PrintWorkTreeStatus(self, output_redir=None, quiet=False, local=False):
        """Prints the status of the repository to stdout.

        Args:
            output_redir: If specified, redirect the output to this object.
            quiet:  If True then only print the project name.  Do not print
                the modified files, branch name, etc.
            local: a boolean, if True, the path is relative to the local
                (sub)manifest.  If false, the path is relative to the outermost
                manifest.
        """
        if not platform_utils.isdir(self.worktree):
            if output_redir is None:
                output_redir = sys.stdout
            print(file=output_redir)
            print("project %s/" % self.RelPath(local), file=output_redir)
            print('  missing (run "repo sync")', file=output_redir)
            return

        self._RefreshIndexStatCache()
        rb = self.IsRebaseInProgress()
        di = self.work_git.DiffZ("diff-index", "-M", "--cached", HEAD)
        df = self.work_git.DiffZ("diff-files")
        do = self.work_git.LsOthers()
        if not rb and not di and not df and not do and not self.CurrentBranch:
            return "CLEAN"

        out = StatusColoring(self.config)
        if output_redir is not None:
            out.redirect(output_redir)
        out.project("project %-40s", self.RelPath(local) + "/ ")

        if quiet:
            out.nl()
            return "DIRTY"

        branch_name = self.CurrentBranch
        if branch_name is None:
            out.nobranch("(*** NO BRANCH ***)")
        else:
            branch_obj = self.GetBranch(branch_name)
            ahead_behind = ""
            try:
                local_merge = branch_obj.LocalMerge
                if local_merge:
                    left_right = self.work_git.rev_list(
                        "--left-right",
                        "--count",
                        f"{local_merge}...{R_HEADS}{branch_name}",
                    )
                    left, right = left_right[0].split()
                    behind = int(left)
                    ahead = int(right)
                    if ahead and behind:
                        ahead_behind = f" [ahead {ahead}, behind {behind}]"
                    elif ahead:
                        ahead_behind = f" [ahead {ahead}]"
                    elif behind:
                        ahead_behind = f" [behind {behind}]"
            except GitError:
                pass
            out.branch("branch %s%s", branch_name, ahead_behind)
        out.nl()

        if rb:
            out.important("prior sync failed; rebase still in progress")
            out.nl()

        paths = []
        paths.extend(di.keys())
        paths.extend(df.keys())
        paths.extend(do)

        for p in sorted(set(paths)):
            try:
                i = di[p]
            except KeyError:
                i = None

            try:
                f = df[p]
            except KeyError:
                f = None

            if i:
                i_status = i.status.upper()
            else:
                i_status = "-"

            if f:
                f_status = f.status.lower()
            else:
                f_status = "-"

            if i and i.src_path:
                line = (
                    f" {i_status}{f_status}\t{i.src_path} => {p} ({i.level}%)"
                )
            else:
                line = f" {i_status}{f_status}\t{p}"

            if i and not f:
                out.added("%s", line)
            elif (i and f) or (not i and f):
                out.changed("%s", line)
            elif not i and not f:
                out.untracked("%s", line)
            else:
                out.write("%s", line)
            out.nl()

        return "DIRTY"

    def PrintWorkTreeDiff(
        self, absolute_paths=False, output_redir=None, local=False
    ):
        """Prints the status of the repository to stdout."""
        out = DiffColoring(self.config)
        if output_redir:
            out.redirect(output_redir)
        cmd = ["diff"]
        if out.is_on:
            cmd.append("--color")
        cmd.append(HEAD)
        if absolute_paths:
            cmd.append("--src-prefix=a/%s/" % self.RelPath(local))
            cmd.append("--dst-prefix=b/%s/" % self.RelPath(local))
        cmd.append("--")
        try:
            p = GitCommand(self, cmd, capture_stdout=True, capture_stderr=True)
            p.Wait()
        except GitError as e:
            out.nl()
            out.project("project %s/" % self.RelPath(local))
            out.nl()
            out.fail("%s", str(e))
            out.nl()
            return False
        if p.stdout:
            out.nl()
            out.project("project %s/" % self.RelPath(local))
            out.nl()
            out.write("%s", p.stdout)
        return p.Wait() == 0

    def WasPublished(self, branch, all_refs=None):
        """Was the branch published (uploaded) for code review?
        If so, returns the SHA-1 hash of the last published
        state for the branch.
        """
        key = R_PUB + branch
        if all_refs is None:
            try:
                return self.bare_git.rev_parse(key)
            except GitError:
                return None
        else:
            try:
                return all_refs[key]
            except KeyError:
                return None

    def CleanPublishedCache(self, all_refs=None):
        """Prunes any stale published refs."""
        if all_refs is None:
            all_refs = self._allrefs
        heads = set()
        canrm = {}
        for name, ref_id in all_refs.items():
            if name.startswith(R_HEADS):
                heads.add(name)
            elif name.startswith(R_PUB):
                canrm[name] = ref_id

        for name, ref_id in canrm.items():
            n = name[len(R_PUB) :]
            if R_HEADS + n not in heads:
                self.bare_git.DeleteRef(name, ref_id)

    def GetUploadableBranches(self, selected_branch=None):
        """List any branches which can be uploaded for review."""
        if selected_branch:
            branch = self.GetBranch(selected_branch)
            if not branch.LocalMerge:
                return []
            head_id = self.bare_ref.get(R_HEADS + selected_branch)
            if not head_id:
                return []
            pub_id = self.bare_ref.get(R_PUB + selected_branch)
            if pub_id and pub_id == head_id:
                return []
            rb = self.GetUploadableBranch(selected_branch)
            if rb:
                rb.branch.current = selected_branch == self.CurrentBranch
                return [rb]
            return []

        # Optimization: Skip scanning _allrefs (which spawns git processes)
        # if no local branches with upstream tracking exist in .git/config.
        if not any(self.config.GetSubSections("branch")):
            return []

        branches = self.GetBranches()

        ready = []
        for branch, branch_config in branches.items():
            if branch_config.published == branch_config.revision:
                continue

            rb = self.GetUploadableBranch(branch)
            if rb:
                rb.branch.current = branch_config.current
                ready.append(rb)
        return ready

    def GetUploadableBranch(self, branch_name):
        """Get a single uploadable branch, or None."""
        branch = self.GetBranch(branch_name)
        base = branch.LocalMerge
        if branch.LocalMerge:
            rb = ReviewableBranch(self, branch, base)
            if rb.commits:
                return rb
        return None

    def UploadForReview(
        self,
        branch=None,
        people=([], []),
        dryrun=False,
        topic=None,
        hashtags=(),
        labels=(),
        private=False,
        notify=None,
        wip=False,
        ready=False,
        dest_branch=None,
        validate_certs=True,
        push_options=None,
        patchset_description=None,
        git_event_log: Optional[EventLog] = None,
    ) -> None:
        """Uploads the named branch for code review."""
        if branch is None:
            branch = self.CurrentBranch
        if branch is None:
            raise GitError("not currently on a branch", project=self.name)

        branch = self.GetBranch(branch)
        if not branch.LocalMerge:
            raise GitError(
                "branch %s does not track a remote" % branch.name,
                project=self.name,
            )
        if not branch.remote.review:
            raise GitError(
                "remote %s has no review url" % branch.remote.name,
                project=self.name,
            )

        # Basic validity check on label syntax.
        for label in labels:
            if not re.match(r"^.+[+-][0-9]+$", label):
                raise UploadError(
                    f'invalid label syntax "{label}": labels use forms like '
                    "CodeReview+1 or Verified-1",
                    project=self.name,
                )

        if dest_branch is None:
            dest_branch = self.dest_branch
        if dest_branch is None:
            dest_branch = branch.merge
        if not dest_branch.startswith(R_HEADS):
            dest_branch = R_HEADS + dest_branch

        if not branch.remote.projectname:
            branch.remote.projectname = self.name
            branch.remote.Save()

        url = branch.remote.ReviewUrl(self.UserEmail, validate_certs)
        if url is None:
            raise UploadError("review not configured", project=self.name)
        cmd = ["push", "--progress"]
        if dryrun:
            cmd.append("-n")

        if url.startswith("ssh://"):
            cmd.append("--receive-pack=gerrit receive-pack")

        # This stops git from pushing all reachable annotated tags when
        # push.followTags is configured. Gerrit does not accept any tags
        # pushed to a CL.
        cmd.append("--no-follow-tags")

        for push_option in push_options or []:
            cmd.append("-o")
            cmd.append(push_option)

        cmd.append(url)

        if dest_branch.startswith(R_HEADS):
            dest_branch = dest_branch[len(R_HEADS) :]

        ref_spec = f"{R_HEADS + branch.name}:refs/for/{dest_branch}"
        opts = []
        if topic is not None:
            opts += [f"topic={topic}"]
        opts += ["t=%s" % p for p in hashtags]
        # NB: No need to encode labels as they've been validated above.
        opts += ["l=%s" % p for p in labels]

        opts += ["r=%s" % p for p in people[0]]
        opts += ["cc=%s" % p for p in people[1]]
        if notify:
            opts += ["notify=" + notify]
        if private:
            opts += ["private"]
        if wip:
            opts += ["wip"]
        if ready:
            opts += ["ready"]
        if patchset_description:
            opts += [
                f"m={self._encode_patchset_description(patchset_description)}"
            ]
        if opts:
            ref_spec = ref_spec + "%" + ",".join(opts)
        cmd.append(ref_spec)

        push_cmd = GitCommand(
            self,
            cmd,
            bare=True,
            verify_command=True,
        )
        push_cmd.Wait()

        cls_urls = self._FindGerritUrls(push_cmd.stderr)

        try:
            rb = ReviewableBranch(self, branch, branch.LocalMerge)
            modified_files_list = rb.modified_files
            if git_event_log:
                git_event_log.LogDataConfigEvents(
                    {
                        "cls": ",".join(cls_urls),
                        "remote": branch.remote.name,
                        "branch": branch.name,
                        "files": ",".join(modified_files_list),
                    },
                    "repo.uploadstate",
                )
        except Exception as e:
            logger.error("Tracing failed: %s", str(e))
        if not dryrun:
            msg = f"posted to {branch.remote.review} for {dest_branch}"
            self.bare_git.UpdateRef(
                R_PUB + branch.name, R_HEADS + branch.name, message=msg
            )

    @staticmethod
    def _FindGerritUrls(stderr: Optional[str]) -> List[str]:
        """Extracts Gerrit review URLs from git push output."""
        if not stderr:
            return []
        return [
            match.group(1)
            for match in re.finditer(r"(https?://[^/]+/c/.+?/\+/\d+)", stderr)
        ]

    @staticmethod
    def _encode_patchset_description(original):
        """Applies percent-encoding for strings sent as patchset description.

        The encoding used is based on but stricter than URL encoding (Section
        2.1 of RFC 3986). The only non-escaped characters are alphanumerics, and
        'SPACE' (U+0020) can be represented as 'LOW LINE' (U+005F) or
        'PLUS SIGN' (U+002B).

        For more information, see the Gerrit docs here:
        https://gerrit-review.googlesource.com/Documentation/user-upload.html#patch_set_description
        """
        SAFE = {ord(x) for x in string.ascii_letters + string.digits}

        def _enc(b):
            if b in SAFE:
                return chr(b)
            elif b == ord(" "):
                return "_"
            else:
                return f"%{b:02x}"

        return "".join(_enc(x) for x in original.encode("utf-8"))

    def _ExtractArchive(self, tarpath, path=None):
        """Extract the given tar on its current location

        Args:
            tarpath: The path to the actual tar file

        """
        try:
            with tarfile.open(tarpath, "r") as tar:
                tar.extractall(path=path)
                return True
        except (OSError, tarfile.TarError) as e:
            logger.error("error: Cannot extract archive %s: %s", tarpath, e)
        return False

    def _ShouldStatelessPrune(
        self, use_superproject: Optional[bool] = None
    ) -> bool:
        """Determines if a stateless prune should be performed.

        Stateless pruning reclaims space by running a reflog expiration and
        garbage collection instead of an incremental fetch. It is only performed
        if the repository is clean and has no local-only state.
        """
        if not self.Exists:
            return False

        if self._CheckForImmutableRevision(use_superproject=use_superproject):
            return False

        # Query the target hash from remote to see if we are up-to-date.
        target_hash = None
        if IsId(self.revisionExpr):
            target_hash = self.revisionExpr
        else:
            output = self._LsRemote(self.upstream or self.revisionExpr)
            if output:
                target_hash = output.splitlines()[0].split()[0]

        if not target_hash:
            return False

        try:
            local_head = self.bare_git.rev_parse("HEAD")
        except GitError:
            local_head = None

        if target_hash == local_head:
            return False

        # Skip if sharing objects with other projects.
        shares_objdir = self.UseAlternates or self.use_git_worktrees
        if not shares_objdir:
            for p in self.manifest.GetProjectsWithName(self.name):
                if p != self and p.objdir == self.objdir:
                    shares_objdir = True
                    break

        if shares_objdir:
            return False

        # Skip if HEAD contains any unpushed local commits.
        try:
            local_commits = self.bare_git.rev_list(
                "--count", "HEAD", "--not", "--remotes", "--tags"
            )
            if int(local_commits[0]) > 0:
                return False
        except (GitError, IndexError, ValueError):
            return False

        if self.IsDirty(consider_untracked=True) or self.HasStash():
            return False

        return True

    def Sync_NetworkHalf(
        self,
        quiet=False,
        verbose=False,
        output_redir=None,
        is_new=None,
        use_superproject=None,
        current_branch_only=None,
        force_sync=False,
        clone_bundle=True,
        tags=None,
        archive=False,
        optimized_fetch=False,
        retry_fetches=0,
        prune=False,
        submodules=False,
        ssh_proxy=None,
        clone_filter=None,
        partial_clone_exclude=None,
        clone_filter_for_depth=None,
    ):
        """Perform only the network IO portion of the sync process.
        Local working directory/branch state is not affected.
        """
        if archive and not isinstance(self, MetaProject):
            if self.remote.url.startswith(("http://", "https://")):
                msg_template = (
                    "%s: Cannot fetch archives from http/https remotes."
                )
                msg_args = self.name
                msg = msg_template % msg_args
                logger.error(msg_template, msg_args)
                return SyncNetworkHalfResult(
                    False, SyncNetworkHalfError(msg, project=self.name)
                )

            name = self.relpath.replace("\\", "/")
            name = name.replace("/", "_")
            tarpath = "%s.tar" % name
            topdir = self.manifest.topdir

            try:
                self._FetchArchive(tarpath, cwd=topdir)
            except GitError as e:
                logger.error("error: %s", e)
                return SyncNetworkHalfResult(False, e)

            # From now on, we only need absolute tarpath.
            tarpath = os.path.join(topdir, tarpath)

            if not self._ExtractArchive(tarpath, path=topdir):
                return SyncNetworkHalfResult(
                    True,
                    SyncNetworkHalfError(
                        f"Unable to Extract Archive {tarpath}",
                        project=self.name,
                    ),
                )
            try:
                platform_utils.remove(tarpath)
            except OSError as e:
                logger.warning("warn: Cannot remove archive %s: %s", tarpath, e)
            self._CopyAndLinkFiles()
            return SyncNetworkHalfResult(True)

        # If the shared object dir already exists, don't try to rebootstrap with
        # a clone bundle download.  We should have the majority of objects
        # already.
        if clone_bundle and os.path.exists(self.objdir):
            clone_bundle = False

        if partial_clone_exclude is None:
            partial_clone_exclude = set()
        if self.name in partial_clone_exclude:
            clone_bundle = True
            clone_filter = None

        if self.sync_strategy == "stateless" and self._ShouldStatelessPrune(
            use_superproject
        ):
            self.stateless_prune_needed = True

        if is_new is None:
            is_new = not self.Exists
        if is_new:
            self._InitGitDir(force_sync=force_sync, quiet=quiet)
        else:
            try:
                # At this point, it's possible that gitdir points to an old
                # objdir (e.g. name changed, but objdir exists). Check
                # references to ensure that's not the case. See
                # https://issues.gerritcodereview.com/40013418 for more
                # details.
                self._CheckDirReference(self.objdir, self.gitdir)

                self._UpdateHooks(quiet=quiet)
            except GitError as e:
                if not force_sync:
                    raise e
                # Let _InitGitDir fix the issue, force_sync is always True here.
                self._InitGitDir(force_sync=True, quiet=quiet)
        self._InitRemote()

        if self.UseAlternates:
            # If gitdir/objects is a symlink, migrate it from the old layout.
            gitdir_objects = os.path.join(self.gitdir, "objects")
            if platform_utils.islink(gitdir_objects):
                platform_utils.remove(gitdir_objects, missing_ok=True)
            gitdir_alt = os.path.join(self.gitdir, "objects/info/alternates")
            if not os.path.exists(gitdir_alt):
                os.makedirs(os.path.dirname(gitdir_alt), exist_ok=True)
                _lwrite(
                    gitdir_alt,
                    os.path.join(
                        os.path.relpath(self.objdir, gitdir_objects), "objects"
                    )
                    + "\n",
                )

        if is_new:
            alt = os.path.join(self.objdir, "objects/info/alternates")
            try:
                with open(alt) as fd:
                    # This works for both absolute and relative alternate
                    # directories.
                    alt_dir = os.path.join(
                        self.objdir, "objects", fd.readline().rstrip()
                    )
            except OSError:
                alt_dir = None
        else:
            alt_dir = None

        if (
            clone_bundle
            and alt_dir is None
            and self._ApplyCloneBundle(
                initial=is_new, quiet=quiet, verbose=verbose
            )
        ):
            is_new = False

        if current_branch_only is None:
            if self.sync_c:
                current_branch_only = True
            elif not self.manifest._loaded:
                # Manifest cannot check defaults until it syncs.
                current_branch_only = False
            elif self.manifest.default.sync_c:
                current_branch_only = True

        if tags is None:
            tags = self.sync_tags

        if self.clone_depth:
            depth = self.clone_depth
        else:
            depth = self.manifest.manifestProject.depth

        # If the project has been manually unshallowed (e.g. via
        # `git fetch --unshallow`), don't re-shallow it during sync.
        if depth and not is_new and not self._HasShallow():
            depth = None

        if depth and clone_filter_for_depth:
            depth = None
            clone_filter = clone_filter_for_depth

        custom_fetch_configured = (
            self.manifest.manifestProject.use_local_gitdirs
            and self.manifest.manifestProject.fetch_cmd
            and not isinstance(self, MetaProject)
        )

        remote_fetched = False
        if custom_fetch_configured:
            # The custom fetch command is executed unconditionally on every sync
            # because it is responsible for updating tracking refs and
            # FETCH_HEAD.
            remote_fetched = True
            try:
                if not self._CustomFetch(verbose=verbose):
                    return SyncNetworkHalfResult(
                        remote_fetched,
                        SyncNetworkHalfError(
                            f"Unable to fetch project {self.name} using "
                            "fetchcmd",
                            project=self.name,
                        ),
                    )
            except RepoError as e:
                return SyncNetworkHalfResult(
                    remote_fetched,
                    e,
                )
        else:
            # See if we can skip the standard network fetch entirely.
            has_shallow = self._HasShallow()
            skip_fetch = (
                optimized_fetch
                and IsId(self.revisionExpr)
                and self._CheckForImmutableRevision(
                    use_superproject=use_superproject,
                    depth=depth,
                )
                and (has_shallow or not self._IsShallow(depth))
            )

            if not skip_fetch:
                remote_fetched = True
                try:
                    if not self._RemoteFetch(
                        initial=is_new,
                        quiet=quiet,
                        verbose=verbose,
                        output_redir=output_redir,
                        alt_dir=alt_dir,
                        use_superproject=use_superproject,
                        current_branch_only=current_branch_only,
                        tags=tags,
                        prune=prune,
                        depth=depth,
                        submodules=submodules,
                        force_sync=force_sync,
                        ssh_proxy=ssh_proxy,
                        clone_filter=clone_filter,
                        retry_fetches=retry_fetches,
                    ):
                        return SyncNetworkHalfResult(
                            remote_fetched,
                            SyncNetworkHalfError(
                                f"Unable to remote fetch project {self.name}",
                                project=self.name,
                            ),
                        )
                except RepoError as e:
                    return SyncNetworkHalfResult(
                        remote_fetched,
                        e,
                    )

        mp = self.manifest.manifestProject
        dissociate = mp.dissociate
        if dissociate:
            alternates_file = os.path.join(
                self.objdir, "objects/info/alternates"
            )
            if os.path.exists(alternates_file):
                cmd = ["repack", "-a", "-d"]
                p = GitCommand(
                    self,
                    cmd,
                    bare=True,
                    capture_stdout=bool(output_redir),
                    merge_output=bool(output_redir),
                )
                if p.stdout and output_redir:
                    output_redir.write(p.stdout)
                if p.Wait() != 0:
                    return SyncNetworkHalfResult(
                        remote_fetched,
                        GitError(
                            "Unable to repack alternates", project=self.name
                        ),
                    )
                platform_utils.remove(alternates_file)

        if self.worktree:
            self._InitMRef()
        else:
            self._InitMirrorHead()
            platform_utils.remove(
                os.path.join(self.gitdir, "FETCH_HEAD"), missing_ok=True
            )
        return SyncNetworkHalfResult(remote_fetched)

    def PostRepoUpgrade(self):
        self._InitHooks()

    def _CopyAndLinkFiles(self):
        if not self.worktree or not platform_utils.isdir(self.worktree):
            return
        for copyfile in self.copyfiles:
            copyfile._Copy()
        for linkfile in self.linkfiles:
            linkfile._Link()

    def GetCommitRevisionId(
        self, all_refs: Optional[Dict[str, str]] = None
    ) -> str:
        """Get revisionId of a commit.

        Use this method instead of GetRevisionId to get the id of the commit
        rather than the id of the current git object (for example, a tag)

        """
        if self.revisionId:
            return self.revisionId
        if not self.revisionExpr.startswith(R_TAGS):
            if all_refs is None:
                all_refs = self._allrefs
            return self.GetRevisionId(all_refs)

        try:
            return self.bare_git.ResolveCommit(self.revisionExpr)
        except GitError:
            raise ManifestInvalidRevisionError(
                f"revision {self.revisionExpr} in {self.name} not found"
            )

    def GetHeadRevisionId(self) -> Optional[str]:
        """Get the commit revision of the checked out HEAD.

        Returns None if worktree is not checked out or HEAD cannot be resolved.
        """
        if self.work_git:
            if not self.use_git_worktrees and self.bare_ref.is_loaded:
                head = self.bare_ref.get(HEAD)
                if head:
                    return head
            try:
                return self.work_git.rev_parse("HEAD")
            except GitError:
                pass
        return None

    def GetRevisionId(self, all_refs=None):
        if self.revisionId:
            return self.revisionId

        rem = self.GetRemote()
        rev = rem.ToLocal(self.revisionExpr)

        if all_refs is not None and rev in all_refs:
            return all_refs[rev]

        try:
            return self.bare_git.ResolveCommit(rev)
        except GitError:
            raise ManifestInvalidRevisionError(
                f"revision {self.revisionExpr} in {self.name} not found"
            )

    def SetRevisionId(self, revisionId):
        if self.revisionExpr:
            self.upstream = self.revisionExpr

        self.revisionId = revisionId

    def Sync_LocalHalf(
        self,
        syncbuf,
        force_sync=False,
        force_checkout=False,
        force_rebase=False,
        submodules=False,
        verbose=False,
    ):
        """Perform only the local IO portion of the sync process.

        Network access is not required.
        """

        def fail(error: Exception):
            syncbuf.fail(self, error)

        if not os.path.exists(self.gitdir):
            fail(
                LocalSyncFail(
                    "Cannot checkout %s due to missing network sync; Run "
                    "`repo sync -n %s` first." % (self.name, self.name),
                    project=self.name,
                )
            )
            return

        self._InitWorkTree(force_sync=force_sync, submodules=submodules)
        # TODO(https://git-scm.com/docs/git-worktree#_bugs): Re-evaluate if
        # submodules can be init when using worktrees once its support is
        # complete.
        if self.parent and not self.use_git_worktrees:
            self._InitSubmodule()
        all_refs = self.bare_ref.all
        self.CleanPublishedCache(all_refs)
        revid = self.GetRevisionId(all_refs)

        # Special case the root of the repo client checkout.  Make sure it
        # doesn't contain files being checked out to dirs we don't allow.
        if self.relpath == ".":
            PROTECTED_PATHS = {".repo"}
            paths = set(
                self.work_git.ls_tree("-z", "--name-only", "--", revid).split(
                    "\0"
                )
            )
            bad_paths = paths & PROTECTED_PATHS
            if bad_paths:
                fail(
                    LocalSyncFail(
                        "Refusing to checkout project that writes to protected "
                        "paths: %s" % (", ".join(bad_paths),),
                        project=self.name,
                    )
                )
                return

        def _doff():
            self._FastForward(revid)
            self._CopyAndLinkFiles()

        def _dorebase():
            self._Rebase(upstream="@{upstream}")

        def _dosubmodules():
            self._SyncSubmodules(quiet=True)

        def _doprune() -> None:
            """Expire reflogs and run prune-now GC for stateless sync."""
            GitCommand(
                self,
                ["reflog", "expire", "--expire=all", "--all"],
                bare=True,
            ).Wait()
            p = GitCommand(
                self,
                ["gc", "--prune=now"],
                bare=True,
                capture_stdout=True,
                capture_stderr=True,
            )
            if p.Wait() != 0:
                logger.warning("warn: %s: stateless gc failed", self.name)

        head = self._GetHead()
        if head and head.startswith(R_HEADS):
            branch = head[len(R_HEADS) :]
            try:
                head = all_refs[head]
            except KeyError:
                head = None
        else:
            branch = None

        if branch is None or syncbuf.detach_head:
            # Currently on a detached HEAD.  The user is assumed to
            # not have any local modifications worth worrying about.
            rebase_in_progress = (
                self.IsRebaseInProgress() or self.IsCherryPickInProgress()
            )
            if rebase_in_progress and force_checkout:
                self._AbortRebase()
                rebase_in_progress = (
                    self.IsRebaseInProgress() or self.IsCherryPickInProgress()
                )
            if rebase_in_progress:
                fail(_PriorSyncFailedError(project=self.name))
                return

            if head == revid:
                # No changes; don't do anything further.
                # Except if the head needs to be detached.
                if not syncbuf.detach_head:
                    # The copy/linkfile config may have changed.
                    self._CopyAndLinkFiles()
                    return
            else:
                lost = self._revlist(not_rev(revid), HEAD)
                if lost and verbose:
                    syncbuf.info(self, "discarding %d commits", len(lost))

            try:
                self._Checkout(revid, force_checkout=force_checkout, quiet=True)
                if submodules:
                    self._SyncSubmodules(quiet=True)
            except GitError as e:
                fail(e)
                return
            self._CopyAndLinkFiles()
            if self.stateless_prune_needed:
                syncbuf.later2(self, _doprune, not verbose)
            return

        if head == revid:
            # No changes; don't do anything further.
            #
            # The copy/linkfile config may have changed.
            self._CopyAndLinkFiles()
            return

        branch = self.GetBranch(branch)

        if not branch.LocalMerge:
            # The current branch has no tracking configuration.
            # Jump off it to a detached HEAD.
            syncbuf.info(
                self, "leaving %s; does not track upstream", branch.name
            )
            try:
                self._Checkout(revid, force_checkout=force_checkout, quiet=True)
                if submodules:
                    self._SyncSubmodules(quiet=True)
            except GitError as e:
                fail(e)
                return
            self._CopyAndLinkFiles()
            return

        upstream_gain = self._revlist(not_rev(HEAD), revid)

        # See if we can perform a fast forward merge.  This can happen if our
        # branch isn't in the exact same state as we last published.
        try:
            self.work_git.merge_base(
                "--is-ancestor", HEAD, revid, log_as_error=False
            )
            # Skip the published logic.
            pub = False
        except GitError:
            pub = self.WasPublished(branch.name, all_refs)

        if pub:
            not_merged = self._revlist(not_rev(revid), pub)
            if not_merged:
                if upstream_gain:
                    if force_rebase:
                        # Try to rebase local published but not merged changes
                        # on top of the upstream changes.
                        syncbuf.later1(self, _dorebase, not verbose)
                    else:
                        # The user has published this branch and some of those
                        # commits are not yet merged upstream.  We do not want
                        # to rewrite the published commits so we punt.
                        fail(
                            LocalSyncFail(
                                "branch %s is published (but not merged) and "
                                "is now %d commits behind. Fix this manually "
                                "or rerun with the --rebase option to force a "
                                "rebase." % (branch.name, len(upstream_gain)),
                                project=self.name,
                            )
                        )
                    return
                syncbuf.later1(self, _doff, not verbose)
                return
            elif pub == head:
                # All published commits are merged, and thus we are a
                # strict subset.  We can fast-forward safely.
                syncbuf.later1(self, _doff, not verbose)
                if submodules:
                    syncbuf.later1(self, _dosubmodules, not verbose)
                return

        # Examine the local commits not in the remote.  Find the
        # last one attributed to this user, if any.
        local_changes = self._revlist(not_rev(revid), HEAD, format="%H %ce")
        last_mine = None
        cnt_mine = 0
        for commit in local_changes:
            commit_id, committer_email = commit.split(" ", 1)
            if committer_email == self.UserEmail:
                last_mine = commit_id
                cnt_mine += 1

        if not upstream_gain and cnt_mine == len(local_changes):
            # The copy/linkfile config may have changed.
            self._CopyAndLinkFiles()
            return

        if self.IsDirty(consider_untracked=False):
            fail(_DirtyError(project=self.name))
            return

        # If the upstream switched on us, warn the user.
        if branch.merge != self.revisionExpr:
            if branch.merge and self.revisionExpr:
                syncbuf.info(
                    self,
                    "manifest switched %s...%s",
                    branch.merge,
                    self.revisionExpr,
                )
            elif branch.merge:
                syncbuf.info(self, "manifest no longer tracks %s", branch.merge)

        if cnt_mine < len(local_changes):
            # Upstream rebased. Not everything in HEAD was created by this user.
            syncbuf.info(
                self,
                "discarding %d commits removed from upstream",
                len(local_changes) - cnt_mine,
            )

        branch.remote = self.GetRemote()
        if not IsId(self.revisionExpr):
            # In case of manifest sync the revisionExpr might be a SHA1.
            branch.merge = self.revisionExpr
            if not branch.merge.startswith("refs/"):
                branch.merge = R_HEADS + branch.merge
        branch.Save()

        if cnt_mine > 0 and self.rebase:

            def _docopyandlink():
                self._CopyAndLinkFiles()

            def _dorebase():
                self._Rebase(upstream="%s^1" % last_mine, onto=revid)

            syncbuf.later2(self, _dorebase, not verbose)
            if submodules:
                syncbuf.later2(self, _dosubmodules, not verbose)
            syncbuf.later2(self, _docopyandlink, not verbose)
        elif local_changes:
            try:
                self._ResetHard(revid)
                if submodules:
                    self._SyncSubmodules(quiet=True)
                self._CopyAndLinkFiles()
            except GitError as e:
                fail(e)
                return
        else:
            syncbuf.later1(self, _doff, not verbose)
            if submodules:
                syncbuf.later1(self, _dosubmodules, not verbose)

        if self.stateless_prune_needed:
            syncbuf.later2(self, _doprune, not verbose)

    def AddCopyFile(self, src, dest, topdir):
        """Mark |src| for copying to |dest| (relative to |topdir|).

        No filesystem changes occur here.  Actual copying happens later on.

        Paths should have basic validation run on them before being queued.
        Further checking will be handled when the actual copy happens.
        """
        self.copyfiles[_CopyFile(self.worktree, src, topdir, dest)] = True

    def AddLinkFile(self, src, dest, topdir):
        """Mark |dest| to create a symlink (relative to |topdir|) pointing to
        |src|.

        No filesystem changes occur here.  Actual linking happens later on.

        Paths should have basic validation run on them before being queued.
        Further checking will be handled when the actual link happens.
        """
        self.linkfiles[_LinkFile(self.worktree, src, topdir, dest)] = True

    def AddAnnotation(self, name, value, keep):
        self.annotations.append(Annotation(name, value, keep))

    def DownloadPatchSet(self, change_id, patch_id):
        """Download a single patch set of a single change to FETCH_HEAD."""
        remote = self.GetRemote()

        cmd = ["fetch", remote.name]
        cmd.append(
            "refs/changes/%2.2d/%d/%d" % (change_id % 100, change_id, patch_id)
        )
        GitCommand(self, cmd, bare=True, verify_command=True).Wait()
        return DownloadedChange(
            self,
            self.GetRevisionId(),
            change_id,
            patch_id,
            self.bare_git.rev_parse("FETCH_HEAD"),
        )

    def DeleteWorktree(self, verbose=False, force=False):
        """Delete the source checkout and any other housekeeping tasks.

        This currently leaves behind the internal .repo/ cache state.  This
        helps when switching branches or manifest changes get reverted as we
        don't have to redownload all the git objects.  But we should do some GC
        at some point.

        Args:
            verbose: Whether to show verbose messages.
            force: Always delete tree even if dirty or corrupted.

        Returns:
            True if the worktree was completely cleaned out.
        """
        is_dirty = False
        is_corrupted = False
        try:
            is_dirty = self.IsDirty()
        except GitError:
            is_corrupted = True

        rel_path = self.RelPath(local=False)

        if is_dirty or is_corrupted:
            if force:
                if is_corrupted:
                    logger.warning(
                        "warning: %s: Removing corrupted project.",
                        rel_path,
                    )
                else:
                    logger.warning(
                        "warning: %s: Removing dirty project: "
                        "uncommitted changes lost.",
                        rel_path,
                    )
            else:
                if is_corrupted:
                    msg = (
                        f"error: {rel_path}: Cannot remove project: "
                        "project is corrupted.\n"
                    )
                    logger.error(msg)
                    raise DeleteWorktreeError(msg, project=self.name)
                else:
                    msg = (
                        f"error: {rel_path}: Cannot remove project: "
                        "uncommitted changes are present.\n"
                    )
                    logger.error(msg)
                    raise DeleteDirtyWorktreeError(msg, project=self.name)

        if verbose:
            print(f"{rel_path}: Deleting obsolete checkout.")

        # Unlock and delink from the main worktree.  We don't use git's worktree
        # remove because it will recursively delete projects -- we handle that
        # ourselves below.  https://crbug.com/git/48
        if self.use_git_worktrees:
            needle = os.path.realpath(self.gitdir)
            # Find the git worktree commondir under .repo/worktrees/.
            try:
                output = self.bare_git.worktree(
                    "list", "--porcelain"
                ).splitlines()[0]
                assert output.startswith("worktree "), output
                commondir = output[9:]
                # Walk each of the git worktrees to see where they point.
                configs = os.path.join(commondir, "worktrees")
                if os.path.exists(configs):
                    for name in os.listdir(configs):
                        gitdir = os.path.join(configs, name, "gitdir")
                        with open(gitdir) as fp:
                            relpath = fp.read().strip()
                        # Resolve the checkout path and see if it
                        # matches this project.
                        fullpath = os.path.realpath(
                            os.path.join(configs, name, relpath)
                        )
                        if fullpath == needle:
                            platform_utils.rmtree(os.path.join(configs, name))
            except GitError as e:
                logger.warning(
                    "warning: %s: Failed to list worktrees, skipping worktree "
                    "cleanup: %s",
                    rel_path,
                    e,
                )

        # Delete the .git directory first, so we're less likely to have a
        # partially working git repository around. There shouldn't be any git
        # projects here, so rmtree works.

        # Try to remove plain files first in case of git worktrees.  If this
        # fails for any reason, we'll fall back to rmtree, and that'll display
        # errors if it can't remove things either.
        try:
            platform_utils.remove(self.gitdir)
        except OSError:
            pass
        try:
            platform_utils.rmtree(self.gitdir)
        except OSError as e:
            if e.errno != errno.ENOENT:
                logger.error("error: %s: %s", self.gitdir, e)
                logger.error(
                    "error: %s: Failed to delete obsolete checkout; remove "
                    "manually, then run `repo sync -l`.",
                    rel_path,
                )
                raise DeleteWorktreeError(aggregate_errors=[e])

        # Delete everything under the worktree, except for directories that
        # contain another git project.
        dirs_to_remove = []
        failed = False
        errors = []
        for root, dirs, files in platform_utils.walk(self.worktree):
            for f in files:
                path = os.path.join(root, f)
                try:
                    platform_utils.remove(path)
                except OSError as e:
                    if e.errno != errno.ENOENT:
                        logger.warning("%s: Failed to remove: %s", path, e)
                        failed = True
                        errors.append(e)
            dirs[:] = [
                d
                for d in dirs
                if not os.path.lexists(os.path.join(root, d, ".git"))
            ]
            dirs_to_remove += [
                os.path.join(root, d)
                for d in dirs
                if os.path.join(root, d) not in dirs_to_remove
            ]
        for d in reversed(dirs_to_remove):
            if platform_utils.islink(d):
                try:
                    platform_utils.remove(d)
                except OSError as e:
                    if e.errno != errno.ENOENT:
                        logger.warning("%s: Failed to remove: %s", d, e)
                        failed = True
                        errors.append(e)
            elif not platform_utils.listdir(d):
                try:
                    platform_utils.rmdir(d)
                except OSError as e:
                    if e.errno != errno.ENOENT:
                        logger.warning("%s: Failed to remove: %s", d, e)
                        failed = True
                        errors.append(e)
        if failed:
            rename_path = (
                f"{self.worktree}_repo_to_be_deleted_{int(time.time())}"
            )
            try:
                platform_utils.rename(self.worktree, rename_path)
                logger.warning(
                    "warning: renamed %s to %s. You can delete it, but you "
                    "might need elevated permissions (e.g. root)",
                    self.worktree,
                    rename_path,
                )
                # Rename successful! Clear the errors.
                errors = []
            except OSError:
                logger.error(
                    "%s: Failed to delete obsolete checkout.\n",
                    "       Remove manually, then run `repo sync -l`.",
                    rel_path,
                )
                raise DeleteWorktreeError(aggregate_errors=errors)

        # Try deleting parent dirs if they are empty.
        path = self.worktree
        while path != self.manifest.topdir:
            try:
                platform_utils.rmdir(path)
            except OSError as e:
                if e.errno != errno.ENOENT:
                    break
            path = os.path.dirname(path)

        return True

    def StartBranch(self, name, branch_merge="", revision=None):
        """Create a new branch off the manifest's revision."""
        if not branch_merge:
            branch_merge = self.revisionExpr
        head = self.work_git.GetHead()
        if head == (R_HEADS + name):
            return True

        all_refs = self.bare_ref.all
        if R_HEADS + name in all_refs:
            GitCommand(
                self, ["checkout", "-q", name, "--"], verify_command=True
            ).Wait()
            return True

        branch = self.GetBranch(name)
        branch.remote = self.GetRemote()
        branch.merge = branch_merge
        if not branch.merge.startswith("refs/") and not IsId(branch_merge):
            branch.merge = R_HEADS + branch_merge

        if revision is None:
            revid = self.GetRevisionId(all_refs)
        else:
            revid = self.work_git.rev_parse(revision)

        if head.startswith(R_HEADS):
            try:
                head = all_refs[head]
            except KeyError:
                head = None
        if revid and head and revid == head:
            ref = R_HEADS + name
            self.work_git.update_ref(ref, revid)
            self.work_git.symbolic_ref(HEAD, ref)
            branch.Save()
            return True

        GitCommand(
            self,
            ["checkout", "-q", "-b", branch.name, revid],
            verify_command=True,
        ).Wait()
        branch.Save()
        return True

    def CheckoutBranch(self, name):
        """Checkout a local topic branch.

        Args:
            name: The name of the branch to checkout.

        Returns:
            True if the checkout succeeded; False if the
            branch doesn't exist.
        """
        rev = R_HEADS + name
        head = self.work_git.GetHead()
        if head == rev:
            # Already on the branch.
            return True

        all_refs = self.bare_ref.all
        try:
            revid = all_refs[rev]
        except KeyError:
            # Branch does not exist in this project.
            return False

        if head.startswith(R_HEADS):
            try:
                head = all_refs[head]
            except KeyError:
                head = None

        if head == revid:
            # Same revision; just update HEAD to point to the new
            # target branch, but otherwise take no other action.
            self.work_git.SetHead(R_HEADS + name)
            return True

        GitCommand(
            self,
            ["checkout", name, "--"],
            capture_stdout=True,
            capture_stderr=True,
            verify_command=True,
        ).Wait()
        return True

    def AbandonBranch(self, name):
        """Destroy a local topic branch.

        Args:
            name: The name of the branch to abandon.

        Returns:
            True if the abandon succeeded; Raises GitCommandError if it didn't;
            None if the branch didn't exist.
        """
        rev = R_HEADS + name
        all_refs = self.bare_ref.all
        if rev not in all_refs:
            # Doesn't exist
            return None

        head = self._GetHead()
        if head == rev:
            # We can't destroy the branch while we are sitting
            # on it.  Switch to a detached HEAD.
            head = all_refs[head]

            revid = self.GetRevisionId(all_refs)
            if head == revid:
                self.work_git.DetachHead(revid)
            else:
                self._Checkout(revid, quiet=True)
        GitCommand(
            self,
            ["branch", "-D", name],
            capture_stdout=True,
            capture_stderr=True,
            verify_command=True,
        ).Wait()
        return True

    def PruneHeads(self):
        """Prune any topic branches already merged into upstream."""
        kill = []
        left = self._allrefs
        cb = self.CurrentBranch
        for name in left.keys():
            if name.startswith(R_HEADS):
                name = name[len(R_HEADS) :]
                if cb is None or name != cb:
                    kill.append(name)

        # Minor optimization: If there's nothing to prune, then don't try to
        # read any project state.
        if not kill and not cb:
            return []

        rev = self.GetCommitRevisionId(left)
        head = left.get(R_HEADS + cb) if cb is not None else None
        if (
            cb is not None
            and head == rev
            and not self.IsDirty(consider_untracked=False)
        ):
            self.work_git.DetachHead(HEAD)
            kill.append(cb)

        if kill:
            if not self.use_git_worktrees:
                old = (
                    head
                    if cb in kill
                    else (self.bare_ref.head or self.bare_git.GetHead())
                )
            else:
                old = self.bare_git.GetHead()

            try:
                self.bare_git.DetachHead(rev)

                b = ["branch", "-d"]
                b.extend(kill)
                b = GitCommand(
                    self, b, bare=True, capture_stdout=True, capture_stderr=True
                )
                b.Wait()
            finally:
                if IsId(old):
                    self.bare_git.DetachHead(old)
                else:
                    branch = (
                        old[len(R_HEADS) :] if old.startswith(R_HEADS) else old
                    )
                    if branch not in kill:
                        self.bare_git.SetHead(old)
                left = self._allrefs

            for branch in kill:
                if (R_HEADS + branch) not in left:
                    self.CleanPublishedCache()
                    break

        if cb and cb not in kill:
            kill.append(cb)
        kill.sort()

        kept = []
        for branch in kill:
            if R_HEADS + branch in left:
                branch = self.GetBranch(branch)
                branch.current = branch.name == cb
                base = branch.LocalMerge
                if not base:
                    base = rev
                kept.append(ReviewableBranch(self, branch, base))
        return kept

    def GetRegisteredSubprojects(self):
        result = []

        def rec(subprojects):
            if not subprojects:
                return
            result.extend(subprojects)
            for p in subprojects:
                rec(p.subprojects)

        rec(self.subprojects)
        return result

    def _GetSubmodules(self):
        # Unfortunately we cannot call `git submodule status --recursive` here
        # because the working tree might not exist yet, and it cannot be used
        # without a working tree in its current implementation.

        def get_submodules(gitdir, rev):
            # Parse .gitmodules for submodule sub_paths and sub_urls.
            sub_paths, sub_urls, sub_shallows = parse_gitmodules(gitdir, rev)
            if not sub_paths:
                return []
            # Run `git ls-tree` to read SHAs of submodule object, which happen
            # to be revision of submodule repository.
            sub_revs = git_ls_tree(gitdir, rev, sub_paths)
            submodules = []
            for sub_path, sub_url, sub_shallow in zip(
                sub_paths, sub_urls, sub_shallows
            ):
                try:
                    sub_rev = sub_revs[sub_path]
                except KeyError:
                    # Ignore non-exist submodules.
                    continue
                submodules.append((sub_rev, sub_path, sub_url, sub_shallow))
            return submodules

        re_path = re.compile(r"^submodule\.(.+)\.path=(.*)$")
        re_url = re.compile(r"^submodule\.(.+)\.url=(.*)$")
        re_shallow = re.compile(r"^submodule\.(.+)\.shallow=(.*)$")

        def parse_gitmodules(gitdir, rev):
            cmd = ["cat-file", "blob", "%s:.gitmodules" % rev]
            try:
                p = GitCommand(
                    None,
                    cmd,
                    capture_stdout=True,
                    capture_stderr=True,
                    bare=True,
                    gitdir=gitdir,
                )
            except GitError:
                return [], [], []
            if p.Wait() != 0:
                return [], [], []

            gitmodules_lines = []
            fd, temp_gitmodules_path = tempfile.mkstemp()
            try:
                os.write(fd, p.stdout.encode("utf-8"))
                os.close(fd)
                cmd = ["config", "--file", temp_gitmodules_path, "--list"]
                p = GitCommand(
                    None,
                    cmd,
                    capture_stdout=True,
                    capture_stderr=True,
                    bare=True,
                    gitdir=gitdir,
                )
                if p.Wait() != 0:
                    return [], [], []
                gitmodules_lines = p.stdout.split("\n")
            except GitError:
                return [], [], []
            finally:
                platform_utils.remove(temp_gitmodules_path)

            names = set()
            paths = {}
            urls = {}
            shallows = {}
            for line in gitmodules_lines:
                if not line:
                    continue
                m = re_path.match(line)
                if m:
                    names.add(m.group(1))
                    paths[m.group(1)] = m.group(2)
                    continue
                m = re_url.match(line)
                if m:
                    names.add(m.group(1))
                    urls[m.group(1)] = m.group(2)
                    continue
                m = re_shallow.match(line)
                if m:
                    names.add(m.group(1))
                    shallows[m.group(1)] = m.group(2)
                    continue
            names = sorted(names)
            return (
                [paths.get(name, "") for name in names],
                [urls.get(name, "") for name in names],
                [shallows.get(name, "") for name in names],
            )

        def git_ls_tree(gitdir, rev, paths):
            cmd = ["ls-tree", rev, "--"]
            cmd.extend(paths)
            try:
                p = GitCommand(
                    None,
                    cmd,
                    capture_stdout=True,
                    capture_stderr=True,
                    bare=True,
                    gitdir=gitdir,
                )
            except GitError:
                return []
            if p.Wait() != 0:
                return []
            objects = {}
            for line in p.stdout.split("\n"):
                if not line.strip():
                    continue
                object_rev, object_path = line.split()[2:4]
                objects[object_path] = object_rev
            return objects

        try:
            rev = self.GetRevisionId()
        except (GitError, ManifestInvalidRevisionError):
            # The git repo may be outdated (i.e. not fetched yet) and querying
            # its submodules using the revision may not work; so return here.
            return []
        return get_submodules(self.gitdir, rev)

    def GetSubmoduleRevisions(self) -> Optional[Dict[str, str]]:
        """Read the gitlinks of our submodules at our current revision.

        Discovered submodules are derived from the revision their parent was
        at when the manifest was loaded, which is before it gets fetched.
        Once it is up-to-date, its gitlinks have to be read again, otherwise
        the submodules would be synced to the revisions of the previous sync.

        Returns:
            The revision of every submodule, keyed by its path inside this
            project, or None if our revision is not available locally. A
            revision that is merely set does not have to be fetched yet, and
            without its objects a removed submodule cannot be told apart from
            one that was never fetched.
        """
        try:
            rev = self.GetRevisionId()
            self.bare_git.rev_list(
                "-1",
                "--missing=allow-any",
                f"{rev}^0",
                "--",
                log_as_error=False,
            )
        except (GitError, ManifestInvalidRevisionError):
            return None

        return {
            path: sha for sha, path, _url, _shallow in self._GetSubmodules()
        }

    def GetDerivedSubprojects(self):
        result = []
        if not self.Exists:
            # If git repo does not exist yet, querying its submodules will
            # mess up its states; so return here.
            return result
        for rev, path, url, shallow in self._GetSubmodules():
            name = self.manifest.GetSubprojectName(self, path)
            (
                relpath,
                worktree,
                gitdir,
                objdir,
            ) = self.manifest.GetSubprojectPaths(self, name, path)
            project = self.manifest.paths.get(relpath)
            if project:
                result.extend(project.GetDerivedSubprojects())
                continue

            if url.startswith(("./", "../")):
                url = urllib.parse.urljoin("%s/" % self.remote.url, url)
            remote = RemoteSpec(
                self.remote.name,
                url=url,
                pushUrl=self.remote.pushUrl,
                review=self.remote.review,
                revision=self.remote.revision,
            )
            clone_depth = 1 if shallow.lower() == "true" else None
            subproject = Project(
                manifest=self.manifest,
                name=name,
                remote=remote,
                gitdir=gitdir,
                objdir=objdir,
                worktree=worktree,
                relpath=relpath,
                revisionExpr=rev,
                revisionId=rev,
                rebase=self.rebase,
                groups=self.groups,
                sync_c=self.sync_c,
                sync_s=self.sync_s,
                sync_tags=self.sync_tags,
                parent=self,
                clone_depth=clone_depth,
                is_derived=True,
                gitlink_path=path,
            )
            result.append(subproject)
            result.extend(subproject.GetDerivedSubprojects())
        return result

    def EnableRepositoryExtension(self, key, value="true", version=1):
        """Enable git repository extension |key| with |value|.

        Args:
            key: The extension to enabled.  Omit the "extensions." prefix.
            value: The value to use for the extension.
            version: The minimum git repository version needed.
        """
        # Make sure the git repo version is new enough already.
        found_version = self.config.GetInt("core.repositoryFormatVersion")
        if found_version is None:
            found_version = 0
        if found_version < version:
            self.config.SetString("core.repositoryFormatVersion", str(version))

        # Enable the extension!
        self.config.SetString(f"extensions.{key}", value)

    def ResolveRemoteHead(self, name=None):
        """Find out what the default branch (HEAD) points to.

        Normally this points to refs/heads/master, but projects are moving to
        main. Support whatever the server uses rather than hardcoding "master"
        ourselves.
        """
        if name is None:
            name = self.remote.name

        # The output will look like (NB: tabs are separators):
        # ref: refs/heads/master	HEAD
        # 5f6803b100bb3cd0f534e96e88c91373e8ed1c44	HEAD
        output = self.bare_git.ls_remote(
            "-q", "--symref", "--exit-code", name, "HEAD"
        )

        for line in output.splitlines():
            lhs, rhs = line.split("\t", 1)
            if rhs == "HEAD" and lhs.startswith("ref:"):
                return lhs[4:].strip()

        return None

    def _CheckForImmutableRevision(
        self,
        use_superproject: Optional[bool] = None,
        depth: Optional[int] = None,
    ) -> bool:
        try:
            # if revision (sha or tag) is not present then following function
            # throws an error.
            revs = [f"{self.revisionExpr}^0"]
            upstream_rev = None
            verify_upstream = self._ShouldVerifyUpstream(
                use_superproject=use_superproject,
                depth=depth,
            )

            # Ensure the local upstream tracking ref also exists in the ODB.
            if verify_upstream:
                upstream_rev = self.GetRemote().ToLocal(self.upstream)
                revs.append(upstream_rev)

            self.bare_git.rev_list(
                "-1",
                "--missing=allow-any",
                *revs,
                "--",
                log_as_error=False,
            )

            # Verify revision is an ancestor of the upstream tracking ref.
            if verify_upstream:
                self.bare_git.merge_base(
                    "--is-ancestor",
                    self.revisionExpr,
                    upstream_rev,
                    log_as_error=False,
                )
            return True
        except GitError:
            # There is no such persistent revision. We have to fetch it.
            return False

    def _HasShallow(self) -> bool:
        """Check if this project has a shallow file in its gitdir."""
        return bool(
            self.gitdir and os.path.exists(os.path.join(self.gitdir, "shallow"))
        )

    def _IsShallow(self, depth: Optional[int] = None) -> bool:
        """Check if the project is shallow or sharing shallow objects."""
        return bool(
            self._HasShallow() or self._SharingProjectHasShallow() or depth
        )

    def _ShouldVerifyUpstream(
        self,
        use_superproject: Optional[bool] = None,
        depth: Optional[int] = None,
    ) -> bool:
        """Whether to verify upstream ancestry during immutable revision
        check."""
        if not (IsId(self.revisionExpr) and self.upstream):
            return False
        if self._UseSuperprojectForUpstream(use_superproject):
            return True
        return not self._IsShallow(depth)

    def _SharingProjectHasShallow(self) -> bool:
        """Check if another project sharing this objdir has a "shallow" file.

        If any project sharing this objdir has a shallow file in its gitdir,
        then the shared objdir may be depth-limited, and every other project
        sharing this objdir needs its own shallow file so that git knows
        where history is truncated.
        """
        other_projects = self.manifest.GetProjectsWithName(
            self.name, all_manifests=True
        )
        for proj in other_projects:
            if proj.objdir == self.objdir and proj.gitdir != self.gitdir:
                if proj._HasShallow():
                    return True
        return False

    def _UseSuperprojectForUpstream(
        self, use_superproject: Optional[bool] = None
    ) -> bool:
        """Whether to check upstream for superprojects."""
        return git_superproject.UseSuperproject(use_superproject, self.manifest)

    def _FetchArchive(self, tarpath, cwd=None):
        cmd = ["archive", "-v", "-o", tarpath]
        cmd.append("--remote=%s" % self.remote.url)
        cmd.append("--prefix=%s/" % self.RelPath(local=False))
        cmd.append(self.revisionExpr)

        command = GitCommand(
            self,
            cmd,
            cwd=cwd,
            capture_stdout=True,
            capture_stderr=True,
            verify_command=True,
        )
        command.Wait()

    def _CustomFetch(self, verbose=False) -> bool:
        """Fetch using a custom command specified by repo.fetchcmd.

        Populates the subshell with project-context environment variables,
        including REPO_TREV (target revision resolved to a commit hash).

        Expects the target commit to be reachable and tracking refs/FETCH_HEAD
        to be updated upon a successful 0 exit code.

        For a detailed contract, environment variables, and postconditions,
        see docs/fetch-cmd.md.
        """
        # Resolve REPO_TREV (target revision resolved to a full commit hash).
        repo_trev = None
        if self.revisionId and IsId(self.revisionId):
            repo_trev = self.revisionId

        if not repo_trev:
            output = self._LsRemote(self.upstream or self.revisionExpr)
            if output:
                lines = output.splitlines()
                if lines:
                    parts = lines[0].split()
                    if parts:
                        repo_trev = parts[0]

        if not repo_trev:
            logger.error("error: Cannot resolve REPO_TREV for %s", self.name)
            return False

        env = os.environ.copy()
        env.update(self.GetEnvVars())
        env["REPO_TREV"] = repo_trev
        cmd_str = self.manifest.manifestProject.fetch_cmd

        if verbose:
            print(f"Running fetchcmd: {cmd_str} for {self.name}")

        stdout = None if verbose else subprocess.PIPE
        stderr = None if verbose else subprocess.PIPE
        try:
            p = subprocess.run(
                cmd_str,
                shell=True,
                cwd=self.manifest.topdir,
                env=env,
                stdout=stdout,
                stderr=stderr,
                text=True,
            )

            if p.returncode != 0:
                logger.error(
                    "error: fetchcmd failed with exit code %d", p.returncode
                )
                if not verbose:
                    logger.error("stderr: %s", p.stderr)
                return False

        except Exception as e:
            logger.error("error: Failed to execute fetchcmd: %s", e)
            return False

        # Verify postconditions.
        try:
            self.bare_git.cat_file("-e", repo_trev)
        except GitError:
            logger.error(
                "error: Postcondition failed: %s not found in object store",
                repo_trev,
            )
            return False

        try:
            ref_name = self.GetRemote().ToLocal(self.revisionExpr)
        except GitError as e:
            logger.error("error: Failed to resolve tracking ref: %s", e)
            return False
        try:
            resolved_ref = self.bare_git.rev_parse(ref_name)
            if resolved_ref != repo_trev:
                logger.error(
                    "error: Postcondition failed: %s is %s, expected %s",
                    ref_name,
                    resolved_ref,
                    repo_trev,
                )
                return False
        except GitError:
            logger.error("error: Postcondition failed: %s not found", ref_name)
            return False

        try:
            fetch_head = self.bare_git.rev_parse("FETCH_HEAD")
            if fetch_head != repo_trev:
                logger.error(
                    "error: Postcondition failed: FETCH_HEAD is %s, "
                    "expected %s",
                    fetch_head,
                    repo_trev,
                )
                return False
        except GitError:
            logger.error("error: Postcondition failed: FETCH_HEAD not found")
            return False

        return True

    def _GetUpstreamFallback(self) -> Optional[str]:
        """Resolve a fallback upstream branch when revisionExpr is a SHA-1.

        Returns manifest default upstream or revision if it names a branch,
        or None to fall back to fetching all heads.
        """
        default = self.manifest.default
        if not default:
            return None
        for cand in (default.upstreamExpr, default.revisionExpr):
            if cand and not IsId(cand) and not cand.startswith(R_TAGS):
                return cand
        return None

    def _RemoteFetch(
        self,
        name=None,
        use_superproject=None,
        current_branch_only=False,
        initial=False,
        quiet=False,
        verbose=False,
        output_redir=None,
        alt_dir=None,
        tags=True,
        prune=False,
        depth=None,
        submodules=False,
        ssh_proxy=None,
        force_sync=False,
        clone_filter=None,
        retry_fetches=2,
        retry_sleep_initial_sec=4.0,
        retry_exp_factor=2.0,
    ) -> bool:
        tag_name = None
        # The depth should not be used when fetching to a mirror because
        # it will result in a shallow repository that cannot be cloned or
        # fetched from.
        # The repo project should also never be synced with partial depth.
        if self.manifest.IsMirror or self.relpath == ".repo/repo":
            depth = None

        if depth:
            current_branch_only = True

        is_sha1 = IsId(self.revisionExpr)
        upstream = self.upstream

        if current_branch_only:
            if is_sha1 and not depth:
                # When syncing a specific commit and --depth is not set:
                # * if upstream is explicitly specified and is not a sha1, fetch
                #   only upstream as users expect only upstream to be fetch.
                #   Note: The commit might not be in upstream in which case the
                #   sync will fail.
                # * otherwise, fetch all branches to make sure we end up with
                #   the specific commit.
                if not upstream:
                    upstream = self._GetUpstreamFallback()

                if upstream:
                    current_branch_only = not IsId(upstream)
                else:
                    current_branch_only = False

            if self.revisionExpr.startswith(R_TAGS):
                # This is a tag and its commit id should never change.
                tag_name = self.revisionExpr[len(R_TAGS) :]
            elif upstream and upstream.startswith(R_TAGS):
                # This is a tag and its commit id should never change.
                tag_name = upstream[len(R_TAGS) :]

            if is_sha1 or tag_name is not None:
                has_shallow = self._HasShallow()
                if self._CheckForImmutableRevision(
                    use_superproject=use_superproject,
                    depth=depth,
                ) and (has_shallow or not self._IsShallow(depth)):
                    if verbose:
                        print(
                            "Skipped fetching project %s (already have "
                            "persistent ref)" % self.name
                        )
                    return True

        if not name:
            name = self.remote.name

        remote = self.GetRemote(name)
        if not remote.PreConnectFetch(ssh_proxy):
            ssh_proxy = None

        alt_tmp_refs = []
        if initial:
            if alt_dir and "objects" == os.path.basename(alt_dir):
                ref_dir = os.path.dirname(alt_dir)

                all_refs = self.bare_ref.all
                ids = set(all_refs.values())

                update_ref_cmds = []

                for r, ref_id in GitRefs(ref_dir).all.items():
                    if r not in all_refs:
                        if r.startswith(R_TAGS) or remote.WritesTo(r):
                            update_ref_cmds.append(f"create {r} {ref_id}\n")
                            all_refs[r] = ref_id
                            ids.add(ref_id)
                            continue

                    if ref_id in ids:
                        continue

                    r = "refs/_alt/%s" % ref_id
                    update_ref_cmds.append(f"create {r} {ref_id}\n")
                    all_refs[r] = ref_id
                    ids.add(ref_id)
                    alt_tmp_refs.append(r)

                if update_ref_cmds:
                    GitCommand(
                        self,
                        ["update-ref", "--stdin"],
                        bare=True,
                        input="".join(update_ref_cmds),
                    ).Wait()
            else:
                alt_dir = None

        cmd = ["fetch"]

        if clone_filter:
            git_require((2, 19, 0), fail=True, msg="partial clones")
            cmd.append("--filter=%s" % clone_filter)
            self.EnableRepositoryExtension("partialclone", self.remote.name)

        if depth:
            cmd.append("--depth=%s" % depth)
        else:
            # If this repo has shallow objects, then we don't know which refs
            # have shallow objects or not. Tell git to unshallow all fetched
            # refs.  Don't do this with projects that don't have shallow
            # objects, since it is less efficient.
            if self._HasShallow():
                cmd.append("--depth=2147483647")

        # Use clone-depth="1" as a heuristic for repositories containing
        # large binaries and disable auto GC to prevent potential hangs.
        # Check the configured depth because the `depth` argument might be None
        # if REPO_ALLOW_SHALLOW=0 converted it to a partial clone.
        effective_depth = (
            self.clone_depth or self.manifest.manifestProject.depth
        )
        if effective_depth == 1 and git_require((2, 23, 0)):
            cmd.append("--no-auto-gc")

        if not verbose:
            cmd.append("--quiet")
        if not quiet and sys.stdout.isatty():
            cmd.append("--progress")
        if not self.worktree:
            cmd.append("--update-head-ok")
        cmd.append(name)

        if force_sync:
            cmd.append("--force")

        if prune:
            cmd.append("--prune")

        # Always pass something for --recurse-submodules, git with GIT_DIR
        # behaves incorrectly when not given `--recurse-submodules=no`.
        # (b/218891912)
        cmd.append(
            f'--recurse-submodules={"on-demand" if submodules else "no"}'
        )

        spec = []
        if not current_branch_only:
            # Fetch whole repo.
            spec.append(
                str(("+refs/heads/*:") + remote.ToLocal("refs/heads/*"))
            )
        elif tag_name is not None:
            spec.append("tag")
            spec.append(tag_name)

        if self.manifest.IsMirror and not current_branch_only:
            branch = None
        else:
            branch = self.revisionExpr
        if not self.manifest.IsMirror and is_sha1 and depth:
            # Shallow checkout of a specific commit, fetch from that commit and
            # not the heads only as the commit might be deeper in the history.
            spec.append(branch)
            if upstream:
                spec.append(upstream)
        else:
            if is_sha1:
                branch = upstream
            if branch is not None and branch.strip():
                if not branch.startswith("refs/"):
                    branch = R_HEADS + branch
                spec.append(str(("+%s:" % branch) + remote.ToLocal(branch)))

        # If mirroring repo and we cannot deduce the tag or branch to fetch,
        # fetch whole repo.
        if self.manifest.IsMirror and not spec:
            spec.append(
                str(("+refs/heads/*:") + remote.ToLocal("refs/heads/*"))
            )

        # If using depth then we should not get all the tags since they may
        # be outside of the depth.
        if not tags or depth:
            cmd.append("--no-tags")
        else:
            cmd.append("--tags")
            spec.append(str(("+refs/tags/*:") + remote.ToLocal("refs/tags/*")))

        cmd.extend(spec)

        # At least one retry minimum due to git remote prune.
        retry_fetches = max(retry_fetches, 2)
        retry_cur_sleep = retry_sleep_initial_sec
        ok = prune_tried = False
        try:
            for try_n in range(retry_fetches):
                verify_command = try_n == retry_fetches - 1
                gitcmd = GitCommand(
                    self,
                    cmd,
                    bare=True,
                    objdir=os.path.join(self.objdir, "objects"),
                    ssh_proxy=ssh_proxy,
                    merge_output=True,
                    capture_stdout=quiet or bool(output_redir),
                    verify_command=verify_command,
                )
                if gitcmd.stdout and not quiet and output_redir:
                    output_redir.write(gitcmd.stdout)
                ret = gitcmd.Wait()
                if ret == 0:
                    ok = True
                    break

                # Retry later due to HTTP 429 Too Many Requests.
                elif (
                    gitcmd.stdout
                    and "error:" in gitcmd.stdout
                    and "HTTP 429" in gitcmd.stdout
                ):
                    # Fallthru to sleep+retry logic at the bottom.
                    pass

                # TODO(b/360889369#comment24): git may gc commits incorrectly.
                # Until the root cause is fixed, retry fetch with --refetch
                # which will bring the repository into a good state.
                elif gitcmd.stdout and (
                    "could not parse commit" in gitcmd.stdout
                    or "unable to parse commit" in gitcmd.stdout
                ):
                    cmd.insert(1, "--refetch")
                    print(
                        "could not parse commit error, retrying with refetch",
                        file=output_redir,
                    )
                    continue

                # Try to prune remote branches once in case there are conflicts.
                # For example, if the remote had refs/heads/upstream, but
                # deleted that and now has refs/heads/upstream/foo.
                elif (
                    gitcmd.stdout
                    and "error:" in gitcmd.stdout
                    and "git remote prune" in gitcmd.stdout
                    and not prune_tried
                ):
                    prune_tried = True
                    prunecmd = GitCommand(
                        self,
                        ["remote", "prune", name],
                        bare=True,
                        ssh_proxy=ssh_proxy,
                    )
                    ret = prunecmd.Wait()
                    if ret:
                        break
                    print(
                        "retrying fetch after pruning remote branches",
                        file=output_redir,
                    )
                    # Continue right away so we don't sleep as we shouldn't
                    # need to.
                    continue
                elif (
                    ret == 128
                    and gitcmd.stdout
                    and "fatal: could not read Username" in gitcmd.stdout
                ):
                    # User needs to be authenticated, and Git wants to prompt
                    # for username and password.
                    print(
                        "git requires authentication, but repo cannot perform "
                        "interactive authentication. Check git credentials.",
                        file=output_redir,
                    )
                    break
                elif (
                    ret == 128
                    and gitcmd.stdout
                    and "remote helper 'sso' aborted session" in gitcmd.stdout
                ):
                    # User needs to be authenticated, and Git wants to prompt
                    # for username and password.
                    print(
                        "git requires authentication, but repo cannot perform "
                        "interactive authentication.",
                        file=output_redir,
                    )
                    raise GitAuthError(gitcmd.stdout)
                    break
                elif current_branch_only and is_sha1 and ret == 128:
                    # Exit code 128 means "couldn't find the ref you asked for";
                    # if we're in sha1 mode, we just tried sync'ing from the
                    # upstream field; it doesn't exist, thus abort the
                    # optimization attempt and do a full sync.
                    break
                elif depth and is_sha1 and ret == 1:
                    # In sha1 mode, when depth is enabled, syncing the revision
                    # from upstream may not work because some servers only allow
                    # fetching named refs. Fetching a specific sha1 may result
                    # in an error like 'server does not allow request for
                    # unadvertised object'. In this case, attempt a full sync
                    # without depth.
                    break
                elif ret < 0:
                    # Git died with a signal, exit immediately.
                    break

                # Figure out how long to sleep before the next attempt, if
                # there is one.
                if not verbose and gitcmd.stdout:
                    print(
                        f"\n{self.name}:\n{gitcmd.stdout}",
                        end="",
                        file=output_redir,
                    )
                if try_n < retry_fetches - 1:
                    print(
                        "%s: sleeping %s seconds before retrying"
                        % (self.name, retry_cur_sleep),
                        file=output_redir,
                    )
                    time.sleep(retry_cur_sleep)
                    retry_cur_sleep = min(
                        retry_exp_factor * retry_cur_sleep,
                        MAXIMUM_RETRY_SLEEP_SEC,
                    )
                    retry_cur_sleep *= 1 - random.uniform(
                        -RETRY_JITTER_PERCENT, RETRY_JITTER_PERCENT
                    )
        finally:
            if initial:
                if alt_tmp_refs:
                    delete_cmds = "".join(
                        f"delete {ref}\n" for ref in alt_tmp_refs
                    )
                    GitCommand(
                        self,
                        ["update-ref", "--stdin"],
                        bare=True,
                        input=delete_cmds,
                        log_as_error=False,
                    ).Wait()

                    for ref in alt_tmp_refs:
                        self.bare_ref.deleted(ref)

                self.bare_git.pack_refs("--all", "--prune")

        if is_sha1 and current_branch_only:
            # We just synced the upstream given branch; verify we
            # got what we wanted, else trigger a second run of all
            # refs.
            if not self._CheckForImmutableRevision(
                use_superproject=use_superproject,
                depth=depth,
            ):
                # Sync the current branch only with depth set to None.
                # We always pass depth=None down to avoid infinite recursion.
                return self._RemoteFetch(
                    name=name,
                    quiet=quiet,
                    verbose=verbose,
                    output_redir=output_redir,
                    use_superproject=use_superproject,
                    current_branch_only=current_branch_only and depth,
                    initial=False,
                    alt_dir=alt_dir,
                    tags=tags,
                    depth=None,
                    ssh_proxy=ssh_proxy,
                    clone_filter=clone_filter,
                )

        return ok

    def _ApplyCloneBundle(self, initial=False, quiet=False, verbose=False):
        if initial and (
            self.manifest.manifestProject.depth or self.clone_depth
        ):
            return False

        remote = self.GetRemote()
        bundle_url = remote.url + "/clone.bundle"
        bundle_url = GitConfig.ForUser().UrlInsteadOf(bundle_url)
        if GetSchemeFromUrl(bundle_url) not in (
            "http",
            "https",
            "persistent-http",
            "persistent-https",
        ):
            return False

        bundle_dst = os.path.join(self.gitdir, "clone.bundle")
        bundle_tmp = os.path.join(self.gitdir, "clone.bundle.tmp")

        exist_dst = os.path.exists(bundle_dst)
        exist_tmp = os.path.exists(bundle_tmp)

        if not initial and not exist_dst and not exist_tmp:
            return False

        if not exist_dst:
            exist_dst = self._FetchBundle(
                bundle_url, bundle_tmp, bundle_dst, quiet, verbose
            )
        if not exist_dst:
            return False

        cmd = ["fetch"]
        if not verbose:
            cmd.append("--quiet")
        if not quiet and sys.stdout.isatty():
            cmd.append("--progress")
        if not self.worktree:
            cmd.append("--update-head-ok")
        cmd.append(bundle_dst)
        for f in remote.fetch:
            cmd.append(str(f))
        cmd.append("+refs/tags/*:refs/tags/*")

        ok = (
            GitCommand(
                self,
                cmd,
                bare=True,
                objdir=os.path.join(self.objdir, "objects"),
            ).Wait()
            == 0
        )
        platform_utils.remove(bundle_dst, missing_ok=True)
        platform_utils.remove(bundle_tmp, missing_ok=True)
        return ok

    def _FetchBundle(self, srcUrl, tmpPath, dstPath, quiet, verbose):
        platform_utils.remove(dstPath, missing_ok=True)

        # We do not use curl's --retry option since it generally doesn't
        # actually retry anything; code 18 for example, it will not retry on.
        cmd = [
            "curl",
            "--fail",
            "--output",
            tmpPath,
            "--netrc-optional",
            "--location",
        ]
        if quiet:
            cmd += ["--silent", "--show-error"]
        if os.path.exists(tmpPath):
            size = os.stat(tmpPath).st_size
            if size >= 1024:
                cmd += ["--continue-at", "%d" % (size,)]
            else:
                platform_utils.remove(tmpPath)
        with GetUrlCookieFile(srcUrl, quiet) as (cookiefile, proxy):
            if cookiefile:
                cmd += ["--cookie", cookiefile]
            if proxy:
                cmd += ["--proxy", proxy]
            elif "http_proxy" in os.environ and "darwin" == sys.platform:
                cmd += ["--proxy", os.environ["http_proxy"]]
            if srcUrl.startswith("persistent-https"):
                srcUrl = "http" + srcUrl[len("persistent-https") :]
            elif srcUrl.startswith("persistent-http"):
                srcUrl = "http" + srcUrl[len("persistent-http") :]
            cmd += [srcUrl]

            proc = None
            with Trace("Fetching bundle: %s", " ".join(cmd)):
                if verbose:
                    print(f"{self.name}: Downloading bundle: {srcUrl}")
                stdout = None if verbose else subprocess.PIPE
                stderr = None if verbose else subprocess.STDOUT
                try:
                    proc = subprocess.Popen(cmd, stdout=stdout, stderr=stderr)
                except OSError:
                    return False

            (output, _) = proc.communicate()
            curlret = proc.returncode

            if curlret in (22, 35, 56, 92):
                # We use --fail so curl exits with unique status.
                # From curl man page:
                # 22: HTTP page not retrieved.  The requested url was not found
                #     or returned another error with the HTTP error code being
                #     400 or above.
                # 35: SSL connect error.  The SSL handshaking failed.  This can
                #     be thrown by Google storage sometimes.
                # 56: Failure in receiving network data.  This shows up with
                #     HTTP/404 on Google storage.
                # 92: Stream error in HTTP/2 framing layer.  Basically the same
                #     as 22 -- Google storage sometimes throws 500's.
                if verbose:
                    print(
                        "%s: Unable to retrieve clone.bundle; ignoring."
                        % self.name
                    )
                    if output:
                        print("Curl output:\n%s" % output)
                return False
            elif curlret and not verbose and output:
                logger.error("%s", output)

        if os.path.exists(tmpPath):
            if curlret == 0 and self._IsValidBundle(tmpPath, quiet):
                platform_utils.rename(tmpPath, dstPath)
                return True
            else:
                platform_utils.remove(tmpPath)
                return False
        else:
            return False

    def _IsValidBundle(self, path, quiet):
        try:
            with open(path, "rb") as f:
                if f.read(16) == b"# v2 git bundle\n":
                    return True
                else:
                    if not quiet:
                        logger.error("Invalid clone.bundle file; ignoring.")
                    return False
        except OSError:
            return False

    def _Checkout(self, rev, force_checkout=False, quiet=False):
        cmd = ["checkout"]
        if quiet:
            cmd.append("-q")
        if force_checkout:
            cmd.append("-f")
        cmd.append(rev)
        cmd.append("--")
        if GitCommand(self, cmd).Wait() != 0:
            if self._allrefs:
                raise GitError(
                    f"{self.name} checkout {rev} ", project=self.name
                )

    def _CherryPick(self, rev, ffonly=False, record_origin=False):
        cmd = ["cherry-pick"]
        if ffonly:
            cmd.append("--ff")
        if record_origin:
            cmd.append("-x")
        cmd.append(rev)
        cmd.append("--")
        if GitCommand(self, cmd).Wait() != 0:
            if self._allrefs:
                raise GitError(
                    f"{self.name} cherry-pick {rev} ", project=self.name
                )

    def _LsRemote(self, refs):
        cmd = ["ls-remote", self.remote.name, refs]
        p = GitCommand(self, cmd, capture_stdout=True)
        if p.Wait() == 0:
            return p.stdout
        return None

    def _Revert(self, rev):
        cmd = ["revert"]
        cmd.append("--no-edit")
        cmd.append(rev)
        cmd.append("--")
        if GitCommand(self, cmd).Wait() != 0:
            if self._allrefs:
                raise GitError(f"{self.name} revert {rev} ", project=self.name)

    def _ResetHard(self, rev, quiet=True):
        cmd = ["reset", "--hard"]
        if quiet:
            cmd.append("-q")
        cmd.append(rev)
        if GitCommand(self, cmd).Wait() != 0:
            raise GitError(
                f"{self.name} reset --hard {rev} ", project=self.name
            )

    def _SyncSubmodules(self, quiet=True):
        cmd = ["submodule", "update", "--init", "--recursive"]
        if quiet:
            cmd.append("-q")
        if GitCommand(self, cmd).Wait() != 0:
            raise GitError(
                "%s submodule update --init --recursive " % self.name,
                project=self.name,
            )

    def _InitSubmodule(self, quiet=True):
        """Initialize the submodule."""
        cmd = ["submodule", "init"]
        if quiet:
            cmd.append("-q")
        cmd.extend(["--", self.worktree])
        max_retries = 3
        base_delay_secs = 1
        jitter_ratio = 1 / 3
        for attempt in range(max_retries):
            git_cmd = GitCommand(
                None,
                cmd,
                cwd=self.parent.worktree,
                capture_stdout=True,
                capture_stderr=True,
            )
            if git_cmd.Wait() == 0:
                return
            error = git_cmd.stderr or git_cmd.stdout
            if "lock" in error:
                delay = base_delay_secs * (2**attempt)
                delay += random.uniform(0, delay * jitter_ratio)
                logger.warning(
                    f"Attempt {attempt+1}/{max_retries}: "
                    + f"git {' '.join(cmd)} failed."
                    + f" Error: {error}."
                    + f" Sleeping {delay:.2f}s before retrying."
                )
                time.sleep(delay)
            else:
                break
        git_cmd.VerifyCommand()

    def _Rebase(self, upstream, onto=None):
        cmd = ["rebase"]
        if onto is not None:
            cmd.extend(["--onto", onto])
        cmd.append(upstream)
        if GitCommand(self, cmd).Wait() != 0:
            raise GitError(f"{self.name} rebase {upstream} ", project=self.name)

    def _FastForward(self, head, ffonly=False, quiet=True):
        cmd = ["merge", "--no-stat", head]
        if ffonly:
            cmd.append("--ff-only")
        if quiet:
            cmd.append("-q")
        if GitCommand(self, cmd).Wait() != 0:
            raise GitError(f"{self.name} merge {head} ", project=self.name)

    def _InitGitDir(self, mirror_git=None, force_sync=False, quiet=False):
        # Prefix for temporary directories created during gitdir initialization.
        TMP_GITDIR_PREFIX = ".tmp-project-initgitdir-"
        init_git_dir = not os.path.exists(self.gitdir)
        init_obj_dir = not os.path.exists(self.objdir)
        tmp_gitdir = None
        curr_gitdir = self.gitdir
        curr_config = self.config
        try:
            # Initialize the bare repository, which contains all of the objects.
            if init_obj_dir:
                os.makedirs(self.objdir)
                self.bare_objdir.init()

                self._UpdateHooks(quiet=quiet)

                if self.use_git_worktrees:
                    # Enable per-worktree config file support if possible.  This
                    # is more a nice-to-have feature for users rather than a
                    # hard requirement.
                    if git_require((2, 20, 0)):
                        self.EnableRepositoryExtension("worktreeConfig")

            # If we have a separate directory to hold refs, initialize it as
            # well.
            if self.objdir != self.gitdir:
                if init_git_dir:
                    os.makedirs(os.path.dirname(self.gitdir), exist_ok=True)
                    tmp_gitdir = tempfile.mkdtemp(
                        prefix=TMP_GITDIR_PREFIX,
                        dir=os.path.dirname(self.gitdir),
                    )
                    curr_config = GitConfig.ForRepository(
                        gitdir=tmp_gitdir, defaults=self.manifest.globalConfig
                    )
                    curr_gitdir = tmp_gitdir

                if init_obj_dir or init_git_dir:
                    self._ReferenceGitDir(
                        self.objdir, curr_gitdir, copy_all=True
                    )
                try:
                    self._CheckDirReference(self.objdir, curr_gitdir)
                except GitError as e:
                    if force_sync:
                        try:
                            rm_dirs = (
                                tmp_gitdir,
                                self.gitdir,
                                self.worktree,
                            )
                            for d in rm_dirs:
                                if d and os.path.exists(d):
                                    platform_utils.rmtree(os.path.realpath(d))
                            return self._InitGitDir(
                                mirror_git=mirror_git,
                                force_sync=False,
                                quiet=quiet,
                            )
                        except Exception:
                            raise e
                    raise e

            if init_git_dir:
                mp = self.manifest.manifestProject
                ref_dir = mp.reference or ""

                def _expanded_ref_dirs():
                    """Iterate through possible git reference dir paths."""
                    name = self.name + ".git"
                    yield mirror_git or os.path.join(ref_dir, name)
                    for prefix in "", self.remote.name:
                        yield os.path.join(
                            ref_dir, ".repo", "project-objects", prefix, name
                        )
                        yield os.path.join(
                            ref_dir, ".repo", "worktrees", prefix, name
                        )

                if ref_dir or mirror_git:
                    found_ref_dir = None
                    for path in _expanded_ref_dirs():
                        if os.path.exists(path):
                            found_ref_dir = path
                            break
                    ref_dir = found_ref_dir

                    if ref_dir:
                        if not os.path.isabs(ref_dir):
                            # The alternate directory is relative to the object
                            # database.
                            ref_dir = os.path.relpath(
                                ref_dir, os.path.join(self.objdir, "objects")
                            )
                        _lwrite(
                            os.path.join(
                                self.objdir, "objects/info/alternates"
                            ),
                            os.path.join(ref_dir, "objects") + "\n",
                        )

                m = self.manifest.manifestProject.config
                for key in ["user.name", "user.email"]:
                    if m.Has(key, include_defaults=False):
                        curr_config.SetString(key, m.GetString(key))
                if not self.manifest.EnableGitLfs:
                    curr_config.SetString(
                        "filter.lfs.smudge", "git-lfs smudge --skip -- %f"
                    )
                    curr_config.SetString(
                        "filter.lfs.process", "git-lfs filter-process --skip"
                    )
                curr_config.SetBoolean(
                    "core.bare", True if self.manifest.IsMirror else None
                )

                if tmp_gitdir:
                    platform_utils.rename(tmp_gitdir, self.gitdir)
                    tmp_gitdir = None
            if not init_obj_dir:
                # The project might be shared (obj_dir already initialized), but
                # such information is not available here. Instead of passing it,
                # set it as shared, and rely to be unset down the execution
                # path.
                if git_require((2, 7, 0)):
                    self.EnableRepositoryExtension("preciousObjects")
                else:
                    self.config.SetString("gc.pruneExpire", "never")

        except Exception:
            if init_obj_dir and os.path.exists(self.objdir):
                platform_utils.rmtree(self.objdir)
            if init_git_dir and os.path.exists(self.gitdir):
                platform_utils.rmtree(self.gitdir)
            raise
        finally:
            # Clean up the temporary directory created during the process,
            # as well as any stale ones left over from previous attempts.
            if tmp_gitdir and os.path.exists(tmp_gitdir):
                platform_utils.rmtree(tmp_gitdir)

            age_threshold = datetime.timedelta(days=1)
            now = datetime.datetime.now()
            for tmp_dir in glob.glob(
                os.path.join(
                    os.path.dirname(self.gitdir), f"{TMP_GITDIR_PREFIX}*"
                )
            ):
                try:
                    mtime = datetime.datetime.fromtimestamp(
                        os.path.getmtime(tmp_dir)
                    )
                    if now - mtime > age_threshold:
                        platform_utils.rmtree(tmp_dir)
                except OSError:
                    pass

    def _UpdateHooks(self, quiet=False):
        if os.path.exists(self.objdir):
            self._InitHooks(quiet=quiet)

    def _InitHooks(self, quiet=False):
        hooks = os.path.realpath(os.path.join(self.objdir, "hooks"))
        if not os.path.exists(hooks):
            os.makedirs(hooks)

        # Delete sample hooks.  They're noise.
        for hook in glob.glob(os.path.join(hooks, "*.sample")):
            try:
                platform_utils.remove(hook, missing_ok=True)
            except PermissionError:
                pass

        for stock_hook in _ProjectHooks():
            name = os.path.basename(stock_hook)

            if (
                name in ("commit-msg",)
                and not self.remote.review
                and self is not self.manifest.manifestProject
            ):
                # Don't install a Gerrit Code Review hook if this
                # project does not appear to use it for reviews.
                #
                # Since the manifest project is one of those, but also
                # managed through gerrit, it's excluded.
                continue

            dst = os.path.join(hooks, name)
            if platform_utils.islink(dst):
                continue
            if os.path.exists(dst):
                # If the files are the same, we'll leave it alone.  We create
                # symlinks below by default but fallback to hardlinks if the OS
                # blocks them. So if we're here, it's probably because we made a
                # hardlink below.
                if not filecmp.cmp(stock_hook, dst, shallow=False):
                    if not quiet:
                        logger.warning(
                            "warn: %s: Not replacing locally modified %s hook",
                            self.RelPath(local=False),
                            name,
                        )
                continue
            try:
                platform_utils.symlink(
                    os.path.relpath(stock_hook, os.path.dirname(dst)), dst
                )
            except OSError as e:
                if e.errno == errno.EPERM:
                    try:
                        os.link(stock_hook, dst)
                    except OSError:
                        raise GitError(
                            self._get_symlink_error_message(), project=self.name
                        )
                else:
                    raise

    def _InitRemote(self):
        if self.remote.url:
            remote = self.GetRemote()
            remote.url = self.remote.url
            remote.pushUrl = self.remote.pushUrl
            remote.review = self.remote.review
            remote.projectname = self.name

            if self.worktree:
                remote.ResetFetch(mirror=False)
            else:
                remote.ResetFetch(mirror=True)
            remote.Save()

        # Disable auto-gc for depth=1 to prevent hangs during lazy fetches
        # inside git checkout for partial clones.
        effective_depth = (
            self.clone_depth or self.manifest.manifestProject.depth
        )
        if effective_depth == 1:
            self.config.SetBoolean("maintenance.auto", False)
            self.config.SetInt("gc.auto", 0)

    def _InitMRef(self):
        """Initialize the pseudo m/<manifest branch> ref."""
        if self.manifest.branch:
            if self.use_git_worktrees:
                # Set up the m/ space to point to the worktree-specific ref
                # space. We'll update the worktree-specific ref space on each
                # checkout.
                ref = R_M + self.manifest.branch
                if not self.bare_ref.symref(ref):
                    self.bare_git.symbolic_ref(
                        "-m",
                        "redirecting to worktree scope",
                        ref,
                        R_WORKTREE_M + self.manifest.branch,
                    )

                # We can't update this ref with git worktrees until it exists.
                # We'll wait until the initial checkout to set it.
                if not os.path.exists(self.worktree):
                    return

                base = R_WORKTREE_M
                active_git = self.work_git

                self._InitAnyMRef(HEAD, self.bare_git, detach=True)
            else:
                base = R_M
                active_git = self.bare_git

            self._InitAnyMRef(base + self.manifest.branch, active_git)

    def _InitMirrorHead(self):
        self._InitAnyMRef(HEAD, self.bare_git)

    def _InitAnyMRef(self, ref, active_git, detach=False):
        """Initialize |ref| in |active_git| to the value in the manifest.

        This points |ref| to the <project> setting in the manifest.

        Args:
            ref: The branch to update.
            active_git: The git repository to make updates in.
            detach: Whether to update target of symbolic refs, or overwrite the
                ref directly (and thus make it non-symbolic).
        """
        cur = self.bare_ref.symref(ref)

        if self.revisionId:
            if cur != "" or self.bare_ref.get(ref) != self.revisionId:
                msg = "manifest set to %s" % self.revisionId
                dst = self.revisionId + "^0"
                active_git.UpdateRef(ref, dst, message=msg, detach=True)
        else:
            remote = self.GetRemote()
            dst = remote.ToLocal(self.revisionExpr)
            if cur != dst:
                msg = "manifest set to %s" % self.revisionExpr
                if detach:
                    active_git.UpdateRef(ref, dst, message=msg, detach=True)
                else:
                    active_git.symbolic_ref("-m", msg, ref, dst)

    def _CheckDirReference(self, srcdir, destdir):
        # Git worktrees don't use symlinks to share at all.
        if self.use_git_worktrees:
            return

        for name in self.shareable_dirs:
            # Try to self-heal a bit in simple cases.
            dst_path = os.path.join(destdir, name)
            src_path = os.path.join(srcdir, name)

            dst = os.path.realpath(dst_path)
            if os.path.lexists(dst):
                src = os.path.realpath(src_path)
                # Fail if the links are pointing to the wrong place.
                if src != dst:
                    logger.error(
                        "error: %s is different in %s vs %s",
                        name,
                        destdir,
                        srcdir,
                    )
                    raise GitError(
                        "--force-sync not enabled; cannot overwrite a local "
                        "work tree. If you're comfortable with the "
                        "possibility of losing the work tree's git metadata,"
                        " use "
                        f"`repo sync --force-sync {self.RelPath(local=False)}` "
                        "to proceed.",
                        project=self.name,
                    )

    def _ReferenceGitDir(self, gitdir, dotgit, copy_all):
        """Update |dotgit| to reference |gitdir|, using symlinks where possible.

        Args:
            gitdir: The bare git repository. Must already be initialized.
            dotgit: The repository you would like to initialize.
            copy_all: If true, copy all remaining files from |gitdir| ->
                |dotgit|. This saves you the effort of initializing |dotgit|
                yourself.
        """
        symlink_dirs = self.shareable_dirs[:]
        to_symlink = symlink_dirs

        to_copy = []
        if copy_all:
            to_copy = platform_utils.listdir(gitdir)

        dotgit = os.path.realpath(dotgit)
        for name in set(to_copy).union(to_symlink):
            try:
                src = os.path.realpath(os.path.join(gitdir, name))
                dst = os.path.join(dotgit, name)

                if os.path.lexists(dst):
                    continue

                # If the source dir doesn't exist, create an empty dir.
                if name in symlink_dirs and not os.path.lexists(src):
                    os.makedirs(src)

                if name in to_symlink:
                    platform_utils.symlink(
                        os.path.relpath(src, os.path.dirname(dst)), dst
                    )
                elif copy_all and not platform_utils.islink(dst):
                    if platform_utils.isdir(src):
                        shutil.copytree(src, dst)
                    elif os.path.isfile(src):
                        shutil.copy(src, dst)

            except OSError as e:
                if e.errno == errno.EPERM:
                    raise DownloadError(self._get_symlink_error_message())
                else:
                    raise

    def _InitGitWorktree(self):
        """Init the project using git worktrees."""
        self.bare_git.worktree("prune")
        self.bare_git.worktree(
            "add",
            "-ff",
            "--checkout",
            "--detach",
            "--lock",
            self.worktree,
            self.GetRevisionId(),
        )

        # Rewrite the internal state files to use relative paths between the
        # checkouts & worktrees.
        dotgit = os.path.join(self.worktree, ".git")
        with open(dotgit) as fp:
            # Figure out the checkout->worktree path.
            setting = fp.read()
            assert setting.startswith("gitdir:")
            git_worktree_path = setting.split(":", 1)[1].strip()

        # `gitdir` maybe be either relative or absolute depending on the
        # behavior of the local copy of git, so only convert the path to
        # relative if it needs to be converted.
        if os.path.isabs(git_worktree_path):
            # Some platforms (e.g. Windows) won't let us update dotgit in situ
            # because of file permissions.  Delete it and recreate it from
            # scratch to avoid.
            platform_utils.remove(dotgit)
            # Use relative path from checkout->worktree & maintain Unix line
            # endings on all OS's to match git behavior.
            with open(dotgit, "w", newline="\n") as fp:
                print(
                    "gitdir:",
                    os.path.relpath(git_worktree_path, self.worktree),
                    file=fp,
                )
            # Use relative path from worktree->checkout & maintain Unix line
            # endings on all OS's to match git behavior.
            with open(
                os.path.join(git_worktree_path, "gitdir"), "w", newline="\n"
            ) as fp:
                print(os.path.relpath(dotgit, git_worktree_path), file=fp)

        self._InitMRef()

    def _InitWorkTree(self, force_sync=False, submodules=False):
        """Setup the worktree .git path.

        This is the user-visible path like src/foo/.git/.

        With non-git-worktrees, this will be a symlink to the .repo/projects/
        path. With git-worktrees, this will be a .git file using "gitdir: ..."
        syntax.

        Older checkouts had .git/ directories.  If we see that, migrate it.

        This also handles changes in the manifest.  Maybe this project was
        backed by "foo/bar" on the server, but now it's "new/foo/bar".  We have
        to update the path we point to under .repo/projects/ to match.
        """
        dotgit = os.path.join(self.worktree, ".git")

        # If bare checkout of the submodule is stored under the subproject dir,
        # migrate it.
        if self.parent:
            self._MigrateOldSubmoduleDir()

        # If using an old layout style (a directory), migrate it.
        if (
            not platform_utils.islink(dotgit)
            and platform_utils.isdir(dotgit)
            and not self.manifest.UseLocalGitDirs
        ):
            self._MigrateOldWorkTreeGitDir(dotgit, project=self.name)

        init_dotgit = not os.path.lexists(dotgit)
        if self.use_git_worktrees:
            if init_dotgit:
                self._InitGitWorktree()
                self._CopyAndLinkFiles()
        else:
            # Remove old directory symbolic links for submodules.
            if self.parent and platform_utils.islink(dotgit):
                platform_utils.remove(dotgit)
                init_dotgit = True

            if not init_dotgit:
                # See if the project has changed.
                self._removeBadGitDirLink(dotgit)

            if init_dotgit or not os.path.exists(dotgit):
                self._createDotGit(dotgit)

            if init_dotgit:
                self.work_git.UpdateRef(HEAD, self.GetRevisionId(), detach=True)

                # Finish checking out the worktree.
                cmd = ["read-tree", "--reset", "-u", "-v", HEAD]
                try:
                    if GitCommand(self, cmd).Wait() != 0:
                        raise GitError(
                            "Cannot initialize work tree for " + self.name,
                            project=self.name,
                        )
                except Exception as e:
                    # Something went wrong with read-tree (perhaps fetching
                    # missing blobs), so remove .git to avoid half initialized
                    # workspace from which repo can't recover on its own.
                    platform_utils.remove(dotgit)
                    raise e

                if submodules:
                    self._SyncSubmodules(quiet=True)
                self._CopyAndLinkFiles()

    def _createDotGit(self, dotgit):
        """Initialize .git path.

        For submodule projects, create a '.git' file using the gitfile
        mechanism, and for the rest, create a symbolic link.
        """
        if self.manifest.UseLocalGitDirs and os.path.normpath(
            self.gitdir
        ) == os.path.normpath(dotgit):
            return

        os.makedirs(self.worktree, exist_ok=True)
        if self.parent:
            _lwrite(
                dotgit,
                f"gitdir: {os.path.relpath(self.gitdir, self.worktree)}\n",
            )
        else:
            platform_utils.symlink(
                os.path.relpath(self.gitdir, self.worktree), dotgit
            )

    def _removeBadGitDirLink(self, dotgit):
        """Verify .git is initialized correctly, otherwise delete it."""
        if self.parent and os.path.isfile(dotgit):
            with open(dotgit) as fp:
                setting = fp.read()
            if not setting.startswith("gitdir:"):
                raise GitError(
                    f"'.git' in {self.worktree} must start with 'gitdir:'",
                    project=self.name,
                )
            gitdir = setting.split(":", 1)[1].strip()
            dotgit_path = os.path.normpath(os.path.join(self.worktree, gitdir))
        else:
            dotgit_path = os.path.realpath(dotgit)
        if os.path.realpath(self.gitdir) != dotgit_path:
            platform_utils.remove(dotgit)

    @classmethod
    def _MigrateOldWorkTreeGitDir(cls, dotgit, project=None):
        """Migrate the old worktree .git/ dir style to a symlink.

        This logic specifically only uses state from |dotgit| to figure out
        where to move content and not |self|.  This way if the backing project
        also changed places, we only do the .git/ dir to .git symlink migration
        here.  The path updates will happen independently.
        """
        # Figure out where in .repo/projects/ it's pointing to.
        gitdir = None
        for name in ("refs", "reftable", "objects"):
            path = os.path.join(dotgit, name)
            if os.path.islink(path):
                gitdir = os.path.dirname(os.path.realpath(path))
                break
        else:
            raise GitError(
                f"{dotgit}: unsupported checkout state", project=project
            )

        # Remove known symlink paths that exist in .repo/projects/.
        KNOWN_LINKS = {
            # go/keep-sorted start
            "config",
            "description",
            "hooks",
            "info",
            "logs",
            "objects",
            "packed-refs",
            "refs",
            "reftable",
            "rr-cache",
            "shallow",
            "svn",
            # go/keep-sorted end
        }
        # Paths that we know will be in both, but are safe to clobber in
        # .repo/projects/.
        SAFE_TO_CLOBBER = {
            "COMMIT_EDITMSG",
            "FETCH_HEAD",
            "HEAD",
            "gc.log",
            "gitk.cache",
            "index",
            "ORIG_HEAD",
        }

        # First see if we'd succeed before starting the migration.
        unknown_paths = []
        for name in platform_utils.listdir(dotgit):
            # Ignore all temporary/backup names.  These are common with vim &
            # emacs.
            if name.endswith("~") or (name[0] == "#" and name[-1] == "#"):
                continue

            dotgit_path = os.path.join(dotgit, name)
            if name in KNOWN_LINKS:
                if not platform_utils.islink(dotgit_path):
                    # In reftable format, refs and reftable can be directories.
                    if name in ("refs", "reftable") and platform_utils.isdir(
                        dotgit_path
                    ):
                        pass
                    else:
                        unknown_paths.append(
                            f"{dotgit_path}: should be a symlink"
                        )

            else:
                gitdir_path = os.path.join(gitdir, name)
                if name not in SAFE_TO_CLOBBER and os.path.exists(gitdir_path):
                    unknown_paths.append(
                        f"{dotgit_path}: unknown file; please file a bug"
                    )
        if unknown_paths:
            raise GitError(
                "Aborting migration: " + "\n".join(unknown_paths),
                project=project,
            )

        # Now walk the paths and sync the .git/ to .repo/projects/.
        for name in platform_utils.listdir(dotgit):
            dotgit_path = os.path.join(dotgit, name)

            # Ignore all temporary/backup names.  These are common with vim &
            # emacs.
            if name.endswith("~") or (name[0] == "#" and name[-1] == "#"):
                platform_utils.remove(dotgit_path)
            elif name in KNOWN_LINKS:
                if (
                    name in ("refs", "reftable")
                    and platform_utils.isdir(dotgit_path)
                    and not platform_utils.islink(dotgit_path)
                ):
                    platform_utils.rmtree(dotgit_path)
                else:
                    platform_utils.remove(dotgit_path)
            else:
                gitdir_path = os.path.join(gitdir, name)
                platform_utils.remove(gitdir_path, missing_ok=True)
                platform_utils.rename(dotgit_path, gitdir_path)

        # Now that the dir should be empty, clear it out, and symlink it over.
        platform_utils.rmdir(dotgit)
        platform_utils.symlink(
            os.path.relpath(gitdir, os.path.dirname(os.path.realpath(dotgit))),
            dotgit,
        )

    def _MigrateOldSubmoduleDir(self):
        """Move the old bare checkout in 'subprojects' to 'modules'
        as bare checkouts of submodules are now in 'modules' dir.
        """
        subprojects = os.path.join(self.parent.gitdir, "subprojects")
        if not platform_utils.isdir(subprojects):
            return

        modules = os.path.join(self.parent.gitdir, "modules")
        old = self.gitdir
        new = os.path.splitext(self.gitdir.replace(subprojects, modules))[0]

        if all(map(platform_utils.isdir, [old, new])):
            platform_utils.rmtree(old, ignore_errors=True)
        else:
            os.makedirs(modules, exist_ok=True)
            platform_utils.rename(old, new)
        self.gitdir = new
        self.UpdatePaths(self.relpath, self.worktree, self.gitdir, self.objdir)
        if platform_utils.isdir(subprojects) and not os.listdir(subprojects):
            platform_utils.rmtree(subprojects, ignore_errors=True)

    def _get_symlink_error_message(self):
        if platform_utils.isWindows():
            return (
                "Unable to create symbolic link. Please re-run the command as "
                "Administrator, or see "
                "https://github.com/git-for-windows/git/wiki/Symbolic-Links "
                "for other options."
            )
        return "filesystem must support symlinks"

    def _revlist(self, *args, **kw):
        a = []
        a.extend(args)
        a.append("--")
        return self.work_git.rev_list(*a, **kw)

    @property
    def _allrefs(self):
        return self.bare_ref.all

    def _getLogs(
        self, rev1, rev2, oneline=False, color=True, pretty_format=None
    ):
        """Get logs between two revisions of this project."""
        comp = ".."
        if rev1:
            revs = [rev1]
            if rev2:
                revs.extend([comp, rev2])
            cmd = ["log", "".join(revs)]
            out = DiffColoring(self.config)
            if out.is_on and color:
                cmd.append("--color")
            if pretty_format is not None:
                cmd.append("--pretty=format:%s" % pretty_format)
            if oneline:
                cmd.append("--oneline")

            try:
                log = GitCommand(
                    self, cmd, capture_stdout=True, capture_stderr=True
                )
                if log.Wait() == 0:
                    return log.stdout
            except GitError:
                # worktree may not exist if groups changed for example. In that
                # case, try in gitdir instead.
                if not os.path.exists(self.worktree):
                    return self.bare_git.log(*cmd[1:])
                else:
                    raise
        return None

    def getAddedAndRemovedLogs(
        self, toProject, oneline=False, color=True, pretty_format=None
    ):
        """Get the list of logs from this revision to given revisionId"""
        logs = {}
        selfId = self.GetRevisionId(self._allrefs)
        toId = toProject.GetRevisionId(toProject._allrefs)

        logs["added"] = self._getLogs(
            selfId,
            toId,
            oneline=oneline,
            color=color,
            pretty_format=pretty_format,
        )
        logs["removed"] = self._getLogs(
            toId,
            selfId,
            oneline=oneline,
            color=color,
            pretty_format=pretty_format,
        )
        return logs

    class _GitGetByExec:
        def __init__(self, project, bare, gitdir):
            self._project = project
            self._bare = bare
            self._gitdir = gitdir

        # __getstate__ and __setstate__ are required for pickling because
        # __getattr__ exists.
        def __getstate__(self):
            return (self._project, self._bare, self._gitdir)

        def __setstate__(self, state):
            self._project, self._bare, self._gitdir = state

        def LsOthers(self):
            p = GitCommand(
                self._project,
                ["ls-files", "-z", "--others", "--exclude-standard"],
                bare=False,
                gitdir=self._gitdir,
                capture_stdout=True,
                capture_stderr=True,
            )
            if p.Wait() == 0:
                out = p.stdout
                if out:
                    # Backslash is not anomalous.
                    return out[:-1].split("\0")
            return []

        def DiffZ(self, name, *args):
            cmd = [name]
            cmd.append("-z")
            cmd.append("--ignore-submodules")
            cmd.extend(args)
            p = GitCommand(
                self._project,
                cmd,
                gitdir=self._gitdir,
                bare=False,
                capture_stdout=True,
                capture_stderr=True,
            )
            p.Wait()
            r = {}
            out = p.stdout
            if out:
                out = iter(out[:-1].split("\0"))
                while out:
                    try:
                        info = next(out)
                        path = next(out)
                    except StopIteration:
                        break

                    class _Info:
                        def __init__(self, path, omode, nmode, oid, nid, state):
                            self.path = path
                            self.src_path = None
                            self.old_mode = omode
                            self.new_mode = nmode
                            self.old_id = oid
                            self.new_id = nid

                            if len(state) == 1:
                                self.status = state
                                self.level = None
                            else:
                                self.status = state[:1]
                                self.level = state[1:]
                                while self.level.startswith("0"):
                                    self.level = self.level[1:]

                    info = info[1:].split(" ")
                    info = _Info(path, *info)
                    if info.status in ("R", "C"):
                        info.src_path = info.path
                        info.path = next(out)
                    r[info.path] = info
            return r

        def GetDotgitPath(self, subpath=None):
            """Return the full path to the .git dir.

            As a convenience, append |subpath| if provided.
            """
            if self._bare:
                dotgit = self._gitdir
            else:
                dotgit = os.path.join(self._project.worktree, ".git")
                if os.path.isfile(dotgit):
                    # Git worktrees use a "gitdir:" syntax to point to the
                    # scratch space.
                    with open(dotgit) as fp:
                        setting = fp.read()
                    assert setting.startswith("gitdir:")
                    gitdir = setting.split(":", 1)[1].strip()
                    dotgit = os.path.normpath(
                        os.path.join(self._project.worktree, gitdir)
                    )

            return dotgit if subpath is None else os.path.join(dotgit, subpath)

        @staticmethod
        def _ParseHead(line: str) -> Optional[str]:
            """Parse the content of a .git/HEAD file.

            Handles both symbolic refs (e.g. 'ref: refs/heads/...') and raw
            commit IDs (40-hex SHA-1 or 64-hex SHA-256).

            Returns:
                The ref name (e.g. 'refs/heads/main') or lowercase commit hash
                if valid, or None if empty or invalid.
            """
            line = line.strip()
            if line.startswith("ref:"):
                ref = line[4:].strip()
                # Ensure the ref is not empty, pure whitespace, or the
                # "refs/heads/.invalid" placeholder used for unborn branches,
                # empty repositories, or when the reftables backend is used
                # (which will be the default in Git 3.0).
                if not ref or ref == R_HEADS + ".invalid":
                    return None
                return ref
            else:
                # Normalize commit IDs to canonical lowercase hexadecimal,
                # matching the output format of `git rev-parse`.
                line_lower = line.lower()
                if IsId(line_lower):
                    return line_lower
            return None

        def GetHead(self):
            """Return the ref that HEAD points to."""
            path = None
            try:
                # Catch AssertionError raised by GetDotgitPath when worktree
                # .git pointer file is malformed (e.g. missing 'gitdir:').
                path = self.GetDotgitPath(subpath=HEAD)
                if not platform_utils.islink(path):
                    with open(
                        path, "r", encoding="utf-8", errors="replace"
                    ) as fd:
                        ref = self._ParseHead(fd.readline())
                    if ref:
                        return ref
            except (OSError, AssertionError):
                pass

            try:
                return self.symbolic_ref("-q", HEAD, log_as_error=False)
            except GitError:
                pass

            try:
                # If symbolic-ref fails, try to treat as detached HEAD.
                return self.rev_parse(HEAD)
            except GitError as e:
                logger.warning(
                    "project %s: unparseable HEAD; trying to recover.\n"
                    "Check that HEAD ref in .git/HEAD is valid. The error "
                    "was: %s",
                    self._project.RelPath(local=False),
                    e,
                )

                # Fallback to direct file reading for compatibility with broken
                # repos, e.g. if HEAD points to an unborn branch.
                if not path:
                    raise NoManifestException(
                        self._project.RelPath(local=False), str(e)
                    )
                try:
                    with open(
                        path, "r", encoding="utf-8", errors="replace"
                    ) as fd:
                        ref = self._ParseHead(fd.readline())
                except OSError:
                    raise NoManifestException(
                        self._project.RelPath(local=False), str(e)
                    )
                if not ref:
                    raise NoManifestException(
                        self._project.RelPath(local=False), str(e)
                    )
                return ref

        def ResolveCommit(self, revision: str) -> str:
            """Resolve |revision| to a commit without option ambiguity."""
            cmdv = ["--verify", "--quiet"]
            if git_require((2, 30, 0)):
                cmdv.append("--end-of-options")
            elif revision.startswith("-"):
                raise GitError(
                    f"invalid revision: {revision}",
                    project=self._project.name,
                )
            cmdv.append(f"{revision}^{{commit}}")
            return self.rev_parse(*cmdv, log_as_error=False)

        def SetHead(self, ref, message=None):
            cmdv = []
            if message is not None:
                cmdv.extend(["-m", message])
            cmdv.append(HEAD)
            cmdv.append(ref)
            self.symbolic_ref(*cmdv)

        def DetachHead(self, new, message=None):
            cmdv = ["--no-deref"]
            if message is not None:
                cmdv.extend(["-m", message])
            cmdv.append(HEAD)
            cmdv.append(new)
            self.update_ref(*cmdv)

        def UpdateRef(self, name, new, old=None, message=None, detach=False):
            cmdv = []
            if message is not None:
                cmdv.extend(["-m", message])
            if detach:
                cmdv.append("--no-deref")
            cmdv.append(name)
            cmdv.append(new)
            if old is not None:
                cmdv.append(old)
            self.update_ref(*cmdv)

        def DeleteRef(self, name, old=None):
            if not old:
                old = self.rev_parse(name)
            self.update_ref("-d", name, old)
            self._project.bare_ref.deleted(name)

        def rev_list(self, *args, log_as_error=True, **kw):
            if "format" in kw:
                cmdv = ["log", "--pretty=format:%s" % kw["format"]]
            else:
                cmdv = ["rev-list"]
            cmdv.extend(args)
            p = GitCommand(
                self._project,
                cmdv,
                bare=self._bare,
                gitdir=self._gitdir,
                capture_stdout=True,
                capture_stderr=True,
                verify_command=True,
                log_as_error=log_as_error,
            )
            p.Wait()
            return p.stdout.splitlines()

        def __getattr__(self, name):
            """Allow arbitrary git commands using pythonic syntax.

            This allows you to do things like:
                git_obj.rev_parse('HEAD')

            Since we don't have a 'rev_parse' method defined, the __getattr__
            will run.  We'll replace the '_' with a '-' and try to run a git
            command. Any other positional arguments will be passed to the git
            command, and the following keyword arguments are supported:
                config: An optional dict of git config options to be passed with
                    '-c'.

            Args:
                name: The name of the git command to call.  Any '_' characters
                    will be replaced with '-'.

            Returns:
                A callable object that will try to call git with the named
                command.
            """
            name = name.replace("_", "-")

            def runner(*args, log_as_error=True, **kwargs):
                cmdv = []
                config = kwargs.pop("config", None)
                for k in kwargs:
                    raise TypeError(
                        f"{name}() got an unexpected keyword argument {k!r}"
                    )
                if config is not None:
                    for k, v in config.items():
                        cmdv.append("-c")
                        cmdv.append(f"{k}={v}")
                cmdv.append(name)
                cmdv.extend(args)
                p = GitCommand(
                    self._project,
                    cmdv,
                    bare=self._bare,
                    gitdir=self._gitdir,
                    capture_stdout=True,
                    capture_stderr=True,
                    verify_command=True,
                    log_as_error=log_as_error,
                )
                p.Wait()
                r = p.stdout
                if r.endswith("\n") and r.index("\n") == len(r) - 1:
                    return r[:-1]
                return r

            return runner


class LocalSyncFail(RepoError):
    """Default error when there is an Sync_LocalHalf error."""


class _PriorSyncFailedError(LocalSyncFail):
    def __str__(self):
        return "prior sync failed; rebase still in progress"


class _DirtyError(LocalSyncFail):
    def __str__(self):
        return "contains uncommitted changes"


class _InfoMessage:
    def __init__(self, project, text):
        self.project = project
        self.text = text

    def Print(self, syncbuf):
        syncbuf.out.info(
            "%s/: %s", self.project.RelPath(local=False), self.text
        )
        syncbuf.out.nl()


class _Failure:
    def __init__(self, project, why):
        self.project = project
        self.why = why

    def Print(self, syncbuf):
        syncbuf.out.fail(
            "error: %s/: %s", self.project.RelPath(local=False), str(self.why)
        )
        syncbuf.out.nl()


class _Later:
    def __init__(self, project, action, quiet):
        self.project = project
        self.action = action
        self.quiet = quiet

    def Run(self, syncbuf):
        out = syncbuf.out
        if not self.quiet:
            out.project("project %s/", self.project.RelPath(local=False))
            out.nl()
        try:
            self.action()
            if not self.quiet:
                out.nl()
            return True
        except GitError as e:
            syncbuf.fail(self.project, e)
            out.nl()
            return False


class _SyncColoring(Coloring):
    def __init__(self, config):
        super().__init__(config, "reposync")
        self.project = self.printer("header", attr="bold")
        self.info = self.printer("info")
        self.fail = self.printer("fail", fg="red")


class SyncBuffer:
    def __init__(self, config, detach_head=False):
        self._messages = []

        # Failures that have not yet been printed. Cleared after printing.
        self._pending_failures = []
        # A persistent record of all failures during the buffer's lifetime.
        self._all_failures = []

        self._later_queue1 = []
        self._later_queue2 = []

        self.out = _SyncColoring(config)
        self.out.redirect(sys.stderr)

        self.detach_head = detach_head
        self.clean = True
        self.recent_clean = True

    def info(self, project, fmt, *args):
        self._messages.append(_InfoMessage(project, fmt % args))

    def fail(self, project, err=None):
        failure = _Failure(project, err)
        self._pending_failures.append(failure)
        self._all_failures.append(failure)
        self._MarkUnclean()

    def later1(self, project, what, quiet):
        self._later_queue1.append(_Later(project, what, quiet))

    def later2(self, project, what, quiet):
        self._later_queue2.append(_Later(project, what, quiet))

    def Finish(self):
        self._PrintMessages()
        self._RunLater()
        self._PrintMessages()
        return self.clean

    def Recently(self):
        recent_clean = self.recent_clean
        self.recent_clean = True
        return recent_clean

    @property
    def errors(self):
        """Returns a list of all exceptions accumulated in the buffer."""
        return [f.why for f in self._all_failures if f.why]

    def _MarkUnclean(self):
        self.clean = False
        self.recent_clean = False

    def _RunLater(self):
        for q in ["_later_queue1", "_later_queue2"]:
            if not self._RunQueue(q):
                return

    def _RunQueue(self, queue):
        for m in getattr(self, queue):
            if not m.Run(self):
                self._MarkUnclean()
                return False
        setattr(self, queue, [])
        return True

    def _PrintMessages(self):
        if self._messages or self._pending_failures:
            if os.isatty(2):
                self.out.write(progress.CSI_ERASE_LINE)
            self.out.write("\r")

        for m in self._messages:
            m.Print(self)
        for m in self._pending_failures:
            m.Print(self)

        self._messages = []
        self._pending_failures = []


@functools.lru_cache(maxsize=None)
def _DefaultBranchFallback() -> str:
    """Return the ref to use when remote default branch can't be resolved."""

    def _git(args: List[str]) -> str:
        p = GitCommand(
            None,
            args,
            capture_stdout=True,
            capture_stderr=True,
            log_as_error=False,
        )
        return p.stdout.strip() if p.Wait() == 0 else ""

    branch = ""
    if git_require((2, 35, 0)):
        branch = _git(["var", "GIT_DEFAULT_BRANCH"])
    if not branch:
        branch = _git(["config", "--get", "init.defaultBranch"])
    return f"refs/heads/{branch or 'master'}"


class MetaProject(Project):
    """A special project housed under .repo."""

    def __init__(self, manifest, name, gitdir, worktree):
        Project.__init__(
            self,
            manifest=manifest,
            name=name,
            gitdir=gitdir,
            objdir=gitdir,
            worktree=worktree,
            remote=RemoteSpec("origin"),
            relpath=".repo/%s" % name,
            revisionExpr=_DefaultBranchFallback(),
            revisionId=None,
            groups=None,
        )

    def PreSync(self):
        if self.Exists:
            cb = self.CurrentBranch
            if cb:
                base = self.GetBranch(cb).merge
                if base:
                    self.revisionExpr = base
                    self.revisionId = None

    def _UseSuperprojectForUpstream(
        self, use_superproject: Optional[bool] = None
    ) -> bool:
        # MetaProjects (the manifest repo and repo itself) never
        # participate in a superproject relationship. Returning False
        # here also avoids loading the manifest during `repo init`,
        # before manifest.xml has been linked into .repo/.
        return False

    def _GetUpstreamFallback(self) -> Optional[str]:
        # MetaProjects (the manifest repo and repo itself) do not have
        # defaults in a manifest. Returning None here also avoids
        # loading the manifest during `repo init`, before manifest.xml
        # has been linked into .repo/.
        return None

    def _SharingProjectHasShallow(self) -> bool:
        # MetaProjects (the manifest repo and repo itself) are never
        # shared with other projects in the manifest. Returning False
        # here also avoids loading the manifest during `repo init`,
        # before manifest.xml has been linked into .repo/.
        return False

    def _ShouldVerifyUpstream(
        self,
        use_superproject: Optional[bool] = None,
        depth: Optional[int] = None,
    ) -> bool:
        """MetaProjects (manifest repo and repo itself) do not verify upstream
        ancestry.
        """
        return False

    @property
    def HasChanges(self):
        """Has the remote received new commits not yet checked out?"""
        if not self.remote or not self.revisionExpr:
            return False

        all_refs = self.bare_ref.all
        revid = self.GetRevisionId(all_refs)
        head = self._GetHead()
        if head and head.startswith(R_HEADS):
            try:
                head = all_refs[head]
            except KeyError:
                head = None

        if revid == head:
            return False
        elif self._revlist("-1", not_rev(HEAD), revid):
            return True
        return False


class RepoProject(MetaProject):
    """The MetaProject for repo itself."""

    @property
    def LastFetch(self):
        try:
            fh = os.path.join(self.gitdir, "FETCH_HEAD")
            return os.path.getmtime(fh)
        except OSError:
            return 0


class ManifestProject(MetaProject):
    """The MetaProject for manifests."""

    def MetaBranchSwitch(self, submodules=False, verbose=False):
        """Prepare for manifest branch switch."""

        # detach and delete manifest branch, allowing a new
        # branch to take over
        syncbuf = SyncBuffer(self.config, detach_head=True)
        self.Sync_LocalHalf(syncbuf, submodules=submodules, verbose=verbose)
        syncbuf.Finish()

        return (
            GitCommand(
                self,
                ["update-ref", "-d", "refs/heads/default"],
                capture_stdout=True,
                capture_stderr=True,
            ).Wait()
            == 0
        )

    @property
    def standalone_manifest_url(self):
        """The URL of the standalone manifest, or None."""
        return self.config.GetString("manifest.standalone")

    @property
    def manifest_groups(self):
        """The manifest groups string."""
        return self.config.GetString("manifest.groups")

    @property
    def reference(self):
        """The --reference for this manifest."""
        return self.config.GetString("repo.reference")

    @property
    def dissociate(self):
        """Whether to dissociate."""
        return self.config.GetBoolean("repo.dissociate")

    @property
    def archive(self):
        """Whether we use archive."""
        return self.config.GetBoolean("repo.archive")

    @property
    def mirror(self):
        """Whether we use mirror."""
        return self.config.GetBoolean("repo.mirror")

    @property
    def use_worktree(self):
        """Whether we use worktree."""
        return self.config.GetBoolean("repo.worktree")

    @property
    def use_local_gitdirs(self):
        """Whether we use local gitdirs."""
        return self.config.GetBoolean("repo.uselocalgitdirs")

    @property
    def fetch_cmd(self):
        """The fetch command to use."""
        return self.config.GetString("repo.fetchcmd")

    @property
    def clone_bundle(self):
        """Whether we use clone_bundle."""
        return self.config.GetBoolean("repo.clonebundle")

    @property
    def submodules(self):
        """Whether we use submodules."""
        return self.config.GetBoolean("repo.submodules")

    @property
    def git_lfs(self):
        """Whether we use git_lfs."""
        return self.config.GetBoolean("repo.git-lfs")

    @property
    def use_superproject(self):
        """Whether we use superproject."""
        return self.config.GetBoolean("repo.superproject")

    @property
    def partial_clone(self):
        """Whether this is a partial clone."""
        return self.config.GetBoolean("repo.partialclone")

    @property
    def depth(self):
        """Partial clone depth."""
        return self.config.GetInt("repo.depth")

    @property
    def clone_filter(self):
        """The clone filter."""
        return self.config.GetString("repo.clonefilter")

    @property
    def partial_clone_exclude(self):
        """Partial clone exclude string"""
        return self.config.GetString("repo.partialcloneexclude")

    @property
    def clone_filter_for_depth(self):
        """Replace shallow clone with partial clone."""
        return self.config.GetString("repo.clonefilterfordepth")

    @property
    def manifest_platform(self):
        """The --platform argument from `repo init`."""
        return self.config.GetString("manifest.platform")

    @property
    def _platform_name(self):
        """Return the name of the platform."""
        return platform.system().lower()

    def SyncWithPossibleInit(
        self,
        submanifest,
        verbose=False,
        current_branch_only=False,
        tags="",
        git_event_log=None,
    ):
        """Sync a manifestProject, possibly for the first time.

        Call Sync() with arguments from the most recent `repo init`.  If this is
        a new sub manifest, then inherit options from the parent's
        manifestProject.

        This is used by subcmds.Sync() to do an initial download of new sub
        manifests.

        Args:
            submanifest: an XmlSubmanifest, the submanifest to re-sync.
            verbose: a boolean, whether to show all output, rather than only
                errors.
            current_branch_only: a boolean, whether to only fetch the current
                manifest branch from the server.
            tags: a boolean, whether to fetch tags.
            git_event_log: an EventLog, for git tracing.
        """
        # TODO(lamontjones): when refactoring sync (and init?) consider how to
        # better get the init options that we should use for new submanifests
        # that are added when syncing an existing workspace.
        git_event_log = git_event_log or EventLog()
        spec = submanifest.ToSubmanifestSpec()
        # Use the init options from the existing manifestProject, or the parent
        # if it doesn't exist.
        #
        # Today, we only support changing manifest_groups on the sub-manifest,
        # with no supported-for-the-user way to change the other arguments from
        # those specified by the outermost manifest.
        #
        # TODO(lamontjones): determine which of these should come from the
        # outermost manifest and which should come from the parent manifest.
        mp = self if self.Exists else submanifest.parent.manifestProject
        return self.Sync(
            manifest_url=spec.manifestUrl,
            manifest_branch=spec.revision,
            standalone_manifest=mp.standalone_manifest_url,
            groups=mp.manifest_groups,
            platform=mp.manifest_platform,
            mirror=mp.mirror,
            dissociate=mp.dissociate,
            reference=mp.reference,
            worktree=mp.use_worktree,
            submodules=mp.submodules,
            archive=mp.archive,
            partial_clone=mp.partial_clone,
            clone_filter=mp.clone_filter,
            partial_clone_exclude=mp.partial_clone_exclude,
            clone_bundle=mp.clone_bundle,
            git_lfs=mp.git_lfs,
            use_superproject=mp.use_superproject,
            verbose=verbose,
            current_branch_only=current_branch_only,
            tags=tags,
            depth=mp.depth,
            git_event_log=git_event_log,
            manifest_name=spec.manifestName,
            this_manifest_only=True,
            outer_manifest=False,
            clone_filter_for_depth=mp.clone_filter_for_depth,
        )

    def Sync(
        self,
        _kwargs_only=(),
        manifest_url="",
        manifest_branch=None,
        standalone_manifest=False,
        groups="",
        mirror=False,
        reference="",
        dissociate=False,
        worktree=False,
        submodules=False,
        archive=False,
        partial_clone=None,
        depth=None,
        clone_filter="blob:none",
        partial_clone_exclude=None,
        clone_bundle=None,
        git_lfs=None,
        use_superproject=None,
        verbose=False,
        current_branch_only=False,
        git_event_log=None,
        platform="",
        manifest_name="default.xml",
        tags="",
        this_manifest_only=False,
        outer_manifest=True,
        clone_filter_for_depth=None,
        use_local_gitdirs=False,
    ):
        """Sync the manifest and all submanifests.

        Args:
            manifest_url: a string, the URL of the manifest project.
            manifest_branch: a string, the manifest branch to use.
            standalone_manifest: a boolean, whether to store the manifest as a
                static file.
            groups: a string, restricts the checkout to projects with the
                specified groups.
            mirror: a boolean, whether to create a mirror of the remote
                repository.
            reference: a string, location of a repo instance to use as a
                reference.
            dissociate: a boolean, whether to dissociate from reference mirrors
                after clone.
            worktree: a boolean, whether to use git-worktree to manage projects.
            submodules: a boolean, whether sync submodules associated with the
                manifest project.
            archive: a boolean, whether to checkout each project as an archive.
                See git-archive.
            partial_clone: a boolean, whether to perform a partial clone.
            depth: an int, how deep of a shallow clone to create.
            clone_filter: a string, filter to use with partial_clone.
            partial_clone_exclude : a string, comma-delimeted list of project
                names to exclude from partial clone.
            clone_bundle: a boolean, whether to enable /clone.bundle on
                HTTP/HTTPS.
            git_lfs: a boolean, whether to enable git LFS support.
            use_superproject: a boolean, whether to use the manifest
                superproject to sync projects.
            verbose: a boolean, whether to show all output, rather than only
                errors.
            current_branch_only: a boolean, whether to only fetch the current
                manifest branch from the server.
            platform: a string, restrict the checkout to projects with the
                specified platform group.
            git_event_log: an EventLog, for git tracing.
            tags: a boolean, whether to fetch tags.
            manifest_name: a string, the name of the manifest file to use.
            this_manifest_only: a boolean, whether to only operate on the
                current sub manifest.
            outer_manifest: a boolean, whether to start at the outermost
                manifest.
            clone_filter_for_depth: a string, when specified replaces shallow
                clones with partial.

        Returns:
            a boolean, whether the sync was successful.
        """
        assert _kwargs_only == (), "Sync only accepts keyword arguments."

        groups = groups or self.manifest.GetDefaultGroupsStr(
            with_platform=False
        )
        platform = platform or "auto"
        git_event_log = git_event_log or EventLog()
        if outer_manifest and self.manifest.is_submanifest:
            # In a multi-manifest checkout, use the outer manifest unless we are
            # told not to.
            return self.client.outer_manifest.manifestProject.Sync(
                manifest_url=manifest_url,
                manifest_branch=manifest_branch,
                standalone_manifest=standalone_manifest,
                groups=groups,
                platform=platform,
                mirror=mirror,
                dissociate=dissociate,
                reference=reference,
                worktree=worktree,
                submodules=submodules,
                archive=archive,
                partial_clone=partial_clone,
                clone_filter=clone_filter,
                partial_clone_exclude=partial_clone_exclude,
                clone_bundle=clone_bundle,
                git_lfs=git_lfs,
                use_superproject=use_superproject,
                verbose=verbose,
                current_branch_only=current_branch_only,
                tags=tags,
                depth=depth,
                git_event_log=git_event_log,
                manifest_name=manifest_name,
                this_manifest_only=this_manifest_only,
                outer_manifest=False,
            )

        # If repo has already been initialized, we take -u with the absence of
        # --standalone-manifest to mean "transition to a standard repo set up",
        # which necessitates starting fresh.
        # If --standalone-manifest is set, we always tear everything down and
        # start anew.
        if self.Exists:
            was_standalone_manifest = self.config.GetString(
                "manifest.standalone"
            )
            if was_standalone_manifest and not manifest_url:
                logger.error(
                    "fatal: repo was initialized with a standlone manifest, "
                    "cannot be re-initialized without --manifest-url/-u"
                )
                return False

            if standalone_manifest or (
                was_standalone_manifest and manifest_url
            ):
                self.config.ClearCache()
                if self.gitdir and os.path.exists(self.gitdir):
                    platform_utils.rmtree(self.gitdir)
                if self.worktree and os.path.exists(self.worktree):
                    platform_utils.rmtree(self.worktree)

        is_new = not self.Exists
        if is_new:
            if not manifest_url:
                logger.error("fatal: manifest url is required.")
                return False

            if verbose:
                print(
                    "Downloading manifest from %s"
                    % (GitConfig.ForUser().UrlInsteadOf(manifest_url),),
                    file=sys.stderr,
                )

            # The manifest project object doesn't keep track of the path on the
            # server where this git is located, so let's save that here.
            mirrored_manifest_git = None
            if reference:
                manifest_git_path = urllib.parse.urlparse(manifest_url).path[1:]
                mirrored_manifest_git = os.path.join(
                    reference, manifest_git_path
                )
                if not mirrored_manifest_git.endswith(".git"):
                    mirrored_manifest_git += ".git"
                if not os.path.exists(mirrored_manifest_git):
                    mirrored_manifest_git = os.path.join(
                        reference, ".repo/manifests.git"
                    )

            self._InitGitDir(mirror_git=mirrored_manifest_git)

        # If standalone_manifest is set, mark the project as "standalone" --
        # we'll still do much of the manifests.git set up, but will avoid actual
        # syncs to a remote.
        if standalone_manifest:
            self.config.SetString("manifest.standalone", manifest_url)
        elif not manifest_url and not manifest_branch:
            # If -u is set and --standalone-manifest is not, then we're not in
            # standalone mode. Otherwise, use config to infer what we were in
            # the last init.
            standalone_manifest = bool(
                self.config.GetString("manifest.standalone")
            )
        if not standalone_manifest:
            self.config.SetString("manifest.standalone", None)

        self._ConfigureDepth(depth)

        # Set the remote URL before the remote branch as we might need it below.
        if manifest_url:
            r = self.GetRemote()
            r.url = manifest_url
            r.ResetFetch()
            r.Save()

        if not standalone_manifest:
            if manifest_branch:
                if manifest_branch == "HEAD":
                    manifest_branch = self.ResolveRemoteHead()
                    if manifest_branch is None:
                        logger.error("fatal: unable to resolve HEAD")
                        return False
                self.revisionExpr = manifest_branch
            else:
                if is_new:
                    default_branch = self.ResolveRemoteHead()
                    if default_branch is None:
                        # If the remote doesn't have HEAD configured, fall back
                        # to whatever git uses as its default branch.
                        default_branch = _DefaultBranchFallback()
                    self.revisionExpr = default_branch
                else:
                    self.PreSync()

        groups = re.split(r"[,\s]+", groups or "")
        all_platforms = ["linux", "darwin", "windows"]
        platformize = lambda x: "platform-" + x
        if platform == "auto":
            if not mirror and not self.mirror:
                groups.append(platformize(self._platform_name))
        elif platform == "all":
            groups.extend(map(platformize, all_platforms))
        elif platform in all_platforms:
            groups.append(platformize(platform))
        elif platform != "none":
            logger.error("fatal: invalid platform flag", file=sys.stderr)
            return False
        self.config.SetString("manifest.platform", platform)

        groups = [x for x in groups if x]
        groupstr = ",".join(groups)
        if (
            platform == "auto"
            and groupstr == self.manifest.GetDefaultGroupsStr()
        ):
            groupstr = None
        self.config.SetString("manifest.groups", groupstr)

        if reference:
            self.config.SetString("repo.reference", reference)

        if dissociate:
            self.config.SetBoolean("repo.dissociate", dissociate)

        if worktree:
            if mirror:
                logger.error("fatal: --mirror and --worktree are incompatible")
                return False
            if submodules:
                logger.error(
                    "fatal: --submodules and --worktree are incompatible"
                )
                return False
            self.config.SetBoolean("repo.worktree", worktree)
            if is_new:
                self.use_git_worktrees = True
            logger.warning("warning: --worktree is experimental!")

        if use_local_gitdirs:
            if mirror:
                logger.error(
                    "fatal: --mirror and --use-local-gitdirs are incompatible"
                )
                return False

            if worktree:
                logger.error(
                    "fatal: --worktree and --use-local-gitdirs are incompatible"
                )
                return False

            if archive:
                logger.error(
                    "fatal: --archive and --use-local-gitdirs are incompatible"
                )
                return False

            if not is_new and not self.use_local_gitdirs:
                logger.error(
                    "fatal: --use-local-gitdirs is only supported when "
                    "initializing a new workspace."
                )
                return False

            self.config.SetBoolean("repo.uselocalgitdirs", use_local_gitdirs)

        if self.fetch_cmd and not (use_local_gitdirs or self.use_local_gitdirs):
            logger.error(
                "fatal: repo.fetchcmd is set but repo.uselocalgitdirs is "
                "not enabled"
            )
            return False

        if archive:
            if is_new:
                self.config.SetBoolean("repo.archive", archive)
            else:
                logger.error(
                    "fatal: --archive is only supported when initializing a "
                    "new workspace."
                )
                logger.error(
                    "Either delete the .repo folder in this workspace, or "
                    "initialize in another location."
                )
                return False

        if mirror:
            if is_new:
                self.config.SetBoolean("repo.mirror", mirror)
            else:
                logger.error(
                    "fatal: --mirror is only supported when initializing a new "
                    "workspace."
                )
                logger.error(
                    "Either delete the .repo folder in this workspace, or "
                    "initialize in another location."
                )
                return False

        if partial_clone is not None:
            if mirror:
                logger.error(
                    "fatal: --mirror and --partial-clone are mutually "
                    "exclusive"
                )
                return False
            self.config.SetBoolean("repo.partialclone", partial_clone)
            if clone_filter:
                self.config.SetString("repo.clonefilter", clone_filter)
        elif self.partial_clone:
            clone_filter = self.clone_filter
        else:
            clone_filter = None

        if partial_clone_exclude is not None:
            self.config.SetString(
                "repo.partialcloneexclude", partial_clone_exclude
            )

        if clone_bundle is None:
            clone_bundle = False if partial_clone else True
        else:
            self.config.SetBoolean("repo.clonebundle", clone_bundle)

        if submodules:
            self.config.SetBoolean("repo.submodules", submodules)

        if git_lfs is not None:
            if git_lfs:
                git_require((2, 17, 0), fail=True, msg="Git LFS support")

            self.config.SetBoolean("repo.git-lfs", git_lfs)
            if not is_new:
                logger.warning(
                    "warning: Changing --git-lfs settings will only affect new "
                    "project checkouts.\n"
                    "         Existing projects will require manual updates.\n"
                )

        if clone_filter_for_depth is not None:
            self.ConfigureCloneFilterForDepth(clone_filter_for_depth)

        if use_superproject is not None:
            self.config.SetBoolean("repo.superproject", use_superproject)

        if not standalone_manifest:
            success = self.Sync_NetworkHalf(
                is_new=is_new,
                quiet=not verbose,
                verbose=verbose,
                clone_bundle=clone_bundle,
                use_superproject=use_superproject,
                current_branch_only=current_branch_only,
                tags=tags,
                submodules=submodules,
                clone_filter=clone_filter,
                partial_clone_exclude=self.manifest.PartialCloneExclude,
                clone_filter_for_depth=self.manifest.CloneFilterForDepth,
            ).success
            if not success:
                r = self.GetRemote()
                logger.error("fatal: cannot obtain manifest %s", r.url)

                # Better delete the manifest git dir if we created it; otherwise
                # next time (when user fixes problems) we won't go through the
                # "is_new" logic.
                if is_new:
                    platform_utils.rmtree(self.gitdir)
                return False

            if manifest_branch:
                self.MetaBranchSwitch(submodules=submodules, verbose=verbose)

            syncbuf = SyncBuffer(self.config)
            self.Sync_LocalHalf(syncbuf, submodules=submodules, verbose=verbose)
            syncbuf.Finish()

            if is_new or self.CurrentBranch is None:
                try:
                    self.StartBranch("default")
                except GitError as e:
                    msg = str(e)
                    logger.error(
                        "fatal: cannot create default in manifest %s", msg
                    )
                    return False

            if not manifest_name:
                logger.error("fatal: manifest name (-m) is required.")
                return False

        elif is_new:
            # This is a new standalone manifest.
            manifest_name = "default.xml"
            manifest_data = fetch.fetch_file(manifest_url, verbose=verbose)
            dest = os.path.join(self.worktree, manifest_name)
            os.makedirs(os.path.dirname(dest), exist_ok=True)
            with open(dest, "wb") as f:
                f.write(manifest_data)

        try:
            self.manifest.Link(manifest_name)
        except ManifestParseError as e:
            logger.error("fatal: manifest '%s' not available", manifest_name)
            logger.error("fatal: %s", e)
            return False

        if not this_manifest_only:
            for submanifest in self.manifest.submanifests.values():
                spec = submanifest.ToSubmanifestSpec()
                submanifest.repo_client.manifestProject.Sync(
                    manifest_url=spec.manifestUrl,
                    manifest_branch=spec.revision,
                    standalone_manifest=standalone_manifest,
                    groups=self.manifest_groups,
                    platform=platform,
                    mirror=mirror,
                    dissociate=dissociate,
                    reference=reference,
                    worktree=worktree,
                    submodules=submodules,
                    archive=archive,
                    partial_clone=partial_clone,
                    clone_filter=clone_filter,
                    partial_clone_exclude=partial_clone_exclude,
                    clone_bundle=clone_bundle,
                    git_lfs=git_lfs,
                    use_superproject=use_superproject,
                    verbose=verbose,
                    current_branch_only=current_branch_only,
                    tags=tags,
                    depth=depth,
                    git_event_log=git_event_log,
                    manifest_name=spec.manifestName,
                    this_manifest_only=False,
                    outer_manifest=False,
                )

        # Lastly, if the manifest has a <superproject> then have the
        # superproject sync it (if it will be used).
        if git_superproject.UseSuperproject(use_superproject, self.manifest):
            sync_result = self.manifest.superproject.Sync(git_event_log)
            if not sync_result.success:
                submanifest = ""
                if self.manifest.path_prefix:
                    submanifest = f"for {self.manifest.path_prefix} "
                logger.warning(
                    "warning: git update of superproject %s failed, "
                    "repo sync will not use superproject to fetch source; "
                    "while this error is not fatal, and you can continue to "
                    "run repo sync, please run repo init with the "
                    "--no-use-superproject option to stop seeing this warning",
                    submanifest,
                )
                if sync_result.fatal and use_superproject is not None:
                    return False

        return True

    def ConfigureCloneFilterForDepth(self, clone_filter_for_depth):
        """Configure clone filter to replace shallow clones.

        Args:
            clone_filter_for_depth: a string or None, e.g. 'blob:none' will
            disable shallow clones and replace with partial clone. None will
            enable shallow clones.
        """
        self.config.SetString(
            "repo.clonefilterfordepth", clone_filter_for_depth
        )

    def _ConfigureDepth(self, depth):
        """Configure the depth we'll sync down.

        Args:
            depth: an int, how deep of a partial clone to create.
        """
        # Opt.depth will be non-None if user actually passed --depth to repo
        # init.
        if depth is not None:
            if depth > 0:
                # Positive values will set the depth.
                depth = str(depth)
            else:
                # Negative numbers will clear the depth; passing None to
                # SetString will do that.
                depth = None

            # We store the depth in the main manifest project.
            self.config.SetString("repo.depth", depth)