From 70da04a1752b763664fe6206345a6dbb07b163ee Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Mon, 19 Oct 2026 09:27:11 +0000
Subject: [PATCH 1/4] feat(workflows): validate sandbox sender use at save time

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../internal/workflows-email-sending-tiers.md |   1 +
 .../cdp/backend/presentation/serializers.py   |   1 +
 products/cdp/backend/test/test_validation.py  |   2 +-
 products/cdp/backend/validation.py            | 112 +++++-
 .../backend/presentation/views/hog_flow.py    |  41 ++-
 .../backend/services/broadcast_conversion.py  |  55 ++-
 .../tests/api/test_broadcast_conversion.py    |  46 +++
 .../api/test_sandbox_sender_validation.py     | 345 ++++++++++++++++++
 8 files changed, 570 insertions(+), 33 deletions(-)
 create mode 100644 products/workflows/backend/tests/api/test_sandbox_sender_validation.py

diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md
index 46e4d4b9d7aa..4617725bb848 100644
--- a/docs/internal/workflows-email-sending-tiers.md
+++ b/docs/internal/workflows-email-sending-tiers.md
@@ -34,6 +34,7 @@ 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 after the flag turns off.
 
 ### Sending tiers
 
diff --git a/products/cdp/backend/presentation/serializers.py b/products/cdp/backend/presentation/serializers.py
index 78de717168e9..f1841ed412c5 100644
--- a/products/cdp/backend/presentation/serializers.py
+++ b/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,
 )
diff --git a/products/cdp/backend/test/test_validation.py b/products/cdp/backend/test/test_validation.py
index fb45b2764a76..7a3f02513889 100644
--- a/products/cdp/backend/test/test_validation.py
+++ b/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(
diff --git a/products/cdp/backend/validation.py b/products/cdp/backend/validation.py
index c0b8ad0de0ad..5e67bf77ae05 100644
--- a/products/cdp/backend/validation.py
+++ b/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,98 @@ def masked_secret_input_keys(stored_inputs: object) -> list[str]:
 
 
 def _sender_integration_ids(from_value: dict) -> set[int]:
+    rotation = from_value.get("integrationIds")
     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 [from_value.get("integrationId"), *(rotation if isinstance(rotation, list) else [])]
+        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 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:
+        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."}
+        )
+    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."
+            }
+        )
+    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
+    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."
+            }
+        )
+    team = get_team()
+    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
+    if not flag_cache[team.id]:
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender is not available for this project. "
+                "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 +232,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 +928,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_)]
diff --git a/products/workflows/backend/presentation/views/hog_flow.py b/products/workflows/backend/presentation/views/hog_flow.py
index 936369623ff1..e5db93ba9346 100644
--- a/products/workflows/backend/presentation/views/hog_flow.py
+++ b/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,6 +2144,23 @@ 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"
+                    ],
+                    # 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={
@@ -2150,6 +2168,7 @@ def _subscribes_to(event_id: str) -> bool:
                         "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 +2179,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 +3464,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 +3586,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)
 
diff --git a/products/workflows/backend/services/broadcast_conversion.py b/products/workflows/backend/services/broadcast_conversion.py
index 63d72ca384eb..0bb24b6db6d4 100644
--- a/products/workflows/backend/services/broadcast_conversion.py
+++ b/products/workflows/backend/services/broadcast_conversion.py
@@ -6,6 +6,8 @@
 from django.db import models, transaction
 from django.utils import timezone
 
+from posthog.models.integration import SANDBOX_EMAIL_PROVIDER, Integration
+
 from products.access_control.backend.facade.user_access_control import UserAccessControl
 from products.workflows.backend.facade.contracts import WorkflowActor, WorkflowSummary
 from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
@@ -37,9 +39,44 @@ def is_convertible_to_broadcast(flow: HogFlow) -> bool:
     config = trigger.get("config") or {}
     if config.get("type") != BROADCAST_TRIGGER_TYPE or (config.get("filters") or {}).get("audience_type") == "accounts":
         return False
+    if _sender_ids(_email_sender(email)) & _sandbox_sender_ids(flow.team_id):
+        return False
     return _joins_in_order(flow, trigger.get("id"), email.get("id"), exit_step.get("id"))
 
 
