feat(alerts): let the cutover move real-time insight alerts

products/alerts/backend/max_tools.py

@@ -344,7 +344,7 @@
                 "details": str(e),
             }
 
     def _save_alert(
         self,
         alert: AlertConfiguration,
         action: UpdateAlertAction,
@@ -477,7 +477,7 @@
                 from products.alerts.backend.platform_ownership import on_updated  # noqa: PLC0415
 
                 try:
-                    on_updated(alert, snooze_changed=False)
+                    on_updated(alert, snooze_changed=False, rescheduled=schedule_reset_required)
                 except (PlatformCannotOwn, InvalidPlatformAlertEdit) as error:
                     # Rolls back the insight row, so neither row takes an edit the platform rejects.
                     transaction.set_rollback(True)

products/alerts/backend/platform_cutover.py

@@ -15,7 +15,7 @@
 from django.utils import timezone
 
 from posthog.dataclasses import frozen
-from posthog.schema_enums import AlertCalculationInterval, AlertState
+from posthog.schema_enums import AlertState
 from posthog.tasks.alerts.utils import due_insight_alerts_q
 
 from products.alerts.backend.facade.destinations import choose_delivery_path
@@ -69,7 +69,7 @@ class CutoverSkip(StrEnum):
     ALREADY_MOVED = "already_moved"
     # Platform rows live on the project's root team, so an alert in a child environment stays.
     ENVIRONMENT = "environment"
-    # The platform evaluates it only in parallel, such as a real-time alert.
+    # The platform cannot copy it, such as an alert with no cadence.
     NOT_ON_PLATFORM = "not_on_platform"
     # The investigation agent reads and writes the insight stack's checks, which the platform does not keep.
     INVESTIGATION = "investigation"
@@ -114,11 +114,6 @@ def cut_over_team(team_id: int, *, now: datetime | None = None) -> tuple[AlertCu
     )
 
 
-def _is_scheduled(alert: AlertConfiguration) -> bool:
-    """The platform evaluates real-time alerts in parallel, but cannot own them yet."""
-    return alert.calculation_interval != AlertCalculationInterval.REAL_TIME
-
-
 def delivers_as_it_does_today(alert: AlertConfiguration) -> bool:
     """Whether the platform can deliver the alert the way its destinations expect: natively, or
     through its HogFunctions when the insight events those listen on are emitted."""
@@ -167,9 +162,9 @@ def in_an_environment(alert: AlertConfiguration) -> bool:
 
 
 def can_move(alert: AlertConfiguration) -> bool:
-    """Whether the platform can own the alert: a scheduled alert it evaluates, with a threshold or a
-    detector, which no investigation agent reads."""
-    return platform_upsert_of(alert) is not None and _is_scheduled(alert) and not alert.investigation_agent_enabled
+    """Whether the platform can own the alert: one it evaluates, with a threshold or a detector, which
+    no investigation agent reads."""
+    return platform_upsert_of(alert) is not None and not alert.investigation_agent_enabled
 
 
 def has_begun_cutover(team_id: int) -> bool:

products/alerts/backend/platform_ownership.py

@@ -30,11 +30,12 @@
     SOURCE_CONDITION_KEY,
     InvalidPlatformAlertEdit,
     PlatformAlertConfigurationEdit,
+    PlatformAlertUpsert,
 )
 from products.alerts_platform.backend.facade.enums import PlatformAlertOwner
 
 # What an edit through either API can change on both rows. The cadence is a schedule, which the
-# platform's edit leaves out, so the check carries it onto the copy instead.
+# platform's edit leaves out, so an insight edit writes it onto the copy separately.
 _EDITED_FIELDS = ("name", "source_config", "schedule_restriction")
 
 
@@ -50,22 +51,25 @@ def on_created(alert: AlertConfiguration, at: datetime) -> bool:
     )
 
 
-def on_updated(alert: AlertConfiguration, *, snooze_changed: bool) -> None:
-    """Writes an edit the insight API saved onto the platform's copy.
+def on_updated(alert: AlertConfiguration, *, snooze_changed: bool, rescheduled: bool = False) -> None:
+    """Writes an edit the insight API saved onto the platform's copy. `rescheduled` says the edit
+    moved the alert's own due time, which the copy then follows.
 
