|
| 1 | +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | + |
| 3 | +import { describe, it, expect, beforeEach } from 'vitest'; |
| 4 | +import { AutomationEngine } from '../engine.js'; |
| 5 | +import type { NodeExecutor } from '../engine.js'; |
| 6 | +import { InMemorySuspendedRunStore } from '../suspended-run-store.js'; |
| 7 | +import { registerMapNode } from './map-node.js'; |
| 8 | + |
| 9 | +function silentLogger() { |
| 10 | + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } } as any; |
| 11 | +} |
| 12 | +function ctx() { |
| 13 | + return { logger: silentLogger(), getService() { throw new Error('none'); } } as any; |
| 14 | +} |
| 15 | + |
| 16 | +/** |
| 17 | + * Wire an engine with the `map` node, a per-item child flow, and a parent flow |
| 18 | + * that maps over `{items}`. `childNodes` is the child flow's middle node(s) |
| 19 | + * (between start and end) — pass a `pauser` to exercise per-item durable pause. |
| 20 | + */ |
| 21 | +function setup(childNodes: Array<{ id: string; type: string }>, captured: unknown[]) { |
| 22 | + const engine = new AutomationEngine(silentLogger()); |
| 23 | + registerMapNode(engine, ctx()); |
| 24 | + |
| 25 | + // Child marker: copies the mapped item (passed as params.val) to the child's |
| 26 | + // output variable `result`. |
| 27 | + engine.registerNodeExecutor({ |
| 28 | + type: 'itemmark', |
| 29 | + async execute(_node, variables, context) { |
| 30 | + variables.set('result', (context as any)?.params?.val); |
| 31 | + return { success: true }; |
| 32 | + }, |
| 33 | + } as NodeExecutor); |
| 34 | + // Pauses the child (stands in for an approval / screen / wait). |
| 35 | + engine.registerNodeExecutor({ |
| 36 | + type: 'pauser', |
| 37 | + async execute() { return { success: true, suspend: true }; }, |
| 38 | + } as NodeExecutor); |
| 39 | + // Fails the child terminally. |
| 40 | + engine.registerNodeExecutor({ |
| 41 | + type: 'failer', |
| 42 | + async execute() { return { success: false, error: 'boom' }; }, |
| 43 | + } as NodeExecutor); |
| 44 | + // Parent checker after the map node: captures the collected results array. |
| 45 | + engine.registerNodeExecutor({ |
| 46 | + type: 'mapcheck', |
| 47 | + async execute(_node, variables) { |
| 48 | + captured.push(variables.get('mapped')); |
| 49 | + return { success: true }; |
| 50 | + }, |
| 51 | + } as NodeExecutor); |
| 52 | + |
| 53 | + const seq = [{ id: 'cs', type: 'start' }, ...childNodes, { id: 'ce', type: 'end' }]; |
| 54 | + engine.registerFlow('child_flow', { |
| 55 | + name: 'child_flow', |
| 56 | + label: 'Child', |
| 57 | + type: 'autolaunched', |
| 58 | + variables: [{ name: 'result', type: 'text', isOutput: true }], |
| 59 | + nodes: seq.map(n => ({ label: n.id, ...n })), |
| 60 | + edges: seq.slice(0, -1).map((n, i) => ({ id: `c${i}`, source: n.id, target: seq[i + 1].id })), |
| 61 | + } as never); |
| 62 | + |
| 63 | + engine.registerFlow('parent_flow', { |
| 64 | + name: 'parent_flow', |
| 65 | + label: 'Parent', |
| 66 | + type: 'autolaunched', |
| 67 | + variables: [{ name: 'items', type: 'list', isInput: true }], |
| 68 | + nodes: [ |
| 69 | + { id: 'ps', type: 'start', label: 'Start' }, |
| 70 | + { |
| 71 | + id: 'do_map', type: 'map', label: 'For each', |
| 72 | + config: { flowName: 'child_flow', collection: '{items}', iteratorVariable: 'item', input: { val: '{item}' }, outputVariable: 'mapped' }, |
| 73 | + }, |
| 74 | + { id: 'chk', type: 'mapcheck', label: 'Check' }, |
| 75 | + { id: 'pe', type: 'end', label: 'End' }, |
| 76 | + ], |
| 77 | + edges: [ |
| 78 | + { id: 'p1', source: 'ps', target: 'do_map' }, |
| 79 | + { id: 'p2', source: 'do_map', target: 'chk' }, |
| 80 | + { id: 'p3', source: 'chk', target: 'pe' }, |
| 81 | + ], |
| 82 | + } as never); |
| 83 | + |
| 84 | + return engine; |
| 85 | +} |
| 86 | + |
| 87 | +const childRunId = (engine: AutomationEngine) => |
| 88 | + engine.listSuspendedRuns().find(r => r.flowName === 'child_flow')?.runId; |
| 89 | + |
| 90 | +describe('map node executor (sequential multi-instance)', () => { |
| 91 | + let captured: unknown[]; |
| 92 | + beforeEach(() => { captured = []; }); |
| 93 | + |
| 94 | + it('runs a synchronous per-item subflow over the collection, collecting results in order', async () => { |
| 95 | + const engine = setup([{ id: 'cm', type: 'itemmark' }], captured); |
| 96 | + const result = await engine.execute('parent_flow', { params: { items: ['a', 'b', 'c'] } }); |
| 97 | + |
| 98 | + expect(result.success).toBe(true); |
| 99 | + expect(result.status).toBeUndefined(); // ran to completion, no pause |
| 100 | + expect(captured).toEqual([[{ result: 'a' }, { result: 'b' }, { result: 'c' }]]); |
| 101 | + expect(engine.listSuspendedRuns()).toHaveLength(0); |
| 102 | + }); |
| 103 | + |
| 104 | + it('completes immediately on an empty collection', async () => { |
| 105 | + const engine = setup([{ id: 'cm', type: 'itemmark' }], captured); |
| 106 | + const result = await engine.execute('parent_flow', { params: { items: [] } }); |
| 107 | + expect(result.success).toBe(true); |
| 108 | + expect(captured).toEqual([[]]); |
| 109 | + }); |
| 110 | + |
| 111 | + it('drives items ONE AT A TIME when each subflow pauses, re-entering on resume', async () => { |
| 112 | + const engine = setup([{ id: 'cp', type: 'pauser' }, { id: 'cm', type: 'itemmark' }], captured); |
| 113 | + |
| 114 | + // Item 0 pauses → the parent suspends at the map node. |
| 115 | + const r0 = await engine.execute('parent_flow', { params: { items: ['a', 'b', 'c'] } }); |
| 116 | + expect(r0.status).toBe('paused'); |
| 117 | + const parent = engine.listSuspendedRuns().find(x => x.flowName === 'parent_flow')!; |
| 118 | + expect(parent.nodeId).toBe('do_map'); |
| 119 | + expect(parent.correlation).toMatch(/^map:/); |
| 120 | + expect(captured).toHaveLength(0); // not done yet |
| 121 | + |
| 122 | + // Resume item 0's child → bubbles → re-enters map → item 1 pauses. |
| 123 | + const id0 = childRunId(engine)!; |
| 124 | + const after0 = await engine.resume(id0); |
| 125 | + expect(after0.success).toBe(true); |
| 126 | + expect(engine.listSuspendedRuns().some(x => x.flowName === 'parent_flow')).toBe(true); // still paused |
| 127 | + expect(captured).toHaveLength(0); |
| 128 | + |
| 129 | + // Resume item 1 → item 2 pauses. |
| 130 | + await engine.resume(childRunId(engine)!); |
| 131 | + expect(captured).toHaveLength(0); |
| 132 | + |
| 133 | + // Resume item 2 → all done → parent continues past the map node. |
| 134 | + await engine.resume(childRunId(engine)!); |
| 135 | + expect(captured).toEqual([[{ result: 'a' }, { result: 'b' }, { result: 'c' }]]); |
| 136 | + expect(engine.listSuspendedRuns()).toHaveLength(0); |
| 137 | + }); |
| 138 | + |
| 139 | + it('fails the map fast when an item subflow fails', async () => { |
| 140 | + const engine = setup([{ id: 'cf', type: 'failer' }], captured); |
| 141 | + const result = await engine.execute('parent_flow', { params: { items: ['a', 'b'] } }); |
| 142 | + expect(result.success).toBe(false); |
| 143 | + expect(result.error).toMatch(/item 0.*failed/i); |
| 144 | + expect(captured).toEqual([]); // downstream never ran |
| 145 | + }); |
| 146 | + |
| 147 | + it('survives a process restart mid-map: resume on a fresh engine continues the sequence', async () => { |
| 148 | + const store = new InMemorySuspendedRunStore(); |
| 149 | + const engine = setup([{ id: 'cp', type: 'pauser' }, { id: 'cm', type: 'itemmark' }], captured); |
| 150 | + engine.setSuspendedRunStore(store); |
| 151 | + |
| 152 | + await engine.execute('parent_flow', { params: { items: ['a', 'b'] } }); |
| 153 | + const id0 = childRunId(engine)!; |
| 154 | + expect((await store.list()).length).toBe(2); // parent (map) + child item 0 |
| 155 | + |
| 156 | + // "Restart": a fresh engine sharing only the durable store + the registry. |
| 157 | + const capturedB: unknown[] = []; |
| 158 | + const engineB = setup([{ id: 'cp', type: 'pauser' }, { id: 'cm', type: 'itemmark' }], capturedB); |
| 159 | + engineB.setSuspendedRunStore(store); |
| 160 | + |
| 161 | + await engineB.resume(id0); // item 0 done → item 1 pauses |
| 162 | + await engineB.resume(childRunId(engineB)!); // item 1 done → all done |
| 163 | + expect(capturedB).toEqual([[{ result: 'a' }, { result: 'b' }]]); |
| 164 | + expect(await store.list()).toHaveLength(0); |
| 165 | + }); |
| 166 | +}); |
0 commit comments