git.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/core/git.py
 
"""
Purpose
-------
Provide the foundation layer for OCT's Git integration - a pure
subprocess wrapper and a set of read-only state queries. No CLI, no
writes, no hook installation. This module is the sole place where
``subprocess.run`` is allowed to invoke the ``git`` binary; every
higher-level ``oct git`` command (Phase 4B–4E) delegates through
``git_run`` and the helpers below.
 
Responsibilities
----------------
- Translate raw subprocess errors into a small, typed ``GitError``
  hierarchy so callers never see ``FileNotFoundError`` or
  ``TimeoutExpired``.
- Enforce subprocess discipline: ``shell=False``, argv-only, explicit
  timeout, ``text=True``, ``encoding="utf-8"``, ``errors="replace"``.
- Redact ``user:token@`` segments from any git output that crosses a
  diagnostics or exception boundary.
- Provide pure read-only query functions for repo state (branch, HEAD,
  changed files, staged files, remotes, last commit) each accepting an
  explicit ``repo_root``.
- Provide branch mutation primitives (create, switch, create-and-switch)
  used by ``oct git branch`` and ``oct git commit`` auto-branching.
- Provide staging primitives (stage, unstage, interactive patch staging)
  used by ``oct git add``.
- Filter path-escape attempts in ``git_changed_files`` / ``git_staged_files``
  by dropping entries whose resolved target lies outside ``repo_root``
  (OI-405 defense-in-depth).
 
Diagnostics
-----------
Domain: GIT
Levels:
    L1 - errors: missing git binary, command failure, path escape
    L2 - lifecycle: git_run entry/exit with redacted argv + returncode
    L3 - command: each public query function's entry with arg summary
    L4 - deep trace: parsed output, per-call timings
 
Contracts
---------
- This module must not import ``click``.
- Every subprocess invocation goes through :func:`git_run`.
- ``is_git_repo`` is the only function that swallows all exceptions and
  returns ``False``; every other public function raises from the
  ``GitError`` hierarchy on failure.
- Credential redaction is applied inside ``git_run`` - callers that
  bubble exception state or log output never see raw credentials.
- Read-only query functions never write to the repo.
- Branch mutation functions (``create_branch``, ``switch_branch``,
  ``create_and_switch_branch``) and staging functions (``stage_paths``,
  ``unstage_paths``, ``stage_paths_interactive``) are the only write
  operations and are clearly documented as such.
- ``stage_paths_interactive`` is the only function that bypasses
  :func:`git_run` because patch-mode staging requires an attached TTY
  for the user to answer hunk prompts. It still uses ``shell=False``
  and validates argv as a list.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger; the shim falls back to a no-op when the optional oc_diagnostics package is absent)
- oct.core.project_root (is_within_project_root, the containment predicate that drops path-escape entries)
- git (not imported; the git binary, run as a subprocess by git_run and stage_paths_interactive)
"""
 
import re
import subprocess
from pathlib import Path
from typing import Iterable, Optional
 
from oct.core.diagnostics import _dbg
 
from oct.core.project_root import is_within_project_root
 
 
# ---------------------------------------------------------------------
# Module constants
# ---------------------------------------------------------------------
 
#: Default timeout for read-only git queries, in seconds. Intentionally
#: generous so that cold-cache operations on large repositories do not
#: spuriously raise :class:`GitTimeoutError`.
DEFAULT_GIT_TIMEOUT: int = 30
 
#: Regex matching the credential segment of an http(s) URL. Captures the
#: scheme in group ``scheme`` so we can substitute the credentials with
#: ``***@`` while preserving the rest of the URL. Intentionally narrow
#: (no backslashes, no spaces, no slashes, no ``@``) so we only match a
#: ``user[:pass]@`` prefix and never a trailing ``@`` inside a path.
_CRED_URL_RE: re.Pattern[str] = re.compile(
    r"(?P<scheme>https?://)[^/@\s]+@",
)
 
 
# ---------------------------------------------------------------------
# Exception hierarchy (G-A1)
# ---------------------------------------------------------------------
 
 
class GitError(Exception):
    """Base class for every error raised by :mod:`oct.core.git`.
 
    Callers that want a catch-all should catch ``GitError``; the
    specific subclasses exist to distinguish "missing binary" from
    "timeout" from "command failed" so that higher-level CLI commands
    can render tailored error messages.
    """
 
 
class GitNotFoundError(GitError):
    """Raised when the ``git`` binary is not found on ``PATH``."""
 
 
