open-swe/agent/utils/github.py

"""GitHub API and git utilities."""

from __future__ import annotations

import logging
import shlex

import httpx
from deepagents.backends.protocol import ExecuteResponse, SandboxBackendProtocol

logger = logging.getLogger(__name__)

# HTTP status codes
HTTP_CREATED = 201
HTTP_UNPROCESSABLE_ENTITY = 422


def is_permanent_github_push_failure(output: str) -> bool:
    """Return whether git push output indicates a permanent auth failure."""
    normalized_output = output.lower()
    return (
        "permanent_failure" in normalized_output
        or "403" in normalized_output
        or "permission" in normalized_output
        or "denied" in normalized_output
    )


def _run_git(
    sandbox_backend: SandboxBackendProtocol, repo_dir: str, command: str
) -> ExecuteResponse:
    """Run a git command in the sandbox repo directory."""
    safe_repo_dir = shlex.quote(repo_dir)
    return sandbox_backend.execute(f"cd {safe_repo_dir} && {command}")


def git_has_uncommitted_changes(sandbox_backend: SandboxBackendProtocol, repo_dir: str) -> bool:
    """Check whether the repo has uncommitted changes."""
    result = _run_git(sandbox_backend, repo_dir, "git status --porcelain")
    return result.exit_code == 0 and bool(result.output.strip())


def git_fetch_origin(sandbox_backend: SandboxBackendProtocol, repo_dir: str) -> ExecuteResponse:
    """Fetch latest from origin (best-effort)."""
    return _run_git(sandbox_backend, repo_dir, "git fetch origin 2>/dev/null || true")


def git_has_unpushed_commits(sandbox_backend: SandboxBackendProtocol, repo_dir: str) -> bool:
    """Check whether there are commits not pushed to upstream."""
    git_log_cmd = (
        "git log --oneline @{upstream}..HEAD 2>/dev/null "
        "|| git log --oneline origin/HEAD..HEAD 2>/dev/null || echo ''"
    )
    result = _run_git(sandbox_backend, repo_dir, git_log_cmd)
    return result.exit_code == 0 and bool(result.output.strip())


def git_current_branch(sandbox_backend: SandboxBackendProtocol, repo_dir: str) -> str:
    """Get the current git branch name."""
    result = _run_git(sandbox_backend, repo_dir, "git rev-parse --abbrev-ref HEAD")
    return result.output.strip() if result.exit_code == 0 else ""


def git_checkout_branch(
    sandbox_backend: SandboxBackendProtocol, repo_dir: str, branch: str
) -> bool:
    """Checkout branch, creating it if needed."""
    safe_branch = shlex.quote(branch)
    checkout_result = _run_git(sandbox_backend, repo_dir, f"git checkout -B {safe_branch}")
    if checkout_result.exit_code == 0:
        return True
    fallback_create = _run_git(sandbox_backend, repo_dir, f"git checkout -b {safe_branch}")
    if fallback_create.exit_code == 0:
        return True
    fallback = _run_git(sandbox_backend, repo_dir, f"git checkout {safe_branch}")
    return fallback.exit_code == 0


def git_checkout_existing_branch(
    sandbox_backend: SandboxBackendProtocol, repo_dir: str, branch: str
) -> ExecuteResponse:
    """Checkout an existing branch without creating or resetting it."""
    safe_branch = shlex.quote(branch)
    return _run_git(sandbox_backend, repo_dir, f"git checkout {safe_branch}")


def git_config_user(
    sandbox_backend: SandboxBackendProtocol,
    repo_dir: str,
    name: str,
    email: str,
) -> None:
    """Configure git user name and email."""
    safe_name = shlex.quote(name)
    safe_email = shlex.quote(email)
    _run_git(sandbox_backend, repo_dir, f"git config user.name {safe_name}")
    _run_git(sandbox_backend, repo_dir, f"git config user.email {safe_email}")


def git_add_all(sandbox_backend: SandboxBackendProtocol, repo_dir: str) -> ExecuteResponse:
    """Stage all changes."""
    return _run_git(sandbox_backend, repo_dir, "git add -A")


