From f18647cc7dd0535759b68b865b334f3b8d0eae61 Mon Sep 17 00:00:00 2001 From: shrimbly Date: Thu, 2 Apr 2026 06:03:40 +1300 Subject: [PATCH] refactor: extract batch execution helper shared across all entry points The batch textItems loop was only in executeWorkflow. regenerateNode and executeSelectedNodes bypassed batch mode entirely, causing inconsistent behavior when array nodes feed into generate nodes. Extracted runBatchIfApplicable() into src/store/execution/batchExecution.ts and wired it into all three execution entry points. Co-Authored-By: Claude Sonnet 4.5 --- src/store/execution/batchExecution.ts | 99 +++++++++++++++++++++++++++ src/store/execution/index.ts | 2 + src/store/workflowStore.ts | 82 ++++------------------ 3 files changed, 115 insertions(+), 68 deletions(-) create mode 100644 src/store/execution/batchExecution.ts diff --git a/src/store/execution/batchExecution.ts b/src/store/execution/batchExecution.ts new file mode 100644 index 00000000..a6df3481 --- /dev/null +++ b/src/store/execution/batchExecution.ts @@ -0,0 +1,99 @@ +/** + * Batch Execution Helper + * + * Detects batch mode (textItems from array nodes) and loops through items, + * executing the appropriate node executor for each. Shared by executeWorkflow, + * regenerateNode, and executeSelectedNodes. + */ + +import { logger } from "@/utils/logger"; +import type { WorkflowNodeData } from "@/types"; +import type { NodeExecutionContext } from "./types"; +import { executeNanoBanana } from "./nanoBananaExecutor"; +import { executeGenerateVideo } from "./generateVideoExecutor"; +import { executeGenerateAudio } from "./generateAudioExecutor"; +import { executeLlmGenerate } from "./llmGenerateExecutor"; + +const BATCH_NODE_TYPES = new Set(["nanoBanana", "generateVideo", "generateAudio", "llmGenerate"]); + +/** + * Attempts to run batch execution for a node. + * + * If the node type supports batching and has textItems from upstream array + * nodes, iterates through each item and runs the executor individually. + * + * @returns `true` if batch execution was performed, `false` if the node + * should proceed with normal single-item execution. + */ +export async function runBatchIfApplicable( + executionCtx: NodeExecutionContext, + options?: { useStoredFallback?: boolean }, +): Promise { + const { node } = executionCtx; + + if (!node.type || !BATCH_NODE_TYPES.has(node.type)) { + return false; + } + + const connectedInputs = executionCtx.getConnectedInputs(node.id); + if (connectedInputs.textItems.length === 0) { + return false; + } + + const items = connectedInputs.textItems; + const totalItems = items.length; + + for (let i = 0; i < totalItems; i++) { + if (executionCtx.signal?.aborted) { + throw new DOMException("Aborted", "AbortError"); + } + + executionCtx.updateNodeData(node.id, { + status: "loading", + error: null, + } as Partial); + + logger.info("node.execution", `Batch ${i + 1} of ${totalItems}`, { + nodeId: node.id, + nodeType: node.type, + batchIndex: i, + batchTotal: totalItems, + }); + + // Wrap context so getConnectedInputs returns current batch item as text + const batchCtx: NodeExecutionContext = { + ...executionCtx, + getConnectedInputs: (nodeId: string) => { + const inputs = executionCtx.getConnectedInputs(nodeId); + return { + ...inputs, + text: items[i], + textItems: [], + }; + }, + }; + + switch (node.type) { + case "nanoBanana": + await executeNanoBanana(batchCtx, options); + break; + case "generateVideo": + await executeGenerateVideo(batchCtx, options); + break; + case "generateAudio": + await executeGenerateAudio(batchCtx, options); + break; + case "llmGenerate": + await executeLlmGenerate(batchCtx, options); + break; + } + + if (i < totalItems - 1) { + executionCtx.updateNodeData(node.id, { + status: "loading", + } as Partial); + } + } + + return true; +} diff --git a/src/store/execution/index.ts b/src/store/execution/index.ts index 3f47465a..fc5fd6bf 100644 --- a/src/store/execution/index.ts +++ b/src/store/execution/index.ts @@ -45,3 +45,5 @@ export { executeVideoTrim, executeVideoFrameGrab, } from "./videoProcessingExecutors"; + +export { runBatchIfApplicable } from "./batchExecution"; diff --git a/src/store/workflowStore.ts b/src/store/workflowStore.ts index 993dc773..a76508cf 100644 --- a/src/store/workflowStore.ts +++ b/src/store/workflowStore.ts @@ -85,6 +85,7 @@ import { executeRouter, executeSwitch, executeConditionalSwitch, + runBatchIfApplicable, } from "./execution"; import type { NodeExecutionContext } from "./execution"; export type { LevelGroup } from "./utils/executionUtils"; @@ -1322,73 +1323,8 @@ const workflowStoreImpl: StateCreator = (set, get) => ({ const executionCtx = get()._buildExecutionContext(node, signal); // Batch mode: for generate-type nodes, detect textItems and loop through them - const batchNodeTypes = new Set(["nanoBanana", "generateVideo", "generateAudio", "llmGenerate"]); - if (node.type && batchNodeTypes.has(node.type)) { - const connectedInputs = get().getConnectedInputs(node.id); - if (connectedInputs.textItems.length > 0) { - const items = connectedInputs.textItems; - const totalItems = items.length; - - for (let i = 0; i < totalItems; i++) { - // Check abort signal before each iteration - if (signal.aborted) { - throw new DOMException('Aborted', 'AbortError'); - } - - // Update status with batch progress - get().updateNodeData(node.id, { - status: "loading", - error: null, - } as Partial); - - logger.info('node.execution', `Batch ${i + 1} of ${totalItems}`, { - nodeId: node.id, - nodeType: node.type, - batchIndex: i, - batchTotal: totalItems, - }); - - // Create a wrapped context where getConnectedInputs returns - // the current item as `text` and clears `textItems` - const batchCtx: NodeExecutionContext = { - ...executionCtx, - getConnectedInputs: (nodeId: string) => { - const inputs = get().getConnectedInputs(nodeId); - return { - ...inputs, - text: items[i], - textItems: [], // Clear so executors don't see batch - }; - }, - }; - - // Execute the appropriate executor - switch (node.type) { - case "nanoBanana": - await executeNanoBanana(batchCtx); - break; - case "generateVideo": - await executeGenerateVideo(batchCtx); - break; - case "generateAudio": - await executeGenerateAudio(batchCtx); - break; - case "llmGenerate": - await executeLlmGenerate(batchCtx); - break; - } - - // After each iteration (except last), reset status to loading for next iteration - if (i < totalItems - 1) { - get().updateNodeData(node.id, { - status: "loading", - } as Partial); - } - } - - // Batch complete — skip the normal switch below - return; - } + if (await runBatchIfApplicable(executionCtx)) { + return; } switch (node.type) { @@ -1605,7 +1541,12 @@ const workflowStoreImpl: StateCreator = (set, get) => ({ const regenOptions = { useStoredFallback: true }; - if (node.type === "nanoBanana") { + // Try batch mode first (handles textItems from array nodes) + const wasBatch = await runBatchIfApplicable(executionCtx, regenOptions); + + if (wasBatch) { + // Batch handled — skip to downstream execution + } else if (node.type === "nanoBanana") { await executeNanoBanana(executionCtx, regenOptions); } else if (node.type === "array") { await executeArray(executionCtx); @@ -1738,6 +1679,11 @@ const workflowStoreImpl: StateCreator = (set, get) => ({ const executionCtx = get()._buildExecutionContext(node, signal); const regenOptions = { useStoredFallback: true }; + // Try batch mode first (handles textItems from array nodes) + if (await runBatchIfApplicable(executionCtx, regenOptions)) { + return; + } + switch (node.type) { case "imageInput": // Data source node - no execution needed