feat(workflows): provision a managed sandbox email sender

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

@@ -24,6 +24,19 @@ Only workflows that send email are subject to the tiers; SMS, push, and webhook
 
 ## Configuration
 
+### Sandbox sender provisioning
+
+The `workflows-sandbox-sender` flag enables a PostHog-managed email integration per project through `POST /api/projects/:project_id/integrations/email_sandbox_sender/`.
+The endpoint requires `integration:write` and project membership. Repeated requests keep the same integration and refresh its display name from the organization name.
+Set `WORKFLOWS_SANDBOX_SENDER_DOMAIN`, `SES_SANDBOX_FROM_ADDRESS`, and `SES_SANDBOX_TENANT_NAME` in Django; any empty value leaves the endpoint unavailable.
+The worker sends from the same `SES_SANDBOX_FROM_ADDRESS`, so set it once for Django and the worker. The address the UI shows is then the address the worker sends from.
+The workflow email step does not offer it.
+Users cannot edit, verify, or remove the managed integration, or register their own integration on the shared domain.
+The managed integration is excluded from project SES tenant migration, identity cleanup, and the reputation page's domain breakdown.
+Deploy the schema, recipient restrictions, daily caps, and tenant pause checks first. Configure the shared SES route, then enable `WORKFLOWS_SANDBOX_SENDER_ENABLED` on the worker and CDP API. Enable `workflows-sandbox-sender` last so provisioning and sender selection follow delivery readiness.
+
+### Sending tiers
+
 Two sides read their own env vars because the worker never reads Django settings.
 Set them together; drift means the UI describes limits the worker does not apply, or the reverse.
 

frontend/src/lib/integrations/utils.ts

@@ -139,7 +139,10 @@
 // surface, so invert them in dark mode (`dark:invert`) wherever the integration icon is rendered.
 export const DARK_MODE_INVERT_ICON_KINDS = new Set<IntegrationKind>(['apns', 'apple-ads', 'github'])
 
