fix(cjson): santization bullmq payloads accurately

Author: icecrasher321

apps/sim/background/workflow-execution.ts

@@ -3,6 +3,7 @@ import { task } from '@trigger.dev/sdk'
 import { v4 as uuidv4 } from 'uuid'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
 import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core/execution-limits'
+import { ensureArray } from '@/lib/core/utils/json-sanitize'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -110,7 +111,7 @@ export async function executeWorkflowJob(payload: WorkflowExecutionPayload) {
       useDraftState: false,
       startTime: new Date().toISOString(),
       isClientSession: false,
-      callChain: payload.callChain,
+      callChain: ensureArray(payload.callChain),
       correlation,
     }
 

apps/sim/executor/execution/block-executor.ts

@@ -148,7 +148,7 @@ export class BlockExecutor {
             block,
             streamingExec,
             resolvedInputs,
-            ctx.selectedOutputs ?? []
+            Array.isArray(ctx.selectedOutputs) ? ctx.selectedOutputs : []
           )
         }
 

apps/sim/executor/execution/executor.ts

@@ -325,7 +325,9 @@ export class DAGExecutor {
         : new Set(),
       workflow: this.workflow,
       stream: this.contextExtensions.stream ?? false,
-      selectedOutputs: this.contextExtensions.selectedOutputs ?? [],
+      selectedOutputs: Array.isArray(this.contextExtensions.selectedOutputs)
+        ? this.contextExtensions.selectedOutputs
+        : [],
       edges: this.contextExtensions.edges ?? [],
       onStream: this.contextExtensions.onStream,
       onBlockStart: this.contextExtensions.onBlockStart,

apps/sim/executor/execution/snapshot-serializer.ts

@@ -119,7 +119,7 @@ export function serializePauseSnapshot(
     context.workflow,
     {},
     context.workflowVariables ?? {},
-    context.selectedOutputs ?? [],
+    Array.isArray(context.selectedOutputs) ? context.selectedOutputs : [],
     state
   )
 

apps/sim/lib/core/utils/json-sanitize.ts

@@ -0,0 +1,68 @@
+/**
+ * Sanitization for JSON data round-tripped through Redis Lua cjson.
+ *
+ * Lua's cjson library cannot distinguish between empty arrays `[]` and empty objects `{}`.
+ * Both serialize to `{}` in Lua tables. When BullMQ's internal Lua scripts touch job data,
+ * any empty array in the payload silently becomes `{}`.
+ *
+ * Applied once at the worker boundary before data enters the execution engine.
+ */
+
+/**
+ * Returns `value` if it's an array, otherwise `[]`.
+ */
+export function ensureArray<T>(value: unknown): T[] {
+  return Array.isArray(value) ? value : []
+}
+
+const EXECUTION_STATE_ARRAY_FIELDS = [
+  'executedBlocks',
+  'blockLogs',
+  'completedLoops',
+  'activeExecutionPath',
+  'pendingQueue',
+  'remainingEdges',
+  'completedPauseContexts',
+]
+
+/**
+ * Normalizes all known array fields on a BullMQ-deserialized workflow execution payload.
+ * Mutates in place — call once before passing into the execution engine.
+ */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export function sanitizeBullMQPayload(payload: any): void {
+  if (!payload) return
+
+  payload.selectedOutputs = ensureArray(payload.selectedOutputs)
+
+  if (payload.metadata) {
+    payload.metadata.callChain = ensureArray(payload.metadata.callChain)
+
+    if (payload.metadata.pendingBlocks !== undefined) {
+      payload.metadata.pendingBlocks = ensureArray(payload.metadata.pendingBlocks)
+    }
+
+    if (payload.metadata.workflowStateOverride?.edges !== undefined) {
+      payload.metadata.workflowStateOverride.edges = ensureArray(
+        payload.metadata.workflowStateOverride.edges
+      )
+    }
+  }
+
+  if (payload.runFromBlock?.sourceSnapshot) {
+    const state = payload.runFromBlock.sourceSnapshot
+    for (const field of EXECUTION_STATE_ARRAY_FIELDS) {
+      if (field in state && !Array.isArray(state[field])) {
+        state[field] = []
+      }
+    }
+
+    if (state.dagIncomingEdges && typeof state.dagIncomingEdges === 'object') {
+      for (const key of Object.keys(state.dagIncomingEdges)) {
+        if (!Array.isArray(state.dagIncomingEdges[key])) {
+          state.dagIncomingEdges[key] = []
+        }
+      }
+    }
+  }
+}

apps/sim/lib/workflows/executor/execute-workflow.ts

@@ -76,7 +76,7 @@ export async function executeWorkflow(
       workflow,
       input,
       workflow.variables || {},
-      streamConfig?.selectedOutputs || []
+      Array.isArray(streamConfig?.selectedOutputs) ? streamConfig.selectedOutputs : []
     )
 
     const result = await executeWorkflowCore({

apps/sim/lib/workflows/executor/queued-workflow-execution.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core/execution-limits'
+import { sanitizeBullMQPayload } from '@/lib/core/utils/json-sanitize'
 import { createExecutionEventWriter, setExecutionMeta } from '@/lib/execution/event-buffer'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -103,12 +104,14 @@ export async function executeQueuedWorkflowJob(
   }
 
   try {
+    sanitizeBullMQPayload(payload)
+
     const snapshot = new ExecutionSnapshot(
       metadata,
       payload.workflow,
       payload.input,
       payload.variables,
-      Array.isArray(payload.selectedOutputs) ? payload.selectedOutputs : []
+      payload.selectedOutputs as string[]
     )
 
     let callbacks = {}