Git Operations

git

Generic git operations for CI pipelines.

Host-side git operations: clone, push, branch creation, diff inspection. All operations use subprocess calls to git.

RemoteLease

Bases: Enum

Special values for push_branch(expected_remote_sha=...).

TRACKING = 'tracking' class-attribute instance-attribute

Bare --force-with-lease: git reads the expected value from the local remote-tracking ref (the default, and the behavior before RHAI-3020).

ABSENT = 'absent' class-attribute instance-attribute

The remote branch must not exist yet: --force-with-lease=refs/heads/<branch>:.

GitControlTamperError

Bases: RuntimeError

The agent replaced .git itself, so the host copy cannot be restored.

GitControlSnapshot(repo_dir, dot_git, entries) dataclass

Host copy of the .git files that decide what host-side git executes.

Taken by :func:snapshot_git_control before an agent can write the repository and put back by :func:restore_git_control afterwards.

GitDiffError

Bases: Exception

Raised when git diff fails (missing ref, not a repo, etc.).

extract_repo_url(text)

Extract a repo URL from text, validating against forge APIs.

Filters out subpaths, file extensions, and placeholder URLs. Returns the first URL that resolves to a real project, or the first unvalidated candidate if no API tokens are available.

Source code in src/agentic_ci/git.py

def extract_repo_url(text: str) -> str | None:
    """Extract a repo URL from text, validating against forge APIs.

    Filters out subpaths, file extensions, and placeholder URLs.
    Returns the first URL that resolves to a real project, or the first
    unvalidated candidate if no API tokens are available.
    """
    candidates = _collect_candidates(text, _GITLAB_URL_RE)
    has_token = bool(os.environ.get("BOT_PAT") or os.environ.get("GITLAB_TOKEN"))
    if candidates and has_token:
        for url in candidates:
            if _validate_gitlab_url(url):
                return url
    if candidates:
        return candidates[0]

    candidates = _collect_candidates(text, _GITHUB_URL_RE)
    if candidates:
        for url in candidates:
            if _validate_github_url(url):
                return url
        return candidates[0]

    return None

extract_all_repo_urls(text)

Extract all distinct repo root URLs from text.

Scans for both GitLab and GitHub URLs, filters out subpaths, file extensions, and placeholder URLs. GitLab URLs that are strict prefixes of other GitLab URLs are collapsed (nested group dedup).

Unlike :func:extract_repo_url, this does not validate URLs against forge APIs -- it returns all plausible candidates.

Source code in src/agentic_ci/git.py

def extract_all_repo_urls(text: str) -> list[str]:
    """Extract all distinct repo root URLs from text.

    Scans for both GitLab and GitHub URLs, filters out subpaths, file
    extensions, and placeholder URLs.  GitLab URLs that are strict
    prefixes of other GitLab URLs are collapsed (nested group dedup).

    Unlike :func:`extract_repo_url`, this does **not** validate URLs
    against forge APIs -- it returns all plausible candidates.
    """
    gitlab_urls = _dedup_gitlab_prefixes(_collect_candidates(text, _GITLAB_URL_RE))
    github_urls = _collect_candidates(text, _GITHUB_URL_RE)
    seen: set[str] = set()
    result: list[str] = []
    for url in gitlab_urls + github_urls:
        if url not in seen:
            seen.add(url)
            result.append(url)
    return result

validate_repo_url(url)

Check that a repo URL points to an allowed host with no path traversal.

Source code in src/agentic_ci/git.py

def validate_repo_url(url: str) -> bool:
    """Check that a repo URL points to an allowed host with no path traversal."""
    if not url:
        return False
    parsed = urlparse(url)
    if parsed.scheme != "https":
        return False
    if not parsed.hostname or parsed.hostname not in ALLOWED_HOSTS:
        return False
    if parsed.username or parsed.password:
        return False
    if ".." in (parsed.path or ""):
        return False
    return True

validate_branch_exists(repo_url, branch)

Check if a branch exists on the remote repository.

Parameters:

Name	Type	Description	Default
repo_url	str	HTTPS URL of the git repository	required
branch	str	Branch name to validate	required

Returns:

Type	Description
bool	True if the branch exists on the remote, False otherwise

Note

Returns False for any error condition (network issues, invalid refs, etc.) to allow graceful fallback in the resolution chain.

Source code in src/agentic_ci/git.py

def validate_branch_exists(repo_url: str, branch: str) -> bool:
    """Check if a branch exists on the remote repository.

    Args:
        repo_url: HTTPS URL of the git repository
        branch: Branch name to validate

    Returns:
        True if the branch exists on the remote, False otherwise

    Note:
        Returns False for any error condition (network issues, invalid refs, etc.)
        to allow graceful fallback in the resolution chain.
    """
    if not _validate_ref(branch):
        log.warning("Invalid branch name rejected: %s", branch)
        return False

    try:
        result = subprocess.run(
            ["git", "ls-remote", "--heads", repo_url, branch],
            capture_output=True,
            text=True,
            timeout=30,
            stdin=_DEVNULL,
        )

        if result.returncode != 0:
            log.debug(
                "git ls-remote failed for %s branch %s: %s", repo_url, branch, result.stderr.strip()
            )
            return False

        output = result.stdout.strip()
        if not output:
            log.debug("Branch %s does not exist on remote %s", branch, repo_url)
            return False

        log.debug("Branch %s exists on remote %s", branch, repo_url)
        return True

    except subprocess.TimeoutExpired:
        log.warning("Branch validation timed out for %s branch %s", repo_url, branch)
        return False
    except (subprocess.CalledProcessError, FileNotFoundError) as exc:
        log.debug("Branch validation failed for %s branch %s: %s", repo_url, branch, exc)
        return False

clone_repo(url, dest, branch=None, depth=None)

Clone a repository. Returns True on success.

Source code in src/agentic_ci/git.py

def clone_repo(url: str, dest: Path, branch: str | None = None, depth: int | None = None) -> bool:
    """Clone a repository. Returns True on success."""
    if not validate_repo_url(url):
        log.error("clone_repo: invalid or disallowed URL: %s", url)
        return False
    if branch and not _validate_ref(branch):
        log.error("clone_repo: invalid branch name: %s", branch)
        return False
    cmd = [
        "git",
        "-c",
        "protocol.ext.allow=never",
        "-c",
        "protocol.file.allow=never",
        "clone",
    ]
    if depth:
        cmd += ["--depth", str(depth)]
    if branch:
        cmd += ["--branch", branch]
    cmd += ["--", url, str(dest)]
    try:
        subprocess.run(
            cmd,
            check=True,
            capture_output=True,
            text=True,
            timeout=GIT_CLONE_TIMEOUT,
            stdin=_DEVNULL,
        )
        subprocess.run(
            ["git", "config", "--global", "--add", "safe.directory", str(dest.resolve())],
            capture_output=True,
            text=True,
        )
        return True
    except subprocess.TimeoutExpired:
        log.error("git clone timed out after %ds for %s", GIT_CLONE_TIMEOUT, url)
        return False
    except subprocess.CalledProcessError as exc:
        log.error("git clone failed: %s", exc.stderr)
        return False

create_branch(repo_dir, branch_name)

Create and checkout a new branch.

Source code in src/agentic_ci/git.py

def create_branch(repo_dir: Path, branch_name: str) -> bool:
    """Create and checkout a new branch."""
    if not _validate_ref(branch_name):
        log.error("create_branch: invalid branch name: %s", branch_name)
        return False
    try:
        subprocess.run(
            ["git", "switch", "-c", branch_name],
            cwd=str(repo_dir),
            check=True,
            capture_output=True,
            text=True,
        )
        return True
    except subprocess.CalledProcessError as exc:
        log.error("git switch -c failed: %s", exc.stderr)
        return False

checkout_branch(repo_dir, branch)

Checkout an existing branch. Returns True on success.

Source code in src/agentic_ci/git.py

def checkout_branch(repo_dir: Path, branch: str) -> bool:
    """Checkout an existing branch. Returns True on success."""
    if not _validate_ref(branch):
        log.error("checkout_branch: invalid branch name: %s", branch)
        return False
    try:
        subprocess.run(
            ["git", "checkout", branch],
            cwd=str(repo_dir),
            check=True,
            capture_output=True,
            text=True,
            stdin=_DEVNULL,
        )
        return True
    except subprocess.CalledProcessError as exc:
        log.error("git checkout failed: %s", exc.stderr)
        return False
    except FileNotFoundError:
        log.error("git binary not found")
        return False

rebase_branch(repo_dir, onto)

Rebase the current branch onto onto. Returns True on success.

On conflict the rebase is aborted so the worktree stays clean.

Source code in src/agentic_ci/git.py

def rebase_branch(repo_dir: Path, onto: str) -> bool:
    """Rebase the current branch onto *onto*. Returns True on success.

    On conflict the rebase is aborted so the worktree stays clean.
    """
    if not _validate_ref(onto):
        log.error("rebase_branch: invalid ref: %s", onto)
        return False
    try:
        subprocess.run(
            ["git", "rebase", onto],
            cwd=str(repo_dir),
            check=True,
            capture_output=True,
            text=True,
            stdin=_DEVNULL,
        )
        return True
    except subprocess.CalledProcessError as exc:
        log.error("git rebase failed: %s", exc.stderr)
        subprocess.run(
            ["git", "rebase", "--abort"],
            cwd=str(repo_dir),
            capture_output=True,
            text=True,
            stdin=_DEVNULL,
        )
        return False
    except FileNotFoundError:
        log.error("git binary not found")
        return False

