dev/release/gh.py

"""GitHub CLI helper functions for the release tool."""

import abc
import enum
import json
import os
import re
import subprocess
import tempfile
from typing import (
    TypedDict,
    override,  # pyrefly: ignore[missing-module-attribute] -- override available in Python 3.12+
)

from dev.release.release_issue import BackportTask
from dev.release.shell import run_cmd

# GitHub label types
RELEASE_LABEL = "type: release"
BACKPORT_LABEL = "type: backport-pr"
RELEASE_PREPARED_LABEL = "release-prepared"
SYNC_CHANGELOG_LABEL = "type: sync-changelog"

# GitHub reaction types
# See: https://docs.github.com/en/rest/reactions/reactions?apiVersion=2022-11-28#about-reactions
GH_REACTION_THUMBS_UP = "+1"
GH_REACTION_THUMBS_DOWN = "-1"
GH_REACTION_LAUGH = "laugh"
GH_REACTION_CONFUSED = "confused"
GH_REACTION_HEART = "heart"
GH_REACTION_HOORAY = "hooray"
GH_REACTION_ROCKET = "rocket"
GH_REACTION_EYES = "eyes"


class BackportTaskStatus(str, enum.Enum):
    """Status strings for backport tasks on a release tracking issue."""

    PENDING = "pending"
    DONE = "done"
    RESOLVED = "resolved"
    OPEN_PR = "open-pr"
    DRAFT_PR = "draft-pr"
    ERROR_NOT_FOUND = "error-not-found"
    ERROR_CLOSED_PR = "error-closed-pr"
    ERROR_NO_MERGE_COMMIT = "error-no-merge-commit"
    ERROR_UNKNOWN = "error-unknown"
    ERROR_RESOLUTION_FAILED = "error-resolution-failed"
    ERROR_MERGE_CONFLICT = "error-merge-conflict"
    ERROR_INVALID_PR = "error-invalid-pr"
    IGNORE = "ignore"

    def __str__(self) -> str:
        return self.value


class IssueDict(TypedDict, total=False):
    """In-memory representation of a GitHub Issue object.

    See GitHub API docs:
    https://docs.github.com/en/rest/issues/issues#get-an-issue
    """

    number: int
    title: str
    body: str
    labels: list[str]
    url: str


class AutoMergeDict(TypedDict, total=False):
    """Representation of auto-merge status on a Pull Request.

    See GitHub API docs:
    https://docs.github.com/en/rest/pulls/pulls#get-a-pull-request
    """

    merge_method: str


class PrFileDict(TypedDict, total=False):
    """In-memory representation of a file in a GitHub Pull Request object.

    See GitHub API docs:
    https://docs.github.com/en/rest/pulls/pulls#list-pull-requests-files
    """

    path: str
    additions: int
    deletions: int
    changeType: str


class PrDict(TypedDict, total=False):
    """In-memory representation of a GitHub Pull Request object.

    See GitHub API docs:
    https://docs.github.com/en/rest/pulls/pulls#get-a-pull-request
    """

    number: int
    title: str
    body: str
    base: str
    head: str
    labels: list[str]
    url: str
    state: str
    isDraft: bool
    mergeCommit: dict[str, str]
    auto_merge: AutoMergeDict
    files: list[PrFileDict]


class GitHubEventPullRequestDict(TypedDict, total=False):
    """Pull request object in a GitHub Actions event payload.

    See GitHub Webhook events docs:
    https://docs.github.com/en/webhooks/webhook-events-and-payloads#pull_request
    """

    number: int


class GitHubEventIssueDict(TypedDict, total=False):
    """Issue object in a GitHub Actions event payload.

    See GitHub Webhook events docs:
    https://docs.github.com/en/webhooks/webhook-events-and-payloads#issues
    """

    number: int


class GitHubEventDict(TypedDict, total=False):
    """Representation of a GitHub Actions event payload JSON ($GITHUB_EVENT_PATH).

    See GitHub Actions events docs:
    https://docs.github.com/en/actions/using-workflows/events-that-trigger-workflows
    """

    inputs: dict[str, object]
    pull_request: GitHubEventPullRequestDict
    issue: GitHubEventIssueDict
    number: int | None


def get_github_event_data() -> GitHubEventDict:
    """Loads JSON data from GITHUB_EVENT_PATH if set."""
    event_path = os.environ.get("GITHUB_EVENT_PATH")
    if not event_path or not os.path.isfile(event_path):
        return {}
    with open(event_path, "r", encoding="utf-8") as f:
        return json.load(f)


