eigent/backend/app/workspace_git/backend.py

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

"""Typed, non-interactive Git CLI backend for Eigent-owned operations."""

from __future__ import annotations

import hashlib
import os
import re
import shutil
import subprocess
import threading
from dataclasses import dataclass
from pathlib import Path, PurePosixPath

_UNSAFE_INHERITED_GIT_ENV = {
    "GIT_ALTERNATE_OBJECT_DIRECTORIES",
    "GIT_ASKPASS",
    "GIT_COMMON_DIR",
    "GIT_CONFIG_COUNT",
    "GIT_CONFIG_PARAMETERS",
    "GIT_DIR",
    "GIT_INDEX_FILE",
    "GIT_OBJECT_DIRECTORY",
    "GIT_PAGER",
    "GIT_PROXY_COMMAND",
    "GIT_SEQUENCE_EDITOR",
    "GIT_SSH",
    "GIT_SSH_COMMAND",
    "GIT_WORK_TREE",
    "GIT_EDITOR",
    "GIT_EXTERNAL_DIFF",
    "SSH_ASKPASS",
}
_ADVANCED_COMMAND_SLOTS = threading.BoundedSemaphore(4)


class GitBackendError(RuntimeError):
    """Base error for typed local Git operations."""


class GitCommandError(GitBackendError):
    def __init__(
        self,
        *,
        args: tuple[str, ...],
        returncode: int,
        stderr: str,
    ) -> None:
        self.args_safe = args
        self.returncode = returncode
        self.stderr = stderr
        super().__init__(
            f"git command failed ({returncode}): {' '.join(args)}: {stderr}"
        )


class GitCommandTimeoutError(GitBackendError):
    """Raised when a bounded Git command exceeds its execution budget."""

    def __init__(
        self, *, args: tuple[str, ...], timeout_seconds: float
    ) -> None:
        self.args_safe = args
        self.timeout_seconds = timeout_seconds
        super().__init__(
            f"git command timed out after {timeout_seconds:g}s: {args[0]}"
        )


class NestedRepositoryError(GitBackendError):
    """Raised when init would silently create a nested repository."""


@dataclass(frozen=True)
class GitCommandResult:
    stdout: str
    stderr: str
    returncode: int
    stdout_truncated: bool = False
    stderr_truncated: bool = False


@dataclass(frozen=True)
class RepositoryProbe:
    requested_root: Path
    is_repository: bool
    repository_root: Path | None
    owns_requested_root: bool
    nested_in_parent: bool
    head_oid: str | None
    branch: str | None


@dataclass(frozen=True)
class RepoStateToken:
    head_oid: str | None
    branch_or_detached_head: str
    index_digest: str
    operation_state: str

    @property
    def digest(self) -> str:
        payload = "\0".join(
            (
                self.head_oid or "unborn",
                self.branch_or_detached_head,
                self.index_digest,
                self.operation_state,
            )
        )
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class RepositoryDiagnostics:
    healthy: bool
    issues: tuple[str, ...]
    state_token: RepoStateToken
    has_submodules: bool
    has_remotes: bool


@dataclass(frozen=True)
class GitWorktreeInfo:
    path: Path
    head_oid: str | None
    ref_name: str | None
    detached: bool


@dataclass(frozen=True)
class GitMergeResult:
    commit_oid: str | None
    source_oid: str
    conflict_paths: tuple[str, ...]


@dataclass(frozen=True)
class GitObjectMetadata:
    oid: str
    object_type: str
    size_bytes: int


@dataclass(frozen=True)
class GitPathChange:
    status: str
    relative_path: str


@dataclass(frozen=True)
class GitPathLineStat:
    relative_path: str
    added_lines: int | None
    removed_lines: int | None


@dataclass(frozen=True)
class GitPathBlob:
    relative_path: str
    oid: str
    size_bytes: int