def git_commit(
    sandbox_backend: SandboxBackendProtocol, repo_dir: str, message: str
) -> ExecuteResponse:
    """Commit staged changes with the given message."""
    safe_message = shlex.quote(message)
    return _run_git(sandbox_backend, repo_dir, f"git commit -m {safe_message}")


def git_get_remote_url(sandbox_backend: SandboxBackendProtocol, repo_dir: str) -> str | None:
    """Get the origin remote URL."""
    result = _run_git(sandbox_backend, repo_dir, "git remote get-url origin")
    if result.exit_code != 0:
        return None
    return result.output.strip()


def git_push(
    sandbox_backend: SandboxBackendProtocol,
    repo_dir: str,
    branch: str,
) -> ExecuteResponse:
    """Push the branch to origin.

    Authentication is handled by the sandbox proxy (configured at sandbox creation
    time via the LangSmith proxy-config API), so no token is needed here.
    """
    safe_branch = shlex.quote(branch)
    return _run_git(sandbox_backend, repo_dir, f"git push origin {safe_branch}")


async def create_github_pr(
    repo_owner: str,
    repo_name: str,
    github_token: str,
    title: str,
    head_branch: str,
    base_branch: str,
    body: str,
    installation_token: str | None = None,
) -> tuple[str | None, int | None, bool]:
    """Create a draft GitHub pull request via the API.

    When *github_token* differs from *installation_token* (e.g. a user
    OAuth token), the function first attempts to create the PR with the
    user token so the user becomes the PR author.  If that fails it
    retries with the installation token.  The ``OpenSWE`` label is
    always added using the installation token.

    Args:
        repo_owner: Repository owner (e.g., "langchain-ai")
        repo_name: Repository name (e.g., "deepagents")
        github_token: GitHub access token (user token preferred)
        title: PR title
        head_branch: Source branch name
        base_branch: Target branch name
        body: PR description
        installation_token: GitHub App installation token used for labeling and as a fallback
            for PR creation. Falls back to github_token when not provided.

    Returns:
        Tuple of (pr_url, pr_number, pr_existing) if successful, (None, None, False) otherwise
    """
    tokens_to_try = [github_token]
    if installation_token and installation_token != github_token:
        tokens_to_try.append(installation_token)
    label_tok = installation_token or github_token

    pr_payload = {
        "title": title,
        "head": head_branch,
        "base": base_branch,
        "body": body,
        "draft": True,
    }

    logger.info(
        "Creating PR: head=%s, base=%s, repo=%s/%s",
        head_branch,
        base_branch,
        repo_owner,
        repo_name,
    )

    async with httpx.AsyncClient() as http_client:
        for token in tokens_to_try:
            try:
                pr_response = await http_client.post(
                    f"https://api.github.com/repos/{repo_owner}/{repo_name}/pulls",
                    headers={
                        "Authorization": f"Bearer {token}",
                        "Accept": "application/vnd.github+json",
                        "X-GitHub-Api-Version": "2022-11-28",
                    },
                    json=pr_payload,
                )

                pr_data = pr_response.json()

                if pr_response.status_code == HTTP_CREATED:
                    pr_url = pr_data.get("html_url")
                    pr_number = pr_data.get("number")
                    await _add_label(
                        http_client,
                        repo_owner,
                        repo_name,
                        label_tok,
                        pr_number,
                    )
                    logger.info("PR created successfully: %s", pr_url)
                    return pr_url, pr_number, False

                if pr_response.status_code == HTTP_UNPROCESSABLE_ENTITY:
                    logger.error("GitHub API validation error (422): %s", pr_data.get("message"))
                    existing = await _find_existing_pr(
                        http_client=http_client,
                        repo_owner=repo_owner,
                        repo_name=repo_name,
                        github_token=token,
                        head_branch=head_branch,
                    )
                    if existing:
                        await _add_label(
                            http_client,
                            repo_owner,
                            repo_name,
                            label_tok,
                            existing[1],
                        )
                        logger.info("Using existing PR for head branch: %s", existing[0])
                        return existing[0], existing[1], True
                    else:
                        logger.debug(
                            "Could not find existing PR with current token, will retry"
                            if token != tokens_to_try[-1]
                            else "Could not find existing PR"
                        )
                else:
                    logger.error(
                        "GitHub API error (%s): %s",
                        pr_response.status_code,
                        pr_data.get("message"),
                    )

                if "errors" in pr_data:
                    logger.error("GitHub API errors detail: %s", pr_data.get("errors"))

                # If this was the user token, fall through to retry with installation token
                if token != tokens_to_try[-1]:
                    logger.info("Retrying PR creation with installation token")
                    continue

                return None, None, False

            except httpx.HTTPError:
                logger.exception("Failed to create PR via GitHub API")
                if token != tokens_to_try[-1]:
                    logger.info("Retrying PR creation with installation token")
                    continue
                return None, None, False

    return None, None, False


