feat(seer): Wire automation schedules into Cadence

src/sentry/conf/server.py

@@ -967,6 +967,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.seer.entrypoints.operator",
     "sentry.seer.entrypoints.slack.messaging",
     "sentry.seer.entrypoints.slack.tasks",
+    "sentry.seer.workflows.scheduled",
     "sentry.seer_workflow_outputs.tasks",
     "sentry.snuba.query_subscriptions.run",
     "sentry.snuba.tasks",

src/sentry/features/temporary.py

@@ -282,6 +282,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:seer-explorer", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable the Seer Automations UI and APIs
     manager.add("organizations:seer-automations", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
+    # Enable execution of scheduled Seer automations through Cadence.
+    manager.add("organizations:seer-scheduled-automations", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable the Seer item in the primary navigation
     manager.add("organizations:seer-primary-nav", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable Seer Agent Index job

src/sentry/seer/endpoints/organization_seer_workflow_configs.py

@@ -25,6 +25,7 @@
 from sentry.apidocs.parameters import CursorQueryParam, GlobalParams
 from sentry.apidocs.utils import inline_sentry_response_serializer
 from sentry.auth.principal import acting_user_id
+from sentry.cadence.api import ScheduledTriggerType, get_schedules
 from sentry.constants import ALL_ACCESS_PROJECT_ID, ALL_ACCESS_PROJECTS
 from sentry.models.organization import Organization
 from sentry.ratelimits.config import RateLimitConfig
@@ -86,6 +87,8 @@ class SeerWorkflowConfigResponse(TypedDict):
     filters: dict[str, Any]
     outputPolicy: dict[str, Any]
     seededDefault: str | None
+    # When the automation next runs on its schedule, or None if it doesn't run on one.
+    nextRunAt: datetime | None
     dateAdded: datetime
     dateUpdated: datetime
     dateLastRun: NotRequired[datetime]
@@ -351,6 +354,8 @@ def get(self, request: Request, organization: Organization) -> Response:
         ``name`` is the user-supplied name, omitted if unset.
         ``createdBy`` is the ID of the user who created the config, omitted if unset.
         ``updatedBy`` is the ID of the user who last changed it, omitted if unset.
+        ``nextRunAt`` is when the automation next runs on its schedule, or null if it
+        isn't scheduled to run.
         ``dateLastRun`` is the latest visible run's creation time and ``lastRunStatus``
         is its status (null for older runs that predate status tracking). Both are
         omitted if no runs are visible under the runs endpoint's project-scope rules.
@@ -747,6 +752,7 @@ def serialize_workflow_configs(
         has_global_access=has_global_access,
     )
     last_runs_by_config_id = {run.config_id: run for run in last_runs}
+    next_run_times = get_next_run_times(configs)
     return {
         "configs": [
             serialize_workflow_config(
@@ -759,6 +765,7 @@ def serialize_workflow_configs(
                     if config.service_account_id
                     else None
                 ),
+                next_run_at=next_run_times.get(config.id),
             )
             for config in configs
         ]
@@ -805,13 +812,30 @@ def get_config_last_runs(
     ]
 
 
+def get_next_run_times(configs: Sequence[SeerWorkflowConfig]) -> dict[int, datetime | None]:
+    """When each config next runs on its schedule, keyed by config id."""
+    by_organization: defaultdict[int, list[int]] = defaultdict(list)
+    for config in configs:
+        by_organization[config.organization_id].append(config.id)
+    return {
+        config_id: info.next_fire_at
+        for organization_id, config_ids in by_organization.items()
+        for config_id, info in get_schedules(
+            organization_id=organization_id,
+            trigger_type=ScheduledTriggerType.SEER_WORKFLOW,
+            target_ids=config_ids,
+        ).items()
+    }
+
+
 def serialize_workflow_config(
     config: SeerWorkflowConfig,
     *,
     selected_project_ids: list[int],
     connected_monitors: list[ConnectedMonitor],
     last_run: WorkflowConfigLastRun | None,
     service_account: SeerServiceAccountResponse | None,
+    next_run_at: datetime | None,
 ) -> SeerWorkflowConfigResponse:
     filters = SeerWorkflowFilters.parse_obj(config.filters)
     monitor_ids = [m.monitor_id for m in connected_monitors]
@@ -842,6 +866,7 @@ def serialize_workflow_config(
         ),
         "outputPolicy": config.output_policy,
         "seededDefault": SeerWorkflowConfigExtras.parse_obj(config.extras).seeded_default,
+        "nextRunAt": next_run_at,
         "dateAdded": config.date_added,
         "dateUpdated": config.date_updated,
     }

src/sentry/seer/workflows/schedule.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import functools
+from typing import assert_never
+
+from sentry.cadence.api import (
+    CronSchedule,
+    Schedule,
+    ScheduledTriggerType,
+    clear_schedule,
+    cron,
+    set_schedule,
+)
+from sentry.seer.models.workflow import (
+    SeerWorkflowConfig,
+    SeerWorkflowSchedule,
+    SeerWorkflowStrategy,
+)
+
+# Strategies that run on their config's schedule. Others are triggered by events.
+SCHEDULED_STRATEGIES = frozenset({SeerWorkflowStrategy.AGENTIC_TRIAGE_V2})
+
+
+@functools.cache
+def _daily() -> CronSchedule:
+    # Matches the legacy night shift's morning run. Firings are spread over the following
+    # hour by the trigger type's spread window; see sentry.seer.workflows.scheduled.
+    return cron("0 10 * * *")
+
+
+def schedule_for(config: SeerWorkflowConfig) -> Schedule | None:
+    """The schedule `config` should run on, or None if it shouldn't run on a schedule."""
+    if not config.enabled or config.strategy not in SCHEDULED_STRATEGIES:
+        return None
+    schedule = SeerWorkflowSchedule(config.schedule)
+    match schedule:
+        case SeerWorkflowSchedule.DAILY:
+            return _daily()
+        case _:
+            assert_never(schedule)
+
+
+def sync_schedule(config: SeerWorkflowConfig) -> None:
+    """
+    Make the config's cadence schedule match `schedule_for`. Safe to call repeatedly, so
+    call it after any change to a config's `enabled` or `schedule`.
+    """
+    schedule = schedule_for(config)
+    if schedule is None:
+        clear_schedule(
+            organization_id=config.organization_id,
+            trigger_type=ScheduledTriggerType.SEER_WORKFLOW,
+            target_id=config.id,
+        )
+    else:
+        set_schedule(
+            organization_id=config.organization_id,
+            trigger_type=ScheduledTriggerType.SEER_WORKFLOW,
+            target_id=config.id,
+            schedule=schedule,
+        )

src/sentry/seer/workflows/scheduled.py

@@ -1,22 +1,164 @@
+"""
+Connects Seer automation schedules to sentry.cadence.
+
+Firings reach the task, but scheduled execution is separately gated off while Seer dispatch
+is unimplemented. Admission and lifecycle handling remain available for that integration.
+No run history is created while the execution flag is off.
+"""
+
 from __future__ import annotations
 
 import logging
 from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any, Literal
+
+from pydantic import BaseModel
 
-from sentry.cadence.api import ScheduledTriggerType, TriggerFiring, register_trigger_action
+from sentry import features
+from sentry.cadence.api import (
+    ScheduledTriggerType,
+    TriggerFiring,
+    clear_schedule,
+    register_trigger_action,
+)
+from sentry.seer.models.workflow import (
+    SeerWorkflowConfig,
+    SeerWorkflowRun,
+    WorkflowRunError,
+)
+from sentry.seer.workflows.runs import SeerWorkflowRunLifecycle
+from sentry.seer.workflows.schedule import schedule_for
+from sentry.seer.workflows.workflow import SeerWorkflow
+from sentry.silo.base import SiloMode
+from sentry.tasks.base import instrumented_task
+from sentry.taskworker.namespaces import seer_tasks
+from sentry.utils import metrics
 
 logger = logging.getLogger(__name__)
 
+SCHEDULED_EXECUTION_FEATURE = "organizations:seer-scheduled-automations"
+
+
+class ScheduledRunInputs(BaseModel):
+    """The inputs of a scheduled run, recorded in its snapshot."""
+
+    source: Literal["cron"] = "cron"
+    scheduled_for: datetime
 
-# TODO: Enqueue a task per firing that creates a cron-sourced run of the SeerWorkflowConfig.
-@register_trigger_action(ScheduledTriggerType.SEER_WORKFLOW)
-def schedule_workflow_runs(firings: Sequence[TriggerFiring]) -> None:
+
+def trigger_key_for(scheduled_for: datetime) -> str:
+    """Identifies a scheduled run, so the same run delivered twice is only created once."""
+    return f"cron:{scheduled_for.astimezone(UTC):%Y-%m-%dT%H:%M}"
+
+
+# Matches the legacy night shift's spread of org runs over the hour after its scheduled time.
+@register_trigger_action(ScheduledTriggerType.SEER_WORKFLOW, spread_window=timedelta(hours=1))
+def enqueue_scheduled_automation_runs(firings: Sequence[TriggerFiring]) -> None:
     for firing in firings:
-        logger.info(
-            "seer.workflow.scheduled_run.stub",
-            extra={
+        record_scheduled_automation_run.apply_async(
+            kwargs={
                 "organization_id": firing.organization_id,
-                "workflow_config_id": firing.target_id,
+                "config_id": firing.target_id,
                 "scheduled_for": firing.scheduled_for.isoformat(),
             },
+            headers={"sentry-propagate-traces": False},
+        )
+
+
+@dataclass(frozen=True)
+class ScheduledRunOutcome:
+    outcome: Literal["accepted", "duplicate", "rejected", "ignored"]
+    reason: str | None = None
+    run: SeerWorkflowRun | None = None
+
+
+@instrumented_task(
+    name="sentry.seer.workflows.scheduled.record_scheduled_automation_run",
+    namespace=seer_tasks,
+    processing_deadline_duration=60,
+    silo_mode=SiloMode.CELL,
+)
+def record_scheduled_automation_run(
+    organization_id: int, config_id: int, scheduled_for: str, **kwargs: Any
+) -> None:
+    """Record the cron-sourced run of an automation config that Cadence says is due."""
+    result = _run(organization_id, config_id, datetime.fromisoformat(scheduled_for))
+    metrics.incr(
+        "seer.workflow.scheduled_run",
+        tags={"outcome": result.outcome, "reason": result.reason or "none"},
+        sample_rate=1.0,
+    )
+    logger.info(
+        "seer.workflow.scheduled_run",
+        extra={
+            "organization_id": organization_id,
+            "config_id": config_id,
+            "scheduled_for": scheduled_for,
+            "outcome": result.outcome,
+            "reason": result.reason,
+            "run_id": result.run.id if result.run else None,
+        },
+    )
+
+
+def _run(organization_id: int, config_id: int, scheduled_for: datetime) -> ScheduledRunOutcome:
+    config = SeerWorkflowConfig.objects.get_or_none(id=config_id, organization_id=organization_id)
+    if config is None or schedule_for(config) is None:
+        # The config was deleted, disabled, or no longer runs on a schedule, but its
+        # schedule wasn't cleared. Clear it now so it stops firing.
+        clear_schedule(
+            organization_id=organization_id,
+            trigger_type=ScheduledTriggerType.SEER_WORKFLOW,
+            target_id=config_id,
         )
+        return ScheduledRunOutcome("ignored", reason="unscheduled")
+    # Entitlement can come back, so keep the schedule and do nothing this time.
+    if not features.has("organizations:seer-automations", config.organization):
+        return ScheduledRunOutcome("ignored", reason="feature_disabled")
+    # Scheduling and stale-registration cleanup stay active while execution is disabled.
+    # Firings consumed during this stage are not replayed when execution is enabled later.
+    if not features.has(SCHEDULED_EXECUTION_FEATURE, config.organization):
+        return ScheduledRunOutcome("ignored", reason="execution_disabled")
+    return ScheduledSeerWorkflow(
+        config_id=config.id, organization_id=config.organization_id
+    ).record_run(scheduled_for)
+
+
+class ScheduledSeerWorkflow(SeerWorkflow):
+    def record_run(self, scheduled_for: datetime) -> ScheduledRunOutcome:
+        """Admit and dispatch the run scheduled for `scheduled_for`, once per config."""
+        try:
+            with self._lock_config() as config:
+                if not config.enabled:
+                    return ScheduledRunOutcome("ignored", reason="configuration_unavailable")
+
+                result = self.get_or_create_run(
+                    trigger_key=trigger_key_for(scheduled_for),
+                    inputs=ScheduledRunInputs(scheduled_for=scheduled_for),
+                )
+                lifecycle = result.lifecycle
+                if not result.created:
+                    return ScheduledRunOutcome("duplicate", run=lifecycle.get_run())
+                if not self.has_quota_available():
+                    run = lifecycle.reject(
+                        error=WorkflowRunError(
+                            code="quota_exceeded", message="Automation quota exceeded."
+                        )
+                    )
+                    return ScheduledRunOutcome("rejected", reason="quota_exceeded", run=run)
+
+                lifecycle.start()
+                self._dispatch_run(lifecycle)
+                return ScheduledRunOutcome("accepted", run=lifecycle.get_run())
+        except SeerWorkflowConfig.DoesNotExist:
+            return ScheduledRunOutcome("ignored", reason="configuration_unavailable")
+
+    def _dispatch_run(self, lifecycle: SeerWorkflowRunLifecycle) -> None:
+        """Persist the strategy's execution plan and durable dispatch before admission commits.
+
+        Seer result/status delivery must complete those executions. Raising here rolls back
+        admission if the execution flag is enabled before this implementation is supplied.
+        """
+        raise NotImplementedError("Scheduled Seer automation dispatch is not implemented")

src/sentry/seer/workflows/workflow.py

@@ -11,6 +11,7 @@
 from django.db.models import Count, Q, QuerySet
 from pydantic import BaseModel
 
+from sentry.cadence.api import ScheduledTriggerType, clear_schedule
 from sentry.constants import ALL_ACCESS_PROJECTS, ObjectStatus
 from sentry.hybridcloud.models.outbox import OutboxDatabaseError, OutboxFlushError, outbox_context
 from sentry.models.organization import Organization
@@ -30,6 +31,7 @@
     WorkflowRunSnapshot,
 )
 from sentry.seer.workflows.runs import SeerWorkflowRunLifecycle
+from sentry.seer.workflows.schedule import sync_schedule
 from sentry.service_accounts.models import (
     MAX_NAME_LENGTH,
     InvalidScopeError,
@@ -232,6 +234,7 @@ def create(
         ) as config:
             config.schedule = schedule
             config.save(using=database, update_fields=["schedule"])
+            sync_schedule(config)
             if project_set:
                 SeerWorkflowConfigProject.objects.using(database).bulk_create(
                     [
@@ -353,6 +356,11 @@ def delete(self) -> None:
         self._delete_service_account()
         with self._lock_config() as config:
             config.delete()
+            clear_schedule(
+                organization_id=self.organization_id,
+                trigger_type=ScheduledTriggerType.SEER_WORKFLOW,
+                target_id=self.config_id,
+            )
 
     def _delete_service_account(self) -> None:
         service_account_id = self.get_config().service_account_id
@@ -392,7 +400,8 @@ def resolve_service_account(self) -> ServiceAccountReplica:
     def update_config(self) -> Iterator[ConfigUpdate]:
         """Lock the config row for a group of changes that commit or roll back together.
 
-        On a normal exit, any marked changes are saved atomically with a single version bump.
+        On a normal exit, any marked changes are saved atomically with a single version bump,
+        and the config's schedule is synced to match.
         Do not early return on error cases - propagate an exception so the transaction can roll back.
         """
         with transaction.atomic(router.db_for_write(SeerWorkflowConfig)):
@@ -403,6 +412,8 @@ def update_config(self) -> Iterator[ConfigUpdate]:
                 update.config.save(
                     update_fields=[*sorted(update.dirty_fields), "version", "date_updated"]
                 )
+            # Even when unchanged, so a schedule that drifted from the config is repaired.
+            sync_schedule(update.config)
 
     @classmethod
     def _new_config_update(cls, config: SeerWorkflowConfig) -> ConfigUpdate: