From 6cd00b99f401eb2a2ca4c5516d88ee4469605f6d Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Mon, 12 Oct 2026 03:57:57 +0000
Subject: [PATCH 1/5] fix(workflows): keep queue and priority when an email
 reschedules

A rescheduled email dropped its queue metadata, so the run returned to
the hogflow queue at the email class priority instead of its own.
Carry the metadata through SES throttles, workflow pacing, team caps
and consecutive inline sends.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../internal/workflows-email-sending-tiers.md |  1 +
 .../services/hog-executor-async.service.ts    |  1 +
 .../hogflows/hogflow-executor.service.test.ts | 64 +++++++++++++++----
 .../hogflows/hogflow-executor.service.ts      |  1 +
 .../services/messaging/email.service.test.ts  |  4 ++
 .../cdp/services/messaging/email.service.ts   |  3 +
 6 files changed, 63 insertions(+), 11 deletions(-)

diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md
index b511cea77ec8..d2b098c6b8a4 100644
--- a/docs/internal/workflows-email-sending-tiers.md
+++ b/docs/internal/workflows-email-sending-tiers.md
@@ -65,6 +65,7 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
 
 - The batch audience cap is decided when the batch is dispatched. Adding an email step to the workflow while a batch is queued does not re-cap it; the send-time buckets still cap every email at execution. This is why enforcement requires the worker caps to be deployed (see the rollout order).
 - Test-panel sends bypass the team buckets on purpose, matching the per-workflow rate limit.
+- Rescheduled emails keep their email queue priority while waiting. After a team-cap, workflow-pacing, or SES-throttle retry, the workflow restores its origin queue and priority when it next leaves the email queue.
 - The buckets are token buckets: a full idle bucket plus refill allows up to roughly twice the stated cap in the very first period. The bucket TTLs exceed the refill periods so this does not recur from idling.
 - A denied send parks until every short bucket has refilled enough to cover it, instead of retrying on a fixed few-minute cadence.
   The computed wait is capped at one hour and then jittered 1x to 2x, so a parked send can wait just under two hours between attempts.
diff --git a/nodejs/src/cdp/services/hog-executor-async.service.ts b/nodejs/src/cdp/services/hog-executor-async.service.ts
index 8a06dcfcf045..dc2f62a33e74 100644
--- a/nodejs/src/cdp/services/hog-executor-async.service.ts
+++ b/nodejs/src/cdp/services/hog-executor-async.service.ts
@@ -179,6 +179,7 @@ export class HogExecutorAsyncService {
                         // the current action's metadata so a throttle retry re-queues under this
                         // send's class rather than the previous email action's.
                         if (invocation.queue === 'email') {
+                            nextInvocation.queueMetadata = invocation.queueMetadata
                             nextInvocation.queuePriority =
                                 EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(nextInvocation.hogFunction.metadata)]
                         }
diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
index c632c352bf77..35f418c01f40 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
@@ -1,4 +1,5 @@
 // sort-imports-ignore
+import { SESv2Client, TooManyRequestsException } from '@aws-sdk/client-sesv2'
 import { PosthogJwtAudience } from '~/cdp/utils/jwt-utils'
 import { ScopedServiceJwt } from '~/cdp/utils/scoped-service-jwt'
 import { DateTime, Duration } from 'luxon'
