feat(workflows): sandbox test sends, save checks and switch confirm

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

@@ -33,10 +33,12 @@ While the block exists, the endpoint returns 404 for the organization's projects
 Set `WORKFLOWS_SANDBOX_SENDER_DOMAIN`, `SES_SANDBOX_FROM_ADDRESS`, and `SES_SANDBOX_TENANT_NAME` in Django; any empty value leaves the endpoint unavailable and lets every team register any domain as before.
 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.
 With the flag on, opening the Channels page creates the sender and shows it in its own row.
-Workflow email steps offer it as a sender. It replaces any other selected sender and has no Reply-To or custom sender. Broadcasts and other email surfaces do not offer it.
+Workflow email steps and template test sends offer it as a sender. In a step it replaces any other selected sender and has no Reply-To or custom sender. Broadcasts and other email surfaces do not offer it.
 Users cannot edit, verify, or remove the managed integration, make it the project default, use it for transactional email, or register their own integration on the shared domain.
 The managed integration is excluded from project SES tenant migration, identity cleanup, data suggestions, and the reputation page's domain breakdown.
 Configure the shared SES route, then enable `WORKFLOWS_SANDBOX_SENDER_ENABLED` on the worker and CDP API. Enable `workflows-sandbox-sender` last, and only for test projects until the worker reads the tenant pause and organization block in every region.
+The sandbox sender is available only in workflow email steps and test sends. Broadcasts, destinations, sender rotation, custom From values, and Reply-To overrides are rejected. Unchanged stored sender settings remain editable.
+Saves check the sandbox sender only while `workflows-sandbox-sender` is on for the project, so projects without the flag skip the integration lookup.
 
 ### Sending tiers
 

frontend/src/scenes/hog-functions/email-templater/emailTemplaterLogic.test.ts

@@ -1,6 +1,8 @@
 import { router } from 'kea-router'
 import { expectLogic } from 'kea-test-utils'
 
+import { LemonDialog } from '@posthog/lemon-ui'
+
 import { FEATURE_FLAGS } from 'lib/constants'
 import { integrationsLogic } from 'lib/integrations/integrationsLogic'
 import { featureFlagLogic } from 'lib/logic/featureFlagLogic'
@@ -145,30 +147,67 @@ describe('emailTemplaterLogic', () => {
             expect(logic.values.sandboxSender).toBeNull()
         })
 