get_default_branch(repo_dir)

Detect the default branch of the remote origin.

Runs git rev-parse --abbrev-ref origin/HEAD and strips the origin/ prefix. Falls back to "main" when the remote HEAD cannot be determined.

Source code in src/agentic_ci/git.py

def get_default_branch(repo_dir: Path) -> str:
    """Detect the default branch of the remote origin.

    Runs ``git rev-parse --abbrev-ref origin/HEAD`` and strips the
    ``origin/`` prefix. Falls back to ``"main"`` when the remote HEAD
    cannot be determined.
    """
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--abbrev-ref", "origin/HEAD"],
            cwd=str(repo_dir),
            check=True,
            capture_output=True,
            text=True,
            stdin=_DEVNULL,
        )
        ref = result.stdout.strip()
        if ref and ref != "origin/HEAD":
            return ref.removeprefix("origin/")
    except (subprocess.CalledProcessError, FileNotFoundError):
        pass
    return "main"

git_output(repo_dir, *args)

Run a git command and return its stripped stdout, or None on error.

This is a thin wrapper around subprocess.run for cases where the caller only needs the text output of a git command.

Source code in src/agentic_ci/git.py

def git_output(repo_dir: Path, *args: str) -> str | None:
    """Run a git command and return its stripped stdout, or None on error.

    This is a thin wrapper around ``subprocess.run`` for cases where
    the caller only needs the text output of a git command.
    """
    try:
        result = subprocess.run(
            ["git", *args],
            cwd=str(repo_dir),
            check=True,
            capture_output=True,
            text=True,
            stdin=_DEVNULL,
        )
        return result.stdout.strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        return None

push_branch(repo_dir, remote='origin', branch=None, *, max_retries=GIT_PUSH_MAX_RETRIES, retry_delay=GIT_PUSH_RETRY_DELAY, expected_remote_sha=RemoteLease.TRACKING)

Push a branch to remote with a force-with-lease. Returns True on success.

Retries up to max_retries times on transient errors (server 5xx, commit_refs failures, lock contention, network resets) with exponential backoff starting at retry_delay seconds. A lease rejection (git reports stale info) is never retried and is logged as such: someone else moved the remote branch.

Parameters:

Name	Type	Description	Default
repo_dir	Path	Local repository to push from.	required
remote	str	Remote name (not a URL).	'origin'
branch	str | None	Branch to push. Defaults to the checked-out branch. Required when expected_remote_sha is given, and then it must be a short name (not starting with refs/) and is pushed as refs/heads/<branch>:refs/heads/<branch>, so the pushed ref is the leased ref even when a tag has the same name.	None
max_retries	int	Retries after the first attempt on transient errors.	GIT_PUSH_MAX_RETRIES
retry_delay	float	Initial backoff in seconds.	GIT_PUSH_RETRY_DELAY
expected_remote_sha	str | RemoteLease	What the remote branch must point at for the push to go through. The default, RemoteLease.TRACKING, runs a bare --force-with-lease, whose expected value comes from the local remote-tracking ref; anyone who can write .git (for example an agent run on the working copy) can move that ref or remap it through remote.<name>.fetch. A full commit SHA (40 or 64 hex characters, recorded before untrusted code ran) runs --force-with-lease=refs/heads/<branch>:<sha>, and RemoteLease.ABSENT runs --force-with-lease=refs/heads/<branch>:, which only creates the branch. Both are checked by the remote against the given value, independent of local refs and config. Any other value fails closed: nothing is pushed, an error is logged and False is returned. Abbreviated SHAs are rejected because git would resolve them in the local repository. The name says "sha" because a commit SHA is the usual value; the two RemoteLease members are the only non-SHA values, one keeping the old default and one meaning "no such branch". The default is kept for compatibility and logs at debug level so unpinned callers can be found.	TRACKING

Source code in src/agentic_ci/git.py