class GitNotARepoError(GitError):
    """Raised when a target path is not inside a git working tree."""
 
 
class GitTimeoutError(GitError):
    """Raised when a git subprocess exceeds its configured timeout."""
 
 
class GitCommandError(GitError):
    """Raised when ``git`` exits with a non-zero status.
 
    Exposes ``returncode``, ``stdout``, and ``stderr`` attributes. Both
    ``stdout`` and ``stderr`` are already credential-redacted.
    """
 
    def __init__(
        self,
        argv: list[str],
        returncode: int,
        stdout: str,
        stderr: str,
    ) -> None:
        # OI-507: redact argv before storing or stringifying so neither the
        # ``.argv`` attribute nor the exception message can leak credentials.
        self.argv = _redact_argv(argv)
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr
        msg = (
            f"git command failed (exit {returncode}): "
            f"{' '.join(self.argv)!r}\nstderr: {stderr.strip() or '<empty>'}"
        )
        super().__init__(msg)
 
 
# ---------------------------------------------------------------------
# Credential redaction (G-A5)
# ---------------------------------------------------------------------
 
 
def _redact_git_url(text: str) -> str:
    """Redact ``user:token@`` segments from http(s) URLs in git output.
 
    Accepts any string (including multi-line output). Empty strings and
    ``None`` are returned unchanged. SSH URLs of the form
    ``git@host:path`` are credential-free (auth is key-based) and are
    deliberately *not* matched by the regex - they pass through intact.
 
    The substitution uses ``\\g<scheme>***@`` so the scheme is preserved
    and the credential segment is replaced with a stable opaque marker.
    """
    func = "_redact_git_url"
    if not text:
        return text
    return _CRED_URL_RE.sub(r"\g<scheme>***@", text)
 
 
def _redact_argv(argv: list[str]) -> list[str]:
    """Return ``argv`` with credentials redacted in each URL-bearing element.
 
    OI-507 - ``git_run`` and ``GitCommandError`` both stringify argv for
    ``_dbg`` and exception messages. Any ``https://user:token@host/...``
    argument would otherwise land in logs and raised exceptions verbatim.
    This helper is the single choke point that keeps the redaction
    invariant (blueprint §11.2) centralised.
    """
    return [_redact_git_url(a) for a in argv]
 
 
# ---------------------------------------------------------------------
# Subprocess wrapper (G-A2)
# ---------------------------------------------------------------------
 
 
def git_run(
    argv: list[str],
    *,
    cwd: Path,
    timeout: int = DEFAULT_GIT_TIMEOUT,
    check: bool = True,
) -> subprocess.CompletedProcess[str]:
    """Run a git command with the project's subprocess discipline.
 
    Parameters
    ----------
    argv
        Git sub-arguments excluding the leading ``git``. For example,
        ``["rev-parse", "--show-toplevel"]``. Must be a list; joined
        strings are rejected to prevent shell injection.
    cwd
        Working directory for the git invocation. Must exist.
    timeout
        Maximum seconds to wait before the subprocess is killed and
        :class:`GitTimeoutError` is raised. Defaults to
        :data:`DEFAULT_GIT_TIMEOUT`.
    check
        When ``True`` (default), a non-zero exit code raises
        :class:`GitCommandError`. When ``False``, the completed process
        is returned with its ``returncode`` intact so callers can
        inspect it (used by :func:`is_git_repo` to avoid exception
        control flow).
 
    Returns
    -------
    ``subprocess.CompletedProcess[str]``
        A completed process whose ``stdout`` and ``stderr`` have already
        been passed through :func:`_redact_git_url`.
 
    Raises
    ------
    GitNotFoundError
        If the ``git`` binary is not on ``PATH``.
    GitTimeoutError
        If the subprocess exceeds ``timeout`` seconds.
    GitCommandError
        If ``check`` is True and the command exits non-zero.
    """
    func = "git_run"
    if not isinstance(argv, list):
        raise GitError(f"argv must be a list, got {type(argv).__name__}")
    full_argv = ["git", *argv]
    # OI-507: redact argv before logging so token-bearing URLs don't land in _dbg.
    _dbg("GIT", func, f"argv={_redact_argv(full_argv)!r} cwd={cwd}", 2)
    try:
        proc = subprocess.run(
            full_argv,
            cwd=str(cwd),
            shell=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
        )
    except FileNotFoundError as exc:
        _dbg("GIT", func, f"git binary not found: {exc}", 1, override=True)
        raise GitNotFoundError(
            "The 'git' executable was not found on PATH. "
            "Install git or adjust PATH."
        ) from exc
    except subprocess.TimeoutExpired as exc:
        # OI-507: both the _dbg line and the raised exception message are
        # credential-safe via _redact_argv.
        redacted_argv = _redact_argv(full_argv)
        _dbg(
            "GIT", func,
            f"git timed out after {timeout}s: {' '.join(redacted_argv)!r}",
            1, override=True,
        )
        raise GitTimeoutError(
            f"git command timed out after {timeout}s: {' '.join(redacted_argv)!r}"
        ) from exc
 
    redacted_out = _redact_git_url(proc.stdout or "")
    redacted_err = _redact_git_url(proc.stderr or "")
    # Replace the captured streams with their redacted form so any
    # caller that logs or re-raises cannot accidentally leak credentials.
    proc.stdout = redacted_out
    proc.stderr = redacted_err
 
    _dbg(
        "GIT", func,
        f"returncode={proc.returncode} stdout_len={len(redacted_out)} "
        f"stderr_len={len(redacted_err)}",
        4,
    )
 
    if check and proc.returncode != 0:
        _dbg(
            "GIT", func,
            f"non-zero exit: {proc.returncode} stderr={redacted_err.strip()!r}",
            1, override=True,
        )
        raise GitCommandError(
            full_argv,
            returncode=proc.returncode,
            stdout=redacted_out,
            stderr=redacted_err,
        )
    return proc
 
 