def get_github_event_pr_number() -> int | None:
    """Extracts PR number from GITHUB_EVENT_PATH if available."""
    data = get_github_event_data()
    if not data:
        return None
    if (
        "inputs" in data
        and isinstance(data["inputs"], dict)
        and data["inputs"].get("pr")
    ):
        pr_val = str(data["inputs"]["pr"]).lstrip("#")
        if pr_val.isdigit():
            return int(pr_val)
    if (
        "pull_request" in data
        and isinstance(data["pull_request"], dict)
        and data["pull_request"].get("number")
    ):
        return int(data["pull_request"]["number"])
    if "number" in data and isinstance(data["number"], int):
        return data["number"]
    return None


def get_github_event_issue_number() -> int | None:
    """Extracts Issue number from GITHUB_EVENT_PATH if available."""
    data = get_github_event_data()
    if not data:
        return None
    if (
        "inputs" in data
        and isinstance(data["inputs"], dict)
        and data["inputs"].get("issue")
    ):
        issue_val = str(data["inputs"]["issue"]).lstrip("#")
        if issue_val.isdigit():
            return int(issue_val)
    if (
        "issue" in data
        and isinstance(data["issue"], dict)
        and data["issue"].get("number")
    ):
        return int(data["issue"]["number"])
    return None


class MultipleTrackingIssuesError(ValueError):
    """Raised when multiple open tracking issues are found for a version."""

    pass


class NoTrackingIssueError(ValueError):
    """Raised when no open tracking issue is found for a version."""

    pass


class CreatePrError(Exception):
    """Raised when creating a pull request fails."""

    pass


class GetPrError(ValueError):
    """Raised when querying a pull request fails."""

    pass


class InvalidPrRefError(ValueError):
    """Raised when a PR reference cannot be resolved."""

    pass


