codeflash-agent/packages/github-app/github_app/git.py

"""Git operations: repo cloning and workspace management."""

from __future__ import annotations

import asyncio
import logging
import tempfile
from pathlib import Path

log = logging.getLogger(__name__)


def _validate_clone_args(
    owner: str,
    repo: str,
    workspace: Path,
) -> None:
    """Reject owner/repo values that could escape the workspace."""
    for name, value in [("owner", owner), ("repo", repo)]:
        if "/" in value or ".." in value:
            msg = f"Invalid {name}: {value!r}"
            raise ValueError(msg)


async def clone_repo(
    owner: str,
    repo: str,
    ref: str,
    token: str,
    workspace: Path,
) -> Path:
    """Shallow-clone a repo at the given ref into a temp directory."""
    _validate_clone_args(owner, repo, workspace)

    # Ensure workspace exists before creating temp dir inside it.
    workspace.mkdir(parents=True, exist_ok=True)

    # Atomic unique directory -- avoids race conditions and rmtree.
    repo_dir = Path(
        tempfile.mkdtemp(
            prefix=f"{owner}_{repo}_",
            dir=workspace,
        ),
    )

    if not str(repo_dir.resolve()).startswith(
        str(workspace.resolve()),
    ):
        msg = f"Path escapes workspace: {repo_dir}"
        raise ValueError(msg)

    clone_url = f"https://x-access-token:{token}@github.com/{owner}/{repo}.git"
    proc = await asyncio.create_subprocess_exec(
        "git",
        "clone",
        "--depth=1",
        "--branch",
        ref,
        clone_url,
        str(repo_dir),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    _, stderr = await proc.communicate()
    if proc.returncode != 0:
        log.error(
            "git clone failed (rc=%d): %s",
            proc.returncode,
            stderr.decode(),
        )
        msg = f"git clone failed for {owner}/{repo} ref={ref}"
        raise RuntimeError(msg)
    return repo_dir


async def commit_and_push(
    repo_dir: Path,
    branch: str,
    owner: str,
    repo: str,
    message: str = "codeflash-agent: optimize code",
) -> bool:
    """Stage all changes, commit, and push back to the PR branch.

    Returns ``True`` if changes were committed and pushed.
    """
    # Stage everything
    proc = await asyncio.create_subprocess_exec(
        "git",
        "add",
        "-A",
        cwd=str(repo_dir),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    await proc.communicate()

    # Check if there are staged changes
    proc = await asyncio.create_subprocess_exec(
        "git",
        "diff",
        "--cached",
        "--quiet",
        cwd=str(repo_dir),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    await proc.communicate()
    if proc.returncode == 0:
        log.info("No changes to commit in %s", repo_dir)
        return False

    # Commit
    proc = await asyncio.create_subprocess_exec(
        "git",
        "commit",
        "-m",
        message,
        cwd=str(repo_dir),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    _, stderr = await proc.communicate()
    if proc.returncode != 0:
        log.error("git commit failed: %s", stderr.decode())
        return False

    # Swap remote URL to use gh credential helper instead of the
    # installation token (which may lack push permission).
    plain_url = f"https://github.com/{owner}/{repo}.git"
    proc = await asyncio.create_subprocess_exec(
        "git",
        "remote",
        "set-url",
        "origin",
        plain_url,
        cwd=str(repo_dir),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    await proc.communicate()

    # Push to the PR branch
    proc = await asyncio.create_subprocess_exec(
        "git",
        "push",
        "origin",
        f"HEAD:{branch}",
        cwd=str(repo_dir),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    _, stderr = await proc.communicate()
    if proc.returncode != 0:
        log.error("git push failed: %s", stderr.decode())
        return False

    log.info("Pushed optimization commit to %s", branch)
    return True