feat(cron-monitors): Add billing seat management to cron monitor detectors

Implements billing seat management for cron monitor detectors via the
MonitorIncidentDetectorValidator. This ensures that cron monitors
created through the detector API properly handle seat assignment,
validation, and removal.

Changes:
- Add validate_enabled() to check seat availability before enabling
- Modify create() to assign seats with graceful degradation when no
  seats are available (detector created but disabled)
- Modify update() to handle enable/disable transitions with seat
  operations and race condition handling
- Add delete() to remove seats immediately when detector is deleted
- Add comprehensive test coverage for all seat management scenarios

Uses generic seat APIs (assign_seat, check_assign_seat, disable_seat,
remove_seat) with DataCategory.MONITOR following the same pattern as
uptime monitors.

Fixes [NEW-619: Ensure CRUD / enable+disable Cron Detectors in new UI handles assigning / unassigning seats](https://linear.app/getsentry/issue/NEW-619/ensure-crud-enabledisable-cron-detectors-in-new-ui-handles-assigning)

Author: evanpurkhiser

src/sentry/monitors/models.py

@@ -50,6 +50,7 @@
 
 if TYPE_CHECKING:
     from sentry.models.project import Project
+    from sentry.workflow_engine.models import Detector
 
 MONITOR_CONFIG = {
     "type": "object",
@@ -812,6 +813,15 @@ class Meta:
         db_table = "sentry_monitorenvbrokendetection"
 
 
+def get_cron_monitor(detector: Detector) -> Monitor:
+    """
+    Given a detector get the matching cron monitor.
+    """
+    data_source = detector.data_sources.first()
+    assert data_source
+    return Monitor.objects.get(id=int(data_source.source_id))
+
+
 @data_source_type_registry.register(DATA_SOURCE_CRON_MONITOR)
 class CronMonitorDataSourceHandler(DataSourceTypeHandler[Monitor]):
     @staticmethod

src/sentry/monitors/validators.py

@@ -18,7 +18,7 @@
 from sentry.api.fields.sentry_slug import SentrySerializerSlugField
 from sentry.api.serializers.rest_framework import CamelSnakeSerializer
 from sentry.api.serializers.rest_framework.project import ProjectField
-from sentry.constants import ObjectStatus
+from sentry.constants import DataCategory, ObjectStatus
 from sentry.db.models import BoundedPositiveIntegerField
 from sentry.db.models.fields.slug import DEFAULT_SLUG_MAX_LENGTH
 from sentry.db.postgres.transactions import in_test_hide_transaction_boundary
@@ -35,6 +35,7 @@
     MonitorLimitsExceeded,
     ScheduleType,
     check_organization_monitor_limit,
+    get_cron_monitor,
 )
 from sentry.monitors.schedule import get_next_schedule, get_prev_schedule
 from sentry.monitors.types import CrontabSchedule, slugify_monitor_slug
@@ -52,7 +53,7 @@
     BaseDataSourceValidator,
     BaseDetectorTypeValidator,
 )