-        it('picking the sandbox sender replaces the selection and clears the custom sender and Reply-To', async () => {
-            enableSandboxFlag()
-            const onChange = jest.fn()
-            await mountWithSenders({
-                sandboxSenderAllowed: true,
-                liveChanges: true,
-                onChange,
-                value: {
-                    ...DEFAULT_EMAIL_TEMPLATE,
-                    from: { integrationId: 5, email: 'hello@acme.example.com', name: 'Acme' },
-                    replyTo: 'replies@acme.example.com',
-                },
-            })
-
-            logic.actions.chooseSenders([5, 7])
-
-            expect(onChange).toHaveBeenLastCalledWith(
-                expect.objectContaining({
-                    from: { integrationId: 7, integrationIds: undefined, email: undefined, name: undefined },
-                    replyTo: '',
+        it.each([
+            {
+                field: 'a custom address',
+                from: { integrationId: 5, email: 'hello@acme.example.com' },
+                replyTo: '',
+                confirm: true,
+            },
+            { field: 'a custom name', from: { integrationId: 5, name: 'Acme' }, replyTo: '', confirm: true },
+            { field: 'a Reply-To', from: { integrationId: 5 }, replyTo: 'replies@acme.example.com', confirm: true },
+            { field: 'a rotation', from: { integrationId: 5, integrationIds: [5, 6] }, replyTo: '', confirm: true },
+            { field: 'one plain sender', from: { integrationId: 5 }, replyTo: '', confirm: false },
+        ])(
+            'picking the sandbox sender over $field replaces it after any confirmation',
+            async ({ from, replyTo, confirm }) => {
+                enableSandboxFlag()
+                useMocks({
+                    get: {
+                        '/api/projects/:team_id/integrations/': {
+                            results: [
+                                SANDBOX_SENDER,
+                                OWN_SENDER,
+                                { ...OWN_SENDER, id: 6, display_name: 'Second <second@acme.example.com>' },
+                            ],
+                        },
+                    },
                 })
-            )
-            expect(logic.values.isSandboxSenderSelected).toBe(true)
-        })
+                const onChange = jest.fn()
+                const dialog = jest.spyOn(LemonDialog, 'open').mockImplementation(jest.fn())
+                const props = { sandboxSenderAllowed: true, liveChanges: true, onChange }
+                await mountWithSenders({ ...props, value: { ...DEFAULT_EMAIL_TEMPLATE, from, replyTo } })
+                onChange.mockClear()
+
+                logic.actions.chooseSenders([...(from.integrationIds ?? [from.integrationId]), 7])
+
+                expect(dialog).toHaveBeenCalledTimes(confirm ? 1 : 0)
+                if (confirm) {
+                    expect(onChange).not.toHaveBeenCalled()
+                    if (from.integrationIds) {
+                        expect(dialog.mock.calls[0][0].description).toContain(
+                            'This replaces Acme <hello@acme.example.com> and Second <second@acme.example.com>'
+                        )
+                    }
+                    emailTemplaterLogic(
+                        makeProps({
+                            ...props,
+                            value: { ...DEFAULT_EMAIL_TEMPLATE, from, replyTo, subject: 'Changed during confirmation' },
+                        })
+                    )
+                    ;(dialog.mock.calls[0][0].primaryButton?.onClick as () => void)()
+                }
+                expect(onChange).toHaveBeenLastCalledWith(
+                    expect.objectContaining({
+                        from: { integrationId: 7, integrationIds: undefined, email: undefined, name: undefined },
+                        replyTo: '',
+                        subject: confirm ? 'Changed during confirmation' : DEFAULT_EMAIL_TEMPLATE.subject,
+                    })
+                )
+                expect(logic.values.isSandboxSenderSelected).toBe(true)
+                dialog.mockRestore()
+            }
+        )
 
         it('offers no Reply-To while the sandbox sender is selected', async () => {
             enableSandboxFlag()

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

@@ -261,6 +261,22 @@
     sandboxSenderAllowed?: boolean
 }
 
+function joinSenderNames(names: string[]): string {
+    return names.length > 1 ? `${names.slice(0, -1).join(', ')} and ${names[names.length - 1]}` : names[0]
+}
+
+function sandboxSwitchDescription(replacedSenderNames: string[], hasCustomSenderDetails: boolean): string {
+    return [
+        replacedSenderNames.length > 1
+            ? `This replaces ${joinSenderNames(replacedSenderNames)} with the sandbox sender.`
+            : '',
+        hasCustomSenderDetails ? 'This removes your custom sender and Reply-To.' : '',
+        'The sandbox sender sends from its own address.',
+    ]
+        .filter(Boolean)
+        .join(' ')
+}
+
 function autoRevealAdvancedFields(
     actions: { revealAdvancedField: (key: EmailMetaFieldKey) => void },
     props: EmailTemplaterLogicProps
@@ -295,6 +311,7 @@
     emailTemplateTouched: boolean
     emailTemplateTouches: Record<string, boolean>
     emailTemplateValidationErrors: DeepPartialMap<EmailTemplate, ValidationErrorType>
+    hasCustomSenderDetails: boolean
     hiddenAdvancedFields: EmailMetaField[]
     isEmailEditorReady: boolean
     isEmailTemplateSubmitting: boolean
@@ -466,6 +483,7 @@
             emailTemplate: EmailTemplate
         ) => IntegrationType | null
         isSandboxSenderSelected: (sandboxSender: IntegrationType | null, emailTemplate: EmailTemplate) => boolean
+        hasCustomSenderDetails: (emailTemplate: EmailTemplate) => boolean
         supportedFields: (arg: EmailTemplaterType, isSandboxSenderSelected: boolean) => EmailMetaField[]
         visibleFields: (
             supportedFields: EmailMetaField[],
@@ -726,6 +744,13 @@
             (sandboxSender: IntegrationType | null, emailTemplate: EmailTemplate): boolean =>
                 !!sandboxSender && selectedSenderIds(senderFrom(emailTemplate)).includes(sandboxSender.id),
         ],
+        hasCustomSenderDetails: [
+            (s) => [s.emailTemplate],
+            (emailTemplate: EmailTemplate): boolean => {
+                const from = senderFrom(emailTemplate)
+                return !!(from.email || from.name || emailTemplate?.replyTo)
+            },
+        ],
         // The sandbox sender always replies from its own identity, so it offers no Reply-To.
         supportedFields: [
             (s) => [(_, props: EmailTemplaterLogicProps) => props.type, s.isSandboxSenderSelected],
@@ -754,8 +779,25 @@
                 return
             }
             if (sandboxSenderId !== undefined && next.integrationId === sandboxSenderId) {
-                actions.setEmailTemplateValues({ from: next, replyTo: '' })
-                actions.hideAdvancedField('replyTo')
+                const selectSandbox = (): void => {
+                    actions.setEmailTemplateValues({ from: next, replyTo: '' })
+                    actions.hideAdvancedField('replyTo')
+                }
+                const replacedSenderIds = selectedSenderIds(senderFrom(values.emailTemplate))
+                if (!values.hasCustomSenderDetails && replacedSenderIds.length < 2) {
+                    selectSandbox()
+                    return
+                }
+                const integrations = integrationsLogic.isMounted() ? (integrationsLogic.values.integrations ?? []) : []
+                const replacedSenderNames = replacedSenderIds.map(
+                    (id) => integrations.find((sender) => sender.id === id)?.display_name || `sender ${id}`
+                )
+                LemonDialog.open({
+                    title: 'Switch to the sandbox sender?',
+                    description: sandboxSwitchDescription(replacedSenderNames, values.hasCustomSenderDetails),
+                    primaryButton: { children: 'Switch sender', onClick: selectSandbox },
+                    secondaryButton: { children: 'Cancel' },
+                })
                 return
             }
             actions.setEmailTemplateValue('from', next)
@@ -1044,7 +1086,7 @@
         },
     })),
 
     propsChanged(({ actions, props, values, cache }, oldProps) => {
         // A transactional email's merge tags follow the variables it uses, so the open editor gets new ones.
         // Built from the new props here: a selector would still return its cached value until the store changes.
         const mergeTags = buildMergeTags(values.personPropertyDefinitions, props.type, props.variables)

products/cdp/backend/api/hog_function.py

@@ -197,6 +197,27 @@
     return snapshot
 
 
+def _stored_email_values_by_key(instance: Optional[HogFunction]) -> dict[str, dict]:
+    """Stored email input values by input key, so a save that leaves one unchanged, such as disabling the function, keeps working.
+
+    Only inputs the stored schema declares as email count. A dictionary input can hold the same shape, and
+    retyping it to email must not carry its value past the sandbox sender checks.
+    """
+    if instance is None:
+        return {}
+    email_keys = {
+        str(schema.get("key"))
+        for schema in instance.inputs_schema or []
+        if isinstance(schema, dict) and schema.get("type") in ("email", "native_email")
+    }
+    stored_inputs = instance.inputs if isinstance(instance.inputs, dict) else {}
+    return {
+        key: entry["value"]
+        for key, entry in stored_inputs.items()
+        if key in email_keys and isinstance(entry, dict) and isinstance(entry.get("value"), dict)
+    }
+
+
 def _named_warehouse_tables(entries: Any) -> list[Any]:
     """The warehouse tables a filters blob actually names, ignoring the picker's placeholder row.
 
@@ -634,6 +655,7 @@
         ):
             raise serializers.ValidationError({"filters": "Select the materialized view to trigger on."})
         self.context["encrypted_inputs"] = instance.encrypted_inputs if instance else {}
+        self.context["existing_email_values_by_key"] = _stored_email_values_by_key(instance)
 
         template = None
         if data["template_id"]:
@@ -701,7 +723,7 @@
             raise serializers.ValidationError("Cannot modify the type of an existing function")
         return value
 
     def validate(self, attrs):
         team = self.context["get_team"]()
         attrs["team"] = team  # NOTE: This has to be done at this level
         hog_type = self.context["function_type"]
@@ -1272,7 +1294,7 @@
             tiebreakers=("name",),
         )
 
     def safely_get_queryset(self, queryset: QuerySet) -> QuerySet:
         queryset = queryset.exclude(type=HogFunctionType.WAREHOUSE_SOURCE_WEBHOOK.value)
 
         if not (self.action == "partial_update" and self.request.data.get("deleted") is False):

products/cdp/backend/presentation/serializers.py

@@ -8,4 +8,5 @@
     InputsSchemaSerializer as InputsSchemaSerializer,
     InputsSerializer as InputsSerializer,
     require_viewer_access_to_added_actions as require_viewer_access_to_added_actions,
+    validate_sandbox_email_sender as validate_sandbox_email_sender,
 )

products/cdp/backend/validation.py

@@ -5,6 +5,7 @@
 from collections.abc import Iterable
 from typing import Any, Optional
 
+from django.conf import settings
 from django.db import models
 
 from drf_spectacular.utils import extend_schema_field
@@ -18,10 +19,11 @@
 from posthog.hogql.parser import parse_program, parse_string_template
 from posthog.hogql.visitor import TraversingVisitor
 
-from posthog.integrations import IntegrationManager
-from posthog.models.integration import POSTHOG_CONNECT_KIND, WAREHOUSE_SOURCE_TYPE_CONFIG_KEY
+from posthog.integrations import IntegrationManager, IntegrationRecord
+from posthog.models.integration import POSTHOG_CONNECT_KIND, SANDBOX_EMAIL_PROVIDER, WAREHOUSE_SOURCE_TYPE_CONFIG_KEY
 from posthog.models.team.team import Team
 from posthog.models.user import User
+from posthog.permissions import posthog_feature_flag_enabled
 
 from products.actions.backend.models.action import Action
 from products.cdp.backend.filters import (
@@ -93,6 +95,118 @@
     }
 
 
+def _delivery_sender_ids(from_value: dict) -> set[int]:
+    rotation = from_value.get("integrationIds")
+    # Mirrors selectEmailSenderIntegrationId in the worker: a non-empty rotation replaces integrationId.
+    configured = rotation if isinstance(rotation, list) and rotation else [from_value.get("integrationId")]
+    return {
+        int(integration_id)
+        for integration_id in configured
+        if isinstance(integration_id, (int, float))
+        and not isinstance(integration_id, bool)
+        and (not isinstance(integration_id, float) or integration_id.is_integer())
+    }
+
+
+def _sandbox_sender_configured() -> bool:
+    # Mirrors sandbox_sender_configured() in the workflows product, which imports this module.
+    return bool(
+        settings.WORKFLOWS_SANDBOX_SENDER_DOMAIN
+        and settings.SES_SANDBOX_FROM_ADDRESS
+        and settings.SES_SANDBOX_TENANT_NAME
+    )
+
+
+def _sandbox_email_integrations(
+    integration_ids: set[int], context: dict[str, Any]
+) -> dict[int, IntegrationRecord | None]:
+    cache: dict[int, IntegrationRecord | None] = context.setdefault("sandbox_email_integration_cache", {})
+    missing_ids = integration_ids - cache.keys()
+    if missing_ids:
+        integrations = IntegrationManager.email.credentials.get_many(context["get_team"]().id, missing_ids)
+        cache.update({integration_id: integrations.get(integration_id) for integration_id in missing_ids})
+    return cache
+
+
+def _sandbox_sender_enabled(team: Team, context: dict[str, Any]) -> bool:
+    flag_cache: dict[int, bool] = context.setdefault("sandbox_sender_enabled_cache", {})
+    if team.id not in flag_cache:
+        try:
+            flag_cache[team.id] = posthog_feature_flag_enabled(
+                "workflows-sandbox-sender", str(team.uuid), organization_id=team.organization_id, team_id=team.id
+            )
+        except Exception:
+            flag_cache[team.id] = False
+    return flag_cache[team.id]
+
+
+def _stored_email_values(context: dict[str, Any]) -> list[object]:
+    # A workflow step passes its stored email values. A hog function passes them by input key,
+    # so only the value stored under the input being validated counts.
+    stored = list(context.get("existing_email_values") or [])
+    by_key = context.get("existing_email_values_by_key")
+    schema = context.get("schema")
+    if isinstance(by_key, dict) and isinstance(schema, dict) and str(schema.get("key")) in by_key:
+        stored.append(by_key[str(schema.get("key"))])
+    return stored
+
+
+def validate_sandbox_email_sender(email_value: object, context: dict[str, Any]) -> None:
+    # Every check returns before any query or flag evaluation while the sandbox settings are empty,
+    # and before any integration lookup while the project lacks the flag.
+    if not _sandbox_sender_configured():
+        return
+    if not isinstance(email_value, dict) or not isinstance(email_value.get("from"), dict):
+        return
+    integration_ids = _delivery_sender_ids(email_value["from"])
+    get_team = context.get("get_team")
+    if not integration_ids or get_team is None or not _sandbox_sender_enabled(get_team(), context):
+        return
+    # Bounds the lookup below. InputsItemSerializer applies the same limit to every save.
+    if len(integration_ids) > MAX_WORKFLOW_EMAIL_SENDERS:
+        raise serializers.ValidationError({"input": f"At most {MAX_WORKFLOW_EMAIL_SENDERS} email senders are allowed."})
+    integrations = _sandbox_email_integrations(integration_ids, context)
+    if not any(
+        (integration := integrations[integration_id]) is not None
+        and (integration.config or {}).get("provider") == SANDBOX_EMAIL_PROVIDER
+        for integration_id in integration_ids
+    ):
+        return
+    if context.get("workflow_origin_product") == "broadcasts":
+        raise serializers.ValidationError(
+            {"input": "The sandbox sender cannot be used in broadcasts. Select a sender on your own verified domain."}
+        )
+    for stored in _stored_email_values(context):
+        if (
+            isinstance(stored, dict)
+            and email_value["from"] == stored.get("from")
+            and (email_value.get("replyTo") or "") == (stored.get("replyTo") or "")
+        ):
+            return
+    is_test_send = getattr(context.get("view"), "action", None) == "invocations"
+    if context.get("workflow_action_type") != "function_email" and not is_test_send:
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender can only be used in workflow email steps and test sends. "
+                "Select a sender on your own verified domain for destinations."
+            }
+        )
+    if len(integration_ids) != 1:
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender must be the only sender. Remove the other senders "
+                "or select a sender on your own verified domain."
+            }
+        )
+    if email_value["from"].get("email") or email_value["from"].get("name") or email_value.get("replyTo"):
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender uses a fixed From address and name and does not support Reply-To. "
+                "Remove these overrides or select a sender on your own verified domain."
+            }
+        )
+
+
 def _validate_not_posthog_connection(integration_ids: list[int], context: dict) -> None:
     # A PostHog connection acts as the user who created it, so only that user may use it. A function
     # runs for the whole team and the runtime never resolves one, so reject it where the author sees why.
@@ -113,7 +227,7 @@
         )
 
 
 def _validate_email_sender_override(from_value: dict, context: dict) -> None:
     """Reject a literal custom sender address the send path would refuse.
 
     The runtime only sends from an address on the selected integration's verified domain; an
@@ -522,7 +636,7 @@
     return input_collector.inputs
 
 
 def generate_template_bytecode(
     obj: Any,
     input_collector: set[str],
     function_type: Optional[str] = None,
@@ -885,6 +999,7 @@
         elif item_type == "email" or item_type == "native_email":
             if not isinstance(value, dict):
                 raise serializers.ValidationError({"input": f"Value must be an email object."})
+            validate_sandbox_email_sender(value, self.context)
             # Report every missing key in one error: these objects are typically authored
             # programmatically, and a one-at-a-time raise forces a round trip per missing key.
             missing = [f"'{key_}'" for key_ in ("from", "to", "subject") if not value.get(key_)]