# ---------------------------------------------------------------------
# State query functions (G-A3)
# ---------------------------------------------------------------------
 
 
def is_git_repo(path: Path) -> bool:
    """Return True if ``path`` is inside a git working tree.
 
    This is the only public function in the module that swallows every
    exception - ``GitNotFoundError``, ``GitCommandError``, ``OSError``,
    and anything else all collapse to ``False``. Use it as a cheap
    predicate before calling any other query function.
    """
    func = "is_git_repo"
    _dbg("GIT", func, f"path={path}", 3)
    try:
        proc = git_run(
            ["rev-parse", "--is-inside-work-tree"],
            cwd=Path(path),
            check=False,
        )
    except GitError:
        return False
    except OSError:
        return False
    result = proc.returncode == 0 and proc.stdout.strip() == "true"
    _dbg("GIT", func, f"result={result}", 4)
    return result
 
 
def get_repo_root(path: Path) -> Path:
    """Return the absolute path of the top-level git repo containing ``path``.
 
    Raises :class:`GitNotARepoError` if ``path`` is not inside a repo.
    """
    func = "get_repo_root"
    _dbg("GIT", func, f"path={path}", 3)
    proc = git_run(
        ["rev-parse", "--show-toplevel"],
        cwd=Path(path),
        check=False,
    )
    if proc.returncode != 0:
        _dbg("GIT", func, f"not a repo: {path}", 1, override=True)
        raise GitNotARepoError(f"{path} is not inside a git repository.")
    root = Path(proc.stdout.strip()).resolve()
    _dbg("GIT", func, f"root={root}", 4)
    return root
 
 
def current_branch(repo_root: Path) -> str:
    """Return the current branch name, or ``"HEAD"`` if detached.
 
    Callers that need to distinguish "on a branch" from "detached"
    should use :func:`is_detached_head` - this function returns the
    literal string ``"HEAD"`` in the detached case, mirroring git's own
    ``--abbrev-ref`` output.
    """
    func = "current_branch"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(
        ["rev-parse", "--abbrev-ref", "HEAD"],
        cwd=Path(repo_root),
    )
    branch = proc.stdout.strip()
    _dbg("GIT", func, f"branch={branch!r}", 4)
    return branch
 
 
def is_detached_head(repo_root: Path) -> bool:
    """Return True if HEAD is detached (not pointing at any branch)."""
    func = "is_detached_head"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    return current_branch(repo_root) == "HEAD"
 
 
def has_uncommitted_changes(repo_root: Path) -> bool:
    """Return True if the working tree has staged or unstaged changes.
 
    Untracked files are ignored - use :func:`has_untracked_files` for
    that. This function checks only tracked files via
    ``git status --porcelain -- uno``.
    """
    func = "has_uncommitted_changes"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(
        ["status", "--porcelain", "--untracked-files=no"],
        cwd=Path(repo_root),
    )
    result = bool(proc.stdout.strip())
    _dbg("GIT", func, f"result={result}", 4)
    return result
 
 
