feat(seer): Run-anchored notify_seer_pr_created RPC for PR attribution (CW-1719)

src/sentry/seer/endpoints/seer_rpc.py

@@ -121,6 +121,7 @@
 from sentry.seer.fetch_issues.utils import NoProjectsForRepoError, get_repo_and_projects
 from sentry.seer.issue_detection import create_issue_occurrence
 from sentry.seer.models.seer_api_models import SeerProjectPreference
+from sentry.seer.pull_requests import notify_seer_pr_created
 from sentry.seer.seer_setup import get_supported_scm_providers
 from sentry.seer.sentry_data_models import (
     AttributeBucket,
@@ -1020,6 +1021,9 @@ def refresh_monitoring_provider_token(
     # PR metrics (judge path)
     "update_pr_metrics": seer_rpc(update_pr_metrics),
     #
+    # PR created (attribution + run link)
+    "notify_seer_pr_created": seer_rpc(notify_seer_pr_created),
+    #
     # Monitoring provider tokens (MCP)
     "get_monitoring_provider_connections": seer_rpc(get_monitoring_provider_connections),
     "refresh_monitoring_provider_token": seer_rpc(refresh_monitoring_provider_token),

src/sentry/seer/pull_requests.py

@@ -6,11 +6,17 @@
 from collections.abc import Mapping, Sequence
 from typing import Any
 
-from sentry import options
-from sentry.models.organization import Organization
+from sentry import features, options
+from sentry.models.organization import Organization, OrganizationStatus
 from sentry.models.pullrequest import PullRequest
+from sentry.pr_metrics.attribution import attribute_seer_created_pull_requests
 from sentry.seer.endpoints.utils import get_seer_run
 from sentry.seer.models.run import SeerRun, SeerRunCodingAgentHandoff, SeerRunPullRequest
+from sentry.seer.sentry_data_models import (
+    NotifySeerPrCreatedErrorResponse,
+    NotifySeerPrCreatedSuccessResponse,
+)
+from sentry.utils import metrics
 
 logger = logging.getLogger(__name__)
 
@@ -119,3 +125,85 @@ def link_pull_request_to_seer_run(
         )
 
     return resolved.pull_request
+
+
+def record_seer_created_pull_requests(
+    *,
+    organization: Organization,
+    run_id: int | None,
+    pull_requests: Sequence[Mapping[str, Any]],
+    group_id: int | None = None,
+) -> None:
+    """Record attribution + a run link for the PRs Seer directly created.
+
+    Attribution is gated on ``organizations:pr-metrics-attribution``; linking is
+    gated only on its own killswitch (checked inside ``link_seer_run_pull_requests``)
+    and always attempted. Both sides are best-effort: any failure is logged and
+    swallowed so the caller's flow is never interrupted.
+    """
+    log_context = {
+        "organization_id": organization.id,
+        "run_id": run_id,
+        "group_id": group_id,
+    }
+
+    if features.has("organizations:pr-metrics-attribution", organization):
+        try:
+            attribute_seer_created_pull_requests(
+                organization=organization,
+                pull_requests=pull_requests,
+                run_id=run_id,
+                group_id=group_id,
+            )
+        except Exception:
+            logger.exception("seer.pr_attribution.failed", extra=log_context)
+
+    try:
+        link_seer_run_pull_requests(
+            organization=organization,
+            seer_run_state_id=run_id,
+            pull_requests=pull_requests,
+        )
+    except Exception:
+        logger.exception("seer.pr_link.failed", extra=log_context)
+
+
+def notify_seer_pr_created(
+    *,
+    organization_id: int,
+    run_id: int,
+    pull_requests: list[dict[str, Any]],
+    group_id: int | None = None,
+) -> NotifySeerPrCreatedSuccessResponse | NotifySeerPrCreatedErrorResponse:
+    """Record attribution + run link for a PR Seer just created (Seer -> Sentry RPC).
+
+    Run-anchored and issue-optional: Seer's PR writer calls this for every PR it
+    opens, whether or not the run is tied to an issue (``group_id`` may be None).
+    This is deliberately independent of the autofix completion hook -- it does no
+    sentry-app broadcast, Activity creation, or analytics, and is not gated on
+    ``SeerAutofixOperator.has_access``. Attribution and linking keep their own
+    existing gates (the ``organizations:pr-metrics-attribution`` flag and the
+    ``seer.pull-request-linking.killswitch.enabled`` killswitch respectively),
+    inherited via ``record_seer_created_pull_requests``.
+    """
+    try:
+        organization = Organization.objects.get(
+            id=organization_id, status=OrganizationStatus.ACTIVE
+        )
+    except Organization.DoesNotExist:
+        logger.exception(
+            "seer.pr_created_notify.organization_not_found_or_not_active",
+            extra={"organization_id": organization_id},
+        )
+        return NotifySeerPrCreatedErrorResponse(error="Organization not found or not active")
+
+    metrics.incr("seer.pr_created_notify", tags={"has_group": group_id is not None})
+
+    record_seer_created_pull_requests(
+        organization=organization,
+        run_id=run_id,
+        pull_requests=pull_requests,
+        group_id=group_id,
+    )
+
+    return NotifySeerPrCreatedSuccessResponse()

src/sentry/seer/sentry_data_models.py

@@ -155,6 +155,21 @@ class SendSeerWebhookErrorResponse(BaseModel):
     error: str
 
 
+class NotifySeerPrCreatedSuccessResponse(BaseModel):
+    """`notify_seer_pr_created` success: `{"success": true}`. The `success` literal
+    is the discriminator against the error shape below."""
+
+    success: Literal[True] = True
+
+
+class NotifySeerPrCreatedErrorResponse(BaseModel):
+    """`notify_seer_pr_created` error: `{"success": false, "error": <message>}`.
+    Only returned when the organization can't be resolved."""
+
+    success: Literal[False] = False
+    error: str
+
+
 class HasRepoCodeMappingsResponse(BaseModel):
     has_code_mappings: bool
     project_slug_to_id: dict[str, int]

tests/sentry/seer/test_pull_requests.py

@@ -1,9 +1,18 @@
 from typing import Any
 from unittest.mock import Mock, patch
 
-from sentry.models.pullrequest import PullRequest
+from sentry.models.pullrequest import (
+    PullRequest,
+    PullRequestAttribution,
+    PullRequestAttributionSignalType,
+    PullRequestAttributionSource,
+)
 from sentry.seer.models.run import SeerRun, SeerRunPullRequest, SeerRunType
-from sentry.seer.pull_requests import link_seer_run_pull_requests
+from sentry.seer.pull_requests import link_seer_run_pull_requests, notify_seer_pr_created
+from sentry.seer.sentry_data_models import (
+    NotifySeerPrCreatedErrorResponse,
+    NotifySeerPrCreatedSuccessResponse,
+)
 from sentry.testutils.cases import TestCase
 
 REPO_NAME = "getsentry/sentry"
@@ -119,3 +128,114 @@ def test_run_lookup_is_org_scoped(self) -> None:
         self._link(self._payload())
 
         assert not SeerRunPullRequest.objects.exists()
+
+
+PR_NUMBER = 7412
+PR_URL = "https://github.com/getsentry/sentry/pull/7412"
+
+
+class NotifySeerPrCreatedTest(TestCase):
+    def setUp(self) -> None:
+        self.repo = self.create_repo(self.project, name=REPO_NAME, provider="integrations:github")
+        self.seer_run = self.create_seer_run(
+            self.organization, type=SeerRunType.FEATURE_RUN, seer_run_state_id=RUN_STATE_ID
+        )
+
+    def _payload(
+        self,
+        pr_number: int = PR_NUMBER,
+        pr_url: str = PR_URL,
+        provider: str = "github",
+    ) -> list[dict[str, Any]]:
+        return [
+            {
+                "provider": provider,
+                "repo_name": REPO_NAME,
+                "pull_request": {"pr_id": 999, "pr_number": pr_number, "pr_url": pr_url},
+            }
+        ]
+
+    def _notify(
+        self,
+        pull_requests: list[dict[str, Any]],
+        *,
+        run_id: int = RUN_STATE_ID,
+        group_id: int | None = None,
+    ) -> NotifySeerPrCreatedSuccessResponse | NotifySeerPrCreatedErrorResponse:
+        return notify_seer_pr_created(
+            organization_id=self.organization.id,
+            run_id=run_id,
+            pull_requests=pull_requests,
+            group_id=group_id,
+        )
+
+    def _linked_pull_request(self) -> PullRequest:
+        return PullRequest.objects.get(repository_id=self.repo.id, key=str(PR_NUMBER))
+
+    def test_groupless_run_links_and_attributes(self) -> None:
+        """A run with no issue (group_id=None) still links the PR and records a
+        SENTRY_APP/SEER_DATA attribution when the attribution flag is on."""
+        with self.feature("organizations:pr-metrics-attribution"):
+            result = self._notify(self._payload(), group_id=None)
+
+        assert isinstance(result, NotifySeerPrCreatedSuccessResponse)
+
+        pull_request = self._linked_pull_request()
+        link = SeerRunPullRequest.objects.get(pull_request=pull_request)
+        assert link.seer_run_id == self.seer_run.id
+
+        attribution = PullRequestAttribution.objects.get(pull_request=pull_request)
+        assert attribution.signal_type == PullRequestAttributionSignalType.SENTRY_APP
+        assert attribution.source == PullRequestAttributionSource.SEER_DATA
+        assert attribution.signal_details == {
+            "run_id": RUN_STATE_ID,
+            "group_ids": [],
+            "pr_url": PR_URL,
+        }
+
+    def test_group_present_populates_attribution_group_ids(self) -> None:
+        with self.feature("organizations:pr-metrics-attribution"):
+            self._notify(self._payload(), group_id=self.group.id)
+
+        attribution = PullRequestAttribution.objects.get(pull_request=self._linked_pull_request())
+        assert attribution.signal_details["group_ids"] == [self.group.id]
+
+    def test_idempotent_on_redelivery(self) -> None:
+        with self.feature("organizations:pr-metrics-attribution"):
+            self._notify(self._payload(), group_id=None)
+            self._notify(self._payload(), group_id=None)
+
+        pull_request = self._linked_pull_request()
+        assert SeerRunPullRequest.objects.filter(pull_request=pull_request).count() == 1
+        assert PullRequestAttribution.objects.filter(pull_request=pull_request).count() == 1
+
+    @patch("sentry.seer.entrypoints.operator.SeerAutofixOperator.has_access", return_value=False)
+    def test_links_regardless_of_operator_access(self, mock_has_access: Mock) -> None:
+        """The new path must not consult ``SeerAutofixOperator.has_access``: linking
+        happens even when operator access would be denied."""
+        with self.feature("organizations:pr-metrics-attribution"):
+            result = self._notify(self._payload(), group_id=None)
+
+        assert isinstance(result, NotifySeerPrCreatedSuccessResponse)
+        assert SeerRunPullRequest.objects.filter(pull_request=self._linked_pull_request()).exists()
+        mock_has_access.assert_not_called()
+
+    def test_links_without_attribution_when_flag_disabled(self) -> None:
+        """Attribution flag off: still links, but writes no attribution row."""
+        result = self._notify(self._payload(), group_id=None)
+
+        assert isinstance(result, NotifySeerPrCreatedSuccessResponse)
+        pull_request = self._linked_pull_request()
+        assert SeerRunPullRequest.objects.filter(pull_request=pull_request).exists()
+        assert not PullRequestAttribution.objects.filter(pull_request=pull_request).exists()
+
+    def test_returns_error_when_organization_missing(self) -> None:
+        result = notify_seer_pr_created(
+            organization_id=self.organization.id + 10_000,
+            run_id=RUN_STATE_ID,
+            pull_requests=self._payload(),
+        )
+
+        assert isinstance(result, NotifySeerPrCreatedErrorResponse)
+        assert result.error == "Organization not found or not active"
+        assert not SeerRunPullRequest.objects.exists()