feat(seer): Add lightweight supergroups backfill task

Add an org-scoped Celery task that iterates all error groups in an
organization (seen in last 90 days) and sends each to Seer's
lightweight RCA clustering endpoint for supergroup backfilling.

The task processes groups in batches of 50 with cursor-based
pagination and self-chains until all groups are processed. Designed
to be triggered from a getsentry job.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: yuvmen

src/sentry/options/defaults.py

@@ -1385,6 +1385,13 @@
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+register(
+    "seer.supergroups_backfill_lightweight.killswitch",
+    type=Bool,
+    default=False,
+    flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 # ## sentry.killswitches
 #
 # The following options are documented in sentry.killswitches in more detail

src/sentry/tasks/seer/backfill_supergroups_lightweight.py

@@ -0,0 +1,186 @@
+import logging
+from datetime import UTC, datetime, timedelta
+
+from sentry import features, options
+from sentry.api.serializers import EventSerializer, serialize
+from sentry.eventstore import backend as eventstore
+from sentry.models.group import DEFAULT_TYPE_ID, Group
+from sentry.models.organization import Organization
+from sentry.models.project import Project
+from sentry.seer.signed_seer_api import (
+    LightweightRCAClusterRequest,
+    SeerViewerContext,
+    make_lightweight_rca_cluster_request,
+)
+from sentry.tasks.base import instrumented_task
+from sentry.taskworker.namespaces import seer_tasks
+from sentry.types.group import GroupSubStatus
+from sentry.utils import metrics
+
+logger = logging.getLogger(__name__)
+
+BACKFILL_LAST_SEEN_DAYS = 90
+BATCH_SIZE = 50
+INTER_BATCH_DELAY_S = 5
+
+
+@instrumented_task(
+    name="sentry.tasks.seer.backfill_supergroups_lightweight.backfill_supergroups_lightweight_for_org",
+    namespace=seer_tasks,
+    processing_deadline_duration=15 * 60,
+)
+def backfill_supergroups_lightweight_for_org(
+    organization_id: int,
+    last_project_id: int = 0,
+    last_group_id: int = 0,
+    **kwargs,
+) -> None:
+    if options.get("seer.supergroups_backfill_lightweight.killswitch"):
+        logger.info("supergroups_backfill_lightweight.killswitch_enabled")
+        return
+
+    try:
+        organization = Organization.objects.get(id=organization_id)
+    except Organization.DoesNotExist:
+        return
+
+    if not features.has("organizations:supergroups-lightweight-rca-clustering", organization):
+        logger.info(
+            "supergroups_backfill_lightweight.feature_not_enabled",
+            extra={"organization_id": organization_id},
+        )
+        return
+
+    project_ids = list(
+        Project.objects.filter(
+            organization_id=organization_id,
+            id__gte=last_project_id,
+        )
+        .order_by("id")
+        .values_list("id", flat=True)
+    )
+
+    if not project_ids:
+        logger.info(
+            "supergroups_backfill_lightweight.org_completed",
+            extra={"organization_id": organization_id},
+        )
+        return
+
+    cutoff = datetime.now(UTC) - timedelta(days=BACKFILL_LAST_SEEN_DAYS)
+
+    group_filter = Group.objects.filter(
+        project_id__in=project_ids,
+        type=DEFAULT_TYPE_ID,
+        last_seen__gte=cutoff,
+        substatus__in=[
+            GroupSubStatus.ONGOING,
+            GroupSubStatus.NEW,
+            GroupSubStatus.ESCALATING,
+            GroupSubStatus.REGRESSED,
+        ],
+    )
+
+    if last_group_id > 0:
+        group_filter = group_filter.filter(
+            project_id=last_project_id, id__gt=last_group_id
+        ) | group_filter.filter(project_id__gt=last_project_id)
+    else:
+        group_filter = group_filter.filter(project_id__gte=last_project_id)
+
+    groups = list(
+        group_filter.select_related("project", "project__organization").order_by(
+            "project_id", "id"
+        )[:BATCH_SIZE]
+    )
+
+    if not groups:
+        logger.info(
+            "supergroups_backfill_lightweight.org_completed",
+            extra={"organization_id": organization_id},
+        )
+        return
+
+    # Phase 1: Batch fetch event data
+    group_event_pairs: list[tuple[Group, dict]] = []
+    for group in groups:
+        event = group.get_latest_event()
+        if not event:
+            continue
+
+        ready_event = eventstore.get_event_by_id(
+            group.project_id, event.event_id, group_id=group.id
+        )
+        if not ready_event:
+            continue
+
+        serialized_event = serialize(ready_event, None, EventSerializer())
+        group_event_pairs.append((group, serialized_event))
+
+    # Phase 2: Send to Seer (per-group for now, bulk-ready)
+    failure_count = 0
+    success_count = 0
+    viewer_context = SeerViewerContext(organization_id=organization_id)
+
+    for group, serialized_event in group_event_pairs:
+        try:
+            body = LightweightRCAClusterRequest(
+                group_id=group.id,
+                issue={
+                    "id": group.id,
+                    "title": group.title,
+                    "short_id": group.qualified_short_id,
+                    "events": [serialized_event],
+                },
+                organization_slug=organization.slug,
+                organization_id=organization_id,
+                project_id=group.project_id,
+            )
+            response = make_lightweight_rca_cluster_request(
+                body, timeout=30, viewer_context=viewer_context
+            )
+            if response.status >= 400:
+                logger.warning(
+                    "supergroups_backfill_lightweight.seer_error",
+                    extra={
+                        "group_id": group.id,
+                        "project_id": group.project_id,
+                        "status": response.status,
+                    },
+                )
+                failure_count += 1
+            else:
+                success_count += 1
+        except Exception:
+            logger.exception(
+                "supergroups_backfill_lightweight.group_failed",
+                extra={"group_id": group.id, "project_id": group.project_id},
+            )
+            failure_count += 1
+
+    metrics.incr(
+        "seer.supergroups_backfill_lightweight.groups_processed",
+        amount=success_count,
+    )
+    metrics.incr(
+        "seer.supergroups_backfill_lightweight.groups_failed",
+        amount=failure_count,
+    )
+
+    # Self-chain if there are more groups to process
+    if len(groups) == BATCH_SIZE:
+        last_group = groups[-1]
+        backfill_supergroups_lightweight_for_org.apply_async(
+            args=[organization_id],
+            kwargs={
+                "last_project_id": last_group.project_id,
+                "last_group_id": last_group.id,
+            },
+            countdown=INTER_BATCH_DELAY_S,
+            headers={"sentry-propagate-traces": False},
+        )
+    else:
+        logger.info(
+            "supergroups_backfill_lightweight.org_completed",
+            extra={"organization_id": organization_id},
+        )

tests/sentry/tasks/seer/test_backfill_supergroups_lightweight.py

@@ -0,0 +1,180 @@
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+from sentry.models.group import DEFAULT_TYPE_ID
+from sentry.tasks.seer.backfill_supergroups_lightweight import (
+    BATCH_SIZE,
+    backfill_supergroups_lightweight_for_org,
+)
+from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers.features import with_feature
+from sentry.types.group import GroupSubStatus
+
+
+class BackfillSupergroupsLightweightForOrgTest(TestCase):
+    def setUp(self):
+        super().setUp()
+        self.event = self.store_event(
+            data={"message": "test error", "level": "error"},
+            project_id=self.project.id,
+        )
+        self.group = self.event.group
+        self.group.substatus = GroupSubStatus.NEW
+        self.group.save(update_fields=["substatus"])
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_processes_groups_and_sends_to_seer(self, mock_request):
+        mock_request.return_value = MagicMock(status=200)
+
+        backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        mock_request.assert_called_once()
+        body = mock_request.call_args.args[0]
+        assert body["group_id"] == self.group.id
+        assert body["project_id"] == self.project.id
+        assert body["organization_id"] == self.organization.id
+        assert body["issue"]["id"] == self.group.id
+        assert len(body["issue"]["events"]) == 1
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_processes_groups_across_projects(self, mock_request):
+        mock_request.return_value = MagicMock(status=200)
+
+        project2 = self.create_project(organization=self.organization)
+        event2 = self.store_event(
+            data={"message": "error in project2", "level": "error"},
+            project_id=project2.id,
+        )
+        event2.group.substatus = GroupSubStatus.NEW
+        event2.group.save(update_fields=["substatus"])
+
+        backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        assert mock_request.call_count == 2
+        project_ids = {call.args[0]["project_id"] for call in mock_request.call_args_list}
+        assert project_ids == {self.project.id, project2.id}
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_self_chains_when_more_groups_exist(self, mock_request):
+        mock_request.return_value = MagicMock(status=200)
+
+        # Create enough groups to fill a batch
+        for i in range(BATCH_SIZE):
+            evt = self.store_event(
+                data={
+                    "message": f"error {i}",
+                    "level": "error",
+                    "fingerprint": [f"group-{i}"],
+                },
+                project_id=self.project.id,
+            )
+            evt.group.substatus = GroupSubStatus.NEW
+            evt.group.save(update_fields=["substatus"])
+
+        with patch(
+            "sentry.tasks.seer.backfill_supergroups_lightweight.backfill_supergroups_lightweight_for_org.apply_async"
+        ) as mock_chain:
+            backfill_supergroups_lightweight_for_org(self.organization.id)
+
+            mock_chain.assert_called_once()
+            call_kwargs = mock_chain.call_args.kwargs["kwargs"]
+            assert call_kwargs["last_project_id"] == self.project.id
+            assert call_kwargs["last_group_id"] > 0
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_does_not_chain_when_batch_incomplete(self, mock_request):
+        mock_request.return_value = MagicMock(status=200)
+
+        with patch(
+            "sentry.tasks.seer.backfill_supergroups_lightweight.backfill_supergroups_lightweight_for_org.apply_async"
+        ) as mock_chain:
+            backfill_supergroups_lightweight_for_org(self.organization.id)
+
+            mock_chain.assert_not_called()
+
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_respects_killswitch(self, mock_request):
+        with self.options({"seer.supergroups_backfill_lightweight.killswitch": True}):
+            backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        mock_request.assert_not_called()
+
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_skips_without_feature_flag(self, mock_request):
+        backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        mock_request.assert_not_called()
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_continues_on_individual_group_failure(self, mock_request):
+        event2 = self.store_event(
+            data={"message": "second error", "level": "error", "fingerprint": ["group2"]},
+            project_id=self.project.id,
+        )
+        event2.group.substatus = GroupSubStatus.NEW
+        event2.group.save(update_fields=["substatus"])
+
+        mock_request.side_effect = [
+            MagicMock(status=500),
+            MagicMock(status=200),
+        ]
+
+        backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        assert mock_request.call_count == 2
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_filters_old_groups(self, mock_request):
+        self.group.last_seen = datetime.now(UTC) - timedelta(days=91)
+        self.group.save(update_fields=["last_seen"])
+
+        backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        mock_request.assert_not_called()
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_skips_non_error_groups(self, mock_request):
+        self.group.type = DEFAULT_TYPE_ID + 1
+        self.group.save(update_fields=["type"])
+
+        backfill_supergroups_lightweight_for_org(self.organization.id)
+
+        mock_request.assert_not_called()
+
+    @with_feature("organizations:supergroups-lightweight-rca-clustering")
+    @patch(
+        "sentry.tasks.seer.backfill_supergroups_lightweight.make_lightweight_rca_cluster_request"
+    )
+    def test_resumes_from_cursor(self, mock_request):
+        backfill_supergroups_lightweight_for_org(
+            self.organization.id,
+            last_project_id=self.project.id,
+            last_group_id=self.group.id,
+        )
+
+        mock_request.assert_not_called()