def has_untracked_files(repo_root: Path) -> bool:
    """Return True if there are any untracked files not ignored by .gitignore."""
    func = "has_untracked_files"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(
        ["ls-files", "--others", "--exclude-standard"],
        cwd=Path(repo_root),
    )
    result = bool(proc.stdout.strip())
    _dbg("GIT", func, f"result={result}", 4)
    return result
 
 
def _resolve_and_filter(
    entries: Iterable[str],
    join_base: Path,
    filter_base: Path,
    extensions: Optional[set[str]],
) -> list[Path]:
    """Resolve each entry against ``join_base`` and keep those under ``filter_base``.
 
    Internal helper used by :func:`git_changed_files` and
    :func:`git_staged_files`. Git emits ``diff``/``ls-files`` paths relative
    to the **git top-level** (no ``--relative`` is passed), so entries must be
    joined onto that top-level (``join_base``) - never onto a subproject
    ``project_root``, which would double the subproject segment into a
    non-existent path (Plan 174: ``<repo>/sub`` + ``sub/pkg/x.py`` ->
    ``<repo>/sub/sub/pkg/x.py``). Every entry is:
 
    1. Joined to ``join_base`` and ``.resolve()``-ed.
    2. Kept only when :func:`is_within_project_root` places it inside
       ``filter_base`` (the caller's ``project_root``). This single predicate
       is both the symlink-escape guard AND the subproject scope filter: a
       staged file in a *sibling* subproject resolves outside ``filter_base``
       and is dropped with an L1 ``_dbg`` log, exactly like a symlink escape.
    3. If ``extensions`` is given, filtered to those suffixes.
 
    When ``join_base == filter_base`` (a repo-root ``project_root``) the
    behaviour is byte-identical to a plain join-against-root.
 
    Deduplicates while preserving first-seen order.
    """
    func = "_resolve_and_filter"
    join_base_resolved = Path(join_base).resolve()
    filter_base_resolved = Path(filter_base).resolve()
    result: list[Path] = []
    seen: set[Path] = set()
    for rel in entries:
        rel = rel.strip()
        if not rel:
            continue
        candidate = (join_base_resolved / rel).resolve()
        if not is_within_project_root(candidate, filter_base_resolved):
            _dbg(
                "GIT", func,
                f"dropping out-of-scope/escape candidate: {rel!r} -> {candidate}",
                1, override=True,
            )
            continue
        if extensions is not None and candidate.suffix not in extensions:
            continue
        if candidate in seen:
            continue
        seen.add(candidate)
        result.append(candidate)
    return result
 
 
def git_changed_files(
    repo_root: Path,
    extensions: Optional[set[str]] = None,
) -> list[Path]:
    """Return absolute paths of files changed since the last commit.
 
    The result is the union of:
 
    - Tracked files with staged changes (``git diff --cached --name-only``).
    - Tracked files with unstaged changes (``git diff --name-only``).
    - Untracked files honouring .gitignore (``git ls-files --others --exclude-standard``).
 
    If ``extensions`` is supplied (e.g. ``{".py"}``), only files with
    those suffixes are returned. Paths whose resolved target lies
    outside ``repo_root`` are silently dropped (defense-in-depth against
    symlink-escape attacks; each drop is logged at L1).
    """
    func = "git_changed_files"
    _dbg("GIT", func, f"repo_root={repo_root} extensions={extensions}", 3)
    # Plan 174: git emits top-relative paths, so join onto the git top-level
    # and filter to ``repo_root``. ``get_repo_root`` raises GitNotARepoError
    # (fails loudly) when ``repo_root`` is not inside a work tree.
    top_level = get_repo_root(Path(repo_root))
    entries: list[str] = []
    for argv in (
        ["diff", "--cached", "--name-only", "--diff-filter=ACMR"],
        ["diff", "--name-only", "--diff-filter=ACMR"],
        ["ls-files", "--others", "--exclude-standard"],
    ):
        proc = git_run(argv, cwd=Path(repo_root))
        entries.extend(proc.stdout.splitlines())
    result = _resolve_and_filter(entries, top_level, Path(repo_root), extensions)
    _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
    return result
 
 
def git_staged_files(
    repo_root: Path,
    extensions: Optional[set[str]] = None,
) -> list[Path]:
    """Return absolute paths of files staged for the next commit.
 
    Same contract as :func:`git_changed_files` but scoped to the index
    only - untracked and unstaged-but-modified files are excluded.
    """
    func = "git_staged_files"
    _dbg("GIT", func, f"repo_root={repo_root} extensions={extensions}", 3)
    # Plan 174: git emits top-relative paths, so join onto the git top-level
    # and filter to ``repo_root``. ``get_repo_root`` raises GitNotARepoError
    # (fails loudly) when ``repo_root`` is not inside a work tree.
    top_level = get_repo_root(Path(repo_root))
    proc = git_run(
        ["diff", "--cached", "--name-only", "--diff-filter=ACMR"],
        cwd=Path(repo_root),
    )
    entries = proc.stdout.splitlines()
    result = _resolve_and_filter(entries, top_level, Path(repo_root), extensions)
    _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
    return result
 
 
def git_staged_deletions(repo_root: Path) -> list[Path]:
    """Return absolute paths of files staged for DELETION in the next commit.
 
    :func:`git_staged_files` filters to ``ACMR`` deliberately: its callers lint,
    format and secret-scan what it returns, and a file staged for deletion has
    no content to check. That is the right question for "what must the gate
    inspect" and the wrong one for "does this commit contain anything", and the
    same list was answering both - so a staged set of pure deletions read as
    empty and ``oct git commit`` refused it with exit 4, "Nothing to commit".
    No deletion-only commit could be made through the tool at all: untracking a
    generated artefact, removing a dead module, retiring a fixture.
 
    This asks only the complementary question. The gate keeps inspecting
    exactly what it inspected before.
 
    Joining and scope filtering are :func:`git_staged_files`'s, so a deletion
    staged in a sibling subproject is dropped by the same predicate. These paths
    are normally absent from disk, which is correct and harmless: resolution
    here is textual (``Path.resolve()`` is non-strict and
    :func:`~oct.core.project_root.is_within_project_root` compares resolved
    prefixes), so a removed file is still placed inside or outside the root.
    """
    func = "git_staged_deletions"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    top_level = get_repo_root(Path(repo_root))
    proc = git_run(
        ["diff", "--cached", "--name-only", "--diff-filter=D"],
        cwd=Path(repo_root),
    )
    entries = proc.stdout.splitlines()
    result = _resolve_and_filter(entries, top_level, Path(repo_root), None)
    _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
    return result
 
 
def last_commit_sha(repo_root: Path, short: bool = False) -> str:
    """Return the SHA of HEAD, or the empty string on a fresh repo.
 
    A fresh repo is one where ``git rev-parse HEAD`` fails because no
    commits exist yet. Any other failure still propagates as
    :class:`GitCommandError`.
    """
    func = "last_commit_sha"
    _dbg("GIT", func, f"repo_root={repo_root} short={short}", 3)
    argv = ["rev-parse"]
    if short:
        argv.append("--short")
    argv.append("HEAD")
    proc = git_run(argv, cwd=Path(repo_root), check=False)
    if proc.returncode != 0:
        # Distinguish "no commits yet" from real failures.
        if "unknown revision" in proc.stderr or "bad revision" in proc.stderr:
            _dbg("GIT", func, "no commits in repo", 4)
            return ""
        raise GitCommandError(
            ["git", *argv],
            returncode=proc.returncode,
            stdout=proc.stdout,
            stderr=proc.stderr,
        )
    sha = proc.stdout.strip()
    _dbg("GIT", func, f"sha={sha}", 4)
    return sha
 
 
def last_commit_message(repo_root: Path) -> str:
    """Return the subject + body of HEAD, or the empty string on fresh repos.
 
    Uses ``git log -1 --pretty=%B`` so both the subject and the body are
    returned together (trailing newline stripped).
    """
    func = "last_commit_message"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(
        ["log", "-1", "--pretty=%B"],
        cwd=Path(repo_root),
        check=False,
    )
    if proc.returncode != 0:
        if "does not have any commits" in proc.stderr or \
           "bad default revision" in proc.stderr:
            _dbg("GIT", func, "no commits in repo", 4)
            return ""
        raise GitCommandError(
            ["git", "log", "-1", "--pretty=%B"],
            returncode=proc.returncode,
            stdout=proc.stdout,
            stderr=proc.stderr,
        )
    message = proc.stdout.rstrip("\n")
    _dbg("GIT", func, f"message_len={len(message)}", 4)
    return message
 
 
