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..ef4e5540855b 100644 --- a/posthog/temporal/ai/anomaly_investigation/notebook.py +++ b/posthog/temporal/ai/anomaly_investigation/notebook.py @@ -1,13 +1,14 @@ 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 -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, @@ -27,10 +28,10 @@ } -@dataclass +@frozen 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 159e25c78638..58fb777ee2cc 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 @@ -25,12 +26,12 @@ 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 ( - _inconclusive_is_suppressed, _should_suppress_notification, dispatch_alert_notification, 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 @@ -41,10 +42,17 @@ 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.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 @@ -70,6 +78,9 @@ ANOMALY_INVESTIGATION_ACTIVITY_HEARTBEAT_TIMEOUT = 5 * 60 # 5 minutes ANOMALY_INVESTIGATION_ACTIVITY_MAX_ATTEMPTS = 2 +# 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" @@ -79,18 +90,21 @@ _MAX_DESCRIPTION_CHARS = 3000 -@dataclass +@dataclass(frozen=False) 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: @@ -99,7 +113,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), @@ -110,35 +124,57 @@ 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, 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(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. @@ -153,10 +189,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 @@ -170,8 +206,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, ) @@ -187,13 +223,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, ) @@ -209,6 +245,79 @@ 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(alerts.finish_investigation, thread_sensitive=False)( + 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 + 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 @@ -217,18 +326,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, ) @@ -244,46 +352,68 @@ 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(alerts.finish_investigation, thread_sensitive=False)( + investigation.team_id, + alert.id, + investigation.id, + 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), +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.""" + 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: @@ -317,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, @@ -329,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, @@ -341,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 @@ -352,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, @@ -361,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. """ @@ -370,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, @@ -554,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 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): @@ -642,30 +774,38 @@ def _build_breach_descriptions( ) -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(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, ) -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(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, 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 @@ -675,9 +815,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/facade/contracts.py b/products/alerts/backend/facade/contracts.py index 50b8222feabf..e74abd41cf03 100644 --- a/products/alerts/backend/facade/contracts.py +++ b/products/alerts/backend/facade/contracts.py @@ -15,7 +15,12 @@ from posthog.dataclasses import frozen from posthog.schema_enums import AlertState -from products.alerts_platform.backend.facade.contracts import DeliveryPath, DestinationType, IncidentAction +from products.alerts_platform.backend.facade.contracts import ( + AlertDeliveryRequest, + DeliveryPath, + DestinationType, + IncidentAction, +) 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 @@ -208,3 +213,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..67131f3e0046 --- /dev/null +++ b/products/alerts/backend/facade/investigations.py @@ -0,0 +1,52 @@ +"""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 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", + "mark_investigation_failed", + "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, + ) diff --git a/products/alerts/backend/investigation_release.py b/products/alerts/backend/investigation_release.py new file mode 100644 index 000000000000..22d29b596a50 --- /dev/null +++ b/products/alerts/backend/investigation_release.py @@ -0,0 +1,171 @@ +"""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.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 +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), + ) + + +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 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. + """ + 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 + ) + 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) + + +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 9732f45cd497..3911a17a596a 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 @@ -66,18 +67,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..38cbe1189a4b 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,106 @@ 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", 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, enabled: bool, action: HoldRelease + ) -> None: + self.alert.enabled = enabled + self.alert.save() + 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 + 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) + + @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 73a51c35991d..fa18022f4591 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 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.""" diff --git a/products/alerts_platform/backend/facade/testing.py b/products/alerts_platform/backend/facade/testing.py index d2d565dd84a9..62d0e5d4d235 100644 --- a/products/alerts_platform/backend/facade/testing.py +++ b/products/alerts_platform/backend/facade/testing.py @@ -13,10 +13,13 @@ 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, DeliveryPath, + HeldDeliveryAddress, PlatformAlertSnapshot, PlatformConfigurationSnapshot, SourceCorrespondence, @@ -68,6 +71,47 @@ 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, + ) + + +@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 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: """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 cef24c91c136..576ed878d4b8 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; } /**