class GitHubInterface(abc.ABC):
    """Abstract interface for GitHub operations."""

    repo: str

    @abc.abstractmethod
    def post_issue_comment(self, issue_num: int, comment_body: str) -> None:
        """Posts a comment on an issue or PR.

        Args:
            issue_num: The issue or PR number.
            comment_body: The body content of the comment.
        """

    @abc.abstractmethod
    def add_comment_reaction(self, comment_id: int, reaction: str) -> None:
        """Adds a reaction to an issue or PR comment.

        Args:
            comment_id: The comment ID.
            reaction: The reaction type (e.g., "+1", "-1", "rocket").
        """

    @abc.abstractmethod
    def enable_auto_merge(self, pr_num: int, method: str = "squash") -> None:
        """Enables auto-merge for a PR.

        Args:
            pr_num: The PR number.
            method: The merge method ('squash', 'rebase', or 'merge').
        """

    @abc.abstractmethod
    def create_issue(
        self, title: str, body: str, labels: list[str] | None = None
    ) -> int:
        """Creates an issue.

        Args:
            title: Title of the issue.
            body: Body text of the issue.
            labels: Optional list of labels to add.

        Returns:
            The created issue number.
        """

    @abc.abstractmethod
    def create_release_tracking_issue(self, version: str, template_content: str) -> int:
        """Creates a release tracking issue from a template.

        Args:
            version: Release version string (e.g., "1.0.0").
            template_content: Content of the issue template markdown file.

        Returns:
            The created issue number.
        """

    @abc.abstractmethod
    def get_issue_body(self, issue_num: int) -> str:
        """Gets the body content of an issue.

        Args:
            issue_num: The issue number.

        Returns:
            The body string of the issue.
        """

    @abc.abstractmethod
    def get_issue_title(self, issue_num: int) -> str:
        """Gets the title of an issue.

        Args:
            issue_num: The issue number.

        Returns:
            The title string of the issue.
        """

    @abc.abstractmethod
    def update_issue_body(self, issue_num: int, body: str) -> None:
        """Updates the body of an issue.

        Args:
            issue_num: The issue number.
            body: The new body content.
        """

    @abc.abstractmethod
    def resolve_pr_number(self, pr_ref: str) -> int:
        """Resolves a PR reference (number, #number, or GitHub URL) to a PR number.

        Args:
            pr_ref: PR number string (e.g., "123", "#123") or URL.

        Returns:
            The integer PR number.

        Raises:
            InvalidPrRefError: If the PR reference cannot be resolved or is for
                another repository.
        """

    @abc.abstractmethod
    def get_release_tracking_issue(self, version: str) -> int:
        """Finds the single open tracking issue for a given version.

        Args:
            version: Version string (e.g., "1.0.0").

        Returns:
            The issue number.

        Raises:
            NoTrackingIssueError: If no open tracking issue is found.
            MultipleTrackingIssuesError: If multiple open tracking issues are
                found.
        """

    @abc.abstractmethod
    def create_pr(
        self,
        title: str,
        body: str,
        base: str = "main",
        labels: list[str] | None = None,
    ) -> str:
        """Creates a pull request.

        Args:
            title: Title of the PR.
            body: Body of the PR.
            base: Base branch to merge into (default: "main").
            labels: Optional list of labels to add.

        Returns:
            The URL of the created PR.
        """

    @abc.abstractmethod
    def get_open_pr(self, branch_name: str) -> PrDict | None:
        """Finds an open PR for the given branch.

        Args:
            branch_name: The head branch name to search for.

        Returns:
            Dictionary containing PR details if open, else None.
        """

    @abc.abstractmethod
    def get_open_tracking_issues(self, version: str | None = None) -> list[IssueDict]:
        """Finds open tracking issues for release.

        Args:
            version: Optional specific version to match (e.g., "1.0.0").

        Returns:
            List of matching open release tracking issue dictionaries.
        """

    @abc.abstractmethod
    def get_pr_info(self, pr_num: int) -> PrDict:
        """Gets info about a PR.

        Args:
            pr_num: The PR number.

        Returns:
            Dictionary containing PR fields (state, isDraft, mergeCommit, etc.).

        Raises:
            GetPrError: If querying the PR fails.
        """

    @abc.abstractmethod
    def get_pr_files(self, pr_num: int) -> list[str]:
        """Gets the list of file paths touched by a PR.

        Args:
            pr_num: The PR number.

        Returns:
            A list of file paths.

        Raises:
            GetPrError: If querying the PR fails.
        """

    @abc.abstractmethod
    def get_pr_comments(self, pr_num: int) -> list[dict]:
        """Gets all comments for a PR.

        Args:
            pr_num: The PR number.

        Returns:
            List of comment objects.

        Raises:
            GetPrError: If querying the PR fails.
        """

    @abc.abstractmethod
    def get_merge_commits_for_prs(
        self, pending_items: list[BackportTask]
    ) -> list[BackportTask]:
        """Resolves PR references in pending backports to their merge commit SHAs.

        Args:
            pending_items: A list of BackportTask items to resolve.

        Returns:
            The list of resolved BackportTask items.
        """


