From 423913740189a0dac6439f8a7a3015b1f4526d9b Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Sat, 31 Oct 2026 17:10:19 +0000
Subject: [PATCH 1/6] feat(cdp): admit sandbox code sends before they are
 queued

The cdp-api send route now runs the sandbox sender's gates for a send
that resolves to the sandbox integration: sender switch and tenant
pause, organization membership over to, cc and bcc, organization block,
and a read-only peek at the daily allowances. A refusal answers before
the send is queued, and the worker keeps its own checks.

The worker's SES envelope now lists each address once.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nodejs/src/cdp/cdp-api.serial.test.ts         | 292 +++++++++++++++++-
 nodejs/src/cdp/cdp-api.ts                     |  37 ++-
 nodejs/src/cdp/cdp-services.ts                |   5 +-
 .../services/hogflows/transactional-email.ts  |   3 +
 .../services/messaging/email.service.test.ts  |  43 ++-
 .../cdp/services/messaging/email.service.ts   | 108 +++----
 .../messaging/sandbox-email-sender.ts         |  86 ++++--
 .../services/messaging/sandbox-envelope.ts    |  72 +++++
 .../messaging/sandbox-send-admission.ts       | 152 +++++++++
 .../rate-limiter/rate-limiter.service.ts      |  57 ++++
 10 files changed, 757 insertions(+), 98 deletions(-)
 create mode 100644 nodejs/src/cdp/services/messaging/sandbox-envelope.ts
 create mode 100644 nodejs/src/cdp/services/messaging/sandbox-send-admission.ts

diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts
index f697fad5e839..dfc7ce18d9d6 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 { SandboxSendAdmission } from './services/messaging/sandbox-send-admission'
 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'
 
