feat: add for_each workflow step type

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
 

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<WorkflowFile>
         }
       }
     }
+    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<string>();
+      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<string, WorkflowStepResult> = { ...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<string, unknown> = { [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<void> {
+  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<typeof setTimeout>;
+    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;

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/);
+});

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 () => {