galaxis-agent/agent/tools/commit_and_open_pr.py

import asyncio
import logging
import os
from typing import Any

from langgraph.config import get_config

from agent.utils.gitea_client import get_gitea_client

from ..utils.git_utils import (
    git_add_all,
    git_checkout_branch,
    git_commit,
    git_config_user,
    git_current_branch,
    git_fetch_origin,
    git_has_uncommitted_changes,
    git_has_unpushed_commits,
    git_push,
)
from ..utils.sandbox_paths import resolve_repo_dir
from ..utils.sandbox_state import get_sandbox_backend_sync

logger = logging.getLogger(__name__)


def commit_and_open_pr(
    title: str,
    body: str,
    commit_message: str | None = None,
) -> dict[str, Any]:
    """Commit all current changes and open a Gitea Pull Request.

    Args:
        title: PR title (under 70 characters)
        body: PR description with ## Description and ## Test Plan
        commit_message: Optional git commit message. If not provided, the PR title is used.

    Returns:
        Dictionary with success, error, pr_url, and pr_existing keys.
    """
    try:
        config = get_config()
        configurable = config.get("configurable", {})
        thread_id = configurable.get("thread_id")

        if not thread_id:
            return {"success": False, "error": "Missing thread_id in config", "pr_url": None}

        repo_config = configurable.get("repo", {})
        repo_owner = repo_config.get("owner")
        repo_name = repo_config.get("name")
        if not repo_owner or not repo_name:
            return {
                "success": False,
                "error": "Missing repo owner/name in config",
                "pr_url": None,
            }

        sandbox_backend = get_sandbox_backend_sync(thread_id)
        if not sandbox_backend:
            return {"success": False, "error": "No sandbox found for thread", "pr_url": None}

        repo_dir = resolve_repo_dir(sandbox_backend, repo_name)

        has_uncommitted_changes = git_has_uncommitted_changes(sandbox_backend, repo_dir)
        git_fetch_origin(sandbox_backend, repo_dir)
        has_unpushed_commits = git_has_unpushed_commits(sandbox_backend, repo_dir)

        if not (has_uncommitted_changes or has_unpushed_commits):
            return {"success": False, "error": "No changes detected", "pr_url": None}

        current_branch = git_current_branch(sandbox_backend, repo_dir)
        target_branch = f"galaxis-agent/{thread_id}"
        if current_branch != target_branch:
            if not git_checkout_branch(sandbox_backend, repo_dir, target_branch):
                return {
                    "success": False,
                    "error": f"Failed to checkout branch {target_branch}",
                    "pr_url": None,
                }

        git_config_user(
            sandbox_backend,
            repo_dir,
            "galaxis-agent[bot]",
            "galaxis-agent@users.noreply.gitea.local",
        )
        git_add_all(sandbox_backend, repo_dir)

        commit_msg = commit_message or title
        if has_uncommitted_changes:
            commit_result = git_commit(sandbox_backend, repo_dir, commit_msg)
            if commit_result.exit_code != 0:
                return {
                    "success": False,
                    "error": f"Git commit failed: {commit_result.output.strip()}",
                    "pr_url": None,
                }

        gitea_token = os.environ.get("GITEA_TOKEN", "")
        if not gitea_token:
            logger.error("commit_and_open_pr missing Gitea token for thread %s", thread_id)
            return {
                "success": False,
                "error": "Missing Gitea token",
                "pr_url": None,
            }

        push_result = git_push(sandbox_backend, repo_dir, target_branch, gitea_token)
        if push_result.exit_code != 0:
            return {
                "success": False,
                "error": f"Git push failed: {push_result.output.strip()}",
                "pr_url": None,
            }

        # --- PR 생성 (GiteaClient) ---
        gitea_external_url = os.environ.get("GITEA_EXTERNAL_URL", "")
        gitea_internal_url = os.environ.get("GITEA_URL", "http://gitea:3000")
        default_branch = os.environ.get("DEFAULT_BRANCH", "main")
        client = get_gitea_client()

        try:
            pr_result = asyncio.run(
                client.create_pull_request(
                    owner=repo_owner,
                    repo=repo_name,
                    title=title,
                    head=target_branch,
                    base=default_branch,
                    body=body,
                )
            )
            pr_url = pr_result.get("html_url", "")
            if gitea_external_url and pr_url:
                pr_url = pr_url.replace(gitea_internal_url, gitea_external_url)

            return {
                "success": True,
                "pr_url": pr_url,
                "pr_number": pr_result.get("number"),
            }
        except Exception as e:
            logger.exception("Failed to create PR (push succeeded)")
            return {
                "success": True,
                "pr_url": "",
                "error": f"Push succeeded but PR creation failed: {e}",
            }

    except Exception as e:
        logger.exception("commit_and_open_pr failed")
        return {"success": False, "error": f"{type(e).__name__}: {e}", "pr_url": None}