feat(seer): Return triggering monitor on workflow runs

src/sentry/apidocs/examples/seer_workflow_examples.py

@@ -9,6 +9,16 @@
                     "id": "101",
                     "configId": "10",
                     "strategy": "breached_metric_investigation",
+                    "trigger": {
+                        "type": "breached_metric",
+                        "detectorId": "401",
+                        "environment": "production",
+                        "eventId": "9f2d4c1b8a7e4f6d9c3b2a1e0f9d8c7b",
+                        "groupId": "501",
+                        "observedAt": "2025-01-01T09:59:30Z",
+                        "occurrenceId": "3b1e6f0a-2c4d-4e8f-9a1b-7c6d5e4f3a2b",
+                        "projectId": "42",
+                    },
                     "status": "complete",
                     "dateAdded": "2025-01-01T10:00:00Z",
                     "dateUpdated": "2025-01-01T10:02:00Z",
@@ -44,6 +54,7 @@
                     "id": "102",
                     "configId": "10",
                     "strategy": "breached_metric_investigation",
+                    "trigger": None,
                     "status": "failed",
                     "dateAdded": "2025-01-02T10:00:00Z",
                     "dateUpdated": "2025-01-02T10:00:20Z",

src/sentry/seer/endpoints/organization_seer_workflow_runs.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
+import logging
 from collections.abc import Sequence
 from datetime import datetime
 from typing import Literal, TypedDict
 
 from django.db.models import F, Prefetch, prefetch_related_objects
 from drf_spectacular.utils import extend_schema
+from pydantic import ValidationError
 from rest_framework import serializers
 from rest_framework.exceptions import NotFound
 from rest_framework.request import Request
@@ -29,13 +31,17 @@
     SeerWorkflowExecutionStatus,
     SeerWorkflowRun,
     SeerWorkflowRunExecution,
+    SeerWorkflowRunExtras,
     SeerWorkflowRunStatus,
     SeerWorkflowStrategy,
 )
 from sentry.seer.monitor_cleanup.runs import create_monitor_cleanup_run
+from sentry.seer.workflows.breached_metrics import MetricBreachTrigger
 from sentry.seer_workflow_outputs.models import SeerWorkflowOutputLink
 from sentry.types.ratelimit import RateLimit, RateLimitCategory
 
+logger = logging.getLogger(__name__)
+
 
 class WorkflowRunCreateSerializer(serializers.Serializer):
     strategy = serializers.ChoiceField(choices=[SeerWorkflowStrategy.DUPLICATE_MONITORS])
@@ -99,10 +105,26 @@ class SeerWorkflowExecutionErrorResponse(SeerWorkflowExecutionResponseBase):
 )
 
 
+class MetricBreachTriggerResponse(TypedDict):
+    type: Literal["breached_metric"]
+    detectorId: str
+    environment: str | None
+    eventId: str
+    groupId: str
+    observedAt: datetime
+    occurrenceId: str
+    projectId: str
+
+
+# Discriminated by `type`; add new trigger variants to this union.
+TriggerResponse = MetricBreachTriggerResponse
+
+
 class SeerWorkflowRunResponseBase(TypedDict):
     id: str
     configId: str
     strategy: str
+    trigger: TriggerResponse | None
     dateAdded: datetime
     dateUpdated: datetime
     dateCompleted: datetime | None
@@ -185,7 +207,9 @@ def get(self, request: Request, organization: Organization) -> Response:
           and each failed execution carries its own ``error``.
 
         Missing error details use code ``unknown``. Historical rows without lifecycle
-        tracking retain a null status. ``seerRunId`` is the linked Seer UUID or null.
+        tracking retain a null status. ``trigger`` describes the metric breach that started
+        the run (detector, event, occurrence, issue, and project), or is null for runs not
+        started by a metric monitor. The detector may have since been deleted. ``seerRunId`` is the linked Seer UUID or null.
         Each execution's ``outputs`` lists linked investigations by type, ID, and title
         when the viewer has investigation access, including archived investigations.
         The same investigation can be linked to multiple executions.
@@ -269,6 +293,7 @@ def serialize_workflow_run(run: SeerWorkflowRun) -> SeerWorkflowRunResponse:
         "id": str(run.id),
         "configId": str(run.workflow_config_id),
         "strategy": getattr(run, "strategy"),
+        "trigger": serialize_trigger(run),
         "dateAdded": run.date_added,
         "dateUpdated": run.date_updated,
         "dateCompleted": run.date_completed,
