feat(workflows): mirror the sandbox ses tenant state in django

.github/scripts/check-idor-model-coverage.py

@@ -327,6 +327,7 @@ def get_scoped_models() -> tuple[dict[str, set[str]], set[str], set[str], set[st
         "DevStackImageTarget",
         "DevStackImageAttempt",
         "License",
+        "SandboxSenderTenantState",  # instance-global mirror of the shared sandbox SES tenant state
         # --- Deprecated ---
         "Prompt",
         "PromptSequence",

docs/internal/workflows-email-sending-tiers.md

@@ -110,6 +110,8 @@ A reached allowance skips the send instead of delaying it. A limiter error, or a
 The worker checks members after the claim and gives the claim back when that check skips the send.
 An SES throttle response also gives the claim back before the delivery retry. Returns are deduplicated, and Valkey ignores a return that arrives more than five minutes after it was issued.
 Test sends run inline in the CDP API, so a sandbox-enabled CDP API opens the SES Valkey pool (`SES_RATE_LIMITER_VALKEY_*`) for the allowances. Without it, every sandbox test send skips.
+Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker.
+A committed state change reaches the workers through the `reload-sandbox-sender-state` message.
 Each delivered send emits a `workflows sandbox email sent` event.
 Test sends never record `email_sent`, so they do not count toward the sending tier.
 Live workflow sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric.

posthog/plugins/plugin_server_api.py

@@ -82,6 +82,11 @@ def reload_team_workflows_config_on_workers(team_id: int):
     publish_message("reload-team-workflows-config", {"teamId": team_id})
 
 
+def reload_sandbox_sender_state_on_workers(tenant_name: str) -> None:
+    logger.info(f"Reloading sandbox sender state for tenant {tenant_name} on workers")
+    publish_message("reload-sandbox-sender-state", {"tenantName": tenant_name})
+
+
 def populate_plugin_capabilities_on_workers(plugin_id: str):
     logger.info(f"Populating plugin capabilities for plugin {plugin_id} on workers")
     publish_message("populate-plugin-capabilities", {"pluginId": plugin_id})

products/workflows/backend/migrations/0040_sandbox_sender_tenant_state.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.2.18 on 2026-10-27 07:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("workflows", "0039_workflowimport"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="SandboxSenderTenantState",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("tenant_name", models.CharField(max_length=64, unique=True)),
+                (
+                    "sending_status",
+                    models.CharField(blank=True, db_default="", default="", max_length=32),
+                ),
+                ("synced_at", models.DateTimeField(blank=True, null=True)),
+            ],
+        ),
+    ]

products/workflows/backend/migrations/max_migration.txt

@@ -1 +1 @@
-0039_workflowimport
+0040_sandbox_sender_tenant_state

products/workflows/backend/models/__init__.py

@@ -4,6 +4,7 @@
 from .hog_flow_optimization import HogFlowOptimization
 from .hog_flow_revision import HogFlowRevision
 from .hog_flow_schedule.hog_flow_schedule import HogFlowSchedule
+from .sandbox_sender_tenant_state import SandboxSenderTenantState
 from .team_workflows_config import TeamWorkflowsConfig
 from .workflow_idea import WorkflowIdea
 from .workflow_import import WorkflowImport
@@ -16,6 +17,7 @@
     "HogFlowRevision",
     "HogFlowSchedule",
     "HogFlowTemplate",
+    "SandboxSenderTenantState",
     "TeamWorkflowsConfig",
     "WorkflowIdea",
     "WorkflowImport",

products/workflows/backend/models/sandbox_sender_tenant_state.py

@@ -0,0 +1,9 @@
+from django.db import models
+
+
+class SandboxSenderTenantState(models.Model):
+    """Instance-level mirror of the shared sandbox SES tenant; platform state, not tenant data, so no team_id."""
+
+    tenant_name = models.CharField(max_length=64, unique=True)
+    sending_status = models.CharField(max_length=32, blank=True, default="", db_default="")
+    synced_at = models.DateTimeField(null=True, blank=True)

products/workflows/backend/providers/ses.py

