ref(seer): Compose automations from a trigger and an action

src/sentry/seer/apps.py

@@ -7,5 +7,5 @@ class SeerConfig(AppConfig):
     def ready(self) -> None:
         # Register all the entrypoints for the operator to use
         import sentry.seer.entrypoints.slack.entrypoint  # noqa: F401
-        import sentry.seer.workflows.action  # noqa: F401
+        import sentry.seer.workflows.breached_metrics  # noqa: F401
         import sentry.seer.workflows.scheduled  # noqa: F401

src/sentry/seer/endpoints/admin_metric_breaches.py

@@ -38,7 +38,7 @@
 from sentry.models.groupopenperiod import should_create_open_periods
 from sentry.models.organization import Organization
 from sentry.seer.models.workflow import SeerWorkflowConfig, SeerWorkflowStrategy
-from sentry.seer.workflows.breached_metrics import BreachedMetricWorkflow
+from sentry.seer.workflows.workflow import SeerWorkflow
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QuerySubscription
 from sentry.workflow_engine.models import DataCondition, DataPacket, DataSourceDetector, Detector
@@ -128,7 +128,7 @@ def prepare_synthetic_breach(
     )
     environment = query.environment.name if query.environment else None
     if not any(
-        BreachedMetricWorkflow(config_id=config.id, organization_id=organization.id).matches_scope(
+        SeerWorkflow(config_id=config.id, organization_id=organization.id).matches_scope(
             detector.project_id, environment
         )
         for config in configs

src/sentry/seer/endpoints/organization_seer_workflow_configs.py

@@ -39,21 +39,19 @@
     SeerWorkflowSchedule,
     SeerWorkflowStrategy,
 )
+from sentry.seer.workflows.base import (
+    AllProjects,
+    SelectedMonitors,
+    SelectedProjects,
+    WorkflowUpdateError,
+)
 from sentry.seer.workflows.breached_metrics import (
-    BreachedMetricConfigUpdate,
-    BreachedMetricWorkflow,
     ConnectedMonitor,
     get_config_monitors_bulk,
     get_monitor_project_ids,
 )
-from sentry.seer.workflows.registry import get_seer_workflow
-from sentry.seer.workflows.workflow import (
-    AllProjects,
-    SeerServiceAccountGrant,
-    SeerWorkflow,
-    SelectedProjects,
-    WorkflowUpdateError,
-)
+from sentry.seer.workflows.definitions import get_definition
+from sentry.seer.workflows.workflow import SeerServiceAccountGrant, SeerWorkflow
 from sentry.service_accounts.models import (
     GRANTABLE_SCOPES,
     ServiceAccountProjectReplica,
@@ -426,7 +424,7 @@ def post(self, request: Request, organization: Organization) -> Response:
 
         project_ids: list[int] = data.get("projectIds") or []
         monitor_ids: list[int] = data.get("monitorIds") or []
-        if strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION:
+        if strategy is SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION:
             denied = _check_project_access(
                 get_monitor_project_ids(monitor_ids, organization),
                 request=request,
@@ -447,27 +445,33 @@ def post(self, request: Request, organization: Organization) -> Response:
             created_by=acting_user_id(request.user),
         )
         try:
-            workflow: SeerWorkflow
-            if strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION:
-                workflow = BreachedMetricWorkflow.create_with_monitors(
-                    organization.id,
+            if strategy is SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION:
+                workflow = SeerWorkflow.create(
+                    organization_id=organization.id,
+                    strategy=strategy,
                     enabled=data["enabled"],
+                    scope=SelectedMonitors(tuple(monitor_ids)),
                     environments=data["environments"],
-                    monitor_ids=monitor_ids,
                     extras=extras,
                     service_account_grant=grant,
                 )
-            else:
+            elif strategy is SeerWorkflowStrategy.AGENTIC_TRIAGE_V2:
                 workflow = SeerWorkflow.create(
                     organization_id=organization.id,
                     strategy=strategy,
                     enabled=data["enabled"],
                     schedule=SeerWorkflowSchedule(data.get("schedule", SeerWorkflowSchedule.DAILY)),
+                    scope=(
+                        AllProjects()
+                        if set(project_ids) == ALL_ACCESS_PROJECTS
+                        else SelectedProjects(tuple(project_ids))
+                    ),
                     environments=data["environments"],
-                    project_ids=project_ids,
                     extras=extras,
                     service_account_grant=grant,
                 )
+            else:
+                raise WorkflowUpdateError(f"{strategy} automations cannot be configured")
         except ValueError as e:
             logger.exception("bad request", extra={ATTRIBUTE_NAMES.HTTP_RESPONSE_STATUS_CODE: 400})
             if isinstance(e, WorkflowUpdateError):
@@ -579,15 +583,18 @@ def put(self, request: Request, organization: Organization, config_id: str) -> R
             )
         except (ValueError, SeerWorkflowConfig.DoesNotExist) as e:
             raise NotFound from e
-        workflow = get_seer_workflow(config)
+        workflow = SeerWorkflow(config_id=config.id, organization_id=organization.id)
 
         enabled: bool | None = serializer.validated_data.get("enabled")
         project_ids: list[int] | None = serializer.validated_data.get("projectIds")
         monitor_ids: list[int] | None = serializer.validated_data.get("monitorIds")
         name: str | None = serializer.validated_data.get("name")
 
         # Checks that depend only on the request run before taking the lock.
-        if monitor_ids is not None and not isinstance(workflow, BreachedMetricWorkflow):
+        if (
+            monitor_ids is not None
+            and get_definition(config.strategy).trigger.scoped_by != "monitors"
+        ):
             return Response(
                 {"detail": "Only metric automations support monitor selection."}, status=400
             )
@@ -646,10 +653,10 @@ def put(self, request: Request, organization: Organization, config_id: str) -> R
                         )
 
                     if all_projects:
-                        update.set_project_selection(AllProjects())
+                        update.set_scope(AllProjects())
                     else:
                         # Replace only what the caller can see; keep current projects they can't.
-                        update.set_project_selection(
+                        update.set_scope(
                             SelectedProjects(
                                 project_ids=tuple(
                                     _with_no_access_project_ids(
@@ -664,16 +671,18 @@ def put(self, request: Request, organization: Organization, config_id: str) -> R
                         )
 
                 if monitor_ids is not None:
-                    assert isinstance(workflow, BreachedMetricWorkflow)
-                    assert isinstance(update, BreachedMetricConfigUpdate)
                     # Replace only what the caller can see; keep monitors in projects they can't.
-                    update.set_monitor_selection(
-                        _with_no_access_monitor_ids(
-                            monitor_ids,
-                            workflow.get_monitors(),
-                            request=request,
-                            organization=organization,
-                            endpoint=self,
+                    update.set_scope(
+                        SelectedMonitors(
+                            tuple(
+                                _with_no_access_monitor_ids(
+                                    monitor_ids,
+                                    get_config_monitors_bulk([update.config]).get(config.id, []),
+                                    request=request,
+                                    organization=organization,
+                                    endpoint=self,
+                                )
+                            )
                         )
                     )
 
@@ -722,7 +731,7 @@ def delete(self, request: Request, organization: Organization, config_id: str) -
         except (ValueError, SeerWorkflowConfig.DoesNotExist) as e:
             raise NotFound from e
 
-        workflow = get_seer_workflow(config)
+        workflow = SeerWorkflow(config_id=config.id, organization_id=organization.id)
         account = workflow.get_service_account()
         workflow.delete()
         if account is not None:

src/sentry/seer/endpoints/organization_seer_workflow_runs.py

@@ -28,7 +28,6 @@
 from sentry.ratelimits.config import RateLimitConfig
 from sentry.seer.models.workflow import (
     SEER_CONFIGURABLE_WORKFLOW_STRATEGIES,
-    SEER_CRON_WORKFLOW_STRATEGIES,
     SeerWorkflowExecutionStatus,
     SeerWorkflowRun,
     SeerWorkflowRunExecution,
@@ -37,7 +36,8 @@
     SeerWorkflowStrategy,
 )
 from sentry.seer.monitor_cleanup.runs import create_monitor_cleanup_run
-from sentry.seer.workflows.breached_metrics import MetricBreachTrigger
+from sentry.seer.workflows.breached_metrics import MetricBreachInputs
+from sentry.seer.workflows.definitions import is_scheduled
 from sentry.seer_workflow_outputs.models import SeerWorkflowOutputLink
 from sentry.types.ratelimit import RateLimit, RateLimitCategory
 from sentry.utils.safe import get_path
@@ -326,14 +326,14 @@ def serialize_workflow_run(run: SeerWorkflowRun) -> SeerWorkflowRunResponse:
 def serialize_trigger(run: SeerWorkflowRun) -> TriggerResponse | None:
     """Serialize the trigger of a run loaded with a `workflow_config__strategy` annotation."""
     strategy = getattr(run, "strategy")
-    if strategy in SEER_CRON_WORKFLOW_STRATEGIES:
+    if is_scheduled(strategy):
         return {"type": "cron"}
     if strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION and get_path(
         run.extras, "snapshot", "inputs"
     ):
         try:
             inputs = SeerWorkflowRunExtras.parse_obj(run.extras).snapshot.inputs
-            trigger = MetricBreachTrigger.parse_obj(inputs)
+            trigger = MetricBreachInputs.parse_obj(inputs)
             return {
                 "type": "monitor",
                 "detectorId": str(trigger.detector_id),

src/sentry/seer/models/workflow.py

@@ -34,9 +34,6 @@ class SeerWorkflowStrategy(models.TextChoices):
     {SeerWorkflowStrategy.AGENTIC_TRIAGE_V2, SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION}
 )
 
-# Configurable strategies whose runs are started on a schedule.
-SEER_CRON_WORKFLOW_STRATEGIES = frozenset({SeerWorkflowStrategy.AGENTIC_TRIAGE_V2})
-
 # Soft bound on configs per organization. Actual count could go over a bit from concurrent writes.
 MAX_WORKFLOW_CONFIGS_PER_ORG = 50
 
@@ -51,6 +48,9 @@ class SeerWorkflowConfigExtras(BaseModel):
     # The user who most recently updated this config through the API, or "bot" for
     # non-user callers. Unset until the first update.
     updated_by: int | Literal["bot"] | None = None
+    # Settings for the automation's trigger and action, in the shapes those classes declare.
+    trigger_settings: dict[str, Any] | None = None
+    action_settings: dict[str, Any] | None = None
 
 
 class SeerWorkflowFilters(BaseModel):