VCS API Reference¶

class VCSError(Exception):
    """Base exception for all VCS operations."""

class BinaryNotFoundError(VCSError):
    """Raised when the VCS binary cannot be located.

    Attributes:
        binary_name: Name of the binary that was not found.
    """

    def __init__(self, binary_name: str) -> None:
        self.binary_name = binary_name
        super().__init__(f"VCS binary not found: {binary_name}")

class CommandError(VCSError):
    """Raised when a VCS command exits with an unexpected return code.

    Attributes:
        command: The full command list that was executed.
        returncode: Process exit code.
        stderr: Captured standard-error output.
        timeout: Timeout value in seconds if this was a timeout, else ``None``.
    """

    def __init__(
        self,
        command: list[str],
        returncode: int,
        stderr: str,
        *,
        timeout: float | None = None,
    ) -> None:
        self.command = command
        self.returncode = returncode
        self.stderr = stderr
        self.timeout = timeout
        cmd_str = " ".join(command)
        if timeout is not None:
            msg = f"Command timed out after {timeout}s: {cmd_str}"
        else:
            msg = f"Command failed (rc={returncode}): {cmd_str}\n{stderr.rstrip()}"
        super().__init__(msg)

class NotARepoError(VCSError):
    """Raised when the given path is not inside a repository.

    Attributes:
        path: Filesystem path that was tested.
    """

    def __init__(self, path: str) -> None:
        self.path = path
        super().__init__(f"Not a repository: {path}")

@dataclasses.dataclass(frozen=True, slots=True)
class FileStatus:
    """Status of a single file in the working tree.

    Attributes:
        path: Relative path of the file.
        status: Single-character status code (e.g. ``'M'``, ``'A'``,
            ``'D'``, ``'R'``, ``'?'``, ``'!'``).
        original_path: For renames/copies, the path before the operation.
    """

    path: str
    status: str
    original_path: str | None = None

@dataclasses.dataclass(frozen=True, slots=True)
class Commit:
    """Metadata for a single commit.

    Attributes:
        hash: Full commit hash.
        short_hash: Abbreviated commit hash.
        author: Author name.
        date: Commit date in ISO 8601 format.
        message: First line of the commit message.
    """

    hash: str
    short_hash: str
    author: str
    date: str
    message: str

@dataclasses.dataclass(frozen=True, slots=True)
class BlameLine:
    """Annotation for a single source line.

    Attributes:
        commit: Commit hash that last changed this line.
        author: Author of the commit.
        date: Commit date.
        line_no: 1-based line number.
        content: Actual text content of the line.
    """

    commit: str
    author: str
    date: str
    line_no: int
    content: str

@dataclasses.dataclass(frozen=True, slots=True)
class WorkspaceInfo:
    """Information about a workspace (Git worktree / Jujutsu workspace).

    Attributes:
        path: Absolute path to the workspace directory.
        head: HEAD commit hash.
        branch: Branch name (Git) or bookmark (Jujutsu), ``None``
            for detached HEAD or unknown.
        is_main: Whether this is the main/default workspace.
    """

    path: str
    head: str
    branch: str | None = None
    is_main: bool = False

