git_refs, project: read files-backend refs in Python

GitRefs._LoadAll spawns `git for-each-ref` to load HEAD and refs/*, and
Project._InitAnyMRef leaves warm snapshots stale after direct UpdateRef
writes. On virtualized checkouts with thousands of repositories, these
spawns add avoidable process overhead.

Read HEAD, packed-refs, and loose refs directly in GitRefs when the
repository uses a plain files ref backend, recording (mode, mtime, size,
inode) stat tuples during the single directory pass and rechecking
packed-refs afterward to guard against concurrent pack-refs updates.
The tuples leave out ctime, which backup agents and file system crawlers
move without changing a ref. Fall back to `git for-each-ref` when
GIT_DIR, GIT_COMMON_DIR, reftable, commondir, custom refStorage config,
symlinks, or lockfiles are present.

Also add GitRefs.TryResolveRef for single-ref lookups without scanning
all of refs/, GitRefs.Invalidate, and GitRefs.UpdateRef to keep cached
snapshots warm after _InitAnyMRef writes a direct ref.

Bug: 543101585
Change-Id: I804ac45712208e9f9e6437012661bc1de58b9825
Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/655963
Tested-by: Gavin Mak <gavinmak@google.com>
Reviewed-by: Brian Gan <brgan@google.com>
Commit-Queue: Gavin Mak <gavinmak@google.com>

git_refs.py

@@ -13,6 +13,9 @@
 # limitations under the License.
 
 import os
+import re
+import stat
+from typing import Dict, Optional, Set, Tuple
 
 import git_command
 import platform_utils
@@ -29,6 +32,26 @@ R_WORKTREE_M = R_WORKTREE + "m/"
 R_M = "refs/remotes/m/"
 # Maximum symbolic-ref indirection depth (matches Git SYMREF_MAXDEPTH).
 _SYMREF_MAXDEPTH = 5
+_HEX_OID_RE = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$")
+_BAD_REF_CHARS = frozenset(" ~^:?*[\\")
+_StatKey = Tuple[int, int, int, int]
+
+
+def _IsValidRefName(name: str) -> bool:
+    """Validate a Git refname per git-check-ref-format rules."""
+    if not name or name.endswith(".") or ".." in name or "@{" in name:
+        return False
+    if any(
+        ch in _BAD_REF_CHARS or ord(ch) < 0x20 or ord(ch) == 0x7F for ch in name
+    ):
+        return False
+    parts = name.split("/")
+    if len(parts) < 2:
+        return False
+    for part in parts:
+        if not part or part.startswith(".") or part.endswith(".lock"):
+            return False
+    return True
 
 
 class GitRefs:
@@ -36,7 +59,8 @@ class GitRefs:
         self._gitdir = gitdir
         self._phyref = None
         self._symref = None
-        self._mtime = {}
+        self._mtime: Dict[str, Optional[_StatKey]] = {}
+        self._files_authoritative = False
 
     @property
     def all(self):
@@ -54,6 +78,18 @@ class GitRefs:
         """Whether a ref snapshot has already been loaded."""
         return self._phyref is not None
 
+    @property
+    def is_files_authoritative(self) -> bool:
+        """Whether the loaded snapshot came from a complete files scan."""
+        return self._phyref is not None and self._files_authoritative
+
+    def Invalidate(self) -> None:
+        """Discard any cached ref snapshot so the next access reloads."""
+        self._phyref = None
+        self._symref = None
+        self._mtime = {}
+        self._files_authoritative = False
+
     def get(self, name):
         try:
             return self.all[name]
@@ -70,6 +106,9 @@ class GitRefs:
 
             if name in self._mtime:
                 del self._mtime[name]
+            norm_os = os.path.join(*name.strip("/").split("/"))
+            if norm_os in self._mtime:
+                del self._mtime[norm_os]
 
     def symref(self, name):
         try:
@@ -78,6 +117,80 @@ class GitRefs:
         except KeyError:
             return ""
 
+    def TryResolveRef(self, name: str) -> Optional[str]:
+        """Resolve a single ref directly on the files backend without git.
+
+        Args:
+            name: Reference name ("HEAD" or "refs/...").
+
+        Returns:
+            The resolved hex object ID if present, "" if the repository uses a
+            clean files backend and the ref is authoritatively absent or
+            unborn, or None if git must be invoked as a fallback.
+        """
+        if name != HEAD and (
+            not name.startswith("refs/") or not _IsValidRefName(name)
+        ):
+            return None
+
+        if self._phyref is not None and not self._NeedUpdate():
+            if name in self._phyref:
+                return self._phyref[name]
+            if self._files_authoritative:
+                return ""
+            return None
+
+        if self._CheckFilesBackendConfig() is None:
+            return None
+        refs_dir = os.path.join(self._gitdir, "refs")
+        if platform_utils.islink(refs_dir) or not os.path.isdir(refs_dir):
+            return None
+        head_file = os.path.join(self._gitdir, HEAD)
+        if platform_utils.islink(head_file) or not os.path.isfile(head_file):
+            return None
+
+        cur = name
+        for _ in range(_SYMREF_MAXDEPTH + 1):
+            parts = cur.split("/")
+            for idx in range(1, len(parts) + 1):
+                if platform_utils.islink(
+                    os.path.join(self._gitdir, *parts[:idx])
+                ):
+                    return None
+            ref_path = os.path.join(self._gitdir, *parts)
+            if os.path.lexists(ref_path + ".lock"):
+                return None
+
+            if os.path.lexists(ref_path):
+                if os.path.isdir(ref_path):
+                    return None
+                parsed = self._ParseLooseRefFile(ref_path)
+                if parsed is None:
+                    return None
+                is_sym, val, _ = parsed
+                if not is_sym:
+                    return val
+                cur = val
+                continue
+
+            if cur == HEAD:
+                return None
+
+            packed_path = os.path.join(self._gitdir, "packed-refs")
+            if os.path.lexists(packed_path):
+                packed_res = self._ReadPackedRefs(packed_path)
+                if packed_res is None:
+                    return None
+                packed_map, _ = packed_res
+            else:
+                packed_map = {}
+
+            if os.path.lexists(ref_path):
+                return None
+            return packed_map.get(cur, "")
+
+        return ""
+
     def UpdateSymref(self, name: str, target: str) -> None:
         """Update a symbolic reference in an already-loaded ref snapshot.
 
@@ -88,7 +201,7 @@ class GitRefs:
         if self._phyref is None:
             return
         if self._HasExternalRefChanges(name):
-            self._phyref = None
+            self.Invalidate()
             return
 
         self._symref[name] = target
@@ -100,38 +213,70 @@ class GitRefs:
             self._TrackMtime(anc)
         self._TrackMtime(os.path.join(*parts))
 
-    def _HasExternalRefChanges(self, updated_symref: str) -> bool:
-        """Check whether refs outside |updated_symref| changed on disk.
+    def UpdateRef(self, name: str) -> None:
+        """Refresh a direct reference in an already-loaded ref snapshot.
 
         Args:
-            updated_symref: Symbolic ref path just updated on disk.
+            name: Full reference name just updated on disk (e.g.
+                "refs/remotes/m/main").
+        """
+        if self._phyref is None:
+            return
+        if self._HasExternalRefChanges(name):
+            self.Invalidate()
+            return
+
+        parts = name.strip("/").split("/")
+        ref_rel = os.path.join(*parts)
+        parsed = self._ParseLooseRefFile(os.path.join(self._gitdir, ref_rel))
+        if parsed is None or parsed[0]:
+            self.Invalidate()
+            return
+
+        _, oid, ref_stat = parsed
+        self._symref.pop(name, None)
+        self._phyref[name] = oid
+        self._ResolveSymrefs()
+
+        for idx in range(1, len(parts)):
+            self._TrackMtime(os.path.join(*parts[:idx]))
+        self._mtime[ref_rel] = ref_stat
+
+    def _HasExternalRefChanges(self, updated_ref: str) -> bool:
+        """Check whether refs outside |updated_ref| changed on disk.
+
+        Args:
+            updated_ref: Reference path just updated on disk.
 
         Returns:
-            True if reftable is in use or any unrelated ref path changed.
+            True if files backend is unsafe or any unrelated ref path changed.
         """
-        if os.path.exists(os.path.join(self._gitdir, "reftable")):
+        if self._CheckFilesBackendConfig() is None:
             return True
-        if "packed-refs" not in self._mtime and os.path.exists(
+        if "packed-refs" not in self._mtime and os.path.lexists(
             os.path.join(self._gitdir, "packed-refs")
         ):
             return True
 
-        parts = updated_symref.strip("/").split("/")
+        parts = updated_ref.strip("/").split("/")
         expected_child = {
             "/".join(parts[:idx]): parts[idx] for idx in range(1, len(parts))
         }
 
-        for tracked, mtime in self._mtime.items():
+        for tracked, prev_stat in self._mtime.items():
             norm = tracked.replace(os.sep, "/")
-            if norm == updated_symref:
+            if norm == updated_ref:
                 continue
             full_path = os.path.join(self._gitdir, tracked)
-            try:
-                curr_mtime = os.path.getmtime(full_path)
-            except OSError:
+            curr_stat = self._StatPath(full_path)
+            if prev_stat is None:
+                if curr_stat is not None:
+                    return True
+                continue
+            if curr_stat is None:
                 return True
             if norm in expected_child:
-                if curr_mtime != mtime:
+                if curr_stat != prev_stat:
                     try:
                         children = platform_utils.listdir(full_path)
                     except OSError:
@@ -140,14 +285,15 @@ class GitRefs:
                     for child in children:
                         if child == next_comp:
                             continue
-                        if os.path.join(tracked, child) not in self._mtime:
+                        child_key = os.path.join(tracked, child)
+                        if self._mtime.get(child_key) is None:
                             return True
-            elif curr_mtime != mtime:
+            elif curr_stat != prev_stat:
                 return True
 
         for anc, next_comp in expected_child.items():
             anc_os = anc.replace("/", os.sep)
-            if anc_os not in self._mtime:
+            if self._mtime.get(anc_os) is None:
                 anc_full = os.path.join(self._gitdir, anc_os)
                 try:
                     if platform_utils.listdir(anc_full) != [next_comp]:
@@ -177,20 +323,23 @@ class GitRefs:
         if self._phyref is None or self._NeedUpdate():
             self._LoadAll()
 
-    def _NeedUpdate(self):
+    def _NeedUpdate(self) -> bool:
         with repo_trace.Trace(": scan refs %s", self._gitdir):
-            for name, mtime in self._mtime.items():
-                try:
-                    if mtime != os.path.getmtime(
-                        os.path.join(self._gitdir, name)
-                    ):
-                        return True
-                except OSError:
+            for name, prev_stat in self._mtime.items():
+                if (
+                    self._StatPath(os.path.join(self._gitdir, name))
+                    != prev_stat
+                ):
                     return True
             return False
 
     def _LoadAll(self):
         with repo_trace.Trace(": load refs %s", self._gitdir):
+            if self._TryReadFilesBackend():
+                self._ResolveSymrefs()
+                return
+
+            self._files_authoritative = False
             self._phyref = {}
             self._symref = {}
             self._mtime = {}
@@ -205,19 +354,238 @@ class GitRefs:
             self._ResolveSymrefs()
             self._RefreshMtimes()
 
+    def _CheckFilesBackendConfig(self) -> Optional[_StatKey]:
+        """Verify standard files ref backend preconditions and stat config."""
+        if "GIT_DIR" in os.environ or "GIT_COMMON_DIR" in os.environ:
+            return None
+        if not os.path.isdir(self._gitdir):
+            return None
+        for special in ("reftable", "commondir"):
+            if os.path.lexists(os.path.join(self._gitdir, special)):
+                return None
+
+        cfg_path = os.path.join(self._gitdir, "config")
+        if platform_utils.islink(cfg_path):
+            return None
+        try:
+            with open(cfg_path, "rb") as cfg_fd:
+                if b"refstorage" in cfg_fd.read().lower():
+                    return None
+                return self._FormatStat(os.fstat(cfg_fd.fileno()))
+        except OSError:
+            return None
+
+    @classmethod
+    def _ParseLooseRefFile(
+        cls, path: str
+    ) -> Optional[Tuple[bool, str, _StatKey]]:
+        """Parse a loose ref or HEAD file into (is_symref, val, stat)."""
+        if platform_utils.islink(path):
+            return None
+        try:
+            with open(path, "r", encoding="utf-8") as fd:
+                raw = fd.read(1024)
+                st = cls._FormatStat(os.fstat(fd.fileno()))
+        except (OSError, UnicodeDecodeError):
+            return None
+
+        line = raw.rstrip("\r\n")
+        if not line or "\n" in line or "\r" in line:
+            return None
+        if line.startswith("ref:"):
+            target = line[4:].strip()
+            if not target or target == R_HEADS + ".invalid":
+                return None
+            if target != HEAD and (
+                not target.startswith("refs/") or not _IsValidRefName(target)
+            ):
+                return None
+            return (True, target, st)
+
+        oid = line.strip()
+        if not _HEX_OID_RE.match(oid) or cls._IsNullRef(oid):
+            return None
+        return (False, oid, st)
+
+    @classmethod
+    def _ReadPackedRefs(
+        cls, packed_path: str
+    ) -> Optional[Tuple[Dict[str, str], _StatKey]]:
+        """Parse a packed-refs file into (ref_map, stat)."""
+        if platform_utils.islink(packed_path):
+            return None
+        packed: Dict[str, str] = {}
+        try:
+            with open(packed_path, "r", encoding="utf-8") as packed_fd:
+                for raw_line in packed_fd:
+                    if raw_line.startswith("#"):
+                        continue
+                    line = raw_line.rstrip("\r\n")
+                    if line.startswith("^"):
+                        if not _HEX_OID_RE.match(line[1:]):
+                            return None
+                        continue
+                    parts = line.split(" ", 1)
+                    if len(parts) != 2:
+                        return None
+                    oid, ref_name = parts
+                    if (
+                        not _HEX_OID_RE.match(oid)
+                        or cls._IsNullRef(oid)
+                        or not ref_name.startswith("refs/")
+                        or not _IsValidRefName(ref_name)
+                    ):
+                        return None
+                    packed[ref_name] = oid
+                st = cls._FormatStat(os.fstat(packed_fd.fileno()))
+        except (OSError, UnicodeDecodeError):
+            return None
+        return (packed, st)
+
+    def _TryReadFilesBackend(self) -> bool:
+        """Load HEAD, packed-refs, and loose refs directly when safe.
+
+        Returns:
+            True if the repository uses a standard files ref backend and all
+            refs and stat fingerprints were read cleanly without invoking git.
+        """
+        cfg_stat = self._CheckFilesBackendConfig()
+        if cfg_stat is None:
+            return False
+
+        mtime: Dict[str, Optional[_StatKey]] = {
+            "config": cfg_stat,
+            "reftable": None,
+        }
+        phyref: Dict[str, str] = {}
+        symref: Dict[str, str] = {}
+
+        parsed_head = self._ParseLooseRefFile(os.path.join(self._gitdir, HEAD))
+        if parsed_head is None:
+            return False
+        head_is_sym, head_val, head_stat = parsed_head
+        mtime[HEAD] = head_stat
+        if head_is_sym:
+            symref[HEAD] = head_val
+        else:
+            phyref[HEAD] = head_val
+
+        packed_path = os.path.join(self._gitdir, "packed-refs")
+        packed_stat: Optional[_StatKey] = None
+        if os.path.lexists(packed_path):
+            packed_res = self._ReadPackedRefs(packed_path)
+            if packed_res is None:
+                return False
+            packed_map, packed_stat = packed_res
+            phyref.update(packed_map)
+        mtime["packed-refs"] = packed_stat
+
+        refs_dir = os.path.join(self._gitdir, "refs")
+        if platform_utils.islink(refs_dir):
+            return False
+        refs_stat = self._StatPath(refs_dir)
+        if refs_stat is None or not stat.S_ISDIR(refs_stat[0]):
+            return False
+        mtime["refs"] = refs_stat
+
+        try:
+            to_scan = [("refs", "refs", refs_dir)]
+            while to_scan:
+                ref_prefix, rel_os_prefix, abs_dir = to_scan.pop()
+                with os.scandir(abs_dir) as entries:
+                    for entry in entries:
+                        if entry.is_symlink():
+                            return False
+                        ref_name = f"{ref_prefix}/{entry.name}"
+                        if not _IsValidRefName(ref_name):
+                            return False
+                        rel_os = os.path.join(rel_os_prefix, entry.name)
+                        if entry.is_dir(follow_symlinks=False):
+                            if ref_name in phyref:
+                                return False
+                            # DirEntry.stat() leaves st_ino == 0 on Windows,
+                            # whereas _StatPath() uses os.lstat() and populates
+                            # st_ino. Stat via _StatPath() so _NeedUpdate() and
+                            # _HasExternalRefChanges() compare matching keys.
+                            d_stat = self._StatPath(entry.path)
+                            if d_stat is None or not stat.S_ISDIR(d_stat[0]):
+                                return False
+                            mtime[rel_os] = d_stat
+                            to_scan.append((ref_name, rel_os, entry.path))
+                        elif entry.is_file(follow_symlinks=False):
+                            parsed = self._ParseLooseRefFile(entry.path)
+                            if parsed is None:
+                                return False
+                            is_sym, val, f_stat = parsed
+                            mtime[rel_os] = f_stat
+                            if is_sym:
+                                phyref.pop(ref_name, None)
+                                symref[ref_name] = val
+                            else:
+                                symref.pop(ref_name, None)
+                                phyref[ref_name] = val
+                        else:
+                            return False
+        except OSError:
+            return False
+
+        # Guard against a concurrent `git pack-refs --prune` writing
+        # packed-refs and unlinking loose refs mid-scan.
+        if self._StatPath(packed_path) != packed_stat:
+            return False
+
+        self._phyref = phyref
+        self._symref = symref
+        self._mtime = mtime
+        self._files_authoritative = True
+        return True
+
     def _RefreshMtimes(self) -> None:
         """Record current modification times for ref storage paths."""
         self._mtime = {}
-        self._TrackMtime(HEAD)
-        self._TrackMtime("config")
-        self._TrackMtime("packed-refs")
+        self._TrackMtime(HEAD, track_missing=True)
+        self._TrackMtime("config", track_missing=True)
+        self._TrackMtime("packed-refs", track_missing=True)
+        self._TrackMtime("refs", track_missing=True)
+        self._TrackMtime("reftable", track_missing=True)
         self._TrackTreeMtimes("refs")
         self._TrackTreeMtimes("reftable")
 
+    @staticmethod
+    def _FormatStat(st: os.stat_result) -> _StatKey:
+        # ctime is left out: tools that only touch metadata, such as backup
+        # agents and file system crawlers, move it without changing a ref.
+        return (st.st_mode, st.st_mtime_ns, st.st_size, st.st_ino)
+
+    @classmethod
+    def _StatPath(cls, path: str) -> Optional[_StatKey]:
+        try:
+            st = os.lstat(path)
+        except OSError:
+            return None
+        mount_tag = getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", -1)
+        is_reparse_dir = getattr(st, "st_reparse_tag", 0) == mount_tag
+        if stat.S_ISLNK(st.st_mode) or is_reparse_dir:
+            # Track the symlink target so mutations to symlinked refs or
+            # packed-refs invalidate the fallback snapshot, while keeping
+            # S_IFLNK in mode and negative size when dangling.
+            link_mode = stat.S_IFLNK | stat.S_IMODE(st.st_mode)
+            try:
+                target_st = os.stat(path)
+            except OSError:
+                return (link_mode, st.st_mtime_ns, -1 - st.st_size, st.st_ino)
+            return (
+                link_mode,
+                target_st.st_mtime_ns,
+                target_st.st_size,
+                target_st.st_ino,
+            )
+        return cls._FormatStat(st)
+
     @staticmethod
     def _IsNullRef(ref_id: str) -> bool:
         """Check if a ref_id is a null object ID."""
-        return ref_id and all(ch == "0" for ch in ref_id)
+        return bool(ref_id) and all(ch == "0" for ch in ref_id)
 
     def _ReadRefs(self) -> bool:
         """Read all references using git for-each-ref.
@@ -299,32 +667,45 @@ class GitRefs:
             if ref_id:
                 self._phyref[name] = ref_id
 
-    def _TrackMtime(self, name: str) -> None:
-        """Track the modification time of a specific gitdir path."""
-        path = os.path.join(self._gitdir, name)
-        try:
-            self._mtime[name] = os.path.getmtime(path)
-        except OSError:
-            return
+    def _TrackMtime(self, name: str, track_missing: bool = False) -> None:
+        """Track the stat fingerprint of a specific gitdir path."""
+        st = self._StatPath(os.path.join(self._gitdir, name))
+        if st is not None or track_missing:
+            self._mtime[name] = st
 
     def _TrackTreeMtimes(self, root: str) -> None:
         """Recursively track modification times for a directory tree."""
         root_path = os.path.join(self._gitdir, root)
         try:
             if not platform_utils.isdir(root_path):
+                self._TrackMtime(root)
                 return
         except OSError:
             return
 
+        visited: Set[Tuple[int, int]] = set()
         to_scan = [root]
         while to_scan:
             name = to_scan.pop()
             self._TrackMtime(name)
             path = os.path.join(self._gitdir, name)
-            if not platform_utils.isdir(path):
+            try:
+                dir_st = os.stat(path)
+            except OSError:
                 continue
+            if not stat.S_ISDIR(dir_st.st_mode):
+                continue
+            if dir_st.st_ino:
+                dir_key = (dir_st.st_dev, dir_st.st_ino)
+                if dir_key in visited:
+                    continue
+                visited.add(dir_key)
 
-            for child in platform_utils.listdir(path):
+            try:
+                children = platform_utils.listdir(path)
+            except OSError:
+                continue
+            for child in children:
                 child_name = os.path.join(name, child)
                 child_path = os.path.join(self._gitdir, child_name)
                 if platform_utils.isdir(child_path):

project.py

@@ -4664,6 +4664,8 @@ class Project:
                 msg = "manifest set to %s" % self.revisionId
                 dst = self.revisionId + "^0"
                 active_git.UpdateRef(ref, dst, message=msg, detach=True)
+                if active_git is self.bare_git:
+                    self.bare_ref.UpdateRef(ref)
         else:
             remote = self.GetRemote()
             dst = remote.ToLocal(self.revisionExpr)
@@ -4671,6 +4673,8 @@ class Project:
                 msg = "manifest set to %s" % self.revisionExpr
                 if detach:
                     active_git.UpdateRef(ref, dst, message=msg, detach=True)
+                    if active_git is self.bare_git:
+                        self.bare_ref.UpdateRef(ref)
                 else:
                     active_git.symbolic_ref("-m", msg, ref, dst)
                     if active_git is self.bare_git:

tests/test_git_refs.py

@@ -16,7 +16,9 @@
 
 import os
 from pathlib import Path
+import stat
 import subprocess
+import time
 from typing import Any, List
 from unittest import mock
 
@@ -25,6 +27,7 @@ import utils_for_test
 
 import git_command
 import git_refs
+import platform_utils
 
 
 def _run(repo: str, *args: str) -> str:
@@ -406,3 +409,334 @@ def test_update_symref_propagates_chained_symrefs(tmp_path: Path) -> None:
         assert refs.get("refs/remotes/m/main") == c2
         assert refs.get("refs/remotes/m/alias") == c2
         assert refs.get("HEAD") == c2
+
+
+def test_files_backend_reads_loose_and_packed_without_git(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Files backend reads packed-refs, loose refs, and symrefs in Python."""
+    repo = _init_repo(tmp_path, reftable=False)
+    branch = _run(repo, "symbolic-ref", "--short", "HEAD")
+    c1 = _run(repo, "rev-parse", "HEAD")
+    _run(repo, "tag", "-a", "-m", "v1", "v1.0")
+    tag_oid = _run(repo, "rev-parse", "refs/tags/v1.0")
+    _run(repo, "branch", "packed-only", c1)
+    _run(repo, "branch", "overridden", c1)
+    _run(repo, "pack-refs", "--all")
+
+    Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8")
+    _run(repo, "add", "a")
+    _run(repo, "commit", "-q", "-m", "second")
+    c2 = _run(repo, "rev-parse", "HEAD")
+    _run(repo, "update-ref", "refs/heads/overridden", c2)
+    _run(repo, "symbolic-ref", "refs/remotes/m/main", "refs/heads/overridden")
+
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_COMMON_DIR", raising=False)
+    seen_cmds: List[List[str]] = []
+
+    class SpyGitCommand(git_command.GitCommand):
+        def __init__(
+            self,
+            proj: Any,
+            cmdv: List[str],
+            *args: Any,
+            **kwargs: Any,
+        ) -> None:
+            seen_cmds.append(list(cmdv))
+            super().__init__(proj, cmdv, *args, **kwargs)
+
+    monkeypatch.setattr(git_command, "GitCommand", SpyGitCommand)
+    refs = git_refs.GitRefs(os.path.join(repo, ".git"))
+    assert refs.get("HEAD") == c2
+    assert refs.is_files_authoritative
+    with mock.patch.object(
+        refs,
+        "_LoadAll",
+        side_effect=AssertionError("expected warm GitRefs snapshot"),
+    ):
+        assert refs.get(f"refs/heads/{branch}") == c2
+        assert refs.get("refs/heads/packed-only") == c1
+        assert refs.get("refs/heads/overridden") == c2
+        assert refs.get("refs/remotes/m/main") == c2
+        assert refs.get("refs/tags/v1.0") == tag_oid
+    assert seen_cmds == []
+
+    # A concurrent .lock file triggers clean fallback to git for-each-ref.
+    lock_file = Path(repo, ".git", "refs", "heads", "packed-only.lock")
+    lock_file.write_text("", encoding="utf-8")
+    refs2 = git_refs.GitRefs(os.path.join(repo, ".git"))
+    assert refs2.get("refs/heads/packed-only") == c1
+    assert not refs2.is_files_authoritative
+    assert any(cmd and cmd[0] == "for-each-ref" for cmd in seen_cmds)
+
+
+def test_try_resolve_ref_loose_packed_symref_and_missing(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """TryResolveRef resolves loose/packed/symref refs without full scan."""
+    repo = _init_repo(tmp_path, reftable=False)
+    c1 = _run(repo, "rev-parse", "HEAD")
+    _run(repo, "branch", "packed-branch", c1)
+    _run(repo, "pack-refs", "--all")
+
+    Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8")
+    _run(repo, "add", "a")
+    _run(repo, "commit", "-q", "-m", "second")
+    c2 = _run(repo, "rev-parse", "HEAD")
+    _run(repo, "update-ref", "refs/remotes/origin/main", c2)
+    _run(
+        repo,
+        "symbolic-ref",
+        "refs/remotes/m/main",
+        "refs/remotes/origin/main",
+    )
+
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_COMMON_DIR", raising=False)
+    gitdir = os.path.join(repo, ".git")
+    refs = git_refs.GitRefs(gitdir)
+
+    assert refs.TryResolveRef("refs/remotes/origin/main") == c2
+    assert refs.TryResolveRef("refs/heads/packed-branch") == c1
+    assert refs.TryResolveRef("refs/remotes/m/main") == c2
+    assert refs.TryResolveRef("refs/remotes/origin/missing") == ""
+    assert not refs.is_loaded
+
+    # A .lock file forces None so callers fall back to git.
+    lock_file = Path(gitdir, "refs", "remotes", "origin", "main.lock")
+    lock_file.write_text("", encoding="utf-8")
+    assert refs.TryResolveRef("refs/remotes/origin/main") is None
+
+
+def test_detects_new_packed_refs_and_same_mtime_replace(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """GitRefs reloads on newly created packed-refs and same-mtime replaces."""
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_COMMON_DIR", raising=False)
+
+    repo = os.path.join(tmp_path, "fresh")
+    utils_for_test.init_git_tree(repo, ref_format="files")
+    gitdir = os.path.join(repo, ".git")
+    refs = git_refs.GitRefs(gitdir)
+    assert refs.all == {}
+
+    c1 = "1" * 40
+    c2 = "2" * 40
+    Path(gitdir, "packed-refs").write_text(
+        f"# pack-refs with: peeled fully-peeled sorted \n"
+        f"{c1} refs/remotes/origin/main\n",
+        encoding="utf-8",
+    )
+    assert refs.get("refs/remotes/origin/main") == c1
+
+    loose = Path(gitdir, "refs", "remotes", "origin", "topic")
+    loose.parent.mkdir(parents=True, exist_ok=True)
+    loose.write_text(c1 + "\n", encoding="utf-8")
+    assert refs.get("refs/remotes/origin/topic") == c1
+
+    # Replace the ref the way git does, by renaming a new file over it, then
+    # put the mtimes back so that only the new inode shows.
+    file_st = os.stat(loose)
+    dir_st = os.stat(loose.parent)
+    replacement = Path(gitdir, "topic.new")
+    replacement.write_text(c2 + "\n", encoding="utf-8")
+    os.replace(replacement, loose)
+    os.utime(loose, ns=(file_st.st_atime_ns, file_st.st_mtime_ns))
+    os.utime(loose.parent, ns=(dir_st.st_atime_ns, dir_st.st_mtime_ns))
+    assert refs.get("refs/remotes/origin/topic") == c2
+
+
+@pytest.mark.skipif(
+    platform_utils.isWindows(),
+    reason="st_ctime is the creation time on Windows",
+)
+def test_ctime_only_change_keeps_snapshot_warm(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A ref file whose ctime alone moves does not invalidate the snapshot."""
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_COMMON_DIR", raising=False)
+    repo = _init_repo(tmp_path, reftable=False)
+    branch = _run(repo, "symbolic-ref", "--short", "HEAD")
+    head = _run(repo, "rev-parse", "HEAD")
+    gitdir = os.path.join(repo, ".git")
+    refs = git_refs.GitRefs(gitdir)
+    assert refs.get(f"refs/heads/{branch}") == head
+
+    # Backup agents and file system crawlers tag files with xattrs, which moves
+    # ctime and nothing else. Flipping a mode bit and putting it back does the
+    # same; repeat it until the timestamp granularity lets the change show.
+    loose = os.path.join(gitdir, "refs", "heads", branch)
+    before = os.lstat(loose)
+    mode = stat.S_IMODE(before.st_mode)
+    for _ in range(1000):
+        os.chmod(loose, mode ^ stat.S_IXUSR)
+        os.chmod(loose, mode)
+        if os.lstat(loose).st_ctime_ns != before.st_ctime_ns:
+            break
+        time.sleep(0.001)
+    else:
+        pytest.fail("chmod did not move ctime")
+
+    with mock.patch.object(
+        refs,
+        "_LoadAll",
+        side_effect=AssertionError("expected warm GitRefs snapshot"),
+    ):
+        assert refs.get(f"refs/heads/{branch}") == head
+
+
+@pytest.mark.parametrize("reftable", [False, True])
+def test_update_ref_keeps_snapshot_warm(
+    tmp_path: Path, reftable: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """UpdateRef stays warm on files backend and reloads on reftable."""
+    if reftable and not utils_for_test.supports_reftable():
+        pytest.skip("reftable not supported")
+
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_COMMON_DIR", raising=False)
+    repo = _init_repo(tmp_path, reftable=reftable)
+    branch = _run(repo, "symbolic-ref", "--short", "HEAD")
+    c1 = _run(repo, "rev-parse", "HEAD")
+    _run(repo, "symbolic-ref", "refs/remotes/m/main", f"refs/heads/{branch}")
+
+    Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8")
+    _run(repo, "add", "a")
+    _run(repo, "commit", "-q", "-m", "second")
+    c2 = _run(repo, "rev-parse", "HEAD")
+
+    refs = git_refs.GitRefs(os.path.join(repo, ".git"))
+    assert refs.symref("refs/remotes/m/main") == f"refs/heads/{branch}"
+    assert refs.get("refs/remotes/m/main") == c2
+
+    _run(repo, "update-ref", "--no-deref", "refs/remotes/m/main", c1)
+    refs.UpdateRef("refs/remotes/m/main")
+
+    if reftable:
+        assert not refs.is_loaded
+        assert refs.symref("refs/remotes/m/main") == ""
+        assert refs.get("refs/remotes/m/main") == c1
+    else:
+        with mock.patch.object(
+            refs,
+            "_LoadAll",
+            side_effect=AssertionError("expected warm GitRefs snapshot"),
+        ):
+            assert refs.symref("refs/remotes/m/main") == ""
+            assert refs.get("refs/remotes/m/main") == c1
+
+        # Setting GIT_COMMON_DIR invalidates instead of reading self._gitdir.
+        monkeypatch.setenv("GIT_COMMON_DIR", os.path.join(repo, ".git"))
+        refs.UpdateRef("refs/remotes/m/main")
+        assert not refs.is_loaded
+
+
+def test_symlinked_ref_fallback_reloads_and_breaks_cycles(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Symlink fallback reloads on target changes and breaks dir cycles."""
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_COMMON_DIR", raising=False)
+    repo = _init_repo(tmp_path, reftable=False)
+    c1 = _run(repo, "rev-parse", "HEAD")
+
+    Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8")
+    _run(repo, "add", "a")
+    _run(repo, "commit", "-q", "-m", "second")
+    c2 = _run(repo, "rev-parse", "HEAD")
+
+    gitdir = os.path.join(repo, ".git")
+    ext_dir = tmp_path / "ext_refs"
+    ext_dir.mkdir()
+    target_ref = ext_dir / "linked"
+    target_ref.write_text(c1 + "\n", encoding="utf-8")
+
+    link_ref = os.path.join(gitdir, "refs", "heads", "linked")
+    loop_dir = os.path.join(gitdir, "refs", "heads", "loop")
+    refs_root = os.path.join(gitdir, "refs")
+    used_junction = False
+    try:
+        os.symlink(str(target_ref), link_ref)
+        os.symlink(refs_root, loop_dir, target_is_directory=True)
+        tracked_ref = "refs/heads/linked"
+    except OSError:
+        if not platform_utils.isWindows():
+            raise
+        used_junction = True
+        ext_link_dir = os.path.join(gitdir, "refs", "remotes", "ext")
+        os.makedirs(os.path.dirname(ext_link_dir), exist_ok=True)
+        subprocess.run(
+            ["cmd", "/c", "mklink", "/J", ext_link_dir, str(ext_dir)],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        subprocess.run(
+            ["cmd", "/c", "mklink", "/J", loop_dir, refs_root],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        tracked_ref = "refs/remotes/ext/linked"
+
+    try:
+        # Remove loop_dir before running git for-each-ref, then verify cycle
+        # handling directly on _TrackTreeMtimes so git itself does not walk
+        # deep cyclic paths.
+        if used_junction:
+            os.rmdir(loop_dir)
+        else:
+            os.unlink(loop_dir)
+
+        refs = git_refs.GitRefs(gitdir)
+        assert refs.get(tracked_ref) == c1
+        assert not refs.is_files_authoritative
+
+        replacement = ext_dir / "linked.new"
+        replacement.write_text(c2 + "\n", encoding="utf-8")
+        st = os.stat(target_ref)
+        os.replace(replacement, target_ref)
+        os.utime(target_ref, ns=(st.st_atime_ns, st.st_mtime_ns))
+        assert refs.get(tracked_ref) == c2
+
+        if used_junction:
+            subprocess.run(
+                ["cmd", "/c", "mklink", "/J", loop_dir, refs_root],
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+        else:
+            os.symlink(refs_root, loop_dir, target_is_directory=True)
+
+        refs._TrackTreeMtimes("refs")
+        assert os.path.join("refs", "heads", "loop") in refs._mtime
+    finally:
+        if used_junction:
+            for junc in (
+                loop_dir,
+                os.path.join(gitdir, "refs", "remotes", "ext"),
+            ):
+                if os.path.lexists(junc):
+                    os.rmdir(junc)
+
+    # Also exercise S_ISLNK target and dangling-link tracking directly so
+    # unprivileged Windows runs cover POSIX symlink stat transitions too.
+    lnk = stat.S_IFLNK | 0o777
+    reg = stat.S_IFREG | 0o644
+    fake_link = os.stat_result((lnk, 10, 1, 1, 0, 0, 8, 1, 1, 1))
+    fake_t1 = os.stat_result((reg, 20, 1, 1, 0, 0, 41, 1, 2, 1))
+    fake_t2 = os.stat_result((reg, 21, 1, 1, 0, 0, 41, 1, 2, 1))
+    with mock.patch("os.lstat", return_value=fake_link):
+        with mock.patch("os.stat", return_value=fake_t1):
+            s_t1 = git_refs.GitRefs._StatPath("sym")
+        with mock.patch("os.stat", return_value=fake_t2):
+            s_t2 = git_refs.GitRefs._StatPath("sym")
+        with mock.patch("os.stat", side_effect=FileNotFoundError):
+            s_dangling = git_refs.GitRefs._StatPath("sym")
+    assert s_t1 != s_t2
+    assert s_t1 != s_dangling
+    assert s_dangling is not None and not stat.S_ISDIR(s_dangling[0])