@@ -395,6 +395,13 @@ def _is_throttling(error: Exception) -> bool:
     return isinstance(error, ClientError) and error.response["Error"]["Code"] in ("Throttling", "ThrottlingException")
 
 
+@frozen
+class TenantSendingState:
+    sending_status: str
+    reputation_impact: str | None
+    tenant_arn: str | None
+
+
 class SESProvider:
     ses_client: "SESClient"
     ses_v2_client: "SESV2Client"
@@ -469,7 +476,17 @@ def get_tenant_reputation(self, team_id: int) -> dict[str, Any] | None:
         provider feedback loops, third-party listings), so this is the authoritative health source;
         our own app metrics only provide the per-workflow diagnosis.
         """
-        tenant_name = self._tenant_name_for_team(team_id)
+        state = self.get_tenant_sending_state(self._tenant_name_for_team(team_id))
+        if state is None:
+            return None
+        return {
+            "sending_status": state.sending_status,
+            "reputation_impact": state.reputation_impact,
+            "findings": self._open_findings(state.tenant_arn) if state.tenant_arn else [],
+        }
+
+    def get_tenant_sending_state(self, tenant_name: str) -> TenantSendingState | None:
+        """Sending status and reputation impact of the named SES tenant, or None when it doesn't exist."""
         try:
             tenant = self.ses_v2_client.get_tenant(TenantName=tenant_name)["Tenant"]
         except ClientError as e:
@@ -479,40 +496,38 @@ def get_tenant_reputation(self, team_id: int) -> dict[str, Any] | None:
 
         sending_status: str = tenant.get("SendingStatus", "ENABLED")
         tenant_arn = tenant.get("TenantArn")
-        reputation_impact: str | None = None
-        findings: list[dict[str, Any]] = []
+        if not tenant_arn:
+            return TenantSendingState(sending_status=sending_status, reputation_impact=None, tenant_arn=None)
 
-        if tenant_arn:
-            try:
-                entity = self.ses_v2_client.get_reputation_entity(
-                    ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE"
-                )["ReputationEntity"]
-            except ClientError as e:
-                # A tenant with no attributed sends yet has no reputation entity.
-                if e.response["Error"]["Code"] != "NotFoundException":
-                    raise
-                entity = {}
-            reputation_impact = entity.get("ReputationImpact")
+        entity = self._reputation_entity(tenant_arn)
+        return TenantSendingState(
             # The aggregate folds in both AWS-managed and customer-managed pauses.
-            sending_status = entity.get("SendingStatusAggregate", sending_status)
-
-            # RESOURCE_ARN is the only filter key AWS documents as usable on its own with this
-            # scoping (see _iter_open_recommendations for why STATUS is filtered locally).
-            findings.extend(
-                {
-                    "finding_type": recommendation.get("Type", ""),
-                    "impact": recommendation.get("Impact", "LOW"),
-                    "description": recommendation.get("Description", ""),
-                    "last_updated_at": recommendation.get("LastUpdatedTimestamp"),
-                }
-                for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn})
-            )
+            sending_status=entity.get("SendingStatusAggregate", sending_status),
+            reputation_impact=entity.get("ReputationImpact"),
+            tenant_arn=tenant_arn,
+        )
 
