feat(issues): Expose mergeable_state on group pull requests

src/sentry/api/serializers/models/pullrequest.py

@@ -19,6 +19,18 @@
 
 PullRequestStatus = Literal["merged", "open", "closed", "draft", "unknown"]
 
+# GitHub's mergeable_state, lowercased: "clean" is ready to merge, "unstable"
+# has failing checks but is technically mergeable, "blocked" is waiting on
+# required checks/reviews, "dirty" has conflicts, "behind" needs a rebase.
+# Anything the provider adds later collapses to "unknown".
+PullRequestMergeableState = Literal[
+    "clean", "unstable", "blocked", "dirty", "behind", "draft", "has_hooks", "unknown"
+]
+
+KNOWN_MERGEABLE_STATES: frozenset[str] = frozenset(
+    ("clean", "unstable", "blocked", "dirty", "behind", "draft", "has_hooks")
+)
+
 
 def get_stored_pull_request_status(pull_request: PullRequest) -> PullRequestStatus | None:
     if pull_request.state == PullRequestLifecycleState.MERGED:
@@ -57,6 +69,9 @@ class LinkedPullRequestSeerAttributionResponse(TypedDict):
 class LinkedPullRequestResponse(PullRequestSerializerResponse):
     attribution: LinkedPullRequestAttributionResponse | None
     dateLinked: datetime
+    # Merge readiness from the provider; None for terminal PRs or when the
+    # provider fetch was unavailable.
+    mergeableState: PullRequestMergeableState | None
 
 
 def get_users_for_pull_requests(item_list, user=None):
@@ -131,9 +146,11 @@ def __init__(
         *,
         date_linked_by_pr_id: Mapping[int, datetime],
         status_by_pr_id: Mapping[int, PullRequestStatus],
+        mergeable_state_by_pr_id: Mapping[int, PullRequestMergeableState | None] | None = None,
     ) -> None:
         self.date_linked_by_pr_id = date_linked_by_pr_id
         self.status_by_pr_id = status_by_pr_id
+        self.mergeable_state_by_pr_id = mergeable_state_by_pr_id or {}
 
     def get_attrs(self, item_list, user, **kwargs):
         attrs = super().get_attrs(item_list, user)
@@ -155,4 +172,5 @@ def serialize(self, obj: PullRequest, attrs, user, **kwargs) -> LinkedPullReques
             "attribution": attrs["attribution"],
             "dateLinked": self.date_linked_by_pr_id[obj.id],
             "status": self.status_by_pr_id[obj.id],
+            "mergeableState": self.mergeable_state_by_pr_id.get(obj.id),
         }

src/sentry/issues/endpoints/group_pull_requests.py

@@ -4,6 +4,7 @@
 from collections.abc import Mapping
 from typing import TypedDict, cast
 
+from django.core.cache import cache
 from django.db.models import Exists, OuterRef
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -13,8 +14,10 @@
 from sentry.api.base import cell_silo_endpoint
 from sentry.api.serializers import serialize
 from sentry.api.serializers.models.pullrequest import (
+    KNOWN_MERGEABLE_STATES,
     LinkedPullRequestResponse,
     LinkedPullRequestSerializer,
+    PullRequestMergeableState,
     PullRequestStatus,
     get_stored_pull_request_status,
 )
@@ -30,10 +33,16 @@
 
 DEFAULT_LIMIT = 5
 
+# How long a provider PR lookup is reused before refetching. Mergeability can
+# flip whenever checks finish, so keep this short — it only needs to absorb
+# bursts of page renders, not stay fresh for long.
+PROVIDER_PR_CACHE_TTL = 120
+
 
 class ProviderPullRequestResponse(TypedDict, total=False):
     draft: bool
     merged: bool
+    mergeable_state: str
     state: str
 
 