def push_branch(
    repo_dir: Path,
    remote: str = "origin",
    branch: str | None = None,
    *,
    max_retries: int = GIT_PUSH_MAX_RETRIES,
    retry_delay: float = GIT_PUSH_RETRY_DELAY,
    expected_remote_sha: str | RemoteLease = RemoteLease.TRACKING,
) -> bool:
    """Push a branch to *remote* with a force-with-lease. Returns True on success.

    Retries up to *max_retries* times on transient errors (server 5xx,
    ``commit_refs`` failures, lock contention, network resets) with
    exponential backoff starting at *retry_delay* seconds. A lease
    rejection (git reports ``stale info``) is never retried and is
    logged as such: someone else moved the remote branch.

    Args:
        repo_dir: Local repository to push from.
        remote: Remote name (not a URL).
        branch: Branch to push. Defaults to the checked-out branch.
            Required when *expected_remote_sha* is given, and then it
            must be a short name (not starting with ``refs/``) and is
            pushed as ``refs/heads/<branch>:refs/heads/<branch>``, so
            the pushed ref is the leased ref even when a tag has the
            same name.
        max_retries: Retries after the first attempt on transient errors.
        retry_delay: Initial backoff in seconds.
        expected_remote_sha: What the remote branch must point at for
            the push to go through. The default,
            ``RemoteLease.TRACKING``, runs a bare ``--force-with-lease``,
            whose expected value comes from the local remote-tracking
            ref; anyone who can write ``.git`` (for example an agent run
            on the working copy) can move that ref or remap it through
            ``remote.<name>.fetch``. A full commit SHA (40 or 64 hex
            characters, recorded before untrusted code ran) runs
            ``--force-with-lease=refs/heads/<branch>:<sha>``, and
            ``RemoteLease.ABSENT`` runs
            ``--force-with-lease=refs/heads/<branch>:``, which only
            creates the branch. Both are checked by the remote against
            the given value, independent of local refs and config. Any
            other value fails closed: nothing is pushed, an error is
            logged and False is returned. Abbreviated SHAs are rejected
            because git would resolve them in the local repository.
            The name says "sha" because a commit SHA is the usual
            value; the two ``RemoteLease`` members are the only
            non-SHA values, one keeping the old default and one
            meaning "no such branch". The default is kept for
            compatibility and logs at debug level so unpinned callers
            can be found.
    """
    if not remote or remote.startswith("-") or ".." in remote or "@{" in remote:
        log.error("push_branch: invalid remote name: %s", remote)
        return False
    if not _SAFE_REMOTE_RE.match(remote):
        log.error("push_branch: invalid remote name: %s", remote)
        return False
    if branch and not _validate_ref(branch):
        log.error("push_branch: invalid branch name: %s", branch)
        return False
    if expected_remote_sha is RemoteLease.TRACKING:
        lease = "--force-with-lease"
        lease_desc = "the remote-tracking ref"
    elif not branch:
        log.error("push_branch: an explicit branch is required with expected_remote_sha")
        return False
    elif branch.startswith("refs/"):
        log.error(
            "push_branch: expected_remote_sha needs a short branch name, not a full ref: %s",
            branch,
        )
        return False
    elif expected_remote_sha is RemoteLease.ABSENT:
        lease = f"--force-with-lease=refs/heads/{branch}:"
        lease_desc = "no remote branch"
    elif isinstance(expected_remote_sha, str) and _FULL_OID_RE.fullmatch(expected_remote_sha):
        sha = expected_remote_sha.lower()
        lease = f"--force-with-lease=refs/heads/{branch}:{sha}"
        lease_desc = sha
    else:
        log.error(
            "push_branch: invalid expected_remote_sha %r: need a full commit SHA "
            "(40 or 64 hex characters) or a RemoteLease value",
            expected_remote_sha,
        )
        return False
    if not branch:
        try:
            result = subprocess.run(
                ["git", "rev-parse", "--abbrev-ref", "HEAD"],
                cwd=str(repo_dir),
                check=True,
                capture_output=True,
                text=True,
                stdin=_DEVNULL,
            )
            branch = result.stdout.strip()
        except subprocess.CalledProcessError:
            log.error("push_branch: could not detect current branch")
            return False
        if not _validate_ref(branch):
            log.error("push_branch: detected invalid branch name: %s", branch)
            return False
    max_retries = max(0, max_retries)
    if not math.isfinite(retry_delay) or retry_delay < 0:
        retry_delay = 5.0

    if expected_remote_sha is RemoteLease.TRACKING:
        log.debug(
            "push_branch: %s/%s uses a bare --force-with-lease (lease from the "
            "remote-tracking ref); pass expected_remote_sha to pin it",
            remote,
            branch,
        )
        refspec = branch
    else:
        # Push exactly the ref the lease names. A bare <branch> would
        # resolve to refs/tags/<branch> when no such local branch exists
        # and push that tag, which the refs/heads/<branch> lease does not
        # cover.
        refspec = f"refs/heads/{branch}:refs/heads/{branch}"

    cmd = ["git", "push", lease, "--set-upstream", remote, refspec]
    total_attempts = 1 + max_retries

    @retry(
        stop=stop_after_attempt(total_attempts),
        wait=wait_exponential(multiplier=retry_delay, exp_base=2, min=0),
        retry=retry_if_exception_type(_TransientPushError),
        sleep=time.sleep,
        reraise=True,
        before_sleep=lambda rs: log.warning(
            "git push failed (attempt %d/%d), retrying in %.0fs: %s",
            rs.attempt_number,
            total_attempts,
            rs.next_action.sleep if rs.next_action else 0,
            str(rs.outcome.exception()) if rs.outcome else "unknown",
        ),
    )
    def _do_push() -> None:
        try:
            subprocess.run(
                cmd,
                cwd=str(repo_dir),
                check=True,
                capture_output=True,
                text=True,
                timeout=GIT_PUSH_TIMEOUT,
                stdin=_DEVNULL,
            )
        except subprocess.TimeoutExpired:
            raise _TransientPushError("git push timed out")
        except subprocess.CalledProcessError as exc:
            stderr = exc.stderr or ""
            if _is_stale_lease_rejection(stderr):
                log.error(
                    "git push rejected: %s/%s no longer matches the lease (expected %s); "
                    "someone else updated it, not overwriting or retrying: %s",
                    remote,
                    branch,
                    lease_desc,
                    stderr.strip(),
                )
                raise
            if _is_transient_push_error(stderr):
                raise _TransientPushError(stderr.strip()) from exc
            log.error("git push failed: %s", stderr.strip())
            raise

    try:
        _do_push()
        return True
    except _TransientPushError as exc:
        log.error("git push failed: %s", str(exc))
        return False
    except subprocess.CalledProcessError:
        return False

