fix(workflows): resume legacy queued jobs

Author: TheodoreSpeaks

apps/sim/background/async-preprocessing-correlation.test.ts

@@ -227,6 +227,98 @@ describe('async preprocessing correlation threading', () => {
     )
   })
 
+  it('restores a legacy authenticated workflow job as its recorded user actor', async () => {
+    mockPreprocessExecution.mockResolvedValueOnce({
+      success: true,
+      actorUserId: 'actor-1',
+      workflowRecord: {
+        id: 'workflow-1',
+        userId: 'owner-1',
+        workspaceId: 'workspace-1',
+        variables: {},
+      },
+      billingAttribution,
+      executionTimeout: {},
+    })
+    mockExecuteWorkflowCore.mockResolvedValueOnce({
+      success: true,
+      status: 'success',
+      output: { ok: true },
+      metadata: { duration: 10, userId: 'actor-1' },
+    })
+
+    await executeWorkflowJob({
+      workflowId: 'workflow-1',
+      userId: 'actor-1',
+      workspaceId: 'workspace-1',
+      billingAttribution,
+      triggerType: 'api',
+      executionId: 'legacy-user-execution',
+      requestId: 'legacy-user-request',
+      enforceCredentialAccess: true,
+    })
+
+    expect(mockExecutionSnapshot).toHaveBeenCalledWith(
+      expect.objectContaining({
+        principal: {
+          kind: 'session',
+          userId: 'actor-1',
+          sessionId: 'legacy-queued-workflow',
+        },
+      }),
+      expect.anything(),
+      undefined,
+      expect.any(Object),
+      expect.any(Array)
+    )
+  })
+
+  it('restores an identity-ambiguous legacy workflow job as actorless', async () => {
+    mockPreprocessExecution.mockResolvedValueOnce({
+      success: true,
+      actorUserId: 'actor-1',
+      workflowRecord: {
+        id: 'workflow-1',
+        userId: 'owner-1',
+        workspaceId: 'workspace-1',
+        variables: {},
+      },
+      billingAttribution,
+      executionTimeout: {},
+    })
+    mockExecuteWorkflowCore.mockResolvedValueOnce({
+      success: true,
+      status: 'success',
+      output: { ok: true },
+      metadata: { duration: 10, userId: 'actor-1' },
+    })
+
+    await executeWorkflowJob({
+      workflowId: 'workflow-1',
+      userId: 'actor-1',
+      workspaceId: 'workspace-1',
+      billingAttribution,
+      triggerType: 'api',
+      executionId: 'legacy-actorless-execution',
+      requestId: 'legacy-actorless-request',
+    })
+
+    expect(mockExecutionSnapshot).toHaveBeenCalledWith(
+      expect.objectContaining({
+        principal: {
+          kind: 'system',
+          serviceId: 'internal',
+          workspaceId: 'workspace-1',
+          workflowId: 'workflow-1',
+        },
+      }),
+      expect.anything(),
+      undefined,
+      expect.any(Object),
+      expect.any(Array)
+    )
+  })
+
   it('passes validated workflow input provenance from the queued payload into core execution', async () => {
     const provenance = {
       version: 1 as const,

apps/sim/background/webhook-execution.test.ts

@@ -25,6 +25,7 @@ const {
   mockLoadDeploymentVersionState,
   mockGetProviderHandler,
   mockSetResolvedSecretTraceRegistry,
+  mockExecutionSnapshot,
 } = vi.hoisted(() => ({
   mockResolveWebhookRecordProviderConfig: vi.fn(),
   mockExecuteWorkflowCore: vi.fn(),
@@ -34,6 +35,7 @@ const {
   mockReleaseExecutionSlot: vi.fn(),
   mockGetProviderHandler: vi.fn(() => ({})),
   mockSetResolvedSecretTraceRegistry: vi.fn(),
+  mockExecutionSnapshot: vi.fn(),
   mockLoadDeploymentVersionState: vi.fn(
     async (_workflowId: string, deploymentVersionId: string) => ({
       blocks: {},
@@ -120,7 +122,7 @@ vi.mock('@/lib/oauth/credential-service', () => ({
 }))
 
 vi.mock('@/executor/execution/snapshot', () => ({
-  ExecutionSnapshot: class {},
+  ExecutionSnapshot: mockExecutionSnapshot,
 }))
 
 vi.mock('@/tools/safe-assign', () => ({ safeAssign: vi.fn() }))
@@ -281,6 +283,55 @@ describe('executeWebhookJob fault vs error handling', () => {
     dbChainMockFns.limit.mockResolvedValue([{ id: 'webhook-1' }])
   })
 
+  it('restores a legacy queued webhook as its canonical system principal', async () => {
+    const legacyPayload = {
+      webhookId: payload.webhookId,
+      workflowId: payload.workflowId,
+      userId: payload.userId,
+      billingAttribution: payload.billingAttribution,
+      executionId: payload.executionId,
+      requestId: payload.requestId,
+      provider: payload.provider,
+      body: payload.body,
+      headers: payload.headers,
+      path: payload.path,
+      workspaceId: payload.workspaceId,
+    }
+    mockExecuteWorkflowCore.mockResolvedValueOnce({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+
+    await executeWebhookJob(legacyPayload)
+
+    expect(mockExecutionSnapshot).toHaveBeenCalledWith(
+      expect.objectContaining({
+        principal: {
+          kind: 'system',
+          serviceId: 'webhook',
+          webhookId: 'webhook-1',
+          workflowId: 'workflow-1',
+          workspaceId: 'workspace-1',
+          provider: 'gmail',
+        },
+      }),
+      expect.anything(),
+      expect.anything(),
+      expect.any(Object),
+      expect.any(Array)
+    )
+  })
+
   it('completes the run (does not throw) when the failure was finalized by core', async () => {
     mockExecuteWorkflowCore.mockRejectedValue(
       new Error('Gmail 2 is missing required fields: Label')

apps/sim/background/webhook-execution.ts

@@ -49,7 +49,10 @@ import {
   resolveWebhookRecordProviderConfig,
   type WebhookEnvResolutionOptions,
 } from '@/lib/webhooks/env-resolver'
-import { assertWebhookExecutionPrincipal } from '@/lib/webhooks/execution-principal'
+import {
+  assertWebhookExecutionPrincipal,
+  createWebhookExecutionPrincipal,
+} from '@/lib/webhooks/execution-principal'
 import { getProviderHandler } from '@/lib/webhooks/providers'
 import {
   executeWorkflowCore,
@@ -165,7 +168,7 @@ function normalizeWebhookAttachments(value: unknown): WebhookAttachment[] {
 }
 
 export function buildWebhookCorrelation(
-  payload: WebhookExecutionPayload
+  payload: WebhookExecutionJobPayload
 ): AsyncExecutionCorrelation {
   const executionId = payload.executionId || generateId()
   const requestId = payload.requestId || payload.correlation?.requestId || executionId.slice(0, 8)
@@ -298,8 +301,26 @@ export type WebhookExecutionPayload = {
   executionTimeoutMs?: number
 }
 
+type LegacyWebhookExecutionPayload = Omit<WebhookExecutionPayload, 'principal'> & {
+  /** Jobs queued before execution principals were introduced omit this field. */
+  principal?: undefined
+}
+
+type WebhookExecutionJobPayload = WebhookExecutionPayload | LegacyWebhookExecutionPayload
+
+/** Reconstructs the exact system authority recorded by the pre-principal webhook payload. */
+function parseWebhookJobPrincipal(payload: WebhookExecutionJobPayload): WorkflowExecutionPrincipal {
+  if (payload.principal !== undefined) return parsePrincipal(payload.principal)
+  return createWebhookExecutionPrincipal({
+    webhookId: payload.webhookId,
+    workflowId: payload.workflowId,
+    workspaceId: payload.workspaceId,
+    provider: payload.provider,
+  })
+}
+
 export async function executeWebhookJob(
-  payload: WebhookExecutionPayload,
+  payload: WebhookExecutionJobPayload,
   externalAbortSignal?: AbortSignal
 ) {
   const correlation = buildWebhookCorrelation(payload)
@@ -308,7 +329,7 @@ export async function executeWebhookJob(
   let payloadBillingAttribution: BillingAttributionSnapshot
   let principal: WorkflowExecutionPrincipal
   try {
-    principal = parsePrincipal(payload.principal)
+    principal = parseWebhookJobPrincipal(payload)
     assertWebhookExecutionPrincipal(principal, payload)
     payloadBillingAttribution = assertBillingAttributionSnapshot(payload.billingAttribution)
     if (
@@ -490,7 +511,7 @@ async function handleExecutionResult(
 }
 
 async function executeWebhookJobInternal(
-  payload: WebhookExecutionPayload,
+  payload: WebhookExecutionJobPayload,
   principal: WorkflowExecutionPrincipal,
   correlation: AsyncExecutionCorrelation,
   timeoutController: ReturnType<typeof createTimeoutAbortController>,
@@ -933,6 +954,6 @@ export const webhookExecution = task({
   queue: {
     concurrencyLimit: WEBHOOK_EXECUTION_CONCURRENCY_LIMIT,
   },
-  run: async (payload: WebhookExecutionPayload, { signal }: { signal: AbortSignal }) =>
+  run: async (payload: WebhookExecutionJobPayload, { signal }: { signal: AbortSignal }) =>
     executeWebhookJob(payload, signal),
 })

apps/sim/background/workflow-execution.ts

@@ -1,4 +1,8 @@
-import { parsePrincipal, type SerializedPrincipalV1 } from '@sim/auth/principal'
+import {
+  parsePrincipal,
+  type SerializedPrincipalV1,
+  type WorkflowExecutionPrincipal,
+} from '@sim/auth/principal'
 import { createLogger, runWithRequestContext } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
@@ -38,9 +42,10 @@ import type { ResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-
 import type { CoreTriggerType } from '@/stores/logs/filters/types'
 
 const logger = createLogger('TriggerWorkflowExecution')
+const LEGACY_WORKFLOW_JOB_SESSION_ID = 'legacy-queued-workflow'
 
 export function buildWorkflowCorrelation(
-  payload: WorkflowExecutionPayload
+  payload: WorkflowExecutionJobPayload
 ): AsyncExecutionCorrelation {
   const executionId = payload.executionId || generateId()
   const requestId = payload.requestId || payload.correlation?.requestId || executionId.slice(0, 8)
@@ -88,13 +93,44 @@ export type WorkflowExecutionPayload = {
   isPublicApiAccess?: boolean
 }
 
+type LegacyWorkflowExecutionPayload = Omit<WorkflowExecutionPayload, 'principal'> & {
+  /** Jobs queued before execution principals were introduced omit this field. */
+  principal?: undefined
+}
+
+type WorkflowExecutionJobPayload = WorkflowExecutionPayload | LegacyWorkflowExecutionPayload
+
+function requireLegacyWorkflowJobString(value: unknown, field: string): string {
+  if (typeof value !== 'string' || !value.trim()) {
+    throw new Error(`Legacy workflow job ${field} must be a non-empty string`)
+  }
+  return value
+}
+
+/** Restores only the actor decision recorded by the pre-principal queue payload. */
+function parseWorkflowJobPrincipal(
+  payload: WorkflowExecutionJobPayload
+): WorkflowExecutionPrincipal {
+  if (payload.principal !== undefined) return parsePrincipal(payload.principal)
+  const workflowId = requireLegacyWorkflowJobString(payload.workflowId, 'workflowId')
+  const workspaceId = requireLegacyWorkflowJobString(payload.workspaceId, 'workspaceId')
+  if (payload.enforceCredentialAccess === true) {
+    return {
+      kind: 'session',
+      userId: requireLegacyWorkflowJobString(payload.userId, 'userId'),
+      sessionId: LEGACY_WORKFLOW_JOB_SESSION_ID,
+    }
+  }
+  return { kind: 'system', serviceId: 'internal', workspaceId, workflowId }
+}
+
 /**
  * Background workflow execution job
  * @see preprocessExecution For detailed information on preprocessing checks
  * @see executeWorkflowCore For the core workflow execution logic
  */
 export async function executeWorkflowJob(
-  payload: WorkflowExecutionPayload,
+  payload: WorkflowExecutionJobPayload,
   externalAbortSignal?: AbortSignal
 ) {
   const workflowId = payload.workflowId
@@ -104,7 +140,7 @@ export async function executeWorkflowJob(
   let principal
   let billingAttribution: BillingAttributionSnapshot
   try {
-    principal = parsePrincipal(payload.principal)
+    principal = parseWorkflowJobPrincipal(payload)
     billingAttribution = assertBillingAttributionSnapshot(payload.billingAttribution)
     if (
       billingAttribution.actorUserId !== payload.userId ||
@@ -309,5 +345,5 @@ export const workflowExecutionTask = task({
   queue: {
     concurrencyLimit: WORKFLOW_EXECUTION_CONCURRENCY_LIMIT,
   },
-  run: (payload: WorkflowExecutionPayload, { signal }) => executeWorkflowJob(payload, signal),
+  run: (payload: WorkflowExecutionJobPayload, { signal }) => executeWorkflowJob(payload, signal),
 })