class GitHub(GitHubInterface):
    """GitHub CLI helper class for the release tool."""

    def __init__(self, repo: str = "bazel-contrib/rules_python"):
        """Initializes the GitHub helper.

        Args:
            repo: The GitHub repository to operate on.
        """
        self.repo = repo

    def _run_gh(
        self, *args: str, check: bool = True, capture_output: bool = True
    ) -> str | None:
        """Runs a 'gh' command.

        Args:
            *args: Arguments for 'gh' (excluding 'gh').
            check: If True, raises CalledProcessError on failure.
            capture_output: If True, captures and returns stdout.

        Returns:
            The stdout of the command, stripped, or None.
        """
        return run_cmd("gh", *args, check=check, capture_output=capture_output)

    def _gh_issue(
        self, *args: str, check: bool = True, capture_output: bool = True
    ) -> str | None:
        """Runs a 'gh issue' command."""
        return self._run_gh(
            "issue",
            *args,
            f"--repo={self.repo}",
            check=check,
            capture_output=capture_output,
        )

    def _gh_pr(
        self, *args: str, check: bool = True, capture_output: bool = True
    ) -> str | None:
        """Runs a 'gh pr' command."""
        return self._run_gh(
            "pr",
            *args,
            f"--repo={self.repo}",
            check=check,
            capture_output=capture_output,
        )

    def list_issues(
        self,
        *,
        fields: str,
        label: str | None = None,
        state: str | None = None,
        search: str | None = None,
    ) -> list[IssueDict]:
        """Helper to list issues using gh CLI.

        Args:
            fields: Comma-separated list of fields to return.
            label: Filter by label.
            state: Filter by state ('open', 'closed', 'all').
            search: Search query.

        Returns:
            A list of issue dictionaries.
        """
        cmd = ["list", f"--json={fields}"]
        if label:
            cmd.append(f"--label={label}")
        if state:
            cmd.append(f"--state={state}")
        if search:
            cmd.append(f"--search={search}")

        output = self._gh_issue(*cmd)
        return json.loads(output) if output else []

    @override
    def get_open_tracking_issues(self, version: str | None = None) -> list[IssueDict]:
        """Finds open tracking issues for release.

        Args:
            version: Optional specific version to match (e.g., "1.0.0").

        Returns:
            List of matching open release tracking issue dictionaries.
        """
        search = f"Release {version}" if version else None
        return self.list_issues(
            fields="number,title,url",
            label=RELEASE_LABEL,
            state="open",
            search=search,
        )

    @override
    def get_release_tracking_issue(self, version: str) -> int:
        """Finds the single open tracking issue for a given version.

        Args:
            version: Version string (e.g. "1.0.0").

        Returns:
            The issue number.

        Raises:
            NoTrackingIssueError: If no open tracking issue is found.
            MultipleTrackingIssuesError: If multiple open tracking issues are found.
        """
        issues = self.get_open_tracking_issues(version)
        matching = [i for i in issues if i["title"] == f"Release {version}"]
        if not matching:
            raise NoTrackingIssueError(
                f"No open tracking issue found for Release {version}"
            )
        if len(matching) > 1:
            raise MultipleTrackingIssuesError(
                f"Multiple open tracking issues found for Release {version}: "
                + ", ".join(str(i["number"]) for i in matching)
            )
        return matching[0]["number"]

    @override
    def create_issue(
        self, title: str, body: str, labels: list[str] | None = None
    ) -> int:
        """Creates an issue using gh CLI.

        Args:
            title: Title of the issue.
            body: Body of the issue.
            labels: List of labels to add.

        Returns:
            The issue number.
        """
        cmd = ["create", f"--title={title}", f"--body={body}"]
        if labels:
            for label in labels:
                cmd.append(f"--label={label}")

        output = self._gh_issue(*cmd)
        if not output:
            raise RuntimeError("gh issue create returned no output")
        # output is URL: https://github.com/owner/repo/issues/123
        return int(output.rstrip("/").split("/")[-1])

    @override
    def create_release_tracking_issue(self, version: str, template_content: str) -> int:
        """Creates a release tracking issue from a template.

        Args:
            version: Release version string (e.g., "1.0.0").
            template_content: Content of the issue template markdown file.

        Returns:
            The created issue number.
        """
        title = f"Release {version}"
        # Strip YAML frontmatter if present
        issue_body = template_content
        if template_content.startswith("---"):
            parts = template_content.split("---", 2)
            if len(parts) >= 3:
                issue_body = parts[2].strip()

        return self.create_issue(title=title, body=issue_body, labels=[RELEASE_LABEL])

    @override
    def get_issue_body(self, issue_num: int) -> str:
        """Gets the body content of an issue.

        Args:
            issue_num: The issue number.

        Returns:
            The body string of the issue.
        """
        output = self._gh_issue("view", str(issue_num), "--json=body")
        if not output:
            return ""
        data = json.loads(output)
        return data.get("body", "")

    @override
    def get_issue_title(self, issue_num: int) -> str:
        """Gets the title of an issue.

        Args:
            issue_num: The issue number.

        Returns:
            The title string of the issue.
        """
        output = self._gh_issue("view", str(issue_num), "--json=title")
        if not output:
            return ""
        data = json.loads(output)
        return data.get("title", "")

    @override
    def update_issue_body(self, issue_num: int, body: str) -> None:
        """Updates the body of an issue.

        Args:
            issue_num: The issue number.
            body: The new body content.
        """
        with tempfile.NamedTemporaryFile(mode="w", delete=False, encoding="utf-8") as f:
            f.write(body)
            f.flush()
            temp_path = f.name

        try:
            self._gh_issue(
                "edit", str(issue_num), f"--body-file={temp_path}", capture_output=False
            )
        finally:
            if os.path.exists(temp_path):
                os.remove(temp_path)

    @override
    def resolve_pr_number(self, pr_ref: str) -> int:
        """Resolves a PR reference (number, #number, or GitHub URL) to a PR number.

        Args:
            pr_ref: PR number string (e.g., "123", "#123") or URL.

        Returns:
            The integer PR number.

        Raises:
            InvalidPrRefError: If the PR reference cannot be resolved or is for another repo.
        """
        clean_ref = pr_ref.lstrip("#")
        if clean_ref.isdigit():
            return int(clean_ref)

        if pr_ref.startswith("http"):
            pattern = rf"github\.com/{re.escape(self.repo)}/pull/(\d+)(/|\?|\Z)"
            match = re.search(pattern, pr_ref, re.IGNORECASE)
            if match:
                return int(match.group(1))
            raise InvalidPrRefError(
                f"URL is not for the configured repository ({self.repo}): {pr_ref}"
            )

        raise InvalidPrRefError(f"Could not resolve PR reference: {pr_ref}")

    def _gh_pr_view(self, pr_num: int, *fields: str) -> str:
        """Helper to run `gh pr view` with specified JSON fields.

        Args:
            pr_num: The PR number.
            *fields: JSON fields to request (e.g., "state", "files").

        Returns:
            The raw JSON output string from gh.

        Raises:
            GetPrError: If querying the PR fails.
        """
        args = ["view", str(pr_num)]
        if fields:
            args.append(f"--json={','.join(fields)}")
        try:
            output = self._gh_pr(*args)
            return output or ""
        except subprocess.CalledProcessError as e:
            raise GetPrError(f"Failed to get PR #{pr_num} on {self.repo}: {e}") from e

    @override
    def get_pr_info(self, pr_num: int) -> PrDict:
        """Gets info about a PR using gh CLI.

        Args:
            pr_num: The PR number.

        Returns:
            Dictionary containing PR fields (state, isDraft, mergeCommit, etc.).
        """
        output = self._gh_pr_view(pr_num, "state", "isDraft", "mergeCommit")
        return json.loads(output) if output else {}

    @override
    def get_pr_files(self, pr_num: int) -> list[str]:
        """Gets the list of file paths touched by a PR using gh CLI.

        Args:
            pr_num: The PR number.

        Returns:
            A list of file paths.

        Raises:
            GetPrError: If querying the PR fails.
        """
        output = self._gh_pr_view(pr_num, "files")
        if not output:
            return []
        data: PrDict = json.loads(output)
        files = data.get("files", [])
        return [f["path"] for f in files]

    @override
    def get_pr_comments(self, pr_num: int) -> list[dict]:
        """Gets all comments for a PR using gh CLI.

        Args:
            pr_num: The PR number.

        Returns:
            List of comment objects (with body, author, etc.).
        """
        output = self._gh_pr_view(pr_num, "comments")
        if not output:
            return []
        data = json.loads(output)
        return data.get("comments", [])

    @override
    def create_pr(
        self,
        title: str,
        body: str,
        base: str = "main",
        labels: list[str] | None = None,
    ) -> str:
        """Creates a pull request.

        Args:
            title: Title of the PR.
            body: Body of the PR.
            base: Base branch to merge into (default: "main").
            labels: Optional list of labels to add.

        Returns:
            The URL of the created PR.

        Raises:
            CreatePrError: If creating the pull request fails.
        """
        cmd = [
            "create",
            f"--title={title}",
            f"--body={body}",
            f"--base={base}",
        ]
        if labels:
            for label in labels:
                cmd.append(f"--label={label}")
        try:
            output = self._gh_pr(*cmd)
        except subprocess.CalledProcessError as e:
            msg = f"Failed to create PR '{title}': {e}"
            if e.stdout:
                msg += (
                    f"\n{'=' * 20} STDOUT BEGIN {'=' * 20}\n"
                    f"{e.stdout}\n"
                    f"{'=' * 20} STDOUT END {'=' * 20}"
                )
            if e.stderr:
                msg += (
                    f"\n{'=' * 20} STDERR BEGIN {'=' * 20}\n"
                    f"{e.stderr}\n"
                    f"{'=' * 20} STDERR END {'=' * 20}"
                )
            raise CreatePrError(msg) from e
        except Exception as e:
            raise CreatePrError(f"Failed to create PR '{title}': {e}") from e
        if not output:
            raise CreatePrError(
                f"Failed to create PR '{title}': gh pr create returned no output"
            )
        return output

    @override
    def enable_auto_merge(self, pr_num: int, method: str = "squash") -> None:
        """Enables auto-merge for a PR.

        Args:
            pr_num: The PR number.
            method: The merge method ('squash', 'rebase', or 'merge').
        """
        cmd = ["merge", str(pr_num), "--auto"]
        if method == "squash":
            cmd.append("--squash")
        elif method == "rebase":
            cmd.append("--rebase")
        elif method == "merge":
            cmd.append("--merge")
        self._gh_pr(*cmd, capture_output=False)

    @override
    def get_open_pr(self, branch_name: str) -> PrDict | None:
        """Finds an open PR for the given branch.

        Args:
            branch_name: The head branch name to search for.

        Returns:
            Dictionary with 'number' and 'url' if an open PR exists, else None.
        """
        cmd = [
            "list",
            f"--head={branch_name}",
            "--state=open",
            "--json=number,url",
        ]
        output = self._gh_pr(*cmd)
        prs = json.loads(output) if output else []
        return prs[0] if prs else None

    @override
    def post_issue_comment(self, issue_num: int, comment_body: str) -> None:
        """Posts a comment on an issue or PR.

        Args:
            issue_num: The issue or PR number.
            comment_body: The body content of the comment.
        """
        self._gh_issue(
            "comment",
            str(issue_num),
            f"--body={comment_body}",
            capture_output=False,
        )

    @override
    def add_comment_reaction(self, comment_id: int, reaction: str) -> None:
        """Adds a reaction to an issue or PR comment.

        Args:
            comment_id: The comment ID (note: gh api endpoint needed for comment reactions).
            reaction: The reaction type (e.g., "+1", "-1", "rocket").
        """
        self._run_gh(
            "api",
            f"repos/{self.repo}/issues/comments/{comment_id}/reactions",
            "-f",
            f"content={reaction}",
            capture_output=False,
        )

    @override
    def get_merge_commits_for_prs(
        self, pending_items: list[BackportTask]
    ) -> list[BackportTask]:
        """Resolves PR references in pending backports to their merge commit SHAs.

        Updates item.status based on PR state if it cannot be resolved.

        Args:
            pending_items: A list of BackportTask items to resolve.

        Returns:
            The list of resolved BackportTask items.
        """
        return resolve_merge_commits_for_prs(self, pending_items)