@runtime_checkable
class VCSBackend(Protocol):
    """Protocol that every VCS backend must satisfy."""

    @property
    def name(self) -> str:
        """Short name of the VCS (e.g. ``"git"``)."""
        ...

    def is_repo(self, path: str) -> bool:
        """Return ``True`` if *path* is inside a repository."""
        ...

    def diff(self, *paths: str, staged: bool = False) -> str:
        """Return a unified diff of uncommitted changes.

        Args:
            *paths: Restrict diff to these paths.
            staged: If ``True``, diff only staged (index) changes.
        """
        ...

    def diff_files(self, path_a: str, path_b: str) -> str:
        """Return a diff between two arbitrary files.

        Args:
            path_a: First file path.
            path_b: Second file path.
        """
        ...

    def apply(self, patch: str) -> None:
        """Apply a unified diff patch to the working tree.

        Args:
            patch: Patch text (unified diff format).
        """
        ...

    def status(self) -> list[FileStatus]:
        """Return the list of changed files in the working tree."""
        ...

    def log(self, n: int = 10) -> list[Commit]:
        """Return the last *n* commits.

        Args:
            n: Maximum number of commits to return.
        """
        ...

    def blame(self, path: str) -> list[BlameLine]:
        """Return per-line annotation for a file.

        Args:
            path: Path to the file (relative to repo root).
        """
        ...

    def current_branch(self) -> str:
        """Return the name of the current branch or bookmark."""
        ...

    def merge_file(self, base: str, ours: str, theirs: str) -> str:
        """Three-way merge of file contents.

        Args:
            base: Common-ancestor file content.
            ours: Content from the first branch.
            theirs: Content from the second branch.

        Returns:
            Merged file content (may contain conflict markers).
        """
        ...

    def workspace_add(
        self,
        path: str,
        *,
        branch: str | None = None,
        rev: str | None = None,
    ) -> str:
        """Create a new workspace at *path*.

        Args:
            path: Directory for the new workspace.
            branch: Branch to create (Git) or bookmark (Jujutsu).
            rev: Starting revision/commit.

        Returns:
            Absolute path to the created workspace.
        """
        ...

    def workspace_remove(self, path: str, *, force: bool = False) -> None:
        """Remove a workspace.

        Args:
            path: Path to the workspace directory.
            force: Force removal even with uncommitted changes.
        """
        ...

    def workspace_list(self) -> list[WorkspaceInfo]:
        """List all workspaces.

        Returns:
            List of :class:`WorkspaceInfo` entries.
        """
        ...

    def branches(self) -> list[str]:
        """List all branch names or bookmarks.

        Returns:
            List of branch/bookmark name strings.
        """
        ...

    def create_branch(self, name: str, *, rev: str | None = None) -> None:
        """Create a new branch or bookmark.

        Args:
            name: Branch/bookmark name.
            rev: Starting point (defaults to current HEAD).
        """
        ...

    def switch(self, target: str) -> None:
        """Switch to a branch or revision.

        Args:
            target: Branch name or revision identifier.
        """
        ...

    def commit(self, message: str, *, paths: list[str] | None = None) -> str:
        """Create a commit/change.

        Args:
            message: Commit message.
            paths: Files to stage and commit (Git-specific).

        Returns:
            Commit hash of the new commit.
        """
        ...

    def rev_parse(self, rev: str) -> str:
        """Resolve a revision string to a full commit hash.

        Args:
            rev: Revision string (branch name, tag, ``"HEAD"``, etc.).

        Returns:
            Full commit hash.
        """
        ...

