feat(code_review): Validate before scheduling (#108545)

<!-- Describe your PR here. -->
Adds pre-scheduling validation for `event_payload` within
`schedule_task` to prevent invalid tasks from being queued.

This addresses issues where model changes could lead to tasks being
scheduled with incompatible payloads when task brokers are updated
before Sentry workers, causing failures downstream. Validation now
occurs before the Celery task is dispatched, catching schema mismatches
early and improving system robustness.

---

Linear Issue:
[CW-837](https://linear.app/getsentry/issue/CW-837/validate-the-event-payload-before-scheduling-as-a-task)

<p><a
href="https://cursor.com/background-agent?bcId=bc-888aa40a-d917-4614-bde4-31cbca092587"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://cursor.com/assets/images/open-in-cursor-dark.png"><source
media="(prefers-color-scheme: light)"
srcset="https://cursor.com/assets/images/open-in-cursor-light.png"><img
alt="Open in Cursor" width="131" height="28"
src="https://cursor.com/assets/images/open-in-cursor-dark.png"></picture></a>&nbsp;<a
href="https://cursor.com/agents?id=bc-888aa40a-d917-4614-bde4-31cbca092587"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://cursor.com/assets/images/open-in-web-dark.png"><source
media="(prefers-color-scheme: light)"
srcset="https://cursor.com/assets/images/open-in-web-light.png"><img
alt="Open in Web" width="114" height="28"
src="https://cursor.com/assets/images/open-in-web-dark.png"></picture></a></p>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Armen Zambrano G. <armenzg@users.noreply.github.com>
Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: 3 people

src/sentry/seer/code_review/webhooks/task.py

@@ -67,9 +67,34 @@ def schedule_task(
         )
         return
 
+    # Validate payload before scheduling to catch schema mismatches early
+    from pydantic import ValidationError
+
+    try:
+        request_type = transformed_event.get("request_type")
+        validated_payload: (
+            SeerCodeReviewTaskRequestForPrClosed | SeerCodeReviewTaskRequestForPrReview
+        )
+        if request_type == "pr-closed":
+            validated_payload = SeerCodeReviewTaskRequestForPrClosed.parse_obj(transformed_event)
+        else:
+            validated_payload = SeerCodeReviewTaskRequestForPrReview.parse_obj(transformed_event)
+        # Convert to dict and handle enum keys (Pydantic v1 converts string keys to enums,
+        # but JSON requires string keys, so we need to convert them back)
+        payload = convert_enum_keys_to_strings(validated_payload.dict())
+        # When upgrading to Pydantic v2, we can remove the convert_enum_keys_to_strings call.
+        # Pydantic v2 will automatically convert enum keys to strings.
+        # payload = validated_payload.model_dump(mode="json")
+    except ValidationError:
+        logger.warning("%s.validation_failed_before_scheduling", PREFIX)
+        record_webhook_filtered(
+            github_event, github_event_action, WebhookFilteredReason.INVALID_PAYLOAD
+        )
+        return
+
     process_github_webhook_event.delay(
         github_event=github_event.value,
-        event_payload=transformed_event,
+        event_payload=payload,
         enqueued_at_str=datetime.now(timezone.utc).isoformat(),
         tags=tags,
     )
@@ -96,7 +121,7 @@ def process_github_webhook_event(
     Args:
         enqueued_at_str: The timestamp when the task was enqueued
         github_event: The GitHub webhook event type from X-GitHub-Event header (e.g., "check_run", "pull_request")
-        event_payload: The payload of the webhook event
+        event_payload: The payload of the webhook event (already validated before scheduling)
         tags: Sentry SDK tags to set on this task's scope for error correlation
         **kwargs: Parameters to pass to webhook handler functions
     """
@@ -106,29 +131,7 @@ def process_github_webhook_event(
         if tags:
             sentry_sdk.set_tags(tags)
         path = get_seer_endpoint_for_event(github_event).value
-
-        # Validate payload with Pydantic (except for CHECK_RUN events which use minimal payload)
-        if github_event != GithubWebhookType.CHECK_RUN:
-            # Parse with appropriate model based on request type to enforce
-            # organization_id and integration_id requirements for PR closed
-            request_type = event_payload.get("request_type")
-            validated_payload: (
-                SeerCodeReviewTaskRequestForPrClosed | SeerCodeReviewTaskRequestForPrReview
-            )
-            if request_type == "pr-closed":
-                validated_payload = SeerCodeReviewTaskRequestForPrClosed.parse_obj(event_payload)
-            else:
-                validated_payload = SeerCodeReviewTaskRequestForPrReview.parse_obj(event_payload)
-            # Convert to dict and handle enum keys (Pydantic v1 converts string keys to enums,
-            # but JSON requires string keys, so we need to convert them back)
-            payload = convert_enum_keys_to_strings(validated_payload.dict())
-            # When upgrading to Pydantic v2, we can remove the convert_enum_keys_to_strings call.
-            # Pydantic v2 will automatically convert enum keys to strings.
-            # payload = validated_payload.model_dump(mode="json")
-        else:
-            payload = event_payload
-
-        make_seer_request(path=path, payload=payload)
+        make_seer_request(path=path, payload=event_payload)
     except Exception as e:
         status = e.__class__.__name__
         # Retryable errors are automatically retried by taskworker.

tests/sentry/seer/code_review/test_webhooks.py

@@ -581,102 +581,6 @@ def test_pr_review_validation_passes_without_organization_id(
 
         assert mock_request.call_count == 1
 
-    @patch("sentry.seer.code_review.utils.make_signed_seer_api_request")
-    def test_pr_closed_validation_fails_without_organization_id(
-        self, mock_request: MagicMock
-    ) -> None:
-        """Test that PR closed validation fails without organization_id (it's required)."""
-        mock_request.return_value = self._mock_response(200, b"{}")
-
-        event_payload = {
-            "request_type": "pr-closed",
-            "external_owner_id": "456",
-            "data": {
-                "repo": {
-                    "provider": "github",
-                    "owner": "test-owner",
-                    "name": "test-repo",
-                    "external_id": "123456",
-                    "base_commit_sha": "abc123",
-                    "integration_id": "99999",
-                    # organization_id intentionally omitted
-                },
-                "pr_id": 123,
-                "bug_prediction_specific_information": {
-                    "organization_id": 789,
-                    "organization_slug": "test-org",
-                },
-                "config": {
-                    "features": {"bug_prediction": True},
-                    "trigger": "on_new_commit",
-                    "trigger_at": "2024-01-15T10:30:00Z",
-                    "sentry_received_trigger_at": "2024-01-15T10:30:00Z",
-                },
-            },
-        }
-
-        # Should raise validation error
-        from pydantic import ValidationError
-
-        with pytest.raises(ValidationError) as exc_info:
-            process_github_webhook_event._func(
-                github_event=GithubWebhookType.PULL_REQUEST,
-                event_payload=event_payload,
-                enqueued_at_str=self.enqueued_at_str,
-            )
-
-        # Verify the error is about organization_id
-        errors = exc_info.value.errors()
-        assert any("organization_id" in str(error) for error in errors)
-
-    @patch("sentry.seer.code_review.utils.make_signed_seer_api_request")
-    def test_pr_closed_validation_fails_without_integration_id(
-        self, mock_request: MagicMock
-    ) -> None:
-        """Test that PR closed validation fails without integration_id (it's required)."""
-        mock_request.return_value = self._mock_response(200, b"{}")
-
-        event_payload = {
-            "request_type": "pr-closed",
-            "external_owner_id": "456",
-            "data": {
-                "repo": {
-                    "provider": "github",
-                    "owner": "test-owner",
-                    "name": "test-repo",
-                    "external_id": "123456",
-                    "base_commit_sha": "abc123",
-                    "organization_id": 789,
-                    # integration_id intentionally omitted
-                },
-                "pr_id": 123,
-                "bug_prediction_specific_information": {
-                    "organization_id": 789,
-                    "organization_slug": "test-org",
-                },
-                "config": {
-                    "features": {"bug_prediction": True},
-                    "trigger": "on_new_commit",
-                    "trigger_at": "2024-01-15T10:30:00Z",
-                    "sentry_received_trigger_at": "2024-01-15T10:30:00Z",
-                },
-            },
-        }
-
-        # Should raise validation error
-        from pydantic import ValidationError
-
-        with pytest.raises(ValidationError) as exc_info:
-            process_github_webhook_event._func(
-                github_event=GithubWebhookType.PULL_REQUEST,
-                event_payload=event_payload,
-                enqueued_at_str=self.enqueued_at_str,
-            )
-
-        # Verify the error is about integration_id
-        errors = exc_info.value.errors()
-        assert any("integration_id" in str(error) for error in errors)
-
     @patch("sentry.seer.code_review.utils.make_signed_seer_api_request")
     def test_pr_closed_validation_passes_with_required_fields(
         self, mock_request: MagicMock

tests/sentry/seer/code_review/webhooks/test_pull_request.py

@@ -381,3 +381,61 @@ def test_pull_request_closed_draft_still_sends_to_seer(self) -> None:
             call_kwargs = self.mock_seer.call_args[1]
             payload = call_kwargs["payload"]
             assert payload["request_type"] == SeerCodeReviewRequestType.PR_CLOSED.value
+
+    def test_validation_happens_before_task_scheduling_pr_closed(self) -> None:
+        """Test that invalid pr-closed payloads are caught before scheduling the Celery task."""
+        with (
+            self.code_review_setup(),
+            self.tasks(),
+            patch(
+                "sentry.seer.code_review.webhooks.task.transform_webhook_to_codegen_request"
+            ) as mock_transform,
+        ):
+            # Return an invalid payload missing required fields for pr-closed
+            mock_transform.return_value = {
+                "request_type": "pr-closed",
+                "data": {
+                    # Missing required fields like repo, pr_id, etc.
+                    "invalid": "payload"
+                },
+            }
+
+            event = orjson.loads(PULL_REQUEST_OPENED_EVENT_EXAMPLE)
+            event["action"] = "closed"
+
+            self._send_webhook_event(
+                GithubWebhookType.PULL_REQUEST,
+                orjson.dumps(event),
+            )
+
+            # Task should NOT be scheduled due to validation failure
+            self.mock_seer.assert_not_called()
+
+    def test_validation_happens_before_task_scheduling_pr_review(self) -> None:
+        """Test that invalid pr-review payloads are caught before scheduling the Celery task."""
+        with (
+            self.code_review_setup(),
+            self.tasks(),
+            patch(
+                "sentry.seer.code_review.webhooks.task.transform_webhook_to_codegen_request"
+            ) as mock_transform,
+        ):
+            # Return an invalid payload missing required fields for pr-review
+            mock_transform.return_value = {
+                "request_type": "pr-review",
+                "data": {
+                    # Missing required fields
+                    "invalid": "payload"
+                },
+            }
+
+            event = orjson.loads(PULL_REQUEST_OPENED_EVENT_EXAMPLE)
+            event["action"] = "opened"
+
+            self._send_webhook_event(
+                GithubWebhookType.PULL_REQUEST,
+                orjson.dumps(event),
+            )
+
+            # Task should NOT be scheduled due to validation failure
+            self.mock_seer.assert_not_called()