setup_git_config(repo_dir, name, email)

Set local git user config.

Source code in src/agentic_ci/git.py

def setup_git_config(repo_dir: Path, name: str, email: str) -> None:
    """Set local git user config."""
    subprocess.run(
        ["git", "config", "user.name", name],
        cwd=str(repo_dir),
        check=True,
        capture_output=True,
        text=True,
    )
    subprocess.run(
        ["git", "config", "user.email", email],
        cwd=str(repo_dir),
        check=True,
        capture_output=True,
        text=True,
    )

harden_git_config(repo_dir)

Apply security hardening to git config (disable hooks, fsmonitor).

Source code in src/agentic_ci/git.py

def harden_git_config(repo_dir: Path) -> None:
    """Apply security hardening to git config (disable hooks, fsmonitor)."""
    for key, value in [
        ("core.hooksPath", "/dev/null"),
        ("core.fsmonitor", "false"),
    ]:
        subprocess.run(
            ["git", "config", key, value],
            cwd=str(repo_dir),
            check=True,
            capture_output=True,
            text=True,
        )

snapshot_git_control(repo_dir)

Record the git control files of the repository at repo_dir.

Call this on the host before an agent can write repo_dir (for example before a sandbox upload or a container bind mount), after any hardening such as :func:harden_git_config. Pass the result to :func:restore_git_control once the agent can no longer write the repository.

Source code in src/agentic_ci/git.py

def snapshot_git_control(repo_dir: Path) -> GitControlSnapshot:
    """Record the git control files of the repository at *repo_dir*.

    Call this on the host before an agent can write *repo_dir* (for example
    before a sandbox upload or a container bind mount), after any hardening
    such as :func:`harden_git_config`. Pass the result to
    :func:`restore_git_control` once the agent can no longer write the
    repository.
    """
    repo_dir = Path(repo_dir)
    dot_git = repo_dir / ".git"
    top = _read_entry(dot_git)
    entries = _read_control(dot_git) if top is not None and top.kind == "dir" else {}
    return GitControlSnapshot(repo_dir=repo_dir, dot_git=top, entries=entries)

restore_git_control(snapshot)

Put the host's git control files back after an agent had write access.

Everything under :data:GIT_CONTROL_PATHS is moved out of git's reach and rewritten from snapshot, so config keys, hooks, attributes and commondir redirects added by the agent are gone before any host git command runs, while its commits, refs and index are kept. Nothing the agent wrote is read or walked before that, so no tree it built can make the restore fail early, and symlinks it planted are never written through.

A .git the agent created in a workdir that had none is removed.

Returns the control paths the agent had changed (best effort, for the log). Raises :class:GitControlTamperError when .git was a directory and is now missing, a symlink or a file (that entry is deleted first), or when the host copy cannot be put back; .git is then moved aside so host git cannot use the agent's config.

Source code in src/agentic_ci/git.py