def remote_urls(repo_root: Path) -> dict[str, str]:
    """Return a ``{remote_name: url}`` map, with credentials redacted.
 
    Uses ``git remote -v`` and collapses the fetch + push variants into
    a single entry per remote (preferring ``fetch`` if both are
    present). Returns an empty dict if no remotes are configured.
    """
    func = "remote_urls"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(["remote", "-v"], cwd=Path(repo_root))
    remotes: dict[str, str] = {}
    for line in proc.stdout.splitlines():
        # Format: "<name>\t<url> (fetch|push)"
        line = line.strip()
        if not line:
            continue
        parts = line.split()
        if len(parts) < 2:
            continue
        name, url = parts[0], parts[1]
        # Prefer the first-seen URL per remote (git outputs fetch then push).
        if name not in remotes:
            remotes[name] = _redact_git_url(url)
    _dbg("GIT", func, f"remotes={list(remotes.keys())}", 4)
    return remotes
 
 
def list_tags(repo_root: Path) -> list[str]:
    """Return tag names sorted by commit date (newest first).
 
    Uses ``git tag --sort=-creatordate``. Returns an empty list if no
    tags exist or if the repo has no commits yet.
    """
    func = "list_tags"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(
        ["tag", "--sort=-creatordate"],
        cwd=Path(repo_root),
        check=False,
    )
    if proc.returncode != 0:
        _dbg("GIT", func, "no tags or git error", 4)
        return []
    tags = [t.strip() for t in proc.stdout.splitlines() if t.strip()]
    _dbg("GIT", func, f"returning {len(tags)} tag(s)", 4)
    return tags
 
 
def get_commit_log(
    repo_root: Path,
    since_ref: str | None = None,
    format_str: str = "%H %s",
) -> list[str]:
    """Return git log lines from HEAD back to *since_ref* (exclusive).
 
    Each line is formatted according to *format_str* (passed to
    ``git log --pretty=<format_str>``).  If *since_ref* is ``None``,
    the entire history is returned.  Returns an empty list on fresh
    repos with no commits.
    """
    func = "get_commit_log"
    _dbg("GIT", func, f"repo_root={repo_root} since_ref={since_ref!r}", 3)
    argv = ["log", f"--pretty={format_str}"]
    if since_ref is not None:
        argv.append(f"{since_ref}..HEAD")
    proc = git_run(argv, cwd=Path(repo_root), check=False)
    if proc.returncode != 0:
        # Fresh repo or invalid ref - return empty.
        _dbg("GIT", func, "no log output", 4)
        return []
    lines = [ln for ln in proc.stdout.splitlines() if ln.strip()]
    _dbg("GIT", func, f"returning {len(lines)} line(s)", 4)
    return lines
 
 
def git_diff_name_only(
    repo_root: Path,
    ref: str,
    extensions: Optional[set[str]] = None,
) -> list[Path]:
    """Return absolute paths of files changed between *ref* and HEAD.
 
    Uses ``git diff --name-only <ref>`` and applies the same
    resolve-and-filter pipeline as :func:`git_changed_files` to drop
    path-escape attempts and optionally filter by extension.
 
    The ``join_base`` argument was missing from the ``_resolve_and_filter`` call
    here, so every invocation raised ``TypeError`` on arity - three arguments to
    a four-parameter function. Nothing in the suite covered it, and its one
    caller is ``oct export-source --diff-ref``, which was therefore broken
    outright rather than subtly wrong. ``get_repo_root`` is resolved AFTER the
    return-code check so a non-repository still returns empty quietly, as this
    function's contract has always promised, instead of raising
    ``GitNotARepoError``.
    """
    func = "git_diff_name_only"
    _dbg("GIT", func, f"repo_root={repo_root} ref={ref!r}", 3)
    proc = git_run(
        ["diff", "--name-only", ref],
        cwd=Path(repo_root),
        check=False,
    )
    if proc.returncode != 0:
        _dbg("GIT", func, "diff failed - returning empty", 4)
        return []
    entries = proc.stdout.splitlines()
    top_level = get_repo_root(Path(repo_root))
    result = _resolve_and_filter(entries, top_level, Path(repo_root), extensions)
    _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
    return result
 
 
def ahead_behind(repo_root: Path) -> tuple[int, int]:
    """Return ``(ahead, behind)`` counts relative to the upstream branch.
 
    Uses ``git rev-list --left-right --count HEAD...@{upstream}``.
    Returns ``(0, 0)`` if no upstream is configured or on a fresh repo.
    """
    func = "ahead_behind"
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
    proc = git_run(
        ["rev-list", "--left-right", "--count", "HEAD...@{upstream}"],
        cwd=Path(repo_root),
        check=False,
    )
    if proc.returncode != 0:
        _dbg("GIT", func, "no upstream or error - returning (0, 0)", 4)
        return 0, 0
    parts = proc.stdout.strip().split()
    if len(parts) != 2:
        return 0, 0
    try:
        ahead = int(parts[0])
        behind = int(parts[1])
    except ValueError:
        return 0, 0
    _dbg("GIT", func, f"ahead={ahead} behind={behind}", 4)
    return ahead, behind
 
 
