From 42370ea6758cc89f7748108b94b83acbf5011618 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 16:43:18 -0800 Subject: [PATCH 1/6] feat(alerts): let the investigation agent run on a platform evaluation The agent takes an investigation id instead of an alert check. It reads what the evaluation measured from the InsightAlertInvestigation row, records status, verdict, summary and notebook there, and releases the firing's held delivery by the verdict. The workflow starts the released delivery behind a patch. A verdict that contradicts the firing's previous one on a firing that already went out calls the platform's verdict-changed announcement, which is a stub until the platform has that kind. The signal contract accepts an investigation id and evaluation key in place of the alert check id, and Today keys an occurrence on either. Co-Authored-By: Claude Opus 5.5 --- posthog/temporal/ai/__init__.py | 7 +- .../ai/anomaly_investigation/__init__.py | 2 + .../ai/anomaly_investigation/evaluated.py | 48 +++ .../ai/anomaly_investigation/notebook.py | 11 +- .../ai/anomaly_investigation/workflow.py | 300 +++++++++++++----- .../ai/test_anomaly_investigation_notebook.py | 5 +- .../ai/test_anomaly_investigation_verdict.py | 102 +++++- .../tests/ai/test_module_integrity.py | 1 + .../alerts/backend/investigation_release.py | 182 +++++++++++ .../alerts/backend/platform_investigation.py | 13 +- .../tests/test_platform_investigation.py | 103 ++++++ .../alerts_platform/backend/facade/api.py | 16 + .../alerts_platform/backend/facade/testing.py | 35 ++ products/signals/backend/contracts.py | 4 +- .../signals/backend/test/test_contracts.py | 37 ++- .../signals/frontend/generated/api.schemas.ts | 4 +- products/today/backend/logic/impact.py | 3 +- services/mcp/src/api/generated.ts | 4 +- 18 files changed, 755 insertions(+), 122 deletions(-) create mode 100644 posthog/temporal/ai/anomaly_investigation/evaluated.py create mode 100644 products/alerts/backend/investigation_release.py diff --git a/posthog/temporal/ai/__init__.py b/posthog/temporal/ai/__init__.py index 292f02bbaaf7..754b62432e88 100644 --- a/posthog/temporal/ai/__init__.py +++ b/posthog/temporal/ai/__init__.py @@ -1,4 +1,8 @@ -from posthog.temporal.ai.anomaly_investigation import AnomalyInvestigationWorkflow, investigate_anomaly_activity +from posthog.temporal.ai.anomaly_investigation import ( + AnomalyInvestigationWorkflow, + investigate_anomaly_activity, + record_investigation_deliveries_started_activity, +) from posthog.temporal.ai.chat_agent import ( AssistantConversationRunnerWorkflow, ChatAgentWorkflow, @@ -45,6 +49,7 @@ process_research_agent_activity, summarize_llm_traces_activity, investigate_anomaly_activity, + record_investigation_deliveries_started_activity, *CHECKPOINT_COMPACTION_ACTIVITIES, ] diff --git a/posthog/temporal/ai/anomaly_investigation/__init__.py b/posthog/temporal/ai/anomaly_investigation/__init__.py index e5a48eca33c3..3f6e39c90c8e 100644 --- a/posthog/temporal/ai/anomaly_investigation/__init__.py +++ b/posthog/temporal/ai/anomaly_investigation/__init__.py @@ -2,10 +2,12 @@ AnomalyInvestigationWorkflow, AnomalyInvestigationWorkflowInputs, investigate_anomaly_activity, + record_investigation_deliveries_started_activity, ) __all__ = [ "AnomalyInvestigationWorkflow", "AnomalyInvestigationWorkflowInputs", "investigate_anomaly_activity", + "record_investigation_deliveries_started_activity", ] diff --git a/posthog/temporal/ai/anomaly_investigation/evaluated.py b/posthog/temporal/ai/anomaly_investigation/evaluated.py new file mode 100644 index 000000000000..51a628cd83df --- /dev/null +++ b/posthog/temporal/ai/anomaly_investigation/evaluated.py @@ -0,0 +1,48 @@ +"""What the investigated check measured, read from whichever stack recorded it. + +The insight stack keeps it on an `AlertCheck`. The alerts platform never writes one, so an +investigation of a platform evaluation carries it on its `InsightAlertInvestigation` row. +""" + +from __future__ import annotations + +from typing import Any + +from posthog.dataclasses import frozen + +from products.alerts.backend.models.alert import AlertCheck +from products.alerts.backend.models.investigation import InsightAlertInvestigation + + +@frozen +class EvaluatedCheck: + calculated_value: float | None + interval: str | None + triggered_dates: list[str] + triggered_points: list[int] + triggered_metadata: dict[str, Any] + reference: str + + @classmethod + def of_alert_check(cls, alert_check: AlertCheck) -> EvaluatedCheck: + return cls( + calculated_value=alert_check.calculated_value, + interval=alert_check.interval, + triggered_dates=list(alert_check.triggered_dates or []), + triggered_points=list(alert_check.triggered_points or []), + triggered_metadata=dict(alert_check.triggered_metadata or {}), + reference=f"Alert check id {alert_check.id}", + ) + + @classmethod + def of_investigation(cls, investigation: InsightAlertInvestigation) -> EvaluatedCheck: + evaluated = investigation.evaluated or {} + value = evaluated.get("value") + return cls( + calculated_value=float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None, + interval=evaluated.get("interval"), + triggered_dates=list(evaluated.get("triggered_dates") or []), + triggered_points=list(evaluated.get("triggered_points") or []), + triggered_metadata=dict(evaluated.get("triggered_metadata") or {}), + reference=f"Investigation id {investigation.id}", + ) diff --git a/posthog/temporal/ai/anomaly_investigation/notebook.py b/posthog/temporal/ai/anomaly_investigation/notebook.py index 1b26cc67fb3f..1e3fb81b3218 100644 --- a/posthog/temporal/ai/anomaly_investigation/notebook.py +++ b/posthog/temporal/ai/anomaly_investigation/notebook.py @@ -5,9 +5,10 @@ from markdown_it import MarkdownIt from markdown_it.token import Token +from posthog.temporal.ai.anomaly_investigation.evaluated import EvaluatedCheck from posthog.temporal.ai.anomaly_investigation.report import InvestigationReport -from products.alerts.backend.models.alert import AlertCheck, AlertConfiguration +from products.alerts.backend.models.alert import AlertConfiguration from products.notebooks.backend.facade.content import ( TipTapContent, TipTapNode, @@ -30,7 +31,7 @@ @dataclass class NotebookRenderContext: alert: AlertConfiguration - alert_check: AlertCheck + evaluated: EvaluatedCheck insight: Insight report: InvestigationReport @@ -122,13 +123,13 @@ def _plain_text(children: list[Token]) -> str: def _summary_line(ctx: NotebookRenderContext) -> str: detector_type = (ctx.alert.detector_config or {}).get("type") or "threshold" - triggered_dates = ctx.alert_check.triggered_dates or [] + triggered_dates = ctx.evaluated.triggered_dates window = "" if triggered_dates: window = f" at {triggered_dates[0]}" if len(triggered_dates) > 1: window = f" from {triggered_dates[0]} to {triggered_dates[-1]}" - value = ctx.alert_check.calculated_value + value = ctx.evaluated.calculated_value value_str = f"{value:.4f}" if isinstance(value, (int, float)) else "n/a" return ( f'Detector {detector_type} flagged an anomaly on insight "{ctx.insight.name or ctx.insight.short_id}"' @@ -138,7 +139,7 @@ def _summary_line(ctx: NotebookRenderContext) -> str: def _footer_line(ctx: NotebookRenderContext) -> str: return ( - f"Alert check id {ctx.alert_check.id}. " + f"{ctx.evaluated.reference}. " f"Tool calls used: {ctx.report.tool_calls_used}. " f"Generated automatically — review before acting." ) diff --git a/posthog/temporal/ai/anomaly_investigation/workflow.py b/posthog/temporal/ai/anomaly_investigation/workflow.py index 96f4cb2633a4..655e4a200325 100644 --- a/posthog/temporal/ai/anomaly_investigation/workflow.py +++ b/posthog/temporal/ai/anomaly_investigation/workflow.py @@ -1,7 +1,8 @@ """Temporal workflow that kicks off the anomaly investigation agent and persists -its findings as a Notebook linked to the AlertCheck. +its findings as a Notebook linked to the investigated check. -Triggered from posthog/temporal/alerts/workflows.py for a firing check the budget allows. +Triggered from posthog/temporal/alerts/workflows.py for a firing check the budget allows, and +for a platform evaluation of an insight alert from the alerts platform's insight source. """ from __future__ import annotations @@ -9,7 +10,7 @@ import json import asyncio from dataclasses import dataclass -from datetime import timedelta +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Optional from uuid import UUID @@ -26,13 +27,13 @@ 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, ) +from posthog.temporal.ai.anomaly_investigation.evaluated import EvaluatedCheck from posthog.temporal.ai.anomaly_investigation.event_provenance import alerted_series_event, describe_event_provenance from posthog.temporal.ai.anomaly_investigation.notebook import NotebookRenderContext, build_investigation_markdown from posthog.temporal.ai.anomaly_investigation.prompts import build_anomaly_context @@ -44,9 +45,21 @@ from posthog.utils import absolute_uri from products.alerts.backend.investigation_episode import EpisodeInvestigations, episode_investigations +from products.alerts.backend.investigation_release import ( + InvestigationRelease, + finish_investigation, + mark_investigation_failed, + mark_investigation_running, + verdict_change_is_news, +) from products.alerts.backend.models.alert import AlertCheck, AlertConfiguration, InvestigationStatus +from products.alerts.backend.models.investigation import InsightAlertInvestigation +from products.alerts_platform.backend.facade import api as platform +from products.alerts_platform.backend.facade.contracts import AlertDeliveryRequest, HeldDeliveryAddress +from products.alerts_platform.backend.facade.temporal import start_deliveries from products.notebooks.backend.facade import api as notebooks from products.notebooks.backend.facade.content import build_markdown_notebook_content +from products.notebooks.backend.facade.contracts import NotebookData from products.product_analytics.backend.facade.api import insights_including_soft_deleted_for_team from products.signals.backend.facade import api as signals @@ -78,6 +91,9 @@ "inconclusive": "Inconclusive", } +# Runs started before the agent released a platform evaluation's held delivery, and replay must not start one. +_PATCH_PLATFORM_RELEASE = "anomaly-investigation-platform-release-2026-11" + # Marker on the check's delivery receipts: the follow-up for a changed verdict is sent once. _VERDICT_CHANGE_FOLLOWUP_KEY = "investigation_verdict_change" @@ -91,14 +107,17 @@ class AnomalyInvestigationWorkflowInputs: team_id: int alert_id: UUID - alert_check_id: UUID + # The insight stack's check. A platform evaluation has none and names its investigation instead. + alert_check_id: Optional[UUID] = None user_id: Optional[int] = None trace_id: Optional[str] = None + investigation_id: Optional[UUID] = None @workflow.defn(name="anomaly-investigation") class AnomalyInvestigationWorkflow(PostHogWorkflow): - """Single-activity workflow — the heavy lifting happens inside the activity.""" + """The agent runs in one activity. A platform evaluation's verdict releases a held delivery, + which only a workflow can start, so the workflow starts it after the activity.""" @staticmethod def parse_inputs(inputs: list[str]) -> AnomalyInvestigationWorkflowInputs: @@ -107,7 +126,7 @@ def parse_inputs(inputs: list[str]) -> AnomalyInvestigationWorkflowInputs: @workflow.run async def run(self, inputs: AnomalyInvestigationWorkflowInputs) -> None: - await workflow.execute_activity( + deliveries = await workflow.execute_activity( investigate_anomaly_activity, inputs, start_to_close_timeout=timedelta(seconds=ANOMALY_INVESTIGATION_ACTIVITY_START_TO_CLOSE), @@ -118,15 +137,54 @@ async def run(self, inputs: AnomalyInvestigationWorkflowInputs) -> None: maximum_attempts=ANOMALY_INVESTIGATION_ACTIVITY_MAX_ATTEMPTS, ), ) + if workflow.patched(_PATCH_PLATFORM_RELEASE) and deliveries: + await start_deliveries(deliveries) + # Without the mark, the safety net starts the delivery again after its grace, and the + # delivery's id dedups it. The mark only saves that second start. + await workflow.execute_activity( + record_investigation_deliveries_started_activity, + deliveries, + start_to_close_timeout=timedelta(seconds=30), + retry_policy=RetryPolicy(maximum_attempts=3), + ) + + +@activity.defn +async def record_investigation_deliveries_started_activity(deliveries: list[AlertDeliveryRequest]) -> None: + now = datetime.now(UTC) + for delivery in deliveries: + await sync_to_async(platform.record_hold_delivery_started, thread_sensitive=False)( + HeldDeliveryAddress( + team_id=delivery.team_id, + configuration_id=delivery.configuration_id, + evaluation_key=delivery.evaluation_key, + ), + now=now, + ) @activity.defn -async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInputs) -> None: - team, alert, alert_check = await asyncio.gather( +async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInputs) -> list[AlertDeliveryRequest]: + """Runs the agent and records its verdict. Returns the platform deliveries the verdict releases, + which the workflow starts. An insight stack check delivers from here and returns none.""" + team, alert = await asyncio.gather( Team.objects.aget(id=inputs.team_id), AlertConfiguration.objects.select_related("insight").aget(id=inputs.alert_id), - AlertCheck.objects.aget(id=inputs.alert_check_id), ) + subject: AlertCheck | InsightAlertInvestigation + if inputs.investigation_id is not None: + subject = await sync_to_async(_load_investigation, thread_sensitive=False)( + inputs.team_id, inputs.investigation_id + ) + if subject.status == InvestigationStatus.DONE: + # A retry after the verdict was recorded repeats only its release, not the agent. + return list((await _finish_platform_investigation(alert, subject)).deliveries) + evaluated = EvaluatedCheck.of_investigation(subject) + elif inputs.alert_check_id is not None: + subject = await AlertCheck.objects.aget(id=inputs.alert_check_id) + evaluated = EvaluatedCheck.of_alert_check(subject) + else: + raise ValueError("An investigation needs an alert check or an investigation id") user: User | None = None if inputs.user_id is not None: @@ -138,15 +196,15 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput user = await _pick_investigation_user(alert) if user is None: - await _mark_failed(alert_check, "No user available to run the investigation agent.") - return + await _mark_failed(subject, "No user available to run the investigation agent.") + return [] - await _update_status(alert_check, InvestigationStatus.RUNNING) + await _mark_running(subject) - insight = await sync_to_async(_evaluated_insight, thread_sensitive=False)(alert, alert_check) + insight = await sync_to_async(_evaluated_insight, thread_sensitive=False)(alert, evaluated) metric_description = insight.name or f"Insight {insight.short_id}" - detector_type = _evaluated_detector_type(alert, alert_check) - series_index = _evaluated_series_index(alert, alert_check) + detector_type = _evaluated_detector_type(alert, evaluated) + series_index = _evaluated_series_index(alert, evaluated) # Measured up front rather than left to a tool call: without it the agent has only the # event's name to go on, and an opaque name invites it to invent the machinery behind it. @@ -161,10 +219,10 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput alert_name=alert.name or "Unnamed alert", metric_description=metric_description, detector_type=detector_type, - triggered_dates=list(alert_check.triggered_dates or []), - triggered_metadata=alert_check.triggered_metadata, - calculated_value=alert_check.calculated_value, - interval=alert_check.interval, + triggered_dates=evaluated.triggered_dates, + triggered_metadata=evaluated.triggered_metadata, + calculated_value=evaluated.calculated_value, + interval=evaluated.interval, # The alerted series, not series 0 — matching how the check and the chart pick it. metric_definition=describe_metric_definition( insight.query, series_index=series_index, alert_config=alert.config @@ -178,8 +236,8 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput anomaly_context = await sync_to_async(_build_multimodal_context, thread_sensitive=False)( alert=alert, context_text=anomaly_context_text, - triggered_dates=list(alert_check.triggered_dates or []), - triggered_points=list(alert_check.triggered_points or []), + triggered_dates=evaluated.triggered_dates, + triggered_points=evaluated.triggered_points, series_index=series_index, detector_type=detector_type, ) @@ -195,13 +253,13 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput ) except Exception as err: logger.exception("anomaly_investigation.agent_failed", alert_id=str(alert.id)) - await _mark_failed(alert_check, f"Agent run failed: {err}") + await _mark_failed(subject, f"Agent run failed: {err}") raise notebook_markdown = build_investigation_markdown( NotebookRenderContext( alert=alert, - alert_check=alert_check, + evaluated=evaluated, insight=insight, report=result.report, ) @@ -217,6 +275,74 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput creation_source=notebooks.NotebookCreationSource.TEMPORAL_AGENT, ) + summary = result.report.summary.strip() + deliveries: list[AlertDeliveryRequest] = [] + if isinstance(subject, InsightAlertInvestigation): + release = await sync_to_async(finish_investigation, thread_sensitive=False)( + alert, subject, verdict=result.report.verdict, summary=summary or None, notebook_id=notebook.id + ) + deliveries = list(release.deliveries) + previous_verdict = release.previous_verdict + episode_source_id = _episode_source_id(subject) + else: + episode = await _deliver_alert_check_outcome( + alert=alert, + alert_check=subject, + user=user, + verdict=result.report.verdict, + summary=summary, + notebook=notebook, + ) + previous_verdict = episode.previous_verdict + episode_source_id = episode.first_check_id + + # Surface the completed investigation to the Signals inbox, gated on the verdict so + # false-positive fires don't become inbox noise (the verdict stays auditable on the + # investigated record and in the notebook). Best-effort: the investigation itself has already + # succeeded and been persisted, so a failure to emit must not fail the activity + # (and trigger a re-run of the whole agent). + if not should_emit_episode_signal(result.report.verdict, previous_verdict, alert.investigation_inconclusive_action): + logger.info( + "anomaly_investigation.signal_skipped", + alert_id=str(alert.id), + **_subject_ids(subject), + verdict=result.report.verdict, + ) + return deliveries + + try: + await _emit_investigation_signal( + team=team, + alert=alert, + subject=subject, + evaluated=evaluated, + source_id=episode_source_id, + previous_verdict=previous_verdict, + insight=insight, + detector_type=detector_type, + report=result.report, + notebook_short_id=notebook.short_id, + ) + except Exception: + logger.exception( + "anomaly_investigation.signal_emission_failed", + alert_id=str(alert.id), + **_subject_ids(subject), + ) + return deliveries + + +async def _deliver_alert_check_outcome( + *, + alert: AlertConfiguration, + alert_check: AlertCheck, + user: User, + verdict: str, + summary: str, + notebook: NotebookData, +) -> EpisodeInvestigations: + """Records the verdict on the insight stack's check and delivers or suppresses its notification. + Returns the check's episode as it stood before this verdict.""" # Rendered before the check is marked DONE: once the status is terminal the safety net's # short grace applies (INVESTIGATION_NOTIFY_GRACE_MINUTES), and a slow render sitting # between the DONE update and the dispatch would let the sweep force-send its fallback @@ -225,18 +351,17 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput alert=alert, alert_check=alert_check, user=user, - verdict=result.report.verdict, + verdict=verdict, ) # Read the episode before this check's own verdict lands, so `previous_verdict` is the # last verdict of an earlier check in the same firing episode. episode = await sync_to_async(episode_investigations, thread_sensitive=False)(alert, alert_check) - summary = result.report.summary.strip() await sync_to_async(AlertCheck.objects.filter(id=alert_check.id).update, thread_sensitive=False)( investigation_notebook_id=notebook.id, investigation_status=InvestigationStatus.DONE, - investigation_verdict=result.report.verdict, + investigation_verdict=verdict, investigation_summary=summary or None, investigation_error=None, ) @@ -252,46 +377,43 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput await sync_to_async(_deliver_investigation_outcome, thread_sensitive=False)( alert=alert, alert_check=alert_check, - verdict=result.report.verdict, + verdict=verdict, previous_verdict=episode.previous_verdict, summary=summary, notebook_short_id=notebook.short_id, insight_chart_url=insight_chart_url, ) + return episode + + +async def _finish_platform_investigation( + alert: AlertConfiguration, investigation: InsightAlertInvestigation +) -> InvestigationRelease: + return await sync_to_async(finish_investigation, thread_sensitive=False)( + alert, + investigation, + verdict=investigation.verdict, + summary=investigation.summary, + notebook_id=investigation.notebook_id, + ) - # Surface the completed investigation to the Signals inbox, gated on the verdict so - # false-positive fires don't become inbox noise (the verdict stays auditable on the - # AlertCheck and in the notebook). Best-effort: the investigation itself has already - # succeeded and been persisted, so a failure to emit must not fail the activity - # (and trigger a re-run of the whole agent). - if not should_emit_episode_signal( - result.report.verdict, episode.previous_verdict, alert.investigation_inconclusive_action - ): - logger.info( - "anomaly_investigation.signal_skipped", - alert_id=str(alert.id), - alert_check_id=str(alert_check.id), - verdict=result.report.verdict, - ) - return - try: - await _emit_investigation_signal( - team=team, - alert=alert, - alert_check=alert_check, - episode=episode, - insight=insight, - detector_type=detector_type, - report=result.report, - notebook_short_id=notebook.short_id, - ) - except Exception: - logger.exception( - "anomaly_investigation.signal_emission_failed", - alert_id=str(alert.id), - alert_check_id=str(alert_check.id), - ) +def _load_investigation(team_id: int, investigation_id: UUID) -> InsightAlertInvestigation: + return InsightAlertInvestigation.objects.for_team(team_id).get(id=investigation_id) + + +def _episode_source_id(investigation: InsightAlertInvestigation) -> str: + """One identity for every investigation of one firing, as the insight stack keys a signal on + its episode's first check. The firing that was open at takeover has no start, and is one episode.""" + started = investigation.episode_started_at.isoformat() if investigation.episode_started_at else "takeover" + return f"{investigation.configuration_id}:{started}" + + +def _subject_ids(subject: AlertCheck | InsightAlertInvestigation) -> dict[str, str]: + """The keys that name the investigated record, on a signal and in logs.""" + if isinstance(subject, InsightAlertInvestigation): + return {"investigation_id": str(subject.id), "evaluation_key": subject.evaluation_key} + return {"alert_check_id": str(subject.id)} def should_emit_investigation_signal(verdict: str | None, inconclusive_action: str | None) -> bool: @@ -325,7 +447,8 @@ def should_emit_episode_signal( def _build_investigation_signal_extra( *, alert: AlertConfiguration, - alert_check: AlertCheck, + subject: AlertCheck | InsightAlertInvestigation, + evaluated: EvaluatedCheck, insight: Insight, detector_type: str, report: InvestigationReport, @@ -337,7 +460,7 @@ def _build_investigation_signal_extra( extra: dict = { "alert_id": str(alert.id), "alert_name": alert.name or "Unnamed alert", - "alert_check_id": str(alert_check.id), + **_subject_ids(subject), "insight_id": str(insight.id), "detector_type": detector_type, "verdict": report.verdict, @@ -349,8 +472,8 @@ def _build_investigation_signal_extra( insight_short_id = getattr(insight, "short_id", None) if insight_short_id: extra["insight_short_id"] = insight_short_id - if alert_check.triggered_dates: - extra["triggered_dates"] = list(alert_check.triggered_dates) + if evaluated.triggered_dates: + extra["triggered_dates"] = list(evaluated.triggered_dates) if notebook_short_id: extra["notebook_short_id"] = notebook_short_id return extra @@ -360,8 +483,10 @@ async def _emit_investigation_signal( *, team: Team, alert: AlertConfiguration, - alert_check: AlertCheck, - episode: EpisodeInvestigations, + subject: AlertCheck | InsightAlertInvestigation, + evaluated: EvaluatedCheck, + source_id: str, + previous_verdict: str | None, insight: Insight, detector_type: str, report: InvestigationReport, @@ -369,8 +494,8 @@ async def _emit_investigation_signal( ) -> None: """Emit an `alerts/anomaly_investigation` signal carrying the agent's verdict and findings. - The source id is the episode's first check, not this one, so every investigation of one - incident carries the same identity. Grouping matches a signal on its description and on + The source id names the episode, not this check, so every investigation of one incident + carries the same identity. Grouping matches a signal on its description and on semantically near signals, not on the source id, so one report per episode is a strong default and not a guarantee: a re-emit can still open a report of its own. """ @@ -378,19 +503,20 @@ async def _emit_investigation_signal( team=team, source_product=SIGNAL_SOURCE_PRODUCT, source_type=SIGNAL_SOURCE_TYPE, - source_id=episode.first_check_id, + source_id=source_id, description=_build_signal_description( alert_name=alert.name or "Unnamed alert", insight_name=insight.name or None, insight_id=str(insight.id), insight_short_id=getattr(insight, "short_id", None), report=report, - previous_verdict=episode.previous_verdict, + previous_verdict=previous_verdict, ), weight=1, extra=_build_investigation_signal_extra( alert=alert, - alert_check=alert_check, + subject=subject, + evaluated=evaluated, insight=insight, detector_type=detector_type, report=report, @@ -562,9 +688,7 @@ def _dispatch_verdict_change_followup( Never raises into the activity: a correction that cannot be sent is not worth a second agent run. """ - if not previous_verdict or verdict == previous_verdict: - return - if _inconclusive_is_suppressed(verdict, alert.investigation_inconclusive_action): + if not verdict_change_is_news(verdict, previous_verdict, alert.investigation_inconclusive_action): return receipts = check.targets_notified or {} if receipts.get(_VERDICT_CHANGE_FOLLOWUP_KEY): @@ -685,30 +809,36 @@ def _build_breach_descriptions( return _BreachDescriptions(lines=lines, summary=summary_line) -async def _update_status(alert_check, status: str) -> None: - await sync_to_async(AlertCheck.objects.filter(id=alert_check.id).update, thread_sensitive=False)( - investigation_status=status, +async def _mark_running(subject: AlertCheck | InsightAlertInvestigation) -> None: + if isinstance(subject, InsightAlertInvestigation): + await sync_to_async(mark_investigation_running, thread_sensitive=False)(subject) + return + await sync_to_async(AlertCheck.objects.filter(id=subject.id).update, thread_sensitive=False)( + investigation_status=InvestigationStatus.RUNNING, ) -async def _mark_failed(alert_check, reason: str) -> None: - await sync_to_async(AlertCheck.objects.filter(id=alert_check.id).update, thread_sensitive=False)( +async def _mark_failed(subject: AlertCheck | InsightAlertInvestigation, reason: str) -> None: + if isinstance(subject, InsightAlertInvestigation): + await sync_to_async(mark_investigation_failed, thread_sensitive=False)(subject, reason) + return + await sync_to_async(AlertCheck.objects.filter(id=subject.id).update, thread_sensitive=False)( investigation_status=InvestigationStatus.FAILED, investigation_error={"message": reason}, ) -def _evaluated_series_index(alert, alert_check) -> int: +def _evaluated_series_index(alert, evaluated) -> int: """The series the check judged, which the alert can have been repointed away from since.""" - saved = (alert_check.triggered_metadata or {}).get("series_index") + saved = (evaluated.triggered_metadata or {}).get("series_index") if isinstance(saved, int) and not isinstance(saved, bool): return saved return (alert.config or {}).get("series_index", 0) -def _evaluated_detector_type(alert, alert_check) -> str: +def _evaluated_detector_type(alert, evaluated) -> str: """The detector that produced the check, which the alert can have been moved off since.""" - metadata = alert_check.triggered_metadata or {} + metadata = evaluated.triggered_metadata or {} saved = metadata.get("detector_type") if isinstance(saved, str) and saved: return saved @@ -718,9 +848,9 @@ def _evaluated_detector_type(alert, alert_check) -> str: return (alert.detector_config or {}).get("type") or "threshold" -def _evaluated_insight(alert, alert_check): +def _evaluated_insight(alert, evaluated): """The insight the check judged, which the alert can have been repointed away from since.""" - saved = (alert_check.triggered_metadata or {}).get("insight_id") + saved = (evaluated.triggered_metadata or {}).get("insight_id") if not isinstance(saved, int) or isinstance(saved, bool) or saved == alert.insight_id: return alert.insight # The judged insight can be soft-deleted by now; its definition is still what the check was about. diff --git a/posthog/temporal/tests/ai/test_anomaly_investigation_notebook.py b/posthog/temporal/tests/ai/test_anomaly_investigation_notebook.py index 551861ea635d..e1bc1f4066c9 100644 --- a/posthog/temporal/tests/ai/test_anomaly_investigation_notebook.py +++ b/posthog/temporal/tests/ai/test_anomaly_investigation_notebook.py @@ -3,6 +3,7 @@ from markdown_it import MarkdownIt +from posthog.temporal.ai.anomaly_investigation.evaluated import EvaluatedCheck from posthog.temporal.ai.anomaly_investigation.notebook import ( NotebookRenderContext, _summary_content, @@ -30,7 +31,9 @@ def _make_context(self, report: InvestigationReport) -> NotebookRenderContext: triggered_dates=["2024-06-01", "2024-06-02"], interval="day", ) - return NotebookRenderContext(alert=alert, alert_check=alert_check, insight=insight, report=report) + return NotebookRenderContext( + alert=alert, evaluated=EvaluatedCheck.of_alert_check(alert_check), insight=insight, report=report + ) def test_builds_markdown_the_cell_tools_accept(self) -> None: report = InvestigationReport( diff --git a/posthog/temporal/tests/ai/test_anomaly_investigation_verdict.py b/posthog/temporal/tests/ai/test_anomaly_investigation_verdict.py index a049e0ebf34a..dbd8778477c5 100644 --- a/posthog/temporal/tests/ai/test_anomaly_investigation_verdict.py +++ b/posthog/temporal/tests/ai/test_anomaly_investigation_verdict.py @@ -1,6 +1,6 @@ """Verifies that the workflow activity persists the agent's verdict onto the AlertCheck.""" -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest from posthog.test.base import NonAtomicBaseTest @@ -11,6 +11,7 @@ from posthog.schema import AlertState +from posthog.models.scoping import team_scope from posthog.temporal.ai.anomaly_investigation.report import InvestigationReport from posthog.temporal.ai.anomaly_investigation.runner import InvestigationRunResult from posthog.temporal.ai.anomaly_investigation.workflow import ( @@ -26,7 +27,9 @@ InvestigationStatus, InvestigationVerdict, ) -from products.alerts_platform.backend.facade.contracts import AlertDelivery +from products.alerts.backend.models.investigation import InsightAlertInvestigation +from products.alerts_platform.backend.facade import testing as platform_testing +from products.alerts_platform.backend.facade.contracts import AlertDelivery, AlertDeliveryRequest, SourceKind from products.cdp.backend.models.hog_functions.hog_function import HogFunction from products.exports.backend.models.exported_asset import ExportedAsset from products.notebooks.backend.facade.content import is_markdown_notebook_content @@ -346,3 +349,98 @@ def test_follow_up_failure_does_not_fail_the_activity(self, _mock_dispatch) -> N self.alert_check.refresh_from_db() assert self.alert_check.targets_notified == {} + + +class TestPlatformEvaluationInvestigation(NonAtomicBaseTest): + CLASS_DATA_LEVEL_SETUP = False + + def setUp(self) -> None: + super().setUp() + self.insight = Insight.objects.create(team=self.team, name="test insight") + self.alert = AlertConfiguration.objects.create( + team=self.team, + insight=self.insight, + name="anomaly alert", + detector_config={"type": "zscore", "threshold": 0.95, "window": 30}, + investigation_agent_enabled=True, + investigation_gates_notifications=True, + created_by=self.user, + ) + with team_scope(self.team.id): + configuration = platform_testing.create_configuration( + team_id=self.team.id, + name="anomaly alert", + source_kind=SourceKind.INSIGHT, + check_interval_minutes=60, + next_check_at=datetime.now(UTC), + ) + self.request = AlertDeliveryRequest( + source=SourceKind.INSIGHT, + team_id=self.team.id, + configuration_id=str(configuration.id), + evaluation_key="slot:now", + destination_alert_id=str(self.alert.id), + event_ids_by_kind={"firing": "$insight_alert_firing"}, + owned_by_platform=True, + ) + platform_testing.hold_delivery( + self.request, held_at=datetime.now(UTC), release_after=datetime.now(UTC) + timedelta(minutes=90) + ) + self.episode_started_at = datetime(2026, 10, 20, 9, tzinfo=UTC) + self.investigation = InsightAlertInvestigation.objects.for_team(self.team.id).create( + team=self.team, + alert_configuration=self.alert, + configuration_id=configuration.id, + evaluation_key="slot:now", + episode_started_at=self.episode_started_at, + kind=InsightAlertInvestigation.Kind.AGENT, + status=InvestigationStatus.PENDING, + evaluated={"value": 42.0, "triggered_dates": ["2026-10-20"], "interval": "day"}, + ) + + def _record_verdict(self) -> None: + InsightAlertInvestigation.objects.for_team(self.team.id).filter(id=self.investigation.id).update( + status=InvestigationStatus.DONE, verdict=InvestigationVerdict.TRUE_POSITIVE + ) + + @parameterized.expand( + [("first run", InvestigationStatus.PENDING), ("retry after the verdict", InvestigationStatus.DONE)] + ) + @pytest.mark.asyncio + @patch("posthog.temporal.ai.anomaly_investigation.workflow.signals.emit_signal") + @patch("posthog.temporal.ai.anomaly_investigation.workflow.run_investigation") + @patch("temporalio.activity.heartbeat") + @patch("temporalio.activity.info") + async def test_a_verdict_hands_the_held_firing_back_to_the_workflow( + self, _name, status, mock_info, _heartbeat, mock_run, mock_emit + ) -> None: + mock_info.return_value.heartbeat_timeout = None + mock_run.return_value = InvestigationRunResult( + report=InvestigationReport( + verdict="true_positive", summary="Confirmed.", hypotheses=[], recommendations=[] + ), + tool_calls_used=0, + model="test-model", + ) + if status == InvestigationStatus.DONE: + await sync_to_async(self._record_verdict)() + + deliveries = await investigate_anomaly_activity( + AnomalyInvestigationWorkflowInputs( + team_id=self.team.id, alert_id=self.alert.id, investigation_id=self.investigation.id + ) + ) + + assert deliveries == [self.request] + await sync_to_async(self.investigation.refresh_from_db)() + assert (self.investigation.status, self.investigation.verdict) == ( + InvestigationStatus.DONE, + InvestigationVerdict.TRUE_POSITIVE, + ) + assert mock_run.called is (status == InvestigationStatus.PENDING) + if status == InvestigationStatus.PENDING: + assert self.investigation.notebook_id is not None + kwargs = mock_emit.await_args.kwargs + assert kwargs["source_id"] == f"{self.investigation.configuration_id}:{self.episode_started_at.isoformat()}" + assert kwargs["extra"]["investigation_id"] == str(self.investigation.id) + assert "alert_check_id" not in kwargs["extra"] diff --git a/posthog/temporal/tests/ai/test_module_integrity.py b/posthog/temporal/tests/ai/test_module_integrity.py index 4382b8e56c7e..0cc26a22fdd9 100644 --- a/posthog/temporal/tests/ai/test_module_integrity.py +++ b/posthog/temporal/tests/ai/test_module_integrity.py @@ -49,6 +49,7 @@ def test_activities_remain_unchanged(self): "process_research_agent_activity", "summarize_llm_traces_activity", "investigate_anomaly_activity", + "record_investigation_deliveries_started_activity", "select_checkpoint_compaction_batch", "compact_checkpoint_conversations", ] diff --git a/products/alerts/backend/investigation_release.py b/products/alerts/backend/investigation_release.py new file mode 100644 index 000000000000..cf4c6883bea0 --- /dev/null +++ b/products/alerts/backend/investigation_release.py @@ -0,0 +1,182 @@ +"""What the investigation agent writes back for a platform evaluation of an insight alert. + +The agent records its status and verdict on the `InsightAlertInvestigation` row, and releases the +firing's held delivery by the verdict. Kept apart from the decision and claim, which read the +insight stack's rules from `posthog.temporal.alerts`, because that package imports the agent. +""" + +from datetime import UTC, datetime, timedelta +from typing import Final +from uuid import UUID + +from django.db import transaction + +import structlog + +from posthog.dataclasses import frozen +from posthog.tasks.alerts.investigation_notifications import ( + INVESTIGATION_NOTIFY_GRACE_MINUTES, + INVESTIGATION_RUNNING_GRACE_MINUTES, +) +from posthog.tasks.alerts.utils import _inconclusive_is_suppressed, _should_suppress_notification + +from products.alerts.backend.models.alert import AlertConfiguration, InvestigationStatus +from products.alerts.backend.models.investigation import InsightAlertInvestigation +from products.alerts_platform.backend.facade import api as platform +from products.alerts_platform.backend.facade.contracts import AlertDeliveryRequest, HeldDeliveryAddress +from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy + +logger = structlog.get_logger(__name__) + +# The insight stack's safety-net graces: a running investigation holds its firing this long after +# the check, and a finished one this long after it finished. +RUNNING_HOLD: Final = timedelta(minutes=INVESTIGATION_RUNNING_GRACE_MINUTES) +TERMINAL_HOLD: Final = timedelta(minutes=INVESTIGATION_NOTIFY_GRACE_MINUTES) + + +def earlier_in_episode( + team_id: int, configuration_id: UUID, episode_started_at: datetime | None, evaluation_key: str +) -> list[tuple[str, str | None]]: + """Status and verdict of the firing's other investigations, newest first. One read, because a + firing holds a few rows. Its own row is left out, because a retry can find it already written.""" + return list( + InsightAlertInvestigation.objects.for_team(team_id) + .filter(configuration_id=configuration_id, episode_started_at=episode_started_at) + .exclude(evaluation_key=evaluation_key) + .order_by("-created_at") + .values_list("status", "verdict") + ) + + +def last_verdict(earlier: list[tuple[str, str | None]]) -> str | None: + return next((verdict for _, verdict in earlier if verdict is not None), None) + + +def previous_verdict_in_episode( + team_id: int, configuration_id: UUID, episode_started_at: datetime | None, evaluation_key: str +) -> str | None: + """The last verdict another investigation of the same firing reached.""" + return last_verdict(earlier_in_episode(team_id, configuration_id, episode_started_at, evaluation_key)) + + +def verdict_change_is_news(verdict: str | None, previous_verdict: str | None, inconclusive_action: str | None) -> bool: + """Whether a verdict that differs from the firing's previous one earns a follow-up on a firing that + already went out. A change to a false positive is sent, because it corrects what the user already + read. An inconclusive verdict still follows the alert's policy.""" + if not previous_verdict or verdict == previous_verdict: + return False + return not _inconclusive_is_suppressed(verdict, inconclusive_action) + + +def release_held( + address: HeldDeliveryAddress, *, suppress: bool, released_by: HoldReleasedBy +) -> AlertDeliveryRequest | None: + """Releases a held firing. Returns the request the caller starts when it goes out. A retry from the + same releaser gets the same request back, so it can start a delivery the first attempt did not.""" + return platform.release_hold( + address, + action=HoldRelease.SUPPRESS if suppress else HoldRelease.SEND, + released_by=released_by, + now=datetime.now(UTC), + ) + + +@frozen +class InvestigationRelease: + """What a finished investigation hands back to the workflow that ran it.""" + + # What the workflow starts: the held firing when the verdict let it go out, or the + # announcement of a changed verdict on a firing that already went out. + deliveries: tuple[AlertDeliveryRequest, ...] + previous_verdict: str | None + + +def _hold_of(investigation: InsightAlertInvestigation) -> HeldDeliveryAddress: + return HeldDeliveryAddress( + team_id=investigation.team_id, + configuration_id=str(investigation.configuration_id), + evaluation_key=investigation.evaluation_key, + ) + + +def mark_investigation_running(investigation: InsightAlertInvestigation) -> None: + """Marks the investigation as running, and holds its firing for the running grace again, because + a retry after a failure would otherwise run on the failure's short grace.""" + with transaction.atomic(): + InsightAlertInvestigation.objects.for_team(investigation.team_id).filter(id=investigation.id).update( + status=InvestigationStatus.RUNNING + ) + platform.reschedule_hold(_hold_of(investigation), release_after=investigation.created_at + RUNNING_HOLD) + + +def mark_investigation_failed(investigation: InsightAlertInvestigation, reason: str) -> None: + """Records the failure, and leaves the firing to the safety net after the terminal grace, as the + insight stack sends a firing whose investigation failed.""" + with transaction.atomic(): + InsightAlertInvestigation.objects.for_team(investigation.team_id).filter(id=investigation.id).update( + status=InvestigationStatus.FAILED, error={"message": reason} + ) + platform.reschedule_hold(_hold_of(investigation), release_after=datetime.now(UTC) + TERMINAL_HOLD) + + +def finish_investigation( + alert: AlertConfiguration, + investigation: InsightAlertInvestigation, + *, + verdict: str | None, + summary: str | None, + notebook_id: UUID | None, +) -> InvestigationRelease: + """Records the verdict, then releases the firing's held delivery by it. + + The terminal grace starts in the transaction that records the verdict, so the safety net cannot + send the firing between the two. A firing that already went out gets a verdict-changed + announcement when its verdict contradicts the firing's previous one. + + Safe to repeat for a recorded verdict: a retry gets back what the first call released. + """ + previous_verdict = previous_verdict_in_episode( + investigation.team_id, + investigation.configuration_id, + investigation.episode_started_at, + investigation.evaluation_key, + ) + with transaction.atomic(): + InsightAlertInvestigation.objects.for_team(investigation.team_id).filter(id=investigation.id).update( + status=InvestigationStatus.DONE, verdict=verdict, summary=summary, notebook_id=notebook_id, error=None + ) + platform.reschedule_hold(_hold_of(investigation), release_after=datetime.now(UTC) + TERMINAL_HOLD) + delivery = release_held( + _hold_of(investigation), + suppress=_should_suppress_notification(verdict, alert.investigation_inconclusive_action), + released_by=HoldReleasedBy.VERDICT, + ) + if delivery is None: + delivery = _announce_verdict_change(alert, investigation, verdict=verdict, previous_verdict=previous_verdict) + return InvestigationRelease(deliveries=(delivery,) if delivery else (), previous_verdict=previous_verdict) + + +def _announce_verdict_change( + alert: AlertConfiguration, + investigation: InsightAlertInvestigation, + *, + verdict: str | None, + previous_verdict: str | None, +) -> AlertDeliveryRequest | None: + """The insight stack's verdict-change follow-up, for a firing whose delivery already went out.""" + if not verdict_change_is_news(verdict, previous_verdict, alert.investigation_inconclusive_action): + return None + address = _hold_of(investigation) + if platform.delivery_held_back(address): + # The firing was suppressed, or the safety net has not sent it yet, so the user read no verdict. + return None + logger.info( + "insight_alert_investigation.verdict_changed", + alert_id=str(alert.id), + investigation_id=str(investigation.id), + verdict=verdict, + previous_verdict=previous_verdict, + ) + return platform.record_verdict_changed( + address, episode_started_at=investigation.episode_started_at, now=datetime.now(UTC) + ) diff --git a/products/alerts/backend/platform_investigation.py b/products/alerts/backend/platform_investigation.py index af37f5c41844..66ae55db10d2 100644 --- a/products/alerts/backend/platform_investigation.py +++ b/products/alerts/backend/platform_investigation.py @@ -23,6 +23,7 @@ ) 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.models.alert import AlertConfiguration, InvestigationStatus from products.alerts.backend.models.investigation import InsightAlertInvestigation @@ -53,18 +54,10 @@ def decide_investigation( """ if not investigates(alert, firing=firing): return InvestigationDecision() - # Newest first, so the first verdict is the last one reached. One read, because a firing holds - # a few rows. Its own row exists when a retry repeats a decision its claim already wrote. - earlier = list( - InsightAlertInvestigation.objects.for_team(alert.team_id) - .filter(configuration_id=configuration_id, episode_started_at=episode_started_at) - .exclude(evaluation_key=evaluation_key) - .order_by("-created_at") - .values_list("status", "verdict") - ) + earlier = earlier_in_episode(alert.team_id, configuration_id, episode_started_at, evaluation_key) return decide_within_budget( started=sum(status in STARTED_INVESTIGATION_STATUSES for status, _ in earlier), - previous_verdict=next((verdict for _, verdict in earlier if verdict is not None), None), + previous_verdict=last_verdict(earlier), ) diff --git a/products/alerts/backend/tests/test_platform_investigation.py b/products/alerts/backend/tests/test_platform_investigation.py index 09fa6e7e98bd..1a78c9f9b8d1 100644 --- a/products/alerts/backend/tests/test_platform_investigation.py +++ b/products/alerts/backend/tests/test_platform_investigation.py @@ -3,15 +3,21 @@ import time_machine from posthog.test.base import BaseTest +from unittest.mock import patch from parameterized import parameterized +from posthog.models.scoping import team_scope from posthog.schema_enums import AlertCalculationInterval from posthog.temporal.alerts.investigation import MAX_INVESTIGATIONS_PER_EPISODE, InvestigationDecision +from products.alerts.backend.investigation_release import RUNNING_HOLD, InvestigationRelease, finish_investigation from products.alerts.backend.models.alert import AlertConfiguration, InvestigationStatus from products.alerts.backend.models.investigation import InsightAlertInvestigation from products.alerts.backend.platform_investigation import InvestigationClaim, claim_investigation, decide_investigation +from products.alerts_platform.backend.facade import testing as platform_testing +from products.alerts_platform.backend.facade.contracts import AlertDeliveryRequest, HeldDeliveryAddress, SourceKind +from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy from products.product_analytics.backend.facade.models import Insight NOW = datetime(2026, 10, 20, 10, tzinfo=UTC) @@ -154,3 +160,100 @@ def test_a_repeated_claim_returns_the_first_answer(self, _name: str, earlier: st assert self._claim() == first assert first.claimed is claimed + + +class TestFinishInvestigation(BaseTest): + def setUp(self) -> None: + super().setUp() + self.alert = AlertConfiguration.objects.create( + team=self.team, + insight=Insight.objects.create(team=self.team, name="Signups"), + name="Signups dropped", + detector_config={"type": "zscore", "threshold": 0.95, "window": 30}, + investigation_agent_enabled=True, + investigation_gates_notifications=True, + ) + with team_scope(self.team.id): + self.configuration = platform_testing.create_configuration( + team_id=self.team.id, + name="Signups dropped", + source_kind=SourceKind.INSIGHT, + check_interval_minutes=60, + next_check_at=NOW, + ) + self.address = HeldDeliveryAddress( + team_id=self.team.id, configuration_id=str(self.configuration.id), evaluation_key="slot:now" + ) + self.request = AlertDeliveryRequest( + source=SourceKind.INSIGHT, + team_id=self.team.id, + configuration_id=str(self.configuration.id), + evaluation_key="slot:now", + destination_alert_id=str(self.alert.id), + event_ids_by_kind={"firing": "$insight_alert_firing"}, + owned_by_platform=True, + ) + + def _row(self, key: str, *, status: str, verdict: str | None = None) -> InsightAlertInvestigation: + return 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, + ) + + def _hold(self, *, released: HoldRelease | None = None) -> None: + platform_testing.hold_delivery( + self.request, + held_at=NOW, + release_after=NOW + RUNNING_HOLD, + released=(released, HoldReleasedBy.SAFETY_NET) if released else None, + ) + + def _finish(self, investigation: InsightAlertInvestigation, verdict: str) -> InvestigationRelease: + return finish_investigation(self.alert, investigation, verdict=verdict, summary="Found it.", notebook_id=None) + + @parameterized.expand( + [ + ("a true positive sends", "true_positive", HoldRelease.SEND), + ("a false positive suppresses", "false_positive", HoldRelease.SUPPRESS), + ] + ) + def test_a_verdict_releases_the_held_firing(self, _name: str, verdict: str, action: HoldRelease) -> None: + self._hold() + investigation = self._row("slot:now", status=InvestigationStatus.RUNNING) + + release = self._finish(investigation, verdict) + repeated = self._finish(investigation, verdict) + + expected = (self.request,) if action == HoldRelease.SEND else () + assert release.deliveries == repeated.deliveries == expected + assert platform_testing.hold_release(self.address)[:2] == (action, HoldReleasedBy.VERDICT) + investigation.refresh_from_db() + assert (investigation.status, investigation.verdict) == (InvestigationStatus.DONE, verdict) + + @parameterized.expand( + [ + ("an ungated firing that already went out", None, "false_positive", True), + ("a firing the safety net already sent", HoldRelease.SEND, "false_positive", True), + ("an unchanged verdict", None, "true_positive", False), + ("a firing that was suppressed", HoldRelease.SUPPRESS, "false_positive", False), + ] + ) + def test_a_changed_verdict_is_announced_only_on_a_firing_that_went_out( + self, _name: str, released: HoldRelease | None, verdict: str, announced: bool + ) -> None: + if released is not None: + self._hold(released=released) + self._row("slot:earlier", status=InvestigationStatus.DONE, verdict="true_positive") + investigation = self._row("slot:now", status=InvestigationStatus.RUNNING) + + with patch("products.alerts.backend.investigation_release.platform.record_verdict_changed") as announce: + release = self._finish(investigation, verdict) + + assert release.previous_verdict == "true_positive" + assert announce.called is announced diff --git a/products/alerts_platform/backend/facade/api.py b/products/alerts_platform/backend/facade/api.py index 85774ea367fa..37f1bf5123ff 100644 --- a/products/alerts_platform/backend/facade/api.py +++ b/products/alerts_platform/backend/facade/api.py @@ -84,6 +84,22 @@ def reschedule_hold(address: HeldDeliveryAddress, *, release_after: datetime) -> return delivery_holds.reschedule_hold(address, release_after=release_after) +def delivery_held_back(address: HeldDeliveryAddress) -> bool: + """Whether an evaluation's delivery is held and not released with a send, so it has not gone out.""" + return delivery_holds.blocks_delivery(address.team_id, address.configuration_id, address.evaluation_key) + + +def record_verdict_changed( + address: HeldDeliveryAddress, *, episode_started_at: datetime | None, now: datetime +) -> AlertDeliveryRequest | None: + """Announces that a source's later verdict contradicts the one a delivered firing carried, in + that firing's thread. Returns the request the caller starts, or None when nothing is to be sent. + + The platform has no verdict-changed kind to record it under, so this records nothing and returns None. + """ + return None + + 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/testing.py b/products/alerts_platform/backend/facade/testing.py index 2677da4fab6e..0fcf1b56c6b2 100644 --- a/products/alerts_platform/backend/facade/testing.py +++ b/products/alerts_platform/backend/facade/testing.py @@ -17,16 +17,19 @@ from products.alerts_platform.backend.facade.contracts import ( AlertDeliveryRequest, DeliveryPath, + HeldDeliveryAddress, PlatformAlertSnapshot, PlatformConfigurationSnapshot, 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, ) @@ -66,6 +69,38 @@ def copy_of(legacy_configuration_id: UUID) -> PlatformConfigurationSnapshot: ) +def hold_delivery( + request: AlertDeliveryRequest, + *, + held_at: datetime, + release_after: datetime, + released: tuple[HoldRelease, HoldReleasedBy] | None = None, +) -> None: + """Holds a request's delivery as a check would, or as one a release already decided.""" + PlatformAlertDeliveryHold.objects.for_team(request.team_id).create( + team_id=request.team_id, + configuration_id=request.configuration_id, + evaluation_key=request.evaluation_key, + reason="test", + request=request.to_stored(), + held_at=held_at, + release_after=release_after, + released_at=held_at if released else None, + release_action=released[0] if released else None, + released_by=released[1] if released else None, + ) + + +def hold_release(address: HeldDeliveryAddress) -> tuple[str | None, str | None, datetime]: + """A hold's release action, who released it, and its deadline.""" + hold = ( + PlatformAlertDeliveryHold.objects.for_team(address.team_id) + .filter(configuration_id=address.configuration_id, evaluation_key=address.evaluation_key) + .get() + ) + return hold.release_action, hold.released_by, hold.release_after + + def set_due_at(configuration_id: UUID, due_at: datetime) -> None: """Move a configuration's due time, so a test can run two checks against one schedule.""" # nosemgrep: idor-lookup-without-team — fail-closed model; the caller holds the team scope diff --git a/products/signals/backend/contracts.py b/products/signals/backend/contracts.py index ed095d320a21..ad3872590931 100644 --- a/products/signals/backend/contracts.py +++ b/products/signals/backend/contracts.py @@ -462,7 +462,6 @@ class ReplayVisionScannerFindingSignalInput(SignalInputBase): class AnalyticsAnomalyInvestigationSignalExtra(SignalExtraBase): alert_id: str alert_name: str - alert_check_id: str insight_id: str detector_type: str verdict: Literal["true_positive", "false_positive", "inconclusive"] @@ -471,6 +470,9 @@ class AnalyticsAnomalyInvestigationSignalExtra(SignalExtraBase): insight_short_id: str | None = None triggered_dates: list[str] | None = None notebook_short_id: str | None = None + alert_check_id: str | None = None + investigation_id: str | None = None + evaluation_key: str | None = None class AnalyticsAnomalyInvestigationSignalInput(SignalInputBase): diff --git a/products/signals/backend/test/test_contracts.py b/products/signals/backend/test/test_contracts.py index 31256c695333..ee37bc2d7c9a 100644 --- a/products/signals/backend/test/test_contracts.py +++ b/products/signals/backend/test/test_contracts.py @@ -1,7 +1,28 @@ +from parameterized import parameterized + from products.signals.backend.contracts import AnalyticsAnomalyInvestigationSignalInput +_EXTRA = { + "alert_id": "alert-id", + "alert_name": "Signups dropped", + "insight_id": "insight-id", + "detector_type": "zscore", + "verdict": "true_positive", + "url": "https://app.posthog.com/notebooks/notebook-short-id", + "insight_name": "Daily signups", + "insight_short_id": "insight-short-id", + "triggered_dates": ["2026-07-14", "2026-07-15"], + "notebook_short_id": "notebook-short-id", +} + -def test_anomaly_investigation_payload_matches_emitter_contract() -> None: +@parameterized.expand( + [ + ("insight stack check", {"alert_check_id": "alert-check-id"}), + ("platform evaluation", {"investigation_id": "investigation-id", "evaluation_key": "slot:2026-07-15T09:00"}), + ] +) +def test_anomaly_investigation_payload_matches_emitter_contract(_name: str, subject: dict[str, str]) -> None: AnalyticsAnomalyInvestigationSignalInput.model_validate( { "source_product": "analytics", @@ -9,18 +30,6 @@ def test_anomaly_investigation_payload_matches_emitter_contract() -> None: "source_id": "alert-check-id", "description": "Anomaly investigation completed.", "weight": 1, - "extra": { - "alert_id": "alert-id", - "alert_name": "Signups dropped", - "alert_check_id": "alert-check-id", - "insight_id": "insight-id", - "detector_type": "zscore", - "verdict": "true_positive", - "url": "https://app.posthog.com/notebooks/notebook-short-id", - "insight_name": "Daily signups", - "insight_short_id": "insight-short-id", - "triggered_dates": ["2026-07-14", "2026-07-15"], - "notebook_short_id": "notebook-short-id", - }, + "extra": {**_EXTRA, **subject}, } ) diff --git a/products/signals/frontend/generated/api.schemas.ts b/products/signals/frontend/generated/api.schemas.ts index b76667ac5650..6d347a3e67f0 100644 --- a/products/signals/frontend/generated/api.schemas.ts +++ b/products/signals/frontend/generated/api.schemas.ts @@ -4139,7 +4139,6 @@ export const InvestigationVerdictEnumApi = { export interface AnalyticsAnomalyInvestigationSignalExtraApi { alert_id: string alert_name: string - alert_check_id: string insight_id: string detector_type: string verdict: InvestigationVerdictEnumApi @@ -4148,6 +4147,9 @@ export interface AnalyticsAnomalyInvestigationSignalExtraApi { insight_short_id?: string | null triggered_dates?: string[] | null notebook_short_id?: string | null + alert_check_id?: string | null + investigation_id?: string | null + evaluation_key?: string | null } export type HealthCheckSignalExtraSeverityEnumApi = diff --git a/products/today/backend/logic/impact.py b/products/today/backend/logic/impact.py index a960d0be9bb5..baa65d378411 100644 --- a/products/today/backend/logic/impact.py +++ b/products/today/backend/logic/impact.py @@ -51,7 +51,8 @@ def _occurrence_of(signal: SignalInput) -> _Occurrence | None: ticket = js_number(signal.extra.get("ticket_number")) or signal.source_id return _Occurrence(kind="tickets", key=f"{signal.source_product}:{ticket}") if ticket else None if signal.source_product == "analytics" and signal.source_type == "anomaly_investigation": - return _Occurrence(kind="alerts", key=text_of(signal.extra.get("alert_check_id")) or signal.source_id) + check = text_of(signal.extra.get("alert_check_id")) or text_of(signal.extra.get("investigation_id")) + return _Occurrence(kind="alerts", key=check or signal.source_id) return None diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 9c380f213e0d..c90351480ff0 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -13447,7 +13447,6 @@ export namespace Schemas { export interface AnalyticsAnomalyInvestigationSignalExtra { alert_id: string; alert_name: string; - alert_check_id: string; insight_id: string; detector_type: string; verdict: InvestigationVerdictEnum; @@ -13456,6 +13455,9 @@ export namespace Schemas { insight_short_id?: string | null; triggered_dates?: string[] | null; notebook_short_id?: string | null; + alert_check_id?: string | null; + investigation_id?: string | null; + evaluation_key?: string | null; } /** From 11a81ed1c893698f5248d443dc1999538f0d5fc5 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 17:27:04 -0800 Subject: [PATCH 2/6] refactor(alerts): split loading out of the investigation agent's activity Co-Authored-By: Claude Opus 5.5 --- .../ai/anomaly_investigation/workflow.py | 51 +++++++++++-------- 1 file changed, 29 insertions(+), 22 deletions(-) diff --git a/posthog/temporal/ai/anomaly_investigation/workflow.py b/posthog/temporal/ai/anomaly_investigation/workflow.py index 655e4a200325..94e59baa9a50 100644 --- a/posthog/temporal/ai/anomaly_investigation/workflow.py +++ b/posthog/temporal/ai/anomaly_investigation/workflow.py @@ -171,29 +171,12 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput Team.objects.aget(id=inputs.team_id), AlertConfiguration.objects.select_related("insight").aget(id=inputs.alert_id), ) - subject: AlertCheck | InsightAlertInvestigation - if inputs.investigation_id is not None: - subject = await sync_to_async(_load_investigation, thread_sensitive=False)( - inputs.team_id, inputs.investigation_id - ) - if subject.status == InvestigationStatus.DONE: - # A retry after the verdict was recorded repeats only its release, not the agent. - return list((await _finish_platform_investigation(alert, subject)).deliveries) - evaluated = EvaluatedCheck.of_investigation(subject) - elif inputs.alert_check_id is not None: - subject = await AlertCheck.objects.aget(id=inputs.alert_check_id) - evaluated = EvaluatedCheck.of_alert_check(subject) - else: - raise ValueError("An investigation needs an alert check or an investigation id") + subject, evaluated = await _load_subject(inputs) + if isinstance(subject, InsightAlertInvestigation) and subject.status == InvestigationStatus.DONE: + # A retry after the verdict was recorded repeats only its release, not the agent. + return list((await _finish_platform_investigation(alert, subject)).deliveries) - user: User | None = None - if inputs.user_id is not None: - try: - user = await User.objects.aget(id=inputs.user_id) - except User.DoesNotExist: - user = None - if user is None: - user = await _pick_investigation_user(alert) + user = await _investigation_user(inputs, alert) if user is None: await _mark_failed(subject, "No user available to run the investigation agent.") @@ -398,10 +381,34 @@ async def _finish_platform_investigation( ) +async def _load_subject( + inputs: AnomalyInvestigationWorkflowInputs, +) -> tuple[AlertCheck | InsightAlertInvestigation, EvaluatedCheck]: + """The record the investigation writes to, and what its check measured.""" + if inputs.investigation_id is not None: + investigation = await sync_to_async(_load_investigation, thread_sensitive=False)( + inputs.team_id, inputs.investigation_id + ) + return investigation, EvaluatedCheck.of_investigation(investigation) + if inputs.alert_check_id is not None: + alert_check = await AlertCheck.objects.aget(id=inputs.alert_check_id) + return alert_check, EvaluatedCheck.of_alert_check(alert_check) + raise ValueError("An investigation needs an alert check or an investigation id") + + def _load_investigation(team_id: int, investigation_id: UUID) -> InsightAlertInvestigation: return InsightAlertInvestigation.objects.for_team(team_id).get(id=investigation_id) +async def _investigation_user(inputs: AnomalyInvestigationWorkflowInputs, alert: AlertConfiguration) -> User | None: + if inputs.user_id is not None: + try: + return await User.objects.aget(id=inputs.user_id) + except User.DoesNotExist: + pass + return await _pick_investigation_user(alert) + + def _episode_source_id(investigation: InsightAlertInvestigation) -> str: """One identity for every investigation of one firing, as the insight stack keys a signal on its episode's first check. The firing that was open at takeover has no start, and is one episode.""" From e5c4687aaca39b45a43579f27225eccb2601dfe0 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 17:54:33 -0800 Subject: [PATCH 3/6] fix(alerts): reach the investigation write-back through the alerts facade The agent imported products.alerts internals that tach does not expose to core, which failed Repo checks. It now imports facade.investigations, kept apart from facade.api to avoid an import cycle with the insight stack's task helpers. The two dataclasses it touched now state frozen= explicitly. Co-Authored-By: Claude Opus 5.5 --- .../ai/anomaly_investigation/notebook.py | 4 ++-- .../ai/anomaly_investigation/workflow.py | 21 +++++++------------ products/alerts/backend/facade/contracts.py | 12 ++++++++++- .../alerts/backend/facade/investigations.py | 18 ++++++++++++++++ .../alerts/backend/investigation_release.py | 12 +---------- 5 files changed, 40 insertions(+), 27 deletions(-) create mode 100644 products/alerts/backend/facade/investigations.py diff --git a/posthog/temporal/ai/anomaly_investigation/notebook.py b/posthog/temporal/ai/anomaly_investigation/notebook.py index 1e3fb81b3218..ef4e5540855b 100644 --- a/posthog/temporal/ai/anomaly_investigation/notebook.py +++ b/posthog/temporal/ai/anomaly_investigation/notebook.py @@ -1,10 +1,10 @@ import json import uuid -from dataclasses import dataclass from markdown_it import MarkdownIt from markdown_it.token import Token +from posthog.dataclasses import frozen from posthog.temporal.ai.anomaly_investigation.evaluated import EvaluatedCheck from posthog.temporal.ai.anomaly_investigation.report import InvestigationReport @@ -28,7 +28,7 @@ } -@dataclass +@frozen class NotebookRenderContext: alert: AlertConfiguration evaluated: EvaluatedCheck diff --git a/posthog/temporal/ai/anomaly_investigation/workflow.py b/posthog/temporal/ai/anomaly_investigation/workflow.py index 94e59baa9a50..51c915c7b1ce 100644 --- a/posthog/temporal/ai/anomaly_investigation/workflow.py +++ b/posthog/temporal/ai/anomaly_investigation/workflow.py @@ -44,14 +44,9 @@ from posthog.temporal.common.heartbeat import Heartbeater from posthog.utils import absolute_uri +from products.alerts.backend.facade import investigations as alerts +from products.alerts.backend.facade.contracts import InvestigationRelease from products.alerts.backend.investigation_episode import EpisodeInvestigations, episode_investigations -from products.alerts.backend.investigation_release import ( - InvestigationRelease, - finish_investigation, - mark_investigation_failed, - mark_investigation_running, - verdict_change_is_news, -) from products.alerts.backend.models.alert import AlertCheck, AlertConfiguration, InvestigationStatus from products.alerts.backend.models.investigation import InsightAlertInvestigation from products.alerts_platform.backend.facade import api as platform @@ -103,7 +98,7 @@ _MAX_DESCRIPTION_CHARS = 3000 -@dataclass +@dataclass(frozen=False) class AnomalyInvestigationWorkflowInputs: team_id: int alert_id: UUID @@ -261,7 +256,7 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput summary = result.report.summary.strip() deliveries: list[AlertDeliveryRequest] = [] if isinstance(subject, InsightAlertInvestigation): - release = await sync_to_async(finish_investigation, thread_sensitive=False)( + release = await sync_to_async(alerts.finish_investigation, thread_sensitive=False)( alert, subject, verdict=result.report.verdict, summary=summary or None, notebook_id=notebook.id ) deliveries = list(release.deliveries) @@ -372,7 +367,7 @@ async def _deliver_alert_check_outcome( async def _finish_platform_investigation( alert: AlertConfiguration, investigation: InsightAlertInvestigation ) -> InvestigationRelease: - return await sync_to_async(finish_investigation, thread_sensitive=False)( + return await sync_to_async(alerts.finish_investigation, thread_sensitive=False)( alert, investigation, verdict=investigation.verdict, @@ -695,7 +690,7 @@ def _dispatch_verdict_change_followup( Never raises into the activity: a correction that cannot be sent is not worth a second agent run. """ - if not verdict_change_is_news(verdict, previous_verdict, alert.investigation_inconclusive_action): + if not alerts.verdict_change_is_news(verdict, previous_verdict, alert.investigation_inconclusive_action): return receipts = check.targets_notified or {} if receipts.get(_VERDICT_CHANGE_FOLLOWUP_KEY): @@ -818,7 +813,7 @@ def _build_breach_descriptions( async def _mark_running(subject: AlertCheck | InsightAlertInvestigation) -> None: if isinstance(subject, InsightAlertInvestigation): - await sync_to_async(mark_investigation_running, thread_sensitive=False)(subject) + await sync_to_async(alerts.mark_investigation_running, thread_sensitive=False)(subject) return await sync_to_async(AlertCheck.objects.filter(id=subject.id).update, thread_sensitive=False)( investigation_status=InvestigationStatus.RUNNING, @@ -827,7 +822,7 @@ async def _mark_running(subject: AlertCheck | InsightAlertInvestigation) -> None async def _mark_failed(subject: AlertCheck | InsightAlertInvestigation, reason: str) -> None: if isinstance(subject, InsightAlertInvestigation): - await sync_to_async(mark_investigation_failed, thread_sensitive=False)(subject, reason) + await sync_to_async(alerts.mark_investigation_failed, thread_sensitive=False)(subject, reason) return await sync_to_async(AlertCheck.objects.filter(id=subject.id).update, thread_sensitive=False)( investigation_status=InvestigationStatus.FAILED, diff --git a/products/alerts/backend/facade/contracts.py b/products/alerts/backend/facade/contracts.py index 29c355e927e3..b9185ce1b555 100644 --- a/products/alerts/backend/facade/contracts.py +++ b/products/alerts/backend/facade/contracts.py @@ -15,7 +15,7 @@ from posthog.dataclasses import frozen from posthog.schema_enums import AlertState -from products.alerts_platform.backend.facade.contracts import DeliveryPath, DestinationType +from products.alerts_platform.backend.facade.contracts import AlertDeliveryRequest, DeliveryPath, DestinationType from products.cdp.backend.internal_events import LEGACY_INSIGHT_ALERT_EVENT # The internal event an insight alert's destinations fire on. Do not take it from @@ -193,3 +193,13 @@ class MovedAlertCheckPage: class PlatformCannotOwn(ValueError): """An edit that would leave an alert the platform owns in a shape the platform cannot evaluate.""" + + +@frozen +class InvestigationRelease: + """What a finished investigation of a platform evaluation hands back to the agent's workflow.""" + + # What the workflow starts: the held firing when the verdict let it go out, or the + # announcement of a changed verdict on a firing that already went out. + deliveries: tuple[AlertDeliveryRequest, ...] + previous_verdict: str | None diff --git a/products/alerts/backend/facade/investigations.py b/products/alerts/backend/facade/investigations.py new file mode 100644 index 000000000000..a42bac6e57dc --- /dev/null +++ b/products/alerts/backend/facade/investigations.py @@ -0,0 +1,18 @@ +"""What the investigation agent writes back for a platform evaluation of an insight alert. + +Apart from `facade.api`, which the insight stack's task helpers import, because these import them. +""" + +from products.alerts.backend.investigation_release import ( + finish_investigation, + mark_investigation_failed, + mark_investigation_running, + verdict_change_is_news, +) + +__all__ = [ + "finish_investigation", + "mark_investigation_failed", + "mark_investigation_running", + "verdict_change_is_news", +] diff --git a/products/alerts/backend/investigation_release.py b/products/alerts/backend/investigation_release.py index cf4c6883bea0..cdbb7503a789 100644 --- a/products/alerts/backend/investigation_release.py +++ b/products/alerts/backend/investigation_release.py @@ -13,13 +13,13 @@ import structlog -from posthog.dataclasses import frozen from posthog.tasks.alerts.investigation_notifications import ( INVESTIGATION_NOTIFY_GRACE_MINUTES, INVESTIGATION_RUNNING_GRACE_MINUTES, ) from posthog.tasks.alerts.utils import _inconclusive_is_suppressed, _should_suppress_notification +from products.alerts.backend.facade.contracts import InvestigationRelease from products.alerts.backend.models.alert import AlertConfiguration, InvestigationStatus from products.alerts.backend.models.investigation import InsightAlertInvestigation from products.alerts_platform.backend.facade import api as platform @@ -81,16 +81,6 @@ def release_held( ) -@frozen -class InvestigationRelease: - """What a finished investigation hands back to the workflow that ran it.""" - - # What the workflow starts: the held firing when the verdict let it go out, or the - # announcement of a changed verdict on a firing that already went out. - deliveries: tuple[AlertDeliveryRequest, ...] - previous_verdict: str | None - - def _hold_of(investigation: InsightAlertInvestigation) -> HeldDeliveryAddress: return HeldDeliveryAddress( team_id=investigation.team_id, From b9f3666945fb2fe98d91fb3f74b3c17262075768 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 18:42:55 -0800 Subject: [PATCH 4/6] fix(alerts): take ids in the investigation facade product:lint requires a facade to take ids and contracts rather than models, so the agent's write-back functions load the alert and investigation rows themselves. Co-Authored-By: Claude Opus 5.5 --- .../ai/anomaly_investigation/workflow.py | 18 ++++++-- .../alerts/backend/facade/investigations.py | 46 ++++++++++++++++--- 2 files changed, 53 insertions(+), 11 deletions(-) diff --git a/posthog/temporal/ai/anomaly_investigation/workflow.py b/posthog/temporal/ai/anomaly_investigation/workflow.py index 51c915c7b1ce..5ce2a520c0db 100644 --- a/posthog/temporal/ai/anomaly_investigation/workflow.py +++ b/posthog/temporal/ai/anomaly_investigation/workflow.py @@ -257,7 +257,12 @@ async def investigate_anomaly_activity(inputs: AnomalyInvestigationWorkflowInput deliveries: list[AlertDeliveryRequest] = [] if isinstance(subject, InsightAlertInvestigation): release = await sync_to_async(alerts.finish_investigation, thread_sensitive=False)( - alert, subject, verdict=result.report.verdict, summary=summary or None, notebook_id=notebook.id + subject.team_id, + alert.id, + subject.id, + verdict=result.report.verdict, + summary=summary or None, + notebook_id=notebook.id, ) deliveries = list(release.deliveries) previous_verdict = release.previous_verdict @@ -368,8 +373,9 @@ async def _finish_platform_investigation( alert: AlertConfiguration, investigation: InsightAlertInvestigation ) -> InvestigationRelease: return await sync_to_async(alerts.finish_investigation, thread_sensitive=False)( - alert, - investigation, + investigation.team_id, + alert.id, + investigation.id, verdict=investigation.verdict, summary=investigation.summary, notebook_id=investigation.notebook_id, @@ -813,7 +819,7 @@ def _build_breach_descriptions( async def _mark_running(subject: AlertCheck | InsightAlertInvestigation) -> None: if isinstance(subject, InsightAlertInvestigation): - await sync_to_async(alerts.mark_investigation_running, thread_sensitive=False)(subject) + await sync_to_async(alerts.mark_investigation_running, thread_sensitive=False)(subject.team_id, subject.id) return await sync_to_async(AlertCheck.objects.filter(id=subject.id).update, thread_sensitive=False)( investigation_status=InvestigationStatus.RUNNING, @@ -822,7 +828,9 @@ async def _mark_running(subject: AlertCheck | InsightAlertInvestigation) -> None async def _mark_failed(subject: AlertCheck | InsightAlertInvestigation, reason: str) -> None: if isinstance(subject, InsightAlertInvestigation): - await sync_to_async(alerts.mark_investigation_failed, thread_sensitive=False)(subject, reason) + await sync_to_async(alerts.mark_investigation_failed, thread_sensitive=False)( + subject.team_id, subject.id, reason + ) return await sync_to_async(AlertCheck.objects.filter(id=subject.id).update, thread_sensitive=False)( investigation_status=InvestigationStatus.FAILED, diff --git a/products/alerts/backend/facade/investigations.py b/products/alerts/backend/facade/investigations.py index a42bac6e57dc..67131f3e0046 100644 --- a/products/alerts/backend/facade/investigations.py +++ b/products/alerts/backend/facade/investigations.py @@ -3,12 +3,13 @@ Apart from `facade.api`, which the insight stack's task helpers import, because these import them. """ -from products.alerts.backend.investigation_release import ( - finish_investigation, - mark_investigation_failed, - mark_investigation_running, - verdict_change_is_news, -) +from uuid import UUID + +from products.alerts.backend import investigation_release +from products.alerts.backend.facade.contracts import InvestigationRelease +from products.alerts.backend.investigation_release import verdict_change_is_news +from products.alerts.backend.models.alert import AlertConfiguration +from products.alerts.backend.models.investigation import InsightAlertInvestigation __all__ = [ "finish_investigation", @@ -16,3 +17,36 @@ "mark_investigation_running", "verdict_change_is_news", ] + + +def _investigation(team_id: int, investigation_id: UUID) -> InsightAlertInvestigation: + return InsightAlertInvestigation.objects.for_team(team_id).get(id=investigation_id) + + +def mark_investigation_running(team_id: int, investigation_id: UUID) -> None: + """Marks the investigation as running and holds its firing for the running grace.""" + investigation_release.mark_investigation_running(_investigation(team_id, investigation_id)) + + +def mark_investigation_failed(team_id: int, investigation_id: UUID, reason: str) -> None: + """Records the investigation's failure and leaves its firing to the platform's safety net.""" + investigation_release.mark_investigation_failed(_investigation(team_id, investigation_id), reason) + + +def finish_investigation( + team_id: int, + alert_id: UUID, + investigation_id: UUID, + *, + verdict: str | None, + summary: str | None, + notebook_id: UUID | None, +) -> InvestigationRelease: + """Records the investigation's verdict and releases its held firing by it. Safe to repeat.""" + return investigation_release.finish_investigation( + AlertConfiguration.objects.get(team_id=team_id, id=alert_id), + _investigation(team_id, investigation_id), + verdict=verdict, + summary=summary, + notebook_id=notebook_id, + ) From f693a2a0abfd8ca11aa77cb926bfe7e3bfa309a5 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 18:50:32 -0800 Subject: [PATCH 5/6] fix(alerts): suppress a held firing whose alert was disabled mid-investigation A verdict on an alert that was switched off while the agent ran now suppresses the held firing, as the safety net does. The hold test door returns a dataclass instead of a tuple. Co-Authored-By: Claude Opus 5.5 --- products/alerts/backend/investigation_release.py | 3 ++- .../backend/tests/test_platform_investigation.py | 14 ++++++++++---- .../alerts_platform/backend/facade/testing.py | 15 +++++++++++++-- 3 files changed, 25 insertions(+), 7 deletions(-) diff --git a/products/alerts/backend/investigation_release.py b/products/alerts/backend/investigation_release.py index cdbb7503a789..6a9b00198cec 100644 --- a/products/alerts/backend/investigation_release.py +++ b/products/alerts/backend/investigation_release.py @@ -138,7 +138,8 @@ def finish_investigation( platform.reschedule_hold(_hold_of(investigation), release_after=datetime.now(UTC) + TERMINAL_HOLD) delivery = release_held( _hold_of(investigation), - suppress=_should_suppress_notification(verdict, alert.investigation_inconclusive_action), + # As the safety net does, so turning the alert back on never sends a firing that old. + suppress=not alert.enabled or _should_suppress_notification(verdict, alert.investigation_inconclusive_action), released_by=HoldReleasedBy.VERDICT, ) if delivery is None: diff --git a/products/alerts/backend/tests/test_platform_investigation.py b/products/alerts/backend/tests/test_platform_investigation.py index 1a78c9f9b8d1..38cbe1189a4b 100644 --- a/products/alerts/backend/tests/test_platform_investigation.py +++ b/products/alerts/backend/tests/test_platform_investigation.py @@ -219,11 +219,16 @@ def _finish(self, investigation: InsightAlertInvestigation, verdict: str) -> Inv @parameterized.expand( [ - ("a true positive sends", "true_positive", HoldRelease.SEND), - ("a false positive suppresses", "false_positive", HoldRelease.SUPPRESS), + ("a true positive sends", "true_positive", True, HoldRelease.SEND), + ("a false positive suppresses", "false_positive", True, HoldRelease.SUPPRESS), + ("a disabled alert suppresses", "true_positive", False, HoldRelease.SUPPRESS), ] ) - def test_a_verdict_releases_the_held_firing(self, _name: str, verdict: str, action: HoldRelease) -> None: + def test_a_verdict_releases_the_held_firing( + self, _name: str, verdict: str, enabled: bool, action: HoldRelease + ) -> None: + self.alert.enabled = enabled + self.alert.save() self._hold() investigation = self._row("slot:now", status=InvestigationStatus.RUNNING) @@ -232,7 +237,8 @@ def test_a_verdict_releases_the_held_firing(self, _name: str, verdict: str, acti expected = (self.request,) if action == HoldRelease.SEND else () assert release.deliveries == repeated.deliveries == expected - assert platform_testing.hold_release(self.address)[:2] == (action, HoldReleasedBy.VERDICT) + hold = platform_testing.hold_release(self.address) + assert (hold.release_action, hold.released_by) == (action, HoldReleasedBy.VERDICT) investigation.refresh_from_db() assert (investigation.status, investigation.verdict) == (InvestigationStatus.DONE, verdict) diff --git a/products/alerts_platform/backend/facade/testing.py b/products/alerts_platform/backend/facade/testing.py index 72fd0080db5c..62d0e5d4d235 100644 --- a/products/alerts_platform/backend/facade/testing.py +++ b/products/alerts_platform/backend/facade/testing.py @@ -13,6 +13,8 @@ from typing import Any from uuid import UUID +from posthog.dataclasses import frozen + from products.alerts_platform.backend.delivery.evaluation import deliver_evaluation from products.alerts_platform.backend.facade.contracts import ( AlertDeliveryRequest, @@ -91,14 +93,23 @@ def hold_delivery( ) -def hold_release(address: HeldDeliveryAddress) -> tuple[str | None, str | None, datetime]: +@frozen +class HoldSnapshot: + release_action: str | None + released_by: str | None + release_after: datetime + + +def hold_release(address: HeldDeliveryAddress) -> HoldSnapshot: """A hold's release action, who released it, and its deadline.""" hold = ( PlatformAlertDeliveryHold.objects.for_team(address.team_id) .filter(configuration_id=address.configuration_id, evaluation_key=address.evaluation_key) .get() ) - return hold.release_action, hold.released_by, hold.release_after + return HoldSnapshot( + release_action=hold.release_action, released_by=hold.released_by, release_after=hold.release_after + ) def set_due_at(configuration_id: UUID, due_at: datetime) -> None: From 77c38d81949989cf1a1941a8eb022dc2598e71bd Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 2 Nov 2026 19:26:28 -0800 Subject: [PATCH 6/6] fix(alerts): release a held firing in the verdict's own transaction A worker that died between recording a false positive and releasing the hold left the safety net to send it. The release now commits with the verdict. Co-Authored-By: Claude Opus 5.5 --- .../alerts/backend/investigation_release.py | 22 +++++++++---------- 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/products/alerts/backend/investigation_release.py b/products/alerts/backend/investigation_release.py index 6a9b00198cec..22d29b596a50 100644 --- a/products/alerts/backend/investigation_release.py +++ b/products/alerts/backend/investigation_release.py @@ -117,11 +117,9 @@ def finish_investigation( summary: str | None, notebook_id: UUID | None, ) -> InvestigationRelease: - """Records the verdict, then releases the firing's held delivery by it. - - The terminal grace starts in the transaction that records the verdict, so the safety net cannot - send the firing between the two. A firing that already went out gets a verdict-changed - announcement when its verdict contradicts the firing's previous one. + """Records the verdict and releases the firing's held delivery by it, in one transaction, so the + safety net can never find a recorded verdict with its firing still held. A firing that already + went out gets a verdict-changed announcement when its verdict contradicts the firing's previous one. Safe to repeat for a recorded verdict: a retry gets back what the first call released. """ @@ -135,13 +133,13 @@ def finish_investigation( InsightAlertInvestigation.objects.for_team(investigation.team_id).filter(id=investigation.id).update( status=InvestigationStatus.DONE, verdict=verdict, summary=summary, notebook_id=notebook_id, error=None ) - platform.reschedule_hold(_hold_of(investigation), release_after=datetime.now(UTC) + TERMINAL_HOLD) - delivery = release_held( - _hold_of(investigation), - # As the safety net does, so turning the alert back on never sends a firing that old. - suppress=not alert.enabled or _should_suppress_notification(verdict, alert.investigation_inconclusive_action), - released_by=HoldReleasedBy.VERDICT, - ) + delivery = release_held( + _hold_of(investigation), + # As the safety net does, so turning the alert back on never sends a firing that old. + suppress=not alert.enabled + or _should_suppress_notification(verdict, alert.investigation_inconclusive_action), + released_by=HoldReleasedBy.VERDICT, + ) if delivery is None: delivery = _announce_verdict_change(alert, investigation, verdict=verdict, previous_verdict=previous_verdict) return InvestigationRelease(deliveries=(delivery,) if delivery else (), previous_verdict=previous_verdict)