def resolve_merge_commits_for_prs(
    gh_client: GitHubInterface, pending_items: list[BackportTask]
) -> list[BackportTask]:
    """Resolves PR references in pending backports to their merge commit SHAs.

    Updates item.status based on PR state if it cannot be resolved.

    Args:
        gh_client: The GitHub client.
        pending_items: A list of BackportTask items to resolve.

    Returns:
        The list of resolved BackportTask items.
    """
    resolved_items = []
    for item in pending_items:
        pr_num = int(item.pr_ref.lstrip("#"))
        print(f"Resolving PR #{pr_num} to merge commit...")
        try:
            pr_info = gh_client.get_pr_info(pr_num)
            if not pr_info:
                print(f"PR #{pr_num} not found. Gating.")
                item.status = BackportTaskStatus.ERROR_NOT_FOUND
            else:
                state = pr_info.get("state")
                is_draft = pr_info.get("isDraft", False)
                if state == "OPEN" or is_draft:
                    print(
                        f"PR #{pr_num} is open or draft (state: {state},"
                        f" draft: {is_draft}). Ignoring."
                    )
                    item.status = (
                        BackportTaskStatus.OPEN_PR
                        if not is_draft
                        else BackportTaskStatus.DRAFT_PR
                    )
                elif state == "CLOSED":
                    print(f"PR #{pr_num} is closed but not merged. Gating.")
                    item.status = BackportTaskStatus.ERROR_CLOSED_PR
                elif state == "MERGED":
                    merge_commit = pr_info.get("mergeCommit")
                    if merge_commit and "oid" in merge_commit:
                        item.commit = merge_commit["oid"]
                        item.status = BackportTaskStatus.RESOLVED
                    else:
                        print(f"PR #{pr_num} has no merge commit SHA. Gating.")
                        item.status = BackportTaskStatus.ERROR_NO_MERGE_COMMIT
                else:
                    print(f"PR #{pr_num} has unknown state: {state}. Gating.")
                    item.status = BackportTaskStatus.ERROR_UNKNOWN
        except Exception as e:
            print(f"Error resolving PR #{pr_num}: {e}. Gating.")
            item.status = BackportTaskStatus.ERROR_RESOLUTION_FAILED
        resolved_items.append(item)
    return resolved_items