+def _email_sender(email_step: dict) -> object:
+    """The step's stored `from` value. Drafts save leniently, so any level can hold another type."""
+    value: object = email_step
+    for key in ("config", "inputs", "email", "value", "from"):
+        value = value.get(key) if isinstance(value, dict) else None
+    return value
+
+
+def _sender_ids(sender: object) -> set[str]:
+    """The integrations a `from` value sends from, as strings, since stored ids can be ints, floats or strings."""
+    if not isinstance(sender, dict):
+        return set()
+    rotation = sender.get("integrationIds")
+    ids = [sender.get("integrationId"), *(rotation if isinstance(rotation, list) else [])]
+    return {_integration_id_key(integration_id) for integration_id in ids if integration_id is not None}
+
+
+def _integration_id_key(integration_id: object) -> str:
+    if isinstance(integration_id, float) and integration_id.is_integer():
+        return str(int(integration_id))
+    return str(integration_id)
+
+
+def _sandbox_sender_ids(team_id: int) -> set[str]:
+    """Broadcasts can't send through the sandbox sender, so a workflow that uses it stays a workflow."""
+    return {
+        str(integration_id)
+        for integration_id in Integration.objects.filter(
+            team_id=team_id, kind="email", config__provider=SANDBOX_EMAIL_PROVIDER
+        ).values_list("id", flat=True)
+    }
+
+
 def _joins_in_order(flow: HogFlow, trigger_id: object, email_id: object, exit_id: object) -> bool:
     """Whether the only edges run trigger to email and email to exit."""
     paths = {(edge.get("from"), edge.get("to")) for edge in flow.edges or [] if isinstance(edge, dict)}
@@ -60,6 +97,15 @@ def _step_count(step_type: str) -> models.Func:
     return models.Func(steps, function="jsonb_array_length", output_field=models.IntegerField())
 
 