@@ -288,6 +313,28 @@ def serialize_workflow_run(run: SeerWorkflowRun) -> SeerWorkflowRunResponse:
     return {**result, "status": status}
 
 
+def serialize_trigger(run: SeerWorkflowRun) -> TriggerResponse | None:
+    if not run.extras.get("snapshot"):
+        return None
+    try:
+        snapshot = SeerWorkflowRunExtras.parse_obj(run.extras).snapshot
+        if snapshot.strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION:
+            trigger = MetricBreachTrigger.parse_obj(snapshot.inputs)
+            return {
+                "type": "breached_metric",
+                "detectorId": str(trigger.detector_id),
+                "environment": trigger.environment,
+                "eventId": trigger.event_id,
+                "groupId": str(trigger.group_id),
+                "observedAt": trigger.observed_at,
+                "occurrenceId": trigger.occurrence_id,
+                "projectId": str(trigger.project_id),
+            }
+    except ValidationError:
+        logger.exception("seer.workflow_run.invalid_trigger", extra={"run_id": run.id})
+    return None
+
+
 def serialize_workflow_execution(
     execution: SeerWorkflowRunExecution, *, organization_id: int
 ) -> SeerWorkflowExecutionResponse:

tests/sentry/seer/endpoints/test_organization_seer_workflow_runs.py

@@ -1,4 +1,5 @@
 from collections.abc import Sequence
+from datetime import UTC, datetime
 from typing import Any
 from unittest.mock import patch
 
@@ -87,6 +88,7 @@ def test_returns_run_lifecycle_and_all_executions(self) -> None:
                     "id": str(run.id),
                     "configId": str(self.config.id),
                     "strategy": "breached_metric_investigation",
+                    "trigger": None,
                     "status": "running",
                     "dateAdded": run.date_added,
                     "dateUpdated": run.date_updated,
@@ -479,8 +481,67 @@ def test_omits_outputs_without_open_membership(self) -> None:
         assert response.data["runs"][0]["executions"][0]["outputs"] == []
         assert investigation.title not in response.content.decode()
 
+    def test_returns_the_triggering_monitor(self) -> None:
+        detector = self.create_detector(project=self.project)
+        run = self.create_run(
+            inputs={
+                "event_id": "a" * 32,
+                "occurrence_id": "abc",
+                "detector_id": detector.id,
+                "evidence": {},
+                "observed_at": "2025-01-01T10:00:00Z",
+                "environment": "production",
+                "group_id": 1,
+                "project_id": self.project.id,
+                "binding_id": 1,
+                "workflow_id": 1,
+                "action_id": 1,
+            }
+        )
+
+        response = self.get_success_response(self.organization.slug)
+
+        (serialized_run,) = response.data["runs"]
+        assert serialized_run["id"] == str(run.id)
+        assert serialized_run["trigger"] == {
+            "type": "breached_metric",
+            "detectorId": str(detector.id),
+            "environment": "production",
+            "eventId": "a" * 32,
+            "groupId": "1",
+            "observedAt": datetime(2025, 1, 1, 10, tzinfo=UTC),
+            "occurrenceId": "abc",
+            "projectId": str(self.project.id),
+        }
+
+    def test_trigger_is_null_for_other_strategies(self) -> None:
+        self.config.update(strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2)
+        self.create_run()
+
+        response = self.get_success_response(self.organization.slug)
+
+        assert response.data["runs"][0]["trigger"] is None
+
+    def test_trigger_is_null_without_a_snapshot(self) -> None:
+        self.create_run(extras={})
+
+        response = self.get_success_response(self.organization.slug)
+
+        assert response.data["runs"][0]["trigger"] is None
+
+    def test_trigger_is_null_for_invalid_inputs(self) -> None:
+        self.create_run(inputs={"detector_id": "not-an-id"})
+
+        response = self.get_success_response(self.organization.slug)
+
+        assert response.data["runs"][0]["trigger"] is None
+
     def create_run(
-        self, *, project_ids: Sequence[int] | None = None, **kwargs: object
+        self,
+        *,
+        project_ids: Sequence[int] | None = None,
+        inputs: dict[str, object] | None = None,
+        **kwargs: object,
     ) -> SeerWorkflowRun:
         kwargs.setdefault("status", SeerWorkflowRunStatus.QUEUED)
         kwargs.setdefault(
@@ -494,7 +555,7 @@ def create_run(
                     if project_ids is not None
                     else [self.project.id],
                     "output_policy": self.config.output_policy,
-                    "inputs": {},
+                    "inputs": inputs or {},
                 },
                 "error": None,
             },