@@ -3218,6 +3219,10 @@ describe('Hogflow Executor', () => {
     })
 
     describe('email queue routing', () => {
+        afterEach(() => {
+            jest.restoreAllMocks()
+        })
+
         it('should route email actions to the email queue', async () => {
             await insertIntegration(hub.postgres, team.id, {
                 id: integrationId,
@@ -3310,7 +3315,7 @@ describe('Hogflow Executor', () => {
             expect(result.invocation.queueParameters?.type).toBe('email')
         })
 
-        it('should complete the full round-trip: hogflow → email queue → email sent → workflow continues', async () => {
+        it('returns consecutive emails to their origin queue and priority: hogflow → email queue → emails sent → workflow continues', async () => {
             await insertIntegration(hub.postgres, team.id, {
                 id: integrationId,
                 kind: 'email',
@@ -3319,9 +3324,14 @@ describe('Hogflow Executor', () => {
                     name: 'Test User',
                     domain: 'posthog.com',
                     verified: true,
-                    provider: 'maildev',
+                    provider: 'ses',
                 },
             })
+            const sesSend = jest.spyOn(SESv2Client.prototype, 'send') as jest.SpyInstance
+            sesSend
+                .mockResolvedValueOnce({ MessageId: 'first-email' })
+                .mockRejectedValueOnce(new TooManyRequestsException({ $metadata: {}, message: 'Too many requests' }))
+                .mockResolvedValue({ MessageId: 'follow-up-email' })
 
             await insertHogFunctionTemplate(hub.postgres, {
                 id: 'template-email-routing-test',
@@ -3377,6 +3387,30 @@ describe('Hogflow Executor', () => {
                                 },
                             },
                         },
+                        email_2: {
+                            type: 'function_email',
+                            config: {
+                                template_id: 'template-email-routing-test',
+                                inputs: {
+                                    email: {
+                                        value: {
+                                            to: { email: 'recipient@example.com', name: 'Recipient' },
+                                            from: { integrationId, email: 'test@posthog.com' },
+                                            subject: 'Follow-up Email',
+                                            text: 'Follow-up text',
+                                            html: '<p>Follow-up html</p>',
+                                        },
+                                    },
+                                },
+                            },
+                        },
+                        webhook: {
+                            type: 'function',
+                            config: {
+                                template_id: 'template-test-hogflow-executor',
+                                inputs: { name: { value: 'Mr Debug' } },
+                            },
+                        },
                         exit: {
                             type: 'exit',
                             config: {},
@@ -3384,7 +3418,9 @@ describe('Hogflow Executor', () => {
                     },
                     edges: [
                         { from: 'trigger', to: 'email_1', type: 'continue' },
-                        { from: 'email_1', to: 'exit', type: 'continue' },
+                        { from: 'email_1', to: 'email_2', type: 'continue' },
+                        { from: 'email_2', to: 'webhook', type: 'continue' },
+                        { from: 'webhook', to: 'exit', type: 'continue' },
                     ],
                 })
                 .build()
@@ -3410,19 +3446,25 @@ describe('Hogflow Executor', () => {
             expect(hogflowResult.invocation.queuePriority).toBe(1)
             expect(hogflowResult.invocation.queueMetadata?.originPriority).toBe(2)
 
-            // Step 2: Email worker picks up the job (queue === 'email') — should send inline and continue
+            const results = []
             let emailResult = await executor.execute(hogflowResult.invocation)
-            while (!emailResult.finished) {
+            results.push(emailResult)
+            while (!emailResult.finished && emailResult.invocation.queue === 'email') {
                 emailResult = await executor.execute(emailResult.invocation)
+                results.push(emailResult)
             }
+            expect(emailResult.invocation).toMatchObject({ queue: 'hogflow', queuePriority: 2 })
 
-            // Workflow should complete
-            expect(emailResult.finished).toBe(true)
-            expect(emailResult.error).toBeUndefined()
+            let workflowResult = emailResult
+            while (!workflowResult.finished) {
+                workflowResult = await executor.execute(workflowResult.invocation)
+                results.push(workflowResult)
+            }
+            expect(workflowResult.error).toBeUndefined()
 
-            // Verify email_sent metric was emitted
-            const emailSentMetrics = emailResult.metrics.filter((m) => m.metric_name === 'email_sent')
-            expect(emailSentMetrics).toHaveLength(1)
+            const emailSentMetrics = results.flatMap((r) => r.metrics).filter((m) => m.metric_name === 'email_sent')
+            expect(emailSentMetrics).toHaveLength(2)
+            expect(sesSend).toHaveBeenCalledTimes(3)
         })
     })
 })
diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts
index ff6d8865003e..6d0971ffc806 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts
@@ -685,6 +685,7 @@ export class HogFlowExecutorService {
         // run whose email action isn't the first action executed.
         const result = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {
             queuePriority: invocation.queuePriority,
+            queueMetadata: invocation.queue === 'email' ? invocation.queueMetadata : undefined,
         })
         result.finished = false // Typically we are never finished unless we error or exit
 
diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts
index 390512f5959d..b4271764739a 100644
--- a/nodejs/src/cdp/services/messaging/email.service.test.ts
+++ b/nodejs/src/cdp/services/messaging/email.service.test.ts
@@ -235,6 +235,7 @@ describe('EmailService', () => {
                 stack: [],
             } as any
             invocation.queueParameters = createEmailParams({ from: { integrationId: 1 } })
+            invocation.queueMetadata = { originQueue: 'hogflow', originPriority: 2 }
 
             // Mock SES v2 send to avoid actual AWS calls
             sendEmailSpy = jest.spyOn(service.sesV2Client!, 'send') as any
@@ -468,6 +469,7 @@ describe('EmailService', () => {
                 expect(scheduledMs).toBeGreaterThanOrEqual(before + 400)
                 expect(scheduledMs).toBeLessThan(before + 2000)
                 expect(result.invocation.queueParameters).toEqual(invocation.queueParameters)
+                expect(result.invocation.queueMetadata).toEqual(invocation.queueMetadata)
                 // No business metric emitted on throttle — the eventual retry
                 // will produce email_sent.
                 expect(result.metrics ?? []).toEqual([])
@@ -653,6 +655,7 @@ describe('EmailService', () => {
                 // The reschedule must carry the email payload forward: without queueParameters the
                 // retry has nothing to send and the throttled email is dropped rather than delayed.
                 expect(result.invocation.queueParameters).toEqual(invocation.queueParameters)
+                expect(result.invocation.queueMetadata).toEqual(invocation.queueMetadata)
                 // Bracketed against both ends of the call, so the bounds hold the delay itself
                 // and not the time the call took.
                 const scheduledMs = result.invocation.queueScheduledAt!.toMillis()
@@ -888,6 +891,7 @@ describe('EmailService', () => {
                 expect(result.finished).toBe(false)
                 // The reschedule must carry the email payload forward, same as the workflow limit.
                 expect(result.invocation.queueParameters).toEqual(invocation.queueParameters)
+                expect(result.invocation.queueMetadata).toEqual(invocation.queueMetadata)
                 const scheduledMs = result.invocation.queueScheduledAt!.toMillis()
                 expect(scheduledMs).toBeGreaterThanOrEqual(before + minMs)
                 expect(scheduledMs).toBeLessThan(before + maxMs + 5000)
diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts
index d99cf20fa3a8..921caba47c04 100644
--- a/nodejs/src/cdp/services/messaging/email.service.ts
+++ b/nodejs/src/cdp/services/messaging/email.service.ts
@@ -667,6 +667,7 @@ export class EmailService {
                     // send. Mirrors the fetch-retry (`result.invocation.queueParameters = params`) and
                     // queue-routing paths, which re-attach the same way.
                     result.invocation.queueParameters = params
+                    result.invocation.queueMetadata = invocation.queueMetadata
                     const retryDelayMs = pickReservedRetryDelayMs(claim.retryAfterMs, refillPerSecond, claim.reserved)
                     emailReservedParkMs.labels('workflow-email').observe(retryDelayMs)
                     result.invocation.queueScheduledAt = DateTime.utc().plus({ milliseconds: retryDelayMs })
@@ -693,6 +694,7 @@ export class EmailService {
                 // per-workflow limit above: createInvocationResult cleared queueParameters, and
                 // without them the rescheduled dequeue resumes the Hog VM and drops the send.
                 result.invocation.queueParameters = params
+                result.invocation.queueMetadata = invocation.queueMetadata
                 result.invocation.queueScheduledAt = DateTime.utc().plus({ milliseconds: capDelay.retryDelayMs })
                 const capRetrySeconds = Math.round(capDelay.retryDelayMs / 1000)
                 addLog(
@@ -768,6 +770,7 @@ export class EmailService {
                 throttled = true
                 result.finished = false
                 result.invocation.queueParameters = params
+                result.invocation.queueMetadata = invocation.queueMetadata
                 result.invocation.queueScheduledAt = DateTime.utc().plus({ milliseconds: error.retryAfterMs })
                 addLog('warn', `SES rate-limited (${error.errorCode}); rescheduling email in ${error.retryAfterMs}ms`)
             } else {

From ff2e9aac8e3abb5a20331aa526208363a793817c Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Mon, 12 Oct 2026 04:24:45 +0000
Subject: [PATCH 2/5] fix(workflows): keep email origin when a push retries on
 the email queue

A push notification that fails transiently while its run is on the email
queue now keeps the origin queue and priority that the email routing
stashed. The run returns to its origin queue at its origin priority when
it leaves the email queue. The retry waits at its own message category's
email queue priority instead of its retry count. Each push action still
starts its own retry budget.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../internal/workflows-email-sending-tiers.md |  2 +-
 .../services/hog-executor-async.service.ts    | 13 +++
 .../hogflows/hogflow-executor.service.test.ts | 97 +++++++++++++++++--
 ...e8779a3d2d4c39e28f7d002d66cdbfb8cd8e2.json |  1 +
 4 files changed, 102 insertions(+), 11 deletions(-)
 create mode 100644 nodejs/tests/fixtures/hog-bytecode/1b3f99defea3027176062c76e92e8779a3d2d4c39e28f7d002d66cdbfb8cd8e2.json

diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md
index d2b098c6b8a4..1fe69735e08b 100644
--- a/docs/internal/workflows-email-sending-tiers.md
+++ b/docs/internal/workflows-email-sending-tiers.md
@@ -65,7 +65,7 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
 
 - The batch audience cap is decided when the batch is dispatched. Adding an email step to the workflow while a batch is queued does not re-cap it; the send-time buckets still cap every email at execution. This is why enforcement requires the worker caps to be deployed (see the rollout order).
 - Test-panel sends bypass the team buckets on purpose, matching the per-workflow rate limit.
-- Rescheduled emails keep their email queue priority while waiting. After a team-cap, workflow-pacing, or SES-throttle retry, the workflow restores its origin queue and priority when it next leaves the email queue.
+- Rescheduled emails keep their email queue priority while waiting. After a team-cap, workflow-pacing, or SES-throttle retry, the workflow restores its origin queue and priority when it next leaves the email queue. A push notification retried while the workflow is on the email queue behaves the same way and waits at the priority of its own message category.
 - The buckets are token buckets: a full idle bucket plus refill allows up to roughly twice the stated cap in the very first period. The bucket TTLs exceed the refill periods so this does not recur from idling.
 - A denied send parks until every short bucket has refilled enough to cover it, instead of retrying on a fixed few-minute cadence.
   The computed wait is capped at one hour and then jittered 1x to 2x, so a parked send can wait just under two hours between attempts.
diff --git a/nodejs/src/cdp/services/hog-executor-async.service.ts b/nodejs/src/cdp/services/hog-executor-async.service.ts
index dc2f62a33e74..79a9abd60baa 100644
--- a/nodejs/src/cdp/services/hog-executor-async.service.ts
+++ b/nodejs/src/cdp/services/hog-executor-async.service.ts
@@ -163,6 +163,9 @@ export class HogExecutorAsyncService {
                         nextInvocation,
                         options?.isTest ?? false
                     )
+                    if (invocation.queue === 'email' && !result.finished) {
+                        this.keepPushRetryInEmailQueueOrder(result, invocation)
+                    }
                 } else if (queueParamsType === 'email') {
                     // Route to the email queue unless this is a test run: tests execute in-process and
                     // never enqueue, so routing would leave the job unworked.
@@ -325,6 +328,16 @@ export class HogExecutorAsyncService {
         return result
     }
 
+    private keepPushRetryInEmailQueueOrder(
+        result: CyclotronJobInvocationResult<CyclotronJobInvocationHogFunction>,
+        entryInvocation: CyclotronJobInvocationHogFunction
+    ): void {
+        const { originQueue, originPriority } = entryInvocation.queueMetadata ?? {}
+        result.invocation.queueMetadata = { ...result.invocation.queueMetadata, originQueue, originPriority }
+        result.invocation.queuePriority =
+            EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(result.invocation.hogFunction.metadata)]
+    }
+
     private routeToQueue(
         invocation: CyclotronJobInvocationHogFunction,
         targetQueue: string,
diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
index 35f418c01f40..d41fa48740df 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
@@ -32,6 +32,7 @@ import { EmailSuppressionService, emailSuppressionConfigFromEnv } from '../messa
 import { EmailValidationService } from '../messaging/email-validation.service'
 import { EmailService } from '../messaging/email.service'
 import { EmailTrackingCodeSigner } from '../messaging/helpers/tracking-code'
+import { PushNotificationService } from '../messaging/push-notification.service'
 import { RecipientPreferencesService } from '../messaging/recipient-preferences.service'
 import { RecipientTokensService } from '../messaging/recipient-tokens.service'
 import { HogFlowExecutorService, createHogFlowInvocation } from './hogflow-executor.service'
@@ -62,6 +63,7 @@ describe('Hogflow Executor', () => {
     let team: Team
     let integrationId: number
     const mockFetch = jest.mocked(fetch)
+    const mockPushProviderFetch = jest.fn()
 
     beforeEach(async () => {
         const fixedTime = DateTime.fromObject({ year: 2025, month: 1, day: 1 }, { zone: 'UTC' })
@@ -127,7 +129,18 @@ describe('Hogflow Executor', () => {
                 hogInputsService,
                 emailService,
                 recipientTokensService,
-                pushNotificationService: undefined as any,
+                pushNotificationService: new PushNotificationService(
+                    hub.integrationManager,
+                    hub.encryptedFields,
+                    {
+                        trackedFetch: mockPushProviderFetch,
+                        maxFetchTimeoutMs: 10000,
+                        maxRetries: 3,
+                        backoffBaseMs: 1000,
+                        backoffMaxMs: 30000,
+                    },
+                    undefined as any
+                ),
             }
         )
         const hogFunctionTemplateManager = new HogFunctionTemplateManagerService(hub.postgres)
@@ -3315,7 +3328,7 @@ describe('Hogflow Executor', () => {
             expect(result.invocation.queueParameters?.type).toBe('email')
         })
 
-        it('returns consecutive emails to their origin queue and priority: hogflow → email queue → emails sent → workflow continues', async () => {
+        it('returns consecutive emails and push retries to their origin queue and priority: hogflow → email queue → messages sent → workflow continues', async () => {
             await insertIntegration(hub.postgres, team.id, {
                 id: integrationId,
                 kind: 'email',
@@ -3327,6 +3340,28 @@ describe('Hogflow Executor', () => {
                     provider: 'ses',
                 },
             })
+            const pushIntegrationId = uniqueTestId()
+            await insertIntegration(hub.postgres, team.id, {
+                id: pushIntegrationId,
+                kind: 'firebase',
+                config: { project_id: 'test-project' },
+                sensitive_config: { access_token: 'test-access-token' },
+            })
+            const fcmResponse = (status: number): Record<string, unknown> => ({
+                fetchError: null,
+                fetchResponse: {
+                    status,
+                    headers: {},
+                    text: () => Promise.resolve('{}'),
+                    dump: () => Promise.resolve(),
+                },
+                fetchDuration: 10,
+            })
+            mockPushProviderFetch
+                .mockResolvedValueOnce(fcmResponse(503))
+                .mockResolvedValueOnce(fcmResponse(200))
+                .mockResolvedValueOnce(fcmResponse(503))
+                .mockResolvedValueOnce(fcmResponse(200))
             const sesSend = jest.spyOn(SESv2Client.prototype, 'send') as jest.SpyInstance
             sesSend
                 .mockResolvedValueOnce({ MessageId: 'first-email' })
@@ -3358,6 +3393,24 @@ describe('Hogflow Executor', () => {
                 ],
             })
 
+            await insertHogFunctionTemplate(hub.postgres, {
+                id: 'template-push-queue-routing-test',
+                name: 'Push Routing Test',
+                code: `let res := sendPushNotification({'integrationIds': [inputs.integrationId], 'distinctId': 'distinct-id', 'payload': {'title': 'Hi', 'body': 'Hello'}})
+if (not res.success) { throw Error(res.error) }`,
+                inputs_schema: [
+                    { type: 'string', key: 'integrationId', label: 'Channel', secret: false, required: true },
+                ],
+            })
+            const transactionalPush = {
+                type: 'function_push',
+                config: {
+                    template_id: 'template-push-queue-routing-test',
+                    message_category_type: 'transactional',
+                    inputs: { integrationId: { value: pushIntegrationId } },
+                },
+            } as const
+
             const hogFlow = new FixtureHogFlowBuilder()
                 .withTeamId(team.id)
                 .withExitCondition('exit_only_at_end')
@@ -3387,6 +3440,8 @@ describe('Hogflow Executor', () => {
                                 },
                             },
                         },
+                        push_1: transactionalPush,
+                        push_2: transactionalPush,
                         email_2: {
                             type: 'function_email',
                             config: {
@@ -3418,19 +3473,29 @@ describe('Hogflow Executor', () => {
                     },
                     edges: [
                         { from: 'trigger', to: 'email_1', type: 'continue' },
-                        { from: 'email_1', to: 'email_2', type: 'continue' },
+                        { from: 'email_1', to: 'push_1', type: 'continue' },
+                        { from: 'push_1', to: 'push_2', type: 'continue' },
+                        { from: 'push_2', to: 'email_2', type: 'continue' },
                         { from: 'email_2', to: 'webhook', type: 'continue' },
                         { from: 'webhook', to: 'exit', type: 'continue' },
                     ],
                 })
                 .build()
 
-            const invocation = createExampleHogFlowInvocation(hogFlow, {
-                event: {
-                    ...createHogExecutionGlobals().event,
-                    event: '$pageview',
+            const invocation = createExampleHogFlowInvocation(
+                hogFlow,
+                {
+                    event: {
+                        ...createHogExecutionGlobals().event,
+                        event: '$pageview',
+                    },
                 },
-            })
+                {
+                    properties: {
+                        '$device_push_subscription_test-project': hub.encryptedFields.encrypt('device-token'),
+                    },
+                }
+            )
 
             // A watcher-degraded flow enters at priority 2; the routing must stash that
             // exact value, not the 0 that per-action result clones reset queuePriority to
@@ -3454,6 +3519,14 @@ describe('Hogflow Executor', () => {
                 results.push(emailResult)
             }
             expect(emailResult.invocation).toMatchObject({ queue: 'hogflow', queuePriority: 2 })
+            const rescheduledOnEmailQueue = results
+                .filter((result) => result.invocation.queue === 'email')
+                .map(({ invocation }) => ({ priority: invocation.queuePriority, metadata: invocation.queueMetadata }))
+            expect(rescheduledOnEmailQueue).toEqual([
+                { priority: 0, metadata: { originQueue: 'hogflow', originPriority: 2, tries: 1 } },
+                { priority: 0, metadata: { originQueue: 'hogflow', originPriority: 2, tries: 1 } },
+                { priority: 1, metadata: expect.objectContaining({ originQueue: 'hogflow', originPriority: 2 }) },
+            ])
 
             let workflowResult = emailResult
             while (!workflowResult.finished) {
@@ -3462,9 +3535,13 @@ describe('Hogflow Executor', () => {
             }
             expect(workflowResult.error).toBeUndefined()
 
-            const emailSentMetrics = results.flatMap((r) => r.metrics).filter((m) => m.metric_name === 'email_sent')
-            expect(emailSentMetrics).toHaveLength(2)
+            const sentMetrics = results
+                .flatMap((r) => r.metrics)
+                .map((m) => m.metric_name)
+                .filter((name) => name === 'email_sent' || name === 'push_sent')
+            expect(sentMetrics).toEqual(['email_sent', 'push_sent', 'push_sent', 'email_sent'])
             expect(sesSend).toHaveBeenCalledTimes(3)
+            expect(mockPushProviderFetch).toHaveBeenCalledTimes(4)
         })
     })
 })
diff --git a/nodejs/tests/fixtures/hog-bytecode/1b3f99defea3027176062c76e92e8779a3d2d4c39e28f7d002d66cdbfb8cd8e2.json b/nodejs/tests/fixtures/hog-bytecode/1b3f99defea3027176062c76e92e8779a3d2d4c39e28f7d002d66cdbfb8cd8e2.json
new file mode 100644
index 000000000000..e21a84fee743
--- /dev/null
+++ b/nodejs/tests/fixtures/hog-bytecode/1b3f99defea3027176062c76e92e8779a3d2d4c39e28f7d002d66cdbfb8cd8e2.json
@@ -0,0 +1 @@
+["_H",1,32,"integrationIds",32,"integrationId",32,"inputs",1,2,43,1,32,"distinctId",32,"distinct-id",32,"payload",32,"title",32,"Hi",32,"body",32,"Hello",42,2,42,3,2,"sendPushNotification",1,36,0,32,"success",45,5,40,9,36,0,32,"error",45,2,"Error",1,49,35]

From 29d303fe1a72cd79a00b088e978be5d38f15becd Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Mon, 12 Oct 2026 04:59:06 +0000
Subject: [PATCH 3/5] chore(workflows): tidy email queue priority lookup and
 push test mock

Extract getEmailQueuePriority so the inline email send and the push
retry share one class-to-priority lookup. Type the push provider fetch
mock and reset it before each test, so a CI retry of a failed attempt
does not start with leftover queued responses.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/cdp/services/hog-executor-async.service.ts  | 13 ++++++++-----
 .../hogflows/hogflow-executor.service.test.ts       | 10 +++++++---
 nodejs/src/cdp/services/messaging/email-priority.ts |  4 ++++
 3 files changed, 19 insertions(+), 8 deletions(-)

diff --git a/nodejs/src/cdp/services/hog-executor-async.service.ts b/nodejs/src/cdp/services/hog-executor-async.service.ts
index 79a9abd60baa..d4286dab465b 100644
--- a/nodejs/src/cdp/services/hog-executor-async.service.ts
+++ b/nodejs/src/cdp/services/hog-executor-async.service.ts
@@ -30,7 +30,12 @@ import { mergeSecretHeaders, resolveSecretHeaders } from '../utils/secret-header
 import { resolveStandardWebhooksKey, signStandardWebhooksRequest } from '../utils/standard-webhooks'
 import { HogExecutorExecuteOptions, HogExecutorPreviousResult, HogExecutorService } from './hog-executor.service'
 import { HogInputsService } from './hog-inputs.service'
-import { EMAIL_QUEUE_PRIORITY, cdpEmailQueuedTotal, getEmailQueuePriorityClass } from './messaging/email-priority'
+import {
+    EMAIL_QUEUE_PRIORITY,
+    cdpEmailQueuedTotal,
+    getEmailQueuePriority,
+    getEmailQueuePriorityClass,
+} from './messaging/email-priority'
 import { EmailService } from './messaging/email.service'
 import { PushNotificationService } from './messaging/push-notification.service'
 import { RecipientTokensService, preferencesTokenSourceForInvocation } from './messaging/recipient-tokens.service'
@@ -183,8 +188,7 @@ export class HogExecutorAsyncService {
                         // send's class rather than the previous email action's.
                         if (invocation.queue === 'email') {
                             nextInvocation.queueMetadata = invocation.queueMetadata
-                            nextInvocation.queuePriority =
-                                EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(nextInvocation.hogFunction.metadata)]
+                            nextInvocation.queuePriority = getEmailQueuePriority(nextInvocation.hogFunction.metadata)
                         }
                         // isTest is forwarded so a test send stays out of the email's engagement tracking.
                         result = await this.deps.emailService.executeSendEmail(nextInvocation, options?.isTest ?? false)
@@ -334,8 +338,7 @@ export class HogExecutorAsyncService {
     ): void {
         const { originQueue, originPriority } = entryInvocation.queueMetadata ?? {}
         result.invocation.queueMetadata = { ...result.invocation.queueMetadata, originQueue, originPriority }
-        result.invocation.queuePriority =
-            EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(result.invocation.hogFunction.metadata)]
+        result.invocation.queuePriority = getEmailQueuePriority(result.invocation.hogFunction.metadata)
     }
 
     private routeToQueue(
diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
index d41fa48740df..0716a38ba90f 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
@@ -32,7 +32,7 @@ import { EmailSuppressionService, emailSuppressionConfigFromEnv } from '../messa
 import { EmailValidationService } from '../messaging/email-validation.service'
 import { EmailService } from '../messaging/email.service'
 import { EmailTrackingCodeSigner } from '../messaging/helpers/tracking-code'
-import { PushNotificationService } from '../messaging/push-notification.service'
+import { PushNotificationFetchUtils, PushNotificationService } from '../messaging/push-notification.service'
 import { RecipientPreferencesService } from '../messaging/recipient-preferences.service'
 import { RecipientTokensService } from '../messaging/recipient-tokens.service'
 import { HogFlowExecutorService, createHogFlowInvocation } from './hogflow-executor.service'
@@ -50,6 +50,8 @@ jest.mock('~/common/utils/request', () => {
     }
 })
 
+type TrackedPushFetch = PushNotificationFetchUtils['trackedFetch']
+
 const cleanLogs = (logs: string[]): string[] => {
     // Replaces the function time with a fixed value to simplify testing
     return logs.map((log) => log.replace(/Function completed in \d+(\.\d+)?ms/, 'Function completed in REPLACEDms'))
@@ -63,9 +65,10 @@ describe('Hogflow Executor', () => {
     let team: Team
     let integrationId: number
     const mockFetch = jest.mocked(fetch)
-    const mockPushProviderFetch = jest.fn()
+    const mockPushProviderFetch = jest.fn<ReturnType<TrackedPushFetch>, Parameters<TrackedPushFetch>>()
 
     beforeEach(async () => {
+        mockPushProviderFetch.mockReset()
         const fixedTime = DateTime.fromObject({ year: 2025, month: 1, day: 1 }, { zone: 'UTC' })
         jest.spyOn(Date, 'now').mockReturnValue(fixedTime.toMillis())
 
@@ -3347,11 +3350,12 @@ describe('Hogflow Executor', () => {
                 config: { project_id: 'test-project' },
                 sensitive_config: { access_token: 'test-access-token' },
             })
-            const fcmResponse = (status: number): Record<string, unknown> => ({
+            const fcmResponse = (status: number): Awaited<ReturnType<TrackedPushFetch>> => ({
                 fetchError: null,
                 fetchResponse: {
                     status,
                     headers: {},
+                    json: () => Promise.resolve({}),
                     text: () => Promise.resolve('{}'),
                     dump: () => Promise.resolve(),
                 },
diff --git a/nodejs/src/cdp/services/messaging/email-priority.ts b/nodejs/src/cdp/services/messaging/email-priority.ts
index b4f3353e94a7..6854218eb47b 100644
--- a/nodejs/src/cdp/services/messaging/email-priority.ts
+++ b/nodejs/src/cdp/services/messaging/email-priority.ts
@@ -39,6 +39,10 @@ export function getEmailQueuePriorityClass(metadata: HogFunctionType['metadata']
     return metadata?.message_category_type === 'transactional' ? 'fast' : 'bulk'
 }
 
+export function getEmailQueuePriority(metadata: HogFunctionType['metadata']): number {
+    return EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(metadata)]
+}
+
 export const cdpEmailQueuedTotal = new Counter({
     name: 'cdp_email_queued_total',
     help: 'Total emails routed to the dedicated email queue',

From fda06d518e45d6f4b878c12b131f5da543204061 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Mon, 2 Nov 2026 06:30:40 +0000
Subject: [PATCH 4/5] chore(workflows): pass the push service's new
 dependencies in the executor test

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/cdp/services/hogflows/hogflow-executor.service.test.ts   | 1 +
 1 file changed, 1 insertion(+)

diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
index 0716a38ba90f..7e59074abdfd 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
@@ -142,6 +142,7 @@ describe('Hogflow Executor', () => {
                         backoffBaseMs: 1000,
                         backoffMaxMs: 30000,
                     },
+                    undefined as any,
                     undefined as any
                 ),
             }

From 0d1bb84901a9bf3b6f444cfc1bbfa39a719596cc Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Mon, 2 Nov 2026 13:01:27 +0000
Subject: [PATCH 5/5] fix(workflows): keep the team cap label on the delayed
 email only

The executor now carries a run's queue metadata between actions on the
email queue, so the teamCapped flag from one capped send reached the
next email in the same dequeue and labeled its queue wait as a cap delay.
Drop the flag from the executor's result; the capped send still reads it
from the job's stored metadata.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../hogflows/hogflow-executor.service.test.ts | 159 ++++++++----------
 .../hogflows/hogflow-executor.service.ts      |  15 +-
 2 files changed, 83 insertions(+), 91 deletions(-)

diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
index 7e59074abdfd..c4a36b9c62de 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts
@@ -3240,7 +3240,7 @@ describe('Hogflow Executor', () => {
             jest.restoreAllMocks()
         })
 
-        it('should route email actions to the email queue', async () => {
+        const insertEmailSetup = async (provider: 'maildev' | 'ses'): Promise<void> => {
             await insertIntegration(hub.postgres, team.id, {
                 id: integrationId,
                 kind: 'email',
@@ -3249,10 +3249,9 @@ describe('Hogflow Executor', () => {
                     name: 'Test User',
                     domain: 'posthog.com',
                     verified: true,
-                    provider: 'maildev',
+                    provider,
                 },
             })
-
             await insertHogFunctionTemplate(hub.postgres, {
                 id: 'template-email-routing-test',
                 name: 'Email Routing Test',
@@ -3277,6 +3276,29 @@ describe('Hogflow Executor', () => {
                     },
                 ],
             })
+        }
+
+        const emailAction = (subject: string) =>
+            ({
+                type: 'function_email',
+                config: {
+                    template_id: 'template-email-routing-test',
+                    inputs: {
+                        email: {
+                            value: {
+                                to: { email: 'recipient@example.com', name: 'Recipient' },
+                                from: { integrationId, email: 'test@posthog.com' },
+                                subject,
+                                text: `${subject} text`,
+                                html: `<p>${subject}</p>`,
+                            },
+                        },
+                    },
+                },
+            }) as const
+
+        it('should route email actions to the email queue', async () => {
+            await insertEmailSetup('maildev')
 
             const hogFlow = new FixtureHogFlowBuilder()
                 .withTeamId(team.id)
@@ -3290,23 +3312,7 @@ describe('Hogflow Executor', () => {
                                 filters: HOG_FILTERS_EXAMPLES.no_filters.filters ?? {},
                             },
                         },
-                        email_1: {
-                            type: 'function_email',
-                            config: {
-                                template_id: 'template-email-routing-test',
-                                inputs: {
-                                    email: {
-                                        value: {
-                                            to: { email: 'recipient@example.com', name: 'Recipient' },
-                                            from: { integrationId, email: 'test@posthog.com' },
-                                            subject: 'Test Email',
-                                            text: 'Test',
-                                            html: '<p>Test</p>',
-                                        },
-                                    },
-                                },
-                            },
-                        },
+                        email_1: emailAction('Test Email'),
                     },
                     edges: [
                         { from: 'trigger', to: 'email_1', type: 'continue' },
@@ -3333,17 +3339,7 @@ describe('Hogflow Executor', () => {
         })
 
         it('returns consecutive emails and push retries to their origin queue and priority: hogflow → email queue → messages sent → workflow continues', async () => {
-            await insertIntegration(hub.postgres, team.id, {
-                id: integrationId,
-                kind: 'email',
-                config: {
-                    email: 'test@posthog.com',
-                    name: 'Test User',
-                    domain: 'posthog.com',
-                    verified: true,
-                    provider: 'ses',
-                },
-            })
+            await insertEmailSetup('ses')
             const pushIntegrationId = uniqueTestId()
             await insertIntegration(hub.postgres, team.id, {
                 id: pushIntegrationId,
@@ -3373,31 +3369,6 @@ describe('Hogflow Executor', () => {
                 .mockRejectedValueOnce(new TooManyRequestsException({ $metadata: {}, message: 'Too many requests' }))
                 .mockResolvedValue({ MessageId: 'follow-up-email' })
 
-            await insertHogFunctionTemplate(hub.postgres, {
-                id: 'template-email-routing-test',
-                name: 'Email Routing Test',
-                code: `sendEmail(inputs.email)`,
-                inputs_schema: [
-                    {
-                        type: 'native_email',
-                        key: 'email',
-                        label: 'Email message',
-                        integration: 'email',
-                        required: true,
-                        default: {
-                            to: { email: '', name: '' },
-                            from: { email: '', name: '' },
-                            subject: '',
-                            text: 'Hello!',
-                            html: '<div>Hello!</div>',
-                        },
-                        secret: false,
-                        description: '',
-                        templating: 'liquid',
-                    },
-                ],
-            })
-
             await insertHogFunctionTemplate(hub.postgres, {
                 id: 'template-push-queue-routing-test',
                 name: 'Push Routing Test',
@@ -3428,42 +3399,10 @@ if (not res.success) { throw Error(res.error) }`,
                                 filters: HOG_FILTERS_EXAMPLES.no_filters.filters ?? {},
                             },
                         },
-                        email_1: {
-                            type: 'function_email',
-                            config: {
-                                template_id: 'template-email-routing-test',
-                                inputs: {
-                                    email: {
-                                        value: {
-                                            to: { email: 'recipient@example.com', name: 'Recipient' },
-                                            from: { integrationId, email: 'test@posthog.com' },
-                                            subject: 'Test Email',
-                                            text: 'Test text',
-                                            html: '<p>Test html</p>',
-                                        },
-                                    },
-                                },
-                            },
-                        },
+                        email_1: emailAction('Test Email'),
                         push_1: transactionalPush,
                         push_2: transactionalPush,
-                        email_2: {
-                            type: 'function_email',
-                            config: {
-                                template_id: 'template-email-routing-test',
-                                inputs: {
-                                    email: {
-                                        value: {
-                                            to: { email: 'recipient@example.com', name: 'Recipient' },
-                                            from: { integrationId, email: 'test@posthog.com' },
-                                            subject: 'Follow-up Email',
-                                            text: 'Follow-up text',
-                                            html: '<p>Follow-up html</p>',
-                                        },
-                                    },
-                                },
-                            },
-                        },
+                        email_2: emailAction('Follow-up Email'),
                         webhook: {
                             type: 'function',
                             config: {
@@ -3548,5 +3487,45 @@ if (not res.success) { throw Error(res.error) }`,
             expect(sesSend).toHaveBeenCalledTimes(3)
             expect(mockPushProviderFetch).toHaveBeenCalledTimes(4)
         })
+
+        it('keeps the team cap label on the delayed email only, not on the email after it', async () => {
+            await insertEmailSetup('ses')
+            const sesSend = jest.spyOn(SESv2Client.prototype, 'send') as jest.SpyInstance
+            sesSend
+                .mockResolvedValueOnce({ MessageId: 'capped-email' })
+                .mockRejectedValueOnce(new TooManyRequestsException({ $metadata: {}, message: 'Too many requests' }))
+
+            const hogFlow = new FixtureHogFlowBuilder()
+                .withTeamId(team.id)
+                .withExitCondition('exit_only_at_end')
+                .withWorkflow({
+                    actions: {
+                        trigger: {
+                            type: 'trigger',
+                            config: { type: 'event', filters: HOG_FILTERS_EXAMPLES.no_filters.filters ?? {} },
+                        },
+                        email_1: emailAction('Capped Email'),
+                        email_2: emailAction('Next Email'),
+                    },
+                    edges: [
+                        { from: 'trigger', to: 'email_1', type: 'continue' },
+                        { from: 'email_1', to: 'email_2', type: 'continue' },
+                        { from: 'email_2', to: 'exit', type: 'continue' },
+                    ],
+                })
+                .build()
+            const invocation = createExampleHogFlowInvocation(hogFlow, {
+                event: { ...createHogExecutionGlobals().event, event: '$pageview' },
+            })
+
+            const routed = await executor.execute(invocation)
+            // As if the team sending cap had delayed email_1 on the email queue.
+            routed.invocation.queueMetadata = { ...routed.invocation.queueMetadata, teamCapped: true }
+
+            const result = await executor.execute(routed.invocation)
+
+            expect(result.invocation.state.currentAction?.id).toBe('email_2')
+            expect(result.invocation.queueMetadata).toEqual({ originQueue: 'hogflow', originPriority: 0 })
+        })
     })
 })
diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts
index d0cfd5242f32..2a6bfa998cfc 100644
--- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts
+++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts
@@ -76,6 +76,19 @@ const TRIGGER_MISMATCH_EXITS: ReadonlySet<string> = new Set([
     'exit_on_trigger_not_matched_or_conversion',
 ])
 
+// The email service sets teamCapped when the team sending cap delays one send, and that send reads it from the
+// job's stored metadata. The actions after it in the same dequeue start from this executor's result, so the
+// flag stays off them. Otherwise their queue waits would also count as cap delays.
+function withoutTeamCapped(
+    metadata: CyclotronJobInvocationHogFlow['queueMetadata']
+): CyclotronJobInvocationHogFlow['queueMetadata'] {
+    if (!metadata || !('teamCapped' in metadata)) {
+        return metadata
+    }
+    const { teamCapped: _teamCapped, ...rest } = metadata
+    return rest
+}
+
 export function createHogFlowInvocation(
     globals: HogFunctionInvocationGlobals,
     hogFlow: HogFlow,
@@ -686,7 +699,7 @@ export class HogFlowExecutorService {
         // run whose email action isn't the first action executed.
         const result = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {
             queuePriority: invocation.queuePriority,
-            queueMetadata: invocation.queue === 'email' ? invocation.queueMetadata : undefined,
+            queueMetadata: invocation.queue === 'email' ? withoutTeamCapped(invocation.queueMetadata) : undefined,
         })
         result.finished = false // Typically we are never finished unless we error or exit