+def _email_sender_value() -> models.Func:
+    return models.Func(
+        models.F("actions"),
+        _json_path('$[*] ? (@.type == "function_email").config.inputs.email.value.from'),
+        function="jsonb_path_query_first",
+        output_field=models.JSONField(),
+    )
+
+
 def _step_id(step_type: str) -> models.Func:
     return models.Func(
         models.F("actions"),
@@ -91,6 +137,7 @@ def list_broadcast_conversion_candidates(
             _exit_count=_step_count("exit"),
             _trigger_id=_step_id("trigger"),
             _email_id=_step_id("function_email"),
+            _email_sender=_email_sender_value(),
             _exit_id=_step_id("exit"),
             _account_audience=models.Func(
                 models.F("actions"),
@@ -111,6 +158,7 @@ def list_broadcast_conversion_candidates(
         .only("id", "team_id", "name", "status", "edges", "created_by")
         .order_by("-updated_at", "id")
     )
+    sandbox_sender_ids = _sandbox_sender_ids(team_id)
     found: list[HogFlow] = []
     offset = 0
     while len(found) < limit:
@@ -118,7 +166,12 @@ def list_broadcast_conversion_candidates(
         if not page:
             break
         offset += len(page)
-        page = [flow for flow in page if _joins_in_order(flow, flow._trigger_id, flow._email_id, flow._exit_id)]
+        page = [
+            flow
+            for flow in page
+            if _joins_in_order(flow, flow._trigger_id, flow._email_id, flow._exit_id)
+            and not _sender_ids(flow._email_sender) & sandbox_sender_ids
+        ]
         if user_access_control is not None:
             # Converting needs editor access, so a workflow the caller can only view is not offered.
             user_access_control.preload_object_access_controls(cast("list[models.Model]", page))
diff --git a/products/workflows/backend/tests/api/test_broadcast_conversion.py b/products/workflows/backend/tests/api/test_broadcast_conversion.py
index 54498d98385f..b6bb1bed62d1 100644
--- a/products/workflows/backend/tests/api/test_broadcast_conversion.py
+++ b/products/workflows/backend/tests/api/test_broadcast_conversion.py
@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from typing import Any
 
 from posthog.test.base import APIBaseTest
@@ -7,6 +8,7 @@
 from rest_framework import status
 
 from posthog.models.activity_logging.activity_log import ActivityLog
+from posthog.models.integration import SANDBOX_EMAIL_INTEGRATION_ID, SANDBOX_EMAIL_PROVIDER, Integration
 
 from products.workflows.backend.models import HogFlow
 
@@ -68,6 +70,11 @@ def _convert(self, flow: HogFlow) -> Any:
             ),
             ("extra step", {"actions": [TRIGGER, EMAIL, DELAY, EXIT]}, False),
             ("edge that skips the email", {"edges": [*EDGES, {"from": "trigger_node", "to": "exit_node"}]}, False),
+            (
+                "email step without a sender object",
+                {"actions": [TRIGGER, {**EMAIL, "config": {"inputs": {"email": {"value": "draft"}}}}, EXIT]},
+                True,
+            ),
         ]
     )
     def test_only_workflows_the_broadcast_editor_can_own_are_offered_and_converted(
@@ -91,6 +98,45 @@ def test_only_workflows_the_broadcast_editor_can_own_are_offered_and_converted(
             assert response.status_code == status.HTTP_400_BAD_REQUEST, response.json()
             assert flow.origin_product == overrides.get("origin_product")
 
+    @parameterized.expand(
+        [
+            ("sandbox sender", {"integrationId": "sandbox"}, int, False),
+            ("sandbox sender stored as a whole-number float", {"integrationId": "sandbox"}, float, False),
+            ("sandbox sender stored as a string", {"integrationId": "sandbox"}, str, False),
+            ("sandbox sender in a rotation", {"integrationIds": ["own", "sandbox"]}, int, False),
+            ("own sender", {"integrationId": "own"}, int, True),
+        ]
+    )
+    def test_a_workflow_sending_through_the_sandbox_sender_does_not_become_a_broadcast(
+        self, _name: str, sender: dict[str, Any], stored_as: Callable[[int], Any], convertible: bool
+    ) -> None:
+        integration_ids = {
+            "sandbox": Integration.objects.create(
+                team=self.team,
+                kind="email",
+                integration_id=SANDBOX_EMAIL_INTEGRATION_ID,
+                config={"provider": SANDBOX_EMAIL_PROVIDER, "verified": True},
+            ).id,
+            "own": Integration.objects.create(
+                team=self.team,
+                kind="email",
+                integration_id="sender@example.com",
+                config={"provider": "ses", "domain": "example.com", "verified": True},
+            ).id,
+        }
+        sender_from = {
+            key: [stored_as(integration_ids[name]) for name in value]
+            if isinstance(value, list)
+            else stored_as(integration_ids[value])
+            for key, value in sender.items()
+        }
+        email = {**EMAIL, "config": {"inputs": {"email": {"value": {"from": sender_from}}}}}
+        flow = self._flow(actions=[TRIGGER, email, EXIT])
+
+        assert (str(flow.id) in self._candidate_ids()) is convertible
+        response = self._convert(flow)
+        assert (response.status_code == status.HTTP_200_OK) is convertible, response.json()
+
     def test_candidates_stop_at_the_cap_and_say_more_exist(self) -> None:
         newest = self._flow(name="Newest")
         self._flow(name="Broken", edges=[{"from": "trigger_node", "to": "exit_node"}])
diff --git a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
new file mode 100644
index 000000000000..41fa33bbf220
--- /dev/null
+++ b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
@@ -0,0 +1,345 @@
+from copy import deepcopy
+from types import SimpleNamespace
+from typing import Any
+
+from posthog.test.base import APIBaseTest
+from unittest.mock import patch
+
+from django.db import connection
+from django.test import override_settings
+from django.test.utils import CaptureQueriesContext
+
+from parameterized import parameterized
+
+from posthog.models.integration import Integration
+
+from products.cdp.backend.api.hog_function import HogFunctionInvocationSerializer
+from products.cdp.backend.models.hog_function_template import HogFunctionTemplate
+from products.workflows.backend.facade.api import ensure_sandbox_email_sender
+from products.workflows.backend.facade.workflows import get_workflow_edit_state
+from products.workflows.backend.models.hog_flow import HogFlow
+from products.workflows.backend.presentation.views.hog_flow import HogFlowInvocationSerializer
+
+
+@override_settings(
+    WORKFLOWS_SANDBOX_SENDER_DOMAIN="sandbox.example.com",
+    SES_SANDBOX_FROM_ADDRESS="hello@sandbox.example.com",
+    SES_SANDBOX_TENANT_NAME="posthog-sandbox",
+)
+class TestSandboxSenderValidation(APIBaseTest):
+    def setUp(self) -> None:
+        super().setUp()
+        flag = patch("posthoganalytics.feature_enabled", return_value=True)
+        self.flag_enabled = flag.start()
+        self.addCleanup(flag.stop)
+        capture = patch("posthoganalytics.capture")
+        capture.start()
+        self.addCleanup(capture.stop)
+        self.sender = Integration.objects.get(
+            team_id=self.team.id, id=ensure_sandbox_email_sender(self.team.id).integration_id
+        )
+        self.email_schema = [{"key": "email", "type": "native_email", "required": True, "templating": "liquid"}]
+        HogFunctionTemplate.objects.create(
+            template_id="template-email",
+            name="Email",
+            type="destination",
+            status="hidden",
+            code="return sendEmail(inputs.email);",
+            inputs_schema=self.email_schema,
+        )
+
+    def _workflow(self) -> dict[str, Any]:
+        return {
+            "name": "Sandbox workflow",
+            "actions": [
+                {
+                    "id": "trigger",
+                    "type": "trigger",
+                    "name": "Trigger",
+                    "config": {
+                        "type": "event",
+                        "filters": {"events": [{"id": "$pageview", "type": "events", "order": 0}]},
+                    },
+                },
+                {
+                    "id": "email",
+                    "type": "function_email",
+                    "name": "Send email",
+                    "config": {"inputs": {"email": {"value": self._email()}}},
+                },
+                {"id": "exit", "type": "exit", "name": "Exit", "config": {}},
+            ],
+            "edges": [
+                {"from": "trigger", "to": "email", "type": "continue"},
+                {"from": "email", "to": "exit", "type": "continue"},
+            ],
+        }
+
+    def _email(self) -> dict[str, Any]:
+        return {
+            "from": {"integrationId": self.sender.id},
+            "to": "member@example.com",
+            "subject": "Hello",
+            "html": "<p>Hello</p>",
+        }
+
+    def _function(self) -> dict[str, Any]:
+        return {
+            "name": "Sandbox destination",
+            "type": "destination",
+            "hog": "return 1;",
+            "inputs_schema": self.email_schema,
+            "inputs": {"email": {"value": self._email()}},
+        }
+
+    @parameterized.expand([("integer",), ("float",), ("malformed_rotation",)])
+    def test_broadcast_cannot_select_the_sandbox_sender(self, sender_shape: str) -> None:
+        workflow = self._workflow()
+        workflow["origin_product"] = "broadcasts"
+        sender = workflow["actions"][1]["config"]["inputs"]["email"]["value"]["from"]
+        if sender_shape == "float":
+            sender["integrationId"] = float(self.sender.id)
+        elif sender_shape == "malformed_rotation":
+            sender["integrationIds"] = self.sender.id
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+
+        assert response.status_code == 400, response.json()
+        assert "sandbox sender" in response.json()["detail"].lower()
+        assert "broadcast" in response.json()["detail"].lower()
+
+    def test_destination_cannot_select_the_sandbox_sender(self) -> None:
+        response = self.client.post(
+            f"/api/projects/{self.team.id}/hog_functions/",
+            self._function(),
+        )
+
+        assert response.status_code == 400, response.json()
+        assert "sandbox sender" in response.json()["detail"].lower()
+        assert "workflow email steps and test sends" in response.json()["detail"].lower()
+
+    def test_generic_workflow_step_cannot_select_the_sandbox_sender_with_a_legacy_email_schema(self) -> None:
+        template = HogFunctionTemplate.objects.get(template_id="template-email")
+        template.inputs_schema = [{**self.email_schema[0], "type": "email"}]
+        template.save()
+        workflow = self._workflow()
+        workflow["actions"][1]["type"] = "function"
+        workflow["actions"][1]["config"]["template_id"] = "template-email"
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+
+        assert response.status_code == 400, response.json()
+        assert "workflow email steps and test sends" in response.json()["detail"].lower()
+
+    def test_workflow_can_select_the_sandbox_sender_alone(self) -> None:
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", self._workflow())
+
+        assert response.status_code == 201, response.json()
+        assert response.json()["actions"][1]["config"]["inputs"]["email"]["value"]["from"] == {
+            "integrationId": self.sender.id
+        }
+
+    @parameterized.expand([("exit",), ("function",)])
+    def test_inert_email_data_is_not_an_existing_sandbox_sender(self, action_type: str) -> None:
+        workflow = self._workflow()
+        action = workflow["actions"][1]
+        action["type"] = action_type
+        action["config"]["template_id"] = "missing-template"
+        email = action["config"]["inputs"]["email"]["value"]
+        email["from"]["name"] = "New sender"
+        email["replyTo"] = "reply@example.com"
+        created = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+        assert created.status_code == 201, created.json()
+        action["type"] = "function_email"
+        action["config"].pop("template_id")
+
+        response = self.client.patch(
+            f"/api/projects/{self.team.id}/hog_flows/{created.json()['id']}",
+            {"actions": workflow["actions"]},
+        )
+
+        assert response.status_code == 400, response.json()
+        assert "fixed From address and name" in response.json()["detail"]
+
+    def test_sandbox_sender_cannot_rotate_with_own_senders(self) -> None:
+        own_sender = Integration.objects.create(
+            team=self.team,
+            kind="email",
+            config={"provider": "ses", "domain": "example.com", "email": "sender@example.com", "verified": True},
+        )
+        workflow = self._workflow()
+        workflow["actions"][1]["config"]["inputs"]["email"]["value"]["from"] = {
+            "integrationIds": [self.sender.id, own_sender.id]
+        }
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+
+        assert response.status_code == 400, response.json()
+        assert "sandbox sender must be the only sender" in response.json()["detail"].lower()
+
+    def test_oversized_sender_rotation_is_rejected_before_looking_up_integrations(self) -> None:
+        workflow = self._workflow()
+        workflow["actions"][1]["config"]["inputs"]["email"]["value"]["from"]["integrationIds"] = list(range(100))
+
+        with CaptureQueriesContext(connection) as queries:
+            response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+
+        assert response.status_code == 400, response.json()
+        assert "At most 10 email senders" in response.json()["detail"]
+        assert not [query for query in queries if 'FROM "posthog_integration"' in query["sql"]]
+
+    @parameterized.expand(
+        [
+            ("custom From address", "email", "custom@sandbox.example.com"),
+            ("templated From address", "email", "{{ person.properties.email }}"),
+            ("custom From name", "name", "Custom sender"),
+            ("Reply-To", "replyTo", "reply@example.com"),
+        ]
+    )
+    def test_sandbox_sender_cannot_use_sender_overrides(self, _name: str, field: str, value: str) -> None:
+        workflow = self._workflow()
+        email = workflow["actions"][1]["config"]["inputs"]["email"]["value"]
+        if field == "replyTo":
+            email[field] = value
+        else:
+            email["from"][field] = value
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+
+        assert response.status_code == 400, response.json()
+        assert "fixed From address and name" in response.json()["detail"]
+        assert "Reply-To" in response.json()["detail"]
+
+    def test_workflow_cannot_select_the_sandbox_sender_after_the_flag_turns_off(self) -> None:
+        self.flag_enabled.return_value = False
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", self._workflow())
+
+        assert response.status_code == 400, response.json()
+        assert "sandbox sender is not available for this project" in response.json()["detail"].lower()
+
+    def test_unchanged_sandbox_sender_keeps_saving_after_the_flag_turns_off(self) -> None:
+        created = self.client.post(f"/api/projects/{self.team.id}/hog_flows", self._workflow())
+        assert created.status_code == 201, created.json()
+        actions = created.json()["actions"]
+        actions[1]["config"]["inputs"]["email"]["value"]["subject"] = "Updated subject"
+        self.flag_enabled.return_value = False
+
+        response = self.client.patch(
+            f"/api/projects/{self.team.id}/hog_flows/{created.json()['id']}",
+            {"name": "Renamed workflow", "actions": actions},
+        )
+
+        assert response.status_code == 200, response.json()
+        email = response.json()["actions"][1]["config"]["inputs"]["email"]["value"]
+        assert email["from"] == {"integrationId": self.sender.id}
+        assert email["subject"] == "Updated subject"
+
+    def test_test_send_configuration_can_select_the_sandbox_sender(self) -> None:
+        serializer = HogFunctionInvocationSerializer(
+            data={"configuration": self._function()},
+            context={"get_team": lambda: self.team, "view": SimpleNamespace(action="invocations")},
+        )
+
+        assert serializer.is_valid(), serializer.errors
+        assert serializer.validated_data["configuration"]["inputs"]["email"]["value"]["from"] == {
+            "integrationId": self.sender.id
+        }
+
+    @parameterized.expand([("missing",), ("cleared",)])
+    def test_broadcast_test_cannot_clear_its_origin(self, origin: str) -> None:
+        workflow = self._workflow()
+        stored = HogFlow.objects.create(
+            team=self.team,
+            name="Broadcast",
+            origin_product="broadcasts",
+            actions=workflow["actions"],
+            edges=workflow["edges"],
+        )
+        if origin == "cleared":
+            workflow["origin_product"] = None
+        serializer = HogFlowInvocationSerializer(
+            data={"configuration": workflow},
+            context={
+                "get_team": lambda: self.team,
+                "view": SimpleNamespace(action="invocations"),
+                "instance": get_workflow_edit_state(
+                    team_id=self.team.id, workflow_id=stored.id, user_access_control=None, required_level=None
+                ),
+            },
+        )
+
+        assert not serializer.is_valid(), serializer.validated_data
+        assert "broadcast" in str(serializer.errors).lower()
+
+    @parameterized.expand([("live",), ("draft",)])
+    def test_unchanged_legacy_sender_survives_unrelated_edits(self, variant: str) -> None:
+        live = self._workflow()
+        live_email = live["actions"][1]["config"]["inputs"]["email"]["value"]
+        live_email["from"].update({"email": "legacy@example.com", "name": "Legacy sender"})
+        live_email["replyTo"] = "legacy-reply@example.com"
+        draft = deepcopy(live)
+        draft_email = draft["actions"][1]["config"]["inputs"]["email"]["value"]
+        draft_email["from"]["name"] = "Draft legacy sender"
+        draft_email["replyTo"] = "draft-reply@example.com"
+        stored = HogFlow.objects.create(
+            team=self.team,
+            name="Legacy workflow",
+            actions=live["actions"],
+            edges=live["edges"],
+            status=HogFlow.State.ACTIVE,
+            draft=draft,
+        )
+        actions = deepcopy(live["actions"] if variant == "live" else draft["actions"])
+        actions[1]["config"]["inputs"]["email"]["value"]["subject"] = "Updated legacy subject"
+        self.flag_enabled.return_value = False
+
+        response = self.client.patch(
+            f"/api/projects/{self.team.id}/hog_flows/{stored.id}",
+            {"name": "Renamed legacy workflow", "actions": actions},
+        )
+
+        assert response.status_code == 200, response.json()
+
+    @parameterized.expand([("email",), ("name",), ("replyTo",)])
+    def test_legacy_sender_overrides_cannot_be_changed(self, field: str) -> None:
+        workflow = self._workflow()
+        email = workflow["actions"][1]["config"]["inputs"]["email"]["value"]
+        email["from"].update({"email": "legacy@example.com", "name": "Legacy sender"})
+        email["replyTo"] = "legacy-reply@example.com"
+        stored = HogFlow.objects.create(
+            team=self.team,
+            name="Legacy workflow",
+            actions=workflow["actions"],
+            edges=workflow["edges"],
+            status=HogFlow.State.ACTIVE,
+        )
+        if field == "replyTo":
+            email[field] = "changed@example.com"
+        else:
+            email["from"][field] = "changed@example.com" if field == "email" else "Changed sender"
+
+        response = self.client.patch(
+            f"/api/projects/{self.team.id}/hog_flows/{stored.id}",
+            {"actions": workflow["actions"]},
+        )
+
+        assert response.status_code == 400, response.json()
+        assert "fixed From address and name" in response.json()["detail"]
+
+    def test_malformed_sender_rotation_returns_a_validation_error(self) -> None:
+        workflow = self._workflow()
+        workflow["actions"][1]["config"]["inputs"]["email"]["value"]["from"]["integrationIds"] = self.sender.id
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow, HTTP_X_POSTHOG_CLIENT="mcp")
+
+        assert response.status_code == 400, response.json()
+        assert "list of Integration IDs" in response.json()["detail"]
+
+    def test_null_email_inputs_return_a_validation_error(self) -> None:
+        workflow = self._workflow()
+        workflow["actions"][1]["config"]["inputs"] = None
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow, HTTP_X_POSTHOG_CLIENT="mcp")
+
+        assert response.status_code == 400, response.json()

From 2526f22f053fe5a487b2404e4a3969c409d18796 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Tue, 27 Oct 2026 06:41:43 +0000
Subject: [PATCH 2/4] fix(workflows): match worker sender selection and keep
 cleanup saves working

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 products/cdp/backend/api/hog_function.py      | 11 +++++++
 products/cdp/backend/validation.py            | 18 +++++-----
 .../backend/services/broadcast_conversion.py  |  7 ++--
 .../tests/api/test_broadcast_conversion.py    |  6 ++++
 .../api/test_sandbox_sender_validation.py     | 33 +++++++++++++++++++
 5 files changed, 65 insertions(+), 10 deletions(-)

diff --git a/products/cdp/backend/api/hog_function.py b/products/cdp/backend/api/hog_function.py
index 1e788be88e0c..d1cf34805cb5 100644
--- a/products/cdp/backend/api/hog_function.py
+++ b/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"]:
diff --git a/products/cdp/backend/validation.py b/products/cdp/backend/validation.py
index 5e67bf77ae05..77e434c6470b 100644
--- a/products/cdp/backend/validation.py
+++ b/products/cdp/backend/validation.py
@@ -83,9 +83,11 @@ 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 {
         int(integration_id)
-        for integration_id in [from_value.get("integrationId"), *(rotation if isinstance(rotation, list) else [])]
+        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())
@@ -127,6 +129,13 @@ def validate_sandbox_email_sender(email_value: object, context: dict[str, Any])
         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(
@@ -135,13 +144,6 @@ def validate_sandbox_email_sender(email_value: object, context: dict[str, Any])
                 "Select a sender on your own verified domain for destinations."
             }
         )
