fix(detectors): Make AlertRuleDetector.objects filter out pending deletion Detectors by default

src/sentry/deletions/defaults/alertrule.py

@@ -1,5 +1,10 @@
 from sentry.deletions.base import BaseRelation, ModelDeletionTask, ModelRelation
 from sentry.incidents.models.alert_rule import AlertRule
+from sentry.workflow_engine.models.alertrule_detector import AlertRuleDetector
+
+
+class AlertRuleDetectorDeletionTask(ModelDeletionTask[AlertRuleDetector]):
+    manager_name = "objects_for_deletion"
 
 
 class AlertRuleDeletionTask(ModelDeletionTask[AlertRule]):
@@ -10,7 +15,15 @@ class AlertRuleDeletionTask(ModelDeletionTask[AlertRule]):
     def get_child_relations(self, instance: AlertRule) -> list[BaseRelation]:
         from sentry.incidents.models.alert_rule import AlertRuleTrigger
         from sentry.incidents.models.incident import Incident
-        from sentry.workflow_engine.models import AlertRuleDetector, AlertRuleWorkflow
+        from sentry.workflow_engine.models import AlertRuleWorkflow
 
-        model_list = (AlertRuleTrigger, Incident, AlertRuleDetector, AlertRuleWorkflow)
-        return [ModelRelation(m, {"alert_rule_id": instance.id}) for m in model_list]
+        return [
+            ModelRelation(AlertRuleTrigger, {"alert_rule_id": instance.id}),
+            ModelRelation(Incident, {"alert_rule_id": instance.id}),
+            ModelRelation(
+                AlertRuleDetector,
+                {"alert_rule_id": instance.id},
+                task=AlertRuleDetectorDeletionTask,
+            ),
+            ModelRelation(AlertRuleWorkflow, {"alert_rule_id": instance.id}),
+        ]

src/sentry/deletions/defaults/detector.py

@@ -26,7 +26,7 @@ def get_child_relations(self, instance: Detector) -> list[BaseRelation]:
         # QuerySubscriptionDeletionTask sees no remaining AlertRule referencing
         # the SnubaQuery and can safely delete it.
         alert_rule_ids = list(
-            AlertRuleDetector.objects.filter(detector_id=instance.id).values_list(
+            AlertRuleDetector.objects_for_deletion.filter(detector_id=instance.id).values_list(
                 "alert_rule_id", flat=True
             )
         )

src/sentry/deletions/defaults/rule.py

@@ -2,6 +2,7 @@
 from collections.abc import Sequence
 
 from sentry.deletions.base import BaseRelation, ModelDeletionTask, ModelRelation
+from sentry.deletions.defaults.alertrule import AlertRuleDetectorDeletionTask
 from sentry.models.rule import Rule
 from sentry.workflow_engine.models import Workflow
 
@@ -19,7 +20,11 @@ def get_child_relations(self, instance: Rule) -> list[BaseRelation]:
             ModelRelation(GroupRuleStatus, {"rule_id": instance.id}),
             ModelRelation(RuleFireHistory, {"rule_id": instance.id}),
             ModelRelation(RuleActivity, {"rule_id": instance.id}),
-            ModelRelation(AlertRuleDetector, {"rule_id": instance.id}),
+            ModelRelation(
+                AlertRuleDetector,
+                {"rule_id": instance.id},
+                task=AlertRuleDetectorDeletionTask,
+            ),
         ]
 
         alert_rule_workflow = AlertRuleWorkflow.objects.filter(rule_id=instance.id).first()

src/sentry/incidents/endpoints/organization_alert_rule_details.py

@@ -155,7 +155,7 @@ def remove_alert_rule(
         try:
             remove_detector(request, organization, target)
             try:
-                ard = AlertRuleDetector.objects.get(detector_id=target.id)
+                ard = AlertRuleDetector.objects_for_deletion.get(detector_id=target.id)
                 target = AlertRule.objects.get(id=ard.alert_rule_id, organization=organization)
                 delete_alert_rule(
                     target,

src/sentry/workflow_engine/models/alertrule_detector.py

@@ -1,12 +1,36 @@
+from typing import ClassVar
+
 from django.db.models import CheckConstraint, Q, UniqueConstraint
 
 from sentry.backup.scopes import RelocationScope
+from sentry.constants import ObjectStatus
 from sentry.db.models import (
     BoundedBigIntegerField,
     DefaultFieldsModel,
     FlexibleForeignKey,
     cell_silo_model,
 )
+from sentry.db.models.manager.base import BaseManager
+from sentry.db.models.manager.base_query_set import BaseQuerySet
+
+
+class AlertRuleDetectorManager(BaseManager["AlertRuleDetector"]):
+    """
+    Default manager that excludes rows whose related Detector is
+    pending deletion or being deleted.
+    """
+
+    def get_queryset(self) -> BaseQuerySet["AlertRuleDetector"]:
+        return (
+            super()
+            .get_queryset()
+            .exclude(
+                detector__status__in=(
+                    ObjectStatus.PENDING_DELETION,
+                    ObjectStatus.DELETION_IN_PROGRESS,
+                )
+            )
+        )
 
 
 @cell_silo_model
@@ -17,6 +41,9 @@ class AlertRuleDetector(DefaultFieldsModel):
 
     __relocation_scope__ = RelocationScope.Organization
 
+    objects: ClassVar[AlertRuleDetectorManager] = AlertRuleDetectorManager()
+    objects_for_deletion: ClassVar[BaseManager["AlertRuleDetector"]] = BaseManager()
+
     alert_rule_id = BoundedBigIntegerField(null=True, db_index=True)
     rule_id = BoundedBigIntegerField(null=True, db_index=True)
     detector = FlexibleForeignKey("workflow_engine.Detector")

tests/sentry/incidents/endpoints/test_organization_alert_rule_details.py

@@ -23,6 +23,7 @@
 from sentry.api.serializers import serialize
 from sentry.auth.access import OrganizationGlobalAccess
 from sentry.conf.server import SEER_ANOMALY_DETECTION_STORE_DATA_URL
+from sentry.constants import ObjectStatus
 from sentry.deletions.tasks.scheduled import run_scheduled_deletions
 from sentry.incidents.endpoints.serializers.alert_rule import DetailedAlertRuleSerializer
 from sentry.incidents.endpoints.serializers.utils import get_fake_id_from_object_id
@@ -321,6 +322,33 @@ def test_workflow_engine_serializer(self) -> None:
         assert resp.data["id"] == str(self.alert_rule.id)
         assert resp.data["name"] == self.detector.name
 
+    @with_feature("organizations:workflow-engine-rule-serializers")
+    @with_feature("organizations:incidents")
+    def test_pending_deletion_detector_returns_404(self) -> None:
+        self.create_team(organization=self.organization, members=[self.user])
+        self.login_as(self.user)
+
+        # Dual-written detector (has an AlertRuleDetector mapping) — look up by alert_rule_id
+        dual_written_detector = self.create_detector(
+            project=self.project,
+            type=MetricIssue.slug,
+            status=ObjectStatus.PENDING_DELETION,
+        )
+        alert_rule = self.create_alert_rule(projects=[self.project])
+        self.create_alert_rule_detector(detector=dual_written_detector, alert_rule_id=alert_rule.id)
+
+        self.get_error_response(self.organization.slug, alert_rule.id, status_code=404)
+
+        # Single-written detector (no ARD) — look up by fake detector ID
+        single_written_detector = self.create_detector(
+            project=self.project,
+            type=MetricIssue.slug,
+            status=ObjectStatus.PENDING_DELETION,
+        )
+        fake_detector_id = get_fake_id_from_object_id(single_written_detector.id)
+
+        self.get_error_response(self.organization.slug, fake_detector_id, status_code=404)
+
     @with_feature("organizations:incidents")
     @freeze_time("2024-12-11 03:21:34")
     def test_workflow_engine_serializer_matches_old_serializer(self) -> None:

tests/sentry/incidents/endpoints/test_organization_incident_index.py

@@ -4,6 +4,7 @@
 from django.utils import timezone
 
 from sentry.api.serializers import serialize
+from sentry.constants import ObjectStatus
 from sentry.incidents.endpoints.serializers.utils import get_fake_id_from_object_id
 from sentry.incidents.grouptype import MetricIssue
 from sentry.incidents.logic import update_incident_status
@@ -27,7 +28,7 @@
     migrate_metric_data_conditions,
     migrate_resolve_threshold_data_condition,
 )
-from sentry.workflow_engine.models import DetectorGroup, IncidentGroupOpenPeriod
+from sentry.workflow_engine.models import Detector, DetectorGroup, IncidentGroupOpenPeriod
 from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.types import DetectorPriorityLevel
 
@@ -444,3 +445,39 @@ def test_filter_by_fake_alert_rule_id(self) -> None:
         )
 
         assert len(resp_empty.data) == 0