def restore_git_control(snapshot: GitControlSnapshot) -> list[str]:
    """Put the host's git control files back after an agent had write access.

    Everything under :data:`GIT_CONTROL_PATHS` is moved out of git's reach and
    rewritten from *snapshot*, so config keys, hooks, attributes and
    ``commondir`` redirects added by the agent are gone before any host git
    command runs, while its commits, refs and index are kept. Nothing the
    agent wrote is read or walked before that, so no tree it built can make
    the restore fail early, and symlinks it planted are never written through.

    A ``.git`` the agent created in a workdir that had none is removed.

    Returns the control paths the agent had changed (best effort, for the
    log). Raises :class:`GitControlTamperError` when ``.git`` was a directory
    and is now missing, a symlink or a file (that entry is deleted first), or
    when the host copy cannot be put back; ``.git`` is then moved aside so
    host git cannot use the agent's config.
    """
    dot_git = snapshot.repo_dir / ".git"
    try:
        current = os.lstat(dot_git)
    except FileNotFoundError:
        current = None
    if snapshot.dot_git is None:
        if current is None:
            return []
        # Host git run from the workdir would use this .git and its config,
        # even when the workdir sits inside another repository.
        try:
            _discard(dot_git)
        except OSError as exc:
            raise GitControlTamperError(
                f"The agent created {dot_git} and it could not be removed ({exc}); "
                "host git must not run in this workdir"
            ) from exc
        log.warning("Removed the .git the agent created in %s", snapshot.repo_dir)
        return [".git"]
    if snapshot.dot_git.kind != "dir":
        # A gitfile or symlink pointing at a git dir outside the workdir,
        # which the agent cannot reach. Put the pointer itself back every
        # time, like the directory case below, so its host mode returns too;
        # only a change that decides what git reads is reported.
        changed = _differs(dot_git, ".git", {".git": snapshot.dot_git})
        try:
            if current is not None:
                _discard(dot_git)
            _write_entry(dot_git, snapshot.dot_git)
        except OSError as exc:
            raise GitControlTamperError(
                f"Could not restore {dot_git} ({exc}); host git must not use this repository"
            ) from exc
        return [".git"] if changed else []
    if current is None or not stat.S_ISDIR(current.st_mode):
        what = "deleted"
        if current is not None:
            kind = "symlink" if stat.S_ISLNK(current.st_mode) else "file"
            what = f"replaced by a {kind}"
            _remove(dot_git)
        raise GitControlTamperError(
            f"{dot_git} was {what} during the agent run; host git must not use this repository"
        )

    # Nothing the agent wrote is read before it is out of git's reach: each
    # control path is renamed into a quarantine directory, which git never
    # reads, and only then compared with the snapshot for the log.
    quarantine = dot_git / f"agentic-ci-untrusted-{uuid.uuid4().hex}"
    try:
        os.chmod(dot_git, snapshot.dot_git.mode)
        os.mkdir(quarantine, 0o700)
        for rel in GIT_CONTROL_PATHS:
            try:
                os.rename(dot_git / rel, quarantine / rel)
            except FileNotFoundError:
                pass
        # Sorted keys create each directory before its children.
        for rel in sorted(snapshot.entries):
            _write_entry(dot_git / rel, snapshot.entries[rel])
    except OSError as exc:
        _fail_closed(dot_git, f"Could not restore the git control files in {dot_git} ({exc})")

    changed = [
        rel for rel in GIT_CONTROL_PATHS if _differs(quarantine / rel, rel, snapshot.entries)
    ]
    try:
        _remove(quarantine)
    except OSError as exc:
        log.warning("Could not delete %s: %s", quarantine, exc)
    if changed:
        log.warning(
            "Agent changed git control files in %s; restored host copy of: %s",
            dot_git,
            ", ".join(changed),
        )
    return changed

discard_git_dir(repo_dir)

Move repo_dir/.git aside and delete it.

For when the host copy cannot be restored safely, for example because an agent process may still be running and able to write .git. Host git then finds no repository in repo_dir instead of the agent's config. The agent's commits are lost with it. Failures are logged, not raised, so the caller's own error is what propagates.

Source code in src/agentic_ci/git.py

def discard_git_dir(repo_dir: Path) -> None:
    """Move ``repo_dir/.git`` aside and delete it.

    For when the host copy cannot be restored safely, for example because an
    agent process may still be running and able to write ``.git``. Host git
    then finds no repository in *repo_dir* instead of the agent's config. The
    agent's commits are lost with it. Failures are logged, not raised, so the
    caller's own error is what propagates.
    """
    dot_git = Path(repo_dir) / ".git"
    if not os.path.lexists(dot_git):
        return
    try:
        _discard(dot_git)
    except OSError as exc:
        log.error("Could not move %s aside (%s); host git must not use it", dot_git, exc)
        return
    log.warning("Moved %s aside: the agent could still write it", dot_git)

get_commit_info(repo_dir)

Get the latest commit info (committer, email, message, sha).

Uses committer identity (not author) so that rebased or cherry-picked commits always reflect the current git config.

Source code in src/agentic_ci/git.py

def get_commit_info(repo_dir: Path) -> dict:
    """Get the latest commit info (committer, email, message, sha).

    Uses committer identity (not author) so that rebased or
    cherry-picked commits always reflect the current git config.
    """
    fmt = "%H%n%ce%n%cn%n%s"
    result = subprocess.run(
        ["git", "log", "-1", f"--format={fmt}"],
        cwd=str(repo_dir),
        capture_output=True,
        text=True,
        check=True,
    )
    lines = result.stdout.strip().split("\n")
    if len(lines) < 4:
        return {}
    return {"sha": lines[0], "email": lines[1], "name": lines[2], "subject": lines[3]}