@@ -103,18 +112,42 @@ def _fetch_pull_request_status_response(
     state = response.get("state")
     if isinstance(state, str):
         provider_response["state"] = state
+    mergeable_state = response.get("mergeable_state")
+    if isinstance(mergeable_state, str):
+        provider_response["mergeable_state"] = mergeable_state.lower()
 
     return provider_response
 
 
-def _get_provider_pull_request_status(
+def _fetch_provider_pull_request_cached(
+    pull_request: PullRequest, repository: Repository
+) -> ProviderPullRequestResponse | None:
+    """Provider PR lookup with a short-lived cache.
+
+    Open PRs are looked up on every request (mergeability changes as checks
+    finish); the cache only absorbs bursts of renders so a page of cards
+    doesn't spend provider rate limit per re-render.
+    """
+    cache_key = f"group-pull-requests:provider-pr:{repository.integration_id}:{pull_request.id}"
+    cached = cache.get(cache_key)
+    if cached is not None:
+        return cast(ProviderPullRequestResponse, cached)
+
+    response = _fetch_pull_request_status_response(pull_request, repository)
+    if response is not None:
+        cache.set(cache_key, response, PROVIDER_PR_CACHE_TTL)
+    return response
+
+
+def _get_provider_pull_request_info(
     pull_request: PullRequest, repository: Repository | None
-) -> PullRequestStatus:
+) -> tuple[PullRequestStatus | None, PullRequestMergeableState | None]:
+    """Live status + merge readiness; (None, None) when the provider is unavailable."""
     if repository is None:
-        return "unknown"
+        return None, None
 
     try:
-        response = _fetch_pull_request_status_response(pull_request, repository)
+        response = _fetch_provider_pull_request_cached(pull_request, repository)
     except Exception:
         logger.info(
             "group_pull_requests.status_fetch_failed",
@@ -125,31 +158,51 @@ def _get_provider_pull_request_status(
                 "repository_id": pull_request.repository_id,
             },
         )
-        return "unknown"
+        return None, None
 
     if response is None:
-        return "unknown"
+        return None, None
 
-    if response.get("merged"):
-        return "merged"
+    status: PullRequestStatus = "unknown"
     state = response.get("state")
-    if state == "closed":
-        return "closed"
-    if response.get("draft"):
-        return "draft"
-    if state == "open":
-        return "open"
-    return "unknown"
+    if response.get("merged"):
+        status = "merged"
+    elif state == "closed":
+        status = "closed"
+    elif response.get("draft"):
+        status = "draft"
+    elif state == "open":
+        status = "open"
+
+    mergeable_state: PullRequestMergeableState | None = None
+    raw_mergeable_state = response.get("mergeable_state")
+    # Merge readiness is only meaningful while the PR can still be merged.
+    if raw_mergeable_state and status not in ("merged", "closed"):
+        mergeable_state = cast(
+            PullRequestMergeableState,
+            raw_mergeable_state if raw_mergeable_state in KNOWN_MERGEABLE_STATES else "unknown",
+        )
+
+    return status, mergeable_state
 
 
-def _get_pull_request_status(
+def _get_pull_request_state(
     pull_request: PullRequest, repository: Repository | None
-) -> PullRequestStatus:
+) -> tuple[PullRequestStatus, PullRequestMergeableState | None]:
+    """Resolve a PR's lifecycle status and merge readiness.
+
+    Terminal PRs (merged/closed) are answered from the stored, webhook-kept
+    fields without a provider round-trip. Everything else asks the provider —
+    the stored row can't answer merge readiness — falling back to the stored
+    status when the provider is unavailable.
+    """
     stored_status = get_stored_pull_request_status(pull_request)
-    if stored_status is not None:
-        return stored_status
+    if stored_status in ("merged", "closed"):
+        return stored_status, None
 
-    return _get_provider_pull_request_status(pull_request, repository)
+    provider_status, mergeable_state = _get_provider_pull_request_info(pull_request, repository)
+    status = provider_status or stored_status or "unknown"
+    return status, mergeable_state
 
 
 @cell_silo_endpoint
@@ -188,12 +241,14 @@ def get(self, request: Request, group: Group) -> Response[GroupPullRequestsRespo
         ]
 
         date_linked_by_pr_id = {link.linked_id: link.datetime for link in group_links}
-        status_by_pr_id = {
-            pull_request.id: _get_pull_request_status(
+        status_by_pr_id: dict[int, PullRequestStatus] = {}
+        mergeable_state_by_pr_id: dict[int, PullRequestMergeableState | None] = {}
+        for pull_request in pull_requests:
+            status, mergeable_state = _get_pull_request_state(
                 pull_request, repositories_by_id.get(pull_request.repository_id)
             )
-            for pull_request in pull_requests
-        }
+            status_by_pr_id[pull_request.id] = status
+            mergeable_state_by_pr_id[pull_request.id] = mergeable_state
 
         # serialize() infers the base PullRequestSerializerResponse from the
         # parent's generic; LinkedPullRequestSerializer returns the narrower type.
@@ -205,6 +260,7 @@ def get(self, request: Request, group: Group) -> Response[GroupPullRequestsRespo
                 serializer=LinkedPullRequestSerializer(
                     date_linked_by_pr_id=date_linked_by_pr_id,
                     status_by_pr_id=status_by_pr_id,
+                    mergeable_state_by_pr_id=mergeable_state_by_pr_id,
                 ),
             ),
         )

tests/sentry/issues/endpoints/test_group_pull_requests.py

@@ -3,6 +3,7 @@
 from datetime import datetime, timedelta
 from unittest.mock import Mock, patch
 
+from django.core.cache import cache
 from django.utils import timezone
 
 from sentry.constants import ObjectStatus
@@ -24,6 +25,9 @@
 class GroupPullRequestsEndpointTest(APITestCase):
     def setUp(self) -> None:
         super().setUp()
+        # The provider PR lookup is cached by pull-request id, which repeats
+        # across test databases.
+        cache.clear()
         self.login_as(user=self.user)
         self.group = self.create_group()
         self.repo = self.create_repo(
@@ -273,29 +277,15 @@ def test_returns_display_pull_request_attribution(self) -> None:
         }
 
     @patch("sentry.issues.endpoints.group_pull_requests.integration_service.get_integration")
-    def test_status_derivation_prefers_stored_lifecycle_fields(
-        self, mock_get_integration: Mock
-    ) -> None:
+    def test_terminal_stored_status_skips_provider(self, mock_get_integration: Mock) -> None:
         self.create_linked_pull_request(
             key="1",
-            linked_delta=timedelta(days=5),
-            state=PullRequestLifecycleState.OPEN,
-            draft=False,
-        )
-        self.create_linked_pull_request(
-            key="2",
-            linked_delta=timedelta(days=4),
-            state=PullRequestLifecycleState.OPEN,
-            draft=True,
-        )
-        self.create_linked_pull_request(
-            key="3",
             linked_delta=timedelta(days=3),
             state=PullRequestLifecycleState.CLOSED,
             draft=True,
         )
         self.create_linked_pull_request(
-            key="4",
+            key="2",
             linked_delta=timedelta(days=2),
             state=PullRequestLifecycleState.MERGED,
             merged_at=timezone.now(),
@@ -307,11 +297,84 @@ def test_status_derivation_prefers_stored_lifecycle_fields(
         assert [item["status"] for item in response.data["pullRequests"]] == [
             "merged",
             "closed",
-            "draft",
-            "open",
         ]
+        assert [item["mergeableState"] for item in response.data["pullRequests"]] == [None, None]
         mock_get_integration.assert_not_called()
 
+    @patch("sentry.issues.endpoints.group_pull_requests.integration_service.get_integration")
+    def test_open_pull_requests_fetch_mergeable_state(self, mock_get_integration: Mock) -> None:
+        self.create_linked_pull_request(
+            key="1",
+            linked_delta=timedelta(days=2),
+            state=PullRequestLifecycleState.OPEN,
+            draft=False,
+        )
+        self.create_linked_pull_request(
+            key="2",
+            linked_delta=timedelta(days=1),
+            state=PullRequestLifecycleState.OPEN,
+            draft=True,
+        )
+
+        client = self.set_provider_pull_request_response(mock_get_integration, {})
+        client.get_pull_request.side_effect = lambda _repo, key: {
+            # Uppercase to prove normalization.
+            "1": {"state": "open", "draft": False, "mergeable_state": "CLEAN"},
+            "2": {"state": "open", "draft": True, "mergeable_state": "dirty"},
+        }[key]
+
+        response = self.client.get(self.path)
+
+        assert response.status_code == 200
+        by_id = {item["id"]: item for item in response.data["pullRequests"]}
+        assert by_id["1"]["status"] == "open"
+        assert by_id["1"]["mergeableState"] == "clean"
+        assert by_id["2"]["status"] == "draft"
+        assert by_id["2"]["mergeableState"] == "dirty"
+
+    @patch("sentry.issues.endpoints.group_pull_requests.integration_service.get_integration")
+    def test_unrecognized_mergeable_state_collapses_to_unknown(
+        self, mock_get_integration: Mock
+    ) -> None:
+        self.create_linked_pull_request(key="1", state=PullRequestLifecycleState.OPEN, draft=False)
+        self.set_provider_pull_request_response(
+            mock_get_integration,
+            {"state": "open", "draft": False, "mergeable_state": "wacky_new_state"},
+        )
+
+        response = self.client.get(self.path)
+
+        assert response.status_code == 200
+        assert response.data["pullRequests"][0]["mergeableState"] == "unknown"
+
+    @patch("sentry.issues.endpoints.group_pull_requests.integration_service.get_integration")
+    def test_provider_failure_falls_back_to_stored_status(self, mock_get_integration: Mock) -> None:
+        self.create_linked_pull_request(key="1", state=PullRequestLifecycleState.OPEN, draft=False)
+        client = self.set_provider_pull_request_response(mock_get_integration, {})
+        client.get_pull_request.side_effect = RuntimeError("nope")
+
+        response = self.client.get(self.path)
+
+        assert response.status_code == 200
+        assert response.data["pullRequests"][0]["status"] == "open"
+        assert response.data["pullRequests"][0]["mergeableState"] is None
+
+    @patch("sentry.issues.endpoints.group_pull_requests.integration_service.get_integration")
+    def test_provider_lookup_is_cached_between_requests(self, mock_get_integration: Mock) -> None:
+        self.create_linked_pull_request(key="1", state=PullRequestLifecycleState.OPEN, draft=False)
+        client = self.set_provider_pull_request_response(
+            mock_get_integration,
+            {"state": "open", "draft": False, "mergeable_state": "blocked"},
+        )
+
+        first = self.client.get(self.path)
+        second = self.client.get(self.path)
+
+        assert first.status_code == 200
+        assert second.status_code == 200
+        assert second.data["pullRequests"][0]["mergeableState"] == "blocked"
+        assert client.get_pull_request.call_count == 1
+
     @patch("sentry.issues.endpoints.group_pull_requests.integration_service.get_integration")
     def test_incomplete_stored_status_falls_back_to_provider(
         self, mock_get_integration: Mock