+
+    @with_feature(
+        [
+            "organizations:incidents",
+            "organizations:performance-view",
+            "organizations:workflow-engine-rule-serializers",
+        ]
+    )
+    def test_filter_by_alert_rule_excludes_pending_deletion_detector(self) -> None:
+        self.create_team(organization=self.organization, members=[self.user])
+        self.login_as(self.user)
+
+        alert_rule = self.create_alert_rule()
+        trigger = self.create_alert_rule_trigger(alert_rule=alert_rule)
+        _, _, _, detector, _, _, _, _ = migrate_alert_rule(alert_rule)
+        migrate_metric_data_conditions(trigger)
+        migrate_resolve_threshold_data_condition(alert_rule)
+
+        with assume_test_silo_mode(SiloMode.CELL):
+            group = self.create_group(type=MetricIssue.type_id, project=self.project)
+            group.priority = PriorityLevel.HIGH.value
+            group.save()
+            DetectorGroup.objects.create(detector=detector, group=group)
+
+        # Verify the incident shows up before deletion
+        resp = self.get_success_response(self.organization.slug, alertRule=str(alert_rule.id))
+        assert len(resp.data) == 1
+
+        # Mark the detector as pending deletion (bypass custom manager)
+        Detector.objects_for_deletion.filter(id=detector.id).update(
+            status=ObjectStatus.PENDING_DELETION
+        )
+
+        # The alert rule filter should no longer find this detector
+        resp = self.get_success_response(self.organization.slug, alertRule=str(alert_rule.id))
+        assert len(resp.data) == 0

tests/sentry/incidents/test_charts.py

@@ -5,6 +5,7 @@
 from django.utils.dateparse import parse_datetime
 
 from sentry.api.serializers import serialize
+from sentry.constants import ObjectStatus
 from sentry.incidents.charts import (
     build_metric_alert_chart,
     fetch_metric_issue_open_periods,
@@ -30,7 +31,7 @@
 from sentry.testutils.helpers.datetime import freeze_time
 from sentry.testutils.helpers.features import with_feature
 from sentry.types.group import PriorityLevel
-from sentry.workflow_engine.models import DetectorGroup
+from sentry.workflow_engine.models import Detector, DetectorGroup
 from tests.sentry.incidents.utils.test_metric_issue_base import BaseMetricIssueTest
 
 now = "2022-05-16T20:00:00"
@@ -272,6 +273,37 @@ def test_get_incidents_from_detector(self) -> None:
         assert created_activity_resp["type"] == IncidentActivityType.CREATED.value
         assert created_activity_resp["dateCreated"] == created_activity.date_added
 
+    @freeze_time(frozen_time)
+    @with_feature("organizations:incidents")
+    def test_pending_deletion_detector_not_resolved_to_alert_rule(self) -> None:
+        self.create_detector()  # dummy so detector ID != alert rule ID
+        detector = self.create_detector(project=self.project)
+        alert_rule = self.create_alert_rule(organization=self.organization, projects=[self.project])
+        self.create_alert_rule_detector(detector=detector, alert_rule_id=alert_rule.id)
+        incident = self.create_incident(
+            date_started=must_parse_datetime("2022-05-16T18:55:00Z"),
+            status=IncidentStatus.CRITICAL.value,
+            alert_rule=alert_rule,
+        )
+        self.create_incident_activity(
+            incident,
+            IncidentActivityType.DETECTED.value,
+            date_added=incident.date_started,
+        )
+
+        time_period = incident_date_range(60, incident.date_started, incident.date_closed)
+
+        # Mark the detector as pending deletion (bypass custom manager)
+        Detector.objects_for_deletion.filter(id=detector.id).update(
+            status=ObjectStatus.PENDING_DELETION
+        )
+
+        # The AlertRuleDetector lookup should skip the deleted detector,
+        # so the detector ID won't be mapped to the alert rule ID and
+        # no incidents will be found
+        chart_data = fetch_metric_issue_open_periods(self.organization, detector.id, time_period)
+        assert len(chart_data) == 0
+
     @freeze_time(frozen_time)
     @with_feature("organizations:incidents")
     @with_feature("organizations:workflow-engine-ui")