-        return {
-            "sending_status": sending_status,
-            "reputation_impact": reputation_impact,
-            "findings": findings,
-        }
+    def _reputation_entity(self, tenant_arn: str) -> Mapping[str, Any]:
+        try:
+            return self.ses_v2_client.get_reputation_entity(
+                ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE"
+            )["ReputationEntity"]
+        except ClientError as e:
+            # A tenant with no attributed sends yet has no reputation entity.
+            if e.response["Error"]["Code"] != "NotFoundException":
+                raise
+            return {}
+
+    def _open_findings(self, tenant_arn: str) -> list[dict[str, Any]]:
+        return [
+            {
+                "finding_type": recommendation.get("Type", ""),
+                "impact": recommendation.get("Impact", "LOW"),
+                "description": recommendation.get("Description", ""),
+                "last_updated_at": recommendation.get("LastUpdatedTimestamp"),
+            }
+            for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn})
+        ]
 
     def _iter_open_recommendations(
         self, finding_filter: dict[Any, str] | None, client: "SESV2Client | None" = None

products/workflows/backend/services/ses_tenant_events.py

@@ -15,21 +15,22 @@
 from collections.abc import Mapping
 from typing import Any
 
+from django.conf import settings
+
 import requests
 import structlog
 
 from posthog.ingress.contracts import WebhookDelivery
 from posthog.ingress.verify.sns_signature import is_valid_sns_url
 
-from products.workflows.backend.tasks.ses_tenant_state import sync_ses_tenant_state_task
+from products.workflows.backend.tasks.ses_tenant_state import sync_sandbox_tenant_state_task, sync_ses_tenant_state_task
 
 logger = structlog.get_logger(__name__)
 
 _TENANT_TEAM_RE = re.compile(r"\bteam-(\d+)\b")
 
 
-def _extract_team_id(event: dict[str, Any]) -> int | None:
-    """Pull the `team-<id>` tenant name out of an EventBridge event, wherever AWS put it."""
+def _tenant_candidates(event: dict[str, Any]) -> list[str]:
     detail = event.get("detail") or {}
     candidates: list[Any] = [
         detail.get("tenantName"),
@@ -39,8 +40,24 @@ def _extract_team_id(event: dict[str, Any]) -> int | None:
     resources = event.get("resources")
     if isinstance(resources, list):
         candidates.extend(resources)
-    for candidate in candidates:
-        if isinstance(candidate, str) and (match := _TENANT_TEAM_RE.search(candidate)):
+    return [candidate for candidate in candidates if isinstance(candidate, str)]
+
+
+def _names_sandbox_tenant(event: dict[str, Any]) -> bool:
+    tenant_name = settings.SES_SANDBOX_TENANT_NAME
+    if not tenant_name:
+        return False
+    tenant_path = re.compile(rf"(?:/|:)tenant/{re.escape(tenant_name)}(?:/|$)")
+    return any(
+        candidate == tenant_name or (candidate.startswith("arn:") and tenant_path.search(candidate) is not None)
+        for candidate in _tenant_candidates(event)
+    )
+
+
+def _extract_team_id(event: dict[str, Any]) -> int | None:
+    """Pull the `team-<id>` tenant name out of an EventBridge event, wherever AWS put it."""
+    for candidate in _tenant_candidates(event):
+        if match := _TENANT_TEAM_RE.search(candidate):
             return int(match.group(1))
     return None
 
@@ -61,7 +78,7 @@ def _confirm_subscription(message: Mapping[str, Any]) -> None:
 
 
 def _handle_notification(message: Mapping[str, Any]) -> None:
-    """Enqueue a tenant-state sync for the team the EventBridge event names.
+    """Enqueue a tenant-state sync for the sandbox tenant or the team the EventBridge event names.
 
     Every unusable payload is dropped rather than raised, because a redelivery would carry the
     same unusable payload.
@@ -73,6 +90,11 @@ def _handle_notification(message: Mapping[str, Any]) -> None:
     if not isinstance(event, dict) or event.get("source") != "aws.ses":
         return
 
+    if _names_sandbox_tenant(event):
+        logger.info("ses_tenant_events_webhook_accepted", tenant="sandbox", detail_type=event.get("detail-type"))
+        sync_sandbox_tenant_state_task.delay()
+        return
+
     team_id = _extract_team_id(event)
     if team_id is None:
         logger.warning("ses_tenant_events_webhook_no_tenant", detail_type=event.get("detail-type"))

products/workflows/backend/services/ses_tenant_state.py

@@ -1,21 +1,26 @@
 from collections.abc import Callable
 from typing import Any
 
+from django.conf import settings
 from django.db import transaction
 from django.utils import timezone
 
 import structlog
 
 from posthog.models.team import Team
-from posthog.plugins.plugin_server_api import reload_team_workflows_config_on_workers
+from posthog.plugins.plugin_server_api import (
+    reload_sandbox_sender_state_on_workers,
+    reload_team_workflows_config_on_workers,
+)
 from posthog.tasks.email import (
     send_email_sending_reputation_finding,
     send_email_sending_suspended,
     send_email_sending_unsuspended,
 )
 
+from products.workflows.backend.models.sandbox_sender_tenant_state import SandboxSenderTenantState
 from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig
-from products.workflows.backend.providers.ses import SESProvider
+from products.workflows.backend.providers.ses import SESProvider, TenantSendingState
 
 logger = structlog.get_logger(__name__)
 
@@ -58,6 +63,39 @@ def _announce_state_change(team_id: int) -> None:
         logger.exception("Failed to announce SES tenant state change to workers", team_id=team_id)
 
 
+def _apply_sandbox_tenant_state(state: SandboxSenderTenantState, tenant: TenantSendingState) -> None:
+    previous_status = state.sending_status
+    state.synced_at = timezone.now()
+    if previous_status == tenant.sending_status:
+        state.save(update_fields=["synced_at"])
+        logger.info("Sandbox SES tenant state unchanged", tenant_name=state.tenant_name, status=previous_status)
+        return
+
+    state.sending_status = tenant.sending_status
+    state.save(update_fields=["sending_status", "synced_at"])
+    logger.info(
+        "Sandbox SES tenant state changed",
+        tenant_name=state.tenant_name,
+        from_status=previous_status,
+        to_status=tenant.sending_status,
+    )
+    tenant_name = state.tenant_name
+    transaction.on_commit(lambda: reload_sandbox_sender_state_on_workers(tenant_name), robust=True)
+
+
+def sync_sandbox_tenant_state(provider: SESProvider | None = None) -> None:
+    # Serialize SES reads so a stale sweep cannot overwrite a newer pause.
+    tenant_name = settings.SES_SANDBOX_TENANT_NAME
+    if not tenant_name:
+        return
+    SandboxSenderTenantState.objects.get_or_create(tenant_name=tenant_name)
+    with transaction.atomic():
+        state = SandboxSenderTenantState.objects.select_for_update().get(tenant_name=tenant_name)
+        tenant = (provider or SESProvider()).get_tenant_sending_state(tenant_name)
+        if tenant is not None:
+            _apply_sandbox_tenant_state(state, tenant)
+
+
 def sync_ses_tenant_state(team_id: int, provider: SESProvider | None = None, *, verify_team: bool = True) -> None:
     """
     Fetch the authoritative AWS SES tenant state for a team and apply it. Called from the

products/workflows/backend/tasks/ses_tenant_state.py

@@ -6,7 +6,7 @@
 from posthog.tasks.utils import CeleryQueue
 
 from products.workflows.backend.providers.ses import SESProvider
-from products.workflows.backend.services.ses_tenant_state import sync_ses_tenant_state
+from products.workflows.backend.services.ses_tenant_state import sync_sandbox_tenant_state, sync_ses_tenant_state
 
 logger = get_logger(__name__)
 
@@ -27,6 +27,19 @@ def sync_ses_tenant_state_task(team_id: int) -> None:
     sync_ses_tenant_state(team_id)
 
 
+@shared_task(
+    ignore_result=True,
+    queue=CeleryQueue.DEFAULT.value,
+    autoretry_for=(Exception,),
+    retry_backoff=60,
+    retry_backoff_max=600,
+    max_retries=5,
+    retry_jitter=True,
+)
+def sync_sandbox_tenant_state_task() -> None:
+    sync_sandbox_tenant_state()
+
+
 @shared_task(ignore_result=True, queue=CeleryQueue.LONG_RUNNING.value)
 @skip_team_scope_audit
 def reconcile_ses_tenant_states() -> None:
@@ -44,6 +57,10 @@ def reconcile_ses_tenant_states() -> None:
     # One provider for the whole sweep: a fresh SESProvider per team would rebuild its boto3
     # clients (and re-resolve credentials) thousands of times.
     provider = SESProvider()
+    try:
+        sync_sandbox_tenant_state(provider=provider)
+    except Exception:
+        logger.exception("SES sandbox tenant reconciliation failed")
     synced = 0
     failed = 0
     for team_id in team_ids.iterator(chunk_size=500):