galaxis-agent/agent/tools/commit_and_open_pr.py

import asyncio
import logging
from typing import Any

from langgraph.config import get_config

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,
                }

        import os
        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,
            }

        # TODO: Phase 2 - use GiteaClient to create PR
        return {"success": True, "pr_url": "pending-gitea-implementation"}

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