feat(cdp): deliver a code send's content on its own transactional email route

nodejs/src/cdp/_tests/fixtures-hogflows.ts

@@ -1,5 +1,5 @@
 import { HogFlow } from '~/cdp/schema/hogflow'
-import { PostgresRouter } from '~/common/utils/db/postgres'
+import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres'
 import { UUIDT } from '~/common/utils/utils'
 import { insertRow } from '~/tests/helpers/sql'
 
@@ -20,6 +20,20 @@ export const insertHogFlow = async (postgres: PostgresRouter, hogFlow: HogFlow):
     return res
 }
 
+export const hideHogFlowsFromReplica = (
+    postgres: PostgresRouter,
+    ids: HogFlow['id'][]
+): jest.SpyInstance<ReturnType<PostgresRouter['query']>, Parameters<PostgresRouter['query']>> => {
+    const query = postgres.query.bind(postgres)
+    return jest.spyOn(postgres, 'query').mockImplementation(async (target, queryString, values, tag, logLevel) => {
+        const result = await query(target, queryString, values, tag, logLevel)
+        if (target === PostgresUse.COMMON_READ && tag === 'fetchHogFlows') {
+            return { ...result, rows: result.rows.filter((row) => !ids.includes(row.id)) }
+        }
+        return result
+    })
+}
+
 export const createHogFlowInvocationContext = (
     data: Partial<HogFlowInvocationContext> = {}
 ): HogFlowInvocationContext => {

nodejs/src/cdp/async-functions/send-email.ts

@@ -1,13 +1,15 @@
 import { DateTime } from 'luxon'
 
 import { CyclotronInvocationQueueParametersEmailSchema } from '~/cdp/schema/cyclotron'
+import { acceptedSendEmailFields } from '~/cdp/services/hogflows/transactional-email'
 
 import { registerAsyncFunction } from '../async-function-registry'
 
 registerAsyncFunction('sendEmail', {
-    execute: (args, _context, result) => {
+    execute: (args, context, result) => {
         result.invocation.queueParameters = CyclotronInvocationQueueParametersEmailSchema.parse({
             ...args[0],
+            ...acceptedSendEmailFields(context.invocation.state.send),
             type: 'email',
         })
     },

nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts

@@ -89,6 +89,9 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker {
         const loadedInvocations: CyclotronJobInvocationHogFlow[] = []
         const failedInvocations: CyclotronJobInvocation[] = []
         const canceledResults: CyclotronJobInvocationResult[] = []
+        const hogFlows = await this.hogFlowManager.getHogFlowsCheckingPrimaryOnMiss(
+            invocations.filter((item) => !item.cancelRequestedAt).map((item) => item.functionId)
+        )
 
         await Promise.all(
             invocations.map(async (item) => {
@@ -103,7 +106,7 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker {
                 }
 
                 const team = await this.deps.teamManager.getTeam(item.teamId)
-                const hogFlow = await this.hogFlowManager.getHogFlow(item.functionId)
+                const hogFlow = hogFlows[item.functionId]
                 if (!hogFlow || !team) {
                     logger.error('⚠️', 'Error finding hog flow', {
                         id: item.functionId,

nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.test.ts

@@ -12,7 +12,7 @@ import { createTeam, createTestTeamFixture, getTeam } from '~/tests/helpers/sql'
 
 import { Hub, InternalPerson, Team } from '../../types'
 import { FixtureHogFlowBuilder } from '../_tests/builders/hogflow.builder'
-import { createHogFlowInvocationContext, insertHogFlow } from '../_tests/fixtures-hogflows'
+import { createHogFlowInvocationContext, hideHogFlowsFromReplica, insertHogFlow } from '../_tests/fixtures-hogflows'
 import {
     CyclotronJobInvocation,
     CyclotronJobInvocationHogFlow,
@@ -88,6 +88,7 @@ describe('CdpCyclotronWorkerHogFlow', () => {
     let team: Team
     let team2: Team
     let hogFlows: HogFlow[]
+    let jobQueue: ReturnType<typeof createMockJobQueue>
 
     const createSerializedHogFlowInvocation = (
         hogFlow: HogFlow,
@@ -160,11 +161,11 @@ describe('CdpCyclotronWorkerHogFlow', () => {
         ]
 
         const mockRepo = createMockPersonReadRepository(testPersons)
-        const mockJobQueue = createMockJobQueue()
+        jobQueue = createMockJobQueue()
         processor = new CdpCyclotronWorkerHogFlow(
             hub,
             { ...createCdpConsumerDeps(hub), personRepository: mockRepo },
-            mockJobQueue
+            jobQueue
         )
 
         hogFlows = []
@@ -587,6 +588,33 @@ describe('CdpCyclotronWorkerHogFlow', () => {
             })
         })
 
+        it('runs a flow the replica does not have yet, and fails one missing from both', async () => {
+            const [lateFlow, missingFlow] = ['Flow only the primary has', 'Flow neither database has'].map((name) =>
+                new FixtureHogFlowBuilder()
+                    .withName(name)
+                    .withTeamId(team.id)
+                    .withStatus('active')
+                    .withSimpleWorkflow()
+                    .build()
+            )
+            await insertHogFlow(hub.postgres, lateFlow)
+            hideHogFlowsFromReplica(hub.postgres, [lateFlow.id])
+            const [lateInvocation, missingInvocation] = [lateFlow, missingFlow].map((flow) =>
+                createSerializedHogFlowInvocation(flow, {
+                    event: { distinct_id: 'distinct_A_1', properties: {} } as any,
+                })
+            )
+
+            const results = await processor.processInvocations([lateInvocation, missingInvocation])
+
+            expect(results).toHaveLength(1)
+            expect(results[0]).toMatchObject({
+                finished: true,
+                invocation: { hogFlow: { id: lateFlow.id } },
+            })
+            expect(jobQueue.dequeueInvocations).toHaveBeenLastCalledWith([missingInvocation])
+        })
+
         it('terminates invocations as canceled when the workflow is disabled after being queued', async () => {
             const hogFlow = hogFlows[0]
 

nodejs/src/cdp/rerun/rerun-job.manager.test.ts

@@ -76,9 +76,11 @@ describe('RerunJobManager', () => {
 
         const jobId = await manager.enqueue(7, 'hog_function', functionId, {
             filter: { ...baseFilter, invocation_ids: ids },
+            replay_code_sends: true,
         })
 
         const state = fetchState(await queryJob(jobId))
+        expect(state.request.replay_code_sends).toBe(true)
         expect(state.function_kind).toBe('hog_function')
         expect(state.function_id).toBe(functionId)
         expect(state.request.filter.invocation_ids).toEqual(ids)
@@ -118,6 +120,7 @@ describe('RerunJobManager', () => {
         expect(state.function_kind).toBe('hog_flow')
         expect(state.request.filter).toMatchObject(filter)
         expect(state.request.filter.invocation_ids).toBeUndefined()
+        expect(state.request.replay_code_sends).toBe(false)
         expect(state.progress.cursor).toBeUndefined()
         expect(state.progress.done).toBe(false)
     })

nodejs/src/cdp/rerun/rerun-job.manager.ts

@@ -83,7 +83,7 @@ export class RerunJobManager {
         const state: RerunJobState = {
             function_kind: functionKind,
             function_id: functionId,
-            request: { filter },
+            request: { filter, replay_code_sends: request.replay_code_sends === true },
             progress: {
                 queued: 0,
                 skipped: 0,

nodejs/src/cdp/rerun/rerun-job.types.ts

@@ -68,6 +68,7 @@ export interface RerunFilter {
 
 export interface RerunRequest {
     filter: RerunFilter
+    replay_code_sends?: boolean
 }
 
 /**

nodejs/src/cdp/rerun/rerun-paginator.replay-fidelity.test.ts

@@ -39,7 +39,11 @@ describe('RerunPaginatorService replay fidelity (hog_flow)', () => {
 
     function buildPaginator(
         clickhouse: ClickHouseClient,
-        hogFlowQueue: jest.Mocked<CyclotronJobQueuePostgresV2>
+        hogFlowQueue: jest.Mocked<CyclotronJobQueuePostgresV2>,
+        monitoring = {
+            queueLogs: jest.fn(),
+            flush: jest.fn().mockResolvedValue(undefined),
+        } as unknown as jest.Mocked<HogFunctionMonitoringService>
     ): RerunPaginatorService {
         const hogFlowManager = {
             getHogFlow: jest.fn().mockResolvedValue({ id: functionId, team_id: teamId, variables: [] }),
@@ -53,11 +57,6 @@ describe('RerunPaginatorService replay fidelity (hog_flow)', () => {
             flush: jest.fn().mockResolvedValue(undefined),
             dropQueuedRowsFor: jest.fn(),
         } as unknown as jest.Mocked<HogInvocationResultsService>
-        const monitoring = {
-            queueLogs: jest.fn(),
-            flush: jest.fn().mockResolvedValue(undefined),
-        } as unknown as jest.Mocked<HogFunctionMonitoringService>
-
         return new RerunPaginatorService(
             clickhouse,
             hogFunctionManager,
@@ -99,6 +98,12 @@ describe('RerunPaginatorService replay fidelity (hog_flow)', () => {
             },
             personId: 'person-1',
             messageOptOuts: [{ identifier: 'opted-out@example.com', categoryId: '$all' }],
+            send: {
+                content: { subject: 'Welcome', html: '<p>Hi</p>', text: 'Hi' },
+                from: { integrationId: 12, email: 'security@example.com', name: 'Example' },
+                replyTo: ['help@example.com'],
+                tracking: { captureEngagement: false },
+            },
         }
         const rows = [
             {
@@ -117,7 +122,11 @@ describe('RerunPaginatorService replay fidelity (hog_flow)', () => {
         } as unknown as jest.Mocked<CyclotronJobQueuePostgresV2>
 
         const paginator = buildPaginator(fakeClickhouse(rows), hogFlowQueue)
-        await paginator.processPage(teamId, state, { jobId: 'rerun-1', createdAt: DateTime.now() })
+        await paginator.processPage(
+            teamId,
+            { ...state, request: { ...state.request, replay_code_sends: true } },
+            { jobId: 'rerun-1', createdAt: DateTime.now() }
+        )
 
         expect(hogFlowQueue.queueInvocations).toHaveBeenCalledTimes(1)
         const [enqueued, opts] = hogFlowQueue.queueInvocations.mock.calls[0]
@@ -133,10 +142,73 @@ describe('RerunPaginatorService replay fidelity (hog_flow)', () => {
         expect(invocation.state?.actionStepCount).toBe(3)
         expect(invocation.state?.customerTaskIdempotencyVersion).toBe(version)
         expect(invocation.state?.variables).toEqual({ ticket_id: '123' })
+        expect(invocation.state?.send).toEqual(persistedState.send)
         // Sticky rerun counter still increments so max_attempts guards apply.
         expect(invocation.state?.rerunAttempts).toBe(1)
     })
 
+    const codeSend = {
+        content: { subject: 'Welcome', html: '<p>Hi</p>', text: 'Hi' },
+        from: { integrationId: 12, email: 'security@example.com', name: 'Example' },
+        tracking: { captureEngagement: false },
+    }
+
+    it.each([
+        ['an email sent from code while replay_code_sends is undefined', { send: codeSend }, undefined, []],
+        ['an email sent from code while replay_code_sends is false', { send: codeSend }, false, []],
+        [
+            'a sensitive message whose content was not stored',
+            { sensitive_variables_omitted: true },
+            true,
+            [
+                expect.objectContaining({
+                    instance_id: 'inv-3',
+                    level: 'warn',
+                    message:
+                        "Not re-run: this run's message is sensitive, so its content wasn't stored and can't be sent again.",
+                }),
+            ],
+        ],
+    ] as const)('skips a run with %s', async (_name, stored, allowed, expectedLogs) => {
+        const persistedState = {
+            event: { uuid: 'evt-3', distinct_id: 'd-3', properties: {}, timestamp: '2026-06-01T09:00:00Z' },
+            actionStepCount: 1,
+            variables: {},
+            ...stored,
+        }
+        const rows = [
+            {
+                invocation_id: 'inv-3',
+                parent_run_id: '',
+                attempts: 0,
+                last_scheduled_at: '2026-06-01 09:00:00.000000',
+                first_scheduled_at: '2026-06-01 09:00:00.000000',
+                invocation_globals: JSON.stringify(persistedState),
+            },
+        ]
+
+        const hogFlowQueue = {
+            queueInvocations: jest.fn().mockResolvedValue(undefined),
+        } as unknown as jest.Mocked<CyclotronJobQueuePostgresV2>
+
+        const monitoring = {
+            queueLogs: jest.fn(),
+            flush: jest.fn().mockResolvedValue(undefined),
+        } as unknown as jest.Mocked<HogFunctionMonitoringService>
+        const paginator = buildPaginator(fakeClickhouse(rows), hogFlowQueue, monitoring)
+        await paginator.processPage(
+            teamId,
+            { ...state, request: { ...state.request, replay_code_sends: allowed } },
+            { jobId: 'rerun-3', createdAt: DateTime.now() }
+        )
+
+        expect(hogFlowQueue.queueInvocations.mock.calls.flatMap(([invocations]) => invocations)).toEqual([])
+        const runLogs = monitoring.queueLogs.mock.calls
+            .flatMap(([entries]) => entries)
+            .filter((entry) => entry.instance_id === 'inv-3')
+        expect(runLogs).toEqual(expectedLogs)
+    })
+
     it('leaves currentAction undefined for a flow that never progressed (fresh start is correct there)', async () => {
         const persistedState = {
             event: { uuid: 'evt-2', distinct_id: 'd-2', properties: {}, timestamp: '2026-06-01T09:00:00Z' },

nodejs/src/cdp/rerun/rerun-paginator.service.ts

@@ -641,7 +641,8 @@ export class RerunPaginatorService {
                         teamId,
                         state.function_kind,
                         state.function_id,
-                        row
+                        row,
+                        state.request.replay_code_sends === true
                     )
                     if (!invocation) {
                         counterRerunInvocationsSkipped.labels(state.function_kind, 'rehydrate_failed').inc()
@@ -682,7 +683,8 @@ export class RerunPaginatorService {
         teamId: number,
         functionKind: RerunFunctionKind,
         functionId: string,
-        row: InvocationRow
+        row: InvocationRow,
+        replayCodeSends: boolean
     ): Promise<CyclotronJobInvocation | null> {
         let parsedGlobals: unknown
         try {
@@ -774,6 +776,29 @@ export class RerunPaginatorService {
                     teamId,
                     invocation_id: row.invocation_id,
                 })
+                this.monitoringService.queueLogs(
+                    [
+                        {
+                            team_id: teamId,
+                            log_source: 'hog_flow',
+                            log_source_id: functionId,
+                            instance_id: row.invocation_id,
+                            timestamp: DateTime.now(),
+                            level: 'warn',
+                            message:
+                                "Not re-run: this run's message is sensitive, so its content wasn't stored and can't be sent again.",
+                        },
+                    ],
+                    'hog_flow'
+                )
+                return null
+            }
+            if (persistedState.send && !replayCodeSends) {
+                logger.warn('⚠️', 'Skipping rerun of an email sent from code while code sends are off', {
+                    functionId,
+                    teamId,
+                    invocation_id: row.invocation_id,
+                })
                 return null
             }
             const eventForFilter = persistedState.event ?? {}
@@ -839,6 +864,7 @@ export class RerunPaginatorService {
                 personId: persistedState.personId,
                 // Later message steps check these before the per-worker preferences cache, which can predate them.
                 messageOptOuts: persistedState.messageOptOuts,
+                send: persistedState.send,
                 // Sticky rerun counter — mirror the hog function path so the
                 // lifecycle row producer can derive `attempts` / `is_retry`
                 // for flows too, and the `max_attempts` guard actually trips.