fix(workflows): keep push test sends out of metrics and assets

nodejs/src/cdp/cdp-api.test.ts

@@ -1617,7 +1617,7 @@ describe('CDP API', () => {
                         ],
                         capturedPostHogEvents: [],
                         warehouseWebhookPayloads: [],
-                        emailAssets: [],
+                        messageAssets: [],
                     })
                 )
         })

nodejs/src/cdp/services/hog-executor.service.test.ts

@@ -141,7 +141,7 @@ describe('Hog Executor', () => {
             expect(result).toEqual({
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
                 invocation: {
                     state: {
                         globals: invocation.state.globals,

nodejs/src/cdp/services/hog-executor.service.ts

@@ -488,7 +488,12 @@ export class HogExecutorService {
                         result = await this.executeFetch(nextInvocation, options)
                     }
                 } else if (queueParamsType === 'sendPushNotification') {
-                    result = await this.pushNotificationService.executeSendPushNotification(nextInvocation)
+                    // Same signal the email branch uses below: `sendEmailsInline` is only set by the
+                    // test panel, so it doubles as "this is a test send".
+                    result = await this.pushNotificationService.executeSendPushNotification(
+                        nextInvocation,
+                        options?.sendEmailsInline ?? false
+                    )
                 } else if (queueParamsType === 'email') {
                     // Route to the email queue only if we're not already there and the
                     // caller hasn't asked for inline-only execution (e.g. the test panel).

nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts

@@ -505,7 +505,7 @@ describe('HogFunctionHandler', () => {
             metrics: [],
             capturedPostHogEvents: [],
             warehouseWebhookPayloads: [],
-            emailAssets: [],
+            messageAssets: [],
         })
 
         const invocationResult = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {

nodejs/src/cdp/services/hogflows/actions/hog_function.ts

@@ -61,7 +61,7 @@ export class HogFunctionHandler implements ActionHandler {
             ...functionResult.warehouseWebhookPayloads,
         ]
         result.metrics = [...result.metrics, ...functionResult.metrics]
-        result.emailAssets = [...result.emailAssets, ...functionResult.emailAssets]
+        result.messageAssets = [...result.messageAssets, ...functionResult.messageAssets]
 
         if (!functionResult.finished) {
             // Set the state of the function result on the substate of the flow for the next execution
@@ -158,7 +158,7 @@ export class HogFunctionHandler implements ActionHandler {
                 metrics,
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
             }
         }
 
@@ -187,7 +187,7 @@ export class HogFunctionHandler implements ActionHandler {
                 ],
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
             }
         }
 

nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts

@@ -280,7 +280,7 @@ describe('Hogflow Executor', () => {
             expect(result).toEqual({
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
                 invocation: {
                     state: {
                         actionStepCount: 1,

nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts

@@ -214,7 +214,7 @@ export class HogFlowExecutorService {
         const logs: MinimalLogEntry[] = []
         const capturedPostHogEvents: HogFunctionCapturedEvent[] = []
         const warehouseWebhookPayloads: WarehouseWebhookPayload[] = []
-        const emailAssets: MessageAssetRow[] = []
+        const messageAssets: MessageAssetRow[] = []
 
         const earlyExitResult = await this.shouldExitEarly(invocation, metrics, capturedPostHogEvents)
         if (earlyExitResult) {
@@ -252,7 +252,7 @@ export class HogFlowExecutorService {
             metrics.push(...result.metrics)
             capturedPostHogEvents.push(...result.capturedPostHogEvents)
             warehouseWebhookPayloads.push(...result.warehouseWebhookPayloads)
-            emailAssets.push(...result.emailAssets)
+            messageAssets.push(...result.messageAssets)
 
             if (this.shouldEndHogFlowExecution(result, logs)) {
                 break
@@ -263,7 +263,7 @@ export class HogFlowExecutorService {
         result.metrics = metrics
         result.capturedPostHogEvents = capturedPostHogEvents
         result.warehouseWebhookPayloads = warehouseWebhookPayloads
-        result.emailAssets = emailAssets
+        result.messageAssets = messageAssets
 
         return result
     }

nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.test.ts

@@ -52,7 +52,7 @@ describe('CyclotronJobQueuePostgresV2', () => {
             metrics: [],
             capturedPostHogEvents: [],
             warehouseWebhookPayloads: [],
-            emailAssets: [],
+            messageAssets: [],
             ...overrides,
         }
     }

nodejs/src/cdp/services/job-queue/shared.test.ts

@@ -266,7 +266,7 @@ describe('createInvocationSanitizer', () => {
             metrics: [],
             capturedPostHogEvents: [],
             warehouseWebhookPayloads: [],
-            emailAssets: [],
+            messageAssets: [],
         }
 
         const [sanitized] = sanitizer.sanitizeResults([result])

nodejs/src/cdp/services/messaging/email.service.ts

@@ -300,7 +300,7 @@ export class EmailService {
             }
 
             if (success && assetRow) {
-                result.emailAssets.push(assetRow)
+                result.messageAssets.push(assetRow)
             }
         }
 

nodejs/src/cdp/services/messaging/message-assets.service.test.ts

@@ -47,7 +47,7 @@ const invocationWithAction = (id: string, teamId = 7): CyclotronJobInvocationHog
 
 const resultWith = (assets: MessageAssetRow[]): CyclotronJobInvocationResult =>
     ({
-        emailAssets: assets,
+        messageAssets: assets,
     }) as unknown as CyclotronJobInvocationResult
 
 const producedRowAt = (outputs: jest.Mocked<IngestionOutputs<'message_assets'>>, index: number): MessageAssetRow => {
@@ -245,7 +245,7 @@ describe('MessageAssetsService', () => {
             expect(outputs.produce).toHaveBeenCalledTimes(1)
         })
 
-        it('skips results that carry an empty emailAssets array', async () => {
+        it('skips results that carry an empty messageAssets array', async () => {
             service.queueInvocationResults([resultWith([]), resultWith([])])
             await service.flush()
             expect(outputs.produce).not.toHaveBeenCalled()

nodejs/src/cdp/services/messaging/message-assets.service.ts

@@ -210,10 +210,10 @@ export class MessageAssetsService {
 
     queueInvocationResults(results: CyclotronJobInvocationResult[]): void {
         for (const result of results) {
-            if (!result.emailAssets || result.emailAssets.length === 0) {
+            if (!result.messageAssets || result.messageAssets.length === 0) {
                 continue
             }
-            for (const row of result.emailAssets) {
+            for (const row of result.messageAssets) {
                 this.queuedRows.push(row)
             }
         }

nodejs/src/cdp/services/messaging/push-notification.service.test.ts

@@ -243,11 +243,11 @@ describe('PushNotificationService', () => {
                 if (captured === 'nothing') {
                     // An asset is a snapshot of what a recipient received; a send that reached nobody
                     // has none, and email behaves the same way (it captures only on success).
-                    expect(result.emailAssets).toEqual([])
+                    expect(result.messageAssets).toEqual([])
                     return
                 }
-                expect(result.emailAssets).toHaveLength(1)
-                expect(result.emailAssets[0]).toMatchObject({
+                expect(result.messageAssets).toHaveLength(1)
+                expect(result.messageAssets[0]).toMatchObject({
                     kind: 'push',
                     status: captured,
                     action_id: 'action_push_1',
@@ -279,12 +279,29 @@ describe('PushNotificationService', () => {
 
                 expect(result.error).toBeUndefined()
                 expect(result.metrics).toContainEqual(expect.objectContaining({ metric_name: 'push_sent' }))
-                expect(result.emailAssets).toEqual([])
+                expect(result.messageAssets).toEqual([])
                 expect(result.logs.map((log) => log.message)).toContainEqual(
                     expect.stringContaining('could not be captured')
                 )
             })
 
+            it('records neither a metric nor an asset for a test send', async () => {
+                // "Run test" really delivers, but it must not show up as a workflow send: email
+                // already skips both, and a test row on the Assets tab reads as a real delivery.
+                respondWith(200)
+                const invocation = createSendPushNotificationInvocation({
+                    '$device_push_subscription_test-project': encryptedFields.encrypt('device-token-123'),
+                })
+                invocation.state.actionId = 'action_push_1'
+
+                const result = await serviceWithAssets.executeSendPushNotification(invocation, true)
+
+                expect(result.error).toBeUndefined()
+                expect(result.metrics).toEqual([])
+                expect(result.messageAssets).toEqual([])
+                expect(result.logs.map((log) => log.message)).toContainEqual(expect.stringContaining('accepted by FCM'))
+            })
+
             it('captures one asset per notification, not one per delivered channel', async () => {
                 respondWith(200)
                 integrationManager.get = jest
@@ -298,7 +315,7 @@ describe('PushNotificationService', () => {
 
                 const result = await serviceWithAssets.executeSendPushNotification(invocation)
 
-                expect(result.emailAssets).toHaveLength(1)
+                expect(result.messageAssets).toHaveLength(1)
             })
         })
 

nodejs/src/cdp/services/messaging/push-notification.service.ts

@@ -136,7 +136,8 @@ export class PushNotificationService {
 
     @instrumented('push-notification.executeSendPushNotification')
     async executeSendPushNotification(
-        invocation: CyclotronJobInvocationHogFunction
+        invocation: CyclotronJobInvocationHogFunction,
+        isTest = false
     ): Promise<CyclotronJobInvocationResult<CyclotronJobInvocationHogFunction>> {
         if (invocation.queueParameters?.type !== 'sendPushNotification') {
             throw new Error('Bad invocation')
@@ -154,7 +155,9 @@ export class PushNotificationService {
         // Business metrics are emitted once, at the terminal outcome below, rather than per channel — a
         // rescheduled attempt must not re-count the same notification's skips or failures on every retry.
         const pushMetric = (metricName: 'push_sent' | 'push_skipped' | 'push_failed', count: number): void => {
-            if (count <= 0) {
+            // A test send from the editor's "Run test" must not land in the workflow's Metrics tab,
+            // matching what the email path already does.
+            if (count <= 0 || isTest) {
                 return
             }
             result.metrics.push({
@@ -278,15 +281,17 @@ export class PushNotificationService {
         // Only a delivered notification is captured, matching email: an asset is a snapshot of what a
         // recipient received, and a skip has no recipient. Skips stay visible as `push_skipped` plus the
         // per-channel run log explaining why.
-        if (this.messageAssetsService && successCount > 0) {
+        // Skipped for a test send for the same reason the metrics are: the Assets tab should show
+        // what real recipients were sent, not what an editor preview produced.
+        if (this.messageAssetsService && successCount > 0 && !isTest) {
             // Best-effort: the notification is already delivered by this point, so a capture failure
             // must not fail the invocation. Throwing here would send the whole batch back for a retry
             // and deliver every notification in it a second time. Losing an Assets row is the cheaper
             // outcome, and it is the same trade the flush path already makes on a Kafka failure.
             try {
                 const assetRow = this.messageAssetsService.buildRowForPush(invocation, params, [...deliveredPlatforms])
                 if (assetRow) {
-                    result.emailAssets.push(assetRow)
+                    result.messageAssets.push(assetRow)
                 }
             } catch (err) {
                 addLog('warn', 'The notification was delivered but could not be captured for the Assets tab.')

nodejs/src/cdp/services/monitoring/hog-invocation-results.service.test.ts

@@ -48,7 +48,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
 
@@ -230,7 +230,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
             await service.flush()
@@ -252,7 +252,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
             await service.flush()
@@ -276,7 +276,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
             await service.flush()

nodejs/src/cdp/types.ts

@@ -304,7 +304,7 @@ export type CyclotronJobInvocationResult<T extends CyclotronJobInvocation = Cycl
     metrics: MinimalAppMetric[]
     capturedPostHogEvents: HogFunctionCapturedEvent[]
     warehouseWebhookPayloads: WarehouseWebhookPayload[]
-    emailAssets: MessageAssetRow[]
+    messageAssets: MessageAssetRow[]
     execResult?: unknown
 }
 

nodejs/src/cdp/utils/invocation-utils.ts

@@ -70,7 +70,7 @@ export function createInvocationResult<T extends CyclotronJobInvocation>(
         | 'finished'
         | 'capturedPostHogEvents'
         | 'warehouseWebhookPayloads'
-        | 'emailAssets'
+        | 'messageAssets'
         | 'logs'
         | 'metrics'
         | 'error'
@@ -82,7 +82,7 @@ export function createInvocationResult<T extends CyclotronJobInvocation>(
         finished: true,
         capturedPostHogEvents: [],
         warehouseWebhookPayloads: [],
-        emailAssets: [],
+        messageAssets: [],
         logs: [],
         metrics: [],
         ...resultParams,

nodejs/src/cdp/workflows-e2e.test.ts

@@ -3385,7 +3385,7 @@ describe('Workflows E2E (email queue)', () => {
     //
     // Email assets used to be produced one-at-a-time via a fire-and-forget Kafka call
     // from `email.service.ts → MessageAssetsService.captureSentEmail`. We've moved that
-    // to a buffer-then-flush pattern that drains `result.emailAssets` at the batch
+    // to a buffer-then-flush pattern that drains `result.messageAssets` at the batch
     // boundary and bulk-produces, gated on broker ack before the consumer commits
     // offsets. These tests pin the end-to-end behavior: one workflow → one asset row in
     // the `message_assets` Kafka topic with the right metadata, and a single batch with