diff --git a/docs/plans/2026-02-03-feat-parallel-workflow-execution-plan.md b/docs/plans/2026-02-03-feat-parallel-workflow-execution-plan.md new file mode 100644 index 00000000..93abf4fe --- /dev/null +++ b/docs/plans/2026-02-03-feat-parallel-workflow-execution-plan.md @@ -0,0 +1,453 @@ +--- +title: "feat: Parallel Workflow Execution" +type: feat +date: 2026-02-03 +--- + +# Parallel Workflow Execution + +## Overview + +Add parallel execution of independent nodes in workflow pipelines to reduce total execution time. Currently, all nodes execute sequentially even when they have no dependencies on each other. This wastes significant time when multiple image generation calls (5-30s each) could run simultaneously. + +## Problem Statement + +**Current behavior:** Workflow with 3 independent Generate nodes takes ~60-90 seconds (3 × 20s average) + +**Desired behavior:** Same workflow completes in ~20-25 seconds (parallel execution + overhead) + +Example workflow that would benefit: +``` +prompt-1 → nanoBanana-1 ──┐ +prompt-2 → nanoBanana-2 ──┼─→ output +prompt-3 → nanoBanana-3 ──┘ +``` + +Currently executes: prompt-1 → prompt-2 → prompt-3 → nanoBanana-1 (wait) → nanoBanana-2 (wait) → nanoBanana-3 (wait) → output + +With parallel: [prompt-1, prompt-2, prompt-3] → [nanoBanana-1, nanoBanana-2, nanoBanana-3] → output + +## Proposed Solution + +### Core Changes + +1. **Level-based topological sort** - Group nodes by dependency depth +2. **Parallel level execution** - Execute each level with `Promise.all` (respecting concurrency limit) +3. **Configurable concurrency** - Default 3 concurrent API calls, adjustable 1-10 +4. **Fail-fast with AbortController** - Cancel sibling requests on first failure + +### Architecture + +``` +┌─────────────────────────────────────────────────────────────┐ +│ executeWorkflow() │ +├─────────────────────────────────────────────────────────────┤ +│ 1. groupNodesByLevel(nodes, edges) │ +│ └─→ Map │ +│ │ +│ 2. for each level (sequential): │ +│ └─→ executeLevel(nodes, concurrencyLimit, abortSignal) │ +│ └─→ Promise.all with chunked batches │ +│ │ +│ 3. On any failure: │ +│ └─→ abortController.abort() │ +│ └─→ Set sibling nodes to 'idle' │ +│ └─→ Show error toast for failed node │ +└─────────────────────────────────────────────────────────────┘ +``` + +## Technical Approach + +### 1. Level Grouping Algorithm + +Replace current linear topological sort with level-aware grouping: + +```typescript +// src/store/workflowStore.ts + +interface LevelGroup { + level: number; + nodeIds: string[]; +} + +function groupNodesByLevel( + nodes: WorkflowNode[], + edges: Edge[] +): LevelGroup[] { + // Calculate in-degree for each node + const inDegree = new Map(); + const adjList = new Map(); + + nodes.forEach(n => { + inDegree.set(n.id, 0); + adjList.set(n.id, []); + }); + + edges.forEach(e => { + inDegree.set(e.target, (inDegree.get(e.target) || 0) + 1); + adjList.get(e.source)?.push(e.target); + }); + + // BFS with level tracking (Kahn's algorithm variant) + const levels: LevelGroup[] = []; + let currentLevel = nodes + .filter(n => inDegree.get(n.id) === 0) + .map(n => n.id); + + let levelNum = 0; + while (currentLevel.length > 0) { + levels.push({ level: levelNum, nodeIds: [...currentLevel] }); + + const nextLevel: string[] = []; + for (const nodeId of currentLevel) { + for (const child of adjList.get(nodeId) || []) { + const newDegree = (inDegree.get(child) || 1) - 1; + inDegree.set(child, newDegree); + if (newDegree === 0) { + nextLevel.push(child); + } + } + } + + currentLevel = nextLevel; + levelNum++; + } + + return levels; +} +``` + +### 2. State Changes + +```typescript +// src/store/workflowStore.ts - WorkflowState interface + +interface WorkflowState { + // Change from single to array + currentNodeIds: string[]; // was: currentNodeId: string | null + + // New: concurrency setting + maxConcurrentCalls: number; // default: 3 + + // New: abort controller reference (not persisted) + _abortController: AbortController | null; +} +``` + +### 3. Execution Loop Changes + +```typescript +// src/store/workflowStore.ts - executeWorkflow() + +executeWorkflow: async (startFromNodeId?: string) => { + const { nodes, edges, maxConcurrentCalls } = get(); + + // Create abort controller for this run + const abortController = new AbortController(); + set({ _abortController: abortController, isRunning: true, currentNodeIds: [] }); + + try { + const levels = groupNodesByLevel(nodes, edges); + + // Find starting level if startFromNodeId specified + let startLevel = 0; + if (startFromNodeId) { + startLevel = levels.findIndex(l => l.nodeIds.includes(startFromNodeId)); + if (startLevel === -1) startLevel = 0; + } + + // Execute levels sequentially + for (let i = startLevel; i < levels.length; i++) { + if (abortController.signal.aborted) break; + + const level = levels[i]; + const executableNodes = level.nodeIds + .map(id => nodes.find(n => n.id === id)!) + .filter(n => !isNodeLocked(n)); // Skip locked nodes + + if (executableNodes.length === 0) continue; + + // Check for pause edges targeting this level + const pauseNode = executableNodes.find(n => hasPauseEdge(n, edges)); + if (pauseNode) { + set({ pausedAtNodeId: pauseNode.id, isRunning: false }); + return; + } + + // Execute level with concurrency limit + await executeLevelParallel( + executableNodes, + maxConcurrentCalls, + abortController.signal + ); + } + } catch (error) { + // Fail-fast triggered + abortController.abort(); + const failedNodeId = get().currentNodeIds[0]; // First to fail + set({ + currentNodeIds: [], + isRunning: false + }); + toast.error(`Workflow failed at node: ${failedNodeId}`); + } finally { + set({ _abortController: null, currentNodeIds: [], isRunning: false }); + } +} +``` + +### 4. Parallel Level Execution + +```typescript +// src/store/workflowStore.ts + +async function executeLevelParallel( + nodes: WorkflowNode[], + concurrencyLimit: number, + signal: AbortSignal +): Promise { + const { set, get } = useWorkflowStore.getState(); + + // Chunk nodes into batches respecting concurrency limit + const batches = chunk(nodes, concurrencyLimit); + + for (const batch of batches) { + if (signal.aborted) throw new Error('Aborted'); + + // Mark batch nodes as executing + const batchIds = batch.map(n => n.id); + set({ currentNodeIds: batchIds }); + batch.forEach(n => updateNodeData(n.id, { status: 'loading' })); + + // Execute batch in parallel + const results = await Promise.all( + batch.map(node => executeNode(node, signal)) + ); + + // Check for failures (fail-fast) + const failed = results.find(r => r.error); + if (failed) { + throw failed.error; + } + + // Mark batch as complete + batch.forEach(n => updateNodeData(n.id, { status: 'complete' })); + } +} +``` + +### 5. AbortController Integration + +Update API fetch calls to respect abort signal: + +```typescript +// src/store/workflowStore.ts - inside node execution + +const response = await fetch('/api/generate', { + method: 'POST', + body: JSON.stringify(payload), + signal: abortController.signal, // NEW +}); +``` + +### 6. UI Updates + +#### BaseNode.tsx - Multiple execution indicators + +```typescript +// src/components/nodes/BaseNode.tsx + +const currentNodeIds = useWorkflowStore((state) => state.currentNodeIds); +const isCurrentlyExecuting = currentNodeIds.includes(id); +``` + +#### FloatingActionBar.tsx - Parallel progress + +```typescript +// src/components/FloatingActionBar.tsx + +const currentNodeIds = useWorkflowStore((state) => state.currentNodeIds); +const runningCount = currentNodeIds.length; + +// Display: "Running 3 nodes..." or "Running Generate1..." +const statusText = runningCount > 1 + ? `Running ${runningCount} nodes...` + : `Running ${getNodeLabel(currentNodeIds[0])}...`; +``` + +#### Settings - Concurrency control + +Add to existing settings UI: + +```typescript +// In settings modal/panel + + +``` + +## Acceptance Criteria + +### Functional Requirements + +- [x] Independent nodes at the same dependency level execute in parallel +- [x] Concurrency limit (default 3) controls max simultaneous API calls +- [x] Fail-fast: first failure aborts sibling requests and stops workflow +- [x] Cancellation (Stop button) aborts all in-flight requests +- [x] Locked nodes are skipped without blocking parallel siblings +- [x] Pause edges halt execution before their target level +- [x] "Run from node" includes parallel siblings at same level +- [ ] Cost tracking works correctly for parallel calls + +### Non-Functional Requirements + +- [x] No race conditions in state updates +- [x] Memory usage stays reasonable (no leaked promises/controllers) +- [ ] Existing sequential workflows behave identically + +### UI Requirements + +- [x] Multiple nodes show blue execution border simultaneously +- [x] FloatingActionBar shows "Running N nodes..." for parallel execution +- [x] Concurrency setting accessible in settings panel +- [x] Error toast identifies which node failed + +## Implementation Phases + +### Phase 1: Core Execution Engine + +**Files to modify:** +- `src/store/workflowStore.ts` - Level grouping, parallel execution loop +- `src/types/index.ts` - Update WorkflowState interface + +**Tasks:** +1. Implement `groupNodesByLevel()` algorithm +2. Change `currentNodeId` to `currentNodeIds: string[]` +3. Add `maxConcurrentCalls` to state with default 3 +4. Refactor `executeWorkflow()` for level-based execution +5. Add `executeLevelParallel()` helper +6. Integrate AbortController + +### Phase 2: API Integration + +**Files to modify:** +- `src/store/workflowStore.ts` - All fetch calls +- `src/app/api/generate/route.ts` - Verify abort handling +- `src/app/api/llm/route.ts` - Verify abort handling + +**Tasks:** +1. Pass abort signal to all fetch calls +2. Handle AbortError gracefully (don't treat as failure) +3. Verify API routes handle client disconnection + +### Phase 3: UI Updates + +**Files to modify:** +- `src/components/nodes/BaseNode.tsx` - Execution indicator +- `src/components/FloatingActionBar.tsx` - Progress display +- `src/components/SettingsPanel.tsx` (or wherever settings live) + +**Tasks:** +1. Update BaseNode to check `currentNodeIds.includes(id)` +2. Update FloatingActionBar for multi-node status +3. Add concurrency slider to settings +4. Store concurrency preference in localStorage + +### Phase 4: Edge Cases & Testing + +**Tasks:** +1. Test locked groups with parallel siblings +2. Test pause edges at level boundaries +3. Test "Run from node" with parallel siblings +4. Test cancellation mid-parallel-execution +5. Test fail-fast with multiple simultaneous failures +6. Test cost tracking accuracy +7. Performance test with 10+ parallel nodes + +## Files to Modify + +| File | Changes | +|------|---------| +| `src/store/workflowStore.ts` | Core execution refactor, new state fields | +| `src/types/index.ts` | WorkflowState interface updates | +| `src/components/nodes/BaseNode.tsx` | Multi-node execution indicator | +| `src/components/FloatingActionBar.tsx` | Parallel progress display | +| `src/components/SettingsPanel.tsx` | Concurrency setting UI | + +## Migration Notes + +- `currentNodeId` → `currentNodeIds` is a breaking change for any code reading this state +- Search codebase for `currentNodeId` references and update all +- localStorage key for concurrency: `node-banana-concurrency-limit` + +## Testing Strategy + +### Unit Tests + +```typescript +describe('groupNodesByLevel', () => { + it('groups independent nodes at level 0', () => { + const nodes = [nodeA, nodeB, nodeC]; // no edges + const levels = groupNodesByLevel(nodes, []); + expect(levels).toEqual([{ level: 0, nodeIds: ['a', 'b', 'c'] }]); + }); + + it('respects dependencies across levels', () => { + // A → C, B → C + const levels = groupNodesByLevel(nodes, edges); + expect(levels[0].nodeIds).toContain('a'); + expect(levels[0].nodeIds).toContain('b'); + expect(levels[1].nodeIds).toEqual(['c']); + }); +}); + +describe('parallel execution', () => { + it('executes independent nodes simultaneously', async () => { + const startTimes: number[] = []; + // Mock API to record call times + // Verify nodes started within 100ms of each other + }); + + it('respects concurrency limit', async () => { + // 5 nodes, limit 2 + // Verify max 2 concurrent at any time + }); + + it('fails fast on first error', async () => { + // Node A fails after 100ms, Node B would take 1000ms + // Verify B is aborted, total time < 200ms + }); +}); +``` + +### Integration Tests + +- Run real workflow with mocked API delays +- Verify timing improvements +- Verify correct outputs + +## Open Questions (Resolved) + +| Question | Resolution | +|----------|------------| +| Fail-fast vs continue? | Fail-fast (user preference) | +| Concurrency limit? | Configurable, default 3 | +| Sibling status on abort? | Set to 'idle' (not 'error') | +| Pause edges within levels? | Pause entire level (simplest) | +| "Run from node" with siblings? | Include siblings at same level | + +## References + +- Current execution: `src/store/workflowStore.ts:924-1919` +- Topological sort: `src/store/workflowStore.ts:945-969` +- Node types: `src/types/index.ts` +- API routes: `src/app/api/generate/route.ts`, `src/app/api/llm/route.ts` diff --git a/src/components/FloatingActionBar.tsx b/src/components/FloatingActionBar.tsx index e02d767b..1fce7fd0 100644 --- a/src/components/FloatingActionBar.tsx +++ b/src/components/FloatingActionBar.tsx @@ -182,6 +182,7 @@ export function FloatingActionBar() { const { nodes, isRunning, + currentNodeIds, executeWorkflow, regenerateNode, stopWorkflow, @@ -193,6 +194,18 @@ export function FloatingActionBar() { modelSearchOpen, modelSearchProvider, } = useWorkflowStore(); + + // Get display text for running nodes + const runningNodeCount = currentNodeIds.length; + const getRunningLabel = () => { + if (runningNodeCount === 0) return "Running..."; + if (runningNodeCount === 1) { + const node = nodes.find((n) => n.id === currentNodeIds[0]); + const nodeName = node?.data?.customTitle || node?.type || "node"; + return `Running ${nodeName}...`; + } + return `Running ${runningNodeCount} nodes...`; + }; const [runMenuOpen, setRunMenuOpen] = useState(false); const runMenuRef = useRef(null); const [mounted, setMounted] = useState(false); @@ -341,7 +354,9 @@ export function FloatingActionBar() { d="M4 12a8 8 0 018-8V0C5.373 0 0 5.373 0 12h4zm2 5.291A7.962 7.962 0 014 12H0c0 3.042 1.135 5.824 3 7.938l3-2.647z" /> - Stop + + {runningNodeCount > 1 ? `${runningNodeCount} nodes` : "Stop"} + ) : ( <> diff --git a/src/components/ProjectSetupModal.tsx b/src/components/ProjectSetupModal.tsx index 3b0f4a79..b708812c 100644 --- a/src/components/ProjectSetupModal.tsx +++ b/src/components/ProjectSetupModal.tsx @@ -82,6 +82,8 @@ export function ProjectSetupModal({ providerSettings, updateProviderApiKey, toggleProvider, + maxConcurrentCalls, + setMaxConcurrentCalls, } = useWorkflowStore(); // Tab state @@ -865,6 +867,33 @@ export function ProjectSetupModal({ + {/* Execution Section */} +
+
+ Execution Settings + + {/* Concurrency slider */} +
+ + setMaxConcurrentCalls(parseInt(e.target.value, 10))} + className="flex-1 h-1 bg-neutral-700 rounded-lg appearance-none cursor-pointer accent-neutral-400" + /> +
+

