diff --git a/CHANGELOG.md b/CHANGELOG.md index a2c5312..b8e9855 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ All notable changes to Lobster will be documented in this file. - Add workflow condition comparison operators `<`, `<=`, `>`, and `>=` with strict numeric semantics (booleans/null do not coerce), including mixed boolean-expression support with `&&`/`||`. Thanks to [@scottgl9](https://github.com/scottgl9) (PR [#71](https://github.com/openclaw/lobster/pull/71)). - Add workflow-level LLM cost tracking with `_meta.cost` summaries, per-step usage attribution, and optional `cost_limit` controls with `warn`/`stop` actions (plus custom pricing via `LOBSTER_LLM_PRICING_JSON`). Thanks to [@scottgl9](https://github.com/scottgl9) (PR [#70](https://github.com/openclaw/lobster/pull/70)). - Add `parallel` workflow steps with branch fan-out, `wait: all|any`, block-level timeout support, and branch result references in downstream steps. Thanks to [@scottgl9](https://github.com/scottgl9) (PR [#69](https://github.com/openclaw/lobster/pull/69)). +- Add `for_each` workflow steps for per-item sub-step execution over arrays, including loop-scoped vars (`item_var`/`index_var`), optional `batch_size` + `pause_ms`, and collected iteration outputs for downstream steps. Thanks to [@scottgl9](https://github.com/scottgl9) (PR [#68](https://github.com/openclaw/lobster/pull/68)). ## 2026.4.6 diff --git a/src/workflows/file.ts b/src/workflows/file.ts index f01f8d5..4844172 100644 --- a/src/workflows/file.ts +++ b/src/workflows/file.ts @@ -57,6 +57,12 @@ export type WorkflowStep = { condition?: unknown; when?: unknown; parallel?: ParallelConfig; + for_each?: string; + item_var?: string; + index_var?: string; + batch_size?: number; + pause_ms?: number; + steps?: WorkflowStep[]; timeout_ms?: number; on_error?: 'stop' | 'continue' | 'skip_rest'; }; @@ -272,21 +278,116 @@ export async function loadWorkflowFile(filePath: string): Promise } } } + if (step.for_each !== undefined && typeof step.for_each !== 'string') { + throw new Error(`Workflow step ${step.id} for_each must be a string (step reference expression)`); + } + const isForEach = typeof step.for_each === 'string'; + if (isForEach) { + if (!Array.isArray(step.steps) || step.steps.length === 0) { + throw new Error(`Workflow step ${step.id} for_each requires a non-empty steps array`); + } + if ( + step.batch_size !== undefined + && ( + typeof step.batch_size !== 'number' + || !Number.isInteger(step.batch_size) + || step.batch_size < 1 + ) + ) { + throw new Error(`Workflow step ${step.id} batch_size must be a positive integer`); + } + if ( + step.pause_ms !== undefined + && ( + typeof step.pause_ms !== 'number' + || !Number.isFinite(step.pause_ms) + || step.pause_ms < 0 + ) + ) { + throw new Error(`Workflow step ${step.id} pause_ms must be a finite non-negative number`); + } + if (isApprovalStep(step.approval)) { + throw new Error(`Workflow step ${step.id} for_each steps cannot define approval (use a separate step after the loop)`); + } + if (isInputStep(step.input)) { + throw new Error(`Workflow step ${step.id} for_each steps cannot define input (use a separate step after the loop)`); + } + if (step.stdin !== undefined && step.stdin !== null) { + throw new Error(`Workflow step ${step.id} for_each steps cannot define stdin (loop input comes from the for_each expression)`); + } + const loopShell = typeof step.run === 'string' ? step.run : step.command; + const loopPipeline = typeof step.pipeline === 'string' ? step.pipeline : undefined; + if (loopShell || loopPipeline || step.workflow || step.parallel) { + throw new Error(`Workflow step ${step.id} for_each cannot also define run, command, pipeline, workflow, or parallel`); + } + if (step.item_var !== undefined && typeof step.item_var !== 'string') { + throw new Error(`Workflow step ${step.id} item_var must be a string`); + } + if (step.index_var !== undefined && typeof step.index_var !== 'string') { + throw new Error(`Workflow step ${step.id} index_var must be a string`); + } + const loopItemVar = step.item_var ?? 'item'; + const loopIndexVar = step.index_var ?? 'index'; + if (loopItemVar === loopIndexVar) { + throw new Error(`Workflow step ${step.id} item_var and index_var cannot be the same`); + } + const subStepIds = new Set(); + for (const sub of step.steps) { + if (!sub || typeof sub !== 'object' || !sub.id || typeof sub.id !== 'string') { + throw new Error(`Workflow step ${step.id} for_each sub-step requires an id`); + } + if (sub.id === loopItemVar || sub.id === loopIndexVar) { + throw new Error(`Workflow step ${step.id} for_each sub-step id '${sub.id}' conflicts with loop variable`); + } + if (subStepIds.has(sub.id)) { + throw new Error(`Workflow step ${step.id} duplicate for_each sub-step id: ${sub.id}`); + } + subStepIds.add(sub.id); + if (isApprovalStep(sub.approval) || isInputStep(sub.input)) { + throw new Error(`Workflow step ${step.id} for_each sub-steps cannot contain approval or input steps`); + } + if (sub.run !== undefined && typeof sub.run !== 'string') { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} run must be a string`); + } + if (sub.command !== undefined && typeof sub.command !== 'string') { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} command must be a string`); + } + if (sub.pipeline !== undefined && typeof sub.pipeline !== 'string') { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} pipeline must be a string`); + } + if (sub.workflow || sub.parallel || sub.for_each) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} cannot define workflow, parallel, or for_each`); + } + const subShell = typeof sub.run === 'string' && sub.run.trim() + ? sub.run + : (typeof sub.command === 'string' && sub.command.trim() ? sub.command : undefined); + const subPipeline = typeof sub.pipeline === 'string' && sub.pipeline.trim() + ? sub.pipeline + : undefined; + if (!subShell && !subPipeline) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} requires run, command, or pipeline`); + } + if (Number(Boolean(subShell)) + Number(Boolean(subPipeline)) > 1) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} can only define one of run, command, or pipeline`); + } + } + } const shellCommand = typeof step.run === 'string' ? step.run : step.command; const pipeline = typeof step.pipeline === 'string' ? step.pipeline : undefined; const workflowRef = typeof step.workflow === 'string' && step.workflow.trim() ? step.workflow : undefined; const executionCount = Number(Boolean(shellCommand)) + Number(Boolean(pipeline)) + Number(Boolean(workflowRef)) - + Number(isParallel); + + Number(isParallel) + + Number(isForEach); if (executionCount === 0 && !isApprovalStep(step.approval) && !isInputStep(step.input)) { - throw new Error(`Workflow step ${step.id} requires run, command, pipeline, workflow, parallel, approval, or input`); + throw new Error(`Workflow step ${step.id} requires run, command, pipeline, workflow, parallel, for_each, approval, or input`); } if (executionCount > 1) { - throw new Error(`Workflow step ${step.id} can only define one of run, command, pipeline, workflow, or parallel`); + throw new Error(`Workflow step ${step.id} can only define one of run, command, pipeline, workflow, parallel, or for_each`); } if (executionCount > 0 && isInputStep(step.input)) { - throw new Error(`Workflow step ${step.id} input steps cannot define run, command, pipeline, workflow, or parallel`); + throw new Error(`Workflow step ${step.id} input steps cannot define run, command, pipeline, workflow, parallel, or for_each`); } if (isApprovalStep(step.approval) && isInputStep(step.input)) { throw new Error(`Workflow step ${step.id} cannot define both approval and input`); @@ -558,6 +659,104 @@ export async function runWorkflowFile({ continue; } + if (typeof step.for_each === 'string' && Array.isArray(step.steps)) { + const itemsRef = resolveInputValue(step.for_each, resolvedArgs, results); + if (!Array.isArray(itemsRef)) { + throw new Error(`Workflow step ${step.id} for_each: expected array, got ${typeof itemsRef}`); + } + + const itemVar = step.item_var ?? 'item'; + const indexVar = step.index_var ?? 'index'; + const batchSize = step.batch_size ?? 1; + const iterationResults: unknown[] = []; + + for (let itemIdx = 0; itemIdx < itemsRef.length; itemIdx++) { + if (step.pause_ms && itemIdx > 0 && itemIdx % batchSize === 0) { + await abortableSleep(step.pause_ms, ctx.signal); + } + + const item = itemsRef[itemIdx]; + const scopedResults: Record = { ...results }; + scopedResults[itemVar] = { + id: itemVar, + json: item, + stdout: typeof item === 'string' ? item : JSON.stringify(item), + }; + scopedResults[indexVar] = { + id: indexVar, + json: itemIdx, + stdout: String(itemIdx), + }; + + for (const subStep of step.steps) { + if (!evaluateCondition(subStep.when ?? subStep.condition, scopedResults)) { + scopedResults[subStep.id] = { id: subStep.id, skipped: true }; + continue; + } + + const loopEnvBase = mergeEnv(ctx.env, workflow.env, step.env, resolvedArgs, scopedResults); + const subEnv = subStep.env + ? mergeEnv(loopEnvBase, undefined, subStep.env, resolvedArgs, scopedResults) + : loopEnvBase; + const subCwd = resolveCwd(subStep.cwd ?? step.cwd ?? workflow.cwd, resolvedArgs) ?? ctx.cwd; + const subExecution = getStepExecution(subStep); + + let subResult: WorkflowStepResult; + if (subExecution.kind === 'shell') { + const command = resolveTemplate(subExecution.value, resolvedArgs, scopedResults); + const stdinValue = resolveShellStdin(subStep.stdin, resolvedArgs, scopedResults); + const { stdout } = await runShellCommand({ command, stdin: stdinValue, env: subEnv, cwd: subCwd, signal: ctx.signal }); + subResult = { id: subStep.id, stdout, json: parseJson(stdout) }; + } else if (subExecution.kind === 'pipeline') { + if (!ctx.registry) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${subStep.id} requires a command registry for pipeline execution`); + } + const pipelineText = resolveTemplate(subExecution.value, resolvedArgs, scopedResults); + const inputValue = resolveInputValue(subStep.stdin, resolvedArgs, scopedResults); + subResult = await runPipelineStep({ + stepId: subStep.id, + pipelineText, + inputValue, + ctx, + env: subEnv, + cwd: subCwd, + }); + } else { + const inputValue = resolveInputValue(subStep.stdin, resolvedArgs, scopedResults); + subResult = createSyntheticStepResult(subStep.id, inputValue); + } + + scopedResults[subStep.id] = subResult; + trackStepCost(costTracker, `${step.id}.${subStep.id}`, subResult); + if (workflow.cost_limit) { + costTracker.checkLimit(workflow.cost_limit, ctx.stderr); + } + } + + const iterResult: Record = { [itemVar]: item, [indexVar]: itemIdx }; + for (const subStep of step.steps) { + const subResult = scopedResults[subStep.id]; + if (subResult && !subResult.skipped) { + iterResult[subStep.id] = subResult.json !== undefined ? subResult.json : subResult.stdout; + } + } + iterationResults.push(iterResult); + } + + const loopResult: WorkflowStepResult = { + id: step.id, + json: iterationResults, + stdout: JSON.stringify(iterationResults), + }; + results[step.id] = loopResult; + lastStepId = step.id; + trackStepCost(costTracker, step.id, loopResult); + if (workflow.cost_limit) { + costTracker.checkLimit(workflow.cost_limit, ctx.stderr); + } + continue; + } + const env = mergeEnv(ctx.env, workflow.env, step.env, resolvedArgs, results); const cwd = resolveCwd(step.cwd ?? workflow.cwd, resolvedArgs) ?? ctx.cwd; const execution = getStepExecution(step); @@ -950,6 +1149,68 @@ function dryRunWorkflow({ continue; } + if (typeof step.for_each === 'string' && Array.isArray(step.steps)) { + lines.push(` ${num}. ${step.id} [for_each]`); + const forEachRef = step.for_each; + const forEachNote = dryRunTemplateNote(forEachRef); + lines.push(` for_each: ${forEachRef}${forEachNote ? ` ${forEachNote}` : ''}`); + if (forEachRef.trim().startsWith('$')) { + try { + resolveInputValue(forEachRef, resolvedArgs, results); + } catch (err: any) { + throw new Error(`Workflow step ${step.id} for_each: ${err?.message ?? String(err)}`); + } + } + const dryItemVar = step.item_var ?? 'item'; + const dryIndexVar = step.index_var ?? 'index'; + lines.push(` item_var: ${dryItemVar}, index_var: ${dryIndexVar}`); + if (step.batch_size) lines.push(` batch_size: ${step.batch_size}`); + if (step.pause_ms) lines.push(` pause_ms: ${step.pause_ms}`); + lines.push(` sub-steps: ${step.steps.length}`); + + const loopScopedResults = { ...results }; + loopScopedResults[dryItemVar] = { id: dryItemVar, json: { _placeholder: true } }; + loopScopedResults[dryIndexVar] = { id: dryIndexVar, json: 0 }; + for (let subIdx = 0; subIdx < step.steps.length; subIdx++) { + const sub = step.steps[subIdx]; + if (!evaluateCondition(sub.when ?? sub.condition, loopScopedResults)) { + lines.push(` ${subIdx + 1}. ${sub.id} [skipped — condition: false]`); + loopScopedResults[sub.id] = { id: sub.id, skipped: true }; + continue; + } + if (sub.stdin !== undefined && sub.stdin !== null) { + try { + resolveInputValue(sub.stdin, resolvedArgs, loopScopedResults); + } catch (err: any) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} stdin: ${err?.message ?? String(err)}`); + } + } + const subExec = getStepExecution(sub); + if (subExec.kind === 'shell') { + const command = resolveDryRunTemplate(subExec.value, resolvedArgs, loopScopedResults); + lines.push(` ${subIdx + 1}. ${sub.id} [shell] run: ${command}`); + } else if (subExec.kind === 'pipeline') { + if (!ctx.registry) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} requires a command registry for pipeline execution`); + } + const pipelineText = resolveDryRunTemplate(subExec.value, resolvedArgs, loopScopedResults); + const stages = parsePipeline(pipelineText); + for (const stage of stages) { + if (hasDeferredDryRunStageName(stage.name)) continue; + if (!ctx.registry.get(stage.name)) { + throw new Error(`Workflow step ${step.id} for_each sub-step ${sub.id} pipeline: unknown command: ${stage.name}`); + } + } + lines.push(` ${subIdx + 1}. ${sub.id} [pipeline] pipeline: ${pipelineText}`); + } else { + lines.push(` ${subIdx + 1}. ${sub.id} [no-op]`); + } + loopScopedResults[sub.id] = { id: sub.id }; + } + results[step.id] = { id: step.id }; + continue; + } + // Validate stdin refs early — throws if a strict ref like '$missing.stdout' // points to a step that doesn't exist at all (real execution would also fail). // We call resolveInputValue with the current results so refs to steps we've @@ -1966,6 +2227,28 @@ function createSyntheticStepResult(stepId: string, value: unknown): WorkflowStep }; } +function abortableSleep(ms: number, signal?: AbortSignal): Promise { + if (!signal) { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + return new Promise((resolve, reject) => { + if (signal.aborted) { + reject(signal.reason ?? new DOMException('The operation was aborted.', 'AbortError')); + return; + } + let timer: ReturnType; + const onAbort = () => { + clearTimeout(timer); + reject(signal.reason ?? new DOMException('The operation was aborted.', 'AbortError')); + }; + timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort); + resolve(); + }, ms); + signal.addEventListener('abort', onAbort, { once: true }); + }); +} + function encodeShellInput(value: unknown) { if (value === null || value === undefined) return null; if (typeof value === 'string') return value; diff --git a/test/for_each.test.ts b/test/for_each.test.ts new file mode 100644 index 0000000..61e073a --- /dev/null +++ b/test/for_each.test.ts @@ -0,0 +1,201 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { promises as fsp } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { PassThrough } from 'node:stream'; + +import { createDefaultRegistry } from '../src/commands/registry.js'; +import { loadWorkflowFile, runWorkflowFile } from '../src/workflows/file.js'; + +async function runWorkflow(workflow: unknown) { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const stateDir = path.join(tmpDir, 'state'); + const filePath = path.join(tmpDir, 'workflow.lobster'); + await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), 'utf8'); + + return runWorkflowFile({ + filePath, + ctx: { + stdin: process.stdin, + stdout: process.stdout, + stderr: process.stderr, + env: { ...process.env, LOBSTER_STATE_DIR: stateDir }, + mode: 'tool', + registry: createDefaultRegistry(), + }, + }); +} + +test('for_each iterates items and collects per-iteration results', async () => { + const result = await runWorkflow({ + steps: [ + { id: 'data', command: 'node -e "process.stdout.write(JSON.stringify([{name:\\"a\\"},{name:\\"b\\"}]))"' }, + { + id: 'loop', + for_each: '$data.json', + steps: [ + { + id: 'transform', + command: 'node -e "process.stdout.write(JSON.stringify({upper: process.env.NAME.toUpperCase()}))"', + env: { NAME: '$item.json.name' }, + }, + ], + }, + ], + }); + assert.equal(result.status, 'ok'); + const output = result.output as any[]; + assert.equal(output.length, 2); + assert.equal(output[0].index, 0); + assert.equal(output[1].index, 1); + assert.equal(output[0].transform.upper, 'A'); + assert.equal(output[1].transform.upper, 'B'); +}); + +test('for_each supports custom item_var and index_var', async () => { + const result = await runWorkflow({ + steps: [ + { id: 'vals', command: 'node -e "process.stdout.write(JSON.stringify([10,20]))"' }, + { + id: 'loop', + for_each: '$vals.json', + item_var: 'num', + index_var: 'idx', + steps: [ + { + id: 'emit', + command: 'node -e "process.stdout.write(JSON.stringify({num:$num.json,idx:$idx.json}))"', + }, + ], + }, + ], + }); + assert.equal(result.status, 'ok'); + assert.deepEqual(result.output, [ + { num: 10, idx: 0, emit: { num: 10, idx: 0 } }, + { num: 20, idx: 1, emit: { num: 20, idx: 1 } }, + ]); +}); + +test('for_each throws when source is not an array', async () => { + await assert.rejects( + () => runWorkflow({ + steps: [ + { id: 'data', command: 'node -e "process.stdout.write(JSON.stringify({x:1}))"' }, + { id: 'loop', for_each: '$data.json', steps: [{ id: 'x', command: 'echo hi' }] }, + ], + }), + /for_each: expected array/, + ); +}); + +test('for_each validation rejects empty sub-step list', async () => { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const filePath = path.join(tmpDir, 'bad.lobster'); + await fsp.writeFile(filePath, JSON.stringify({ + steps: [{ id: 'loop', for_each: '$x.json', steps: [] }], + }), 'utf8'); + await assert.rejects(() => loadWorkflowFile(filePath), /for_each requires a non-empty steps array/); +}); + +test('for_each validation rejects run/command/pipeline/workflow/parallel on loop step', async () => { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const filePath = path.join(tmpDir, 'bad.lobster'); + await fsp.writeFile(filePath, JSON.stringify({ + steps: [{ id: 'loop', for_each: '$x.json', run: 'echo no', steps: [{ id: 's', command: 'echo hi' }] }], + }), 'utf8'); + await assert.rejects(() => loadWorkflowFile(filePath), /for_each cannot also define run, command, pipeline, workflow, or parallel/); +}); + +test('for_each validation rejects approval/input in sub-steps', async () => { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const filePath = path.join(tmpDir, 'bad.lobster'); + await fsp.writeFile(filePath, JSON.stringify({ + steps: [{ id: 'loop', for_each: '$x.json', steps: [{ id: 's', command: 'echo hi', approval: true }] }], + }), 'utf8'); + await assert.rejects(() => loadWorkflowFile(filePath), /cannot contain approval or input/); +}); + +test('for_each validation rejects duplicate sub-step ids', async () => { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const filePath = path.join(tmpDir, 'bad.lobster'); + await fsp.writeFile(filePath, JSON.stringify({ + steps: [{ + id: 'loop', + for_each: '$x.json', + steps: [ + { id: 'dup', command: 'echo a' }, + { id: 'dup', command: 'echo b' }, + ], + }], + }), 'utf8'); + await assert.rejects(() => loadWorkflowFile(filePath), /duplicate for_each sub-step id/); +}); + +test('for_each validation rejects item_var/index_var collisions', async () => { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const filePath = path.join(tmpDir, 'bad.lobster'); + await fsp.writeFile(filePath, JSON.stringify({ + steps: [{ + id: 'loop', + for_each: '$x.json', + item_var: 'x', + index_var: 'x', + steps: [{ id: 's', command: 'echo hi' }], + }], + }), 'utf8'); + await assert.rejects(() => loadWorkflowFile(filePath), /item_var and index_var cannot be the same/); +}); + +test('for_each pause_ms and batch_size are accepted and executable', async () => { + const result = await runWorkflow({ + steps: [ + { id: 'vals', command: 'node -e "process.stdout.write(JSON.stringify([1,2,3]))"' }, + { + id: 'loop', + for_each: '$vals.json', + batch_size: 2, + pause_ms: 10, + steps: [{ id: 'emit', command: 'node -e "process.stdout.write(JSON.stringify({v:$item.json}))"' }], + }, + ], + }); + assert.equal(result.status, 'ok'); + assert.equal((result.output as any[]).length, 3); +}); + +test('for_each dry-run renders loop structure', async () => { + const workflow = { + steps: [ + { id: 'vals', command: 'node -e "process.stdout.write(JSON.stringify([1,2]))"' }, + { id: 'loop', for_each: '$vals.json', batch_size: 2, steps: [{ id: 'emit', command: 'echo hi' }] }, + ], + }; + + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-foreach-')); + const stateDir = path.join(tmpDir, 'state'); + const filePath = path.join(tmpDir, 'workflow.lobster'); + await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), 'utf8'); + + const stderr = new PassThrough(); + let out = ''; + stderr.on('data', (d: Buffer | string) => { out += String(d); }); + + await runWorkflowFile({ + filePath, + ctx: { + stdin: process.stdin, + stdout: process.stdout, + stderr, + env: { ...process.env, LOBSTER_STATE_DIR: stateDir }, + mode: 'tool', + dryRun: true, + registry: createDefaultRegistry(), + }, + }); + + assert.match(out, /\[for_each\]/); + assert.match(out, /sub-steps: 1/); + assert.match(out, /batch_size: 2/); +}); diff --git a/test/workflow_composition.test.ts b/test/workflow_composition.test.ts index c11e6f6..9f72127 100644 --- a/test/workflow_composition.test.ts +++ b/test/workflow_composition.test.ts @@ -113,7 +113,10 @@ test('workflow validation rejects workflow combined with run', async () => { await fsp.writeFile(filePath, JSON.stringify({ steps: [{ id: 'x', workflow: 'child.lobster', run: 'echo hi' }], }), 'utf8'); - await assert.rejects(() => loadWorkflowFile(filePath), /can only define one of run, command, pipeline, workflow, or parallel/); + await assert.rejects( + () => loadWorkflowFile(filePath), + /can only define one of run, command, pipeline, workflow, parallel, or for_each/, + ); }); test('workflow validation rejects workflow combined with pipeline', async () => { @@ -122,7 +125,10 @@ test('workflow validation rejects workflow combined with pipeline', async () => await fsp.writeFile(filePath, JSON.stringify({ steps: [{ id: 'x', workflow: 'child.lobster', pipeline: 'json' }], }), 'utf8'); - await assert.rejects(() => loadWorkflowFile(filePath), /can only define one of run, command, pipeline, workflow, or parallel/); + await assert.rejects( + () => loadWorkflowFile(filePath), + /can only define one of run, command, pipeline, workflow, parallel, or for_each/, + ); }); test('workflow validation rejects blank workflow path', async () => {