# ---------------------------------------------------------------------
# Branch mutation functions (G-F1)
# ---------------------------------------------------------------------
 
 
def branch_exists(repo_root: Path, name: str) -> bool:
    """Return True if a local branch *name* exists.
 
    Uses ``git rev-parse --verify refs/heads/<name>`` with
    ``check=False`` so it never raises on a missing branch.
    """
    func = "branch_exists"
    _dbg("GIT", func, f"repo_root={repo_root} name={name!r}", 3)
    proc = git_run(
        ["rev-parse", "--verify", f"refs/heads/{name}"],
        cwd=Path(repo_root),
        check=False,
    )
    result = proc.returncode == 0
    _dbg("GIT", func, f"result={result}", 4)
    return result
 
 
def create_branch(
    repo_root: Path,
    name: str,
    start_point: str | None = None,
) -> None:
    """Create a new branch *name* pointing at *start_point* (default HEAD).
 
    Does **not** switch to the new branch - use :func:`switch_branch` or
    :func:`create_and_switch_branch` for that.
 
    Raises :class:`GitCommandError` if the branch already exists or the
    start-point is invalid.
    """
    func = "create_branch"
    _dbg("GIT", func, f"name={name!r} start_point={start_point!r}", 3)
    argv = ["branch", name]
    if start_point:
        argv.append(start_point)
    git_run(argv, cwd=Path(repo_root))
    _dbg("GIT", func, f"created branch {name!r}", 4)
 
 
def switch_branch(repo_root: Path, name: str) -> None:
    """Switch the working tree to an existing branch *name*.
 
    Uses ``git checkout`` for maximum git-version compatibility.
    Raises :class:`GitCommandError` if the branch does not exist or
    the checkout fails (e.g. uncommitted changes would be overwritten).
    """
    func = "switch_branch"
    _dbg("GIT", func, f"name={name!r}", 3)
    git_run(["checkout", name], cwd=Path(repo_root))
    _dbg("GIT", func, f"switched to {name!r}", 4)
 
 
def create_and_switch_branch(
    repo_root: Path,
    name: str,
    start_point: str | None = None,
) -> None:
    """Atomically create and switch to a new branch *name*.
 
    Equivalent to ``git checkout -b <name> [start_point]``. The index
    and working tree are preserved - staged changes carry over to the
    new branch.
 
    Raises :class:`GitCommandError` if the branch already exists.
    """
    func = "create_and_switch_branch"
    _dbg("GIT", func, f"name={name!r} start_point={start_point!r}", 3)
    argv = ["checkout", "-b", name]
    if start_point:
        argv.append(start_point)
    git_run(argv, cwd=Path(repo_root))
    _dbg("GIT", func, f"created and switched to {name!r}", 4)
 
 
def list_branches(repo_root: Path, *, remote: bool = False) -> list[str]:
    """Return a list of local (or remote) branch names.
 
    Uses ``git branch --format=%(refname:short)`` for clean output
    without the ``*`` marker or whitespace prefix.
    """
    func = "list_branches"
    _dbg("GIT", func, f"repo_root={repo_root} remote={remote}", 3)
    argv = ["branch", "--format=%(refname:short)"]
    if remote:
        argv.insert(1, "-r")
    proc = git_run(argv, cwd=Path(repo_root))
    branches = [b.strip() for b in proc.stdout.splitlines() if b.strip()]
    _dbg("GIT", func, f"returning {len(branches)} branch(es)", 4)
    return branches
 
 