class GitBackend:
    """Small typed backend; arbitrary command execution is intentionally absent."""

    def __init__(
        self,
        git_executable: str | Path | None = None,
        *,
        timeout_seconds: float = 30.0,
        max_output_chars: int = 64_000,
        hooks_path: Path | None = None,
    ) -> None:
        executable = (
            str(git_executable)
            if git_executable is not None
            else os.environ.get("EIGENT_BUNDLED_GIT") or shutil.which("git")
        )
        if not executable:
            raise GitBackendError("Git runtime is not available")
        executable_path = Path(executable).expanduser()
        if executable_path.is_absolute() and not executable_path.is_file():
            raise GitBackendError(
                f"Git executable does not exist: {executable_path}"
            )
        self.git_executable = str(executable_path)
        self.timeout_seconds = timeout_seconds
        self.max_output_chars = max_output_chars
        self.hooks_path = hooks_path or Path(os.devnull)

    def run_advanced_argv(
        self,
        repository_root: Path,
        args: tuple[str, ...],
        *,
        identity: tuple[str, str] | None = None,
    ) -> GitCommandResult:
        """Execute classifier-approved argv with bounded streaming output.

        Classification and policy evaluation intentionally live above this
        transport primitive. This method never invokes a shell, never reads
        stdin, and drains output without retaining more than the configured
        per-stream limit in memory.
        """

        if not args or any(not value or "\x00" in value for value in args):
            raise GitBackendError("advanced Git argv is invalid")
        root = repository_root.expanduser().resolve()
        command = self._command(root, args)
        if not _ADVANCED_COMMAND_SLOTS.acquire(timeout=self.timeout_seconds):
            raise GitBackendError("advanced Git concurrency limit reached")
        try:
            return self._run_advanced_argv_acquired(
                command=command,
                args=args,
                identity=identity,
            )
        finally:
            _ADVANCED_COMMAND_SLOTS.release()

    def _run_advanced_argv_acquired(
        self,
        *,
        command: tuple[str, ...],
        args: tuple[str, ...],
        identity: tuple[str, str] | None,
    ) -> GitCommandResult:
        try:
            process = subprocess.Popen(
                command,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=self._environment(identity=identity),
            )
        except OSError as exc:
            raise GitBackendError(
                f"failed to execute advanced Git operation: {args[0]}"
            ) from exc

        stdout = bytearray()
        stderr = bytearray()
        truncated = {"stdout": False, "stderr": False}

        def drain(name: str, pipe, destination: bytearray) -> None:
            while True:
                chunk = pipe.read(16 * 1024)
                if not chunk:
                    return
                remaining = self.max_output_chars - len(destination)
                if remaining > 0:
                    destination.extend(chunk[:remaining])
                if len(chunk) > remaining:
                    truncated[name] = True

        assert process.stdout is not None
        assert process.stderr is not None
        threads = (
            threading.Thread(
                target=drain,
                args=("stdout", process.stdout, stdout),
                daemon=True,
            ),
            threading.Thread(
                target=drain,
                args=("stderr", process.stderr, stderr),
                daemon=True,
            ),
        )
        for thread in threads:
            thread.start()
        try:
            returncode = process.wait(timeout=self.timeout_seconds)
        except subprocess.TimeoutExpired as exc:
            process.kill()
            process.wait()
            for thread in threads:
                thread.join(timeout=1)
            raise GitCommandTimeoutError(
                args=args,
                timeout_seconds=self.timeout_seconds,
            ) from exc
        for thread in threads:
            thread.join(timeout=1)
        return GitCommandResult(
            stdout=stdout.decode("utf-8", errors="replace"),
            stderr=stderr.decode("utf-8", errors="replace"),
            returncode=returncode,
            stdout_truncated=truncated["stdout"],
            stderr_truncated=truncated["stderr"],
        )

    def probe(self, root: Path) -> RepositoryProbe:
        requested = root.expanduser().resolve()
        if not requested.is_dir():
            raise GitBackendError(
                f"repository root is not a directory: {root}"
            )
        top = self._run(
            requested,
            ("rev-parse", "--show-toplevel"),
            check=False,
        )
        if top.returncode != 0:
            return RepositoryProbe(
                requested_root=requested,
                is_repository=False,
                repository_root=None,
                owns_requested_root=False,
                nested_in_parent=False,
                head_oid=None,
                branch=None,
            )
        repository_root = Path(top.stdout.strip()).expanduser().resolve()
        owns_root = repository_root == requested
        head = self._run(
            repository_root,
            ("rev-parse", "--verify", "HEAD"),
            check=False,
        )
        branch = self._run(
            repository_root,
            ("symbolic-ref", "--quiet", "--short", "HEAD"),
            check=False,
        )
        return RepositoryProbe(
            requested_root=requested,
            is_repository=True,
            repository_root=repository_root,
            owns_requested_root=owns_root,
            nested_in_parent=not owns_root,
            head_oid=head.stdout.strip() if head.returncode == 0 else None,
            branch=branch.stdout.strip() if branch.returncode == 0 else None,
        )

    def init_repository(
        self,
        root: Path,
        *,
        initial_branch: str = "main",
    ) -> RepositoryProbe:
        requested = root.expanduser().resolve()
        requested.mkdir(parents=True, exist_ok=True)
        before = self.probe(requested)
        if before.nested_in_parent:
            raise NestedRepositoryError(
                f"refusing to initialize nested repository inside "
                f"{before.repository_root}"
            )
        if before.is_repository:
            return before
        self._run(
            requested,
            ("init", f"--initial-branch={initial_branch}", "--"),
        )
        after = self.probe(requested)
        if not after.is_repository or not after.owns_requested_root:
            raise GitBackendError("Git init did not create the expected repo")
        return after

    def current_head(self, repository_root: Path) -> str | None:
        result = self._run(
            repository_root,
            ("rev-parse", "--verify", "HEAD"),
            check=False,
        )
        return result.stdout.strip() if result.returncode == 0 else None

    def object_metadata(
        self,
        repository_root: Path,
        object_ids: tuple[str, ...],
    ) -> tuple[GitObjectMetadata, ...]:
        """Return bounded object type/size metadata without reading blobs."""

        if not object_ids:
            return ()
        for oid in object_ids:
            self._validate_object_name(oid)
        result = self._run(
            repository_root,
            (
                "cat-file",
                "--batch-check=%(objectname) %(objecttype) %(objectsize)",
            ),
            input_text="\n".join(object_ids) + "\n",
        )
        records: list[GitObjectMetadata] = []
        for line in result.stdout.splitlines():
            oid, separator, remainder = line.partition(" ")
            object_type, separator_two, raw_size = remainder.partition(" ")
            if not separator or not separator_two or not raw_size.isdigit():
                raise GitBackendError("Git returned invalid object metadata")
            records.append(
                GitObjectMetadata(
                    oid=oid,
                    object_type=object_type,
                    size_bytes=int(raw_size),
                )
            )
        if len(records) != len(object_ids):
            raise GitBackendError("Git object metadata response was truncated")
        return tuple(records)

    def is_ancestor(
        self,
        repository_root: Path,
        ancestor_oid: str,
        descendant_oid: str,
    ) -> bool:
        self._validate_object_name(ancestor_oid)
        self._validate_object_name(descendant_oid)
        result = self._run(
            repository_root,
            ("merge-base", "--is-ancestor", ancestor_oid, descendant_oid),
            check=False,
        )
        if result.returncode not in {0, 1}:
            raise GitCommandError(
                args=("merge-base", "--is-ancestor"),
                returncode=result.returncode,
                stderr=result.stderr,
            )
        return result.returncode == 0

    def merge_owned_ref(
        self,
        worktree_root: Path,
        *,
        source_ref: str,
        operation_id: str,
        message: str,
    ) -> GitMergeResult:
        """Merge one Eigent branch and abort cleanly on conflicts."""

        self._validate_eigent_ref(source_ref, branch_only=True)
        if not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", operation_id):
            raise ValueError("invalid Git operation id")
        if not message.strip():
            raise ValueError("merge message is required")
        root = worktree_root.expanduser().resolve()
        owned = next(
            (
                item
                for item in self.list_worktrees(root)
                if item.path == root
                and item.ref_name is not None
                and item.ref_name.startswith("refs/heads/eigent/")
            ),
            None,
        )
        if owned is None:
            raise GitBackendError("merge requires an Eigent-owned worktree")
        if not self.is_worktree_clean(root):
            raise GitBackendError("merge target worktree is not clean")
        source_oid = self.ref_oid(root, source_ref)
        current_oid = self.current_head(root)
        if source_oid is None or current_oid is None:
            raise GitBackendError("merge source or target HEAD is missing")
        if self.is_ancestor(root, source_oid, current_oid):
            return GitMergeResult(
                commit_oid=current_oid,
                source_oid=source_oid,
                conflict_paths=(),
            )
        result = self._run(
            root,
            (
                "merge",
                "--no-ff",
                "--no-edit",
                "-m",
                f"{message}\n\nEigent-Operation: {operation_id}",
                source_ref,
            ),
            check=False,
            identity=("Eigent", "noreply@eigent.ai"),
        )
        if result.returncode != 0:
            conflicts = self._run(
                root,
                ("diff", "--name-only", "--diff-filter=U", "-z"),
                check=False,
            )
            conflict_paths = tuple(
                sorted(path for path in conflicts.stdout.split("\0") if path)
            )
            self._run(root, ("merge", "--abort"), check=False)
            if conflict_paths:
                return GitMergeResult(
                    commit_oid=None,
                    source_oid=source_oid,
                    conflict_paths=conflict_paths,
                )
            raise GitCommandError(
                args=("merge", source_ref),
                returncode=result.returncode,
                stderr=result.stderr,
            )
        merged_oid = self.current_head(root)
        if merged_oid is None:
            raise GitBackendError("merge completed without a target HEAD")
        return GitMergeResult(
            commit_oid=merged_oid,
            source_oid=source_oid,
            conflict_paths=(),
        )

    def repo_state_token(self, repository_root: Path) -> RepoStateToken:
        root = repository_root.expanduser().resolve()
        probe = self.probe(root)
        if not probe.is_repository or not probe.owns_requested_root:
            raise GitBackendError(f"not an owned Git root: {root}")
        index = self._run(root, ("ls-files", "--stage", "-z"))
        status = self._run(
            root,
            ("status", "--porcelain=v1", "-z", "--untracked-files=all"),
        )
        worktree_metadata = self._status_metadata(root, status.stdout)
        index_digest = hashlib.sha256(
            (
                index.stdout + "\0" + status.stdout + "\0" + worktree_metadata
            ).encode("utf-8")
        ).hexdigest()
        return RepoStateToken(
            head_oid=probe.head_oid,
            branch_or_detached_head=(probe.branch or "DETACHED"),
            index_digest=index_digest,
            operation_state=self._operation_state(root),
        )

    def diagnostics(self, repository_root: Path) -> RepositoryDiagnostics:
        root = repository_root.expanduser().resolve()
        token = self.repo_state_token(root)
        issues: list[str] = []
        if token.operation_state != "clean":
            issues.append(f"operation_in_progress:{token.operation_state}")
        if token.head_oid is not None:
            connectivity = self._run(
                root,
                ("cat-file", "-e", f"{token.head_oid}^{{commit}}"),
                check=False,
            )
            if connectivity.returncode != 0:
                issues.append("object_database_unhealthy")
        staged = self._run(root, ("ls-files", "--stage"))
        has_submodules = any(
            line.startswith("160000 ") for line in staged.stdout.splitlines()
        )
        remotes = self._run(root, ("remote",))
        return RepositoryDiagnostics(
            healthy=not issues,
            issues=tuple(issues),
            state_token=token,
            has_submodules=has_submodules,
            has_remotes=bool(remotes.stdout.strip()),
        )

    def changed_paths(
        self,
        repository_root: Path,
        paths: tuple[Path, ...],
    ) -> tuple[str, ...]:
        return tuple(self.path_status(repository_root, paths))

    def path_status(
        self,
        repository_root: Path,
        paths: tuple[Path, ...],
    ) -> dict[str, str]:
        pathspecs = self._relative_pathspecs(repository_root, paths)
        result = self._run(
            repository_root,
            ("status", "--porcelain=v1", "--", *pathspecs),
        )
        return {
            line[3:]: line[:2]
            for line in result.stdout.splitlines()
            if len(line) > 3
        }

    def is_tracked(self, repository_root: Path, path: Path) -> bool:
        pathspec = self._relative_pathspecs(repository_root, (path,))[0]
        result = self._run(
            repository_root,
            ("ls-files", "--error-unmatch", "--", pathspec),
            check=False,
        )
        return result.returncode == 0

    def commit_paths(
        self,
        repository_root: Path,
        paths: tuple[Path, ...],
        *,
        message: str,
        author_name: str = "Eigent",
        author_email: str = "noreply@eigent.ai",
    ) -> str:
        if not message.strip():
            raise ValueError("Git commit message is required")
        pathspecs = self._relative_pathspecs(repository_root, paths)
        self.assert_no_clean_filters(repository_root, pathspecs)
        self._run(repository_root, ("add", "--", *pathspecs))
        staged = self._run(
            repository_root,
            ("diff", "--cached", "--name-only", "--", *pathspecs),
        )
        if not staged.stdout.strip():
            head = self.current_head(repository_root)
            if head is None:
                raise GitBackendError(
                    "configuration repository has no commit and no changes"
                )
            return head
        self._run(
            repository_root,
            (
                "-c",
                f"user.name={author_name}",
                "-c",
                f"user.email={author_email}",
                "-c",
                "commit.gpgSign=false",
                "commit",
                "--only",
                "--no-verify",
                "-m",
                message,
                "--",
                *pathspecs,
            ),
            identity=(author_name, author_email),
        )
        head = self.current_head(repository_root)
        if head is None:
            raise GitBackendError("Git commit did not create HEAD")
        return head

    def empty_tree_oid(self, repository_root: Path) -> str:
        """Return Git's canonical empty-tree object without creating a commit."""

        result = self._run(repository_root, ("mktree",), input_text="")
        oid = result.stdout.strip()
        if not oid:
            raise GitBackendError("Git did not create the empty tree object")
        return oid

    def show_commit_paths(
        self,
        repository_root: Path,
        commit: str = "HEAD",
    ) -> tuple[str, ...]:
        result = self._run(
            repository_root,
            (
                "diff-tree",
                "--no-commit-id",
                "--name-only",
                "-r",
                "--root",
                commit,
            ),
        )
        return tuple(line for line in result.stdout.splitlines() if line)

    def relative_paths(
        self,
        repository_root: Path,
        paths: tuple[Path, ...],
    ) -> tuple[str, ...]:
        return self._relative_pathspecs(repository_root, paths)

    def diff_paths(
        self,
        repository_root: Path,
        paths: tuple[Path, ...],
        *,
        cached: bool = False,
        source: str | None = None,
    ) -> str:
        pathspecs = self._relative_pathspecs(repository_root, paths)
        args = ["diff", "--no-ext-diff", "--no-textconv"]
        if cached:
            args.append("--cached")
        if source is not None:
            self._validate_object_name(source)
            args.append(source)
        args.extend(("--", *pathspecs))
        return self._run(repository_root, tuple(args)).stdout

    def changed_paths_between(
        self,
        repository_root: Path,
        *,
        base_commit: str,
        target_commit: str,
    ) -> tuple[GitPathChange, ...]:
        """Return exact non-rename path changes between two commits."""

        self._validate_object_name(base_commit)
        self._validate_object_name(target_commit)
        result = self._run(
            repository_root,
            (
                "diff",
                "--name-status",
                "-z",
                "--no-renames",
                base_commit,
                target_commit,
                "--",
            ),
        )
        fields = result.stdout.split("\0")
        if fields and fields[-1] == "":
            fields.pop()
        if len(fields) % 2:
            raise GitBackendError("Git returned malformed path changes")
        changes: list[GitPathChange] = []
        for offset in range(0, len(fields), 2):
            status = fields[offset]
            path = self._normalize_relative_git_path(fields[offset + 1])
            if status not in {"A", "M", "D", "T"}:
                raise GitBackendError(
                    f"unsupported Git path change status: {status!r}"
                )
            changes.append(GitPathChange(status=status, relative_path=path))
        return tuple(changes)

    def path_line_stats_between(
        self,
        repository_root: Path,
        *,
        base_commit: str,
        target_commit: str,
    ) -> tuple[GitPathLineStat, ...]:
        """Return text line totals per path; binary files use null totals."""

        self._validate_object_name(base_commit)
        self._validate_object_name(target_commit)
        result = self._run(
            repository_root,
            (
                "diff",
                "--numstat",
                "-z",
                "--no-renames",
                base_commit,
                target_commit,
                "--",
            ),
        )
        stats: list[GitPathLineStat] = []
        for record in result.stdout.split("\0"):
            if not record:
                continue
            raw_added, separator, remainder = record.partition("\t")
            raw_removed, separator_two, raw_path = remainder.partition("\t")
            if not separator or not separator_two or not raw_path:
                raise GitBackendError("Git returned malformed line stats")
            path = self._normalize_relative_git_path(raw_path)
            if raw_added == "-" and raw_removed == "-":
                added_lines = None
                removed_lines = None
            elif raw_added.isdigit() and raw_removed.isdigit():
                added_lines = int(raw_added)
                removed_lines = int(raw_removed)
            else:
                raise GitBackendError("Git returned invalid line stats")
            stats.append(
                GitPathLineStat(
                    relative_path=path,
                    added_lines=added_lines,
                    removed_lines=removed_lines,
                )
            )
        return tuple(stats)

    def hash_worktree_file(
        self,
        repository_root: Path,
        path: Path,
    ) -> str:
        """Hash one regular file as a raw Git blob without clean filters."""

        root = repository_root.expanduser().resolve()
        target = path.expanduser().resolve()
        try:
            target.relative_to(root)
        except ValueError as exc:
            raise GitBackendError(
                "file is outside the repository root"
            ) from exc
        if target.is_symlink() or not target.is_file():
            raise GitBackendError("Git blob hashing requires a regular file")
        result = self._run(
            root,
            ("hash-object", "--no-filters", "--", str(target)),
        )
        oid = result.stdout.strip()
        self._validate_object_name(oid)
        return oid

    def commit_parent(
        self,
        repository_root: Path,
        commit_oid: str,
    ) -> str | None:
        self._validate_object_name(commit_oid)
        result = self._run(
            repository_root,
            ("rev-parse", f"{commit_oid}^"),
            check=False,
        )
        return result.stdout.strip() if result.returncode == 0 else None

    def find_commit_by_operation(
        self,
        repository_root: Path,
        operation_id: str,
    ) -> str | None:
        if not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", operation_id):
            raise ValueError("invalid Git operation id")
        result = self._run(
            repository_root,
            (
                "log",
                "--all",
                "--fixed-strings",
                f"--grep=Eigent-Operation: {operation_id}",
                "-1",
                "--format=%H",
            ),
            check=False,
        )
        value = result.stdout.strip()
        return value if result.returncode == 0 and value else None

    def update_eigent_ref(
        self,
        repository_root: Path,
        ref_name: str,
        commit_oid: str,
        *,
        expected_oid: str | None = None,
    ) -> str:
        self._validate_eigent_ref(ref_name)
        self._validate_object_name(commit_oid)
        args = ["update-ref", ref_name, commit_oid]
        if expected_oid is not None:
            self._validate_object_name(expected_oid)
            args.append(expected_oid)
        self._run(repository_root, tuple(args))
        return commit_oid

    def archive_eigent_branch_ref(
        self,
        repository_root: Path,
        *,
        active_ref: str,
        archive_ref: str,
        expected_oid: str,
    ) -> str:
        self._validate_eigent_ref(active_ref, branch_only=True)
        self._validate_eigent_ref(archive_ref)
        if not archive_ref.startswith("refs/eigent/archive/"):
            raise ValueError(
                "archive ref must use the Eigent archive namespace"
            )
        self._validate_object_name(expected_oid)
        root = repository_root.expanduser().resolve()
        active_oid = self.ref_oid(root, active_ref)
        if active_oid not in {None, expected_oid}:
            raise GitBackendError("active Eigent ref changed before archive")
        archived_oid = self.ref_oid(root, archive_ref)
        if archived_oid is None:
            self.update_eigent_ref(root, archive_ref, expected_oid)
        elif archived_oid != expected_oid:
            raise GitBackendError("archive ref already points elsewhere")
        if active_oid == expected_oid:
            self._run(
                root,
                ("update-ref", "-d", active_ref, expected_oid),
            )
        return archive_ref

    def ref_oid(
        self,
        repository_root: Path,
        ref_name: str,
    ) -> str | None:
        self._validate_eigent_ref(ref_name)
        result = self._run(
            repository_root,
            ("rev-parse", "--verify", ref_name),
            check=False,
        )
        return result.stdout.strip() if result.returncode == 0 else None

    def blob_oid_at_path(
        self,
        repository_root: Path,
        commit_oid: str,
        relative_path: str,
    ) -> str | None:
        """Resolve one regular Git blob without checking out a worktree."""

        self._validate_object_name(commit_oid)
        path = self._normalize_relative_git_path(relative_path)
        result = self._run(
            repository_root,
            ("ls-tree", "-z", commit_oid, "--", path),
            check=False,
        )
        if result.returncode != 0 or not result.stdout:
            return None
        metadata, separator, observed_path = result.stdout.partition("\t")
        if separator != "\t" or observed_path.rstrip("\0") != path:
            return None
        parts = metadata.split()
        if len(parts) != 3 or parts[1] != "blob":
            return None
        oid = parts[2]
        self._validate_object_name(oid)
        return oid

    def blobs_at_paths(
        self,
        repository_root: Path,
        commit_oid: str,
        relative_paths: tuple[str, ...],
    ) -> tuple[GitPathBlob, ...]:
        """Resolve regular blobs and sizes for many paths in one Git call."""

        if not relative_paths:
            return ()
        self._validate_object_name(commit_oid)
        paths = tuple(
            self._normalize_relative_git_path(path) for path in relative_paths
        )
        result = self._run(
            repository_root,
            ("ls-tree", "-z", "-l", commit_oid, "--", *paths),
        )
        blobs: list[GitPathBlob] = []
        for record in result.stdout.split("\0"):
            if not record:
                continue
            metadata, separator, raw_path = record.partition("\t")
            parts = metadata.split()
            if separator != "\t" or len(parts) != 4:
                raise GitBackendError("Git returned malformed blob metadata")
            _mode, object_type, oid, raw_size = parts
            if object_type != "blob" or not raw_size.isdigit():
                continue
            self._validate_object_name(oid)
            blobs.append(
                GitPathBlob(
                    relative_path=self._normalize_relative_git_path(raw_path),
                    oid=oid,
                    size_bytes=int(raw_size),
                )
            )
        return tuple(blobs)

    def object_size(self, repository_root: Path, object_oid: str) -> int:
        self._validate_object_name(object_oid)
        result = self._run(repository_root, ("cat-file", "-s", object_oid))
        try:
            value = int(result.stdout.strip())
        except ValueError as exc:
            raise GitBackendError(
                "Git returned an invalid object size"
            ) from exc
        if value < 0:
            raise GitBackendError("Git returned a negative object size")
        return value

    def read_blob_range(
        self,
        repository_root: Path,
        object_oid: str,
        *,
        start_offset: int,
        max_bytes: int,
    ) -> bytes:
        """Stream a bounded blob range without buffering the whole object."""

        self._validate_object_name(object_oid)
        if start_offset < 0 or max_bytes < 1:
            raise ValueError("invalid Git blob byte range")
        command = self._command(
            repository_root,
            ("cat-file", "blob", object_oid),
        )
        try:
            process = subprocess.Popen(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=self._environment(),
            )
        except OSError as exc:
            raise GitBackendError("failed to stream Git blob") from exc
        assert process.stdout is not None
        try:
            remaining = start_offset
            while remaining:
                chunk = process.stdout.read(min(remaining, 64 * 1024))
                if not chunk:
                    break
                remaining -= len(chunk)
            data = process.stdout.read(max_bytes)
        finally:
            if process.poll() is None:
                process.terminate()
            try:
                _, stderr = process.communicate(timeout=self.timeout_seconds)
            except subprocess.TimeoutExpired:
                process.kill()
                _, stderr = process.communicate()
        if remaining:
            return b""
        if process.returncode not in {0, -15}:
            raise GitCommandError(
                args=("cat-file", "blob", object_oid),
                returncode=process.returncode or -1,
                stderr=stderr.decode("utf-8", errors="replace").strip(),
            )
        return data

    def create_anchor_commit(
        self,
        repository_root: Path,
        *,
        message: str,
    ) -> str:
        if not message.strip():
            raise ValueError("anchor commit message is required")
        empty_tree = self._run(
            repository_root,
            ("mktree",),
            input_text="",
        ).stdout.strip()
        if not empty_tree:
            raise GitBackendError("Git did not create an empty tree")
        commit = self._run(
            repository_root,
            ("commit-tree", empty_tree, "-m", message),
            identity=("Eigent", "noreply@eigent.ai"),
        ).stdout.strip()
        self._validate_object_name(commit)
        return commit

    def list_worktrees(
        self,
        repository_root: Path,
    ) -> tuple[GitWorktreeInfo, ...]:
        result = self._run(
            repository_root,
            ("worktree", "list", "--porcelain", "-z"),
        )
        records: list[GitWorktreeInfo] = []
        fields: dict[str, str] = {}
        for value in result.stdout.split("\0"):
            if not value:
                if "worktree" in fields:
                    records.append(self._worktree_from_fields(fields))
                    fields = {}
                continue
            key, _, item = value.partition(" ")
            fields[key] = item
        if "worktree" in fields:
            records.append(self._worktree_from_fields(fields))
        return tuple(records)

    def is_worktree_clean(self, worktree_root: Path) -> bool:
        result = self._run(
            worktree_root,
            ("status", "--porcelain=v1", "-z", "--untracked-files=all"),
        )
        return not result.stdout

    def worktree_status(
        self,
        worktree_root: Path,
        *,
        limit: int = 500,
    ) -> dict[str, str]:
        if limit < 1:
            raise ValueError("status limit must be positive")
        result = self._run(
            worktree_root,
            ("status", "--porcelain=v1", "-z", "--untracked-files=all"),
        )
        values = result.stdout.split("\0")
        records: dict[str, str] = {}
        index = 0
        while index < len(values):
            record = values[index]
            index += 1
            if not record or len(record) < 4:
                continue
            state = record[:2]
            relative_path = record[3:]
            records[relative_path] = state
            if "R" in state or "C" in state:
                index += 1
            if len(records) > limit:
                raise GitBackendError(
                    f"worktree delta exceeds the {limit}-path limit"
                )
        return records

    def restore_owned_worktree_path(
        self,
        worktree_root: Path,
        *,
        relative_path: str,
        source_commit: str | None,
    ) -> None:
        """Restore one private worktree path; never targets User Worktree."""

        path = self._normalize_relative_git_path(relative_path)
        root = worktree_root.expanduser().resolve()
        owned = next(
            (
                item
                for item in self.list_worktrees(root)
                if item.path == root
                and item.ref_name is not None
                and item.ref_name.startswith("refs/heads/eigent/")
            ),
            None,
        )
        if owned is None:
            raise GitBackendError(
                "owned path restore requires an attached private worktree"
            )
        if source_commit is not None:
            self._validate_object_name(source_commit)
        source_blob = (
            self.blob_oid_at_path(root, source_commit, path)
            if source_commit is not None
            else None
        )
        if source_blob is not None:
            self._run(
                root,
                (
                    "restore",
                    f"--source={source_commit}",
                    "--worktree",
                    "--",
                    path,
                ),
            )
            return
        target = root / path
        if target.is_symlink() or (target.exists() and not target.is_file()):
            raise GitBackendError(
                "refusing to remove non-regular private projection path"
            )
        target.unlink(missing_ok=True)

    def restore_owned_paths_from_ref(
        self,
        worktree_root: Path,
        *,
        source_ref: str,
        relative_paths: tuple[str, ...],
    ) -> None:
        """Project selected paths from an Eigent ref into a clean worktree."""

        self._validate_eigent_ref(source_ref, branch_only=True)
        if not relative_paths:
            raise ValueError("at least one conflict path is required")
        normalized: list[str] = []
        for value in relative_paths:
            path = PurePosixPath(value)
            if not value or path.is_absolute() or ".." in path.parts:
                raise GitBackendError("conflict paths must be relative")
            normalized.append(path.as_posix())
        if len(set(normalized)) != len(normalized):
            raise GitBackendError("conflict paths must be unique")
        if not self.is_worktree_clean(worktree_root):
            raise GitBackendError(
                "Run integration changed before conflict resolution"
            )
        self._run(
            worktree_root,
            (
                "restore",
                "--source",
                source_ref,
                "--worktree",
                "--",
                *normalized,
            ),
        )

    def worktree_matches_commit(
        self,
        worktree_root: Path,
        commit_oid: str,
    ) -> bool:
        self._validate_object_name(commit_oid)
        changed = self._run(
            worktree_root,
            ("diff", "--quiet", commit_oid, "--"),
            check=False,
        )
        if changed.returncode != 0:
            return False
        untracked = self._run(
            worktree_root,
            ("ls-files", "--others", "--exclude-standard", "-z"),
        )
        return not untracked.stdout

    def refresh_owned_worktree(
        self,
        worktree_root: Path,
        *,
        expected_projected_head: str,
        target_head: str,
    ) -> None:
        self._validate_object_name(expected_projected_head)
        self._validate_object_name(target_head)
        current_head = self.current_head(worktree_root)
        if current_head == target_head and self.worktree_matches_commit(
            worktree_root, target_head
        ):
            return
        if not self.worktree_matches_commit(
            worktree_root,
            expected_projected_head,
        ):
            raise GitBackendError(
                "Project worktree contains external or uncheckpointed changes"
            )
        self._run(worktree_root, ("reset", "--hard", target_head))
        if not self.worktree_matches_commit(worktree_root, target_head):
            raise GitBackendError("Project worktree refresh did not converge")

    def ensure_worktree(
        self,
        repository_root: Path,
        *,
        worktree_path: Path,
        ref_name: str,
        commit_oid: str,
    ) -> GitWorktreeInfo:
        self._validate_eigent_ref(ref_name, branch_only=True)
        self._validate_object_name(commit_oid)
        root = repository_root.expanduser().resolve()
        destination = worktree_path.expanduser().resolve()
        if destination == root or root in destination.parents:
            raise GitBackendError(
                "Eigent worktrees must live outside the User Worktree"
            )
        existing_ref = self.ref_oid(root, ref_name)
        if existing_ref is None:
            self.update_eigent_ref(root, ref_name, commit_oid)
        elif existing_ref != commit_oid:
            raise GitBackendError(
                f"worktree ref {ref_name!r} points to another commit"
            )
        for item in self.list_worktrees(root):
            if item.path == destination:
                if item.ref_name != ref_name:
                    raise GitBackendError(
                        "worktree path is registered for another ref"
                    )
                return item
            if item.ref_name == ref_name:
                raise GitBackendError(
                    "Eigent ref is already checked out in another worktree"
                )
        if destination.exists():
            if not destination.is_dir() or any(destination.iterdir()):
                raise GitBackendError(
                    f"worktree destination is not empty: {destination}"
                )
        destination.parent.mkdir(parents=True, exist_ok=True)
        self._run(
            root,
            (
                "worktree",
                "add",
                str(destination),
                ref_name.removeprefix("refs/heads/"),
            ),
        )
        for item in self.list_worktrees(root):
            if item.path == destination and item.ref_name == ref_name:
                return item
        raise GitBackendError("Git did not register the requested worktree")

    def remove_owned_worktree(
        self,
        repository_root: Path,
        *,
        worktree_path: Path,
        expected_ref: str,
    ) -> None:
        self._validate_eigent_ref(expected_ref, branch_only=True)
        root = repository_root.expanduser().resolve()
        target = worktree_path.expanduser().resolve()
        if target == root or root in target.parents:
            raise GitBackendError("refusing to remove the User Worktree")
        registered = next(
            (
                item
                for item in self.list_worktrees(root)
                if item.path == target
            ),
            None,
        )
        if registered is None:
            if target.exists():
                raise GitBackendError(
                    "unregistered worktree path requires manual attention"
                )
            return
        if registered.ref_name != expected_ref:
            raise GitBackendError("worktree is registered for another ref")
        if not self.is_worktree_clean(target):
            raise GitBackendError("refusing to remove a dirty worktree")
        self._run(root, ("worktree", "remove", str(target)))
        if any(item.path == target for item in self.list_worktrees(root)):
            raise GitBackendError("Git did not remove the owned worktree")

    @staticmethod
    def _worktree_from_fields(fields: dict[str, str]) -> GitWorktreeInfo:
        return GitWorktreeInfo(
            path=Path(fields["worktree"]).expanduser().resolve(),
            head_oid=fields.get("HEAD") or None,
            ref_name=fields.get("branch") or None,
            detached="detached" in fields,
        )

    @staticmethod
    def _validate_eigent_ref(
        ref_name: str,
        *,
        branch_only: bool = False,
    ) -> None:
        allowed_prefixes = (
            ("refs/heads/eigent/",)
            if branch_only
            else ("refs/eigent/", "refs/heads/eigent/")
        )
        if not ref_name.startswith(allowed_prefixes) or not re.fullmatch(
            r"refs/(?:heads/)?eigent/[A-Za-z0-9._/-]+",
            ref_name,
        ):
            raise ValueError("ref must be inside an Eigent-owned namespace")

    def _operation_state(self, repository_root: Path) -> str:
        markers = (
            ("MERGE_HEAD", "merge"),
            ("rebase-merge", "rebase"),
            ("rebase-apply", "rebase"),
            ("CHERRY_PICK_HEAD", "cherry-pick"),
            ("REVERT_HEAD", "revert"),
        )
        for marker, state in markers:
            result = self._run(
                repository_root,
                ("rev-parse", "--git-path", marker),
            )
            path = Path(result.stdout.strip())
            if not path.is_absolute():
                path = repository_root / path
            if path.exists():
                return state
        return "clean"

    @staticmethod
    def _validate_object_name(value: str) -> None:
        if not re.fullmatch(r"[0-9a-fA-F]{4,64}", value):
            raise ValueError("Git object id must be hexadecimal")

    @staticmethod
    def _normalize_relative_git_path(value: str) -> str:
        path = PurePosixPath(value)
        if (
            not value
            or path.is_absolute()
            or ".." in path.parts
            or value.startswith(("~/", "\\\\"))
            or (len(value) > 1 and value[1] == ":")
        ):
            raise ValueError("Git path must stay inside the repository")
        normalized = path.as_posix()
        if normalized in {"", "."}:
            raise ValueError("Git path must name a file")
        return normalized

    def _relative_pathspecs(
        self,
        repository_root: Path,
        paths: tuple[Path, ...],
    ) -> tuple[str, ...]:
        if not paths:
            raise ValueError("at least one Git path is required")
        root = repository_root.expanduser().resolve()
        pathspecs: list[str] = []
        for path in paths:
            candidate = path.expanduser()
            if not candidate.is_absolute():
                candidate = root / candidate
            resolved = candidate.resolve()
            try:
                relative = resolved.relative_to(root)
            except ValueError as exc:
                raise GitBackendError(
                    f"Git path escapes repository root: {path}"
                ) from exc
            if not relative.parts:
                raise GitBackendError(
                    "repository root is not a valid pathspec"
                )
            pathspecs.append(relative.as_posix())
        return tuple(pathspecs)

    @staticmethod
    def _status_metadata(repository_root: Path, status_output: str) -> str:
        records = status_output.split("\0")
        metadata: list[str] = []
        skip_next = False
        for record in records:
            if not record:
                continue
            if skip_next:
                skip_next = False
                continue
            if len(record) < 4:
                continue
            state = record[:2]
            relative_path = record[3:]
            if "R" in state or "C" in state:
                skip_next = True
            path = repository_root / relative_path
            try:
                stat = path.lstat()
                value = (
                    f"{relative_path}\0{stat.st_mode}\0{stat.st_size}\0"
                    f"{stat.st_mtime_ns}\0{stat.st_ino}"
                )
            except FileNotFoundError:
                value = f"{relative_path}\0missing"
            metadata.append(value)
        return "\0".join(sorted(metadata))

    def assert_no_clean_filters(
        self,
        repository_root: Path,
        pathspecs: tuple[str, ...],
    ) -> None:
        attributes = self._run(
            repository_root,
            ("check-attr", "-a", "-z", "--", *pathspecs),
        ).stdout.split("\0")
        for index in range(0, len(attributes) - 2, 3):
            path, attribute, value = attributes[index : index + 3]
            if attribute == "filter" and value not in {
                "unspecified",
                "unset",
                "",
            }:
                raise GitBackendError(
                    f"refusing to execute clean filter {value!r} for {path!r}"
                )

    def _run(
        self,
        cwd: Path,
        args: tuple[str, ...],
        *,
        check: bool = True,
        identity: tuple[str, str] | None = None,
        input_text: str | None = None,
    ) -> GitCommandResult:
        environment = self._environment(identity=identity)
        command = self._command(cwd, args)
        try:
            completed = subprocess.run(
                command,
                check=False,
                capture_output=True,
                text=True,
                input=input_text,
                timeout=self.timeout_seconds,
                env=environment,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            raise GitBackendError(
                f"failed to execute typed Git operation: {args[0]}"
            ) from exc
        stdout = completed.stdout[: self.max_output_chars]
        stderr = completed.stderr[: self.max_output_chars]
        if check and completed.returncode != 0:
            raise GitCommandError(
                args=args,
                returncode=completed.returncode,
                stderr=stderr.strip(),
            )
        return GitCommandResult(
            stdout=stdout,
            stderr=stderr,
            returncode=completed.returncode,
        )

    def _environment(
        self,
        *,
        identity: tuple[str, str] | None = None,
    ) -> dict[str, str]:
        environment: dict[str, str] = dict(os.environ)
        for key in tuple(environment):
            if (
                key in _UNSAFE_INHERITED_GIT_ENV
                or key.startswith("GIT_AUTHOR_")
                or key.startswith("GIT_COMMITTER_")
                or key.startswith("GIT_CONFIG_KEY_")
                or key.startswith("GIT_CONFIG_VALUE_")
            ):
                environment.pop(key)
        environment.update(
            {
                "GIT_EDITOR": "true",
                "GIT_CONFIG_NOSYSTEM": "1",
                "GIT_CONFIG_GLOBAL": os.devnull,
                "GIT_LITERAL_PATHSPECS": "1",
                "GIT_MERGE_AUTOEDIT": "no",
                "GIT_PAGER": "cat",
                "GIT_SEQUENCE_EDITOR": "true",
                "GIT_TERMINAL_PROMPT": "0",
                "LC_ALL": "C",
                "PAGER": "cat",
            }
        )
        if identity is not None:
            name, email = identity
            environment.update(
                {
                    "GIT_AUTHOR_NAME": name,
                    "GIT_AUTHOR_EMAIL": email,
                    "GIT_COMMITTER_NAME": name,
                    "GIT_COMMITTER_EMAIL": email,
                }
            )
        return environment

    def _command(self, cwd: Path, args: tuple[str, ...]) -> tuple[str, ...]:
        return (
            self.git_executable,
            "-c",
            f"core.hooksPath={self.hooks_path}",
            "-c",
            "core.fsmonitor=false",
            "-c",
            "core.untrackedCache=false",
            "-c",
            "protocol.ext.allow=never",
            "-c",
            "protocol.file.allow=never",
            "-C",
            str(cwd),
            *args,
        )