_OPENSWE_LABEL = "OpenSWE"


async def _add_label(
    http_client: httpx.AsyncClient,
    repo_owner: str,
    repo_name: str,
    github_token: str,
    pr_number: int | None,
) -> None:
    """Add the 'OpenSWE' label to a PR without failing PR creation on errors."""
    if not pr_number:
        return

    try:
        response = await http_client.post(
            f"https://api.github.com/repos/{repo_owner}/{repo_name}/issues/{pr_number}/labels",
            headers={
                "Authorization": f"Bearer {github_token}",
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
            },
            json={"labels": [_OPENSWE_LABEL]},
        )
        if response.is_success:
            logger.info("Added '%s' label to PR #%s", _OPENSWE_LABEL, pr_number)
        else:
            logger.warning(
                "Failed to add label to PR #%s (%s)",
                pr_number,
                response.status_code,
            )
    except httpx.HTTPError:
        logger.warning("Failed to add label to PR #%s", pr_number, exc_info=True)


async def _find_existing_pr(
    http_client: httpx.AsyncClient,
    repo_owner: str,
    repo_name: str,
    github_token: str,
    head_branch: str,
) -> tuple[str | None, int | None]:
    """Find an existing PR for the given head branch."""
    headers = {
        "Authorization": f"Bearer {github_token}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    head_ref = f"{repo_owner}:{head_branch}"
    for state in ("open", "all"):
        response = await http_client.get(
            f"https://api.github.com/repos/{repo_owner}/{repo_name}/pulls",
            headers=headers,
            params={"head": head_ref, "state": state, "per_page": 1},
        )
        if response.status_code != 200:  # noqa: PLR2004
            continue
        data = response.json()
        if not data:
            continue
        pr = data[0]
        return pr.get("html_url"), pr.get("number")
    return None, None


async def get_github_default_branch(
    repo_owner: str,
    repo_name: str,
    github_token: str,
) -> str:
    """Get the default branch of a GitHub repository via the API.

    Args:
        repo_owner: Repository owner (e.g., "langchain-ai")
        repo_name: Repository name (e.g., "deepagents")
        github_token: GitHub access token

    Returns:
        The default branch name (e.g., "main" or "master")
    """
    try:
        async with httpx.AsyncClient() as http_client:
            response = await http_client.get(
                f"https://api.github.com/repos/{repo_owner}/{repo_name}",
                headers={
                    "Authorization": f"Bearer {github_token}",
                    "Accept": "application/vnd.github+json",
                    "X-GitHub-Api-Version": "2022-11-28",
                },
            )

            if response.status_code == 200:  # noqa: PLR2004
                repo_data = response.json()
                default_branch = repo_data.get("default_branch", "main")
                logger.debug("Got default branch from GitHub API: %s", default_branch)
                return default_branch

            logger.warning(
                "Failed to get repo info from GitHub API (%s), falling back to 'main'",
                response.status_code,
            )
            return "main"

    except httpx.HTTPError:
        logger.exception("Failed to get default branch from GitHub API, falling back to 'main'")
        return "main"