def stage_paths(
    repo_root: Path,
    paths: list[Path] | list[str],
    *,
    update: bool = False,
    all_: bool = False,
) -> None:
    """Stage *paths* for the next commit (whole-file staging).
 
    Equivalent to ``git add [<paths>...]`` with optional ``-u`` (update
    tracked only) or ``-A`` (all changes including untracked).
 
    Parameters
    ----------
    repo_root
        Git repository root.
    paths
        Paths to stage. May be empty when ``update`` or ``all_`` is set.
    update
        Pass ``-u`` so only already-tracked files are staged.
    all_
        Pass ``-A`` so untracked files are also staged.
 
    Raises
    ------
    GitCommandError
        If git rejects the operation (e.g. path outside repo).
    """
    func = "stage_paths"
    _dbg(
        "GIT", func,
        f"repo_root={repo_root} paths={len(paths)} update={update} all={all_}",
        3,
    )
    argv: list[str] = ["add"]
    if update:
        argv.append("-u")
    if all_:
        argv.append("-A")
    if paths:
        argv.append("--")
        argv.extend(str(p) for p in paths)
    elif not (update or all_):
        # Nothing to do; refuse silently rather than running ``git add``
        # with no arguments which is a no-op in modern git but warns.
        _dbg("GIT", func, "no paths and no -u/-A; nothing to stage", 4)
        return
    git_run(argv, cwd=Path(repo_root))
    _dbg("GIT", func, "stage complete", 4)
 
 
def unstage_paths(
    repo_root: Path,
    paths: list[Path] | list[str],
) -> None:
    """Remove *paths* from the index (unstage), leaving working tree intact.
 
    Equivalent to ``git reset HEAD -- <paths>``. Provided for symmetry
    with :func:`stage_paths`; not currently exposed via CLI but used in
    tests and reserved for a future ``oct git unstage`` command.
 
    Raises
    ------
    GitCommandError
        If git rejects the operation.
    """
    func = "unstage_paths"
    _dbg("GIT", func, f"repo_root={repo_root} paths={len(paths)}", 3)
    if not paths:
        _dbg("GIT", func, "no paths; nothing to unstage", 4)
        return
    argv: list[str] = ["reset", "HEAD", "--"]
    argv.extend(str(p) for p in paths)
    git_run(argv, cwd=Path(repo_root))
    _dbg("GIT", func, "unstage complete", 4)
 
 
def restore_paths(
    repo_root: Path,
    paths: list[Path] | list[str],
    *,
    staged: bool = False,
) -> None:
    """Restore *paths* from the index (working tree) or HEAD (staged).
 
    L3 / Phase 4G-followup. Equivalent to:
 
    - ``staged=False``: ``git restore <paths>`` - drop unstaged
      working-tree edits, restoring from the index.
    - ``staged=True``: ``git restore --staged <paths>`` - unstage
      *paths* without touching the working tree.
 
    Raises
    ------
    GitCommandError
        If git rejects the operation.
    """
    func = "restore_paths"
    _dbg(
        "GIT", func,
        f"repo_root={repo_root} paths={len(paths)} staged={staged}",
        3,
    )
    if not paths:
        _dbg("GIT", func, "no paths; nothing to restore", 4)
        return
    argv: list[str] = ["restore"]
    if staged:
        argv.append("--staged")
    argv.append("--")
    argv.extend(str(p) for p in paths)
    git_run(argv, cwd=Path(repo_root))
    _dbg("GIT", func, "restore complete", 4)
 
 
def stage_paths_interactive(
    repo_root: Path,
    paths: list[Path] | list[str],
) -> int:
    """Run ``git add -p`` so the user can choose hunks interactively.
 
    This is the only public function in this module that bypasses
    :func:`git_run`. Patch-mode staging requires an attached TTY so the
    user can answer hunk prompts (``y/n/s/...``); ``git_run`` always
    captures stdout/stderr which would break that interaction. The
    bypass still uses ``shell=False`` and validates argv.
 
    Parameters
    ----------
    repo_root
        Git repository root.
    paths
        Optional list of paths to scope the patch session to. When
        empty, ``git`` walks every modified tracked file.
 
    Returns
    -------
    int
        Process exit code from ``git add -p``. ``0`` on success.
 
    Raises
    ------
    GitNotFoundError
        If the ``git`` executable is not on PATH.
    """
    func = "stage_paths_interactive"
    _dbg("GIT", func, f"repo_root={repo_root} paths={len(paths)}", 3)
    argv: list[str] = ["git", "add", "-p"]
    if paths:
        argv.append("--")
        argv.extend(str(p) for p in paths)
    if not isinstance(argv, list):
        raise GitError(
            f"argv must be a list, got {type(argv).__name__}",
        )
    try:
        proc = subprocess.run(
            argv,
            cwd=str(repo_root),
            shell=False,
            check=False,
        )
    except FileNotFoundError as exc:
        _dbg("GIT", func, f"git binary not found: {exc}", 1, override=True)
        raise GitNotFoundError(
            "The 'git' executable was not found on PATH. "
            "Install git or adjust PATH."
        ) from exc
    _dbg("GIT", func, f"interactive stage exit={proc.returncode}", 4)
    return proc.returncode