@@ -2823,6 +2835,282 @@ 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
+            ): SandboxEmailSender =>
+                new SandboxEmailSender(
+                    {
+                        enabled: true,
+                        tenantName: 'sandbox-tenant',
+                        configurationSetName: 'sandbox-email',
+                        fromAddress: 'fixed-sandbox@example.com',
+                        dailyTeamCap: '25',
+                        dailyRecipientCap: '5',
+                        ...settings,
+                    },
+                    hub.teamManager,
+                    sandboxRedis,
+                    new RateLimiterService(limiterRedis, { name: 'sandbox-email' }),
+                    new SandboxSenderStateService(hub.postgres, hub.pubSub)
+                )
+            const admitThrough = (
+                sender: SandboxEmailSender,
+                members = new OrganizationMembersService(hub.postgres)
+            ) => {
+                api['sandboxSendAdmission'] = new SandboxSendAdmission(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('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(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 }}' }],
+            ])('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 after its daily allowance is spent', 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 sandboxSend(address('member'), { idempotency_key })
+
+                expect(first.status).toEqual(202)
+                expect([retry.status, retry.body]).toEqual([
+                    202,
+                    { status: 'queued', invocation_id: first.body.invocation_id, replayed: true },
+                ])
+                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..0311f2e9dee8 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,11 @@ 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 {
+    SandboxSendAdmission,
+    SandboxSendRefusal,
+    SandboxSendRequest,
+} from './services/messaging/sandbox-send-admission'
 import {
     HogWatcherService,
     HogWatcherState,
@@ -173,6 +178,13 @@ 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,
+}
+
 const transactionalEmailFingerprint = (
     hogFlowId: string,
     { to, distinct_id, variables, send }: TransactionalEmailRequest
@@ -237,6 +249,7 @@ export class CdpApi {
     private emailValidationService: EmailValidationService
     private teamWorkflowsConfigService: TeamWorkflowsConfigService
     private emailStatusService: EmailStatusService
+    private sandboxSendAdmission: SandboxSendAdmission
     private resumeBatchJwt: ScopedServiceJwt
 
     constructor(
@@ -262,6 +275,11 @@ export class CdpApi {
         this.hogWatcherMirror = services.hogWatcherMirror
         this.invocationResultsService = services.invocationResultsService
         this.emailStatusService = services.emailStatusService
+        this.sandboxSendAdmission = new SandboxSendAdmission(
+            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.
@@ -1322,7 +1340,13 @@ export class CdpApi {
                 }
                 try {
                     stages.start('checks')
-                    const refusal = await this.refuseTransactionalEmail(team.id, hogFlow, to)
+                    const refusal = await this.refuseTransactionalEmail({
+                        teamId: team.id,
+                        invocationId: invocation.id,
+                        hogFlow,
+                        send,
+                        to,
+                    })
                     if (refusal) {
                         if (idempotencyKey) {
                             stages.start('idempotency')
@@ -1379,10 +1403,9 @@ export class CdpApi {
     // 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
+        request: SandboxSendRequest
     ): Promise<{ outcome: TransactionalEmailOutcome; status: number } | null> {
+        const { teamId, hogFlow, to } = request
         if (hogFlow.status !== 'active') {
             return { outcome: 'not_sendable', status: 400 }
         }
@@ -1399,6 +1422,10 @@ export class CdpApi {
         if (isLimited) {
             return { outcome: 'quota_limited', status: 429 }
         }
+        const sandboxRefusal = await this.sandboxSendAdmission.refusal(request)
+        if (sandboxRefusal) {
+            return { outcome: sandboxRefusal, status: SANDBOX_REFUSAL_STATUS[sandboxRefusal] }
+        }
         // 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 }
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/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts
index f7d3fdca0612..4d5d84ea0105 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({
diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts
index f5aa7c8b9986..b94e54576660 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,24 @@ 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.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 +899,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 +944,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 +1192,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..e65ea7844690 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 { CyclotronJobInvocationHogFlow, CyclotronJobInvocationHogFunction } from '~/cdp/types'
 import { RedisV2 } from '~/common/redis/redis-v2'
 import { logger } from '~/common/utils/logger'
 import { captureTeamEvent } from '~/common/utils/posthog'
@@ -52,6 +53,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 +97,22 @@ 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 runsUnderApiTrigger(invocation) ? 'api' : 'workflow'
+}
+
+function runsUnderApiTrigger(invocation: CyclotronJobInvocationHogFunction): boolean {
+    const { hogFlow } = invocation as Partial<Pick<CyclotronJobInvocationHogFlow, 'hogFlow'>>
+    return hogFlow?.trigger?.type === 'api'
+}
+
 function htmlBody(document: DefaultTreeAdapterMap['document']): DefaultTreeAdapterMap['element'] | undefined {
     const root = document.childNodes.find(defaultTreeAdapter.isElementNode)
     return root?.childNodes.find(
@@ -199,15 +229,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 +271,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 +303,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 +320,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 +353,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-admission.ts b/nodejs/src/cdp/services/messaging/sandbox-send-admission.ts
new file mode 100644
index 000000000000..50cbf5351a4c
--- /dev/null
+++ b/nodejs/src/cdp/services/messaging/sandbox-send-admission.ts
@@ -0,0 +1,152 @@
+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 SandboxSendRequest = {
+    teamId: number
+    invocationId: string
+    hogFlow: HogFlow
+    send: AcceptedSend | undefined
+    to: string
+}
+
+const StoredEmailStepSchema = z.object({
+    from: z
+        .object({ integrationId: z.number().int().nullish(), integrationIds: z.array(z.number().int()).optional() })
+        .optional(),
+    cc: z.string().optional(),
+    bcc: z.string().optional(),
+})
+
+type StoredEmailStep = z.infer<typeof StoredEmailStepSchema>
+
+const isTemplated = (copies: string): boolean => /[{}]/.test(copies)
+
+export class SandboxSendAdmission {
+    constructor(
+        private sandboxSender: SandboxEmailSender,
+        private organizationMembers: OrganizationMembersService,
+        private integrationManager: IntegrationManagerService
+    ) {}
+
+    public async refusal(request: SandboxSendRequest): Promise<SandboxSendRefusal | null> {
+        const step = storedEmailStep(request.hogFlow)
+        if (!(await this.sendsThroughSandbox(request, step))) {
+            return null
+        }
+        const suspension = await this.suspensionRefusal(request.teamId)
+        if (suspension) {
+            return suspension
+        }
+        if (!step || isTemplated(step.cc ?? '') || isTemplated(step.bcc ?? '')) {
+            return await this.refuse(request.teamId, 'sandbox_recipient_not_allowed', 'recipient_not_member', 1)
+        }
+        const recipients = sandboxEnvelopeRecipients(sandboxEnvelope(request.to, step.cc, step.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, step: StoredEmailStep | null): Promise<boolean> {
+        const sender = request.send?.from ?? storedSender(step)
+        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 this.sandboxSender.captureBlocked(teamId, 'api', reason, blockedRecipientCount)
+        return refusal
+    }
+}
+
+function storedSender(step: StoredEmailStep | null): { integrationId: number; integrationIds?: number[] } | null {
+    const integrationId = step?.from?.integrationId
+    return integrationId == null ? null : { integrationId, integrationIds: step?.from?.integrationIds }
+}
+
+function storedEmailStep(hogFlow: HogFlow): StoredEmailStep | null {
+    const action = hogFlow.actions.find((candidate) => candidate.type === 'function_email')
+    const value = action?.type === 'function_email' ? action.config.inputs?.email?.value : undefined
+    const parsed = StoredEmailStepSchema.safeParse(value ?? {})
+    return parsed.success ? parsed.data : null
+}
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[]

From ef32ebb6335e309a60fce08ef51c167c72fcee29 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Sat, 31 Oct 2026 17:19:04 +0000
Subject: [PATCH 2/6] feat(cdp): answer a code send with the delivery mode it
 went out with

The send route now says whether an accepted send goes out through the
project's own sender or the sandbox sender. The idempotency record keeps
that answer, so a replay reports the mode of the first send even when
the project's sender changed since.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nodejs/src/cdp/cdp-api.serial.test.ts         | 30 ++++++--
 nodejs/src/cdp/cdp-api.ts                     | 72 ++++++++++++-------
 .../cdp/services/hogflows/idempotency-keys.ts | 43 ++++++-----
 ...send-admission.ts => sandbox-send-gate.ts} | 19 ++++-
 4 files changed, 114 insertions(+), 50 deletions(-)
 rename nodejs/src/cdp/services/messaging/{sandbox-send-admission.ts => sandbox-send-gate.ts} (91%)

diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts
index dfc7ce18d9d6..4ce07eb87975 100644
--- a/nodejs/src/cdp/cdp-api.serial.test.ts
+++ b/nodejs/src/cdp/cdp-api.serial.test.ts
@@ -51,7 +51,7 @@ import {
 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 { SandboxSendAdmission } from './services/messaging/sandbox-send-admission'
+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'
@@ -2342,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([
@@ -2376,6 +2377,7 @@ describe('CDP API', () => {
                     status: 'queued',
                     invocation_id: sent.body.invocation_id,
                     replayed: true,
+                    delivery_mode: 'own',
                 })
                 expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
             })
@@ -2546,6 +2548,7 @@ describe('CDP API', () => {
                     status: 'queued',
                     invocation_id: first.body.invocation_id,
                     replayed: true,
+                    delivery_mode: 'own',
                 })
                 expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
             })
@@ -2892,7 +2895,7 @@ describe('CDP API', () => {
                 sender: SandboxEmailSender,
                 members = new OrganizationMembersService(hub.postgres)
             ) => {
-                api['sandboxSendAdmission'] = new SandboxSendAdmission(sender, members, cdpDeps.integrationManager)
+                api['sandboxSendGate'] = new SandboxSendGate(sender, members, cdpDeps.integrationManager)
             }
             const sandboxSend = (to: string, extra: Record<string, unknown> = {}) =>
                 sendAccepted(hogFlow.id, {
@@ -2980,6 +2983,7 @@ describe('CDP API', () => {
                 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)
             })
 
@@ -3094,19 +3098,33 @@ describe('CDP API', () => {
                 expect(mockQueueInvocations).not.toHaveBeenCalled()
             })
 
-            it('replays a keyed send after its daily allowance is spent', async () => {
+            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 sandboxSend(address('member'), { idempotency_key })
+                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 },
+                    {
+                        status: 'queued',
+                        invocation_id: first.body.invocation_id,
+                        replayed: true,
+                        delivery_mode: 'sandbox',
+                    },
                 ])
                 expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
             })
diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts
index 0311f2e9dee8..919e9652cf42 100644
--- a/nodejs/src/cdp/cdp-api.ts
+++ b/nodejs/src/cdp/cdp-api.ts
@@ -76,10 +76,11 @@ import { EmailValidationService } from './services/messaging/email-validation.se
 import { EmailTrackingCodeSigner } from './services/messaging/helpers/tracking-code'
 import { RecipientTokensService } from './services/messaging/recipient-tokens.service'
 import {
-    SandboxSendAdmission,
+    DeliveryMode,
+    SandboxSendGate,
     SandboxSendRefusal,
     SandboxSendRequest,
-} from './services/messaging/sandbox-send-admission'
+} from './services/messaging/sandbox-send-gate'
 import {
     HogWatcherService,
     HogWatcherState,
@@ -185,6 +186,21 @@ const SANDBOX_REFUSAL_STATUS: Record<SandboxSendRefusal, number> = {
     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
@@ -244,12 +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 sandboxSendAdmission: SandboxSendAdmission
+    private sandboxSendGate: SandboxSendGate
     private resumeBatchJwt: ScopedServiceJwt
 
     constructor(
@@ -275,7 +291,7 @@ export class CdpApi {
         this.hogWatcherMirror = services.hogWatcherMirror
         this.invocationResultsService = services.invocationResultsService
         this.emailStatusService = services.emailStatusService
-        this.sandboxSendAdmission = new SandboxSendAdmission(
+        this.sandboxSendGate = new SandboxSendGate(
             services.sandboxEmailSender,
             services.organizationMembers,
             deps.integrationManager
@@ -338,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'
         )
@@ -1329,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' })
@@ -1338,22 +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({
+                    const check = await this.checkTransactionalEmail({
                         teamId: team.id,
                         invocationId: invocation.id,
                         hogFlow,
                         send,
                         to,
                     })
-                    if (refusal) {
+                    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 {
@@ -1373,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' })
@@ -1388,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),
@@ -1402,35 +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(
-        request: SandboxSendRequest
-    ): 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 sandboxRefusal = await this.sandboxSendAdmission.refusal(request)
-        if (sandboxRefusal) {
-            return { outcome: sandboxRefusal, status: SANDBOX_REFUSAL_STATUS[sandboxRefusal] }
+        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/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/messaging/sandbox-send-admission.ts b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
similarity index 91%
rename from nodejs/src/cdp/services/messaging/sandbox-send-admission.ts
rename to nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
index 50cbf5351a4c..a8623c7de1ed 100644
--- a/nodejs/src/cdp/services/messaging/sandbox-send-admission.ts
+++ b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
@@ -15,6 +15,12 @@ export type SandboxSendRefusal =
     | '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
@@ -35,18 +41,25 @@ type StoredEmailStep = z.infer<typeof StoredEmailStepSchema>
 
 const isTemplated = (copies: string): boolean => /[{}]/.test(copies)
 
-export class SandboxSendAdmission {
+export class SandboxSendGate {
     constructor(
         private sandboxSender: SandboxEmailSender,
         private organizationMembers: OrganizationMembersService,
         private integrationManager: IntegrationManagerService
     ) {}
 
-    public async refusal(request: SandboxSendRequest): Promise<SandboxSendRefusal | null> {
+    public async admit(request: SandboxSendRequest): Promise<SandboxSendAdmission> {
         const step = storedEmailStep(request.hogFlow)
         if (!(await this.sendsThroughSandbox(request, step))) {
-            return null
+            return { deliveryMode: 'own' }
         }
+        return { deliveryMode: 'sandbox', refusal: await this.refusal(request, step) }
+    }
+
+    private async refusal(
+        request: SandboxSendRequest,
+        step: StoredEmailStep | null
+    ): Promise<SandboxSendRefusal | null> {
         const suspension = await this.suspensionRefusal(request.teamId)
         if (suspension) {
             return suspension

From 08f928264c7baadadad98b486380d3606d01f6ca Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Sat, 31 Oct 2026 18:10:52 +0000
Subject: [PATCH 3/6] fix(cdp): admit a sandbox send whose stored copies can't
 be read

The stored sender is read on its own, so a stored cc or bcc that isn't
text no longer hides a sandbox sender from the admission check. Worker
sandbox events name an API send by its transactional email event.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nodejs/src/cdp/cdp-api.serial.test.ts         |  1 +
 .../services/messaging/email.service.test.ts  | 15 +++++++
 .../messaging/sandbox-email-sender.ts         | 10 ++---
 .../services/messaging/sandbox-send-gate.ts   | 43 +++++++++++--------
 4 files changed, 43 insertions(+), 26 deletions(-)

diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts
index 4ce07eb87975..317a4ab3f0f6 100644
--- a/nodejs/src/cdp/cdp-api.serial.test.ts
+++ b/nodejs/src/cdp/cdp-api.serial.test.ts
@@ -3078,6 +3078,7 @@ describe('CDP API', () => {
                 ['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')!
diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts
index b94e54576660..ba1254f4396f 100644
--- a/nodejs/src/cdp/services/messaging/email.service.test.ts
+++ b/nodejs/src/cdp/services/messaging/email.service.test.ts
@@ -633,6 +633,21 @@ describe('EmailService', () => {
                 })
             })
 
+            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)
diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts
index e65ea7844690..c43d3f16a104 100644
--- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts
+++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts
@@ -2,7 +2,7 @@ import { createHash } from 'crypto'
 import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5'
 
 import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron'
-import { CyclotronJobInvocationHogFlow, CyclotronJobInvocationHogFunction } from '~/cdp/types'
+import { CyclotronJobInvocationHogFunction } from '~/cdp/types'
 import { RedisV2 } from '~/common/redis/redis-v2'
 import { logger } from '~/common/utils/logger'
 import { captureTeamEvent } from '~/common/utils/posthog'
@@ -10,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
@@ -105,12 +106,7 @@ export function sandboxSendSource(invocation: CyclotronJobInvocationHogFunction,
     if (isTest) {
         return 'test'
     }
-    return runsUnderApiTrigger(invocation) ? 'api' : 'workflow'
-}
-
-function runsUnderApiTrigger(invocation: CyclotronJobInvocationHogFunction): boolean {
-    const { hogFlow } = invocation as Partial<Pick<CyclotronJobInvocationHogFlow, 'hogFlow'>>
-    return hogFlow?.trigger?.type === 'api'
+    return isTransactionalEmailSend(invocation) ? 'api' : 'workflow'
 }
 
 function htmlBody(document: DefaultTreeAdapterMap['document']): DefaultTreeAdapterMap['element'] | undefined {
diff --git a/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
index a8623c7de1ed..3dd0dce8e231 100644
--- a/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
+++ b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
@@ -29,15 +29,17 @@ export type SandboxSendRequest = {
     to: string
 }
 
-const StoredEmailStepSchema = z.object({
-    from: z
-        .object({ integrationId: z.number().int().nullish(), integrationIds: z.array(z.number().int()).optional() })
-        .optional(),
+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 StoredEmailStep = z.infer<typeof StoredEmailStepSchema>
+type StoredSender = z.infer<typeof StoredSenderSchema>['from']
+type StoredCopies = z.infer<typeof StoredCopiesSchema>
 
 const isTemplated = (copies: string): boolean => /[{}]/.test(copies)
 
@@ -50,24 +52,24 @@ export class SandboxSendGate {
 
     public async admit(request: SandboxSendRequest): Promise<SandboxSendAdmission> {
         const step = storedEmailStep(request.hogFlow)
-        if (!(await this.sendsThroughSandbox(request, step))) {
+        if (!(await this.sendsThroughSandbox(request, storedSender(step)))) {
             return { deliveryMode: 'own' }
         }
-        return { deliveryMode: 'sandbox', refusal: await this.refusal(request, step) }
+        return { deliveryMode: 'sandbox', refusal: await this.refusal(request, storedCopies(step)) }
     }
 
     private async refusal(
         request: SandboxSendRequest,
-        step: StoredEmailStep | null
+        copies: StoredCopies | null
     ): Promise<SandboxSendRefusal | null> {
         const suspension = await this.suspensionRefusal(request.teamId)
         if (suspension) {
             return suspension
         }
-        if (!step || isTemplated(step.cc ?? '') || isTemplated(step.bcc ?? '')) {
+        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, step.cc, step.bcc))
+        const recipients = sandboxEnvelopeRecipients(sandboxEnvelope(request.to, copies.cc, copies.bcc))
         return (
             (await this.recipientRefusal(request.teamId, recipients)) ??
             (await this.dailyCapRefusal(request.teamId, recipients))
@@ -103,8 +105,8 @@ export class SandboxSendGate {
         }
     }
 
-    private async sendsThroughSandbox(request: SandboxSendRequest, step: StoredEmailStep | null): Promise<boolean> {
-        const sender = request.send?.from ?? storedSender(step)
+    private async sendsThroughSandbox(request: SandboxSendRequest, stored: StoredSender | null): Promise<boolean> {
+        const sender = request.send?.from ?? stored
         if (!sender) {
             return false
         }
@@ -152,14 +154,17 @@ export class SandboxSendGate {
     }
 }
 
-function storedSender(step: StoredEmailStep | null): { integrationId: number; integrationIds?: number[] } | null {
-    const integrationId = step?.from?.integrationId
-    return integrationId == null ? null : { integrationId, integrationIds: step?.from?.integrationIds }
+function storedSender(step: unknown): StoredSender | null {
+    const parsed = StoredSenderSchema.safeParse(step)
+    return parsed.success ? parsed.data.from : null
 }
 
-function storedEmailStep(hogFlow: HogFlow): StoredEmailStep | null {
-    const action = hogFlow.actions.find((candidate) => candidate.type === 'function_email')
-    const value = action?.type === 'function_email' ? action.config.inputs?.email?.value : undefined
-    const parsed = StoredEmailStepSchema.safeParse(value ?? {})
+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) ?? {}
+}

From f3832e2a05a4050d691e0391c70b66d1124a900a Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Sat, 31 Oct 2026 19:07:16 +0000
Subject: [PATCH 4/6] fix(cdp): read an integration the replica lacks from the
 primary

A sandbox code send provisions its integration and queues the send in one
request, so the first lookup can reach the read replica before the row does.
The loader cached that miss, the admission gate took the send for an own
sender, and the worker could fail it as a missing integration.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../integration-manager.service.test.ts        |  8 ++++++++
 .../managers/integration-manager.service.ts    | 18 ++++++++++++++----
 2 files changed, 22 insertions(+), 4 deletions(-)

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..6afaef9b6a94 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,14 @@ describe('IntegrationManager', () => {
         expect(item?.config).toEqual({ team: 'foobar' })
     })
 
+    it('finds an integration the read replica does not have yet', async () => {
+        jest.spyOn(hub.postgres, 'query').mockResolvedValueOnce({ rows: [] } as any)
+
+        const item = await manager.get(integrations[0].id)
+
+        expect(item?.config).toEqual({ team: 'foobar' })
+    })
+
     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,
             })

From c864a8489b107c019a393a05b4155fb86fb372b2 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Sat, 31 Oct 2026 19:07:22 +0000
Subject: [PATCH 5/6] fix(cdp): answer a sandbox refusal without waiting on its
 blocked event

The blocked event's first-in-window check uses a Redis client with no
command deadline, so a store that stops answering held the refusal until
the caller timed out. The gate now waits at most a second for it.

The checks stage's metric help now names the sandbox checks it runs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nodejs/src/cdp/cdp-api.serial.test.ts         | 19 +++++++++++++++++--
 .../services/hogflows/transactional-email.ts  |  2 +-
 .../services/messaging/sandbox-send-gate.ts   |  8 +++++++-
 3 files changed, 25 insertions(+), 4 deletions(-)

diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts
index 317a4ab3f0f6..86235669747f 100644
--- a/nodejs/src/cdp/cdp-api.serial.test.ts
+++ b/nodejs/src/cdp/cdp-api.serial.test.ts
@@ -2874,7 +2874,8 @@ describe('CDP API', () => {
             }
             const sandboxSender = (
                 settings: Partial<SandboxEmailSenderConfig> = {},
-                limiterRedis: RedisV2 = sandboxRedis
+                limiterRedis: RedisV2 = sandboxRedis,
+                blockedEventRedis: RedisV2 = sandboxRedis
             ): SandboxEmailSender =>
                 new SandboxEmailSender(
                     {
@@ -2887,7 +2888,7 @@ describe('CDP API', () => {
                         ...settings,
                     },
                     hub.teamManager,
-                    sandboxRedis,
+                    blockedEventRedis,
                     new RateLimiterService(limiterRedis, { name: 'sandbox-email' }),
                     new SandboxSenderStateService(hub.postgres, hub.pubSub)
                 )
@@ -2976,6 +2977,20 @@ describe('CDP API', () => {
                 )
             })
 
+            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' }))
 
diff --git a/nodejs/src/cdp/services/hogflows/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts
index 4d5d84ea0105..f49ae3561114 100644
--- a/nodejs/src/cdp/services/hogflows/transactional-email.ts
+++ b/nodejs/src/cdp/services/hogflows/transactional-email.ts
@@ -91,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/messaging/sandbox-send-gate.ts b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
index 3dd0dce8e231..2bfb9107a53a 100644
--- a/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
+++ b/nodejs/src/cdp/services/messaging/sandbox-send-gate.ts
@@ -1,3 +1,4 @@
+import { setTimeout as delay } from 'timers/promises'
 import { z } from 'zod'
 
 import { HogFlow } from '~/cdp/schema/hogflow'
@@ -41,6 +42,8 @@ const StoredCopiesSchema = z.object({
 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 {
@@ -149,7 +152,10 @@ export class SandboxSendGate {
         reason: Parameters<SandboxEmailSender['captureBlocked']>[2],
         blockedRecipientCount: number
     ): Promise<SandboxSendRefusal> {
-        await this.sandboxSender.captureBlocked(teamId, 'api', reason, blockedRecipientCount)
+        await Promise.race([
+            this.sandboxSender.captureBlocked(teamId, 'api', reason, blockedRecipientCount),
+            delay(BLOCKED_EVENT_WAIT_MS, undefined, { ref: false }),
+        ])
         return refusal
     }
 }

From 87af9d607ca788bb257de522b7b80276d7ed47bc Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Sat, 31 Oct 2026 19:45:31 +0000
Subject: [PATCH 6/6] chore(cdp): check that an integration the replica lacks
 comes from the primary

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../services/managers/integration-manager.service.test.ts    | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

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 6afaef9b6a94..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,12 +70,13 @@ describe('IntegrationManager', () => {
         expect(item?.config).toEqual({ team: 'foobar' })
     })
 
-    it('finds an integration the read replica does not have yet', async () => {
-        jest.spyOn(hub.postgres, 'query').mockResolvedValueOnce({ rows: [] } as any)
+    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 () => {