open-swe/agent/utils/github_checks.py

"""GitHub Checks API helpers for the reviewer's PR check run.

A check run named ``Open SWE Review`` is created on the PR head SHA when an
auto-review is dispatched, so the PR's checks section shows the review as
in-progress. ``publish_review`` (or the after-agent fallback) completes it.

All calls are best-effort: check runs require the GitHub App's
``Checks: Read & write`` permission, and a missing permission must never
break review dispatch or publish.
"""

from __future__ import annotations

import logging
from collections.abc import Mapping
from datetime import UTC, datetime
from typing import Literal

import httpx

from .github_http import (
    GITHUB_API_BASE,
    github_client,
    github_headers,  # re-exported for backward compatibility
    github_request,
)

__all__ = ["github_headers"]

logger = logging.getLogger(__name__)

REVIEW_CHECK_RUN_NAME = "Open SWE Review"
AUTOFIX_CHECK_RUN_NAME = "Open SWE Auto-fix"

_GITHUB_API_BASE = GITHUB_API_BASE

CheckConclusion = Literal["success", "neutral", "failure"]


def _utc_now_iso() -> str:
    return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


async def create_review_check_run(
    *,
    owner: str,
    repo: str,
    head_sha: str,
    token: str,
    details_url: str | None = None,
) -> int | None:
    """Create an in-progress ``Open SWE Review`` check run on ``head_sha``.

    Returns the check run id, or ``None`` on any failure (most commonly the
    App lacking the Checks permission).
    """
    payload: dict[str, object] = {
        "name": REVIEW_CHECK_RUN_NAME,
        "head_sha": head_sha,
        "status": "in_progress",
        "started_at": _utc_now_iso(),
        "output": {
            "title": "Review in progress",
            "summary": "Open SWE is reviewing this pull request…",
        },
    }
    if details_url:
        payload["details_url"] = details_url
    url = f"{_GITHUB_API_BASE}/repos/{owner}/{repo}/check-runs"
    try:
        async with github_client(token=token) as client:
            response = await github_request(client, "POST", url, json=payload)
            response.raise_for_status()
    except httpx.HTTPError:
        logger.exception(
            "Failed to create review check run for %s/%s@%s "
            "(does the GitHub App have Checks: Read & write?)",
            owner,
            repo,
            head_sha,
        )
        return None
    data = response.json()
    check_run_id = data.get("id") if isinstance(data, dict) else None
    return check_run_id if isinstance(check_run_id, int) else None


async def complete_review_check_run(
    *,
    owner: str,
    repo: str,
    check_run_id: int,
    token: str,
    conclusion: CheckConclusion,
    title: str,
    summary: str,
) -> bool:
    """Mark a review check run as completed. Returns True on success."""
    url = f"{_GITHUB_API_BASE}/repos/{owner}/{repo}/check-runs/{check_run_id}"
    payload = {
        "status": "completed",
        "conclusion": conclusion,
        "completed_at": _utc_now_iso(),
        "output": {"title": title, "summary": summary},
    }
    try:
        async with github_client(token=token) as client:
            response = await github_request(client, "PATCH", url, json=payload)
            response.raise_for_status()
    except httpx.HTTPError:
        logger.exception(
            "Failed to complete review check run %s on %s/%s", check_run_id, owner, repo
        )
        return False
    return True


async def post_autofix_status_check(
    *,
    owner: str,
    repo: str,
    head_sha: str,
    token: str,
    title: str,
    summary: str,
    details_url: str | None = None,
) -> bool:
    """Post an informational, completed ``Open SWE Auto-fix`` check on ``head_sha``.

    Completed immediately as ``neutral`` so it's non-blocking and never leaves a
    dangling in-progress check that could gate branch protection. Used as the
    auto-fix status channel instead of a PR comment (PR comments can trigger
    ``issue_comment`` automation like Atlantis/Terraform).
    """
    payload: dict[str, object] = {
        "name": AUTOFIX_CHECK_RUN_NAME,
        "head_sha": head_sha,
        "status": "completed",
        "conclusion": "neutral",
        "started_at": _utc_now_iso(),
        "completed_at": _utc_now_iso(),
        "output": {"title": title, "summary": summary},
    }
    if details_url:
        payload["details_url"] = details_url
    url = f"{_GITHUB_API_BASE}/repos/{owner}/{repo}/check-runs"
    try:
        async with github_client(token=token) as client:
            response = await github_request(client, "POST", url, json=payload)
            response.raise_for_status()
    except httpx.HTTPError:
        logger.warning("Failed to post auto-fix status check for %s/%s@%s", owner, repo, head_sha)
        return False
    return True


def review_check_conclusion(
    publish_outcome: Mapping[str, object],
) -> tuple[CheckConclusion, str, str]:
    """Map the authoritative publish outcome to a check conclusion."""
    verdict_submitted = publish_outcome.get("verdict_submitted") is True
    verdict_event = publish_outcome.get("verdict_event")
    if verdict_submitted and verdict_event == "APPROVE":
        return (
            "success",
            "Review approved",
            "Open SWE recorded an approving review on this pull request.",
        )
    if verdict_submitted and verdict_event == "REQUEST_CHANGES":
        return (
            "failure",
            "Changes requested",
            "Open SWE recorded a request-changes review on this pull request.",
        )

    blocking_count_raw = publish_outcome.get("blocking_finding_count")
    blocking_count = (
        blocking_count_raw
        if isinstance(blocking_count_raw, int) and not isinstance(blocking_count_raw, bool)
        else None
    )
    ignored_reason = publish_outcome.get("verdict_ignored_reason")
    verdict_authorization = publish_outcome.get("verdict_authorization")
    if (
        blocking_count is not None
        and blocking_count > 0
        and verdict_authorization
        in {
            "requested",
            "consistent",
        }
        and ignored_reason
        in {
            None,
            "approve_with_open_findings",
            "self_review",
        }
    ):
        issue_word = "issue" if blocking_count == 1 else "issues"
        return (
            "failure",
            f"Found {blocking_count} blocking {issue_word}",
            f"Open SWE found {blocking_count} blocking {issue_word} on this pull request.",
        )
    if ignored_reason and ignored_reason != "self_review":
        return (
            "neutral",
            "Verdict withheld",
            f"Open SWE published without a verdict ({str(ignored_reason).replace('_', ' ')}).",
        )
    if blocking_count == 0:
        return (
            "success",
            "No issues found",
            "Open SWE reviewed this pull request and found no blocking issues.",
        )

    surfaced_count_raw = publish_outcome.get("surfaced_count")
    surfaced_count = (
        surfaced_count_raw
        if isinstance(surfaced_count_raw, int) and not isinstance(surfaced_count_raw, bool)
        else 0
    )
    if surfaced_count > 0:
        issue_word = "issue" if surfaced_count == 1 else "issues"
        title = f"Found {surfaced_count} potential {issue_word}"
    else:
        title = "Review completed without verdict"
    return (
        "neutral",
        title,
        "Open SWE completed the review without an authoritative verdict.",
    )