-    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
     if len(integration_ids) != 1:
         raise serializers.ValidationError(
             {
diff --git a/products/workflows/backend/services/broadcast_conversion.py b/products/workflows/backend/services/broadcast_conversion.py
index 0bb24b6db6d4..eb150b2c5350 100644
--- a/products/workflows/backend/services/broadcast_conversion.py
+++ b/products/workflows/backend/services/broadcast_conversion.py
@@ -53,11 +53,14 @@ def _email_sender(email_step: dict) -> object:
 
 
 def _sender_ids(sender: object) -> set[str]:
-    """The integrations a `from` value sends from, as strings, since stored ids can be ints, floats or strings."""
+    """The integrations a `from` value sends from, as the worker picks them: a non-empty rotation replaces `integrationId`.
+
+    Ids are compared as strings, since stored ids can be ints, floats or strings.
+    """
     if not isinstance(sender, dict):
         return set()
     rotation = sender.get("integrationIds")
-    ids = [sender.get("integrationId"), *(rotation if isinstance(rotation, list) else [])]
+    ids = rotation if isinstance(rotation, list) and rotation else [sender.get("integrationId")]
     return {_integration_id_key(integration_id) for integration_id in ids if integration_id is not None}
 
 
diff --git a/products/workflows/backend/tests/api/test_broadcast_conversion.py b/products/workflows/backend/tests/api/test_broadcast_conversion.py
index b6bb1bed62d1..24dad6eb6238 100644
--- a/products/workflows/backend/tests/api/test_broadcast_conversion.py
+++ b/products/workflows/backend/tests/api/test_broadcast_conversion.py
@@ -105,6 +105,12 @@ def test_only_workflows_the_broadcast_editor_can_own_are_offered_and_converted(
             ("sandbox sender stored as a string", {"integrationId": "sandbox"}, str, False),
             ("sandbox sender in a rotation", {"integrationIds": ["own", "sandbox"]}, int, False),
             ("own sender", {"integrationId": "own"}, int, True),
+            (
+                "own rotation with a stale sandbox fallback",
+                {"integrationId": "sandbox", "integrationIds": ["own"]},
+                int,
+                True,
+            ),
         ]
     )
     def test_a_workflow_sending_through_the_sandbox_sender_does_not_become_a_broadcast(
diff --git a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
index 41fa33bbf220..d1cf1ec68b0a 100644
--- a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
+++ b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
@@ -15,6 +15,7 @@
 
 from products.cdp.backend.api.hog_function import HogFunctionInvocationSerializer
 from products.cdp.backend.models.hog_function_template import HogFunctionTemplate
+from products.cdp.backend.models.hog_functions.hog_function import HogFunction
 from products.workflows.backend.facade.api import ensure_sandbox_email_sender
 from products.workflows.backend.facade.workflows import get_workflow_edit_state
 from products.workflows.backend.models.hog_flow import HogFlow
@@ -118,6 +119,38 @@ def test_destination_cannot_select_the_sandbox_sender(self) -> None:
         assert "sandbox sender" in response.json()["detail"].lower()
         assert "workflow email steps and test sends" in response.json()["detail"].lower()
 
+    @parameterized.expand([("disabled", {"enabled": False}), ("deleted", {"deleted": True})])
+    def test_stored_destination_with_the_sandbox_sender_can_still_be(self, _name: str, change: dict) -> None:
+        function = HogFunction.objects.create(
+            team=self.team,
+            name="Stored destination",
+            type="destination",
+            hog="return 1;",
+            inputs_schema=self.email_schema,
+            inputs={"email": {"value": self._email()}},
+            enabled=True,
+        )
+
+        response = self.client.patch(f"/api/projects/{self.team.id}/hog_functions/{function.id}/", change)
+
+        assert response.status_code == 200, response.json()
+
+    def test_destination_with_an_own_rotation_ignores_a_stale_sandbox_fallback(self) -> None:
+        own_sender = Integration.objects.create(
+            team=self.team,
+            kind="email",
+            config={"provider": "ses", "domain": "example.com", "email": "sender@example.com", "verified": True},
+        )
+        function = self._function()
+        function["inputs"]["email"]["value"]["from"] = {
+            "integrationId": self.sender.id,
+            "integrationIds": [own_sender.id],
+        }
+
+        response = self.client.post(f"/api/projects/{self.team.id}/hog_functions/", function)
+
+        assert response.status_code == 201, response.json()
+
     def test_generic_workflow_step_cannot_select_the_sandbox_sender_with_a_legacy_email_schema(self) -> None:
         template = HogFunctionTemplate.objects.get(template_id="template-email")
         template.inputs_schema = [{**self.email_schema[0], "type": "email"}]

From 30aef2fc2f975ca989829f76424a2d5a423e2ca4 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Tue, 27 Oct 2026 08:08:36 +0000
Subject: [PATCH 3/4] fix(workflows): keep the step-type rule when a sandbox
 email step changes type

A stored email step that turns into a generic function step no longer keeps
its sandbox sender as an unchanged value, so the step-type rule rejects it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../backend/presentation/views/hog_flow.py      |  4 +++-
 .../tests/api/test_sandbox_sender_validation.py | 17 +++++++++++++++--
 2 files changed, 18 insertions(+), 3 deletions(-)

diff --git a/products/workflows/backend/presentation/views/hog_flow.py b/products/workflows/backend/presentation/views/hog_flow.py
index e5db93ba9346..986e2914d06e 100644
--- a/products/workflows/backend/presentation/views/hog_flow.py
+++ b/products/workflows/backend/presentation/views/hog_flow.py
@@ -2149,7 +2149,9 @@ def _subscribes_to(event_id: str) -> bool:
                     **self.context,
                     "workflow_action_type": data.get("type"),
                     "existing_email_values": [
-                        email for action_type, email in existing_emails if action_type == "function_email"
+                        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).
diff --git a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
index d1cf1ec68b0a..8d21545e6037 100644
--- a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
+++ b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
@@ -151,15 +151,28 @@ def test_destination_with_an_own_rotation_ignores_a_stale_sandbox_fallback(self)
 
         assert response.status_code == 201, response.json()
 
-    def test_generic_workflow_step_cannot_select_the_sandbox_sender_with_a_legacy_email_schema(self) -> None:
+    @parameterized.expand([("created",), ("changed_from_an_email_step",)])
+    def test_generic_workflow_step_cannot_select_the_sandbox_sender_with_a_legacy_email_schema(
+        self, origin: str
+    ) -> None:
         template = HogFunctionTemplate.objects.get(template_id="template-email")
         template.inputs_schema = [{**self.email_schema[0], "type": "email"}]
         template.save()
         workflow = self._workflow()
+        created_id = None
+        if origin == "changed_from_an_email_step":
+            created = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+            assert created.status_code == 201, created.json()
+            created_id = created.json()["id"]
         workflow["actions"][1]["type"] = "function"
         workflow["actions"][1]["config"]["template_id"] = "template-email"
 
-        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+        if created_id is None:
+            response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", workflow)
+        else:
+            response = self.client.patch(
+                f"/api/projects/{self.team.id}/hog_flows/{created_id}", {"actions": workflow["actions"]}
+            )
 
         assert response.status_code == 400, response.json()
         assert "workflow email steps and test sends" in response.json()["detail"].lower()

From d233d18682a6e94ecf8bc24dde4a924b43a62689 Mon Sep 17 00:00:00 2001
From: sim <sim@example.com>
Date: Wed, 28 Oct 2026 09:52:37 +0000
Subject: [PATCH 4/4] feat(workflows): check the sandbox sender on save only
 behind its flag

Projects without workflows-sandbox-sender skip the sandbox sender
checks and their integration lookup, so the save path stays as it was
for everyone else.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../internal/workflows-email-sending-tiers.md |  3 +-
 products/cdp/backend/validation.py            | 30 ++++++++-----------
 .../api/test_sandbox_sender_validation.py     | 29 +++++-------------
 3 files changed, 22 insertions(+), 40 deletions(-)

diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md
index 4617725bb848..f85b1832321d 100644
--- a/docs/internal/workflows-email-sending-tiers.md
+++ b/docs/internal/workflows-email-sending-tiers.md
@@ -34,7 +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 after the flag turns off.
+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
 
diff --git a/products/cdp/backend/validation.py b/products/cdp/backend/validation.py
index 77e434c6470b..c09fe55f4219 100644
--- a/products/cdp/backend/validation.py
+++ b/products/cdp/backend/validation.py
@@ -108,6 +108,18 @@ def _email_integrations(integration_ids: set[int], context: dict[str, Any]) -> d
     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
@@ -116,7 +128,7 @@ def validate_sandbox_email_sender(email_value: object, context: dict[str, Any])
         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:
+    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(
@@ -158,22 +170,6 @@ def validate_sandbox_email_sender(email_value: object, context: dict[str, Any])
                 "Remove these overrides or select a sender on your own verified domain."
             }
         )
-    team = get_team()
-    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
-    if not flag_cache[team.id]:
-        raise serializers.ValidationError(
-            {
-                "input": "The sandbox sender is not available for this project. "
-                "Select a sender on your own verified domain."
-            }
-        )
 
 
 def _validate_not_posthog_connection(integration_ids: list[int], context: dict) -> None:
diff --git a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
index 8d21545e6037..ccc67fb4dd30 100644
--- a/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
+++ b/products/workflows/backend/tests/api/test_sandbox_sender_validation.py
@@ -256,30 +256,16 @@ def test_sandbox_sender_cannot_use_sender_overrides(self, _name: str, field: str
         assert "fixed From address and name" in response.json()["detail"]
         assert "Reply-To" in response.json()["detail"]
 
-    def test_workflow_cannot_select_the_sandbox_sender_after_the_flag_turns_off(self) -> None:
+    @parameterized.expand([("hog_flows",), ("hog_functions/",)])
+    def test_saves_skip_the_sandbox_checks_while_the_flag_is_off(self, endpoint: str) -> None:
         self.flag_enabled.return_value = False
+        payload = self._workflow() if endpoint == "hog_flows" else self._function()
 
-        response = self.client.post(f"/api/projects/{self.team.id}/hog_flows", self._workflow())
-
-        assert response.status_code == 400, response.json()
-        assert "sandbox sender is not available for this project" in response.json()["detail"].lower()
-
-    def test_unchanged_sandbox_sender_keeps_saving_after_the_flag_turns_off(self) -> None:
-        created = self.client.post(f"/api/projects/{self.team.id}/hog_flows", self._workflow())
-        assert created.status_code == 201, created.json()
-        actions = created.json()["actions"]
-        actions[1]["config"]["inputs"]["email"]["value"]["subject"] = "Updated subject"
-        self.flag_enabled.return_value = False
-
-        response = self.client.patch(
-            f"/api/projects/{self.team.id}/hog_flows/{created.json()['id']}",
-            {"name": "Renamed workflow", "actions": actions},
-        )
+        with CaptureQueriesContext(connection) as queries:
+            response = self.client.post(f"/api/projects/{self.team.id}/{endpoint}", payload)
 
-        assert response.status_code == 200, response.json()
-        email = response.json()["actions"][1]["config"]["inputs"]["email"]["value"]
-        assert email["from"] == {"integrationId": self.sender.id}
-        assert email["subject"] == "Updated subject"
+        assert response.status_code == 201, response.json()
+        assert not [query for query in queries.captured_queries if '"posthog_integration"' in query["sql"]]
 
     def test_test_send_configuration_can_select_the_sandbox_sender(self) -> None:
         serializer = HogFunctionInvocationSerializer(
@@ -338,7 +324,6 @@ def test_unchanged_legacy_sender_survives_unrelated_edits(self, variant: str) ->
         )
         actions = deepcopy(live["actions"] if variant == "live" else draft["actions"])
         actions[1]["config"]["inputs"]["email"]["value"]["subject"] = "Updated legacy subject"
-        self.flag_enabled.return_value = False
 
         response = self.client.patch(
             f"/api/projects/{self.team.id}/hog_flows/{stored.id}",