feat(alerts): investigate platform-owned insight firings with the agent

products/alerts/backend/platform_investigation.py

@@ -8,6 +8,7 @@
 write, and runs once the evaluation is recorded.
 """
 
+from dataclasses import replace
 from datetime import UTC, datetime
 from typing import Any, Final
 from uuid import UUID
@@ -17,17 +18,42 @@
 from posthog.dataclasses import frozen
 from posthog.temporal.alerts.investigation import (
     InvestigationDecision,
+    carried_verdict_suppresses,
     decide_within_budget,
     investigates,
     refusing_investigations,
 )
 
 from products.alerts.backend.investigation_episode import STARTED_INVESTIGATION_STATUSES
-from products.alerts.backend.investigation_release import earlier_in_episode, last_verdict
+from products.alerts.backend.investigation_release import (
+    RUNNING_HOLD,
+    earlier_in_episode,
+    last_verdict,
+    previous_verdict_in_episode,
+    release_held,
+)
 from products.alerts.backend.models.alert import AlertConfiguration, InvestigationStatus
 from products.alerts.backend.models.investigation import InsightAlertInvestigation
+from products.alerts.backend.platform_investigation_detail import (
+    INVESTIGATION_REQUESTED_KEY,
+    InvestigationClaims,
+    InvestigationStart,
+    requested_investigation,
+)
+from products.alerts_platform.backend.facade.contracts import (
+    AlertEventKind,
+    HeldDelivery,
+    HeldDeliveryAddress,
+    PlatformAlertCheckInput,
+    PlatformAlertOutcome,
+    SourceBatchEvaluation,
+)
+from products.alerts_platform.backend.facade.enums import HoldReleasedBy
+from products.alerts_platform.backend.facade.lifecycle import AlertState, NotificationAction
 
 _CLAIM_LOCK_SQL: Final = "SELECT pg_advisory_xact_lock(hashtext(%s))"
+HOLD_REASON: Final = "investigation"
+_EVALUATED_KEYS: Final = ("anomaly_scores", "triggered_points", "triggered_dates", "interval", "triggered_metadata")
 # A verdict-change follow-up is recorded under the key of the firing it follows up plus this suffix,
 # so it gets a delivery of its own.
 _VERDICT_CHANGE_KEY_SUFFIX: Final = "#verdict"
@@ -111,3 +137,104 @@ def claim_investigation(
             evaluated=evaluated,
         )
     return InvestigationClaim(investigation_id=investigation.id, claimed=not refused)
+
+
+def with_investigation(check: PlatformAlertCheckInput, evaluation: SourceBatchEvaluation) -> SourceBatchEvaluation:
+    """The evaluation of a check the platform owns, with the insight stack's investigation rules applied.
+
+    A firing the agent investigates asks for its claim in `source_detail`, and a gated alert holds the
+    firing's delivery until the verdict. A firing past the budget is muted when the firing's last
+    verdict suppresses it. Only an owned check, because a parallel run must never pay for an agent run.
+    """
+    if not check.owned_by_platform or check.legacy_configuration_id is None or len(evaluation.outcomes) != 1:
+        return evaluation
+    outcome = evaluation.outcomes[0]
+    # A skipped check keeps a firing alert's state without evaluating it, so it has nothing to investigate.
+    if outcome.skipped or len(outcome.groups) != 1 or outcome.groups[0].new_state != AlertState.FIRING.value:
+        return evaluation
+    alert = AlertConfiguration.objects.filter(team_id=check.team_id, id=check.legacy_configuration_id).first()
+    if alert is None:
+        return evaluation
+    group = outcome.groups[0]
+    decision = decide_investigation(
+        alert,
+        firing=True,
+        configuration_id=check.id,
+        evaluation_key=outcome.evaluation_key,
+        episode_started_at=group.firing_episode.started_at if group.firing_episode else None,
+    )
+    if decision.should_investigate:
+        group = replace(group, source_detail={**group.source_detail, INVESTIGATION_REQUESTED_KEY: str(alert.id)})
+        held = None
+        if evaluation.deliveries and alert.investigation_gates_notifications:
+            (request,) = evaluation.deliveries
+            held = HeldDelivery(
+                request=request, reason=HOLD_REASON, release_after_seconds=int(RUNNING_HOLD.total_seconds())
+            )
+        return replace(
+            evaluation,
+            outcomes=(replace(outcome, groups=(group,), held_delivery=held),),
+            deliveries=() if held else evaluation.deliveries,
+        )
+    if group.kind == AlertEventKind.FIRING and carried_verdict_suppresses(alert, decision):
+        # As the insight stack: the firing still moves the alert's state and `last_notified_at`.
+        group = replace(group, kind=AlertEventKind.CHECK, muted_notification=NotificationAction.FIRE.value)
+        return replace(evaluation, outcomes=(replace(outcome, groups=(group,)),), deliveries=())
+    return evaluation
+
+
+def claim_requested_investigations(team_id: int, outcomes: tuple[PlatformAlertOutcome, ...]) -> InvestigationClaims:
+    """Claims the investigation each recorded firing asked for, and returns the agent runs to start.
+
+    A refused claim releases the firing's held delivery by the firing's last verdict, as the insight
+    stack mutes a check that runs no investigation of its own.
+    """
+    starts: list[InvestigationStart] = []
+    deliveries = []
+    for outcome in outcomes:
+        requested = requested_investigation(outcome)
+        if requested is None:
+            continue
+        alert = AlertConfiguration.objects.filter(team_id=team_id, id=requested.alert_id).first()
+        if alert is None:
+            # The safety net sends the held firing at its deadline.
+            continue
+        group = outcome.groups[0]
+        claim = claim_investigation(
+            alert,
+            configuration_id=outcome.configuration_id,
+            evaluation_key=outcome.evaluation_key,
+            episode_started_at=requested.episode_started_at,
+            kind=InsightAlertInvestigation.Kind.AGENT,
+            evaluated={
+                "value": group.value,
+                **{key: group.source_detail[key] for key in _EVALUATED_KEYS if key in group.source_detail},
+            },
+        )
+        if claim.claimed:
+            starts.append(
+                InvestigationStart(
+                    team_id=team_id,
+                    alert_id=str(alert.id),
+                    investigation_id=str(claim.investigation_id),
+                    configuration_id=str(outcome.configuration_id),
+                    evaluation_key=outcome.evaluation_key,
+                    user_id=alert.created_by_id,
+                )
+            )
+            continue
+        if outcome.held_delivery is None:
+            continue
+        previous_verdict = previous_verdict_in_episode(
+            team_id, outcome.configuration_id, requested.episode_started_at, outcome.evaluation_key
+        )
+        request = release_held(
+            HeldDeliveryAddress(
+                team_id=team_id, configuration_id=str(outcome.configuration_id), evaluation_key=outcome.evaluation_key
+            ),
+            suppress=carried_verdict_suppresses(alert, InvestigationDecision(previous_verdict=previous_verdict)),
+            released_by=HoldReleasedBy.CLAIM_REFUSED,
+        )
+        if request is not None:
+            deliveries.append(request)
+    return InvestigationClaims(starts=tuple(starts), deliveries=tuple(deliveries))

products/alerts/backend/platform_investigation_detail.py

@@ -0,0 +1,65 @@
+"""The investigation an insight firing asks for in its `source_detail`, claimed after the record.
+
+A check only decides that its firing gets an investigation, because its evaluation can be retried.
+The evaluation workflow claims it once the firing is recorded, and starts the agent. This module
+has no Django imports, so the workflow can call it.
+"""
+
+from datetime import datetime
+
+from posthog.dataclasses import frozen
+
+from products.alerts_platform.backend.facade.contracts import AlertDeliveryRequest, PlatformAlertOutcome
+
+# The legacy alert whose agent investigates the firing. Recorded with the firing, so history shows
+# which checks asked for an investigation.
+INVESTIGATION_REQUESTED_KEY = "investigation_alert_id"
+
+
+@frozen
+class InvestigationStart:
+    """An agent run the claim granted, as the workflow starts it."""
+
+    team_id: int
+    alert_id: str
+    investigation_id: str
+    configuration_id: str
+    evaluation_key: str
+    user_id: int | None = None
+
+    @property
+    def workflow_id(self) -> str:
+        return f"anomaly-investigation-{self.configuration_id}:{self.evaluation_key}"
+
+    def workflow_inputs(self) -> dict[str, object]:
+        """The agent's `AnomalyInvestigationWorkflowInputs`, as JSON. The agent's module imports Django,
+        which the workflow sandbox refuses, so the inputs cross by name."""
+        return {
+            "team_id": self.team_id,
+            "alert_id": self.alert_id,
+            "investigation_id": self.investigation_id,
+            "user_id": self.user_id,
+        }
+
+
+@frozen
+class InvestigationClaims:
+    starts: tuple[InvestigationStart, ...] = ()
+    # Held firings whose claim was refused, released by the firing's last verdict and due to start.
+    deliveries: tuple[AlertDeliveryRequest, ...] = ()
+
+
+@frozen
+class RequestedInvestigation:
+    alert_id: str
+    episode_started_at: datetime | None
+
+
+def requested_investigation(outcome: PlatformAlertOutcome) -> RequestedInvestigation | None:
+    """The investigation an outcome asks for, or None. Only an ungrouped firing asks for one."""
+    for group in outcome.groups:
+        alert_id = group.source_detail.get(INVESTIGATION_REQUESTED_KEY)
+        if alert_id is not None:
+            started_at = group.firing_episode.started_at if group.firing_episode else None
+            return RequestedInvestigation(alert_id=str(alert_id), episode_started_at=started_at)
+    return None

products/alerts/backend/platform_source_cycle.py

@@ -59,6 +59,7 @@
 from products.alerts.backend.models.alert import AlertConfiguration
 from products.alerts.backend.platform_alert_wording import insight_alert_url
 from products.alerts.backend.platform_chart_detail import CHART_PENDING_KEY, RenderedChart
+from products.alerts.backend.platform_investigation import with_investigation
 from products.alerts_platform.backend.facade.api import due_check_ids, due_checks, refresh_settings, slot_of
 from products.alerts_platform.backend.facade.contracts import (
     SOURCE_CONDITION_KEY,
@@ -495,10 +496,17 @@
             raise
         if outcome is None:
             return None
-        return _evaluation(checks[0], outcome)
+        evaluation = _evaluation(checks[0], outcome)
     finally:
         if not retrying:
             release_evaluation_slot(configuration_id, held_until=held_until, key=INFLIGHT_KEY)
+    # After the slot is free, because the investigation's reads need no ClickHouse slot.
+    try:
+        return with_investigation(checks[0], evaluation)
+    except Exception:
+        # The firing still goes out, uninvestigated, rather than the check being lost as errored.
+        logger.exception("Failed to decide a platform insight firing's investigation", check_id=configuration_id)
+        return evaluation
 
 
 def fail_insight_check(
@@ -677,7 +685,7 @@
         return None
 
 
 def _decide(
     check: PlatformAlertCheckInput,
     now: datetime,
     *,
@@ -891,7 +899,7 @@
 GROUPED_INSIGHT_ALERT_POLICY: Final = replace(INSIGHT_ALERT_POLICY, mute_gates_notification_only=True)
 
 
 def _decide_grouped(
     check: PlatformAlertCheckInput,
     snapshot: AlertSnapshot,
     alert: AlertConfiguration,