feat(workflows): validate sandbox sender use at save time

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

@@ -34,6 +34,8 @@ Workflow email steps and test sends offer it as a sender.
 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.
+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
 

products/cdp/backend/api/hog_function.py

@@ -195,6 +195,16 @@ def snapshot_hog_function_content(hog_function: HogFunction) -> dict:
     return snapshot
 
 
+def _stored_input_values(instance: Optional[HogFunction]) -> list[dict]:
+    """Stored input values, so a save that leaves an email value unchanged, such as disabling the function, keeps working."""
+    stored_inputs = instance.inputs if instance and isinstance(instance.inputs, dict) else {}
+    return [
+        entry["value"]
+        for entry in stored_inputs.values()
+        if 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.
 
@@ -632,6 +642,7 @@ def to_internal_value(self, data):
         ):
             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"] = _stored_input_values(instance)
 
         template = None
         if data["template_id"]:

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/test/test_validation.py

@@ -633,7 +633,7 @@ def _validate_email_from(self, from_value, existing_from=None, get_team=True, ca
         if get_team:
             context_extra["get_team"] = lambda: self.team
         if cache is not None:
-            context_extra["email_integration_domain_cache"] = cache
+            context_extra["email_integration_cache"] = cache
         return validate_inputs(inputs_schema, {"email": {"value": value}}, context_extra=context_extra)
 
     @parameterized.expand(

products/cdp/backend/validation.py

@@ -18,9 +18,10 @@
 from posthog.hogql.parser import parse_program, parse_string_template
 from posthog.hogql.visitor import TraversingVisitor
 
-from posthog.models.integration import POSTHOG_CONNECT_KIND, Integration
+from posthog.models.integration import POSTHOG_CONNECT_KIND, SANDBOX_EMAIL_PROVIDER, Integration
 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 (
@@ -81,13 +82,96 @@ def masked_secret_input_keys(stored_inputs: object) -> list[str]:
 
 
 def _sender_integration_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 {
-        integration_id
-        for integration_id in [from_value.get("integrationId"), *(from_value.get("integrationIds") or [])]
-        if isinstance(integration_id, int) and not isinstance(integration_id, bool)
+        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 _email_integrations(integration_ids: set[int], context: dict[str, Any]) -> dict[int, Integration | None]:
+    cache: dict[int, Integration | None] = context.setdefault("email_integration_cache", {})
+    missing_ids = integration_ids - cache.keys()
+    if missing_ids:
+        integrations = {
+            integration.id: integration
+            for integration in Integration.objects.filter(
+                team_id=context["get_team"]().id, id__in=missing_ids, kind="email"
+            )
+        }
+        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 validate_sandbox_email_sender(email_value: object, context: dict[str, Any]) -> None:
+    if not isinstance(email_value, dict) or not isinstance(email_value.get("from"), dict):
+        return
+    rotation = email_value["from"].get("integrationIds")
+    if isinstance(rotation, list) and len(rotation) > MAX_WORKFLOW_EMAIL_SENDERS:
+        raise serializers.ValidationError({"input": f"At most {MAX_WORKFLOW_EMAIL_SENDERS} email senders are allowed."})
+    integration_ids = _sender_integration_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
+    integrations = _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 context.get("existing_email_values") or []:
+        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.
@@ -146,20 +230,11 @@ def _validate_email_sender_override(from_value: dict, context: dict) -> None:
         return
 
     override_domain = override.split("@")[1].lower()
-    # An empty cached domain means the id resolved to no email integration for this team; the
-    # save is not blocked on it (there is no domain to compare), matching the uncached behavior.
-    shared_cache = context.get("email_integration_domain_cache")
-    domain_cache: dict[int, str] = shared_cache if isinstance(shared_cache, dict) else {}
-    missing_ids = [integration_id for integration_id in integration_ids if integration_id not in domain_cache]
-    if missing_ids:
-        for integration in Integration.objects.filter(team_id=get_team().id, id__in=missing_ids, kind="email"):
-            config = integration.config or {}
-            domain_cache[integration.id] = (config.get("domain") or (config.get("email") or "").split("@")[-1]).lower()
-        for integration_id in missing_ids:
-            domain_cache.setdefault(integration_id, "")
-
+    integrations = _email_integrations(integration_ids, context)
     for integration_id in sorted(integration_ids):
-        integration_domain = domain_cache.get(integration_id) or ""
+        integration = integrations[integration_id]
+        config = (integration.config or {}) if integration is not None else {}
+        integration_domain = (config.get("domain") or (config.get("email") or "").split("@")[-1]).lower()
         if integration_domain and override_domain != integration_domain:
             raise serializers.ValidationError(
                 {
@@ -851,6 +926,7 @@ def validate(self, attrs):
         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_)]

products/workflows/backend/presentation/views/hog_flow.py

@@ -126,6 +126,7 @@
     InputsSchemaSerializer,
     InputsSerializer,
     require_viewer_access_to_added_actions,
+    validate_sandbox_email_sender,
 )
 from products.cohorts.backend.facade.api import get_cohort_ref
 from products.messaging.backend.facade.categories import MessageCategoryMissing, get_category
@@ -1095,14 +1096,14 @@ def _has_exact_string_filter(filters: dict, key: str) -> bool:
     return False
 
 
-def _existing_email_from_by_action(instance: WorkflowEditState) -> dict[str, list[dict]]:
-    """Stored email sender overrides keyed by action id, live and draft variants both kept.
+def _existing_email_values_by_action(instance: WorkflowEditState) -> dict[str, list[tuple[str | None, dict]]]:
+    """Stored email inputs keyed by action id, live and draft variants both kept.
 
     Which variant a save should be compared against depends on how the request routes (a
     builder save targets the draft, a raw API write targets live), so validation grandfathers
     a value matching either - both are values already stored on the workflow.
     """
-    result: dict[str, list[dict]] = {}
+    result: dict[str, list[tuple[str | None, dict]]] = {}
     draft = instance.draft if isinstance(instance.draft, dict) else {}
     for actions in (instance.actions, draft.get("actions")):
         for stored_action in actions or []:
@@ -1111,8 +1112,8 @@ def _existing_email_from_by_action(instance: WorkflowEditState) -> dict[str, lis
             email_input = ((stored_action.get("config") or {}).get("inputs") or {}).get("email")
             value = email_input.get("value") if isinstance(email_input, dict) else None
             from_value = value.get("from") if isinstance(value, dict) else None
-            if isinstance(from_value, dict):
-                result.setdefault(stored_action["id"], []).append(from_value)
+            if isinstance(value, dict) and isinstance(from_value, dict):
+                result.setdefault(stored_action["id"], []).append((stored_action.get("type"), value))
     return result
 
 
@@ -2143,13 +2144,33 @@ def _subscribes_to(event_id: str) -> bool:
             else:
                 input_schema = template.inputs_schema
                 inputs = data.get("config", {}).get("inputs", {})
+                existing_emails = (self.context.get("existing_action_emails") or {}).get(data.get("id")) or []
+                input_context = {
+                    **self.context,
+                    "workflow_action_type": data.get("type"),
+                    "existing_email_values": [
+                        email
+                        for action_type, email in existing_emails
+                        if action_type == "function_email" and data.get("type") == "function_email"
+                    ],
+                    # Request-scoped: a drip sequence's steps share senders, and the actions
+                    # list validates one action at a time (mirrors _message_template_cache).
+                    "email_integration_cache": self.context.setdefault("_email_integration_cache", {}),
+                    "sandbox_sender_enabled_cache": self.context.setdefault("_sandbox_sender_enabled_cache", {}),
+                }
+                for schema in input_schema or []:
+                    if schema.get("type") in {"email", "native_email"} and isinstance(inputs, dict):
+                        email_input = inputs.get(schema["key"])
+                        if isinstance(email_input, dict):
+                            validate_sandbox_email_sender(email_input.get("value"), input_context)
 
                 function_config_serializer = HogFlowConfigFunctionInputsSerializer(
                     data={
                         "inputs_schema": input_schema,
                         "inputs": inputs,
                     },
                     context={
+                        **input_context,
                         "function_type": template.type,
                         "is_dwh_source": self.context.get("is_dwh_source", False),
                         # The existing (decrypted) secret inputs for this action, so a resent
@@ -2160,14 +2181,7 @@ def _subscribes_to(event_id: str) -> bool:
                         # sender address is checked against the integration's verified domain at
                         # save time, while an unchanged stored value is grandfathered.
                         "get_team": self.context.get("get_team"),
-                        "existing_email_from": (self.context.get("existing_action_email_from") or {}).get(
-                            data.get("id")
-                        ),
-                        # Request-scoped: a drip sequence's steps share senders, and the actions
-                        # list validates one action at a time (mirrors _message_template_cache).
-                        "email_integration_domain_cache": self.context.setdefault(
-                            "_email_integration_domain_cache", {}
-                        ),
+                        "existing_email_from": [email["from"] for _, email in existing_emails],
                     },
                 )
 
@@ -3452,6 +3466,7 @@ def to_internal_value(self, data):
         # When used as a nested field (the `configuration` override on test invocations) DRF never
         # binds `self.instance`, so fall back to the flow passed in via context so recovery still works.
         instance = cast(Optional[WorkflowEditState], self.instance) or self.context.get("instance")
+        self.context["workflow_origin_product"] = instance.origin_product if instance else data.get("origin_product")
 
         # Who a "Create AI task" step runs as: the existing creator for an update, or the
         # requesting user for a brand-new flow (matches the `created_by` a create() actually
@@ -3573,7 +3588,7 @@ def to_internal_value(self, data):
             # Stored sender overrides, keyed by action id, so child action validation only holds
             # newly written custom sender addresses to the verified-domain rule. Draft wins over
             # live for the same reason as secrets: it is the value the client last saw.
-            self.context["existing_action_email_from"] = _existing_email_from_by_action(instance)
+            self.context["existing_action_emails"] = _existing_email_values_by_action(instance)
             # Live "Run scout" notes, so only a new or changed note needs skill editor access.
             self.context["live_run_scout_notes"] = _live_run_scout_notes_by_action(instance)