-from sentry.workflow_engine.models import DataSource, Detector
+from sentry.workflow_engine.models import Detector
 
 MONITOR_STATUSES = {
     "active": ObjectStatus.ACTIVE,
@@ -686,16 +687,58 @@ class MonitorIncidentDetectorValidator(BaseDetectorTypeValidator):
 
     data_sources = serializers.ListField(child=MonitorDataSourceValidator(), required=False)
 
+    def validate_enabled(self, value: bool) -> bool:
+        """
+        Validate that enabling a detector is allowed based on seat availability.
+        """
+        detector = self.instance
+        if detector and value and not detector.enabled:
+            monitor = get_cron_monitor(detector)
+            result = quotas.backend.check_assign_seat(DataCategory.MONITOR, monitor)
+            if not result.assignable:
+                raise serializers.ValidationError(result.reason)
+        return value
+
+    def create(self, validated_data):
+        detector = super().create(validated_data)
+        monitor = get_cron_monitor(detector)
+
+        # Try to assign a seat for the monitor
+        seat_outcome = quotas.backend.assign_seat(DataCategory.MONITOR, monitor)
+        if seat_outcome != Outcome.ACCEPTED:
+            detector.update(enabled=False)
+            monitor.update(status=ObjectStatus.DISABLED)
+
+        return detector
+
     def update(self, instance: Detector, validated_data: dict[str, Any]) -> Detector:
+        was_enabled = instance.enabled
+        enabled = validated_data.get("enabled", was_enabled)
+
+        # Handle enable/disable seat operations
+        if was_enabled != enabled:
+            monitor = get_cron_monitor(instance)
+
+            if enabled:
+                seat_outcome = quotas.backend.assign_seat(DataCategory.MONITOR, monitor)
+                # We should have already validated that a seat was available in
+                # validate_enabled, avoid races by failing here if we can't
+                # accept the seat
+                if seat_outcome != Outcome.ACCEPTED:
+                    raise serializers.ValidationError("Failed to update monitor")
+                monitor.update(status=ObjectStatus.ACTIVE)
+            else:
+                quotas.backend.disable_seat(DataCategory.MONITOR, monitor)
+                monitor.update(status=ObjectStatus.DISABLED)
+
         super().update(instance, validated_data)
 
         data_source_data = None
         if "data_sources" in validated_data:
             data_source_data = validated_data.pop("data_sources")[0]
 
         if data_source_data is not None:
-            data_source = DataSource.objects.get(detectors=instance)
-            monitor = Monitor.objects.get(id=data_source.source_id)
+            monitor = get_cron_monitor(instance)
 
             monitor_validator = MonitorDataSourceValidator(
                 instance=monitor,
@@ -709,3 +752,12 @@ def update(self, instance: Detector, validated_data: dict[str, Any]) -> Detector
                     monitor_validator.save()
 
         return instance
+
+    def delete(self) -> None:
+        assert self.instance is not None
+        monitor = get_cron_monitor(self.instance)
+
+        # Remove the seat immediately
+        quotas.backend.remove_seat(DataCategory.MONITOR, monitor)
+
+        super().delete()

tests/sentry/monitors/test_validators.py

@@ -6,22 +6,25 @@
 from django.test import RequestFactory
 from django.test.utils import override_settings
 from django.utils import timezone
+from rest_framework.exceptions import ErrorDetail
 
 from sentry.analytics.events.cron_monitor_created import CronMonitorCreated, FirstCronMonitorCreated
-from sentry.constants import ObjectStatus
+from sentry.constants import DataCategory, ObjectStatus
 from sentry.models.rule import Rule, RuleSource
-from sentry.monitors.models import Monitor, MonitorStatus, ScheduleType
+from sentry.monitors.models import Monitor, MonitorStatus, ScheduleType, get_cron_monitor
+from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
 from sentry.monitors.validators import (
     MonitorDataSourceValidator,
     MonitorIncidentDetectorValidator,
     MonitorValidator,
 )
+from sentry.quotas.base import SeatAssignmentResult
 from sentry.testutils.cases import MonitorTestCase
 from sentry.testutils.helpers.analytics import assert_any_analytics_event
 from sentry.types.actor import Actor
 from sentry.utils.outcomes import Outcome
 from sentry.utils.slug import DEFAULT_SLUG_ERROR_MESSAGE
-from sentry.workflow_engine.models import DataConditionGroup
+from sentry.workflow_engine.models import DataConditionGroup, DataSource, DataSourceDetector
 
 
 class MonitorValidatorCreateTest(MonitorTestCase):
@@ -166,7 +169,6 @@ def test_monitor_organization_limit(self):
         }
         validator = MonitorValidator(data=data, context=self.context)
         assert not validator.is_valid()
-        from rest_framework.exceptions import ErrorDetail
 
         assert validator.errors["nonFieldErrors"] == [
             ErrorDetail(
@@ -1123,3 +1125,203 @@ def test_create_detector_validates_data_source(self):
         assert validator.is_valid(), validator.errors
         assert "_creator" in validator.validated_data["data_sources"][0]
         assert validator.validated_data["data_sources"][0]["data_source_type"] == "cron_monitor"
+
+    @patch("sentry.quotas.backend.assign_seat", return_value=Outcome.ACCEPTED)
+    def test_create_enabled_assigns_seat(self, mock_assign_seat):
+        """Test that creating an enabled detector assigns a billing seat."""
+
+        condition_group = DataConditionGroup.objects.create(
+            organization_id=self.organization.id,
+            logic_type=DataConditionGroup.Type.ANY,
+        )
+        context = {**self.context, "condition_group": condition_group}
+        validator = MonitorIncidentDetectorValidator(
+            data=self.valid_data,
+            context=context,
+        )
+        assert validator.is_valid(), validator.errors
+        detector = validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is True
+
+        # Verify seat was assigned
+        monitor = get_cron_monitor(detector)
+        mock_assign_seat.assert_called_with(DataCategory.MONITOR, monitor)
+
+    @patch("sentry.quotas.backend.assign_seat", return_value=Outcome.RATE_LIMITED)
+    def test_create_enabled_no_seat_available(self, mock_assign_seat):
+        """
+        Test that creating a detector with no seats available creates it but
+        leaves it disabled.
+        """
+        condition_group = DataConditionGroup.objects.create(
+            organization_id=self.organization.id,
+            logic_type=DataConditionGroup.Type.ANY,
+        )
+        context = {**self.context, "condition_group": condition_group}
+        validator = MonitorIncidentDetectorValidator(
+            data=self.valid_data,
+            context=context,
+        )
+        assert validator.is_valid(), validator.errors
+        detector = validator.save()
+
+        detector.refresh_from_db()
+        # Detector created but not enabled due to no seat assignment
+        assert detector.enabled is False
+        monitor = get_cron_monitor(detector)
+        assert monitor.status == ObjectStatus.DISABLED
+
+        # Verify seat assignment was attempted
+        mock_assign_seat.assert_called_with(DataCategory.MONITOR, monitor)
+
+    @patch("sentry.quotas.backend.assign_seat", return_value=Outcome.ACCEPTED)
+    def test_update_enable_assigns_seat(self, mock_assign_seat):
+        """
+        Test that enabling a previously disabled detector assigns a seat.
+        """
+        # Create a disabled detector
+        detector = self.create_detector(
+            project=self.project,
+            name="Test Detector",
+            type="monitor_check_in_failure",
+            enabled=False,
+        )
+        monitor = self._create_monitor(
+            name="Test Monitor",
+            slug="test-monitor",
+            status=ObjectStatus.DISABLED,
+        )
+        data_source = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            organization_id=self.organization.id,
+            source_id=str(monitor.id),
+        )
+        DataSourceDetector.objects.create(data_source=data_source, detector=detector)
+
+        validator = MonitorIncidentDetectorValidator(
+            instance=detector, data={"enabled": True}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        monitor.refresh_from_db()
+        assert detector.enabled is True
+        assert monitor.status == ObjectStatus.ACTIVE
+
+        # Verify seat was assigned
+        mock_assign_seat.assert_called_with(DataCategory.MONITOR, monitor)
+
+    @patch(
+        "sentry.quotas.backend.check_assign_seat",
+        return_value=SeatAssignmentResult(assignable=False, reason="No seats available"),
+    )
+    def test_update_enable_no_seat_available(self, mock_check_seat):
+        """
+        Test that enabling fails with validation error when no seats are
+        available.
+        """
+        # Create a disabled detector
+        detector = self.create_detector(
+            project=self.project,
+            name="Test Detector",
+            type="monitor_check_in_failure",
+            enabled=False,
+        )
+        monitor = self._create_monitor(
+            name="Test Monitor",
+            slug="test-monitor",
+            status=ObjectStatus.DISABLED,
+        )
+        data_source = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            organization_id=self.organization.id,
+            source_id=str(monitor.id),
+        )
+        DataSourceDetector.objects.create(data_source=data_source, detector=detector)
+
+        validator = MonitorIncidentDetectorValidator(
+            instance=detector, data={"enabled": True}, context=self.context, partial=True
+        )
+
+        # Validation should fail due to no seats available
+        assert not validator.is_valid()
+        assert "enabled" in validator.errors
+        assert validator.errors["enabled"] == ["No seats available"]
+
+        # Detector and monitor should still be disabled
+        detector.refresh_from_db()
+        monitor.refresh_from_db()
+        assert detector.enabled is False
+        assert monitor.status == ObjectStatus.DISABLED
+
+        # Verify seat availability check was performed
+        mock_check_seat.assert_called_with(DataCategory.MONITOR, monitor)
+
+    @patch("sentry.quotas.backend.disable_seat")
+    def test_update_disable_disables_seat(self, mock_disable_seat):
+        """Test that disabling a previously enabled detector disables the seat."""
+        # Create an enabled detector
+        detector = self.create_detector(
+            project=self.project,
+            name="Test Detector",
+            type="monitor_check_in_failure",
+            enabled=True,
+        )
+        monitor = self._create_monitor(
+            name="Test Monitor",
+            slug="test-monitor",
+            status=ObjectStatus.ACTIVE,
+        )
+        data_source = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            organization_id=self.organization.id,
+            source_id=str(monitor.id),
+        )
+        DataSourceDetector.objects.create(data_source=data_source, detector=detector)
+
+        validator = MonitorIncidentDetectorValidator(
+            instance=detector, data={"enabled": False}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        monitor.refresh_from_db()
+        assert detector.enabled is False
+        assert monitor.status == ObjectStatus.DISABLED
+
+        # Verify disable_seat was called
+        mock_disable_seat.assert_called_with(DataCategory.MONITOR, monitor)
+
+    @patch("sentry.quotas.backend.remove_seat")
+    def test_delete_removes_seat(self, mock_remove_seat: MagicMock) -> None:
+        """Test that deleting a detector removes its billing seat immediately."""
+        detector = self.create_detector(
+            project=self.project,
+            name="Test Detector",
+            type="monitor_check_in_failure",
+            enabled=True,
+        )
+        monitor = self._create_monitor(
+            name="Test Monitor",
+            slug="test-monitor",
+            status=ObjectStatus.ACTIVE,
+        )
+        data_source = DataSource.objects.create(
+            type=DATA_SOURCE_CRON_MONITOR,
+            organization_id=self.organization.id,
+            source_id=str(monitor.id),
+        )
+        DataSourceDetector.objects.create(data_source=data_source, detector=detector)
+
+        validator = MonitorIncidentDetectorValidator(
+            instance=detector, data={}, context=self.context
+        )
+
+        validator.delete()
+
+        # Verify remove_seat was called immediately
+        mock_remove_seat.assert_called_with(DataCategory.MONITOR, monitor)