From f80812153c2b5d7eac2a0f9e55627429a99bcd08 Mon Sep 17 00:00:00 2001 From: Lorchie Date: Tue, 14 Jul 2026 00:59:16 +0200 Subject: [PATCH] feat(workflows): add For Each iterator nodes (image / text / mesh) Adds a For Each source node that walks a folder alphabetically and emits one file per loop pass, re-running the downstream nodes for each item. Supports image, text and mesh modes with Pause / Continue / Retry controls, and runs several iterators in lockstep when they share a downstream body. - ForEachNode UI + palette entry, container-type detection generalized - runner: iterator resolution, loop table, per-file execution - fs.listFiles / fs.selectTextFile IPC helpers for folder listing - preflight labels and output typing for the new node --- electron/main/ipc-handlers.ts | 32 +++ electron/preload/electron-api.ts | 4 + src/areas/workflows/WorkflowsPage.tsx | 28 +- src/areas/workflows/nodes/ForEachNode.tsx | 139 ++++++++++ src/areas/workflows/preflight.ts | 16 ++ src/areas/workflows/workflowRunStore.ts | 301 +++++++++++++++++++--- src/shared/stores/workflowsStore.ts | 2 +- src/shared/types/electron.d.ts | 2 + 8 files changed, 473 insertions(+), 51 deletions(-) create mode 100644 src/areas/workflows/nodes/ForEachNode.tsx diff --git a/electron/main/ipc-handlers.ts b/electron/main/ipc-handlers.ts index 695bda0f..60b8413b 100644 --- a/electron/main/ipc-handlers.ts +++ b/electron/main/ipc-handlers.ts @@ -677,6 +677,38 @@ export function setupIpcHandlers(pythonBridge: PythonBridge, getWindow: WindowGe } }) + // List files (not directories) in a folder, optionally filtered by extension. + // `extensions` are lowercase without the dot (e.g. ['txt', 'png']). + ipcMain.handle('fs:listFiles', async (_, dirPath: string, extensions?: string[]) => { + try { + const wanted = extensions?.map(e => e.toLowerCase().replace(/^\./, '')) + const entries = await readdir(dirPath, { withFileTypes: true }) + return entries + .filter(e => e.isFile()) + .map(e => e.name) + .filter(name => { + if (!wanted || wanted.length === 0) return true + const ext = name.split('.').pop()?.toLowerCase() ?? '' + return wanted.includes(ext) + }) + .sort((a, b) => a.localeCompare(b, undefined, { numeric: true })) + } catch { + return [] + } + }) + + // Text-file picker for the standalone Load Prompt node. + ipcMain.handle('fs:selectTextFile', async () => { + const win = getWindow() + if (!win) return null + const result = await dialog.showOpenDialog(win, { + title: 'Select a prompt text file', + filters: [{ name: 'Text', extensions: ['txt', 'md', 'prompt'] }], + properties: ['openFile'], + }) + return result.canceled ? null : result.filePaths[0] + }) + ipcMain.handle('fs:moveDirectory', async (_, { src, dest }: { src: string; dest: string }) => { try { await mkdir(dest, { recursive: true }) diff --git a/electron/preload/electron-api.ts b/electron/preload/electron-api.ts index 6ced29c3..150d0678 100644 --- a/electron/preload/electron-api.ts +++ b/electron/preload/electron-api.ts @@ -70,6 +70,10 @@ export function createElectronApi(ipcRenderer: IpcRendererLike, webFrame: WebFra ipcRenderer.invoke('fs:savePath', args) as Promise, listDir: (dirPath: string): Promise => ipcRenderer.invoke('fs:listDir', dirPath) as Promise, + listFiles: (dirPath: string, extensions?: string[]): Promise => + ipcRenderer.invoke('fs:listFiles', dirPath, extensions) as Promise, + selectTextFile: (): Promise => + ipcRenderer.invoke('fs:selectTextFile') as Promise, moveDirectory: (args: { src: string; dest: string }): Promise<{ success: boolean; error?: string }> => ipcRenderer.invoke('fs:moveDirectory', args) as Promise<{ success: boolean; error?: string }>, deleteDirectory: (dirPath: string): Promise<{ success: boolean; error?: string }> => diff --git a/src/areas/workflows/WorkflowsPage.tsx b/src/areas/workflows/WorkflowsPage.tsx index a542284f..f40ab73b 100644 --- a/src/areas/workflows/WorkflowsPage.tsx +++ b/src/areas/workflows/WorkflowsPage.tsx @@ -30,13 +30,19 @@ import Load3DMeshNode from './nodes/Load3DMeshNode' import PreviewImageNode from './nodes/PreviewImageNode' import WaitNode from './nodes/WaitNode' import WhileNode from './nodes/WhileNode' +import ForEachNode from './nodes/ForEachNode' import WorkflowEdge from './nodes/WorkflowEdge' // ─── Constants ──────────────────────────────────────────────────────────────── const DRAG_KEY = 'modly/extension-id' const DRAG_NODE_KEY = 'modly/node-type' -const NODE_TYPES = { extensionNode: ExtensionNode, imageNode: ImageNode, textNode: TextNode, outputNode: AddToSceneNode, meshNode: Load3DMeshNode, previewNode: PreviewImageNode, waitNode: WaitNode, whileNode: WhileNode } +const NODE_TYPES = { extensionNode: ExtensionNode, imageNode: ImageNode, textNode: TextNode, outputNode: AddToSceneNode, meshNode: Load3DMeshNode, previewNode: PreviewImageNode, waitNode: WaitNode, whileNode: WhileNode, forEachNode: ForEachNode } + +// Loop-container node types: resizable frames whose children form a loop body. +// (For Each iterators are plain source nodes, not containers.) +const CONTAINER_TYPES = new Set(['whileNode']) +const isContainerType = (type: string | undefined): boolean => !!type && CONTAINER_TYPES.has(type) const EDGE_TYPES = { workflowEdge: WorkflowEdge } const DEFAULT_EDGE_OPTS = { type: 'workflowEdge' } @@ -45,7 +51,7 @@ const DEFAULT_EDGE_OPTS = { type: 'workflowEdge' } // auto-parent nodes dropped (or created) inside a While so they join its loop body. function findWhileContainerAt(nodes: Node[], pos: { x: number; y: number }): Node | undefined { return nodes.find((n) => { - if (n.type !== 'whileNode') return false + if (!isContainerType(n.type)) return false const gw = (n.measured?.width ?? n.width ?? (n.style?.width as number)) || 0 const gh = (n.measured?.height ?? n.height ?? (n.style?.height as number)) || 0 return pos.x >= n.position.x && pos.x <= n.position.x + gw @@ -92,6 +98,7 @@ const PANEL_BUILTIN_NODES = [ { type: 'previewNode', label: 'Preview Views', color: '#38bdf8', icon: <> }, { type: 'waitNode', label: 'Wait', color: '#71717a', icon: <> }, { type: 'whileNode', label: 'While', color: '#f59e0b', icon: <> }, + { type: 'forEachNode', label: 'For Each', color: '#38bdf8', icon: <> }, ] function ExtGroupHeader({ title, author, expanded, onToggle, count }: { title: string; author?: string; expanded: boolean; onToggle: () => void; count: number }) { @@ -337,6 +344,7 @@ const BUILTIN_NODES = [ { type: 'previewNode', label: 'Preview Views', color: '#38bdf8', description: 'Displays multi-view image outputs in a 2×3 grid' }, { type: 'waitNode', label: 'Wait', color: '#71717a', description: 'Pauses the workflow until you click Continue' }, { type: 'whileNode', label: 'While', color: '#f59e0b', description: 'Container: wrap nodes to loop them N times or with Continue/Retry' }, + { type: 'forEachNode', label: 'For Each', color: '#38bdf8', description: 'Iterates a folder (image / text / mesh) alphabetically, one item per run of the downstream nodes' }, ] type PaletteItem = @@ -879,8 +887,8 @@ function WorkflowCanvasInner({ // bail only on a real node or a handle. const target = event.target as Element const nodeEl = target.closest('.react-flow__node') - const onWhile = nodeEl?.classList.contains('react-flow__node-whileNode') ?? false - if (target.closest('.react-flow__handle') || (nodeEl && !onWhile)) { + const onContainer = !!nodeEl && [...CONTAINER_TYPES].some((t) => nodeEl.classList.contains(`react-flow__node-${t}`)) + if (target.closest('.react-flow__handle') || (nodeEl && !onContainer)) { pendingConnectionRef.current = null return } @@ -900,7 +908,7 @@ function WorkflowCanvasInner({ const nodeType = e.dataTransfer.getData(DRAG_NODE_KEY) if (nodeType) { - const isContainer = nodeType === 'whileNode' + const isContainer = isContainerType(nodeType) setNodes((nds) => { const parent = isContainer ? undefined : findWhileContainerAt(nds, position) const node: Node = { @@ -959,7 +967,7 @@ function WorkflowCanvasInner({ pendingDropPos ?? { x: window.innerWidth / 2, y: window.innerHeight / 2 } ) const newNodeId = newId() - const isContainer = type === 'whileNode' + const isContainer = isContainerType(type) setNodes((nds) => { const parent = isContainer ? undefined : findWhileContainerAt(nds, position) const node: Node = { @@ -1001,7 +1009,7 @@ function WorkflowCanvasInner({ // When a While container is deleted (button or keyboard), detach its children // to absolute coordinates so they don't get orphaned to the canvas origin. const onNodesDelete = useCallback((deleted: Node[]) => { - const removedContainers = deleted.filter((n) => n.type === 'whileNode') + const removedContainers = deleted.filter((n) => isContainerType(n.type)) if (removedContainers.length === 0) return setNodes((nds) => nds.map((n) => { const container = removedContainers.find((c) => c.id === n.parentId) @@ -1014,9 +1022,9 @@ function WorkflowCanvasInner({ // When a node is dropped, attach/detach it to a While container based on overlap. // Children get a parentId + parent-relative position (no extent, so they can be dragged back out). const onNodeDragStop = useCallback((_e: unknown, dragged: Node) => { - if (dragged.type === 'whileNode') return + if (isContainerType(dragged.type)) return setNodes((nds) => { - const containers = nds.filter((n) => n.type === 'whileNode') + const containers = nds.filter((n) => isContainerType(n.type)) if (containers.length === 0 && !dragged.parentId) return nds const parent = dragged.parentId ? nds.find((n) => n.id === dragged.parentId) : undefined @@ -1036,7 +1044,7 @@ function WorkflowCanvasInner({ const newParentId = container?.id if (newParentId === dragged.parentId) return nds // no change - let next: Node[] = nds.map((n) => { + const next: Node[] = nds.map((n) => { if (n.id !== dragged.id) return n if (container) { // parentId (no extent) → child moves with the container but can still be dragged out diff --git a/src/areas/workflows/nodes/ForEachNode.tsx b/src/areas/workflows/nodes/ForEachNode.tsx new file mode 100644 index 00000000..3b14d3fa --- /dev/null +++ b/src/areas/workflows/nodes/ForEachNode.tsx @@ -0,0 +1,139 @@ +import { useLayoutEffect, useRef, useState } from 'react' +import { Handle, Position, useReactFlow } from '@xyflow/react' +import type { WFNodeData } from '@shared/types/electron.d' +import BaseNode from './BaseNode' +import { useWorkflowRunStore } from '../workflowRunStore' + +// For Each — an iterator source. It walks a folder alphabetically and emits one +// file per loop pass; every node wired downstream re-runs for each file. Runs to +// completion unless the user hits Stop, with an optional Pause (then Continue → +// next file / Retry → same file). A dropdown picks what it emits: image / text / mesh. +type Mode = 'image' | 'text' | 'mesh' + +const MODES: Record = { + image: { title: 'For Each Image', color: '#38bdf8', hint: 'images' }, + text: { title: 'For Each Text', color: '#fbbf24', hint: 'text files' }, + mesh: { title: 'For Each Mesh', color: '#a78bfa', hint: 'meshes' }, +} + +export default function ForEachNode({ id, data, selected }: { id: string; data: WFNodeData; selected?: boolean }) { + const { updateNodeData } = useReactFlow() + const mode = ((data.params.mode as Mode | undefined) ?? 'image') + const variant = MODES[mode] ?? MODES.image + + const status = useWorkflowRunStore((s) => s.runState.status) + const activeNodeId = useWorkflowRunStore((s) => s.activeNodeId) + const continueWhile = useWorkflowRunStore((s) => s.continueWhile) + const retryWhile = useWorkflowRunStore((s) => s.retryWhile) + const pauseWhile = useWorkflowRunStore((s) => s.pauseWhile) + const progress = useWorkflowRunStore((s) => s.whileProgress[id]) + const pausedGroup = useWorkflowRunStore((s) => s.pausedGroup) + const isPaused = status === 'paused' && (activeNodeId === id || pausedGroup.includes(id)) + const isLooping = status === 'running' && progress != null && !isPaused + const locked = status === 'running' || status === 'paused' + + const ioRowRef = useRef(null) + const [handleTop, setHandleTop] = useState('50%') + useLayoutEffect(() => { + if (ioRowRef.current) setHandleTop(`${ioRowRef.current.offsetTop + ioRowRef.current.offsetHeight / 2}px`) + }, []) + + const dir = (data.params.dir as string | undefined) ?? '' + const dirLabel = dir ? dir.split(/[\\/]/).filter(Boolean).pop() : 'Pick folder…' + + const pickDir = async (): Promise => { + const picked = await window.electron.fs.selectDirectory(dir || undefined) + if (picked) updateNodeData(id, { params: { ...data.params, dir: picked } }) + } + + return ( + + + + } + subheader={ +
+ {progress && ( + + {progress.total != null ? `${progress.current}/${progress.total}` : progress.current} + + )} + + {mode} + +
+ } + handles={ + + } + > +
+ {/* Iterated type */} + + + + + {/* Run controls */} + {isLooping && ( + + )} + {isPaused && ( +
+ + +
+ )} +
+ + ) +} diff --git a/src/areas/workflows/preflight.ts b/src/areas/workflows/preflight.ts index f8493987..b33c1851 100644 --- a/src/areas/workflows/preflight.ts +++ b/src/areas/workflows/preflight.ts @@ -16,6 +16,10 @@ function nodeLabel(node: WFNode, allExtensions: WorkflowExtension[]): string { if (node.type === 'meshNode') return 'Load 3D Mesh' if (node.type === 'outputNode') return 'Add to Scene' if (node.type === 'previewNode') return 'Preview Views' + if (node.type === 'forEachNode') { + const mode = (node.data.params?.mode as string) ?? 'image' + return mode === 'text' ? 'For Each Text' : mode === 'mesh' ? 'For Each Mesh' : 'For Each Image' + } if (node.type === 'extensionNode') { return getWorkflowExtension(node.data.extensionId ?? '', allExtensions)?.name ?? 'Extension' } @@ -40,6 +44,10 @@ function getNodeOutputType(node: WFNode, allExtensions: WorkflowExtension[]): Da if (node.type === 'textNode') return 'text' if (node.type === 'meshNode' || node.type === 'outputNode') return 'mesh' if (node.type === 'previewNode') return 'image' + if (node.type === 'forEachNode') { + const mode = (node.data.params?.mode as DataType | undefined) ?? 'image' + return mode === 'text' || mode === 'mesh' ? mode : 'image' + } if (node.type === 'extensionNode') { return getWorkflowExtension(node.data.extensionId ?? '', allExtensions)?.output } @@ -78,6 +86,14 @@ export function validateWorkflowPreflight( }) } + if (node.type === 'forEachNode' && !((node.data.params?.dir as string | undefined)?.trim())) { + pushIssue(issues, { + key: `${node.id}:foreach-no-folder`, + nodeId: node.id, + message: `${nodeLabel(node, allExtensions)} needs a folder selected.`, + }) + } + // A node fed by two different Wait branches can't be scheduled into a single // branch — it would run before either branch produces its mesh. if ( diff --git a/src/areas/workflows/workflowRunStore.ts b/src/areas/workflows/workflowRunStore.ts index a858ec67..df654e51 100644 --- a/src/areas/workflows/workflowRunStore.ts +++ b/src/areas/workflows/workflowRunStore.ts @@ -32,6 +32,8 @@ const _activeJobId = { current: null as string | null } // While container (manual mode) pause/resume — set by continueWhile()/retryWhile(). const _resume = { current: null as (() => void) | null } const _retry = { current: false } +// For Each auto-loop: the user asked to pause at the next iteration boundary. +const _pauseRequested = { current: false } // Live node params — the UI pushes edits here so a looping/paused run re-reads the // latest values when a body node starts (instead of the snapshot from run start). const _liveParams = { current: new Map>() } @@ -48,6 +50,67 @@ function isSceneMeshOutput(output: NodeOutput | undefined): output is NodeOutput return output?.outputType === 'mesh' && typeof output.filePath === 'string' } +// ─── For Each iterator (image / text / mesh) ─────────────────────────────────── +// A "For Each" node walks a folder alphabetically and emits one file per loop +// iteration. Its loop body = the executable nodes reachable downstream, which +// re-run for every file. The `mode` param picks what it emits and which files it +// matches. + +const FOR_EACH_MODES: Record = { + image: { exts: ['png', 'jpg', 'jpeg', 'webp'], outputType: 'image' }, + text: { exts: ['txt', 'md', 'prompt'], outputType: 'text' }, + mesh: { exts: ['glb', 'gltf', 'obj', 'stl', 'ply', 'fbx'], outputType: 'mesh' }, +} + +function iteratorConfig(node: WFNode): { exts: string[]; outputType: 'image' | 'text' | 'mesh' } { + return FOR_EACH_MODES[(node.data.params?.mode as string) ?? 'image'] ?? FOR_EACH_MODES.image +} + +function isIterator(type: string | undefined): boolean { + return type === 'forEachNode' +} + +/** True for nodes the runner executes (and can re-run inside a loop body). */ +function isExecutable(node: WFNode): boolean { + if (isIterator(node.type)) return true + return node.type === 'extensionNode' && !!node.data.enabled +} + +/** Absolute, alphabetically-sorted paths of an iterator's files (listFiles sorts). */ +async function listIteratorFiles(dir: string, exts: string[]): Promise { + const names = await window.electron.fs.listFiles(dir, exts) + const norm = dir.replace(/\\/g, '/').replace(/\/+$/, '') + return names.map((n) => `${norm}/${n}`) +} + +/** Read a UTF-8 text file through the base64 IPC bridge. */ +async function readTextFile(filePath: string): Promise { + const b64 = await window.electron.fs.readFileBase64(filePath) + return new TextDecoder('utf-8').decode(Uint8Array.from(atob(b64), (c) => c.charCodeAt(0))) +} + +/** + * Executable nodes reachable downstream from `startId` (its loop body). Traversal + * stops at Wait boundaries — those nodes belong to branches, not the pre-phase loop. + */ +function reachableExecutable(startId: string, edges: WFEdge[], nodeMap: Map): Set { + const body = new Set([startId]) + const stack = [startId] + const seen = new Set([startId]) + while (stack.length > 0) { + const id = stack.pop()! + for (const e of edges) { + if (e.source !== id || seen.has(e.target)) continue + seen.add(e.target) + const t = nodeMap.get(e.target) + if (!t || isBranchStarter(t.type)) continue + if (isExecutable(t)) body.add(e.target) + stack.push(e.target) + } + } + return body +} + function toWorkspaceUrl(filePath: string, workspaceDir: string): string | undefined { const norm = filePath.replace(/\\/g, '/') if (!norm.startsWith(workspaceDir)) return undefined @@ -70,6 +133,8 @@ interface RunContext { waitIds: string[] /** waitId → nearest upstream waitId (null = top-level, runnable from the start) */ parentWait: Map + /** iterator node id → its resolved file paths, one per loop iteration */ + iteratorFiles: Map /** workspace URL of the most recently pushed scene mesh (last branch the user ran wins) */ lastSceneMesh?: string } @@ -159,7 +224,7 @@ function identifyBranches(workflow: Workflow): { // Wait → … → Wait chains nest: nodes after the 2nd Wait belong to it, not the 1st. const branchOwner = new Map() for (const node of workflow.nodes) { - if (isBranchStarter(node.type)) continue + if (isBranchStarter(node.type) || !isExecutable(node)) continue const nearest = nearestUpstreamWaits(node.id, workflow.edges, nodeMap) if (nearest.size === 1) branchOwner.set(node.id, [...nearest][0]) } @@ -175,7 +240,7 @@ function identifyBranches(workflow: Workflow): { for (const w of waitIds) branches.set(w, []) const preExecExtNodes: WFNode[] = [] for (const node of ordered) { - if (node.type !== 'extensionNode' || !node.data.enabled) continue + if (!isExecutable(node)) continue const owner = branchOwner.get(node.id) if (owner) branches.get(owner)!.push(node) else preExecExtNodes.push(node) @@ -184,6 +249,34 @@ function identifyBranches(workflow: Workflow): { return { preExecExtNodes, branches, waitIds, parentWait, ordered } } +// ─── For Each iterator execution ─────────────────────────────────────────────── +// Emits the current iteration's file. Image iterators emit an image path; text +// iterators read the file and emit its text. The iteration index comes from the +// loop's progress (its own node id keys the loop). + +async function executeIteratorNode( + node: WFNode, + ctx: RunContext, + setRunState: (updater: (s: WorkflowRunState) => WorkflowRunState) => void, +): Promise { + const files = ctx.iteratorFiles.get(node.id) ?? [] + const current = useWorkflowRunStore.getState().whileProgress[node.id]?.current ?? 1 + const path = files[current - 1] + if (!path) throw new Error('For Each: no file for this iteration') + + const kind = iteratorConfig(node) + const name = path.split(/[\\/]/).pop() + setRunState((s) => ({ ...s, blockProgress: 30, blockStep: `Reading ${name}` })) + + if (kind.outputType === 'text') { + const text = await readTextFile(path) + ctx.nodeOutputs.set(node.id, { text, outputType: 'text' }) + } else { + ctx.nodeOutputs.set(node.id, { filePath: path, outputType: kind.outputType }) + } + setRunState((s) => ({ ...s, blockProgress: 100, blockStep: `Loaded ${name}` })) +} + // ─── Per-node execution ────────────────────────────────────────────────────── // Resolves inputs (walking through Wait passthroughs), runs the extension // (model or process), updates nodeOutputs, and pushes the mesh to the scene @@ -194,6 +287,11 @@ async function executeExtensionNode( ctx: RunContext, setRunState: (updater: (s: WorkflowRunState) => WorkflowRunState) => void, ): Promise { + if (isIterator(node.type)) { + await executeIteratorNode(node, ctx, setRunState) + return + } + const { workflow, allExtensions, client, workspaceDir, nodeOutputs, nodeMap, selectedImagePath, selectedImageData } = ctx @@ -394,6 +492,8 @@ interface WorkflowRunStore { runningBranchId: string | null /** whileId → current iteration / total (total null = manual/unbounded) */ whileProgress: Record + /** iterator ids paused together at a shared boundary (lockstep For Each group) */ + pausedGroup: string[] run: (workflow: Workflow, allExtensions: WorkflowExtension[], overrideImageData?: string) => Promise cancel: () => void @@ -403,6 +503,8 @@ interface WorkflowRunStore { continueWhile: () => void /** While container: resume and re-run the loop body once more (Retry) */ retryWhile: () => void + /** For Each container: request a pause at the next file boundary */ + pauseWhile: () => void /** UI → runner: push the latest params for a node so a looping run uses them */ setLiveNodeParams: (nodeId: string, params: Record) => void } @@ -458,6 +560,7 @@ export const useWorkflowRunStore = create((set, get) => { activeNodeId: null, runningBranchId: null, whileProgress: {}, + pausedGroup: [], waitStates: finalWaitStates ?? s.waitStates, nodeImageOutputs: collectImageOutputs(ctx), runState: { @@ -481,9 +584,11 @@ export const useWorkflowRunStore = create((set, get) => { waitStates: {}, runningBranchId: null, whileProgress: {}, + pausedGroup: [], async run(workflow, allExtensions, overrideImageData?) { _cancel.current = false + _pauseRequested.current = false // Seed live params from the snapshot; UI edits during the run override these. _liveParams.current = new Map(workflow.nodes.map((n) => [n.id, { ...(n.data.params ?? {}) }])) @@ -493,15 +598,37 @@ export const useWorkflowRunStore = create((set, get) => { const { preExecExtNodes, branches, waitIds, parentWait, ordered } = identifyBranches(workflow) const branchSteps = waitIds.reduce((acc, w) => acc + (branches.get(w)?.length ?? 0), 0) - // ── While containers → loop over their pre-phase body members ───────────── - // A While's body = the extension nodes parented to it (parentId). They run in - // the pre-phase; we re-run those members either N times (data.iterations) or, - // in manual mode, on Continue/Retry. The body need not be contiguous in topo - // order, so re-runs filter by parentId (not an index range) to avoid replaying - // unrelated pre-phase nodes that happen to sort between body members. Body - // nodes gated behind a Wait land in a branch instead and are out of scope. - interface LoopInfo { whileId: string; firstIdx: number; lastIdx: number; bodyIds: Set; iterations: number | null } + const nodeMap = new Map(workflow.nodes.map((n) => [n.id, n])) + + // ── For Each iterators → resolve their folders up front ──────────────────── + // The loop count is driven by the folder contents, so the listing must resolve + // before the loop table (and its progress totals) below. + const iteratorFiles = new Map() + for (const w of workflow.nodes) { + if (!isIterator(w.type)) continue + const dir = (w.data.params?.dir as string | undefined)?.trim() + const fail = (msg: string, step: string): void => { + set((s) => ({ runState: { ...s.runState, status: 'error', error: msg, blockStep: step }, activeNodeId: null })) + } + if (!dir) { fail('For Each: pick a folder first', 'No folder selected'); return } + try { + const files = await listIteratorFiles(dir, iteratorConfig(w).exts) + if (files.length === 0) { fail(`For Each: no matching files in ${dir}`, 'Empty folder'); return } + iteratorFiles.set(w.id, files) + } catch (err) { + fail(String(err), 'Failed to read folder'); return + } + } + + // ── Loop table ───────────────────────────────────────────────────────────── + // While containers loop their geometric body N× (or manually). For Each + // iterators loop the executable nodes reachable downstream, once per file. + // Replays filter by bodyIds membership (not a contiguous range), so unrelated + // pre-phase nodes sorting between body members aren't replayed. + interface LoopInfo { whileId: string; kind: 'while' | 'forEach'; firstIdx: number; lastIdx: number; bodyIds: Set; iterations: number | null } const loops: LoopInfo[] = [] + const indexOf = new Map(preExecExtNodes.map((n, i) => [n.id, i])) + for (const w of workflow.nodes) { if (w.type !== 'whileNode') continue const bounds = whileBounds(w) @@ -512,17 +639,54 @@ export const useWorkflowRunStore = create((set, get) => { if (idxs.length === 0) continue const iters = Number(w.data?.iterations) loops.push({ - whileId: w.id, + whileId: w.id, + kind: 'while', + firstIdx: Math.min(...idxs), + lastIdx: Math.max(...idxs), + bodyIds: new Set(idxs.map((i) => preExecExtNodes[i].id)), + iterations: Number.isFinite(iters) && iters > 0 ? Math.floor(iters) : null, + }) + } + + for (const [iterId, files] of iteratorFiles) { + const bodyIds = new Set([...reachableExecutable(iterId, workflow.edges, nodeMap)].filter((id) => indexOf.has(id))) + const idxs = [...bodyIds].map((id) => indexOf.get(id)!) + if (idxs.length === 0) continue + loops.push({ + whileId: iterId, + kind: 'forEach', firstIdx: Math.min(...idxs), lastIdx: Math.max(...idxs), - bodyIds: new Set(idxs.map((i) => preExecExtNodes[i].id)), - iterations: Number.isFinite(iters) && iters > 0 ? Math.floor(iters) : null, + bodyIds, + iterations: files.length, // one pass per file }) } const loopCounters = new Map(loops.map((l) => [l.whileId, l.iterations])) // Auto-mode loops replay their body N times; count the extra passes so the - // progress total reflects the real work (manual loops stay unbounded). - const loopExtraSteps = loops.reduce((acc, l) => acc + (l.iterations != null ? (l.iterations - 1) * l.bodyIds.size : 0), 0) + // progress total reflects the real work (manual loops stay unbounded). While + // loops are independent. For Each loops sharing a boundary (same lastIdx) run + // in lockstep over the union of their bodies, so that union is replayed once + // per pass — count it once, for max(files) − 1 extra passes. + const forEachGroups = new Map() + let loopExtraSteps = 0 + for (const l of loops) { + if (l.kind === 'while') { + loopExtraSteps += l.iterations != null ? (l.iterations - 1) * l.bodyIds.size : 0 + } else { + const g = forEachGroups.get(l.lastIdx) ?? [] + g.push(l) + forEachGroups.set(l.lastIdx, g) + } + } + for (const group of forEachGroups.values()) { + const union = new Set() + let maxIter = 0 + for (const l of group) { + l.bodyIds.forEach((id) => union.add(id)) + if (l.iterations != null) maxIter = Math.max(maxIter, l.iterations) + } + if (maxIter > 0) loopExtraSteps += (maxIter - 1) * union.size + } const totalSteps = preExecExtNodes.length + branchSteps + loopExtraSteps const selectedImagePath = appState.selectedImagePath ?? undefined @@ -536,6 +700,7 @@ export const useWorkflowRunStore = create((set, get) => { waitStates: Object.fromEntries(waitIds.map((id) => [id, parentWait.get(id) ? 'blocked' as WaitState : 'pending' as WaitState])), runningBranchId: null, whileProgress: Object.fromEntries(loops.map((l) => [l.whileId, { current: 1, total: l.iterations }])), + pausedGroup: [], runState: { status: 'running', blockIndex: 0, blockTotal: totalSteps, blockProgress: 0, blockStep: 'Starting…', @@ -559,7 +724,6 @@ export const useWorkflowRunStore = create((set, get) => { window.electron.fs.deleteDirectory(tmpAbsPath).catch(() => {}) const nodeOutputs = new Map() - const nodeMap = new Map(workflow.nodes.map((n) => [n.id, n])) // Pre-populate source nodes for (const node of ordered) { @@ -592,14 +756,15 @@ export const useWorkflowRunStore = create((set, get) => { const ctx: RunContext = { workflow, allExtensions, client, workspaceDir, selectedImagePath, selectedImageData, - overrideImageData, nodeOutputs, nodeMap, ordered, branches, waitIds, parentWait, + overrideImageData, nodeOutputs, nodeMap, ordered, branches, waitIds, parentWait, iteratorFiles, } _ctx.current = ctx - // While the runner is replaying a loop body, this holds the active While id; - // re-iterations then execute only that While's members and skip everything - // else in the range. null = first pass / no active loop (run all nodes once). - let activeLoopId: string | null = null + // While the runner is replaying a loop body, this holds the body nodes of the + // active loop (or the union of several For Each loops sharing a boundary); + // re-iterations then execute only those members and skip everything else in + // the range. null = first pass / no active loop (run all nodes once). + let activeLoopBody: Set | null = null // End-of-body handler for While containers. Called after each pre-phase node; // when the index is a loop's last body node, it either jumps back (auto N× or @@ -612,22 +777,76 @@ export const useWorkflowRunStore = create((set, get) => { ) const handleLoopEnd = async (idx: number): Promise => { - const loop = loops.find((l) => l.lastIdx === idx) - if (!loop) return undefined - const remaining = loopCounters.get(loop.whileId) + const forEachLoops = loops.filter((l) => l.lastIdx === idx && l.kind === 'forEach') + const whileLoop = loops.find((l) => l.lastIdx === idx && l.kind === 'while') + + // ── For Each: run through every file automatically. Several iterators can + // share the same downstream body (e.g. an image folder + a mesh folder both + // feeding one node); they advance together, in lockstep. It only stops if + // the user hit Pause; then Continue advances to the next file(s) and Retry + // re-runs the current one. No forced pause at the end — it just finishes. + if (forEachLoops.length > 0) { + const groupBody = new Set() + let jumpTo = Infinity + for (const loop of forEachLoops) { + loop.bodyIds.forEach((id) => groupBody.add(id)) + jumpTo = Math.min(jumpTo, loop.firstIdx) + } + + if (_pauseRequested.current) { + _pauseRequested.current = false + _retry.current = false + // Pause every iterator of the group together (they show the same state). + const groupIds = forEachLoops.map((l) => l.whileId) + set({ activeNodeId: groupIds[0], runningBranchId: groupIds[0], pausedGroup: groupIds }) + setRunState((s) => ({ ...s, status: 'paused', blockStep: 'Paused — Continue or Retry' })) + await new Promise((resolve) => { _resume.current = resolve }) + if (_cancel.current) return 'cancel' + set({ runningBranchId: null, pausedGroup: [] }) + setRunState((s) => ({ ...s, status: 'running' })) + if (_retry.current) { // re-run the current file(s), no advance + _retry.current = false + activeLoopBody = groupBody + return jumpTo + } + // Continue → fall through to the normal advance below + } + + // Advance every iterator that still has files left; they move together. + let anyMore = false + for (const loop of forEachLoops) { + const remaining = loopCounters.get(loop.whileId) + if (remaining != null && remaining > 1) { + loopCounters.set(loop.whileId, remaining - 1) + bumpWhileProgress(loop.whileId) + anyMore = true + } + } + if (anyMore) { + setRunState((s) => ({ ...s, blockStep: 'Next file…' })) + activeLoopBody = groupBody + return jumpTo + } + activeLoopBody = null + return undefined + } + + if (!whileLoop) return undefined + const remaining = loopCounters.get(whileLoop.whileId) + // Auto mode with iterations left → loop back automatically. if (remaining != null && remaining > 1) { - loopCounters.set(loop.whileId, remaining - 1) - bumpWhileProgress(loop.whileId) + loopCounters.set(whileLoop.whileId, remaining - 1) + bumpWhileProgress(whileLoop.whileId) setRunState((s) => ({ ...s, blockStep: `Looping… ${remaining - 1} left` })) - activeLoopId = loop.whileId - return loop.firstIdx + activeLoopBody = whileLoop.bodyIds + return whileLoop.firstIdx } // Otherwise the auto counter is exhausted (or it's manual mode): pause on // the While and wait for Continue (proceed) or Retry (run the body again). // runningBranchId blocks Wait branches while the pre-phase is parked here. _retry.current = false - set({ activeNodeId: loop.whileId, runningBranchId: loop.whileId }) + set({ activeNodeId: whileLoop.whileId, runningBranchId: whileLoop.whileId }) setRunState((s) => ({ ...s, status: 'paused', blockStep: 'Loop finished — Continue or Retry' })) await new Promise((resolve) => { _resume.current = resolve }) if (_cancel.current) return 'cancel' @@ -635,11 +854,11 @@ export const useWorkflowRunStore = create((set, get) => { setRunState((s) => ({ ...s, status: 'running' })) if (_retry.current) { _retry.current = false - bumpWhileProgress(loop.whileId) - activeLoopId = loop.whileId - return loop.firstIdx + bumpWhileProgress(whileLoop.whileId) + activeLoopBody = whileLoop.bodyIds + return whileLoop.firstIdx } - activeLoopId = null // Continue → resume normal forward execution + activeLoopBody = null // Continue → resume normal forward execution return undefined } @@ -648,11 +867,8 @@ export const useWorkflowRunStore = create((set, get) => { for (let i = 0; i < preExecExtNodes.length; i++) { if (_cancel.current) { _ctx.current = null; set({ runState: IDLE, activeNodeId: null }); return } const node = preExecExtNodes[i] - // During a loop replay, only re-run that While's own body members. - if (activeLoopId) { - const body = loops.find((l) => l.whileId === activeLoopId)?.bodyIds - if (body && !body.has(node.id)) continue - } + // During a loop replay, only re-run the active loop's body members. + if (activeLoopBody && !activeLoopBody.has(node.id)) continue set((s) => ({ activeNodeId: node.id, runState: { ...s.runState, blockIndex: stepsDone, blockProgress: 0, blockStep: 'Starting…' }, @@ -776,6 +992,7 @@ export const useWorkflowRunStore = create((set, get) => { cancel() { _cancel.current = true + _pauseRequested.current = false flushResume() // unblock a manual While pause so the run can tear down if (_activeJobId.current) { const apiUrl = useAppStore.getState().apiUrl @@ -783,13 +1000,13 @@ export const useWorkflowRunStore = create((set, get) => { _activeJobId.current = null } _ctx.current = null - set({ runState: IDLE, activeNodeId: null, activeWorkflowId: null, nodeImageOutputs: {}, waitStates: {}, runningBranchId: null, whileProgress: {} }) + set({ runState: IDLE, activeNodeId: null, activeWorkflowId: null, nodeImageOutputs: {}, waitStates: {}, runningBranchId: null, whileProgress: {}, pausedGroup: [] }) useAppStore.getState().setCurrentJob(null) }, reset() { _ctx.current = null - set({ runState: IDLE, activeNodeId: null, activeWorkflowId: null, nodeImageOutputs: {}, waitStates: {}, runningBranchId: null, whileProgress: {} }) + set({ runState: IDLE, activeNodeId: null, activeWorkflowId: null, nodeImageOutputs: {}, waitStates: {}, runningBranchId: null, whileProgress: {}, pausedGroup: [] }) }, continueWhile() { @@ -801,6 +1018,10 @@ export const useWorkflowRunStore = create((set, get) => { flushResume() }, + pauseWhile() { + _pauseRequested.current = true + }, + setLiveNodeParams(nodeId, params) { _liveParams.current.set(nodeId, params) }, diff --git a/src/shared/stores/workflowsStore.ts b/src/shared/stores/workflowsStore.ts index c3a65a9c..9cd28a3c 100644 --- a/src/shared/stores/workflowsStore.ts +++ b/src/shared/stores/workflowsStore.ts @@ -38,7 +38,7 @@ interface LegacyWorkflow { // Source-only nodes have no target handle; sink-only nodes have no source handle. // An edge into/out of the wrong side can't resolve a handle and makes React Flow // warn ("Couldn't create edge for target handle id: null") on every render. -export const NODE_TYPES_WITHOUT_TARGET = new Set(['imageNode', 'textNode', 'meshNode', 'inputNode']) +export const NODE_TYPES_WITHOUT_TARGET = new Set(['imageNode', 'textNode', 'meshNode', 'inputNode', 'forEachNode']) export const NODE_TYPES_WITHOUT_SOURCE = new Set(['outputNode', 'previewNode']) function sanitizeEdges(nodes: WFNode[], edges: WFEdge[]): WFEdge[] { diff --git a/src/shared/types/electron.d.ts b/src/shared/types/electron.d.ts index ca35e853..7e500b85 100644 --- a/src/shared/types/electron.d.ts +++ b/src/shared/types/electron.d.ts @@ -157,6 +157,8 @@ declare global { selectDirectory: (defaultPath?: string) => Promise savePath: (args: { filters: { name: string; extensions: string[] }[]; defaultPath?: string }) => Promise listDir: (dirPath: string) => Promise + listFiles: (dirPath: string, extensions?: string[]) => Promise + selectTextFile: () => Promise moveDirectory: (args: { src: string; dest: string }) => Promise<{ success: boolean; error?: string }> deleteDirectory: (dirPath: string) => Promise<{ success: boolean; error?: string }> readScreenshotDataUrl: (filename: string) => Promise