get_changed_files(repo_dir, base_ref='HEAD~1', *, include_deleted=True)

Return files changed between base_ref and HEAD (committed state only).

Uses the two-ref form git diff <base_ref> HEAD so that files still in HEAD after a failed git commit --amend are detected even when git rm --cached already removed them from the index.

Rename detection is off, so a moved file is listed by both its old and its new path: a check on protected paths sees a file moved out of them. Names are read NUL-separated and decoded with os.fsdecode, so they are never C-quoted and a name that is not UTF-8 cannot raise. git replace refs, which the agent can write, are ignored. With include_deleted False, paths that no longer exist in HEAD (deleted files and the old side of a move) are left out.

Raises GitDiffError if the git command fails.

Source code in src/agentic_ci/git.py

def get_changed_files(
    repo_dir: Path, base_ref: str = "HEAD~1", *, include_deleted: bool = True
) -> list[str]:
    """Return files changed between *base_ref* and HEAD (committed state only).

    Uses the two-ref form ``git diff <base_ref> HEAD`` so that files still in
    HEAD after a failed ``git commit --amend`` are detected even when
    ``git rm --cached`` already removed them from the index.

    Rename detection is off, so a moved file is listed by both its old and
    its new path: a check on protected paths sees a file moved out of them.
    Names are read NUL-separated and decoded with ``os.fsdecode``, so they
    are never C-quoted and a name that is not UTF-8 cannot raise.  ``git
    replace`` refs, which the agent can write, are ignored.  With
    *include_deleted* False, paths that no longer exist in HEAD (deleted
    files and the old side of a move) are left out.

    Raises GitDiffError if the git command fails.
    """
    if not _validate_ref(base_ref):
        raise GitDiffError(f"Invalid ref name: {base_ref}")
    cmd = ["git", "diff", "--no-renames", "--name-only", "-z"]
    if not include_deleted:
        cmd.append("--diff-filter=d")
    try:
        result = subprocess.run(
            [*cmd, base_ref, "HEAD"],
            cwd=str(repo_dir),
            capture_output=True,
            check=True,
            env={**os.environ, "GIT_NO_REPLACE_OBJECTS": "1"},
        )
    except subprocess.CalledProcessError as exc:
        stderr = (exc.stderr or b"").decode(errors="replace").strip()
        raise GitDiffError(f"git diff failed for base_ref={base_ref}: {stderr}") from exc
    return [os.fsdecode(name) for name in result.stdout.split(b"\0") if name]

unquote_git_path(path)

Undo git's C-quoting of a path ("caf\303\251" becomes café).

Git quotes names with unusual characters in its default (non -z) output. Only git's own escapes are decoded: the letter escapes, and octal escapes for the bytes git writes in octal (control characters, DEL and non-ASCII). A path without the surrounding double quotes is returned as is, and so is one with an escape git never writes, such as \057 for /, so an escape cannot hide a / or other printable character from a caller matching the result.

Source code in src/agentic_ci/git.py

def unquote_git_path(path: str) -> str:
    """Undo git's C-quoting of a path (``"caf\\303\\251"`` becomes ``café``).

    Git quotes names with unusual characters in its default (non ``-z``)
    output.  Only git's own escapes are decoded: the letter escapes, and
    octal escapes for the bytes git writes in octal (control characters,
    DEL and non-ASCII).  A path without the surrounding double quotes is
    returned as is, and so is one with an escape git never writes, such as
    ``\\057`` for ``/``, so an escape cannot hide a ``/`` or other
    printable character from a caller matching the result.
    """
    if len(path) < 2 or not (path.startswith('"') and path.endswith('"')):
        return path
    body = path[1:-1]
    raw = bytearray()
    i = 0
    while i < len(body):
        char = body[i]
        if char != "\\":
            raw += os.fsencode(char)
            i += 1
            continue
        octal = body[i + 1 : i + 4]
        if len(octal) == 3 and all(c in "01234567" for c in octal) and octal[0] in "0123":
            value = int(octal, 8)
            if 0x20 <= value < 0x7F:
                return path
            raw.append(value)
            i += 4
        elif body[i + 1 : i + 2] in _GIT_QUOTE_ESCAPES:
            raw.append(_GIT_QUOTE_ESCAPES[body[i + 1]])
            i += 2
        else:
            return path
    return os.fsdecode(bytes(raw))

display_path(path)

Return path safe to show in a log line or gate message.

get_changed_files() returns names as git stores them, so an agent-chosen name can hold a newline or other control character that would forge extra log lines. Such names come back with those characters (and any non-ASCII, undecodable bytes included) escaped; a printable name is returned unchanged.

