feat(messaging): send code emails through the sandbox sender without a domain

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

@@ -36,7 +36,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.
+The sandbox sender is available only in workflow email steps, test sends, and code sends that name no `from` (see [Sandbox sends from code](#sandbox-sends-from-code)). 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
@@ -101,13 +101,14 @@ Every To, CC, and BCC address must belong to an active member of the team's acti
 Membership snapshots expire within one minute, and the worker checks recipients again right before delivery.
 A failed or stale membership check skips the send and lists the blocked addresses in the run log.
 The blocked event fires at most once an hour per team, reason, and send kind (workflow or test).
+The SES envelope lists each address once, compared without case, so an address repeated across To, CC, and BCC gets one copy.
 A send that passes every check goes out through the fixed SES tenant, configuration set, and From address, whatever the step's custom sender and Reply-To say.
 Sandbox sends skip the team's tier budget, click and open tracking, and unsubscribe headers, because they reach only members of the sending organization.
 The worker parses and normalizes the HTML, then appends a footer that names the sender and the organization ID.
 It overrides root clipping and visibility styles and keeps the template's typography, so malformed markup may render differently from an own-domain send.
 A template whose HTML cannot keep the footer fails the send.
 The visible footer is best effort across email clients and template CSS; the signed tracking header still attributes the message.
-Right before delivery, every sandbox send, test sends included, claims a daily allowance per project (`WORKFLOWS_SANDBOX_DAILY_TEAM_CAP`, one token per recipient copy) and per recipient address across the organization (`WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP`).
+Right before delivery, every sandbox send, test sends included, claims a daily allowance per project (`WORKFLOWS_SANDBOX_DAILY_TEAM_CAP`, one token per distinct recipient) and per recipient address across the organization (`WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP`).
 The allowances are token buckets that refill their capacity over 24 hours, not calendar-day counters.
 A reached allowance skips the send instead of delaying it. A limiter error, or an empty or invalid cap value, skips every sandbox send.
 The worker checks members after the claim and gives the claim back when that check skips the send.
@@ -125,3 +126,30 @@ A successful live sandbox send records `email_sandbox_sent` next to `email_sent`
 The sending tier subtracts `email_sandbox_sent` from `email_sent` per day, and the team's sending allowance per workflow, both floored at zero. Sandbox bounces and complaints still count.
 The CDP API's SES webhook captures `workflows sandbox email delivered` for a delivery on `SES_SANDBOX_CONFIGURATION_SET`, test sends included. Without that setting on the CDP API, the event never fires.
 Deploy this metric to the worker and Django before live workflow sends use the switch in production.
+
+### Sandbox sends from code
+
+A code send (`POST /api/projects/:project_id/messaging/send/email/` with a `template_key`) that names no `from` goes through the sandbox sender when the project has no verified sender of its own.
+Django then provisions the sandbox integration with the provisioning endpoint's checks, so `workflows-sandbox-sender`, the organization block, and the three sandbox settings all apply.
+When one of them fails, the send gets `sender_not_verified`, the same as before the sandbox sender existed, so a blocked organization learns nothing about the block.
+A project secret API key with the send scope can provision the sandbox integration this way. Only the provisioning endpoint captures `workflows sandbox sender provisioned`.
+The CDP API and the worker read an integration the read replica doesn't have yet from the primary, so the first send after provisioning finds it.
+A project with a verified sender of its own but no default keeps the `sender_not_verified` refusal.
+A send by `template_id` keeps the sender its workflow stores.
+A `from` that names the sandbox address, or any address on its domain, is refused with `sender_not_verified` on `from`, and a `reply_to` on a sandbox send is refused with `invalid_request` on `reply_to`.
+
+The CDP API runs the worker's checks on a sandbox code send before it queues it, after the idempotency replay:
+
+| Check                                                                                                                  | Answer                                                      |
+| ---------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------- |
+| A recipient in To, CC, or BCC is not a verified member, or a stored copy address can't be read without the send's data | `400 sandbox_recipient_not_allowed`                         |
+| The project or recipient allowance is used up (a read-only peek, so the check claims nothing)                          | `400 sandbox_limit_reached`                                 |
+| The worker switch is off, the shared tenant is paused, or the organization is blocked after Django resolved the sender | `422 sending_suspended`                                     |
+| The members, the pause state, or the allowance can't be read                                                           | `503 delivery_unavailable` in Django, and nothing is queued |
+
+Each refusal emits `workflows sandbox email blocked` with `source: api`.
+The worker still runs every check again and claims the allowance at delivery, so the CDP API check can let through a send that the worker then skips.
+The 202 answer carries `delivery_mode` (`own` or `sandbox`), and the idempotency record stores it, so a replay reports the mode of the first send.
+Django adds `delivery_mode` to the `transactional email api send` usage event and to the `email first code send` event. The API response doesn't carry it.
+
+Deploy the CDP API change before the Django change. A CDP API without the checks accepts a sandbox code send with 202, and only the worker checks then decide whether it goes out.

products/messaging/backend/facade/contracts.py

@@ -112,6 +112,19 @@ class TemplateOutdated(TransactionalEmailError):
     code = "template_outdated"
 
 
+class SandboxRecipientNotAllowed(TransactionalEmailError):
+    code = "sandbox_recipient_not_allowed"
+
+
+class SandboxLimitReached(TransactionalEmailError):
+    code = "sandbox_limit_reached"
+
+
+class SandboxReplyToNotAllowed(TransactionalEmailError):
+    code = "invalid_request"
+    attr = "reply_to"
+
+
 @frozen
 class CodeEmail:
     key: str
@@ -138,11 +151,21 @@ class WorkflowVersionBehind(DeliveryUnavailable):
     pass
 
 
+class SandboxCheckFailed(DeliveryUnavailable):
+    pass
+
+
+class DeliveryMode(LabeledStrEnum):
+    OWN = "own", "Own sender"
+    SANDBOX = "sandbox", "Sandbox sender"
+
+
 @frozen
 class QueuedEmail:
     message_id: str
     # True when an earlier request with the same idempotency key already queued this email.
     replayed: bool = False
+    delivery_mode: DeliveryMode | None = None
 
 
 class SupabaseEmailActionType(LabeledStrEnum):

products/messaging/backend/facade/transactional_email.py

@@ -6,6 +6,7 @@
     EmailDeliveryStatus,
     EmailStatus,
     QueuedEmail,
+    SandboxCheckFailed,
     TemplateOutdated,
     TemplateSendOptions,
     TransactionalEmailError,
@@ -35,6 +36,7 @@
     "EmailDeliveryStatus",
     "EmailStatus",
     "QueuedEmail",
+    "SandboxCheckFailed",
     "TemplateOutdated",
     "TemplateSendOptions",
     "TransactionalEmailError",

products/messaging/backend/presentation/views/message_templates.py

@@ -91,6 +91,7 @@
     CodeEmail,
     DeliveryUnavailable,
     QueuedEmail,
+    SandboxCheckFailed,
     TemplateOutdated,
     TemplateSendOptions,
     TransactionalEmailError,
@@ -750,6 +751,13 @@ class PipelineHoldsAnOlderWorkflow(TransactionalEmailUnavailable):
     )
 
 
+class SandboxChecksUnavailable(TransactionalEmailUnavailable):
+    default_detail = (
+        "PostHog couldn't run the sandbox sender's checks, so nothing was sent. Retry in a few seconds, with the "
+        "same Idempotency-Key if you sent one."
+    )
+
+
 class UnconfirmedEmailSend(TransactionalEmailUnavailable):
     default_detail = (
         "The email pipeline did not confirm the send, so the email may still go out. To retry, send it again "
@@ -959,20 +967,27 @@ def _queue_send(
         try:
             queued = send(idempotency_key)
             outcome["outcome"] = "replayed" if queued.replayed else "queued"
+            if queued.delivery_mode is not None:
+                outcome["delivery_mode"] = queued.delivery_mode.value
         except TransactionalEmailError as e:
             outcome.update({"outcome": "rejected", "error_code": e.code})
             raise TransactionalEmailRejected(e)
         except WorkflowVersionBehind:
             outcome["outcome"] = "unavailable"
             raise PipelineHoldsAnOlderWorkflow() from None
+        except SandboxCheckFailed:
+            outcome["outcome"] = "unavailable"
+            raise SandboxChecksUnavailable() from None
         except DeliveryUnavailable:
             outcome["outcome"] = "unavailable"
             raise unavailable() from None
         finally:
             with transactional_email_stage("usage_event"):
                 self._report_transactional_email_send(request, {**addressed, **outcome})
         if not queued.replayed:
-            self._report_first_code_send_after_commit(request, route, addressed.get("addressed_by", "template_id"))
+            self._report_first_code_send_after_commit(
+                request, route, _first_send_properties(addressed.get("addressed_by", "template_id"), queued)
+            )
         data = SendTransactionalEmailResponseSerializer(queued).data
         return Response({**data, "warnings": list(warnings)} if warnings else data, status=status.HTTP_202_ACCEPTED)
 
@@ -1001,29 +1016,29 @@ def _require_workflow_create(self) -> None:
         if not self.user_access_control.check_access_level_for_resource("hog_flow", "editor"):
             raise exceptions.PermissionDenied("You need editor access to workflows to send under a new key.")
 
-    def _report_first_code_send_after_commit(self, request: Any, route: str, addressed_by: str) -> None:
+    def _report_first_code_send_after_commit(self, request: Any, route: str, properties: dict[str, str]) -> None:
         if route != "send_email" or not isinstance(request.user, ProjectSecretAPIKeyUser):
             return
         if not email_from_code_enabled(self.team):
             return
         key = request.user.project_secret_api_key
         if key.created_by_id is not None:
-            transaction.on_commit(lambda: self._report_first_code_send(key, addressed_by))
+            transaction.on_commit(lambda: self._report_first_code_send(key, properties))
 
-    def _report_first_code_send(self, key: ProjectSecretAPIKey, addressed_by: str) -> None:
+    def _report_first_code_send(self, key: ProjectSecretAPIKey, properties: dict[str, str]) -> None:
         try:
-            report_first_code_send(self.team_id, lambda: self._capture_first_code_send(key, addressed_by))
+            report_first_code_send(self.team_id, lambda: self._capture_first_code_send(key, properties))
         except Exception as e:
             logger.warning("Failed to capture the first code send", team_id=self.team_id, error=str(e))
 
-    def _capture_first_code_send(self, key: ProjectSecretAPIKey, addressed_by: str) -> bool:
+    def _capture_first_code_send(self, key: ProjectSecretAPIKey, properties: dict[str, str]) -> bool:
         key_creator = _current_creator(key)
         if key_creator is None:
             return False
         queued = posthoganalytics.capture(
             distinct_id=key_creator.distinct_id,
             event=FIRST_CODE_SEND_EVENT,
-            properties={"team_id": self.team_id, "addressed_by": addressed_by},
+            properties={"team_id": self.team_id, **properties},
             groups=groups(self.organization, self.team),
         )
         return queued is not None
@@ -1769,5 +1784,11 @@ def _template_send_options(data: Mapping[str, Any]) -> TemplateSendOptions:
     )
 
 
+def _first_send_properties(addressed_by: str, queued: QueuedEmail) -> dict[str, str]:
+    if queued.delivery_mode is None:
+        return {"addressed_by": addressed_by}
+    return {"addressed_by": addressed_by, "delivery_mode": queued.delivery_mode.value}
+
+
 def _distinct_id(data: Mapping[str, Any]) -> str | None:
     return data.get("distinct_id") or (data.get("identifiers") or {}).get("distinct_id") or None

products/messaging/backend/services/transactional_email.py

@@ -25,6 +25,7 @@
 
 from products.messaging.backend.facade.contracts import (
     CodeEmail,
+    DeliveryMode,
     DeliveryUnavailable,
     IdempotencyKeyInFlight,
     IdempotencyKeyReused,
@@ -36,6 +37,10 @@
     QuotaLimited,
     RecipientSuppressed,
     RecipientUnreachable,
+    SandboxCheckFailed,
+    SandboxLimitReached,
+    SandboxRecipientNotAllowed,
+    SandboxReplyToNotAllowed,
     SenderNotVerified,
     SenderUnavailable,
     SendingPaused,
@@ -57,6 +62,7 @@
 from products.workflows.backend.facade.transactional_emails import (
     TransactionalEmailContent,
     TransactionalEmailContentInvalid,
+    TransactionalEmailSender,
     TransactionalEmailSenderUnavailable,
     code_send_sender,
     create_code_email_workflow,
@@ -169,6 +175,16 @@ def _observe_cdp_call(seconds: float) -> None:
         "so the email would bounce. Check the address for a typo.",
     ),
     "not_sendable": (TransactionalEmailPaused, "The template's transactional email isn't active."),
+    "sandbox_recipient_not_allowed": (
+        SandboxRecipientNotAllowed,
+        "The sandbox sender only sends to verified members of your organization, so nothing was sent. "
+        "Send to a member, or verify your own domain to send to anyone.",
+    ),
+    "sandbox_limit_reached": (
+        SandboxLimitReached,
+        "The sandbox sender's daily limit for this project or recipient is used up, so nothing was sent. "
+        "Try again later, or verify your own domain to send without these limits.",
+    ),
 }
 
 IDEMPOTENCY_KEY_PATTERN = re.compile(r"[!-~]{1,255}")
@@ -531,11 +547,17 @@ def _send_payload(
 
 
 def _accepted_send(team_id: int, email: CodeEmail) -> dict[str, Any]:
+    sender = _code_send_sender(team_id, email.sender)
+    if sender.is_sandbox and email.reply_to:
+        raise SandboxReplyToNotAllowed(
+            "The sandbox sender can't set a reply-to address. "
+            "Remove `reply_to`, or verify your own domain to choose where replies go."
+        )
     accepted: dict[str, Any] = {
         "content": _without_empty(
             {"subject": email.subject, "html": email.html, "text": email.text or text_from_html(email.html)}
         ),
-        "from": _without_empty(_code_send_sender(team_id, email.sender)),
+        "from": _sender_fields(sender),
         "tracking": {"captureEngagement": email.capture_engagement},
     }
     if email.reply_to:
@@ -557,26 +579,31 @@ def _graded_version(team_id: int, hog_flow_id: UUID) -> int:
 def _accepted_template_send(team_id: int, options: TemplateSendOptions) -> dict[str, Any]:
     accepted: dict[str, Any] = {"tracking": {"captureEngagement": options.capture_engagement}}
     if options.sender:
-        accepted["from"] = _without_empty(_code_send_sender(team_id, options.sender))
+        accepted["from"] = _sender_fields(_code_send_sender(team_id, options.sender))
     if options.reply_to:
         accepted["replyTo"] = list(options.reply_to)
     return accepted
 
 
-def _code_send_sender(team_id: int, address: str | None) -> dict[str, Any]:
+def _code_send_sender(team_id: int, address: str | None) -> TransactionalEmailSender:
     name, email = parseaddr(address) if address else ("", "")
     try:
-        sender = code_send_sender(team_id=team_id, email=email or None, name=name or None)
+        return code_send_sender(team_id=team_id, email=email or None, name=name or None)
     except TransactionalEmailSenderUnavailable as e:
         if address:
             raise NamedSenderNotVerified(f"No verified sender in this project can send from `{email}`.") from e
         raise SenderNotVerified("The project has no verified default email sender.") from e
-    return {
-        "integrationId": sender.integration_id,
-        "email": sender.email,
-        "name": sender.name,
-        "isDefault": not address,
-    }
+
+
+def _sender_fields(sender: TransactionalEmailSender) -> dict[str, Any]:
+    return _without_empty(
+        {
+            "integrationId": sender.integration_id,
+            "email": sender.email,
+            "name": sender.name,
+            "isDefault": sender.email is None,
+        }
+    )
 
 
 def _without_empty(fields: dict[str, Any]) -> dict[str, Any]:
@@ -621,10 +648,20 @@ def _queue(
         )
         if refusal_code == "workflow_version_behind":
             raise WorkflowVersionBehind()
+        if refusal_code == "sandbox_check_failed":
+            raise SandboxCheckFailed()
         raise DeliveryUnavailable()
 
     body = response.json()
-    return QueuedEmail(message_id=str(body["invocation_id"]), replayed=bool(body.get("replayed", False)))
+    return QueuedEmail(
+        message_id=str(body["invocation_id"]),
+        replayed=bool(body.get("replayed", False)),
+        delivery_mode=_delivery_mode(body.get("delivery_mode")),
+    )
+
+
+def _delivery_mode(value: object) -> DeliveryMode | None:
+    return DeliveryMode(value) if value in DeliveryMode.values else None
 
 
 def _refusal_code(response: requests.Response) -> str | None: