From 96df83e5bc57162e9c61a7421d56a4583d7cbf69 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 16:29:09 -0800 Subject: [PATCH] feat(alerts): word platform insight firings with their investigation The insight describer, Slack renderer and legacy-event emitter read the evaluation's InsightAlertInvestigation row and the platform's delivery hold, so a platform-owned firing states the verdict, summary, notebook link or safety-net line that the insight stack sends. The verdict-change follow-up is worded for the verdict_changed kind. Legacy's verdict and safety-net lines move to one shared module so both stacks send the same text. Co-Authored-By: Claude Opus 5.5 --- .../alerts/investigation_notifications.py | 20 +- posthog/tasks/alerts/investigation_wording.py | 81 ++++++ .../ai/anomaly_investigation/workflow.py | 71 +---- .../backend/platform_alert_legacy_events.py | 22 +- .../alerts/backend/platform_alert_slack.py | 31 ++- .../alerts/backend/platform_alert_wording.py | 25 +- .../alerts/backend/platform_investigation.py | 13 + .../backend/platform_investigation_message.py | 130 +++++++++ .../test_platform_investigation_message.py | 246 ++++++++++++++++++ .../alerts_platform/backend/facade/api.py | 6 + .../backend/facade/contracts.py | 5 +- .../alerts_platform/backend/facade/testing.py | 20 ++ .../backend/logic/delivery_holds.py | 15 ++ .../backend/logic/platform_alert_events.py | 1 + 14 files changed, 593 insertions(+), 93 deletions(-) create mode 100644 posthog/tasks/alerts/investigation_wording.py create mode 100644 products/alerts/backend/platform_investigation_message.py create mode 100644 products/alerts/backend/test/test_platform_investigation_message.py diff --git a/posthog/tasks/alerts/investigation_notifications.py b/posthog/tasks/alerts/investigation_notifications.py index f4154e796aaf..9a63c82284e3 100644 --- a/posthog/tasks/alerts/investigation_notifications.py +++ b/posthog/tasks/alerts/investigation_notifications.py @@ -26,6 +26,7 @@ from posthog.schema import AlertState +from posthog.tasks.alerts.investigation_wording import safety_net_breaches from posthog.tasks.alerts.utils import ( dispatch_alert_notification, prepare_alert_insight_chart_url, @@ -109,7 +110,9 @@ def run_investigation_notification_safety_net() -> int: locked = AlertCheck.objects.select_for_update().get(id=check.id) if locked.notification_sent_at is not None or locked.notification_suppressed_by_agent: continue - breaches = _fallback_breach_descriptions(locked) + breaches = safety_net_breaches( + triggered_dates=locked.triggered_dates or [], value=locked.calculated_value + ) deliveries = dispatch_alert_notification( alert, locked, breaches, extra_properties=extra_properties, render_chart=False ) @@ -135,18 +138,3 @@ def run_investigation_notification_safety_net() -> int: notified += 1 return notified - - -def _fallback_breach_descriptions(alert_check: AlertCheck) -> list[str]: - """Minimal breach description for a safety-net dispatch (no verdict yet).""" - triggered_dates = alert_check.triggered_dates or [] - if triggered_dates: - if len(triggered_dates) == 1: - head = f"Anomaly detected on {triggered_dates[0]}." - else: - head = f"Anomaly detected from {triggered_dates[0]} to {triggered_dates[-1]}." - elif alert_check.calculated_value is not None: - head = f"Calculated value at fire: {alert_check.calculated_value}." - else: - head = "Anomaly detected." - return [head, "(Investigation agent did not complete before the notification grace period expired.)"] diff --git a/posthog/tasks/alerts/investigation_wording.py b/posthog/tasks/alerts/investigation_wording.py new file mode 100644 index 000000000000..cbc5fad061d2 --- /dev/null +++ b/posthog/tasks/alerts/investigation_wording.py @@ -0,0 +1,81 @@ +"""The lines an investigated insight alert notification states. + +The insight stack and the alerts platform both send these, so one definition keeps the two word +for word the same. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Final + +from posthog.dataclasses import frozen +from posthog.tasks.alerts.utils import InvestigationSummaryLine, investigation_summary_teaser +from posthog.utils import absolute_uri + +VERDICT_LABELS: Final = { + "true_positive": "True positive", + "false_positive": "False positive", + "inconclusive": "Inconclusive", +} + +SAFETY_NET_LINE: Final = "(Investigation agent did not complete before the notification grace period expired.)" + + +@frozen +class InvestigationBreaches: + lines: list[str] + summary: InvestigationSummaryLine | None + + +def notebook_url(notebook_short_id: str) -> str: + return absolute_uri(f"/notebooks/{notebook_short_id}") + + +def _anomaly_head(triggered_dates: Sequence[str], value: float | None) -> str: + if triggered_dates: + if len(triggered_dates) == 1: + return f"Anomaly detected on {triggered_dates[0]}." + return f"Anomaly detected from {triggered_dates[0]} to {triggered_dates[-1]}." + if value is not None: + return f"Calculated value at fire: {value}." + return "Anomaly detected." + + +def verdict_breaches( + *, + triggered_dates: Sequence[str], + value: float | None, + verdict: str | None, + previous_verdict: str | None, + summary: str, + notebook_short_id: str | None, +) -> InvestigationBreaches: + """The breach lines of a notification sent once the agent reached its verdict. + + The summary line holds a plain-text teaser. Email renders the full Markdown in its place. + """ + lines: list[str] = [] + if previous_verdict and verdict and verdict != previous_verdict: + lines.append( + f"Investigation verdict changed from {VERDICT_LABELS.get(previous_verdict, previous_verdict)} " + f"to {VERDICT_LABELS.get(verdict, verdict)} while this alert keeps firing." + ) + lines.append(_anomaly_head(triggered_dates, value)) + + verdict_label = VERDICT_LABELS.get(verdict or "", "") + if verdict_label: + lines.append(f"Investigation verdict: {verdict_label}.") + summary_line = None + teaser = investigation_summary_teaser(summary) + if teaser: + summary_line = InvestigationSummaryLine(index=len(lines), markdown=summary) + lines.append(teaser) + if notebook_short_id: + lines.append(f"See {notebook_url(notebook_short_id)} for the full investigation.") + return InvestigationBreaches(lines=lines, summary=summary_line) + + +def safety_net_breaches(*, triggered_dates: Sequence[str], value: float | None) -> list[str]: + """Minimal breach description for a notification the safety net sent without a verdict.""" + return [_anomaly_head(triggered_dates, value), SAFETY_NET_LINE] diff --git a/posthog/temporal/ai/anomaly_investigation/workflow.py b/posthog/temporal/ai/anomaly_investigation/workflow.py index 96f4cb2633a4..159e25c78638 100644 --- a/posthog/temporal/ai/anomaly_investigation/workflow.py +++ b/posthog/temporal/ai/anomaly_investigation/workflow.py @@ -20,16 +20,14 @@ from temporalio import activity, workflow from temporalio.common import RetryPolicy -from posthog.dataclasses import frozen from posthog.models import Team, User from posthog.tasks.alerts.charts import png_to_b64, render_series_chart +from posthog.tasks.alerts.investigation_wording import InvestigationBreaches, notebook_url, verdict_breaches from posthog.tasks.alerts.metric_definition import describe_metric_definition from posthog.tasks.alerts.utils import ( - InvestigationSummaryLine, _inconclusive_is_suppressed, _should_suppress_notification, dispatch_alert_notification, - investigation_summary_teaser, prepare_alert_insight_chart_url, record_alert_delivery, ) @@ -72,12 +70,6 @@ ANOMALY_INVESTIGATION_ACTIVITY_HEARTBEAT_TIMEOUT = 5 * 60 # 5 minutes ANOMALY_INVESTIGATION_ACTIVITY_MAX_ATTEMPTS = 2 -_VERDICT_LABELS = { - "true_positive": "True positive", - "false_positive": "False positive", - "inconclusive": "Inconclusive", -} - # Marker on the check's delivery receipts: the follow-up for a changed verdict is sent once. _VERDICT_CHANGE_FOLLOWUP_KEY = "investigation_verdict_change" @@ -333,7 +325,7 @@ def _build_investigation_signal_extra( ) -> dict: """The `extra` payload for an anomaly-investigation signal — a plain dict validated against `AnomalyInvestigationSignalExtra` inside `emit_signal`.""" - notebook_url = absolute_uri(f"/notebooks/{notebook_short_id}") if notebook_short_id else absolute_uri("/alerts") + url = notebook_url(notebook_short_id) if notebook_short_id else absolute_uri("/alerts") extra: dict = { "alert_id": str(alert.id), "alert_name": alert.name or "Unnamed alert", @@ -341,7 +333,7 @@ def _build_investigation_signal_extra( "insight_id": str(insight.id), "detector_type": detector_type, "verdict": report.verdict, - "url": notebook_url, + "url": url, } # Optional fields — omit when absent so the schema's None defaults apply. if insight.name: @@ -513,7 +505,7 @@ def _deliver_investigation_outcome( # block falls back to a divider when absent). extra_properties: dict[str, str] = {} if notebook_short_id: - extra_properties["investigation_notebook_url"] = absolute_uri(f"/notebooks/{notebook_short_id}") + extra_properties["investigation_notebook_url"] = notebook_url(notebook_short_id) if insight_chart_url: extra_properties["insight_chart_url"] = insight_chart_url try: @@ -577,9 +569,7 @@ def _dispatch_verdict_change_followup( summary=summary, notebook_short_id=notebook_short_id, ) - extra_properties = ( - {"investigation_notebook_url": absolute_uri(f"/notebooks/{notebook_short_id}")} if notebook_short_id else None - ) + extra_properties = {"investigation_notebook_url": notebook_url(notebook_short_id)} if notebook_short_id else None try: # The savepoint keeps a database failure inside the dispatch from poisoning the # caller's transaction once the handler below swallows it. @@ -634,12 +624,6 @@ def _dispatch_verdict_change_followup( ) -@frozen -class _BreachDescriptions: - lines: list[str] - summary: InvestigationSummaryLine | None - - def _build_breach_descriptions( *, alert_check, @@ -647,42 +631,15 @@ def _build_breach_descriptions( previous_verdict: str | None, summary: str, notebook_short_id: str | None, -) -> _BreachDescriptions: - """Compose the strings that populate the `match_descriptions` list in the - existing alert email template. Keeps the current template working while - giving gated notifications richer body content. - - The summary line holds a plain-text teaser. Email renders the full Markdown in its place. - """ - lines: list[str] = [] - if previous_verdict and verdict and verdict != previous_verdict: - lines.append( - f"Investigation verdict changed from {_VERDICT_LABELS.get(previous_verdict, previous_verdict)} " - f"to {_VERDICT_LABELS.get(verdict, verdict)} while this alert keeps firing." - ) - triggered_dates = alert_check.triggered_dates or [] - if triggered_dates: - if len(triggered_dates) == 1: - lines.append(f"Anomaly detected on {triggered_dates[0]}.") - else: - lines.append(f"Anomaly detected from {triggered_dates[0]} to {triggered_dates[-1]}.") - elif alert_check.calculated_value is not None: - lines.append(f"Calculated value at fire: {alert_check.calculated_value}.") - else: - lines.append("Anomaly detected.") - - verdict_label = _VERDICT_LABELS.get(verdict or "", "") - if verdict_label: - lines.append(f"Investigation verdict: {verdict_label}.") - summary_line = None - teaser = investigation_summary_teaser(summary) - if teaser: - summary_line = InvestigationSummaryLine(index=len(lines), markdown=summary) - lines.append(teaser) - if notebook_short_id: - notebook_url = absolute_uri(f"/notebooks/{notebook_short_id}") - lines.append(f"See {notebook_url} for the full investigation.") - return _BreachDescriptions(lines=lines, summary=summary_line) +) -> InvestigationBreaches: + return verdict_breaches( + triggered_dates=alert_check.triggered_dates or [], + value=alert_check.calculated_value, + verdict=verdict, + previous_verdict=previous_verdict, + summary=summary, + notebook_short_id=notebook_short_id, + ) async def _update_status(alert_check, status: str) -> None: diff --git a/products/alerts/backend/platform_alert_legacy_events.py b/products/alerts/backend/platform_alert_legacy_events.py index f4ea948604b4..48835f14c7a1 100644 --- a/products/alerts/backend/platform_alert_legacy_events.py +++ b/products/alerts/backend/platform_alert_legacy_events.py @@ -23,6 +23,12 @@ from products.alerts.backend.logic.destinations import emit_alert_internal_events from products.alerts.backend.models.alert import AlertConfiguration from products.alerts.backend.platform_alert_slack import insight_transition_properties +from products.alerts.backend.platform_alert_wording import announces_firing +from products.alerts.backend.platform_investigation_message import ( + FiringInvestigation, + firing_breaches, + firing_investigation, +) from products.alerts_platform.backend.facade.contracts import ( AlertDeliveryRequest, AlertEventKind, @@ -45,7 +51,8 @@ def emit_insight_legacy_events( firings = [ emission.transition for emission in emissions - if emission.sends_message and emission.transition.kind == AlertEventKind.FIRING + # A verdict-change follow-up goes out as a firing event, as the insight stack sends it. + if emission.sends_message and announces_firing(emission.transition) ] if not firings: return @@ -84,14 +91,16 @@ def insight_chart_image(*, team_id: int, transition: AnnouncedTransition) -> Mes return MessageImage(url=url, alt_text="Insight chart") if url else None -def _formatting_properties(transition: AnnouncedTransition) -> dict[str, Any]: +def _formatting_properties( + transition: AnnouncedTransition, investigation: FiringInvestigation | None +) -> dict[str, Any]: """The Slack card fields, or none of them. These enrich the event; the destinations it reaches do not depend on them. A stored detail the formatter cannot read must cost the firing its card, never its delivery. """ try: - return insight_transition_properties(transition) + return insight_transition_properties(transition, investigation) except Exception: logger.exception("alerts_insight_legacy_event_formatting_failed") return {} @@ -99,16 +108,19 @@ def _formatting_properties(transition: AnnouncedTransition) -> dict[str, Any]: def _firing_event(alert: AlertConfiguration, transition: AnnouncedTransition) -> AlertInternalEvent: detail = transition.source_detail + investigation = firing_investigation(alert.team_id, transition) properties: dict[str, Any] = { # From the transition, not the alert row: the platform never writes `state` or # `last_checked_at`, so a platform-owned alert past its shadow window would report the # frozen `not_firing` default on the very event that announces a firing. **alert.internal_event_properties(state=AlertState.FIRING.value, last_checked_at=transition.occurred_at), # Joined with newlines so each breach renders on its own line, as the insight stack sends it. - "breaches": "\n".join(detail.get("breaches") or []), + "breaches": "\n".join(firing_breaches(transition, investigation)), **detector_verdict_event_fields(detail.get("triggered_metadata")), - **_formatting_properties(transition), + **_formatting_properties(transition, investigation), } + if investigation is not None and investigation.notebook_url: + properties["investigation_notebook_url"] = investigation.notebook_url if chart_url := insight_chart_url_of(alert.team_id, detail): properties["insight_chart_url"] = chart_url # The occurrence is in the seed too, so two firings recorded without an evaluation key keep distinct ids. diff --git a/products/alerts/backend/platform_alert_slack.py b/products/alerts/backend/platform_alert_slack.py index 5de654765896..c2c98339c9c9 100644 --- a/products/alerts/backend/platform_alert_slack.py +++ b/products/alerts/backend/platform_alert_slack.py @@ -9,8 +9,13 @@ from products.alerts.backend.logic.insight_alert_destinations import _slack_attachments, _slack_blocks from products.alerts.backend.logic.insight_alert_message import insight_alert_slack_properties +from products.alerts.backend.platform_alert_wording import announces_firing +from products.alerts.backend.platform_investigation_message import ( + FiringInvestigation, + firing_breaches, + firing_investigation, +) from products.alerts_platform.backend.facade.contracts import ( - AlertEventKind, AnnouncedTransition, MessageImage, MessageLink, @@ -33,9 +38,11 @@ def _zone(name: str) -> ZoneInfo: return ZoneInfo("UTC") -def insight_transition_properties(transition: AnnouncedTransition) -> dict[str, Any]: +def insight_transition_properties( + transition: AnnouncedTransition, investigation: FiringInvestigation | None +) -> dict[str, Any]: detail = transition.source_detail - breaches = detail.get("breaches") or [] + breaches = firing_breaches(transition, investigation) # Rows come from a stored detail, so their shape is checked rather than assumed. The formatter # reads `observed` directly, so a row without it would raise there. rows = [row for row in (detail.get("breach_details") or []) if isinstance(row, dict)] @@ -58,11 +65,9 @@ def insight_transition_properties(transition: AnnouncedTransition) -> dict[str, metric_name=detail.get("metric_name") or "Insight", is_anomaly=bool(detail.get("is_anomaly")), tags=detail.get("tags") or [], - # Native delivery sends as the check finishes, before any anomaly investigation runs, so it - # has no summary, verdict or next step to show. The HogFunction path sends later and does. - summary=None, - verdict=None, - investigation_status=None, + summary=investigation.chat_summary if investigation else None, + verdict=investigation.verdict if investigation else None, + investigation_status=investigation.status if investigation else None, detected_at=detected_at, allowed_root=absolute_uri("/"), ) @@ -77,12 +82,14 @@ def render_insight_slack_message( project: MessageLink | None, image: MessageImage | None, ) -> SlackMessageContent | None: - if transition.kind != AlertEventKind.FIRING: + if not announces_firing(transition): return None alert_id = transition.source_detail.get("alert_id") if not alert_id: return None - properties = insight_transition_properties(transition) + investigation = firing_investigation(team_id, transition) + properties = insight_transition_properties(transition, investigation) + notebook_url = investigation.notebook_url if investigation else None kind = "Anomaly alert" if properties["slack_alert_type"] == "Anomaly detected" else "Insight alert" project_url = project.url if project else absolute_uri(f"/project/{team_id}") project_name = project.label if project else transition.source_detail.get("team_name") or "" @@ -92,8 +99,8 @@ def render_insight_slack_message( blocks = _slack_blocks( chart={"type": "image", "image_url": image.url, "alt_text": image.alt_text} if image else {"type": "divider"}, alert_id=str(alert_id), - primary_url=alert_url, - primary_label="🔎 View alert", + primary_url=notebook_url or alert_url, + primary_label="🔎 View investigation" if notebook_url else "🔎 View alert", insight_url=absolute_uri(f"/project/{team_id}/insights/{insight_short_id}") if insight_short_id else None, ) attachments = _slack_attachments( diff --git a/products/alerts/backend/platform_alert_wording.py b/products/alerts/backend/platform_alert_wording.py index 1da56ed23278..47a12e8b12cc 100644 --- a/products/alerts/backend/platform_alert_wording.py +++ b/products/alerts/backend/platform_alert_wording.py @@ -9,6 +9,8 @@ from __future__ import annotations +from typing import Final + from posthog.utils import absolute_uri from products.alerts.backend.logic.insight_alert_destinations import snooze_select @@ -21,12 +23,21 @@ SourceDescription, ) +# The platform kind of the follow-up sent when an investigation's verdict contradicts the previous +# verdict of the same firing. +VERDICT_CHANGED_KIND: Final = "verdict_changed" + + +def announces_firing(transition: AnnouncedTransition) -> bool: + """Whether a transition is worded as a firing: the firing itself, or a follow-up to its verdict.""" + return transition.kind in (AlertEventKind.FIRING, VERDICT_CHANGED_KIND) + def describe_insight_transition(*, project_id: int, transition: AnnouncedTransition) -> SourceDescription: detail = transition.source_detail # A failure keeps the platform's own wording, so only a firing states breaches or offers a snooze. - firing = transition.kind == AlertEventKind.FIRING - breaches = (detail.get("breaches") or []) if firing else [] + firing = announces_firing(transition) + breaches = _firing_breaches(project_id, transition) if firing else [] alert_id = detail.get("alert_id") if firing else None insight_name = detail.get("insight_name") insight_short_id = detail.get("insight_short_id") @@ -41,6 +52,16 @@ def describe_insight_transition(*, project_id: int, transition: AnnouncedTransit ) +def _firing_breaches(project_id: int, transition: AnnouncedTransition) -> list[str]: + from products.alerts.backend.platform_investigation_message import ( # noqa: PLC0415 - keeps models off the startup path + firing_breaches, + firing_investigation, + ) + + # The platform passes the team's id as the project id. + return firing_breaches(transition, firing_investigation(project_id, transition)) + + def _breach_label(transition: AnnouncedTransition) -> str: condition = transition.source_config.get(SOURCE_CONDITION_KEY) return "Anomaly detected" if isinstance(condition, dict) and condition.get("detector") else "Threshold breached" diff --git a/products/alerts/backend/platform_investigation.py b/products/alerts/backend/platform_investigation.py index af37f5c41844..9732f45cd497 100644 --- a/products/alerts/backend/platform_investigation.py +++ b/products/alerts/backend/platform_investigation.py @@ -27,6 +27,9 @@ from products.alerts.backend.models.investigation import InsightAlertInvestigation _CLAIM_LOCK_SQL: Final = "SELECT pg_advisory_xact_lock(hashtext(%s))" +# 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" @frozen @@ -36,6 +39,16 @@ class InvestigationClaim: claimed: bool +def verdict_change_evaluation_key(evaluation_key: str) -> str: + """The key the verdict-change follow-up of an investigated firing is recorded under.""" + return f"{evaluation_key}{_VERDICT_CHANGE_KEY_SUFFIX}" + + +def investigated_evaluation_key(verdict_change_key: str) -> str: + """The key of the firing a verdict-change follow-up follows up.""" + return verdict_change_key.removesuffix(_VERDICT_CHANGE_KEY_SUFFIX) + + def decide_investigation( alert: AlertConfiguration, *, diff --git a/products/alerts/backend/platform_investigation_message.py b/products/alerts/backend/platform_investigation_message.py new file mode 100644 index 000000000000..f4ef2bf7b586 --- /dev/null +++ b/products/alerts/backend/platform_investigation_message.py @@ -0,0 +1,130 @@ +"""What a platform message about an investigated insight firing says about its investigation. + +The insight stack words a firing from its `AlertCheck`. The platform records no check, so the +describer, the Slack renderer and the legacy-event emitter read the evaluation's +`InsightAlertInvestigation` row, and the platform's hold on its delivery, as the message goes out. + +The verdict arrives after the firing is recorded, so it cannot sit on the history row the message +is built from. +""" + +from __future__ import annotations + +import structlog + +from posthog.dataclasses import frozen +from posthog.tasks.alerts.investigation_wording import notebook_url, safety_net_breaches, verdict_breaches +from posthog.tasks.alerts.utils import investigation_summary_teaser + +from products.alerts.backend.models.alert import InvestigationStatus +from products.alerts.backend.models.investigation import InsightAlertInvestigation +from products.alerts.backend.platform_alert_wording import VERDICT_CHANGED_KIND, announces_firing +from products.alerts.backend.platform_investigation import investigated_evaluation_key +from products.alerts_platform.backend.facade.api import hold_released_by +from products.alerts_platform.backend.facade.contracts import AnnouncedTransition, HeldDeliveryAddress +from products.alerts_platform.backend.facade.enums import HoldReleasedBy + +logger = structlog.get_logger(__name__) + + +@frozen +class FiringInvestigation: + """An investigation of a firing, in the fields a message states.""" + + status: str + verdict: str | None + chat_summary: str | None + # The lines that replace the firing's own breaches. None keeps the firing's breaches. + breaches: tuple[str, ...] | None + # Set only when the message carries the verdict, as the insight stack links the notebook then. + notebook_url: str | None + + +def firing_breaches(transition: AnnouncedTransition, investigation: FiringInvestigation | None) -> list[str]: + """The breach lines a firing states: its investigation's, or the ones its check recorded.""" + if investigation is not None and investigation.breaches is not None: + return list(investigation.breaches) + return transition.source_detail.get("breaches") or [] + + +def firing_investigation(team_id: int, transition: AnnouncedTransition) -> FiringInvestigation | None: + """The investigation of the firing a transition announces, or None when it had none. + + Never raises: a failed read costs the message its investigation lines, never the message. + """ + if not announces_firing(transition) or not transition.configuration_id: + return None + try: + return _read(team_id, transition) + except Exception: + logger.exception("alerts_insight_investigation_read_failed", configuration_id=transition.configuration_id) + return None + + +def _read(team_id: int, transition: AnnouncedTransition) -> FiringInvestigation | None: + follow_up = transition.kind == VERDICT_CHANGED_KIND + evaluation_key = investigated_evaluation_key(transition.evaluation_key) if follow_up else transition.evaluation_key + rows = InsightAlertInvestigation.objects.for_team(team_id).filter(configuration_id=transition.configuration_id) + row = rows.filter(evaluation_key=evaluation_key).values_list( + "status", + "verdict", + "summary", + "notebook__short_id", + "evaluated", + "kind", + "episode_started_at", + "created_at", + ) + found = row.first() + if found is None: + return None + status, verdict, summary, notebook_short_id, evaluated, kind, episode_started_at, created_at = found + # A refused claim investigated nothing, so its firing reads as one the agent never saw. + if status == InvestigationStatus.SKIPPED: + return None + + released_by = ( + None + if follow_up + else hold_released_by( + HeldDeliveryAddress( + team_id=team_id, configuration_id=transition.configuration_id, evaluation_key=evaluation_key + ) + ) + ) + triggered_dates = [str(date) for date in evaluated.get("triggered_dates") or []] + value = evaluated.get("value") + breaches: tuple[str, ...] | None = None + link: str | None = None + if follow_up or released_by == HoldReleasedBy.VERDICT: + previous_verdict = ( + rows.filter(episode_started_at=episode_started_at, created_at__lt=created_at, verdict__isnull=False) + .order_by("-created_at") + .values_list("verdict", flat=True) + .first() + if verdict + else None + ) + breaches = tuple( + verdict_breaches( + triggered_dates=triggered_dates, + value=value, + verdict=verdict, + previous_verdict=previous_verdict, + summary=summary or "", + notebook_short_id=notebook_short_id, + ).lines + ) + link = notebook_url(notebook_short_id) if notebook_short_id else None + elif released_by == HoldReleasedBy.SAFETY_NET: + breaches = tuple(safety_net_breaches(triggered_dates=triggered_dates, value=value)) + + return FiringInvestigation( + status=status, + verdict=verdict, + # The agent writes Markdown, which chat shows as is, so it gets the teaser. The metrics + # investigation writes plain text. + chat_summary=investigation_summary_teaser(summary) if kind == InsightAlertInvestigation.Kind.AGENT else summary, + breaches=breaches, + notebook_url=link, + ) diff --git a/products/alerts/backend/test/test_platform_investigation_message.py b/products/alerts/backend/test/test_platform_investigation_message.py new file mode 100644 index 000000000000..0540d95a50b0 --- /dev/null +++ b/products/alerts/backend/test/test_platform_investigation_message.py @@ -0,0 +1,246 @@ +from datetime import UTC, datetime, timedelta +from typing import Any, cast + +import time_machine +from posthog.test.base import APIBaseTest +from unittest.mock import patch + +from parameterized import parameterized + +from posthog.models.scoping import team_scope +from posthog.schema_enums import AlertState +from posthog.tasks.alerts.investigation_wording import SAFETY_NET_LINE +from posthog.utils import absolute_uri + +from products.alerts.backend.models.alert import InvestigationStatus +from products.alerts.backend.models.investigation import InsightAlertInvestigation +from products.alerts.backend.platform_alert_legacy_events import emit_insight_legacy_events +from products.alerts.backend.platform_alert_slack import render_insight_slack_message +from products.alerts.backend.platform_alert_wording import VERDICT_CHANGED_KIND, describe_insight_transition +from products.alerts.backend.platform_investigation import verdict_change_evaluation_key +from products.alerts.backend.test.insight_alerts import create_insight_alert +from products.alerts_platform.backend.facade import testing as platform_testing +from products.alerts_platform.backend.facade.contracts import ( + AlertDeliveryRequest, + AlertEventKind, + AnnouncedTransition, + EvaluationAnnouncement, + LegacyEventEmission, + MessageDetail, + SourceKind, +) +from products.alerts_platform.backend.facade.enums import HoldReleasedBy +from products.notebooks.backend.facade import api as notebooks + +EPISODE = datetime(2026, 10, 20, 8, 0, tzinfo=UTC) +FIRED_AT = datetime(2026, 10, 20, 9, 0, tzinfo=UTC) +KEY = "slot:2026-10-20T09:00:00+00:00" +OWN_BREACH = "Signups (50) is less than lower threshold (100)" +ANOMALY_HEAD = "Anomaly detected on 2026-10-20." +SUMMARY = "Signups fell after the **release**." +TEASER = "Signups fell after the release." + + +class TestInvestigatedFiringMessages(APIBaseTest): + def setUp(self) -> None: + super().setUp() + self.alert = create_insight_alert(self.team, name="Signups dropped", detector_config={"type": "zscore"}) + with team_scope(self.team.id): + self.configuration_id = platform_testing.create_configuration( + team_id=self.team.id, + name="alert", + source_kind=SourceKind.INSIGHT, + check_interval_minutes=60, + recurrence_unit="hour", + legacy_configuration_id=self.alert.id, + ).id + self.notebook = notebooks.create_notebook(self.team.id, title="Investigation", content=None) + self.notebook_url = absolute_uri(f"/notebooks/{self.notebook.short_id}") + + def _investigation( + self, key: str, *, status: str = InvestigationStatus.DONE, verdict: str | None = None, at: datetime = FIRED_AT + ) -> None: + with time_machine.travel(at, tick=False): + InsightAlertInvestigation.objects.for_team(self.team.id).create( + team=self.team, + alert_configuration=self.alert, + configuration_id=self.configuration_id, + evaluation_key=key, + episode_started_at=EPISODE, + kind=InsightAlertInvestigation.Kind.AGENT, + status=status, + verdict=verdict, + summary=SUMMARY if verdict else None, + notebook_id=self.notebook.id if verdict else None, + evaluated={"value": 50.0, "triggered_dates": ["2026-10-20"]}, + ) + + def _released_by(self, released_by: HoldReleasedBy) -> None: + platform_testing.release_held_delivery( + team_id=self.team.id, + configuration_id=self.configuration_id, + evaluation_key=KEY, + released_by=released_by, + at=FIRED_AT, + ) + + def _transition(self, kind: str = AlertEventKind.FIRING, key: str = KEY) -> AnnouncedTransition: + return AnnouncedTransition( + grouping_key="", + kind=cast(AlertEventKind, kind), + episode_started_at=EPISODE, + value=50.0, + labels={}, + condition={}, + source_config={}, + error_message=None, + occurred_at=FIRED_AT, + evaluation_key=key, + source_detail={"breaches": [OWN_BREACH], "alert_id": str(self.alert.id), "is_anomaly": True}, + configuration_id=str(self.configuration_id), + ) + + def _event_properties(self, transition: AnnouncedTransition) -> dict[str, Any]: + request = AlertDeliveryRequest( + source=SourceKind.INSIGHT, + team_id=self.team.id, + configuration_id=str(self.configuration_id), + evaluation_key=transition.evaluation_key, + destination_alert_id=str(self.alert.id), + event_ids_by_kind={}, + ) + announcement = EvaluationAnnouncement( + configuration_id=str(self.configuration_id), + source=SourceKind.INSIGHT, + alert_name=self.alert.name, + consecutive_failures=0, + transitions=(transition,), + ) + emission = LegacyEventEmission(transition=transition, sends_message=True, incident_action=None) + with patch("products.alerts.backend.platform_alert_legacy_events.emit_alert_internal_events") as emit: + emit_insight_legacy_events(request=request, announcement=announcement, emissions=[emission]) + (event,) = emit.call_args.kwargs["events"] + return event.properties + + @staticmethod + def _investigation_label(properties: dict[str, Any]) -> str | None: + for section in properties["slack_alert_details"]["elements"]: + texts = [element.get("text") for element in section.get("elements", [])] + if texts[:1] == ["Investigation: "]: + return texts[1] + return None + + @parameterized.expand( + [ + ("the_agent_is_off", None, None, None, [OWN_BREACH], False, None), + # A firing of an alert that does not gate goes out before the agent answers. + ("an_ungated_firing", InvestigationStatus.PENDING, None, None, [OWN_BREACH], False, "In progress"), + ("a_refused_claim", InvestigationStatus.SKIPPED, None, None, [OWN_BREACH], False, None), + ( + "a_verdict_released_the_firing", + InvestigationStatus.DONE, + "true_positive", + HoldReleasedBy.VERDICT, + [ANOMALY_HEAD, "Investigation verdict: True positive.", TEASER], + True, + "Unusual change confirmed", + ), + ( + "the_safety_net_sent_it_without_a_verdict", + InvestigationStatus.RUNNING, + None, + HoldReleasedBy.SAFETY_NET, + [ANOMALY_HEAD, SAFETY_NET_LINE], + False, + "In progress", + ), + ( + "the_claim_was_refused_after_the_hold", + InvestigationStatus.PENDING, + None, + HoldReleasedBy.CLAIM_REFUSED, + [OWN_BREACH], + False, + "In progress", + ), + ] + ) + def test_a_firing_states_what_its_investigation_found( + self, + _name: str, + status: str | None, + verdict: str | None, + released_by: HoldReleasedBy | None, + breaches: list[str], + links_notebook: bool, + label: str | None, + ) -> None: + if status is not None: + self._investigation(KEY, status=status, verdict=verdict) + if released_by is not None: + self._released_by(released_by) + expected_breaches = [ + *breaches, + *([f"See {self.notebook_url} for the full investigation."] if links_notebook else []), + ] + + properties = self._event_properties(self._transition()) + + assert properties["breaches"] == "\n".join(expected_breaches) + assert properties.get("investigation_notebook_url") == (self.notebook_url if links_notebook else None) + assert self._investigation_label(properties) == label + description = describe_insight_transition(project_id=self.team.id, transition=self._transition()) + assert description.details == (MessageDetail(label="Threshold breached", value="\n".join(expected_breaches)),) + + def test_a_changed_verdict_follows_up_with_the_change(self) -> None: + self._investigation("slot:2026-10-20T08:00:00+00:00", verdict="true_positive", at=EPISODE) + self._investigation(KEY, verdict="false_positive") + follow_up = self._transition(VERDICT_CHANGED_KIND, verdict_change_evaluation_key(KEY)) + expected = [ + "Investigation verdict changed from True positive to False positive while this alert keeps firing.", + ANOMALY_HEAD, + "Investigation verdict: False positive.", + TEASER, + f"See {self.notebook_url} for the full investigation.", + ] + + properties = self._event_properties(follow_up) + + assert properties["breaches"] == "\n".join(expected) + assert properties["state"] == AlertState.FIRING.value + assert properties["investigation_notebook_url"] == self.notebook_url + slack = render_insight_slack_message( + team_id=self.team.id, + transition=follow_up, + alert_name=self.alert.name, + alert_url="https://example.com/alert", + project=None, + image=None, + ) + assert slack is not None + (actions,) = [block for block in slack.attachments[-1]["blocks"] if block["type"] == "actions"] + button = actions["elements"][0] + assert button == { + "url": self.notebook_url, + "text": {"text": "🔎 View investigation", "type": "plain_text"}, + "type": "button", + } + + def test_a_firing_of_an_earlier_episode_does_not_count_as_the_previous_verdict(self) -> None: + with time_machine.travel(EPISODE - timedelta(days=1), tick=False): + InsightAlertInvestigation.objects.for_team(self.team.id).create( + team=self.team, + alert_configuration=self.alert, + configuration_id=self.configuration_id, + evaluation_key="slot:yesterday", + episode_started_at=EPISODE - timedelta(days=1), + kind=InsightAlertInvestigation.Kind.AGENT, + status=InvestigationStatus.DONE, + verdict="true_positive", + ) + self._investigation(KEY, verdict="inconclusive") + self._released_by(HoldReleasedBy.VERDICT) + + properties = self._event_properties(self._transition()) + + assert properties["breaches"].splitlines()[0] == ANOMALY_HEAD diff --git a/products/alerts_platform/backend/facade/api.py b/products/alerts_platform/backend/facade/api.py index 85774ea367fa..4cc2e549fe69 100644 --- a/products/alerts_platform/backend/facade/api.py +++ b/products/alerts_platform/backend/facade/api.py @@ -84,6 +84,12 @@ def reschedule_hold(address: HeldDeliveryAddress, *, release_after: datetime) -> return delivery_holds.reschedule_hold(address, release_after=release_after) +def hold_released_by(address: HeldDeliveryAddress) -> HoldReleasedBy | None: + """Who released an evaluation's held delivery, so a source can word the message that the release sent. + None when the evaluation was never held or is still held.""" + return delivery_holds.hold_released_by(address) + + def slot_of(next_check_at: datetime | None, cutoff: datetime) -> str: """The minute a configuration is due for.""" return platform_lifecycle.slot_of(next_check_at, cutoff) diff --git a/products/alerts_platform/backend/facade/contracts.py b/products/alerts_platform/backend/facade/contracts.py index c35ee855e5fe..0a06726a10f0 100644 --- a/products/alerts_platform/backend/facade/contracts.py +++ b/products/alerts_platform/backend/facade/contracts.py @@ -773,6 +773,8 @@ class AnnouncedTransition: # The evaluation this transition came from, so a source can read back what it put in the key. evaluation_key: str = "" source_detail: dict[str, Any] = field(default_factory=dict) + # With `evaluation_key`, names the evaluation, so a source can find what it recorded about it later. + configuration_id: str = "" @frozen @@ -1407,7 +1409,8 @@ class SourceDescriber(Protocol): """How a source describes its transitions. A source registers one for native delivery. It runs once per destination on the send path, inside a delivery that a held thread or a - failure repeats. So it builds its answer from the transition alone and does no I/O. + failure repeats. So it builds its answer from the transition, plus at most a point read of + what the source recorded about this evaluation since, such as an investigation's verdict. """ def __call__(self, *, project_id: int, transition: AnnouncedTransition) -> SourceDescription: ... diff --git a/products/alerts_platform/backend/facade/testing.py b/products/alerts_platform/backend/facade/testing.py index 2677da4fab6e..d2d565dd84a9 100644 --- a/products/alerts_platform/backend/facade/testing.py +++ b/products/alerts_platform/backend/facade/testing.py @@ -22,11 +22,13 @@ SourceCorrespondence, SourceKind, ) +from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy from products.alerts_platform.backend.logic.platform_alert_events import PlatformAlertEventRow, insert_events from products.alerts_platform.backend.logic.platform_reads import instance_view from products.alerts_platform.backend.models import ( PlatformAlert, PlatformAlertConfiguration, + PlatformAlertDeliveryHold, PlatformAlertDeliveryPath, PlatformAlertThread, ) @@ -72,6 +74,24 @@ def set_due_at(configuration_id: UUID, due_at: datetime) -> None: PlatformAlertConfiguration.objects.filter(id=configuration_id).update(next_check_at=due_at) +def release_held_delivery( + *, team_id: int, configuration_id: UUID, evaluation_key: str, released_by: HoldReleasedBy, at: datetime +) -> None: + """An evaluation whose delivery was held and then released with a send, as a message reader finds it.""" + PlatformAlertDeliveryHold.objects.for_team(team_id).create( + team_id=team_id, + configuration_id=configuration_id, + evaluation_key=evaluation_key, + reason="test", + request={}, + held_at=at, + release_after=at, + released_at=at, + release_action=HoldRelease.SEND, + released_by=released_by, + ) + + def deliver(request: AlertDeliveryRequest) -> int: """Send a recorded evaluation's delivery as the delivery activity does. Returns how many destinations took it.""" return deliver_evaluation(request).sent diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py index c8d1f1e34727..23cff87d7962 100644 --- a/products/alerts_platform/backend/logic/delivery_holds.py +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -149,6 +149,21 @@ def reschedule_hold(address: HeldDeliveryAddress, *, release_after: datetime) -> return updated > 0 +def hold_released_by(address: HeldDeliveryAddress) -> HoldReleasedBy | None: + """Who released an evaluation's hold. None when the evaluation was never held or is still held.""" + value = ( + PlatformAlertDeliveryHold.objects.for_team(address.team_id) + .filter( + configuration_id=address.configuration_id, + evaluation_key=address.evaluation_key, + released_at__isnull=False, + ) + .values_list("released_by", flat=True) + .first() + ) + return HoldReleasedBy(value) if value else None + + def blocks_delivery(team_id: int, configuration_id: str, evaluation_key: str) -> bool: """Whether an evaluation has a hold that is not released with a send. diff --git a/products/alerts_platform/backend/logic/platform_alert_events.py b/products/alerts_platform/backend/logic/platform_alert_events.py index 048e4b10a414..c9241002a788 100644 --- a/products/alerts_platform/backend/logic/platform_alert_events.py +++ b/products/alerts_platform/backend/logic/platform_alert_events.py @@ -244,6 +244,7 @@ def announcement( occurred_at=occurred_at, evaluation_key=evaluation_key, source_detail=_snapshot(source_detail), + configuration_id=configuration_id, ) for grouping_key, kind, episode_started_at, value, labels, condition_snapshot, source_config_snapshot, error_message, occurred_at, source_detail, _, _ in rows )