Source code in src/agentic_ci/git.py

def display_path(path: str) -> str:
    """Return *path* safe to show in a log line or gate message.

    ``get_changed_files()`` returns names as git stores them, so an
    agent-chosen name can hold a newline or other control character that
    would forge extra log lines.  Such names come back with those
    characters (and any non-ASCII, undecodable bytes included) escaped; a
    printable name is returned unchanged.
    """
    if path.isprintable():
        return path
    return path.encode("unicode_escape", errors="backslashreplace").decode("ascii")

strip_committed_files(repo_dir, patterns, base_ref='origin/HEAD')

Remove files matching patterns from the latest commit.

Agents can bypass .git/info/exclude by explicitly naming files in git add. This function detects any committed files that match the given fnmatch patterns and amends the commit to remove them, keeping the working-tree copies intact.

Returns the list of file paths actually stripped (empty if none matched or all removals failed).

Source code in src/agentic_ci/git.py

def strip_committed_files(
    repo_dir: Path,
    patterns: list[str],
    base_ref: str = "origin/HEAD",
) -> list[str]:
    """Remove files matching *patterns* from the latest commit.

    Agents can bypass ``.git/info/exclude`` by explicitly naming files in
    ``git add``.  This function detects any committed files that match the
    given fnmatch *patterns* and amends the commit to remove them, keeping
    the working-tree copies intact.

    Returns the list of file paths actually stripped (empty if none matched
    or all removals failed).
    """
    try:
        changed = get_changed_files(repo_dir, base_ref=base_ref, include_deleted=False)
    except GitDiffError:
        return []

    to_remove = []
    for filepath in changed:
        name = Path(filepath).name
        for pattern in patterns:
            if fnmatch.fnmatch(name, pattern) or fnmatch.fnmatch(filepath, pattern):
                to_remove.append(filepath)
                break

    if not to_remove:
        return []

    log.warning(
        "Stripping %d artifact file(s) from commit: %s",
        len(to_remove),
        ", ".join(display_path(f) for f in to_remove),
    )
    actually_removed = []
    for filepath in to_remove:
        # The name is agent-chosen: "--" keeps a leading "-" from reading as
        # an option, and literal pathspecs keep "*" or ":" from matching
        # other files.
        result = subprocess.run(
            ["git", "rm", "--cached", "--quiet", "--", filepath],
            cwd=str(repo_dir),
            capture_output=True,
            text=True,
            env={**os.environ, "GIT_LITERAL_PATHSPECS": "1"},
        )
        if result.returncode != 0:
            log.error(
                "git rm --cached failed for %s (rc=%d): %s",
                display_path(filepath),
                result.returncode,
                result.stderr.strip(),
            )
        else:
            actually_removed.append(filepath)

    if actually_removed:
        result = subprocess.run(
            ["git", "commit", "--amend", "--no-edit", "--allow-empty"],
            cwd=str(repo_dir),
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            log.error(
                "git commit --amend failed after stripping (rc=%d): %s",
                result.returncode,
                result.stderr.strip(),
            )

    return actually_removed

setup_git_credentials(repo_url, *, github_token_resolver=None)

Configure git url.insteadOf for the forge hosting repo_url.

Sets up transparent credential injection so that clone_repo() and push_branch() (which use bare HTTPS URLs) can authenticate without modification.

For GitLab, reads BOT_PAT from the environment. For GitHub, calls github_token_resolver(repo_url) to obtain a short-lived token. If no resolver is provided for GitHub URLs, returns False.

Idempotent and safe to call multiple times. Returns True on success, False if credentials are unavailable.

Source code in src/agentic_ci/git.py

def setup_git_credentials(
    repo_url: str,
    *,
    github_token_resolver: Callable[[str], str | None] | None = None,
) -> bool:
    """Configure ``git url.insteadOf`` for the forge hosting *repo_url*.

    Sets up transparent credential injection so that ``clone_repo()`` and
    ``push_branch()`` (which use bare HTTPS URLs) can authenticate
    without modification.

    For **GitLab**, reads ``BOT_PAT`` from the environment.
    For **GitHub**, calls *github_token_resolver(repo_url)* to obtain a
    short-lived token. If no resolver is provided for GitHub URLs,
    returns False.

    Idempotent and safe to call multiple times. Returns True on success,
    False if credentials are unavailable.
    """
    if not repo_url:
        return False

    parsed = urlparse(repo_url)
    hostname = (parsed.hostname or "").lower()

    if hostname == "gitlab.com":
        return _setup_gitlab_credentials()
    elif hostname == "github.com":
        if github_token_resolver is None:
            log.error("No github_token_resolver provided for GitHub URL: %s", repo_url)
            return False
        return _setup_github_credentials(repo_url, github_token_resolver)

    log.info("No credential setup needed for URL: %s", repo_url)
    return True