+export const isSandboxEmailSender = (integration: { kind: string; config?: unknown }): boolean =>
+    integration.kind === 'email' && (integration.config as { provider?: unknown } | undefined)?.provider === 'sandbox'
+
 export const getIntegrationNameFromKind = (kind: string): string => {
     switch (kind) {
         case 'google-pubsub':
             return 'Google Cloud Pub/Sub'

frontend/src/scenes/hog-functions/email-templater/EmailTemplater.test.tsx

@@ -18,28 +18,40 @@ jest.mock('kea', () => ({
 
 jest.mock('@posthog/lemon-ui', () => ({
     ...jest.requireActual('@posthog/lemon-ui'),
-    LemonInputSelect: () => <div data-attr="multiple-email-sender-select" />,
-    LemonSelect: () => <div data-attr="single-email-sender-select" />,
+    LemonInputSelect: ({ options }: { options: { key: string; label: string }[] }) => (
+        <div data-attr="multiple-email-sender-select">
+            {options.map((option) => (
+                <span key={option.key}>{option.label}</span>
+            ))}
+        </div>
+    ),
+    LemonSelect: ({ options }: { options: { options: { value: number; label: string }[] }[] }) => (
+        <div data-attr="single-email-sender-select">
+            {options
+                .flatMap((section) => section.options)
+                .map((option) => (
+                    <span key={option.value}>{option.label}</span>
+                ))}
+        </div>
+    ),
 }))
 
+const ALICE = { id: 1, kind: 'email', display_name: 'Alice <alice@example.com>', config: { provider: 'ses' } }
+const BOB = { id: 2, kind: 'email', display_name: 'Bob <bob@example.com>', config: { provider: 'ses' } }
+const SANDBOX = {
+    id: 3,
+    kind: 'email',
+    display_name: 'Acme via PostHog <hello@sandbox.example.com>',
+    config: { provider: 'sandbox' },
+}
+
 describe('NativeEmailIntegrationChoice', () => {
     afterEach(cleanup)
 
     beforeEach(() => {
         jest.mocked(useValues).mockReturnValue({
             integrationsLoading: false,
-            integrations: [
-                {
-                    id: 1,
-                    kind: 'email',
-                    display_name: 'Alice <alice@example.com>',
-                },
-                {
-                    id: 2,
-                    kind: 'email',
-                    display_name: 'Bob <bob@example.com>',
-                },
-            ],
+            integrations: [ALICE, BOB, SANDBOX],
         })
     })
 
@@ -54,4 +66,24 @@ describe('NativeEmailIntegrationChoice', () => {
         expect(screen.getByTestId(expectedPicker)).toBeInTheDocument()
         expect(screen.getByText('Custom sender')).toBeInTheDocument()
     })
+
+    it.each([false, true])(
+        'leaves the sandbox sender out of the picker when the rotation flag is %s',
+        (flagEnabled) => {
+            mockSenderRotationEnabled = flagEnabled
+
+            render(<NativeEmailIntegrationChoice label="From" value={{ integrationId: 1 }} onChange={jest.fn()} />)
+
+            expect(screen.getByText(ALICE.display_name)).toBeInTheDocument()
+            expect(screen.queryByText(SANDBOX.display_name)).not.toBeInTheDocument()
+        }
+    )
+
+    it('treats a project with only the sandbox sender as having no email sender', () => {
+        jest.mocked(useValues).mockReturnValue({ integrationsLoading: false, integrations: [SANDBOX] })
+
+        render(<NativeEmailIntegrationChoice label="From" value={{}} onChange={jest.fn()} />)
+
+        expect(screen.getByText('No email senders configured yet')).toBeInTheDocument()
+    })
 })

frontend/src/scenes/hog-functions/email-templater/EmailTemplater.tsx

@@ -26,6 +26,7 @@
 } from 'lib/components/CyclotronJob/cyclotronJobTemplateSuggestionsLogic'
 import { useFeatureFlag } from 'lib/hooks/useFeatureFlag'
 import { integrationsLogic } from 'lib/integrations/integrationsLogic'
+import { isSandboxEmailSender } from 'lib/integrations/utils'
 import { LemonField } from 'lib/lemon-ui/LemonField'
 import { LemonInput } from 'lib/lemon-ui/LemonInput/LemonInput'
 import { LemonTextArea } from 'lib/lemon-ui/LemonTextArea'
@@ -269,7 +270,7 @@
     )
 }
 
 export function NativeEmailIntegrationChoice({
     label,
     onChange,
     value,
@@ -282,7 +283,7 @@
     const { integrationsLoading, integrations } = useValues(integrationsLogic)
     const { logicProps } = useValues(emailTemplaterLogic)
     const senderRotationEnabled = useFeatureFlag('WORKFLOWS_EMAIL_SENDER_ROTATION')
-    const integrationsOfKind = integrations?.filter((x) => x.kind === 'email')
+    const integrationsOfKind = integrations?.filter((x) => x.kind === 'email' && !isSandboxEmailSender(x))
     const selectedIntegrationIds = value?.integrationIds?.length
         ? value.integrationIds
         : value?.integrationId
@@ -532,7 +533,7 @@
     )
 }
 
 function NativeEmailTemplaterForm({
     mode,
     fieldsHidden,
 }: {

posthog/api/integration.py

@@ -147,7 +147,12 @@
 from products.slack_app.backend.services.slack_auth import SLACK_AUTH_FAILURE_CODES
 from products.tasks.backend.facade.api import get_in_progress_runs_for_github_integration
 from products.tasks.backend.facade.contracts import InProgressGithubRunsDTO
-from products.workflows.backend.facade.api import get_active_workflows_using_integration, lookup_existing_email_tools
+from products.workflows.backend.facade.api import (
+    SandboxSenderUnavailable,
+    ensure_sandbox_email_sender,
+    get_active_workflows_using_integration,
+    lookup_existing_email_tools,
+)
 from products.workflows.backend.facade.contracts import EmailProviderUnavailableError
 from products.workflows.backend.facade.enums import EmailDomainSetupMethod
 
@@ -1533,6 +1538,7 @@ class IntegrationViewSet(
         "github_link_existing",
         "github_oauth_authorize",
         "email_verify",
+        "email_sandbox_sender",
         # Side-effecting POST (emails admins) — a read-only token must not be able to trigger it.
         "request_access",
         "push_tokens",
@@ -1572,7 +1578,13 @@ def dangerously_get_permissions(self):
         ]
         # Adding an integration only requires project membership. Every edit and removal uses the
         # viewset permission class, including the creator exception for Google account removal.
-        if self.action in ("create", "github_link_existing", "github_oauth_authorize", "request_access"):
+        if self.action in (
+            "create",
+            "github_link_existing",
+            "github_oauth_authorize",
+            "request_access",
+            "email_sandbox_sender",
+        ):
             return base_permissions
         if self.action == "refresh_github_repos":
             return [*base_permissions, TeamMemberLightManagementPermission()]
@@ -1586,6 +1598,9 @@ def get_throttles(self):
         return super().get_throttles()
 
     def perform_destroy(self, instance: Integration) -> None:
+        if instance.kind == "email":
+            EmailIntegration(instance).ensure_user_managed()
+
         flows_using_integration = get_active_workflows_using_integration(
             team_id=instance.team_id, integration_id=instance.id
         )
@@ -2710,6 +2725,23 @@ def push_tokens(self, request: ValidatedRequest, **kwargs) -> Response:
             ).data
         )
 
+    @extend_schema(
+        request=None,
+        responses={200: IntegrationSerializer, 404: OpenApiResponse(description="No sandbox sender for this project.")},
+    )
+    @action(methods=["POST"], detail=False, url_path="email_sandbox_sender")
+    def email_sandbox_sender(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        try:
+            sandbox_sender = ensure_sandbox_email_sender(self.team_id)
+        except SandboxSenderUnavailable:
+            raise NotFound("The sandbox sender is not available for this project.")
+        if sandbox_sender.created:
+            report_user_action(
+                cast(User, request.user), "workflows sandbox sender provisioned", team=self.team, request=request
+            )
+        integration = self.get_queryset().get(id=sandbox_sender.integration_id)
+        return Response(IntegrationSerializer(integration, context=self.get_serializer_context()).data)
+
     @extend_schema(
         request=None,
         responses={200: EmailDomainVerificationSerializer},
@@ -2786,14 +2818,13 @@ def _email_domain_setup_method(request: Request, instance: Integration) -> str:
     @extend_schema(responses={200: IntegrationSerializer})
     @action(methods=["PATCH"], detail=True, url_path="email")
     def email_update(self, request, **kwargs) -> Response:
-        instance = self.get_object()
-        config = request.data.get("config", {})
+        email = EmailIntegration(self.get_object())
+        email.ensure_user_managed()
 
-        serializer = NativeEmailIntegrationSerializer(data=config)
+        serializer = NativeEmailIntegrationSerializer(data=request.data.get("config", {}))
         serializer.is_valid(raise_exception=True)
 
-        email = EmailIntegration(instance)
-        email.update_native_integration(serializer.validated_data, instance.team_id)
+        email.update_native_integration(serializer.validated_data, email.integration.team_id)
 
         return Response(IntegrationSerializer(email.integration).data)
 

posthog/api/team/team_config.py

@@ -7,6 +7,7 @@
 
 from posthog.models import Team, TeamRevenueAnalyticsConfig
 from posthog.models.filters.utils import validate_group_type_index
+from posthog.models.integration.email import SANDBOX_EMAIL_PROVIDER
 from posthog.models.integration.model import Integration
 
 from products.access_control.backend.presentation.access_control import UserAccessControlSerializerMixin
@@ -427,6 +428,10 @@ def validate_default_email_integration_id(self, value: int | None) -> int | None
         )
         if sender is None:
             raise serializers.ValidationError("Choose an email sender from this project.")
+        if sender.config.get("provider") == SANDBOX_EMAIL_PROVIDER:
+            raise serializers.ValidationError(
+                "The sandbox sender can't be the project default. Choose a sender on your own verified domain."
+            )
         if not sender.config.get("verified"):
             raise serializers.ValidationError("Verify this sender's domain before making it the default.")
         return value

posthog/models/integration/__init__.py

@@ -38,7 +38,12 @@
     dot_get,
 )
 from .databricks import DatabricksIntegration, DatabricksIntegrationError
-from .email import EmailIntegration, cleanup_ses_identity_on_integration_delete
+from .email import (
+    SANDBOX_EMAIL_INTEGRATION_ID,
+    SANDBOX_EMAIL_PROVIDER,
+    EmailIntegration,
+    cleanup_ses_identity_on_integration_delete,
+)
 from .external_issues import (
     SUPPORTED_EXTERNAL_ISSUE_PROVIDERS,
     external_issue_url,
@@ -221,6 +226,8 @@
     "ClickUpIntegration",
     "EmailIntegration",
     "cleanup_ses_identity_on_integration_delete",
+    "SANDBOX_EMAIL_INTEGRATION_ID",
+    "SANDBOX_EMAIL_PROVIDER",
     "LinearIntegration",
     "SUPPORTED_EXTERNAL_ISSUE_PROVIDERS",
     "external_issue_url",

posthog/models/integration/email.py

@@ -39,6 +39,16 @@ class EmailDomainStatusReport:
     checked_at: datetime
 
 
+SANDBOX_EMAIL_PROVIDER = "sandbox"
+SANDBOX_EMAIL_INTEGRATION_ID = "posthog-sandbox"
+SANDBOX_SENDER_MANAGED_MESSAGE = "The sandbox sender is managed by PostHog."
+
+
+def is_sandbox_sender_domain(domain: str) -> bool:
+    sandbox_domain = settings.WORKFLOWS_SANDBOX_SENDER_DOMAIN
+    return bool(sandbox_domain) and (domain == sandbox_domain or domain.endswith(f".{sandbox_domain}"))
+
+
 class EmailIntegration:
     integration: model.Integration
 
@@ -47,6 +57,14 @@ def __init__(self, integration: model.Integration) -> None:
             raise Exception("EmailIntegration init called with Integration with wrong 'kind'")
         self.integration = integration
 
+    @property
+    def is_sandbox_sender(self) -> bool:
+        return self.integration.config.get("provider") == SANDBOX_EMAIL_PROVIDER
+
+    def ensure_user_managed(self) -> None:
+        if self.is_sandbox_sender:
+            raise ValidationError(SANDBOX_SENDER_MANAGED_MESSAGE)
+
     @classmethod
     def create_native_integration(
         cls, config: dict, team_id: int, organization_id: str, created_by: User | None = None
@@ -60,6 +78,11 @@ def create_native_integration(
         if domain in free_email_domains_list or domain in disposable_email_domains_list:
             raise ValidationError(f"Email domain {domain} is not supported. Please use a custom domain.")
 
+        if is_sandbox_sender_domain(domain):
+            raise ValidationError(
+                f"Email domain {domain} belongs to the PostHog sandbox sender. Please use a domain you own."
+            )
+
         # Check if any other integration already exists in a different team with the same domain,
         # if so, ensure this team is part of the same organization. If not, we block creation.
         same_domain_integrations = model.Integration.objects.filter(kind="email", config__domain=domain)
@@ -152,6 +175,7 @@ def update_native_integration(self, config: dict, team_id: int) -> model.Integra
         return self.integration
 
     def verify(self, *, on_domain_verified: Callable[[], None] | None = None) -> "EmailDomainVerification":
+        self.ensure_user_managed()
         domain = self.integration.config.get("domain")
         provider = self.integration.config.get("provider", "ses")
         mail_from_subdomain = self.integration.config.get("mail_from_subdomain", "feedback")
@@ -192,6 +216,7 @@ def status(
             EmailDomainCheck,  # noqa: PLC0415 — keeps the workflows facade off the model import path
         )
 
+        self.ensure_user_managed()
         cache_key = f"email_domain_status:{self.integration.id}"
         if not refresh and isinstance(cached := self._cached_status(cache_key), str):
             entry = json.loads(cached)

posthog/settings/ses.py

@@ -29,6 +29,10 @@
     if cs.strip()
 ]
 
+WORKFLOWS_SANDBOX_SENDER_DOMAIN: str = os.getenv("WORKFLOWS_SANDBOX_SENDER_DOMAIN", "").strip().lower()
+SES_SANDBOX_FROM_ADDRESS: str = os.getenv("SES_SANDBOX_FROM_ADDRESS", "").strip().lower()
+SES_SANDBOX_TENANT_NAME: str = os.getenv("SES_SANDBOX_TENANT_NAME", "").strip()
+
 # SNS topics allowed to deliver SES tenant reputation events (EventBridge -> SNS -> webhook).
 # Empty (the default) leaves the webhook inert: the SNS signature proves a message came from AWS,
 # but only the allowlist proves it came from *our* topic.

products/workflows/backend/facade/api.py

@@ -15,6 +15,7 @@
     EmailDomainVerification,
     FlowUtmUpdate,
     RecentWorkflow,
+    SandboxSenderUnavailable,
     TeamUtmDefaults,
     TierDecision,
     TwilioAccount,
@@ -47,6 +48,7 @@
 )
 from products.workflows.backend.services.integration_usage import get_active_hog_flows_using_integration
 from products.workflows.backend.services.owned_workflow_health import list_owned_workflow_health
+from products.workflows.backend.services.sandbox_sender import ensure_sandbox_email_sender
 from products.workflows.backend.services.template_input_usage import (
     filter_hog_flow_references_by_access_level,
     get_hog_flows_referencing_template_input_keys,
@@ -64,8 +66,10 @@
     "WorkflowNotFound",
     "MIN_EMAIL_SENDING_TIER",
     "clear_default_email_sender",
+    "SandboxSenderUnavailable",
     "compute_next_occurrences",
     "create_batch_job",
+    "ensure_sandbox_email_sender",
     "ensure_workflows_config",
     "filter_hog_flow_references_by_access_level",
     "get_email_sending_state",

products/workflows/backend/facade/contracts.py

@@ -281,6 +281,16 @@ class EmailSendingTierLimits:
     max_batch_audience: int
 
 
+@frozen
+class SandboxEmailSender:
+    integration_id: int
+    created: bool
+
+
+class SandboxSenderUnavailable(Exception):
+    pass
+
+
 @frozen
 class TierDecision:
     team_id: int