diff --git a/src/components/WorkflowCanvas.tsx b/src/components/WorkflowCanvas.tsx index c30d7deb..51904ad1 100644 --- a/src/components/WorkflowCanvas.tsx +++ b/src/components/WorkflowCanvas.tsx @@ -1207,6 +1207,10 @@ export function WorkflowCanvas() { } else if (nodeType === "generateAudio") { // GenerateAudio outputs audio (no audio input to wire to) sourceHandleIdForNewNode = "audio"; + } else if (nodeType === "generateVideo") { + // GenerateVideo accepts audio input and outputs video + targetHandleId = "audio"; + sourceHandleIdForNewNode = "video"; } else if (nodeType === "videoStitch") { // VideoStitch accepts audio targetHandleId = "audio"; 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/undoHistory.ts b/src/store/undoHistory.ts index f820c945..579ef318 100644 --- a/src/store/undoHistory.ts +++ b/src/store/undoHistory.ts @@ -57,9 +57,13 @@ export class UndoManager { * snapshots, but strings (immutable in JS) are returned by reference. * This avoids duplicating multi-MB base64 blobs across undo history. * - * Matches JSON.parse(JSON.stringify()) semantics: + * Matches JSON.parse(JSON.stringify()) semantics for plain JSON-like + * objects/arrays: * - `undefined` values are dropped from objects, become `null` in arrays * - functions are dropped from objects, become `null` in arrays + * + * Does NOT call toJSON() on objects. Objects with custom toJSON methods + * are treated as plain objects (their enumerable own properties are cloned). */ export function clonePreservingStrings(value: T): T { if (value === null || typeof value !== "object") { diff --git a/src/store/utils/connectedInputs.ts b/src/store/utils/connectedInputs.ts index d5250e21..563699a1 100644 --- a/src/store/utils/connectedInputs.ts +++ b/src/store/utils/connectedInputs.ts @@ -229,7 +229,8 @@ export function getConnectedInputsPure( if (dimmedNodeIds && dimmedNodeIds.has(sourceNode.id)) return; // Array batch mode — send all items as textItems instead of a single item - if (sourceNode.type === "array" && (edge.data as Record | undefined)?.arrayBatchAll === true) { + // Derive from source node's current batchMode (not edge metadata which can go stale) + if (sourceNode.type === "array" && (sourceNode.data as ArrayNodeData).batchMode === true) { const arrayData = sourceNode.data as ArrayNodeData; const items = arrayData.outputItems; if (items.length > 0) { diff --git a/src/store/workflowStore.ts b/src/store/workflowStore.ts index 993dc773..e97e3478 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"; @@ -145,11 +146,8 @@ function buildConnectionEdgeData( if (sourceNode?.type === "array" && (connection.sourceHandle || "text") === "text") { const sourceData = sourceNode.data as Record; - // Batch mode: all items sent through a single connection - if (sourceData.batchMode === true) { - baseData.arrayBatchAll = true; - return baseData; - } + // Batch mode is now derived dynamically in connectedInputs.ts from + // the source node's batchMode — no need to stamp edge metadata. const selectedIndex = sourceData.selectedOutputIndex; const outputItems = Array.isArray(sourceData.outputItems) ? sourceData.outputItems : []; @@ -909,8 +907,8 @@ const workflowStoreImpl: StateCreator = (set, get) => ({ ); // Deep clone the nodes and edges to avoid reference issues - const clonedNodes = JSON.parse(JSON.stringify(selectedNodes)) as WorkflowNode[]; - const clonedEdges = JSON.parse(JSON.stringify(connectedEdges)) as WorkflowEdge[]; + const clonedNodes = clonePreservingStrings(selectedNodes) as WorkflowNode[]; + const clonedEdges = clonePreservingStrings(connectedEdges) as WorkflowEdge[]; set({ clipboard: { nodes: clonedNodes, edges: clonedEdges } }); }, @@ -948,7 +946,7 @@ const workflowStoreImpl: StateCreator = (set, get) => ({ width: undefined, height: undefined, measured: undefined, - data: JSON.parse(JSON.stringify(node.data)), + data: clonePreservingStrings(node.data), }; }); @@ -1322,73 +1320,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 +1538,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 +1676,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 @@ -2593,12 +2536,12 @@ const workflowStoreImpl: StateCreator = (set, get) => ({ captureSnapshot: () => { const state = get(); // Deep copy the current workflow state to avoid reference sharing - const snapshot = { - nodes: JSON.parse(JSON.stringify(state.nodes)), - edges: JSON.parse(JSON.stringify(state.edges)), - groups: JSON.parse(JSON.stringify(state.groups)), + const snapshot = clonePreservingStrings({ + nodes: state.nodes, + edges: state.edges, + groups: state.groups, edgeStyle: state.edgeStyle, - }; + }); set({ previousWorkflowSnapshot: snapshot, manualChangeCount: 0,