class Git:
    """Git CLI backend.

    Args:
        repo_path: Path to the repository (defaults to ``"."``).
        binary: Explicit path to the ``git`` binary; discovered
            automatically if ``None``.
        encoding: Text encoding for command I/O.
        timeout: Default subprocess timeout in seconds.
    """

    def __init__(
        self,
        repo_path: str = ".",
        *,
        binary: str | None = None,
        encoding: str = "utf-8",
        timeout: float = 30.0,
    ) -> None:
        self._binary = binary or _find_binary("git")
        self._repo = os.path.abspath(repo_path)
        self._encoding = encoding
        self._timeout = timeout

    # -- helpers --

    def _git(
        self,
        *args: str,
        allowed_returncodes: tuple[int, ...] = (0,),
        input: str | None = None,  # noqa: A002
    ) -> subprocess.CompletedProcess[str]:
        """Run a git sub-command inside the repository.

        Args:
            *args: Arguments after ``git``.
            allowed_returncodes: Acceptable exit codes.
            input: Text to pass via stdin.

        Returns:
            Completed process result.
        """
        cmd = [self._binary, *args]
        return _run(
            cmd,
            cwd=self._repo,
            input=input,
            timeout=self._timeout,
            encoding=self._encoding,
            allowed_returncodes=allowed_returncodes,
        )

    # -- protocol implementation --

    @property
    def name(self) -> str:
        """Return ``"git"``."""
        return "git"

    def is_repo(self, path: str) -> bool:
        """Check whether *path* is inside a git repository.

        Args:
            path: Directory to test.

        Returns:
            ``True`` if *path* is a git working tree or bare repo.
        """
        git_dir = os.path.join(path, ".git")
        if os.path.isdir(git_dir) or os.path.isfile(git_dir):
            return True
        try:
            _run(
                [self._binary, "-C", path, "rev-parse", "--git-dir"],
                timeout=self._timeout,
                encoding=self._encoding,
            )
            return True
        except (VCSError, OSError):
            return False

    def diff(self, *paths: str, staged: bool = False) -> str:
        """Return a unified diff of uncommitted changes.

        Args:
            *paths: Restrict diff to these paths.
            staged: If ``True``, show only staged changes.

        Returns:
            Diff text (may be empty).
        """
        cmd: list[str] = ["diff"]
        if staged:
            cmd.append("--staged")
        if paths:
            cmd.append("--")
            cmd.extend(paths)
        result = self._git(*cmd, allowed_returncodes=(0, 1))
        return result.stdout

    def diff_files(self, path_a: str, path_b: str) -> str:
        """Return a diff between two arbitrary files.

        Args:
            path_a: First file path.
            path_b: Second file path.

        Returns:
            Diff text.
        """
        result = self._git(
            "diff", "--no-index", "--", path_a, path_b, allowed_returncodes=(0, 1)
        )
        return result.stdout

    def apply(self, patch: str) -> None:
        """Apply a unified diff patch via ``git apply``.

        Args:
            patch: Patch text.
        """
        self._git("apply", "-", input=patch)

    def status(self) -> list[FileStatus]:
        """Return the list of changed files via ``git status --porcelain``.

        Returns:
            List of :class:`FileStatus` entries.
        """
        result = self._git("status", "--porcelain=v1")
        entries: list[FileStatus] = []
        for line in result.stdout.splitlines():
            if len(line) < 4:
                continue
            xy = line[:2]
            rest = line[3:]
            # Pick the most informative status character:
            # index status (X) or work-tree status (Y).
            status_char = xy[0] if xy[0] != " " else xy[1]
            if status_char == "R" and " -> " in rest:
                old, new = rest.split(" -> ", 1)
                entries.append(FileStatus(path=new, status="R", original_path=old))
            elif " -> " in rest and xy[0] == "R":
                old, new = rest.split(" -> ", 1)
                entries.append(FileStatus(path=new, status="R", original_path=old))
            else:
                entries.append(FileStatus(path=rest, status=status_char))
        return entries

    def log(self, n: int = 10) -> list[Commit]:
        """Return the last *n* commits.

        Args:
            n: Maximum number of commits.

        Returns:
            List of :class:`Commit` entries, newest first.
        """
        fmt = "%H%n%h%n%an%n%aI%n%s"
        result = self._git("log", f"--format={fmt}", "-n", str(n))
        lines = result.stdout.strip().splitlines()
        commits: list[Commit] = []
        for i in range(0, len(lines) - 4, 5):
            commits.append(
                Commit(
                    hash=lines[i],
                    short_hash=lines[i + 1],
                    author=lines[i + 2],
                    date=lines[i + 3],
                    message=lines[i + 4],
                )
            )
        return commits

    def blame(self, path: str) -> list[BlameLine]:
        """Return per-line blame annotation via ``git blame --porcelain``.

        Args:
            path: File path relative to the repository root.

        Returns:
            List of :class:`BlameLine` entries, one per source line.
        """
        result = self._git("blame", "--porcelain", path)
        blame_lines: list[BlameLine] = []
        current_hash = ""
        current_author = ""
        current_date = ""
        current_lineno = 0
        for raw in result.stdout.splitlines():
            # Header line: <hash> <orig-line> <final-line> [<num-lines>]
            m = re.match(r"^([0-9a-f]{40})\s+\d+\s+(\d+)", raw)
            if m:
                current_hash = m.group(1)
                current_lineno = int(m.group(2))
                continue
            if raw.startswith("author "):
                current_author = raw[7:]
            elif raw.startswith("author-time "):
                current_date = raw[12:]
            elif raw.startswith("\t"):
                blame_lines.append(
                    BlameLine(
                        commit=current_hash,
                        author=current_author,
                        date=current_date,
                        line_no=current_lineno,
                        content=raw[1:],
                    )
                )
        return blame_lines

    def current_branch(self) -> str:
        """Return the current branch name.

        Falls back to the short HEAD hash when in detached-HEAD state.

        Returns:
            Branch name or abbreviated commit hash.
        """
        result = self._git("branch", "--show-current")
        branch = result.stdout.strip()
        if branch:
            return branch
        # Detached HEAD — return short hash
        result = self._git("rev-parse", "--short", "HEAD")
        return result.stdout.strip()

    def merge_file(self, base: str, ours: str, theirs: str) -> str:
        """Three-way merge of file contents via ``git merge-file``.

        Writes the three versions to temporary files, runs
        ``git merge-file -p``, and returns the merged output.  Conflict
        markers are included when the merge is non-clean.

        Args:
            base: Common-ancestor file content.
            ours: Content from the first branch.
            theirs: Content from the second branch.

        Returns:
            Merged file content (may contain conflict markers).
        """
        tmp_base = tmp_ours = tmp_theirs = None
        try:
            tmp_base = tempfile.NamedTemporaryFile(
                mode="w",
                suffix=".base",
                delete=False,
                encoding=self._encoding,
            )
            tmp_base.write(base)
            tmp_base.close()

            tmp_ours = tempfile.NamedTemporaryFile(
                mode="w",
                suffix=".ours",
                delete=False,
                encoding=self._encoding,
            )
            tmp_ours.write(ours)
            tmp_ours.close()

            tmp_theirs = tempfile.NamedTemporaryFile(
                mode="w",
                suffix=".theirs",
                delete=False,
                encoding=self._encoding,
            )
            tmp_theirs.write(theirs)
            tmp_theirs.close()

            result = self._git(
                "merge-file",
                "-p",
                tmp_ours.name,
                tmp_base.name,
                tmp_theirs.name,
                allowed_returncodes=(0, 1),
            )
            return result.stdout
        finally:
            for tmp in (tmp_base, tmp_ours, tmp_theirs):
                if tmp is not None:
                    # Tier 3: best-effort silent — temp file cleanup
                    try:
                        os.unlink(tmp.name)
                    except OSError:
                        pass

    # -- workspace / branch / commit operations --

    def workspace_add(
        self,
        path: str,
        *,
        branch: str | None = None,
        rev: str | None = None,
    ) -> str:
        """Create a new workspace via ``git worktree add``.

        Args:
            path: Directory for the new workspace.
            branch: If given, create this branch in the new workspace.
            rev: Starting commit-ish.

        Returns:
            Absolute path to the created workspace.
        """
        abs_path = os.path.abspath(path)
        cmd: list[str] = ["worktree", "add"]
        if branch:
            cmd.extend(["-b", branch])
        cmd.append(abs_path)
        if rev:
            cmd.append(rev)
        self._git(*cmd)
        return abs_path

    def workspace_remove(self, path: str, *, force: bool = False) -> None:
        """Remove a workspace via ``git worktree remove``.

        Args:
            path: Path to the workspace directory.
            force: Force removal even with uncommitted changes.
        """
        abs_path = os.path.abspath(path)
        cmd: list[str] = ["worktree", "remove"]
        if force:
            cmd.append("--force")
        cmd.append(abs_path)
        self._git(*cmd)

    def workspace_list(self) -> list[WorkspaceInfo]:
        """List all workspaces via ``git worktree list --porcelain``.

        Returns:
            List of :class:`WorkspaceInfo` entries.
        """
        result = self._git("worktree", "list", "--porcelain")
        workspaces: list[WorkspaceInfo] = []
        path = ""
        head = ""
        branch: str | None = None
        for line in result.stdout.splitlines():
            if line.startswith("worktree "):
                # Flush previous entry
                if path:
                    workspaces.append(
                        WorkspaceInfo(
                            path=path,
                            head=head,
                            branch=branch,
                            is_main=len(workspaces) == 0,
                        )
                    )
                path = line[9:]
                head = ""
                branch = None
            elif line.startswith("HEAD "):
                head = line[5:]
            elif line.startswith("branch "):
                branch = line[7:].removeprefix("refs/heads/")
            elif line == "detached":
                branch = None
        # Flush last entry
        if path:
            workspaces.append(
                WorkspaceInfo(
                    path=path,
                    head=head,
                    branch=branch,
                    is_main=len(workspaces) == 0,
                )
            )
        return workspaces

    def branches(self) -> list[str]:
        """List all branches via ``git branch``.

        Returns:
            List of branch name strings.
        """
        result = self._git("branch", "--format=%(refname:short)")
        return [b.strip() for b in result.stdout.splitlines() if b.strip()]

    def create_branch(self, name: str, *, rev: str | None = None) -> None:
        """Create a new branch via ``git branch``.

        Args:
            name: Branch name.
            rev: Starting commit-ish (defaults to HEAD).
        """
        cmd: list[str] = ["branch", name]
        if rev:
            cmd.append(rev)
        self._git(*cmd)

    def switch(self, target: str) -> None:
        """Switch to a branch or revision via ``git switch``.

        Falls back to ``git switch --detach`` for non-branch revisions.

        Args:
            target: Branch name or revision identifier.
        """
        try:
            self._git("switch", target)
        except CommandError:
            self._git("switch", "--detach", target)

    def commit(self, message: str, *, paths: list[str] | None = None) -> str:
        """Create a commit via ``git commit``.

        If *paths* is given, stages those files first.  Otherwise commits
        whatever is already staged.

        Args:
            message: Commit message.
            paths: Files to stage before committing.

        Returns:
            Full commit hash of the new commit.
        """
        if paths:
            self._git("add", "--", *paths)
        self._git("commit", "-m", message)
        result = self._git("rev-parse", "HEAD")
        return result.stdout.strip()

    def rev_parse(self, rev: str) -> str:
        """Resolve a revision string via ``git rev-parse``.

        Args:
            rev: Revision string.

        Returns:
            Full commit hash.
        """
        result = self._git("rev-parse", rev)
        return result.stdout.strip()