+ Maximum number of nodes to execute in parallel during workflow execution. + Higher values may improve speed but increase API rate limit risk. +

+
+
+

These defaults are applied when creating nodes via keyboard shortcuts (Shift+G, Shift+L, etc).

diff --git a/src/components/__tests__/AnnotationNode.test.tsx b/src/components/__tests__/AnnotationNode.test.tsx index f32132a4..e8061bc2 100644 --- a/src/components/__tests__/AnnotationNode.test.tsx +++ b/src/components/__tests__/AnnotationNode.test.tsx @@ -73,7 +73,7 @@ describe("AnnotationNode", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { updateNodeData: mockUpdateNodeData, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/BaseNode.test.tsx b/src/components/__tests__/BaseNode.test.tsx index 36f23496..741484d2 100644 --- a/src/components/__tests__/BaseNode.test.tsx +++ b/src/components/__tests__/BaseNode.test.tsx @@ -36,7 +36,7 @@ describe("BaseNode", () => { // Default mock implementation mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: null, @@ -419,7 +419,7 @@ describe("BaseNode", () => { it("should not render lock badge when node is not in a locked group", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: { "group-1": { id: "group-1", locked: false } }, nodes: [{ id: "test-node-1", groupId: "group-1" }], }; @@ -440,7 +440,7 @@ describe("BaseNode", () => { it("should render lock badge when node is in a locked group", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: { "group-1": { id: "group-1", locked: true } }, nodes: [{ id: "test-node-1", groupId: "group-1" }], }; @@ -461,7 +461,7 @@ describe("BaseNode", () => { it("should not render lock badge when node has no groupId", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: { "group-1": { id: "group-1", locked: true } }, nodes: [{ id: "test-node-1" }], // No groupId }; @@ -487,7 +487,7 @@ describe("BaseNode", () => { it("should not render navigation arrows when commentNavigation is not provided even when focused", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: "test-node-1", // This node is focused @@ -510,7 +510,7 @@ describe("BaseNode", () => { it("should render navigation arrows in tooltip when focused, commentNavigation provided, and comment exists", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: "test-node-1", // This node is focused @@ -544,7 +544,7 @@ describe("BaseNode", () => { it("should render index indicator in tooltip showing current position when focused", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: "test-node-1", @@ -577,7 +577,7 @@ describe("BaseNode", () => { it("should call onPrevious when previous button is clicked in tooltip", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: "test-node-1", @@ -614,7 +614,7 @@ describe("BaseNode", () => { it("should call onNext when next button is clicked in tooltip", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: "test-node-1", @@ -651,7 +651,7 @@ describe("BaseNode", () => { it("should not render arrows when no comment exists even when focused with commentNavigation prop", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], focusedCommentNodeId: "test-node-1", @@ -706,10 +706,10 @@ describe("BaseNode", () => { expect(nodeDiv).toBeInTheDocument(); }); - it("should apply executing styling when currentNodeId matches", () => { + it("should apply executing styling when currentNodeIds includes the node", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { - currentNodeId: "test-node-1", + currentNodeIds: ["test-node-1"], groups: {}, nodes: [], }; diff --git a/src/components/__tests__/GenerateImageNode.test.tsx b/src/components/__tests__/GenerateImageNode.test.tsx index e5dd7928..e55f5071 100644 --- a/src/components/__tests__/GenerateImageNode.test.tsx +++ b/src/components/__tests__/GenerateImageNode.test.tsx @@ -95,7 +95,7 @@ describe("GenerateImageNode", () => { providerSettings: defaultProviderSettings, generationsPath: "/test/generations", isRunning: false, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], recentModels: [], @@ -446,7 +446,7 @@ describe("GenerateImageNode", () => { providerSettings: defaultProviderSettings, generationsPath: "/test/generations", isRunning: true, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), @@ -544,7 +544,7 @@ describe("GenerateImageNode", () => { }, generationsPath: "/test/generations", isRunning: false, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/GenerateVideoNode.test.tsx b/src/components/__tests__/GenerateVideoNode.test.tsx index a36317ff..8e49500e 100644 --- a/src/components/__tests__/GenerateVideoNode.test.tsx +++ b/src/components/__tests__/GenerateVideoNode.test.tsx @@ -94,7 +94,7 @@ describe("GenerateVideoNode", () => { providerSettings: defaultProviderSettings, generationsPath: "/test/generations", isRunning: false, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], recentModels: [], @@ -255,7 +255,7 @@ describe("GenerateVideoNode", () => { }, generationsPath: "/test/generations", isRunning: false, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), @@ -568,7 +568,7 @@ describe("GenerateVideoNode", () => { providerSettings: defaultProviderSettings, generationsPath: "/test/generations", isRunning: true, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/ImageInputNode.test.tsx b/src/components/__tests__/ImageInputNode.test.tsx index e21a5bb3..07e12a1f 100644 --- a/src/components/__tests__/ImageInputNode.test.tsx +++ b/src/components/__tests__/ImageInputNode.test.tsx @@ -10,7 +10,7 @@ vi.mock("@/store/workflowStore", () => ({ useWorkflowStore: vi.fn((selector) => { const state = { updateNodeData: mockUpdateNodeData, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/LLMGenerateNode.test.tsx b/src/components/__tests__/LLMGenerateNode.test.tsx index d32d4d09..9d9996a4 100644 --- a/src/components/__tests__/LLMGenerateNode.test.tsx +++ b/src/components/__tests__/LLMGenerateNode.test.tsx @@ -33,7 +33,7 @@ describe("LLMGenerateNode", () => { updateNodeData: mockUpdateNodeData, regenerateNode: mockRegenerateNode, isRunning: false, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), @@ -318,7 +318,7 @@ describe("LLMGenerateNode", () => { updateNodeData: mockUpdateNodeData, regenerateNode: mockRegenerateNode, isRunning: true, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/OutputNode.test.tsx b/src/components/__tests__/OutputNode.test.tsx index 97b7ff24..9c0e3f07 100644 --- a/src/components/__tests__/OutputNode.test.tsx +++ b/src/components/__tests__/OutputNode.test.tsx @@ -39,7 +39,7 @@ describe("OutputNode", () => { mockUseWorkflowStore.mockImplementation((selector) => { const state = { updateNodeData: mockUpdateNodeData, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/PromptNode.test.tsx b/src/components/__tests__/PromptNode.test.tsx index eb402d3a..871c10ee 100644 --- a/src/components/__tests__/PromptNode.test.tsx +++ b/src/components/__tests__/PromptNode.test.tsx @@ -14,7 +14,7 @@ vi.mock("@/store/workflowStore", () => ({ updateNodeData: mockUpdateNodeData, incrementModalCount: mockIncrementModalCount, decrementModalCount: mockDecrementModalCount, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], edges: [], diff --git a/src/components/__tests__/SplitGridNode.test.tsx b/src/components/__tests__/SplitGridNode.test.tsx index 32333611..6cd39066 100644 --- a/src/components/__tests__/SplitGridNode.test.tsx +++ b/src/components/__tests__/SplitGridNode.test.tsx @@ -48,7 +48,7 @@ describe("SplitGridNode", () => { updateNodeData: mockUpdateNodeData, regenerateNode: mockRegenerateNode, isRunning: false, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), @@ -280,7 +280,7 @@ describe("SplitGridNode", () => { updateNodeData: mockUpdateNodeData, regenerateNode: mockRegenerateNode, isRunning: true, - currentNodeId: null, + currentNodeIds: [], groups: {}, nodes: [], getNodesWithComments: vi.fn(() => []), diff --git a/src/components/__tests__/WorkflowCanvas.test.tsx b/src/components/__tests__/WorkflowCanvas.test.tsx index 2fccc937..f424d0f3 100644 --- a/src/components/__tests__/WorkflowCanvas.test.tsx +++ b/src/components/__tests__/WorkflowCanvas.test.tsx @@ -131,7 +131,7 @@ const createDefaultState = (overrides = {}) => ({ clipboard: null, providerSettings: defaultProviderSettings, edgeStyle: "angular" as const, - currentNodeId: null, + currentNodeIds: [], navigationTarget: null, setNavigationTarget: vi.fn(), getNodesWithComments: vi.fn(() => []), diff --git a/src/components/nodes/BaseNode.tsx b/src/components/nodes/BaseNode.tsx index 275fcd6e..467eb8d7 100644 --- a/src/components/nodes/BaseNode.tsx +++ b/src/components/nodes/BaseNode.tsx @@ -55,12 +55,12 @@ export function BaseNode({ titlePrefix, commentNavigation, }: BaseNodeProps) { - const currentNodeId = useWorkflowStore((state) => state.currentNodeId); + const currentNodeIds = useWorkflowStore((state) => state.currentNodeIds); const groups = useWorkflowStore((state) => state.groups); const nodes = useWorkflowStore((state) => state.nodes); const focusedCommentNodeId = useWorkflowStore((state) => state.focusedCommentNodeId); const setFocusedCommentNodeId = useWorkflowStore((state) => state.setFocusedCommentNodeId); - const isCurrentlyExecuting = currentNodeId === id; + const isCurrentlyExecuting = currentNodeIds.includes(id); const { getNodes, setNodes } = useReactFlow(); // Check if node is in a locked group diff --git a/src/store/__tests__/workflowStore.integration.test.ts b/src/store/__tests__/workflowStore.integration.test.ts index e5744a7e..192f37bd 100644 --- a/src/store/__tests__/workflowStore.integration.test.ts +++ b/src/store/__tests__/workflowStore.integration.test.ts @@ -907,7 +907,7 @@ describe("workflowStore integration tests", () => { // Workflow should complete successfully expect(useWorkflowStore.getState().isRunning).toBe(false); - expect(useWorkflowStore.getState().currentNodeId).toBeNull(); + expect(useWorkflowStore.getState().currentNodeIds).toEqual([]); }); it("should throw error on cycle detection", async () => { @@ -1228,7 +1228,7 @@ describe("workflowStore integration tests", () => { expect(useWorkflowStore.getState().isRunning).toBe(true); }); - it("should clear currentNodeId after execution completes", async () => { + it("should clear currentNodeIds after execution completes", async () => { useWorkflowStore.setState({ nodes: [ createTestNode("prompt-1", "prompt", { prompt: "test" }), @@ -1239,7 +1239,7 @@ describe("workflowStore integration tests", () => { const store = useWorkflowStore.getState(); await store.executeWorkflow(); - expect(useWorkflowStore.getState().currentNodeId).toBeNull(); + expect(useWorkflowStore.getState().currentNodeIds).toEqual([]); }); }); }); @@ -1785,7 +1785,7 @@ describe("workflowStore integration tests", () => { vi.unstubAllGlobals(); }); - it("should set currentNodeId to null after completion", async () => { + it("should set currentNodeIds to empty after completion", async () => { vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true, json: () => Promise.resolve({ success: true, image: "data:image/png;base64,test" }), @@ -1809,7 +1809,7 @@ describe("workflowStore integration tests", () => { const store = useWorkflowStore.getState(); await store.executeWorkflow(); - expect(useWorkflowStore.getState().currentNodeId).toBeNull(); + expect(useWorkflowStore.getState().currentNodeIds).toEqual([]); vi.unstubAllGlobals(); }); @@ -1846,7 +1846,7 @@ describe("workflowStore integration tests", () => { await store.executeWorkflow("output-1"); expect(useWorkflowStore.getState().isRunning).toBe(false); - expect(useWorkflowStore.getState().currentNodeId).toBeNull(); + expect(useWorkflowStore.getState().currentNodeIds).toEqual([]); vi.unstubAllGlobals(); }); diff --git a/src/store/workflowStore.ts b/src/store/workflowStore.ts index 6c3888d8..62b4efa4 100644 --- a/src/store/workflowStore.ts +++ b/src/store/workflowStore.ts @@ -122,11 +122,14 @@ interface WorkflowStore { // Execution isRunning: boolean; - currentNodeId: string | null; + currentNodeIds: string[]; // Changed from currentNodeId for parallel execution pausedAtNodeId: string | null; + maxConcurrentCalls: number; // Configurable concurrency limit (1-10) + _abortController: AbortController | null; // Internal: for cancellation executeWorkflow: (startFromNodeId?: string) => Promise; regenerateNode: (nodeId: string) => Promise; stopWorkflow: () => void; + setMaxConcurrentCalls: (value: number) => void; // Save/Load saveWorkflow: (name?: string) => void; @@ -299,6 +302,97 @@ async function waitForPendingImageSyncs(): Promise { await Promise.all(pendingImageSyncs.values()); } +// Concurrency settings +export const CONCURRENCY_SETTINGS_KEY = "node-banana-concurrency-limit"; +const DEFAULT_MAX_CONCURRENT_CALLS = 3; + +// Load/save concurrency setting from localStorage +const loadConcurrencySetting = (): number => { + if (typeof window === "undefined") return DEFAULT_MAX_CONCURRENT_CALLS; + const stored = localStorage.getItem(CONCURRENCY_SETTINGS_KEY); + if (stored) { + const parsed = parseInt(stored, 10); + if (!isNaN(parsed) && parsed >= 1 && parsed <= 10) { + return parsed; + } + } + return DEFAULT_MAX_CONCURRENT_CALLS; +}; + +const saveConcurrencySetting = (value: number): void => { + if (typeof window === "undefined") return; + localStorage.setItem(CONCURRENCY_SETTINGS_KEY, String(value)); +}; + +// Level grouping for parallel execution +export interface LevelGroup { + level: number; + nodeIds: string[]; +} + +/** + * Groups nodes by dependency level using Kahn's algorithm variant. + * Nodes at the same level can be executed in parallel. + * Level 0 = nodes with no incoming edges (roots) + * Level N = nodes whose dependencies are all at levels < N + */ +function groupNodesByLevel( + nodes: WorkflowNode[], + edges: WorkflowEdge[] +): LevelGroup[] { + // Calculate in-degree for each node + const inDegree = new Map(); + const adjList = new Map(); + + nodes.forEach((n) => { + inDegree.set(n.id, 0); + adjList.set(n.id, []); + }); + + edges.forEach((e) => { + inDegree.set(e.target, (inDegree.get(e.target) || 0) + 1); + adjList.get(e.source)?.push(e.target); + }); + + // BFS with level tracking (Kahn's algorithm variant) + const levels: LevelGroup[] = []; + let currentLevel = nodes + .filter((n) => inDegree.get(n.id) === 0) + .map((n) => n.id); + + let levelNum = 0; + while (currentLevel.length > 0) { + levels.push({ level: levelNum, nodeIds: [...currentLevel] }); + + const nextLevel: string[] = []; + for (const nodeId of currentLevel) { + for (const child of adjList.get(nodeId) || []) { + const newDegree = (inDegree.get(child) || 1) - 1; + inDegree.set(child, newDegree); + if (newDegree === 0) { + nextLevel.push(child); + } + } + } + + currentLevel = nextLevel; + levelNum++; + } + + return levels; +} + +/** + * Chunk an array into smaller arrays of specified size + */ +function chunk(array: T[], size: number): T[][] { + const chunks: T[][] = []; + for (let i = 0; i < array.length; i += size) { + chunks.push(array.slice(i, i + size)); + } + return chunks; +} + // Clear all imageRefs from nodes (used when saving to a different directory) function clearNodeImageRefs(nodes: WorkflowNode[]): WorkflowNode[] { return nodes.map(node => { @@ -328,8 +422,10 @@ export const useWorkflowStore = create((set, get) => ({ isModalOpen: false, showQuickstart: true, isRunning: false, - currentNodeId: null, + currentNodeIds: [], // Changed from currentNodeId for parallel execution pausedAtNodeId: null, + maxConcurrentCalls: loadConcurrencySetting(), // Default 3, configurable 1-10 + _abortController: null, // Internal: for cancellation globalImageHistory: [], // Auto-save initial state @@ -922,15 +1018,17 @@ export const useWorkflowStore = create((set, get) => ({ }, executeWorkflow: async (startFromNodeId?: string) => { - const { nodes, edges, groups, updateNodeData, getConnectedInputs, isRunning } = get(); + const { nodes, edges, groups, updateNodeData, getConnectedInputs, isRunning, maxConcurrentCalls } = get(); if (isRunning) { logger.warn('workflow.start', 'Workflow already running, ignoring execution request'); return; } + // Create AbortController for this execution run + const abortController = new AbortController(); const isResuming = startFromNodeId === get().pausedAtNodeId; - set({ isRunning: true, pausedAtNodeId: null }); + set({ isRunning: true, pausedAtNodeId: null, currentNodeIds: [], _abortController: abortController }); // Start logging session await logger.startSession(); @@ -940,102 +1038,80 @@ export const useWorkflowStore = create((set, get) => ({ edgeCount: edges.length, startFromNodeId, isResuming, + maxConcurrentCalls, }); - // Topological sort - const sorted: WorkflowNode[] = []; - const visited = new Set(); - const visiting = new Set(); - - const visit = (nodeId: string) => { - if (visited.has(nodeId)) return; - if (visiting.has(nodeId)) { - logger.error('workflow.validation', 'Cycle detected in workflow', { nodeId }); - throw new Error("Cycle detected in workflow"); - } + // Group nodes by level for parallel execution + const levels = groupNodesByLevel(nodes, edges); - visiting.add(nodeId); - - // Visit all nodes that this node depends on - edges - .filter((e) => e.target === nodeId) - .forEach((e) => visit(e.source)); - - visiting.delete(nodeId); - visited.add(nodeId); - - const node = nodes.find((n) => n.id === nodeId); - if (node) sorted.push(node); - }; + // Find starting level if startFromNodeId specified + let startLevel = 0; + if (startFromNodeId) { + const foundLevel = levels.findIndex((l) => l.nodeIds.includes(startFromNodeId)); + if (foundLevel !== -1) startLevel = foundLevel; + } - try { - nodes.forEach((node) => visit(node.id)); - - // If starting from a specific node, find its index and skip earlier nodes - let startIndex = 0; - if (startFromNodeId) { - const nodeIndex = sorted.findIndex((n) => n.id === startFromNodeId); - if (nodeIndex !== -1) { - startIndex = nodeIndex; - } + // Helper to execute a single node - returns true if successful, throws on error + const executeSingleNode = async (node: WorkflowNode, signal: AbortSignal): Promise => { + // Check for abort before starting + if (signal.aborted) { + throw new DOMException('Aborted', 'AbortError'); } - // Execute nodes in order, starting from startIndex - for (let i = startIndex; i < sorted.length; i++) { - const node = sorted[i]; - if (!get().isRunning) break; - - // Check for pause edges on incoming connections (skip if resuming from this exact node) - const isResumingThisNode = isResuming && node.id === startFromNodeId; - if (!isResumingThisNode) { - const incomingEdges = edges.filter((e) => e.target === node.id); - const pauseEdge = incomingEdges.find((e) => e.data?.hasPause); - if (pauseEdge) { - logger.info('workflow.end', 'Workflow paused at node', { - nodeId: node.id, - nodeType: node.type, - }); - set({ pausedAtNodeId: node.id, isRunning: false, currentNodeId: null }); - useToast.getState().show("Workflow paused - click Run to continue", "warning"); - - // Save logs to server (on pause) - const session = logger.getCurrentSession(); - if (session) { - session.endTime = new Date().toISOString(); - fetch('/api/logs', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ session }), - }).catch((err) => { - console.error('Failed to save log session:', err); - }); - } - - await logger.endSession(); - return; - } - } - - // Check if node is in a locked group - if so, skip execution - const nodeGroup = node.groupId ? groups[node.groupId] : null; - if (nodeGroup?.locked) { - logger.info('node.execution', `Skipping node in locked group`, { + // Check for pause edges on incoming connections (skip if resuming from this exact node) + const isResumingThisNode = isResuming && node.id === startFromNodeId; + if (!isResumingThisNode) { + const incomingEdges = edges.filter((e) => e.target === node.id); + const pauseEdge = incomingEdges.find((e) => e.data?.hasPause); + if (pauseEdge) { + logger.info('workflow.end', 'Workflow paused at node', { nodeId: node.id, nodeType: node.type, - groupId: node.groupId, - groupName: nodeGroup.name, }); - continue; // Skip to next node - } + set({ pausedAtNodeId: node.id, isRunning: false, currentNodeIds: [], _abortController: null }); + useToast.getState().show("Workflow paused - click Run to continue", "warning"); + + // Save logs to server (on pause) + const session = logger.getCurrentSession(); + if (session) { + session.endTime = new Date().toISOString(); + fetch('/api/logs', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ session }), + }).catch((err) => { + console.error('Failed to save log session:', err); + }); + } - set({ currentNodeId: node.id }); + await logger.endSession(); + // Signal to stop the entire workflow + abortController.abort(); + return; + } + } - logger.info('node.execution', `Executing ${node.type} node`, { + // Check if node is in a locked group - if so, skip execution + const nodeGroup = node.groupId ? groups[node.groupId] : null; + if (nodeGroup?.locked) { + logger.info('node.execution', `Skipping node in locked group`, { nodeId: node.id, nodeType: node.type, + groupId: node.groupId, + groupName: nodeGroup.name, }); + return; // Skip this node but continue with others + } - switch (node.type) { + logger.info('node.execution', `Executing ${node.type} node`, { + nodeId: node.id, + nodeType: node.type, + }); + + // NOTE: The switch statement below executes the node based on its type + // The signal parameter should be passed to any fetch calls for cancellation + + switch (node.type) { case "imageInput": // Nothing to execute, data is already set break; @@ -1129,7 +1205,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Missing text input", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1200,6 +1276,7 @@ export const useWorkflowStore = create((set, get) => ({ method: "POST", headers, body: JSON.stringify(requestPayload), + signal, // Pass abort signal for cancellation }); if (!response.ok) { @@ -1224,7 +1301,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: errorMessage, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1302,7 +1379,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: result.error || "Generation failed", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1330,7 +1407,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: errorMessage, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1350,7 +1427,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Missing required inputs", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1367,7 +1444,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "No model selected", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1429,6 +1506,7 @@ export const useWorkflowStore = create((set, get) => ({ method: "POST", headers, body: JSON.stringify(requestPayload), + signal, // Pass abort signal for cancellation }); if (!response.ok) { @@ -1453,7 +1531,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: errorMessage, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1538,7 +1616,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: result.error || "Video generation failed", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1564,7 +1642,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: errorMessage, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1586,7 +1664,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Missing text input - connect a prompt node or set internal prompt", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1639,6 +1717,7 @@ export const useWorkflowStore = create((set, get) => ({ temperature: nodeData.temperature, maxTokens: nodeData.maxTokens, }), + signal, // Pass abort signal for cancellation }); if (!response.ok) { @@ -1659,7 +1738,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: errorMessage, }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1681,7 +1760,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: result.error || "LLM generation failed", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1693,7 +1772,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: error instanceof Error ? error.message : "LLM generation failed", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1709,7 +1788,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "No input image connected", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); return; } @@ -1720,7 +1799,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Node not configured - open settings first", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); return; } @@ -1769,7 +1848,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: error instanceof Error ? error.message : "Failed to split image", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -1878,10 +1957,67 @@ export const useWorkflowStore = create((set, get) => ({ break; } } + }; // End of executeSingleNode helper + + try { + // Execute levels sequentially, but nodes within each level in parallel + for (let levelIdx = startLevel; levelIdx < levels.length; levelIdx++) { + // Check if execution was stopped + if (abortController.signal.aborted || !get().isRunning) break; + + const level = levels[levelIdx]; + const levelNodes = level.nodeIds + .map((id) => nodes.find((n) => n.id === id)) + .filter((n): n is WorkflowNode => n !== undefined); + + if (levelNodes.length === 0) continue; + + // Execute nodes in batches respecting concurrency limit + const batches = chunk(levelNodes, maxConcurrentCalls); + + for (const batch of batches) { + if (abortController.signal.aborted || !get().isRunning) break; + + // Update currentNodeIds to show which nodes are executing + const batchIds = batch.map((n) => n.id); + set({ currentNodeIds: batchIds }); + + logger.info('node.execution', `Executing level ${levelIdx} batch`, { + level: levelIdx, + nodeCount: batch.length, + nodeIds: batchIds, + }); + + // Execute batch in parallel + const results = await Promise.allSettled( + batch.map((node) => executeSingleNode(node, abortController.signal)) + ); + + // Check for failures (fail-fast behavior) + const failed = results.find( + (r): r is PromiseRejectedResult => + r.status === 'rejected' && + !(r.reason instanceof DOMException && r.reason.name === 'AbortError') + ); + + if (failed) { + // Log the failure and abort remaining executions + logger.error('workflow.error', 'Node execution failed in parallel batch', { + level: levelIdx, + error: failed.reason instanceof Error ? failed.reason.message : String(failed.reason), + }); + abortController.abort(); + throw failed.reason; + } + } } - logger.info('workflow.end', 'Workflow execution completed successfully'); - set({ isRunning: false, currentNodeId: null }); + // Check if we completed or were aborted + if (!abortController.signal.aborted && get().isRunning) { + logger.info('workflow.end', 'Workflow execution completed successfully'); + } + + set({ isRunning: false, currentNodeIds: [], _abortController: null }); // Save logs to server const session = logger.getCurrentSession(); @@ -1898,8 +2034,18 @@ export const useWorkflowStore = create((set, get) => ({ await logger.endSession(); } catch (error) { - logger.error('workflow.error', 'Workflow execution failed', {}, error instanceof Error ? error : undefined); - set({ isRunning: false, currentNodeId: null }); + // Handle AbortError gracefully (user cancelled) + if (error instanceof DOMException && error.name === 'AbortError') { + logger.info('workflow.end', 'Workflow execution cancelled by user'); + } else { + logger.error('workflow.error', 'Workflow execution failed', {}, error instanceof Error ? error : undefined); + // Show error toast for the failed node + useToast.getState().show( + `Workflow failed: ${error instanceof Error ? error.message : 'Unknown error'}`, + "error" + ); + } + set({ isRunning: false, currentNodeIds: [], _abortController: null }); // Save logs to server (even on error) const session = logger.getCurrentSession(); @@ -1919,7 +2065,18 @@ export const useWorkflowStore = create((set, get) => ({ }, stopWorkflow: () => { - set({ isRunning: false, currentNodeId: null }); + // Abort any in-flight requests + const controller = get()._abortController; + if (controller) { + controller.abort(); + } + set({ isRunning: false, currentNodeIds: [], _abortController: null }); + }, + + setMaxConcurrentCalls: (value: number) => { + const clamped = Math.max(1, Math.min(10, value)); + saveConcurrencySetting(clamped); + set({ maxConcurrentCalls: clamped }); }, regenerateNode: async (nodeId: string) => { @@ -1936,7 +2093,7 @@ export const useWorkflowStore = create((set, get) => ({ return; } - set({ isRunning: true, currentNodeId: nodeId }); + set({ isRunning: true, currentNodeIds: [nodeId] }); await logger.startSession(); logger.info('node.execution', 'Regenerating node', { @@ -1966,7 +2123,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Missing text input", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2044,7 +2201,7 @@ export const useWorkflowStore = create((set, get) => ({ errorMessage, }); updateNodeData(nodeId, { status: "error", error: errorMessage }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2130,7 +2287,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Missing text input", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2197,7 +2354,7 @@ export const useWorkflowStore = create((set, get) => ({ errorMessage, }); updateNodeData(nodeId, { status: "error", error: errorMessage }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2238,7 +2395,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Missing text input", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2251,7 +2408,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "No model selected", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2325,7 +2482,7 @@ export const useWorkflowStore = create((set, get) => ({ errorMessage, }); updateNodeData(nodeId, { status: "error", error: errorMessage }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2422,7 +2579,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "No input image connected", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2435,7 +2592,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: "Node not configured - open settings first", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } @@ -2496,14 +2653,14 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: error instanceof Error ? error.message : "Failed to split image", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); await logger.endSession(); return; } } logger.info('node.execution', 'Node regeneration completed successfully', { nodeId }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); // Save logs to server const session = logger.getCurrentSession(); @@ -2527,7 +2684,7 @@ export const useWorkflowStore = create((set, get) => ({ status: "error", error: error instanceof Error ? error.message : "Regeneration failed", }); - set({ isRunning: false, currentNodeId: null }); + set({ isRunning: false, currentNodeIds: [] }); // Save logs to server (even on error) const session = logger.getCurrentSession(); @@ -2651,7 +2808,7 @@ export const useWorkflowStore = create((set, get) => ({ edgeStyle: hydratedWorkflow.edgeStyle || "angular", groups: hydratedWorkflow.groups || {}, isRunning: false, - currentNodeId: null, + currentNodeIds: [], // Restore workflow ID and paths from localStorage if available workflowId: workflow.id || null, workflowName: workflow.name, @@ -2681,7 +2838,7 @@ export const useWorkflowStore = create((set, get) => ({ edges: [], groups: {}, isRunning: false, - currentNodeId: null, + currentNodeIds: [], // Reset auto-save state when clearing workflow workflowId: null, workflowName: null,