fix(workflow, custom): billing attribution passthrough (#5657)

Author: icecrasher321

apps/sim/executor/execution/executor.ts

@@ -427,6 +427,9 @@ export class DAGExecutor {
       blockLogs: overrides?.runFromBlockContext ? [] : (snapshotState?.blockLogs ?? []),
       metadata: {
         ...this.contextExtensions.metadata,
+        ...(this.contextExtensions.billingAttribution
+          ? { billingAttribution: this.contextExtensions.billingAttribution }
+          : {}),
         startTime: new Date().toISOString(),
         duration: 0,
         useDraftState:

apps/sim/executor/execution/types.ts

@@ -168,6 +168,13 @@ export interface ContextExtensions {
   fileKeys?: string[]
   allowLargeValueWorkflowScope?: boolean
   userId?: string
+  /**
+   * Immutable actor/payer decision for this execution. Child workflow
+   * executions receive it here (they carry no full metadata), so internal
+   * tool calls inside the child still attach the billing attribution header.
+   * Takes precedence over `metadata.billingAttribution` when both are set.
+   */
+  billingAttribution?: BillingAttributionSnapshot
   stream?: boolean
   selectedOutputs?: string[]
   edges?: Array<{ source: string; target: string }>

apps/sim/executor/handlers/workflow/workflow-handler.test.ts

@@ -9,17 +9,27 @@ import {
 import type { ExecutionContext } from '@/executor/types'
 import type { SerializedBlock } from '@/serializer/types'
 
-const { mockExecutorExecute, mockCreateSnapshot } = vi.hoisted(() => ({
-  mockExecutorExecute: vi.fn(),
-  mockCreateSnapshot: vi.fn(),
-}))
+const { mockExecutorExecute, mockCreateSnapshot, mockResolveBillingAttribution, executorOptions } =
+  vi.hoisted(() => ({
+    mockExecutorExecute: vi.fn(),
+    mockCreateSnapshot: vi.fn(),
+    mockResolveBillingAttribution: vi.fn(),
+    executorOptions: [] as Array<Record<string, any>>,
+  }))
 
 vi.mock('@/executor', () => ({
   Executor: class {
+    constructor(options: Record<string, any>) {
+      executorOptions.push(options)
+    }
     execute = mockExecutorExecute
   },
 }))
 
+vi.mock('@/lib/billing/core/billing-attribution', () => ({
+  resolveBillingAttribution: mockResolveBillingAttribution,
+}))
+
 vi.mock('@/lib/logs/execution/snapshot/service', () => ({
   snapshotService: { createSnapshotWithDeduplication: mockCreateSnapshot },
 }))
@@ -92,6 +102,7 @@ describe('WorkflowBlockHandler', () => {
 
     // Reset all mocks
     vi.clearAllMocks()
+    executorOptions.length = 0
 
     // Setup default fetch mock
     mockFetch.mockResolvedValue({
@@ -273,6 +284,43 @@ describe('WorkflowBlockHandler', () => {
       expect(mockExecutorExecute).toHaveBeenCalledWith('child-workflow-id')
     })
 
+    it('threads the parent billing attribution into the child execution context', async () => {
+      const billingAttribution = {
+        actorUserId: 'actor-1',
+        workspaceId: 'workspace-parent',
+        organizationId: 'org-1',
+        billedAccountUserId: 'owner-1',
+        billingEntity: { type: 'organization', id: 'org-1' },
+        billingPeriod: { start: '2026-07-01T00:00:00.000Z', end: '2026-08-01T00:00:00.000Z' },
+        payerSubscription: null,
+      }
+      const ctx = {
+        ...mockContext,
+        workspaceId: 'workspace-parent',
+        metadata: { ...mockContext.metadata, billingAttribution },
+      } as ExecutionContext
+
+      mockFetch.mockResolvedValueOnce({
+        ok: true,
+        json: () =>
+          Promise.resolve({
+            data: {
+              name: 'Child Workflow',
+              workspaceId: 'workspace-parent',
+              state: { blocks: {}, edges: [], loops: {}, parallels: {} },
+            },
+          }),
+      })
+      mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
+      mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
+
+      await handler.execute(ctx, mockBlock, inputs)
+
+      expect(executorOptions).toHaveLength(1)
+      expect(executorOptions[0].contextExtensions.billingAttribution).toBe(billingAttribution)
+      expect(mockResolveBillingAttribution).not.toHaveBeenCalled()
+    })
+
     it('should fail closed when the executing context has no workspace', async () => {
       mockFetch.mockResolvedValueOnce({
         ok: true,

apps/sim/executor/handlers/workflow/workflow-handler.ts

@@ -1,6 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
+import { resolveBillingAttribution } from '@/lib/billing/core/billing-attribution'
 import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import { buildNextCallChain, validateCallChain } from '@/lib/execution/call-chain'
 import { calculateCostSummary } from '@/lib/logs/execution/logging-factory'
@@ -360,6 +361,7 @@ export class WorkflowBlockHandler implements BlockHandler {
       let childUserId = ctx.userId
       let childWorkspaceId = ctx.workspaceId
       let childEnvVarValues = ctx.environmentVariables
+      let childBillingAttribution = ctx.metadata.billingAttribution
       if (isCustomBlock) {
         if (!loadUserId) {
           throw new Error('Custom block source workflow has no owner')
@@ -371,6 +373,15 @@ export class WorkflowBlockHandler implements BlockHandler {
         childWorkspaceId = childWorkflow.workspaceId
         const ownerEnv = await getPersonalAndWorkspaceEnv(loadUserId, childWorkflow.workspaceId)
         childEnvVarValues = { ...ownerEnv.personalDecrypted, ...ownerEnv.workspaceDecrypted }
+        // Custom-block children authenticate internal tool calls as the source
+        // owner in the source workspace, so the consumer's snapshot would fail
+        // the internal routes' actor/workspace scope match. Resolve the
+        // source-scoped payer instead — the same decision those routes made
+        // themselves before attribution headers became required.
+        childBillingAttribution = await resolveBillingAttribution({
+          actorUserId: loadUserId,
+          workspaceId: childWorkflow.workspaceId,
+        })
       }
 
       const subExecutor = new Executor({
@@ -388,6 +399,10 @@ export class WorkflowBlockHandler implements BlockHandler {
           workspaceId: childWorkspaceId,
           userId: childUserId,
           executionId: ctx.executionId,
+          // Same-workspace children share the parent's frozen payer decision so
+          // internal tool calls (knowledge, guardrails, MCP, Mothership) can
+          // attach the required billing attribution header.
+          billingAttribution: childBillingAttribution,
           abortSignal: ctx.abortSignal,
           // Propagate in-flight block-output redaction into child workflows so
           // nested blocks mask outputs too (recurses: each child forwards it).