-    Raises `PlatformCannotOwn` for an edit the platform cannot evaluate, such as a real-time check, and
+    Raises `PlatformCannotOwn` for an edit the platform cannot evaluate, such as one the investigation agent reads, and
     `InvalidPlatformAlertEdit` for one the platform's schema rejects.
     """
     if not can_move(alert):
-        raise PlatformCannotOwn(
-            "The alerts platform runs this alert, and it cannot run real-time checks or the investigation agent yet."
-        )
+        raise PlatformCannotOwn("The alerts platform runs this alert, and it cannot run the investigation agent yet.")
+    upsert = platform_upsert_of(alert)
+    assert upsert is not None
     update_owned_copy(
         team_id=alert.team_id,
         legacy_configuration_id=alert.id,
-        change=_platform_change_of(alert),
+        change=_platform_change_of(upsert),
         snooze_until=alert.snoozed_until,
         snooze_changed=snooze_changed,
+        cadence=upsert,
+        rescheduled=rescheduled,
     )
 
 
@@ -76,14 +80,14 @@ def hand_over_new_alert(*, team_id: int, alert_id: UUID, at: datetime) -> bool:
     return alert is not None and on_created(alert, at)
 
 
-def write_moved_alert_edit(*, team_id: int, alert_id: UUID, snooze_changed: bool) -> None:
+def write_moved_alert_edit(*, team_id: int, alert_id: UUID, snooze_changed: bool, rescheduled: bool = False) -> None:
     """Writes an edit the insight API saved in the caller's transaction onto the platform's copy.
 
     Raises like `on_updated`.
     """
     alert = _locked(team_id, alert_id)
     if alert is not None and alert.platform_owned:
-        on_updated(alert, snooze_changed=snooze_changed)
+        on_updated(alert, snooze_changed=snooze_changed, rescheduled=rescheduled)
 
 
 def close_moved_alert_firings(*, team_id: int, alert_id: UUID) -> None:
@@ -118,12 +122,10 @@ def _locked(team_id: int, alert_id: UUID) -> AlertConfiguration | None:
 
 
 def _platform_change_of(
-    alert: AlertConfiguration,
+    upsert: PlatformAlertUpsert,
 ) -> Callable[[PlatformAlertConfigurationEdit], PlatformAlertConfigurationEdit]:
     # Only the settings an insight alert has. The platform's own, such as grouping, keep what the
     # copy stores, so an edit through the insight API never resets them.
-    upsert = platform_upsert_of(alert)
-    assert upsert is not None
     shared = {field: getattr(upsert, field) for field in _EDITED_FIELDS}
     return lambda stored: replace(stored, enabled=upsert.enabled, **shared)
 
@@ -134,7 +136,7 @@ def mirror_platform_edit(
     """Locks the insight row of an alert edited through the platform API, and returns its writer."""
     alert = (
         AlertConfiguration.objects.select_for_update(no_key=True, of=("self",))
-        .select_related("threshold", "insight")
+        .select_related("threshold", "insight", "team__organization")
         # A copy lives on the project's team. An insight alert in one of its environments is never
         # handed over, so the copy of one is the insight stack's and refuses the edit.
         .filter(team_id=team_id, id=legacy_configuration_id)
@@ -156,6 +158,20 @@ def _write_insight_row(alert: AlertConfiguration, edit: PlatformAlertConfigurati
             "enabled",
             "An AI detector alert can't be turned back on through this API. Turn it on in the insight's alert settings.",
         )
+    if edit.enabled and not alert.enabled:
+        # The insight API and the Max tool check the plan and the active real-time limit before they
+        # turn an alert on. Without the plan check, the next evaluation would switch it off again.
+        organization = alert.team.organization
+        if admission_error := AlertConfiguration.interval_entitlement_error(
+            calculation_interval=alert.calculation_interval, organization=organization
+        ) or AlertConfiguration.real_time_alert_validation_error(
+            team_id=alert.team_id,
+            organization=organization,
+            calculation_interval=alert.calculation_interval,
+            enabled=True,
+            existing=alert,
+        ):
+            raise InvalidPlatformAlertEdit("enabled", admission_error)
     condition = edit.source_config[SOURCE_CONDITION_KEY]
     if condition.get("detector") != alert.detector_config:
         # A detector config is checked against its insight and its access rules in the insight API only.

products/alerts/backend/presentation/views/alert.py

@@ -927,7 +927,7 @@
 
     @transaction.atomic
     def update(self, instance, validated_data):
         current_insight_id = AlertConfiguration.objects.get(pk=instance.pk).insight_id
         target_insight = validated_data.get("insight")
         insight_ids = {current_insight_id, target_insight.id if target_insight else current_insight_id}
         # Use the evaluation lock order so query edits cannot pass validation during this save.
@@ -1001,7 +1001,8 @@
             "calculation_interval" in validated_data
             and validated_data["calculation_interval"] != instance.calculation_interval
         )
-        if enable_now or evaluation_changed or calculation_interval_changed:
+        rescheduled = enable_now or evaluation_changed or calculation_interval_changed
+        if rescheduled:
             if evaluation_changed:
                 apply_threshold_change(instance)
             # Keep the due timestamp so the scheduler metric can measure a
@@ -1039,6 +1040,7 @@
             schedule_start_time_changed = validated_data["schedule_start_time"] != instance.schedule_start_time
 
         instance = super().update(instance, validated_data)
+        rescheduled = rescheduled or schedule_restriction_changed or schedule_start_time_changed
         if schedule_restriction_changed or schedule_start_time_changed:
             instance.next_check_at = next_check_at_after_schedule_restriction_change(instance)
             instance.save(update_fields=["next_check_at"])
@@ -1048,6 +1050,7 @@
                     team_id=instance.team_id,
                     alert_id=instance.id,
                     snooze_changed=snoozed_until_param is not serializers.empty,
+                    rescheduled=rescheduled,
                 )
             except PlatformCannotOwn as error:
                 raise ValidationError(str(error)) from error
@@ -1101,7 +1104,7 @@
             raise serializers.ValidationError("Invalid schedule restriction.")
 
     def validate(self, attrs):
         if attrs.get("insight") and attrs["insight"].team.id != self.context["team_id"]:
             raise ValidationError({"insight": ["This insight does not belong to your team."]})
 
         condition = attrs.get("condition", self.instance.condition if self.instance else None)

products/alerts/backend/test/test_platform_cutover.py

@@ -39,6 +39,7 @@ def _alert(self, **overrides: Any) -> AlertConfiguration:
             ("disabled", {"enabled": False, "next_check_at": None}, "not_firing"),
             ("statistical_detector", {"detector_config": {"type": "zscore"}}, "not_firing"),
             ("ai_detector", {"detector_config": {"type": "llm"}}, "not_firing"),
+            ("real_time", {"calculation_interval": AlertCalculationInterval.REAL_TIME.value}, "not_firing"),
         ]
     )
     def test_an_alert_moves_with_its_state_and_schedule(
@@ -66,11 +67,6 @@ def test_an_alert_moves_with_its_state_and_schedule(
             ("firing", {"state": AlertState.FIRING}, CutoverSkip.FIRING),
             ("check_due", {"next_check_at": NOW}, CutoverSkip.CHECK_DUE),
             ("never_scheduled", {"next_check_at": None}, CutoverSkip.CHECK_DUE),
-            (
-                "real_time",
-                {"calculation_interval": AlertCalculationInterval.REAL_TIME.value},
-                CutoverSkip.NOT_ON_PLATFORM,
-            ),
             ("investigated", {"investigation_agent_enabled": True}, CutoverSkip.INVESTIGATION),
         ]
     )

products/alerts/backend/tests/api/test_alert_platform_owned.py

@@ -292,11 +292,27 @@ def edit(stored: PlatformAlertConfigurationEdit) -> PlatformAlertConfigurationEd
         assert (self.alert.threshold.configuration["type"], self.alert.detector_config) == ("absolute", None)
         assert self._copy().source_config == stored_config
 
-    def test_a_platform_edit_cannot_turn_an_ai_alert_back_on(self) -> None:
+    @parameterized.expand(
+        [
+            ("ai_alert", {"detector_config": {"type": "llm"}}, True, None),
+            ("real_time_alert_over_the_limit", {"calculation_interval": "real_time"}, True, "Limit reached."),
+            ("real_time_alert_without_the_plan", {"calculation_interval": "real_time"}, False, None),
+        ]
+    )
+    def test_a_platform_edit_cannot_turn_an_alert_back_on_past_its_admission(
+        self, _name: str, stored: dict[str, Any], has_real_time_plan: bool, real_time_limit_error: str | None
+    ) -> None:
+        self.organization.available_product_features = (
+            [{"key": AvailableFeature.REAL_TIME_ALERTS, "name": "Real-time alerts"}] if has_real_time_plan else []
+        )
+        self.organization.save()
         self._edit_on_the_platform(lambda stored: replace(stored, enabled=False))
-        AlertConfiguration.objects.filter(id=self.alert.id).update(detector_config={"type": "llm"})
+        AlertConfiguration.objects.filter(id=self.alert.id).update(**stored)
 
-        with pytest.raises(InvalidPlatformAlertEdit) as refused:
+        with (
+            patch.object(AlertConfiguration, "check_real_time_alert_limit", return_value=real_time_limit_error),
+            pytest.raises(InvalidPlatformAlertEdit) as refused,
+        ):
             self._edit_on_the_platform(lambda stored: replace(stored, enabled=True))
 
         assert refused.value.field_name == "enabled"
@@ -322,23 +338,35 @@ def test_a_snooze_the_platform_refuses_leaves_both_rows_unsnoozed(self) -> None:
         self.alert.refresh_from_db()
         assert (self.alert.snoozed_until, self._copy().snooze_until) == (None, None)
 
-    def test_an_edit_the_platform_cannot_run_is_refused_on_both_rows(self) -> None:
-        # Real-time passes the insight API's own checks on this insight, so only the platform refuses it.
+    @parameterized.expand(
+        [
+            ("to_real_time", {"calculation_interval": AlertCalculationInterval.REAL_TIME.value}, True),
+            ("threshold", {"threshold": {"configuration": {"type": "absolute", "bounds": {"upper": 50}}}}, True),
+            ("name_only", {"name": "Renamed"}, False),
+        ]
+    )
+    def test_an_edit_reschedules_the_copy_when_it_reschedules_the_insight_alert(
+        self, _name: str, edit: dict[str, Any], rescheduled: bool
+    ) -> None:
         self.organization.available_product_features = [
             {"key": AvailableFeature.REAL_TIME_ALERTS, "name": "Real-time alerts"}
         ]
         self.organization.save()
+        with team_scope(self.team.id):
+            due_before = platform_testing.copy_of(self.alert.id).next_check_at
 
-        response = self.client.patch(
-            f"{self.url}/{self.alert.id}/",
-            {"calculation_interval": AlertCalculationInterval.REAL_TIME.value},
-            format="json",
-        )
+        response = self.client.patch(f"{self.url}/{self.alert.id}/", edit, format="json")
 
-        assert response.status_code == status.HTTP_400_BAD_REQUEST, response.json()
-        assert "alerts platform" in str(response.json())
+        assert response.status_code == status.HTTP_200_OK, response.json()
         self.alert.refresh_from_db()
-        assert self.alert.calculation_interval == AlertCalculationInterval.DAILY.value
+        with team_scope(self.team.id):
+            due_after = platform_testing.copy_of(self.alert.id).next_check_at
+        assert self.alert.owner == "platform"
+        assert due_before is not None and due_after is not None and self.alert.next_check_at is not None
+        if rescheduled:
+            assert self.alert.next_check_at + timedelta(minutes=1) <= due_after < due_before
+        else:
+            assert due_after == due_before
 
     @parameterized.expand([("a_team_on_the_platform", True), ("a_team_off_the_platform", False)])
     def test_a_new_alert_is_handed_over_only_in_a_team_on_the_platform(self, _name: str, begun: bool) -> None:

products/alerts_platform/backend/facade/api.py

@@ -311,8 +311,12 @@ def update_owned_copy(
     change: Callable[[PlatformAlertConfigurationEdit], PlatformAlertConfigurationEdit],
     snooze_until: datetime | None = None,
     snooze_changed: bool = False,
+    cadence: PlatformAlertUpsert | None = None,
+    rescheduled: bool = False,
 ) -> None:
-    """Applies an edit a source made to its own row onto the copy the platform owns.
+    """Applies an edit a source made to its own row onto the copy the platform owns. When `cadence`
+    changes the copy's cadence, or `rescheduled` says the source moved its own due time, the copy is
+    next due at the cadence's `next_check_at`.
 
     Raises `InvalidPlatformAlertEdit` for an edit the source kind's schema rejects.
     """
