feat(seer): Add Night Shift nightly autofix cron scaffolding

src/sentry/conf/server.py

@@ -991,6 +991,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.tasks.seer.explorer_index",
     "sentry.tasks.seer.context_engine_index",
     "sentry.tasks.seer.lightweight_rca_cluster",
+    "sentry.tasks.seer.night_shift",
     # Used for tests
     "sentry.taskworker.tasks.examples",
 )
@@ -1173,6 +1174,11 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         # Run once a month at midnight
         "schedule": crontab("0", "0", "*", "1", "*"),
     },
+    "seer-night-shift": {
+        "task": "seer:sentry.tasks.seer.night_shift.schedule_night_shift",
+        # Run daily at 10:00 AM UTC (2/3 AM Pacific)
+        "schedule": crontab("0", "10", "*", "*", "*"),
+    },
     "refresh-artifact-bundles-in-use": {
         "task": "attachments:sentry.debug_files.tasks.refresh_artifact_bundles_in_use",
         "schedule": crontab("*/1", "*", "*", "*", "*"),

src/sentry/features/temporary.py

@@ -300,6 +300,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:seer-explorer", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable Seer Explorer Index job
     manager.add("organizations:seer-explorer-index", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
+    # Enable Seer Night Shift nightly autofix cron
+    manager.add("organizations:seer-night-shift", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable streaming responses for Seer Explorer
     manager.add("organizations:seer-explorer-streaming", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable context engine for Seer Explorer

src/sentry/options/defaults.py

@@ -1373,6 +1373,17 @@
     default=0.0,
     flags=FLAG_MODIFIABLE_RATE | FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "seer.night_shift.enable",
+    type=Bool,
+    default=False,
+    flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE,
+)
+register(
+    "seer.night_shift.issues_per_org",
+    default=5,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 
 # Supergroups / Lightweight RCA
 register(

src/sentry/seer/autofix/constants.py

@@ -54,10 +54,12 @@ class AutofixReferrer(enum.StrEnum):
     ISSUE_SUMMARY_POST_PROCESS_FIXABILITY = "issue_summary.post_process_fixability"
     SLACK = "slack"
     ON_COMPLETION_HOOK = "autofix.on_completion_hook"
+    NIGHT_SHIFT = "night_shift"
     UNKNOWN = "unknown"
 
 
 class SeerAutomationSource(enum.Enum):
     ISSUE_DETAILS = "issue_details"
     ALERT = "alert"
     POST_PROCESS = "post_process"
+    NIGHT_SHIFT = "night_shift"

src/sentry/tasks/seer/night_shift.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from datetime import timedelta
+
+import sentry_sdk
+
+from sentry import features, options
+from sentry.models.organization import Organization, OrganizationStatus
+from sentry.tasks.base import instrumented_task
+from sentry.taskworker.namespaces import seer_tasks
+from sentry.utils.iterators import chunked
+from sentry.utils.query import RangeQuerySetWrapper
+
+logger = logging.getLogger("sentry.tasks.seer.night_shift")
+
+NIGHT_SHIFT_DISPATCH_STEP_SECONDS = 37
+NIGHT_SHIFT_SPREAD_DURATION = timedelta(hours=4)
+
+FEATURE_NAMES = [
+    "organizations:seer-night-shift",
+    "organizations:gen-ai-features",
+]
+
+
+@instrumented_task(
+    name="sentry.tasks.seer.night_shift.schedule_night_shift",
+    namespace=seer_tasks,
+    processing_deadline_duration=15 * 60,
+)
+def schedule_night_shift() -> None:
+    """
+    Nightly scheduler: iterates active orgs in batches, checks feature flags
+    in bulk, and dispatches per-org worker tasks with jitter.
+    """
+    if not options.get("seer.night_shift.enable"):
+        return
+
+    spread_seconds = int(NIGHT_SHIFT_SPREAD_DURATION.total_seconds())
+    batch_index = 0
+
+    for org_batch in chunked(
+        RangeQuerySetWrapper[Organization](
+            Organization.objects.filter(status=OrganizationStatus.ACTIVE),
+            step=1000,
+        ),
+        100,
+    ):
+        for org in _get_eligible_orgs_from_batch(org_batch):
+            if bool(org.get_option("sentry:hide_ai_features")):
+                continue
+
+            delay = (batch_index * NIGHT_SHIFT_DISPATCH_STEP_SECONDS) % spread_seconds
+
+            run_night_shift_for_org.apply_async(
+                args=[org.id],
+                countdown=delay,
+            )
+            batch_index += 1
+
+    logger.info(
+        "night_shift.schedule_complete",
+        extra={"orgs_dispatched": batch_index},
+    )
+
+
+@instrumented_task(
+    name="sentry.tasks.seer.night_shift.run_night_shift_for_org",
+    namespace=seer_tasks,
+    processing_deadline_duration=5 * 60,
+)
+def run_night_shift_for_org(organization_id: int) -> None:
+    try:
+        organization = Organization.objects.get(
+            id=organization_id, status=OrganizationStatus.ACTIVE
+        )
+    except Organization.DoesNotExist:
+        return
+
+    sentry_sdk.set_tags(
+        {
+            "organization_id": organization.id,
+            "organization_slug": organization.slug,
+        }
+    )
+
+    logger.info(
+        "night_shift.org_dispatched",
+        extra={
+            "organization_id": organization_id,
+            "organization_slug": organization.slug,
+        },
+    )
+
+
+def _get_eligible_orgs_from_batch(
+    orgs: Sequence[Organization],
+) -> list[Organization]:
+    """
+    Check feature flags for a batch of orgs using batch_has_for_organizations.
+    Returns orgs that have all required feature flags enabled.
+    """
+    eligible = list(orgs)
+
+    for feature_name in FEATURE_NAMES:
+        batch_result = features.batch_has_for_organizations(feature_name, eligible)
+        if batch_result is None:
+            raise RuntimeError(f"batch_has_for_organizations returned None for {feature_name}")
+
+        eligible = [org for org in eligible if batch_result.get(f"organization:{org.id}", False)]
+
+        if not eligible:
+            return []
+
+    return eligible

tests/sentry/tasks/seer/test_night_shift.py

@@ -0,0 +1,79 @@
+from unittest.mock import patch
+
+from sentry.tasks.seer.night_shift import (
+    run_night_shift_for_org,
+    schedule_night_shift,
+)
+from sentry.testutils.cases import TestCase
+from sentry.testutils.pytest.fixtures import django_db_all
+
+
+@django_db_all
+class TestScheduleNightShift(TestCase):
+    def test_disabled_by_option(self) -> None:
+        with (
+            self.options({"seer.night_shift.enable": False}),
+            patch("sentry.tasks.seer.night_shift.run_night_shift_for_org") as mock_worker,
+        ):
+            schedule_night_shift()
+            mock_worker.apply_async.assert_not_called()
+
+    def test_dispatches_eligible_orgs(self) -> None:
+        org = self.create_organization()
+
+        with (
+            self.options({"seer.night_shift.enable": True}),
+            self.feature(
+                {
+                    "organizations:seer-night-shift": [org.slug],
+                    "organizations:gen-ai-features": [org.slug],
+                }
+            ),
+            patch("sentry.tasks.seer.night_shift.run_night_shift_for_org") as mock_worker,
+        ):
+            schedule_night_shift()
+            mock_worker.apply_async.assert_called_once()
+            assert mock_worker.apply_async.call_args.kwargs["args"] == [org.id]
+
+    def test_skips_ineligible_orgs(self) -> None:
+        self.create_organization()
+
+        with (
+            self.options({"seer.night_shift.enable": True}),
+            patch("sentry.tasks.seer.night_shift.run_night_shift_for_org") as mock_worker,
+        ):
+            schedule_night_shift()
+            mock_worker.apply_async.assert_not_called()
+
+    def test_skips_orgs_with_hidden_ai(self) -> None:
+        org = self.create_organization()
+        org.update_option("sentry:hide_ai_features", True)
+
+        with (
+            self.options({"seer.night_shift.enable": True}),
+            self.feature(
+                {
+                    "organizations:seer-night-shift": [org.slug],
+                    "organizations:gen-ai-features": [org.slug],
+                }
+            ),
+            patch("sentry.tasks.seer.night_shift.run_night_shift_for_org") as mock_worker,
+        ):
+            schedule_night_shift()
+            mock_worker.apply_async.assert_not_called()
+
+
+@django_db_all
+class TestRunNightShiftForOrg(TestCase):
+    def test_logs_for_valid_org(self) -> None:
+        org = self.create_organization()
+
+        with patch("sentry.tasks.seer.night_shift.logger") as mock_logger:
+            run_night_shift_for_org(org.id)
+            mock_logger.info.assert_called_once()
+            assert mock_logger.info.call_args.args[0] == "night_shift.org_dispatched"
+
+    def test_nonexistent_org(self) -> None:
+        with patch("sentry.tasks.seer.night_shift.logger") as mock_logger:
+            run_night_shift_for_org(999999999)
+            mock_logger.info.assert_not_called()