diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts
index f697fad5e839..86235669747f 100644
--- a/nodejs/src/cdp/cdp-api.serial.test.ts
+++ b/nodejs/src/cdp/cdp-api.serial.test.ts
@@ -12,17 +12,24 @@ import { PosthogJwtAudience } from '~/cdp/utils/jwt-utils'
 import { ScopedServiceJwt } from '~/cdp/utils/scoped-service-jwt'
 import { setupExpressApp } from '~/common/api/router'
 import { deleteKeysWithPrefix } from '~/common/redis/_tests/redis'
-import { createRedisV2PoolFromConfig } from '~/common/redis/redis-v2'
+import { RedisClient, RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2'
 import { closeHub, createHub } from '~/common/utils/db/hub'
 import { PostgresUse } from '~/common/utils/db/postgres'
 import { parseJSON } from '~/common/utils/json-parse'
 import { logger } from '~/common/utils/logger'
+import * as posthog from '~/common/utils/posthog'
 import { UUIDT } from '~/common/utils/utils'
 import { waitForExpect } from '~/tests/helpers/expectations'
 
 import { createCdpConsumerDeps } from '../../tests/helpers/cdp'
 import { forSnapshot } from '../../tests/helpers/snapshots'
-import { createTeam, getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql'
+import {
+    createOrganizationMembership,
+    createTeam,
+    createUser,
+    getFirstTeam,
+    resetTestDatabase,
+} from '../../tests/helpers/sql'
 import { Hub, Team } from '../types'
 import { FixtureHogFlowBuilder } from './_tests/builders/hogflow.builder'
 import { HOG_EXAMPLES, HOG_FILTERS_EXAMPLES, HOG_INPUTS_EXAMPLES } from './_tests/examples'
@@ -41,7 +48,12 @@ import {
     transactionalEmailRequestsTotal,
     transactionalEmailStageDurationMs,
 } from './services/hogflows/transactional-email'
+import { OrganizationMembersService } from './services/managers/organization-members.service'
+import { SandboxSenderStateService } from './services/managers/sandbox-sender-state.service'
+import { SandboxEmailSender, SandboxEmailSenderConfig } from './services/messaging/sandbox-email-sender'
+import { SandboxSendGate } from './services/messaging/sandbox-send-gate'
 import { BASE_REDIS_KEY, HogWatcherState } from './services/monitoring/hog-watcher.service'
+import { RateLimiterService } from './services/rate-limiter/rate-limiter.service'
 import { compileHog } from './templates/compiler'
 import { HogFunctionInvocationGlobals, HogFunctionType } from './types'
 
@@ -2330,12 +2342,13 @@ describe('CDP API', () => {
                     const retry = await sendKeyedOn(route, payload)
 
                     expect(first.status).toEqual(202)
-                    expect(first.body).toMatchObject({ replayed: false })
+                    expect(first.body).toMatchObject({ replayed: false, delivery_mode: 'own' })
                     expect(retry.status).toEqual(202)
                     expect(retry.body).toEqual({
                         status: 'queued',
                         invocation_id: first.body.invocation_id,
                         replayed: true,
+                        delivery_mode: 'own',
                     })
                     expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
                     expect(emailStatusRows()).toEqual([
@@ -2364,6 +2377,7 @@ describe('CDP API', () => {
                     status: 'queued',
                     invocation_id: sent.body.invocation_id,
                     replayed: true,
+                    delivery_mode: 'own',
                 })
                 expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
             })
@@ -2534,6 +2548,7 @@ describe('CDP API', () => {
                     status: 'queued',
                     invocation_id: first.body.invocation_id,
                     replayed: true,
+                    delivery_mode: 'own',
                 })
                 expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
             })
@@ -2823,6 +2838,313 @@ describe('CDP API', () => {
             expect(res.status).toEqual(401)
             expect(mockQueueInvocations).not.toHaveBeenCalled()
         })
+
+        describe('through the sandbox sender', () => {
+            let sandboxIntegrationId: number
+            let userIds: number[]
+            let capture: jest.SpyInstance
+            const address = (name: string): string => `${name}-${team.id}@example.com`
+            const sandboxRedis: RedisV2 = {
+                useClient: async (_options, callback) => {
+                    const client = await hub.redisPool.acquire()
+                    try {
+                        return await callback(client as unknown as RedisClient)
+                    } finally {
+                        await hub.redisPool.release(client)
+                    }
+                },
+                usePipeline: () => Promise.reject(new Error('The sandbox tests do not use pipelines')),
+            }
+            const addUser = async (email: string, { member = true, verified = true } = {}): Promise<void> => {
+                const id = await createUser(hub.postgres, email)
+                userIds.push(id)
+                if (member) {
+                    await createOrganizationMembership(hub.postgres, team.organization_id, id)
+                }
+                await hub.postgres.query(
+                    PostgresUse.COMMON_WRITE,
+                    'UPDATE posthog_user SET email = $1, is_active = true, is_email_verified = $2 WHERE id = $3',
+                    [email, verified, id],
+                    'test:sandbox-api-user'
+                )
+            }
+            const unreachableRedis: RedisV2 = {
+                useClient: () => Promise.reject(new Error('Valkey is unreachable')),
+                usePipeline: () => Promise.reject(new Error('Valkey is unreachable')),
+            }
+            const sandboxSender = (
+                settings: Partial<SandboxEmailSenderConfig> = {},
+                limiterRedis: RedisV2 = sandboxRedis,
+                blockedEventRedis: RedisV2 = sandboxRedis
+            ): SandboxEmailSender =>
+                new SandboxEmailSender(
+                    {
+                        enabled: true,
+                        tenantName: 'sandbox-tenant',
+                        configurationSetName: 'sandbox-email',
+                        fromAddress: 'fixed-sandbox@example.com',
+                        dailyTeamCap: '25',
+                        dailyRecipientCap: '5',
+                        ...settings,
+                    },
+                    hub.teamManager,
+                    blockedEventRedis,
+                    new RateLimiterService(limiterRedis, { name: 'sandbox-email' }),
+                    new SandboxSenderStateService(hub.postgres, hub.pubSub)
+                )
+            const admitThrough = (
+                sender: SandboxEmailSender,
+                members = new OrganizationMembersService(hub.postgres)
+            ) => {
+                api['sandboxSendGate'] = new SandboxSendGate(sender, members, cdpDeps.integrationManager)
+            }
+            const sandboxSend = (to: string, extra: Record<string, unknown> = {}) =>
+                sendAccepted(hogFlow.id, {
+                    ...body,
+                    to,
+                    send: {
+                        content: codeSend.content,
+                        from: { integrationId: sandboxIntegrationId, isDefault: true },
+                        tracking: codeSend.tracking,
+                    },
+                    ...extra,
+                })
+
+            const pausableTenant = (): string => `sandbox-tenant-${team.id}`
+
+            beforeEach(async () => {
+                capture = jest.spyOn(posthog, 'captureTeamEvent').mockImplementation(() => {})
+                userIds = []
+                await addUser(address('member'))
+                const integration = await insertIntegration(hub.postgres, team.id, {
+                    kind: 'email',
+                    config: {
+                        provider: 'sandbox',
+                        email: 'sandbox@example.com',
+                        domain: 'example.com',
+                        name: 'Example organization via PostHog',
+                        verified: true,
+                    },
+                })
+                sandboxIntegrationId = integration.id
+                admitThrough(sandboxSender())
+            })
+
+            afterEach(async () => {
+                await hub.postgres.query(
+                    PostgresUse.COMMON_WRITE,
+                    'DELETE FROM posthog_organizationmembership WHERE user_id = ANY($1::integer[])',
+                    [userIds],
+                    'test:delete-sandbox-api-memberships'
+                )
+                await hub.postgres.query(
+                    PostgresUse.COMMON_WRITE,
+                    'DELETE FROM posthog_user WHERE id = ANY($1::integer[])',
+                    [userIds],
+                    'test:delete-sandbox-api-users'
+                )
+                await deleteKeysWithPrefix(sandboxRedis, `@posthog/workflows-sandbox-blocked/${team.id}/`)
+                await deleteKeysWithPrefix(sandboxRedis, `@posthog/workflows-sandbox-daily/{${team.organization_id}}`)
+                await hub.postgres.query(
+                    PostgresUse.COMMON_WRITE,
+                    'DELETE FROM workflows_sandboxsenderorganizationblock WHERE organization_id = $1',
+                    [team.organization_id],
+                    'test:delete-sandbox-organization-block'
+                )
+                await hub.postgres.query(
+                    PostgresUse.COMMON_WRITE,
+                    'DELETE FROM workflows_sandboxsendertenantstate WHERE tenant_name = $1',
+                    [pausableTenant()],
+                    'test:delete-sandbox-tenant-status'
+                )
+            })
+
+            it.each([
+                ['someone outside the organization', { member: false }],
+                ['a member with an unverified address', { verified: false }],
+            ])('refuses a send to %s without saying which it is', async (_desc, user) => {
+                await addUser(address('other'), user)
+
+                const res = await sandboxSend(address('other'))
+
+                expect(res.status).toEqual(400)
+                expect(res.body).toEqual({ error: 'sandbox_recipient_not_allowed' })
+                expect(mockQueueInvocations).not.toHaveBeenCalled()
+                expect(capture).toHaveBeenCalledWith(
+                    expect.objectContaining({ id: team.id }),
+                    'workflows sandbox email blocked',
+                    expect.objectContaining({ source: 'api', blocked_recipient_count: 1 })
+                )
+            })
+
+            it('answers a refusal while the blocked event store does not answer', async () => {
+                const silentRedis: RedisV2 = {
+                    useClient: () => new Promise(() => {}),
+                    usePipeline: () => new Promise(() => {}),
+                }
+                admitThrough(sandboxSender({}, sandboxRedis, silentRedis))
+                await addUser(address('other'), { member: false })
+
+                const res = await sandboxSend(address('other'))
+
+                expect(res.status).toEqual(400)
+                expect(res.body).toEqual({ error: 'sandbox_recipient_not_allowed' })
+            })
+
+            it('queues a send to a member and leaves the daily allowance for the worker to claim', async () => {
+                admitThrough(sandboxSender({ dailyTeamCap: '1', dailyRecipientCap: '1' }))
+
+                const first = await sandboxSend(address('member'))
+                const second = await sandboxSend(address('member'))
+
+                expect([first.status, second.status]).toEqual([202, 202])
+                expect([first.body.delivery_mode, second.body.delivery_mode]).toEqual(['sandbox', 'sandbox'])
+                expect(mockQueueInvocations).toHaveBeenCalledTimes(2)
+            })
+
+            it.each([
+                ['the project', { dailyTeamCap: '1', dailyRecipientCap: '5' }, 'other-member'],
+                ['the recipient', { dailyTeamCap: '25', dailyRecipientCap: '1' }, 'member'],
+            ])('refuses a send once %s used up its daily allowance', async (_desc, caps, spentOn) => {
+                await addUser(address('other-member'))
+                const sender = sandboxSender(caps)
+                admitThrough(sender)
+                expect(await sender.claimDailyCaps(team.id, [address(spentOn)])).toMatchObject({ type: 'granted' })
+
+                const res = await sandboxSend(address('member'))
+
+                expect(res.status).toEqual(400)
+                expect(res.body).toEqual({ error: 'sandbox_limit_reached' })
+                expect(mockQueueInvocations).not.toHaveBeenCalled()
+            })
+
+            it.each([
+                [
+                    'the sandbox tenant is paused',
+                    async () => {
+                        await hub.postgres.query(
+                            PostgresUse.COMMON_WRITE,
+                            `INSERT INTO workflows_sandboxsendertenantstate (tenant_name, sending_status, synced_at)
+                             VALUES ($1, 'DISABLED', now())`,
+                            [pausableTenant()],
+                            'test:pause-sandbox-tenant'
+                        )
+                        admitThrough(sandboxSender({ tenantName: pausableTenant() }))
+                    },
+                ],
+                [
+                    'the organization is blocked',
+                    async () => {
+                        await hub.postgres.query(
+                            PostgresUse.COMMON_WRITE,
+                            `INSERT INTO workflows_sandboxsenderorganizationblock (organization_id, reason, created_at)
+                             VALUES ($1, 'Abuse report', now())`,
+                            [team.organization_id],
+                            'test:block-sandbox-organization'
+                        )
+                        await Promise.resolve()
+                    },
+                ],
+                [
+                    'the sandbox sender is switched off',
+                    async () => {
+                        admitThrough(sandboxSender({ enabled: false }))
+                        await Promise.resolve()
+                    },
+                ],
+            ])('refuses a send while %s', async (_desc, suspend) => {
+                await suspend()
+
+                const res = await sandboxSend(address('member'))
+
+                expect(res.status).toEqual(422)
+                expect(res.body).toEqual({ error: 'sending_suspended' })
+                expect(mockQueueInvocations).not.toHaveBeenCalled()
+            })
+
+            it.each([
+                [
+                    'the membership lookup',
+                    () => {
+                        const members = new OrganizationMembersService(hub.postgres)
+                        jest.spyOn(members, 'checkRecipients').mockRejectedValue(new Error('Postgres is unreachable'))
+                        admitThrough(sandboxSender(), members)
+                    },
+                ],
+                ['the daily allowance', () => admitThrough(sandboxSender({}, unreachableRedis))],
+                [
+                    'the pause state',
+                    () =>
+                        jest
+                            .spyOn(SandboxSenderStateService.prototype, 'isPaused')
+                            .mockRejectedValue(new Error('Postgres is unreachable')),
+                ],
+            ])('answers 503 and queues nothing when %s cannot be checked', async (_desc, breakCheck) => {
+                breakCheck()
+
+                const res = await sandboxSend(address('member'))
+
+                expect(res.status).toEqual(503)
+                expect(res.body).toEqual({ error: 'sandbox_check_failed' })
+                expect(mockQueueInvocations).not.toHaveBeenCalled()
+            })
+
+            it.each([
+                ['a stored cc outside the organization', { cc: 'outsider@example.com' }],
+                ['a stored bcc outside the organization', { bcc: 'Outsider <outsider@example.com>' }],
+                ['a stored cc that renders per send', { cc: '{{ person.properties.manager }}' }],
+                ['a stored cc that is not text', { cc: ['outsider@example.com'] }],
+            ])('refuses a template send with %s', async (_desc, copies) => {
+                const storedFlow = transactionalFlow('api', 'active')
+                const email = storedFlow.actions.find((action) => action.type === 'function_email')!
+                email.config = {
+                    ...email.config,
+                    inputs: { email: { value: { from: { integrationId: sandboxIntegrationId }, ...copies } } },
+                } as typeof email.config
+                hogFlow = await insertHogFlow(storedFlow)
+
+                const res = await sendAccepted(hogFlow.id, {
+                    ...body,
+                    to: address('member'),
+                    send: { tracking: { captureEngagement: false } },
+                })
+
+                expect(res.status).toEqual(400)
+                expect(res.body).toEqual({ error: 'sandbox_recipient_not_allowed' })
+                expect(mockQueueInvocations).not.toHaveBeenCalled()
+            })
+
+            it('replays a keyed send as it first went out after its allowance is spent and the default sender changed', async () => {
+                const sender = sandboxSender({ dailyTeamCap: '1', dailyRecipientCap: '1' })
+                admitThrough(sender)
+                const idempotency_key = `key-${new UUIDT().toString()}`
+
+                const first = await sandboxSend(address('member'), { idempotency_key })
+                expect(await sender.claimDailyCaps(team.id, [address('member')])).toMatchObject({ type: 'granted' })
+                const retry = await sendAccepted(hogFlow.id, {
+                    ...body,
+                    to: address('member'),
+                    idempotency_key,
+                    send: {
+                        content: codeSend.content,
+                        from: { integrationId: codeSend.from.integrationId, isDefault: true },
+                        tracking: codeSend.tracking,
+                    },
+                })
+
+                expect(first.status).toEqual(202)
+                expect([retry.status, retry.body]).toEqual([
+                    202,
+                    {
+                        status: 'queued',
+                        invocation_id: first.body.invocation_id,
+                        replayed: true,
+                        delivery_mode: 'sandbox',
+                    },
+                ])
+                expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
+            })
+        })
     })
 
     describe('hogflow batch job resume', () => {
diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts
index 392c5426a519..919e9652cf42 100644
--- a/nodejs/src/cdp/cdp-api.ts
+++ b/nodejs/src/cdp/cdp-api.ts
@@ -34,7 +34,7 @@ import {
 import { HogTransformerService, createHogTransformerService } from './hog-transformations/hog-transformer.service'
 import { RerunJobManager } from './rerun/rerun-job.manager'
 import { RerunRequest } from './rerun/rerun-job.types'
-import { HogFlow, HogFlowAction } from './schema/hogflow'
+import { HogFlowAction } from './schema/hogflow'
 import { BatchExportHogFunctionService, NotFoundError, ParseError } from './services/batch-export-hog-function.service'
 import type { CyclotronV2JobProducer } from './services/cyclotron-v2'
 import { HogExecutorAsyncService, HogExecutorExecuteAsyncOptions } from './services/hog-executor-async.service'
@@ -75,6 +75,12 @@ import { EmailTrackingService } from './services/messaging/email-tracking.servic
 import { EmailValidationService } from './services/messaging/email-validation.service'
 import { EmailTrackingCodeSigner } from './services/messaging/helpers/tracking-code'
 import { RecipientTokensService } from './services/messaging/recipient-tokens.service'
+import {
+    DeliveryMode,
+    SandboxSendGate,
+    SandboxSendRefusal,
+    SandboxSendRequest,
+} from './services/messaging/sandbox-send-gate'
 import {
     HogWatcherService,
     HogWatcherState,
@@ -173,6 +179,28 @@ function parseRealtimeCohort(value: unknown): BatchResolverState['realtimeCohort
 
 const ACCEPTED_SEND_FINGERPRINT_MARKER = 'accepted_send'
 
+const SANDBOX_REFUSAL_STATUS: Record<SandboxSendRefusal, number> = {
+    sandbox_recipient_not_allowed: 400,
+    sandbox_limit_reached: 400,
+    sending_suspended: 422,
+    sandbox_check_failed: 503,
+}
+
+type TransactionalEmailAnswer = { deliveryMode: DeliveryMode }
+
+type TransactionalEmailCheck =
+    | { type: 'refused'; outcome: TransactionalEmailOutcome; status: number }
+    | { type: 'admitted'; deliveryMode: DeliveryMode }
+
+const refused = (outcome: TransactionalEmailOutcome, status: number): TransactionalEmailCheck => ({
+    type: 'refused',
+    outcome,
+    status,
+})
+
+const deliveryModeBody = (answer: TransactionalEmailAnswer | undefined): { delivery_mode?: DeliveryMode } =>
+    answer ? { delivery_mode: answer.deliveryMode } : {}
+
 const transactionalEmailFingerprint = (
     hogFlowId: string,
     { to, distinct_id, variables, send }: TransactionalEmailRequest
@@ -232,11 +260,12 @@ export class CdpApi {
     private cancelBatchJwt: ScopedServiceJwt
     private stepResumeJwt: ScopedServiceJwt
     private transactionalEmailJwt: ScopedServiceJwt
-    private transactionalEmailIdempotencyKeys: IdempotencyKeys
+    private transactionalEmailIdempotencyKeys: IdempotencyKeys<TransactionalEmailAnswer>
     private emailSuppressionService: EmailSuppressionService
     private emailValidationService: EmailValidationService
     private teamWorkflowsConfigService: TeamWorkflowsConfigService
     private emailStatusService: EmailStatusService
+    private sandboxSendGate: SandboxSendGate
     private resumeBatchJwt: ScopedServiceJwt
 
     constructor(
@@ -262,6 +291,11 @@ export class CdpApi {
         this.hogWatcherMirror = services.hogWatcherMirror
         this.invocationResultsService = services.invocationResultsService
         this.emailStatusService = services.emailStatusService
+        this.sandboxSendGate = new SandboxSendGate(
+            services.sandboxEmailSender,
+            services.organizationMembers,
+            deps.integrationManager
+        )
 
         // API-only services. The hog-transformer's monitoring service reuses the same
         // resolved outputs registry as the core CDP services — no separate construction.
@@ -320,7 +354,7 @@ export class CdpApi {
             PosthogJwtAudience.WORKFLOWS_SEND_TRANSACTIONAL_EMAIL,
             config.WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET || ''
         )
-        this.transactionalEmailIdempotencyKeys = new IdempotencyKeys(
+        this.transactionalEmailIdempotencyKeys = new IdempotencyKeys<TransactionalEmailAnswer>(
             deps.idempotencyValkey ?? null,
             'transactional-email'
         )
@@ -1311,6 +1345,7 @@ export class CdpApi {
                                 status: 'queued',
                                 invocation_id: reservation.invocationId,
                                 replayed: true,
+                                ...deliveryModeBody(reservation.answer),
                             })
                         case 'in_flight':
                             return reply('idempotency_key_in_flight', 409, { error: 'idempotency_key_in_flight' })
@@ -1320,16 +1355,24 @@ export class CdpApi {
                             return reply('idempotency_unavailable', 503, { error: 'idempotency_unavailable' })
                     }
                 }
+                let deliveryMode: DeliveryMode
                 try {
                     stages.start('checks')
-                    const refusal = await this.refuseTransactionalEmail(team.id, hogFlow, to)
-                    if (refusal) {
+                    const check = await this.checkTransactionalEmail({
+                        teamId: team.id,
+                        invocationId: invocation.id,
+                        hogFlow,
+                        send,
+                        to,
+                    })
+                    if (check.type === 'refused') {
                         if (idempotencyKey) {
                             stages.start('idempotency')
                             await idempotencyKeys.release(team.id, idempotencyKey, invocation.id)
                         }
-                        return reply(refusal.outcome, refusal.status, { error: refusal.outcome })
+                        return reply(check.outcome, check.status, { error: check.outcome })
                     }
+                    deliveryMode = check.deliveryMode
                     stages.start('queue_insert')
                     this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running')
                     try {
@@ -1349,7 +1392,9 @@ export class CdpApi {
                 }
                 if (idempotencyKey) {
                     stages.start('idempotency')
-                    await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id)
+                    await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id, {
+                        deliveryMode,
+                    })
                 }
                 stages.start('metrics_flush')
                 this.emailStatusService.queue({ teamId: team.id, messageId: invocation.id, status: 'queued' })
@@ -1364,7 +1409,12 @@ export class CdpApi {
                     await this.invocationResultsService.invocationResultsRowsService.handOffQueuedRows()
                 void produced.catch(warnTelemetryFlushFailed)
 
-                return reply('queued', 202, { status: 'queued', invocation_id: invocation.id, replayed: false })
+                return reply('queued', 202, {
+                    status: 'queued',
+                    invocation_id: invocation.id,
+                    replayed: false,
+                    delivery_mode: deliveryMode,
+                })
             } catch (e) {
                 logger.error('Error queueing transactional email', {
                     error: e instanceof Error ? e.message : String(e),
@@ -1378,32 +1428,33 @@ export class CdpApi {
     // The email worker skips a send for these reasons too, but only after the caller has its 202. Checking them
     // here tells the caller that the email will not go out. The worker keeps its checks, because the state can
     // change while the job waits on the queue.
-    private async refuseTransactionalEmail(
-        teamId: number,
-        hogFlow: HogFlow,
-        to: string
-    ): Promise<{ outcome: TransactionalEmailOutcome; status: number } | null> {
+    private async checkTransactionalEmail(request: SandboxSendRequest): Promise<TransactionalEmailCheck> {
+        const { teamId, hogFlow, to } = request
         if (hogFlow.status !== 'active') {
-            return { outcome: 'not_sendable', status: 400 }
+            return refused('not_sendable', 400)
         }
         if (await this.teamWorkflowsConfigService.getEmailSendingSuspension(teamId)) {
-            return { outcome: 'sending_suspended', status: 422 }
+            return refused('sending_suspended', 422)
         }
         if (hogFlow.email_sending_paused_at) {
-            return { outcome: 'sending_paused', status: 422 }
+            return refused('sending_paused', 422)
         }
         if (await this.emailSuppressionService.isSuppressed(teamId, to, { transactional: true })) {
-            return { outcome: 'recipient_suppressed', status: 422 }
+            return refused('recipient_suppressed', 422)
         }
         const { isLimited } = await checkHogFlowQuotaLimits(hogFlow, teamId, this.deps.quotaLimiting)
         if (isLimited) {
-            return { outcome: 'quota_limited', status: 429 }
+            return refused('quota_limited', 429)
+        }
+        const admission = await this.sandboxSendGate.admit(request)
+        if (admission.deliveryMode === 'sandbox' && admission.refusal) {
+            return refused(admission.refusal, SANDBOX_REFUSAL_STATUS[admission.refusal])
         }
         // Last, because a cache miss costs a DNS lookup. A DNS failure lets the send through.
         if (await this.emailValidationService.getRecipientSkipReason(teamId, to)) {
-            return { outcome: 'recipient_unreachable', status: 422 }
+            return refused('recipient_unreachable', 422)
         }
-        return null
+        return { type: 'admitted', deliveryMode: admission.deliveryMode }
     }
 
     private verifyScopedWorkflowJwt(
diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts
index 30946d936cfa..1134310144e0 100644
--- a/nodejs/src/cdp/cdp-services.ts
+++ b/nodejs/src/cdp/cdp-services.ts
@@ -121,6 +121,7 @@ export interface CdpCoreServices {
     outputs: CdpOutputs
     emailService: EmailService
     sandboxEmailSender: SandboxEmailSender
+    organizationMembers: OrganizationMembersService
     /** Buffers rendered-email asset rows and bulk-flushes them at the batch boundary. */
     messageAssetsService: MessageAssetsService
     /** Buffers transactional email status rows. `invocationResultsService.flush()` produces them. */
@@ -489,6 +490,7 @@ export function createCdpCoreServices(
         sandboxEmailRateLimiter,
         new SandboxSenderStateService(deps.postgres, deps.pubSub)
     )
+    const organizationMembers = new OrganizationMembersService(deps.postgres)
     const emailService = new EmailService(
         {
             sesAccessKeyId: config.SES_ACCESS_KEY_ID,
@@ -513,7 +515,7 @@ export function createCdpCoreServices(
         workflowEmailRateLimiter,
         teamEmailRateLimiter,
         sandboxEmailSender,
-        new OrganizationMembersService(deps.postgres)
+        organizationMembers
     )
     const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL)
     const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields)
@@ -653,6 +655,7 @@ export function createCdpCoreServices(
         outputs,
         emailService,
         sandboxEmailSender,
+        organizationMembers,
         messageAssetsService,
         emailStatusService,
     }
diff --git a/nodejs/src/cdp/services/hogflows/idempotency-keys.ts b/nodejs/src/cdp/services/hogflows/idempotency-keys.ts
index cd4a0e78480a..841a2cc405fe 100644
--- a/nodejs/src/cdp/services/hogflows/idempotency-keys.ts
+++ b/nodejs/src/cdp/services/hogflows/idempotency-keys.ts
@@ -111,32 +111,33 @@ const sortKeys = (value: unknown): unknown => {
     return value
 }
 
-export type IdempotencyReservation =
+export type IdempotencyReservation<Answer extends object = object> =
     | { status: 'reserved' }
     // The key already queued a run for the same request.
-    | { status: 'replayed'; invocationId: string }
+    | { status: 'replayed'; invocationId: string; answer?: Answer }
     // The first request with the key has not finished queueing yet.
     | { status: 'in_flight' }
     // The key was used for a different request.
     | { status: 'conflict' }
     | { status: 'unavailable' }
 
-type StoredKey = { fingerprint: string; invocationId: string; queued: boolean }
+type StoredKey<Answer extends object> = { fingerprint: string; invocationId: string; queued: boolean; answer?: Answer }
 
-type PendingCompletion = { redisKey: string; value: StoredKey; expiresAt: number }
+type PendingCompletion<Answer extends object> = { redisKey: string; value: StoredKey<Answer>; expiresAt: number }
 
 // Keyed by run, not by Redis key. Several runs of one key can wait at once, for example a webhook
 // retry that queued without a reservation during the outage. Only the store knows which run owns
 // the key, so every run keeps its entry until its own write succeeds, loses ownership, or expires.
-const pendingCompletionKey = ({ redisKey, value }: PendingCompletion): string => `${redisKey}/${value.invocationId}`
+const pendingCompletionKey = <Answer extends object>({ redisKey, value }: PendingCompletion<Answer>): string =>
+    `${redisKey}/${value.invocationId}`
 
 /**
  * Caller-supplied idempotency keys, scoped to a team. A key reserves a run before it is queued, and
  * holds the run's id for 24 hours once it is. A queue failure releases the key, so the caller's
  * retry can queue the run.
  */
-export class IdempotencyKeys {
-    private pendingCompletions = new Map<string, PendingCompletion>()
+export class IdempotencyKeys<Answer extends object = object> {
+    private pendingCompletions = new Map<string, PendingCompletion<Answer>>()
     private draining = false
 
     constructor(
@@ -156,7 +157,7 @@ export class IdempotencyKeys {
         key: string,
         fingerprint: string,
         invocationId: string
-    ): Promise<IdempotencyReservation> {
+    ): Promise<IdempotencyReservation<Answer>> {
         const reservation = await this.reserveInStore(teamId, key, fingerprint, invocationId)
         this.count('reserve', reservation.status)
         return reservation
@@ -167,12 +168,12 @@ export class IdempotencyKeys {
         key: string,
         fingerprint: string,
         invocationId: string
-    ): Promise<IdempotencyReservation> {
+    ): Promise<IdempotencyReservation<Answer>> {
         const valkey = this.valkey
         if (!valkey) {
             return { status: 'unavailable' }
         }
-        const value: StoredKey = { fingerprint, invocationId, queued: false }
+        const value: StoredKey<Answer> = { fingerprint, invocationId, queued: false }
         let existing: string | null
         try {
             // SET ... NX GET sets the key when absent and returns the existing value when present.
@@ -196,17 +197,25 @@ export class IdempotencyKeys {
         if (existing === null) {
             return { status: 'reserved' }
         }
-        const stored = parseJSON(existing) as StoredKey
+        const stored = parseJSON(existing) as StoredKey<Answer>
         if (stored.fingerprint !== fingerprint) {
             return { status: 'conflict' }
         }
-        return stored.queued ? { status: 'replayed', invocationId: stored.invocationId } : { status: 'in_flight' }
+        return stored.queued
+            ? { status: 'replayed', invocationId: stored.invocationId, answer: stored.answer }
+            : { status: 'in_flight' }
     }
 
-    async markQueued(teamId: number, key: string, fingerprint: string, invocationId: string): Promise<void> {
-        const completion: PendingCompletion = {
+    async markQueued(
+        teamId: number,
+        key: string,
+        fingerprint: string,
+        invocationId: string,
+        answer?: Answer
+    ): Promise<void> {
+        const completion: PendingCompletion<Answer> = {
             redisKey: this.redisKey(teamId, key),
-            value: { fingerprint, invocationId, queued: true },
+            value: { fingerprint, invocationId, queued: true, answer },
             expiresAt: Date.now() + QUEUED_TTL_SECONDS * 1000,
         }
         for (let attempt = 1; attempt <= MARK_QUEUED_INLINE_ATTEMPTS; attempt++) {
@@ -225,7 +234,7 @@ export class IdempotencyKeys {
         this.retryInBackground(completion)
     }
 
-    private retryInBackground(completion: PendingCompletion): void {
+    private retryInBackground(completion: PendingCompletion<Answer>): void {
         if (this.pendingCompletions.size >= MAX_PENDING_COMPLETIONS) {
             logger.error('🔑', 'idempotency key completion dropped, too many waiting for the store', {
                 scope: this.scope,
@@ -270,7 +279,7 @@ export class IdempotencyKeys {
         return true
     }
 
-    private async writeQueued({ redisKey, value, expiresAt }: PendingCompletion): Promise<void> {
+    private async writeQueued({ redisKey, value, expiresAt }: PendingCompletion<Answer>): Promise<void> {
         const ttlSeconds = Math.ceil((expiresAt - Date.now()) / 1000)
         const written = await this.timed(
             'mark_queued',
diff --git a/nodejs/src/cdp/services/hogflows/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts
index f7d3fdca0612..f49ae3561114 100644
--- a/nodejs/src/cdp/services/hogflows/transactional-email.ts
+++ b/nodejs/src/cdp/services/hogflows/transactional-email.ts
@@ -66,6 +66,9 @@ export type TransactionalEmailOutcome =
     | 'sending_paused'
     | 'recipient_suppressed'
     | 'recipient_unreachable'
+    | 'sandbox_recipient_not_allowed'
+    | 'sandbox_limit_reached'
+    | 'sandbox_check_failed'
     | 'error'
 
 export const transactionalEmailRequestsTotal = new Counter({
@@ -88,7 +91,7 @@ export type TransactionalEmailStage = (typeof TRANSACTIONAL_EMAIL_STAGES)[number
 
 export const transactionalEmailStageDurationMs = new Histogram({
     name: 'cdp_transactional_email_stage_duration_ms',
-    help: 'Milliseconds one transactional email request spends in each stage: auth (the JWT), validate (the body), load (the team and the workflow), build (the run and its idempotency fingerprint), idempotency (every Idempotency-Key call to Valkey), checks (suspension, pause, suppression, quota and MX), queue_insert (the email queue insert), metrics_flush (the hand-off of the result and status rows to the Kafka producer, without the wait for the broker), and total. A request records only the stages it reached, so a refusal shows in the stage that refused it.',
+    help: 'Milliseconds one transactional email request spends in each stage: auth (the JWT), validate (the body), load (the team and the workflow), build (the run and its idempotency fingerprint), idempotency (every Idempotency-Key call to Valkey), checks (suspension, pause, suppression, quota, the sandbox member, pause and allowance checks, and MX), queue_insert (the email queue insert), metrics_flush (the hand-off of the result and status rows to the Kafka producer, without the wait for the broker), and total. A request records only the stages it reached, so a refusal shows in the stage that refused it.',
     labelNames: ['stage'],
     buckets: [1, 2.5, 5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000],
 })
diff --git a/nodejs/src/cdp/services/managers/integration-manager.service.test.ts b/nodejs/src/cdp/services/managers/integration-manager.service.test.ts
index 45f2dd712ecb..ebbba431dfb2 100644
--- a/nodejs/src/cdp/services/managers/integration-manager.service.test.ts
+++ b/nodejs/src/cdp/services/managers/integration-manager.service.test.ts
@@ -70,6 +70,15 @@ describe('IntegrationManager', () => {
         expect(item?.config).toEqual({ team: 'foobar' })
     })
 
+    it('finds an integration the read replica does not have yet on the primary', async () => {
+        const query = jest.spyOn(hub.postgres, 'query').mockResolvedValueOnce({ rows: [] } as any)
+
+        const item = await manager.get(integrations[0].id)
+
+        expect(item?.config).toEqual({ team: 'foobar' })
+        expect(query.mock.calls.map(([use]) => use)).toEqual([PostgresUse.COMMON_READ, PostgresUse.COMMON_WRITE])
+    })
+
     it('updates cached integration data when integration changes', async () => {
         // First check - initial state
         const item = await manager.get(integrations[0].id)
diff --git a/nodejs/src/cdp/services/managers/integration-manager.service.ts b/nodejs/src/cdp/services/managers/integration-manager.service.ts
index 83f3cd76267d..6c4d5479e48a 100644
--- a/nodejs/src/cdp/services/managers/integration-manager.service.ts
+++ b/nodejs/src/cdp/services/managers/integration-manager.service.ts
@@ -39,16 +39,26 @@ export class IntegrationManagerService {
     private async fetchIntegrations(ids: string[]): Promise<Record<string, IntegrationType | undefined>> {
         logger.info('[IntegrationManager]', 'Fetching integrations', { ids })
 
+        const fromReplica = await this.queryIntegrations(PostgresUse.COMMON_READ, ids)
+        const missingFromReplica = ids.filter((id) => !fromReplica[id])
+        if (!missingFromReplica.length) {
+            return fromReplica
+        }
+        return { ...fromReplica, ...(await this.queryIntegrations(PostgresUse.COMMON_WRITE, missingFromReplica)) }
+    }
+
+    private async queryIntegrations(
+        use: PostgresUse,
+        ids: string[]
+    ): Promise<Record<string, IntegrationType | undefined>> {
         const response = await this.postgres.query<IntegrationType>(
-            PostgresUse.COMMON_READ,
+            use,
             `SELECT id, team_id, kind, config, sensitive_config FROM posthog_integration WHERE id = ANY($1)`,
             [ids],
             'fetchIntegrations'
         )
 
-        const items = response.rows
-
-        return items.reduce<Record<string, IntegrationType | undefined>>((acc, item) => {
+        return response.rows.reduce<Record<string, IntegrationType | undefined>>((acc, item) => {
             const sensitiveConfig = this.encryptedFields.decryptObject(item.sensitive_config || {}, {
                 ignoreDecryptionErrors: true,
             })
diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts
index f5aa7c8b9986..ba1254f4396f 100644
--- a/nodejs/src/cdp/services/messaging/email.service.test.ts
+++ b/nodejs/src/cdp/services/messaging/email.service.test.ts
@@ -369,7 +369,12 @@ describe('EmailService', () => {
                 expect(capture).toHaveBeenLastCalledWith(
                     expect.objectContaining({ id: team.id }),
                     'workflows sandbox email blocked',
-                    { reason, is_test: isTest, blocked_recipient_count: blockedRecipientCount }
+                    {
+                        reason,
+                        is_test: isTest,
+                        blocked_recipient_count: blockedRecipientCount,
+                        source: isTest ? 'test' : 'workflow',
+                    }
                 )
             }
 
@@ -610,6 +615,39 @@ describe('EmailService', () => {
                 )
             })
 
+            it('delivers to each member once when an address repeats across to, cc and bcc', async () => {
+                service = createSandboxService(true)
+                invocation.queueParameters = createSandboxParams({
+                    to: { email: memberEmail() },
+                    cc: `${memberEmail().toUpperCase()}, ${memberEmail('cc')}, ${memberEmail('cc')}`,
+                    bcc: `Example colleague <${memberEmail('cc')}>, ${memberEmail('bcc')}, ${memberEmail()}`,
+                })
+
+                const result = await service.executeSendEmail(invocation)
+
+                expect(result.error).toBeUndefined()
+                expect((sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input.Destination).toEqual({
+                    ToAddresses: [memberEmail()],
+                    CcAddresses: [memberEmail('cc')],
+                    BccAddresses: [memberEmail('bcc')],
+                })
+            })
+
+            it('records a send from the transactional email API as an api send', async () => {
+                service = createSandboxService(true)
+                invocation.queueParameters = createSandboxParams({ to: { email: memberEmail() } })
+                invocation.state.globals.event.event = TRANSACTIONAL_EMAIL_EVENT
+
+                const result = await service.executeSendEmail(invocation)
+
+                expect(result.error).toBeUndefined()
+                expect(capture).toHaveBeenCalledWith(
+                    expect.objectContaining({ id: team.id }),
+                    'workflows sandbox email sent',
+                    { recipient_count: 1, source: 'api', is_test: false }
+                )
+            })
+
             it.each(['cc', 'bcc'] as const)('delivers a quoted %s mailbox whole', async (field) => {
                 const email = `"example,colleague-${team.id}"@example.com`
                 await createMember(email)
@@ -876,7 +914,12 @@ describe('EmailService', () => {
                     expect(capture).toHaveBeenCalledWith(
                         expect.objectContaining({ id: team.id }),
                         'workflows sandbox email blocked',
-                        { reason: 'paused', is_test: isTest, blocked_recipient_count: 0 }
+                        {
+                            reason: 'paused',
+                            is_test: isTest,
+                            blocked_recipient_count: 0,
+                            source: isTest ? 'test' : 'workflow',
+                        }
                     )
                 })
 
@@ -916,7 +959,7 @@ describe('EmailService', () => {
                     expect(capture).toHaveBeenCalledWith(
                         expect.objectContaining({ id: team.id }),
                         'workflows sandbox email blocked',
-                        { reason: 'check_failed', is_test: false, blocked_recipient_count: 0 }
+                        { reason: 'check_failed', is_test: false, blocked_recipient_count: 0, source: 'workflow' }
                     )
                 })
 
@@ -1164,19 +1207,20 @@ describe('EmailService', () => {
                     }
                 )
 
-                it('charges a repeated address once per address and once per copy for the project', async () => {
+                it('charges a repeated address once, as it is delivered once', async () => {
                     service = createSandboxService(true, {
-                        senderSettings: { dailyTeamCap: '3', dailyRecipientCap: '1' },
+                        senderSettings: { dailyTeamCap: '2', dailyRecipientCap: '1' },
                     })
 
                     expectSent(await send(false, { cc: memberEmail().toUpperCase(), bcc: ` ${memberEmail()} ` }))
+                    expectSent(await send(false, { to: { email: memberEmail('cc') } }))
                     expectCapSkip(
-                        await send(false, { to: { email: memberEmail('cc') } }),
+                        await send(false, { to: { email: memberEmail('bcc') } }),
                         false,
                         TEAM_CAP_REACHED,
                         'cap_reached'
                     )
-                    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+                    expect(sendEmailSpy).toHaveBeenCalledTimes(2)
                 })
 
                 it('refills a spent allowance over a day rather than seconds', async () => {
diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts
index be8b9aecfdc3..225373e894d3 100644
--- a/nodejs/src/cdp/services/messaging/email.service.ts
+++ b/nodejs/src/cdp/services/messaging/email.service.ts
@@ -44,7 +44,13 @@ import {
     RecipientTokensService,
     preferencesTokenSourceForInvocation,
 } from './recipient-tokens.service'
-import { SandboxDailyCapGrant, SandboxEmailBlockReason, SandboxEmailSender } from './sandbox-email-sender'
+import {
+    SandboxDailyCapGrant,
+    SandboxEmailBlockReason,
+    SandboxEmailSender,
+    sandboxSendSource,
+} from './sandbox-email-sender'
+import { sandboxEnvelope, sandboxEnvelopeRecipients } from './sandbox-envelope'
 
 const sesThrottleResponsesTotal = new Counter({
     name: 'cdp_ses_throttle_responses_total',
@@ -408,60 +414,8 @@ export function parseAddressList(value?: string): string[] | undefined {
     return result.length > 0 ? result : undefined
 }
 
-type QuoteScan = { unquotedIndexes: number[]; balanced: boolean }
-
-function scanQuotes(text: string): QuoteScan {
-    const unquotedIndexes: number[] = []
-    let quoted = false
-    let escaped = false
-    for (let index = 0; index < text.length; index++) {
-        const character = text[index]
-        if (escaped) {
-            escaped = false
-        } else if (quoted && character === '\\') {
-            escaped = true
-        } else if (character === '"') {
-            quoted = !quoted
-        } else if (!quoted) {
-            unquotedIndexes.push(index)
-        }
-    }
-    return { unquotedIndexes, balanced: !quoted }
-}
-
-function splitOutsideQuotes(text: string): string[] {
-    const commas = scanQuotes(text).unquotedIndexes.filter((index) => text[index] === ',')
-    return [-1, ...commas].map((comma, position) => text.slice(comma + 1, commas[position] ?? text.length))
-}
-
-function angleAddress(entry: string): string | undefined {
-    const { unquotedIndexes, balanced } = scanQuotes(entry)
-    const opening = unquotedIndexes.find((index) => entry[index] === '<')
-    const closing = entry.length - 1
-    if (!balanced || opening === undefined || entry[closing] !== '>' || /[\r\n]/.test(entry)) {
-        return undefined
-    }
-    const unquotedBetween = (start: number, end: number): string =>
-        unquotedIndexes
-            .filter((index) => index >= start && index < end)
-            .map((index) => entry[index])
-            .join('')
-    const mailbox = entry.slice(opening + 1, closing).trim()
-    const isFramed = !/[>@;]/.test(unquotedBetween(0, opening)) && !/[<>]/.test(unquotedBetween(opening + 1, closing))
-    return isFramed && mailbox ? mailbox : undefined
-}
-
-// Unlike `extractEmailsFromAddressList`, this keeps quoted commas inside one entry and leaves an
-// ambiguous mailbox whole, so the member check sees exactly the addresses SES would deliver to.
-function sandboxAddressList(value?: string): string[] {
-    return splitOutsideQuotes(value ?? '')
-        .map((entry) => entry.trim())
-        .filter(Boolean)
-        .map((entry) => angleAddress(entry) ?? entry)
-}
-
 function sandboxRecipientsOf(params: CyclotronInvocationQueueParametersEmailType): string[] {
-    return [params.to.email.trim(), ...sandboxAddressList(params.cc), ...sandboxAddressList(params.bcc)]
+    return sandboxEnvelopeRecipients(sandboxEnvelope(params.to.email, params.cc, params.bcc))
 }
 
 type SandboxRecipientBlockReason = Exclude<SandboxEmailBlockReason, 'switch_off' | 'paused' | 'organization_blocked'>
@@ -687,7 +641,12 @@ export class EmailService {
                     'Skipping send: the sandbox sender is unavailable right now. Verify your own domain to keep sending.',
                     'sending_suspended'
                 )
-                await this.sandboxSender?.captureBlocked(invocation.teamId, isTest, 'switch_off', 0)
+                await this.sandboxSender?.captureBlocked(
+                    invocation.teamId,
+                    sandboxSendSource(invocation, isTest),
+                    'switch_off',
+                    0
+                )
                 return result
             }
             if (isSandbox && !(await this.passesSandboxPauseGate(result, isTest))) {
@@ -920,7 +879,11 @@ export class EmailService {
                         return result
                     }
                     await this.sendSandboxEmail(result, deliveryParams, from, isTest, queueWait, route, capGrant)
-                    await this.sandboxSender!.captureSent(invocation.teamId, isTest, deliveryRecipients.length)
+                    await this.sandboxSender!.captureSent(
+                        invocation.teamId,
+                        sandboxSendSource(invocation, isTest),
+                        deliveryRecipients.length
+                    )
                     break
                 }
 
@@ -1206,7 +1169,12 @@ export class EmailService {
         }
         if (organizationBlocked) {
             this.skipSandboxSend(result, isTest, SANDBOX_ORGANIZATION_BLOCKED_MESSAGE, 'sending_suspended')
-            await this.sandboxSender?.captureBlocked(teamId, isTest, 'organization_blocked', 0)
+            await this.sandboxSender?.captureBlocked(
+                teamId,
+                sandboxSendSource(result.invocation, isTest),
+                'organization_blocked',
+                0
+            )
             return false
         }
         if (!blockedRecipients.length) {
@@ -1225,7 +1193,12 @@ export class EmailService {
             return true
         }
         this.skipSandboxSend(result, isTest, SANDBOX_PAUSE_SKIP_MESSAGES[gate], 'sending_suspended')
-        await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, gate, 0)
+        await this.sandboxSender!.captureBlocked(
+            result.invocation.teamId,
+            sandboxSendSource(result.invocation, isTest),
+            gate,
+            0
+        )
         return false
     }
 
@@ -1242,7 +1215,12 @@ export class EmailService {
             `${explanation}${label}${blockedRecipients.join(', ')}. ${guidance}`,
             reason === 'cap_reached' ? 'frequency_capped' : 'recipient_unreachable'
         )
-        await this.sandboxSender?.captureBlocked(result.invocation.teamId, isTest, reason, blockedRecipients.length)
+        await this.sandboxSender?.captureBlocked(
+            result.invocation.teamId,
+            sandboxSendSource(result.invocation, isTest),
+            reason,
+            blockedRecipients.length
+        )
     }
 
     private async claimSandboxDailyCaps(
@@ -1265,7 +1243,12 @@ export class EmailService {
             claim.type === 'check_failed' ? 'sending_suspended' : 'frequency_capped'
         )
         const reason = claim.type === 'check_failed' ? 'check_failed' : 'cap_reached'
-        await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, reason, recipients.length)
+        await this.sandboxSender!.captureBlocked(
+            result.invocation.teamId,
+            sandboxSendSource(result.invocation, isTest),
+            reason,
+            recipients.length
+        )
         return null
     }
 
@@ -1366,12 +1349,13 @@ export class EmailService {
     // set, and no recipient gets an unsubscribe link: only members of the sending organization
     // receive it. CC and BCC are parsed once so the final member check covers exactly this list.
     private sandboxSesRoute(params: CyclotronInvocationQueueParametersEmailType): SesRoute {
+        const envelope = sandboxEnvelope(params.to.email, params.cc, params.bcc)
         return {
             tenantName: this.sandboxSender!.config.tenantName,
             configurationSetName: this.sandboxSender!.config.configurationSetName,
             unsubscribeHeaders: false,
-            ccAddresses: sandboxAddressList(params.cc),
-            bccAddresses: sandboxAddressList(params.bcc),
+            ccAddresses: envelope.cc,
+            bccAddresses: envelope.bcc,
         }
     }
 
diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts
index e9dd35401e8f..c43d3f16a104 100644
--- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts
+++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts
@@ -2,6 +2,7 @@ import { createHash } from 'crypto'
 import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5'
 
 import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron'
+import { CyclotronJobInvocationHogFunction } from '~/cdp/types'
 import { RedisV2 } from '~/common/redis/redis-v2'
 import { logger } from '~/common/utils/logger'
 import { captureTeamEvent } from '~/common/utils/posthog'
@@ -9,6 +10,7 @@ import { TeamManager } from '~/common/utils/team-manager'
 
 import { SandboxSenderStateService } from '../managers/sandbox-sender-state.service'
 import { ClaimRequest, IssuedClaim, RateLimiterService } from '../rate-limiter/rate-limiter.service'
+import { isTransactionalEmailSend } from './email-status.service'
 import { maybeAddPreheaderToEmail } from './helpers/preheader'
 
 const SECONDS_PER_DAY = 86400
@@ -52,6 +54,19 @@ export type SandboxDailyCapClaim =
     | { type: 'recipient_cap_reached'; addresses: string[] }
     | { type: 'check_failed' }
 
+type DailyCapBuckets = { limiter: RateLimiterService; buckets: ClaimRequest[]; addresses: string[] }
+
+type SandboxDailyCapDenial = { type: 'project_cap_reached' } | { type: 'recipient_cap_reached'; addresses: string[] }
+
+export type SandboxDailyCapPeek = { type: 'available' } | SandboxDailyCapDenial | { type: 'check_failed' }
+
+function dailyCapDenial({ addresses }: DailyCapBuckets, deniedIndexes: number[]): SandboxDailyCapDenial {
+    if (deniedIndexes.includes(0)) {
+        return { type: 'project_cap_reached' }
+    }
+    return { type: 'recipient_cap_reached', addresses: deniedIndexes.map((index) => addresses[index - 1]) }
+}
+
 function dailyBucket(key: string, requested: number, capacity: number): ClaimRequest {
     return { key, requested, capacity, refillPerSecond: capacity / SECONDS_PER_DAY, ttlSeconds: DAILY_CAP_TTL_SECONDS }
 }
@@ -83,6 +98,17 @@ function distinctAddresses(recipients: string[]): string[] {
 
 const BLOCKED_EVENT_WINDOW_SECONDS = 3600
 
+export type SandboxSendSource = 'test' | 'workflow' | 'api'
+
+const BLOCKED_EVENT_WINDOW_NAMES: Record<SandboxSendSource, string> = { test: 'test', workflow: 'live', api: 'api' }
+
+export function sandboxSendSource(invocation: CyclotronJobInvocationHogFunction, isTest: boolean): SandboxSendSource {
+    if (isTest) {
+        return 'test'
+    }
+    return isTransactionalEmailSend(invocation) ? 'api' : 'workflow'
+}
+
 function htmlBody(document: DefaultTreeAdapterMap['document']): DefaultTreeAdapterMap['element'] | undefined {
     const root = document.childNodes.find(defaultTreeAdapter.isElementNode)
     return root?.childNodes.find(
@@ -199,15 +225,39 @@ export class SandboxEmailSender {
     }
 
     public async claimDailyCaps(teamId: number, recipients: string[]): Promise<SandboxDailyCapClaim> {
+        const caps = await this.dailyCapBuckets(teamId, recipients)
+        if (!caps) {
+            return { type: 'check_failed' }
+        }
+        const claim = await caps.limiter.claimAllOrNothing(caps.buckets)
+        if (claim.granted) {
+            return { type: 'granted', claim, buckets: caps.buckets }
+        }
+        return claim.deniedIndexes === null ? { type: 'check_failed' } : dailyCapDenial(caps, claim.deniedIndexes)
+    }
+
+    public async peekDailyCaps(teamId: number, recipients: string[]): Promise<SandboxDailyCapPeek> {
+        const caps = await this.dailyCapBuckets(teamId, recipients)
+        if (!caps) {
+            return { type: 'check_failed' }
+        }
+        const peek = await caps.limiter.peekAllOrNothing(caps.buckets)
+        if (!peek) {
+            return { type: 'check_failed' }
+        }
+        return peek.deniedIndexes.length ? dailyCapDenial(caps, peek.deniedIndexes) : { type: 'available' }
+    }
+
+    private async dailyCapBuckets(teamId: number, recipients: string[]): Promise<DailyCapBuckets | null> {
         const dailyTeamCap = parseDailyCap(this.config.dailyTeamCap)
         const dailyRecipientCap = parseDailyCap(this.config.dailyRecipientCap)
         if (!this.dailyCapLimiter || dailyTeamCap === null || dailyRecipientCap === null) {
             logger.error('Sandbox email daily caps are not configured correctly', { teamId })
-            return { type: 'check_failed' }
+            return null
         }
         const team = await this.teamManager.getTeam(teamId)
         if (!team) {
-            return { type: 'check_failed' }
+            return null
         }
         const prefix = dailyCapKeyPrefix(team.organization_id)
         const addresses = distinctAddresses(recipients)
@@ -217,17 +267,7 @@ export class SandboxEmailSender {
                 dailyBucket(`${prefix}/recipient/${addressDigest(address)}`, 1, dailyRecipientCap)
             ),
         ]
-        const claim = await this.dailyCapLimiter.claimAllOrNothing(buckets)
-        if (claim.granted) {
-            return { type: 'granted', claim, buckets }
-        }
-        if (claim.deniedIndexes === null) {
-            return { type: 'check_failed' }
-        }
-        if (claim.deniedIndexes.includes(0)) {
-            return { type: 'project_cap_reached' }
-        }
-        return { type: 'recipient_cap_reached', addresses: claim.deniedIndexes.map((index) => addresses[index - 1]) }
+        return { limiter: this.dailyCapLimiter, buckets, addresses }
     }
 
     public async returnDailyCaps(grant: SandboxDailyCapGrant): Promise<void> {
@@ -259,11 +299,11 @@ export class SandboxEmailSender {
         }
     }
 
-    public async captureSent(teamId: number, isTest: boolean, recipientCount: number): Promise<void> {
+    public async captureSent(teamId: number, source: SandboxSendSource, recipientCount: number): Promise<void> {
         await this.captureEvent(teamId, 'workflows sandbox email sent', {
             recipient_count: recipientCount,
-            source: isTest ? 'test' : 'workflow',
-            is_test: isTest,
+            source,
+            is_test: source === 'test',
         })
     }
 
@@ -276,15 +316,15 @@ export class SandboxEmailSender {
 
     public async captureBlocked(
         teamId: number,
-        isTest: boolean,
+        source: SandboxSendSource,
         reason: SandboxEmailBlockReason,
         blockedRecipientCount: number
     ): Promise<void> {
         await this.captureEvent(
             teamId,
             'workflows sandbox email blocked',
-            { reason, blocked_recipient_count: blockedRecipientCount, is_test: isTest },
-            () => this.isFirstBlockInWindow(teamId, isTest, reason)
+            { reason, blocked_recipient_count: blockedRecipientCount, is_test: source === 'test', source },
+            () => this.isFirstBlockInWindow(teamId, source, reason)
         )
     }
 
@@ -309,14 +349,14 @@ export class SandboxEmailSender {
 
     private async isFirstBlockInWindow(
         teamId: number,
-        isTest: boolean,
+        source: SandboxSendSource,
         reason: SandboxEmailBlockReason
     ): Promise<boolean> {
         const firstBlock = await this.blockedEventRedis.useClient(
             { name: 'sandbox-blocked-event', timeout: 1000 },
             (client) =>
                 client.set(
-                    `@posthog/workflows-sandbox-blocked/${teamId}/${reason}/${isTest ? 'test' : 'live'}`,
+                    `@posthog/workflows-sandbox-blocked/${teamId}/${reason}/${BLOCKED_EVENT_WINDOW_NAMES[source]}`,
                     '1',
                     'EX',
                     BLOCKED_EVENT_WINDOW_SECONDS,
diff --git a/nodejs/src/cdp/services/messaging/sandbox-envelope.ts b/nodejs/src/cdp/services/messaging/sandbox-envelope.ts
new file mode 100644
index 000000000000..410ad87259d1
--- /dev/null
+++ b/nodejs/src/cdp/services/messaging/sandbox-envelope.ts
@@ -0,0 +1,72 @@
+export type SandboxEnvelope = { to: string; cc: string[]; bcc: string[] }
+
+type QuoteScan = { unquotedIndexes: number[]; balanced: boolean }
+
+function scanQuotes(text: string): QuoteScan {
+    const unquotedIndexes: number[] = []
+    let quoted = false
+    let escaped = false
+    for (let index = 0; index < text.length; index++) {
+        const character = text[index]
+        if (escaped) {
+            escaped = false
+        } else if (quoted && character === '\\') {
+            escaped = true
+        } else if (character === '"') {
+            quoted = !quoted
+        } else if (!quoted) {
+            unquotedIndexes.push(index)
+        }
+    }
+    return { unquotedIndexes, balanced: !quoted }
+}
+
+function splitOutsideQuotes(text: string): string[] {
+    const commas = scanQuotes(text).unquotedIndexes.filter((index) => text[index] === ',')
+    return [-1, ...commas].map((comma, position) => text.slice(comma + 1, commas[position] ?? text.length))
+}
+
+function angleAddress(entry: string): string | undefined {
+    const { unquotedIndexes, balanced } = scanQuotes(entry)
+    const opening = unquotedIndexes.find((index) => entry[index] === '<')
+    const closing = entry.length - 1
+    if (!balanced || opening === undefined || entry[closing] !== '>' || /[\r\n]/.test(entry)) {
+        return undefined
+    }
+    const unquotedBetween = (start: number, end: number): string =>
+        unquotedIndexes
+            .filter((index) => index >= start && index < end)
+            .map((index) => entry[index])
+            .join('')
+    const mailbox = entry.slice(opening + 1, closing).trim()
+    const isFramed = !/[>@;]/.test(unquotedBetween(0, opening)) && !/[<>]/.test(unquotedBetween(opening + 1, closing))
+    return isFramed && mailbox ? mailbox : undefined
+}
+
+// Unlike `extractEmailsFromAddressList`, this keeps quoted commas inside one entry and leaves an
+// ambiguous mailbox whole, so the member check sees exactly the addresses SES would deliver to.
+function sandboxAddressList(value?: string): string[] {
+    return splitOutsideQuotes(value ?? '')
+        .map((entry) => entry.trim())
+        .filter(Boolean)
+        .map((entry) => angleAddress(entry) ?? entry)
+}
+
+export function sandboxEnvelope(to: string, cc?: string, bcc?: string): SandboxEnvelope {
+    const seen = new Set<string>()
+    const firstSightings = (addresses: string[]): string[] =>
+        addresses.filter((address) => {
+            const key = address.toLowerCase()
+            if (seen.has(key)) {
+                return false
+            }
+            seen.add(key)
+            return true
+        })
+    const [envelopeTo] = firstSightings([to.trim()])
+    return { to: envelopeTo, cc: firstSightings(sandboxAddressList(cc)), bcc: firstSightings(sandboxAddressList(bcc)) }
+}
+
+export function sandboxEnvelopeRecipients({ to, cc, bcc }: SandboxEnvelope): string[] {
+    return [to, ...cc, ...bcc]
+}
diff --git a/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
new file mode 100644
index 000000000000..2bfb9107a53a
--- /dev/null
+++ b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
@@ -0,0 +1,176 @@
+import { setTimeout as delay } from 'timers/promises'
+import { z } from 'zod'
+
+import { HogFlow } from '~/cdp/schema/hogflow'
+
+import { AcceptedSend } from '../hogflows/transactional-email'
+import { IntegrationManagerService } from '../managers/integration-manager.service'
+import { OrganizationMembersService } from '../managers/organization-members.service'
+import { selectEmailSenderIntegrationId } from './email-sender-selection'
+import { SandboxEmailSender } from './sandbox-email-sender'
+import { sandboxEnvelope, sandboxEnvelopeRecipients } from './sandbox-envelope'
+
+export type SandboxSendRefusal =
+    | 'sandbox_recipient_not_allowed'
+    | 'sandbox_limit_reached'
+    | 'sending_suspended'
+    | 'sandbox_check_failed'
+
+export type DeliveryMode = 'own' | 'sandbox'
+
+export type SandboxSendAdmission =
+    | { deliveryMode: 'own' }
+    | { deliveryMode: 'sandbox'; refusal: SandboxSendRefusal | null }
+
+export type SandboxSendRequest = {
+    teamId: number
+    invocationId: string
+    hogFlow: HogFlow
+    send: AcceptedSend | undefined
+    to: string
+}
+
+const StoredSenderSchema = z.object({
+    from: z.object({ integrationId: z.number().int(), integrationIds: z.array(z.number().int()).optional() }),
+})
+
+const StoredCopiesSchema = z.object({
+    cc: z.string().optional(),
+    bcc: z.string().optional(),
+})
+
+type StoredSender = z.infer<typeof StoredSenderSchema>['from']
+type StoredCopies = z.infer<typeof StoredCopiesSchema>
+
+const BLOCKED_EVENT_WAIT_MS = 1000
+
+const isTemplated = (copies: string): boolean => /[{}]/.test(copies)
+
+export class SandboxSendGate {
+    constructor(
+        private sandboxSender: SandboxEmailSender,
+        private organizationMembers: OrganizationMembersService,
+        private integrationManager: IntegrationManagerService
+    ) {}
+
+    public async admit(request: SandboxSendRequest): Promise<SandboxSendAdmission> {
+        const step = storedEmailStep(request.hogFlow)
+        if (!(await this.sendsThroughSandbox(request, storedSender(step)))) {
+            return { deliveryMode: 'own' }
+        }
+        return { deliveryMode: 'sandbox', refusal: await this.refusal(request, storedCopies(step)) }
+    }
+
+    private async refusal(
+        request: SandboxSendRequest,
+        copies: StoredCopies | null
+    ): Promise<SandboxSendRefusal | null> {
+        const suspension = await this.suspensionRefusal(request.teamId)
+        if (suspension) {
+            return suspension
+        }
+        if (!copies || isTemplated(copies.cc ?? '') || isTemplated(copies.bcc ?? '')) {
+            return await this.refuse(request.teamId, 'sandbox_recipient_not_allowed', 'recipient_not_member', 1)
+        }
+        const recipients = sandboxEnvelopeRecipients(sandboxEnvelope(request.to, copies.cc, copies.bcc))
+        return (
+            (await this.recipientRefusal(request.teamId, recipients)) ??
+            (await this.dailyCapRefusal(request.teamId, recipients))
+        )
+    }
+
+    private async suspensionRefusal(teamId: number): Promise<SandboxSendRefusal | null> {
+        if (!this.sandboxSender.canSend) {
+            return await this.refuse(teamId, 'sending_suspended', 'switch_off', 0)
+        }
+        const gate = await this.sandboxSender.pauseGate()
+        switch (gate) {
+            case 'open':
+                return null
+            case 'paused':
+                return await this.refuse(teamId, 'sending_suspended', 'paused', 0)
+            case 'check_failed':
+                return await this.refuse(teamId, 'sandbox_check_failed', 'check_failed', 0)
+        }
+    }
+
+    private async dailyCapRefusal(teamId: number, recipients: string[]): Promise<SandboxSendRefusal | null> {
+        const peek = await this.sandboxSender.peekDailyCaps(teamId, recipients)
+        switch (peek.type) {
+            case 'available':
+                return null
+            case 'check_failed':
+                return await this.refuse(teamId, 'sandbox_check_failed', 'check_failed', recipients.length)
+            case 'project_cap_reached':
+                return await this.refuse(teamId, 'sandbox_limit_reached', 'cap_reached', recipients.length)
+            case 'recipient_cap_reached':
+                return await this.refuse(teamId, 'sandbox_limit_reached', 'cap_reached', peek.addresses.length)
+        }
+    }
+
+    private async sendsThroughSandbox(request: SandboxSendRequest, stored: StoredSender | null): Promise<boolean> {
+        const sender = request.send?.from ?? stored
+        if (!sender) {
+            return false
+        }
+        const integration = await this.integrationManager.get(
+            selectEmailSenderIntegrationId(request.invocationId, sender)
+        )
+        return (
+            integration?.team_id === request.teamId &&
+            integration.kind === 'email' &&
+            integration.config.provider === 'sandbox'
+        )
+    }
+
+    private async recipientRefusal(teamId: number, recipients: string[]): Promise<SandboxSendRefusal | null> {
+        let check
+        try {
+            check = await this.organizationMembers.checkRecipients(teamId, recipients)
+        } catch {
+            return await this.refuse(teamId, 'sandbox_check_failed', 'check_failed', recipients.length)
+        }
+        if (check.organizationBlocked) {
+            return await this.refuse(teamId, 'sending_suspended', 'organization_blocked', 0)
+        }
+        if (check.blockedRecipients.length) {
+            return await this.refuse(
+                teamId,
+                'sandbox_recipient_not_allowed',
+                check.blockedRecipients.every(({ unverifiedMember }) => unverifiedMember)
+                    ? 'recipient_unverified'
+                    : 'recipient_not_member',
+                check.blockedRecipients.length
+            )
+        }
+        return null
+    }
+
+    private async refuse(
+        teamId: number,
+        refusal: SandboxSendRefusal,
+        reason: Parameters<SandboxEmailSender['captureBlocked']>[2],
+        blockedRecipientCount: number
+    ): Promise<SandboxSendRefusal> {
+        await Promise.race([
+            this.sandboxSender.captureBlocked(teamId, 'api', reason, blockedRecipientCount),
+            delay(BLOCKED_EVENT_WAIT_MS, undefined, { ref: false }),
+        ])
+        return refusal
+    }
+}
+
+function storedSender(step: unknown): StoredSender | null {
+    const parsed = StoredSenderSchema.safeParse(step)
+    return parsed.success ? parsed.data.from : null
+}
+
+function storedCopies(step: unknown): StoredCopies | null {
+    const parsed = StoredCopiesSchema.safeParse(step)
+    return parsed.success ? parsed.data : null
+}
+
+function storedEmailStep(hogFlow: HogFlow): unknown {
+    const action = hogFlow.actions.find((candidate) => candidate.type === 'function_email')
+    return (action?.type === 'function_email' ? action.config.inputs?.email?.value : undefined) ?? {}
+}
diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts
index f5d72eee2efd..f93b00bc25e6 100644
--- a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts
+++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts
@@ -300,6 +300,31 @@ redis.call('SET', claimKey, cjson.encode(granted), 'NX', 'PX', replayTtlMs)
 return granted
 `
 
+const PEEK_ALL_OR_NOTHING_LUA = `
+local time = redis.call('TIME')
+local now = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
+local denied = {0}
+for i = 1, #KEYS do
+    local offset = (i - 1) * 3
+    local requested = tonumber(ARGV[offset + 1])
+    local capacity = tonumber(ARGV[offset + 2])
+    local refillPerSecond = tonumber(ARGV[offset + 3])
+    local existing = redis.call('hmget', KEYS[i], 'ts', 'pool')
+    local avail = capacity
+    if existing[1] ~= false and existing[2] ~= false then
+        local elapsedMs = math.max(0, now - tonumber(existing[1]))
+        avail = math.min(capacity, tonumber(existing[2]) + (elapsedMs / 1000.0) * refillPerSecond)
+    end
+    if avail < requested then
+        table.insert(denied, i)
+    end
+end
+if #denied > 1 then
+    return denied
+end
+return {1}
+`
+
 // ioredis can replay unanswered commands, so returns need deduplication and an expiry.
 const RETURN_CLAIM_LUA = `
 local time = redis.call('TIME')
@@ -447,6 +472,38 @@ export class RateLimiterService {
         }
     }
 
+    public async peekAllOrNothing(buckets: ClaimRequest[]): Promise<{ deniedIndexes: number[] } | null> {
+        const keys = buckets.map((bucket) => bucket.key)
+        try {
+            const result = await this.valkey.useClient(
+                { name: `rate-limiter:${this.config.name}:peekAllOrNothing`, timeout: 1000 },
+                (client) =>
+                    client.eval(
+                        PEEK_ALL_OR_NOTHING_LUA,
+                        buckets.length,
+                        ...keys,
+                        ...buckets.flatMap((bucket) =>
+                            [bucket.requested, bucket.capacity, bucket.refillPerSecond].map(String)
+                        )
+                    )
+            )
+            if (Array.isArray(result) && result.length === 1 && result[0] === 1) {
+                return { deniedIndexes: [] }
+            }
+            if (Array.isArray(result) && result[0] === 0 && result.length > 1) {
+                return { deniedIndexes: result.slice(1).map((index) => Number(index) - 1) }
+            }
+            logger.warn('🪙', `RateLimiterService(${this.config.name}) peek returned invalid result`, {
+                keys,
+                raw: result,
+            })
+            return null
+        } catch (err) {
+            logger.warn('🪙', `RateLimiterService(${this.config.name}) peek threw`, { keys, error: String(err) })
+            return null
+        }
+    }
+
     // All bucket keys must share a hash tag because clustered Valkey rejects cross-slot Lua calls.
     public async claimAllOrNothing(
         buckets: ClaimRequest[]