@@ -322,6 +326,8 @@ def update_owned_copy(
         change=change,
         snooze_until=snooze_until,
         snooze_changed=snooze_changed,
+        cadence=cadence,
+        rescheduled=rescheduled,
     )
 
 

products/alerts_platform/backend/logic/platform_lifecycle.py

@@ -515,7 +515,7 @@
     return replace(group, kind=AlertEventKind.CHECK, notified=False, muted_notification=held)
 
 
 def record_outcomes(team_id: int, outcomes: Sequence[PlatformAlertOutcome], now: datetime) -> int:
     """Persists a batch's decisions and advances each configuration's schedule.
 
     Two statements rather than two per alert, in one transaction, so a crash between them cannot
@@ -958,6 +958,37 @@
     )
 
 
+def write_owned_cadence(
+    configuration: PlatformAlertConfiguration, upsert: PlatformAlertUpsert, *, rescheduled: bool
+) -> None:
+    """Writes a changed cadence onto a copy the platform owns, locked by the caller, and makes it due
+    when its source says.
+
+    The source computes that due time when the edit changes the cadence, or when `rescheduled` says
+    the source moved its own due time, as `upsert_configuration` takes it for a copy the source
+    owns. Any other edit keeps the copy's own due time.
+
+    Raises `ValueError` on a cadence `upsert_configuration` would refuse.
+    """
+    if upsert.recurrence_unit is not None:
+        to_recurrence_interval(upsert.recurrence_unit)
+    cadence = {
+        "check_interval_minutes": upsert.check_interval_minutes,
+        "recurrence_unit": upsert.recurrence_unit,
+        "anchor_time": validate_and_normalize_schedule_start_time(upsert.anchor_time),
+    }
+    if not rescheduled and all(getattr(configuration, key) == value for key, value in cadence.items()):
+        return
+    for key, value in cadence.items():
+        setattr(configuration, key, value)
+    now = django_timezone.now()
+    # A minute out at the earliest, so the recheck gets a slot of its own. A due time in the minute
+    # of a check that just finished would reuse that check's evaluation key, and the recheck's
+    # history and notification would be dropped as duplicates.
+    configuration.next_check_at = max(upsert.next_check_at or now, now + _HANDOVER_FENCE)
+    configuration.save(update_fields=[*_CADENCE_FIELDS, "next_check_at", "updated_at"])
+
+
 _HANDOVER_FENCE: Final = timedelta(minutes=1)
 
 

products/alerts_platform/backend/logic/platform_writes.py

@@ -31,6 +31,7 @@
     PlatformAlertConfigurationView,
     PlatformAlertNotEditable,
     PlatformAlertSnapshot,
+    PlatformAlertUpsert,
     SourceCreate,
     SourceKind,
     UnknownPlatformAlertGroup,
@@ -66,7 +67,7 @@
     )
 
 
 def _validated(
     source_kind: SourceKind,
     current: PlatformAlertConfigurationEdit | None,
     edit: PlatformAlertConfigurationEdit,
@@ -289,10 +290,12 @@
     change: Callable[[PlatformAlertConfigurationEdit], PlatformAlertConfigurationEdit],
     snooze_until: datetime | None,
     snooze_changed: bool,
+    cadence: PlatformAlertUpsert | None = None,
+    rescheduled: bool = False,
 ) -> None:
     """Applies an edit the source made to its own row onto the platform's copy. Nothing is mirrored
     back, because the source already wrote its row. A snooze the edit set or cleared becomes the
-    copy's mute."""
+    copy's mute. A changed `cadence`, or `rescheduled`, makes the copy due when the source is."""
     with transaction.atomic():
         configuration = _locked_owned_copy(team_id, legacy_configuration_id)
         if configuration is None:
@@ -301,6 +304,8 @@
         # A snooze ends the firing, as on the logs stack. A firing after it opens a new incident.
         snoozing = snooze_changed and snooze_until is not None and snooze_until > now
         _, pending = _apply(configuration, change, now=now, snoozing=snoozing)
+        if cadence is not None:
+            platform_lifecycle.write_owned_cadence(configuration, cadence, rescheduled=rescheduled)
         if snooze_changed:
             platform_lifecycle.write_snooze(configuration, snooze_until, now=now)
         control_plane.announce_after_commit(pending)