feat: add parallel workflow step execution

CHANGELOG.md

@@ -9,6 +9,7 @@ All notable changes to Lobster will be documented in this file.
 - Add per-step workflow `timeout_ms` handling, including timeout-triggered aborts, `SIGKILL` for timed shell steps, and dry-run annotations. Thanks to [@scottgl9](https://github.com/scottgl9) (PR [#74](https://github.com/openclaw/lobster/pull/74)).
 - 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)).
 
 ## 2026.4.6
 

src/workflows/file.ts

@@ -26,6 +26,22 @@ export type WorkflowFile = {
   cost_limit?: CostLimit;
 };
 
+export type ParallelBranch = {
+  id: string;
+  run?: string;
+  command?: string;
+  pipeline?: string;
+  env?: Record<string, string>;
+  cwd?: string;
+  stdin?: unknown;
+};
+
+export type ParallelConfig = {
+  wait?: 'all' | 'any';
+  timeout_ms?: number;
+  branches: ParallelBranch[];
+};
+
 export type WorkflowStep = {
   id: string;
   command?: string;
@@ -40,6 +56,7 @@ export type WorkflowStep = {
   input?: WorkflowInputRequest;
   condition?: unknown;
   when?: unknown;
+  parallel?: ParallelConfig;
   timeout_ms?: number;
   on_error?: 'stop' | 'continue' | 'skip_rest';
 };
@@ -192,18 +209,84 @@ export async function loadWorkflowFile(filePath: string): Promise<WorkflowFile>
         throw new Error(`Workflow step ${step.id} workflow_args must be a plain object`);
       }
     }
+    if (step.parallel !== undefined && (!step.parallel || typeof step.parallel !== 'object' || Array.isArray(step.parallel))) {
+      throw new Error(`Workflow step ${step.id} parallel must be an object`);
+    }
+    const isParallel = Boolean(step.parallel && typeof step.parallel === 'object' && !Array.isArray(step.parallel));
+    if (isParallel) {
+      const parallel = step.parallel as ParallelConfig;
+      if (!Array.isArray(parallel.branches) || parallel.branches.length === 0) {
+        throw new Error(`Workflow step ${step.id} parallel requires a non-empty branches array`);
+      }
+      if (parallel.wait !== undefined && parallel.wait !== 'all' && parallel.wait !== 'any') {
+        throw new Error(`Workflow step ${step.id} parallel wait must be "all" or "any"`);
+      }
+      if (
+        parallel.timeout_ms !== undefined
+        && (
+          typeof parallel.timeout_ms !== 'number'
+          || !Number.isFinite(parallel.timeout_ms)
+          || !Number.isInteger(parallel.timeout_ms)
+          || parallel.timeout_ms < 1
+          || parallel.timeout_ms > 2_147_483_647
+        )
+      ) {
+        throw new Error(`Workflow step ${step.id} parallel timeout_ms must be a positive integer between 1 and 2147483647`);
+      }
+
+      const branchIds = new Set<string>();
+      for (const branch of parallel.branches) {
+        if (!branch || typeof branch !== 'object') {
+          throw new Error(`Workflow step ${step.id} parallel branches must be objects`);
+        }
+        if (!branch.id || typeof branch.id !== 'string') {
+          throw new Error(`Workflow step ${step.id} parallel branch requires an id`);
+        }
+        if (branch.id === step.id) {
+          throw new Error(`Workflow step ${step.id} parallel branch id cannot match the step id`);
+        }
+        if (branchIds.has(branch.id)) {
+          throw new Error(`Workflow step ${step.id} duplicate parallel branch id: ${branch.id}`);
+        }
+        if (seen.has(branch.id)) {
+          throw new Error(`Duplicate workflow id across steps/parallel branches: ${branch.id}`);
+        }
+        branchIds.add(branch.id);
+        const branchShell = typeof branch.run === 'string' ? branch.run : branch.command;
+        const branchPipeline = typeof branch.pipeline === 'string' ? branch.pipeline : undefined;
+        const branchExecCount = Number(Boolean(branchShell)) + Number(Boolean(branchPipeline));
+        if (branchExecCount === 0) {
+          throw new Error(`Workflow step ${step.id} parallel branch ${branch.id} requires run, command, or pipeline`);
+        }
+        if (branchExecCount > 1) {
+          throw new Error(`Workflow step ${step.id} parallel branch ${branch.id} can only define one of run, command, or pipeline`);
+        }
+        if (branch.run !== undefined && typeof branch.run !== 'string') {
+          throw new Error(`Workflow step ${step.id} parallel branch ${branch.id} run must be a string`);
+        }
+        if (branch.command !== undefined && typeof branch.command !== 'string') {
+          throw new Error(`Workflow step ${step.id} parallel branch ${branch.id} command must be a string`);
+        }
+        if (branch.pipeline !== undefined && typeof branch.pipeline !== 'string') {
+          throw new Error(`Workflow step ${step.id} parallel branch ${branch.id} pipeline must be a string`);
+        }
+      }
+    }
     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));
+    const executionCount = Number(Boolean(shellCommand))
+      + Number(Boolean(pipeline))
+      + Number(Boolean(workflowRef))
+      + Number(isParallel);
     if (executionCount === 0 && !isApprovalStep(step.approval) && !isInputStep(step.input)) {
-      throw new Error(`Workflow step ${step.id} requires run, command, pipeline, workflow, approval, or input`);
+      throw new Error(`Workflow step ${step.id} requires run, command, pipeline, workflow, parallel, approval, or input`);
     }
     if (executionCount > 1) {
-      throw new Error(`Workflow step ${step.id} can only define one of run, command, pipeline, or workflow`);
+      throw new Error(`Workflow step ${step.id} can only define one of run, command, pipeline, workflow, or parallel`);
     }
     if (executionCount > 0 && isInputStep(step.input)) {
-      throw new Error(`Workflow step ${step.id} input steps cannot define run, command, pipeline, or workflow`);
+      throw new Error(`Workflow step ${step.id} input steps cannot define run, command, pipeline, workflow, or parallel`);
     }
     if (isApprovalStep(step.approval) && isInputStep(step.input)) {
       throw new Error(`Workflow step ${step.id} cannot define both approval and input`);
@@ -256,6 +339,12 @@ export async function loadWorkflowFile(filePath: string): Promise<WorkflowFile>
     if (seen.has(step.id)) {
       throw new Error(`Duplicate workflow step id: ${step.id}`);
     }
+    if (isParallel) {
+      const parallel = step.parallel as ParallelConfig;
+      for (const branch of parallel.branches) {
+        seen.add(branch.id);
+      }
+    }
     seen.add(step.id);
   }
 
@@ -488,8 +577,109 @@ export async function runWorkflowFile({
     }
 
     let result: WorkflowStepResult;
+    let parallelBranchResults: Record<string, WorkflowStepResult> | null = null;
     try {
-      if (execution.kind === 'workflow') {
+      if (execution.kind === 'parallel') {
+        const parallel = execution.value;
+        const wait = parallel.wait ?? 'all';
+        const branchAbortController = new AbortController();
+        const branchSignal = stepSignal
+          ? AbortSignal.any([stepSignal, branchAbortController.signal])
+          : branchAbortController.signal;
+        const shouldForceKill = Boolean(step.timeout_ms || parallel.timeout_ms);
+        const runBranch = async (branch: ParallelBranch): Promise<{ branchId: string; result: WorkflowStepResult }> => {
+          const mergedBranchEnv = { ...(step.env ?? {}), ...(branch.env ?? {}) };
+          const branchEnv = mergeEnv(ctx.env, workflow.env, mergedBranchEnv, resolvedArgs, results);
+          const branchCwd = resolveCwd(branch.cwd ?? step.cwd ?? workflow.cwd, resolvedArgs) ?? ctx.cwd;
+          const branchShell = typeof branch.run === 'string' ? branch.run : branch.command;
+          const branchExec = typeof branch.pipeline === 'string' && branch.pipeline.trim()
+            ? { kind: 'pipeline' as const, value: branch.pipeline }
+            : (typeof branchShell === 'string' && branchShell.trim()
+              ? { kind: 'shell' as const, value: branchShell }
+              : { kind: 'none' as const });
+
+          if (branchExec.kind === 'shell') {
+            const command = resolveTemplate(branchExec.value, resolvedArgs, results);
+            const stdinValue = resolveShellStdin(branch.stdin, resolvedArgs, results);
+            const { stdout } = await runShellCommand({
+              command,
+              stdin: stdinValue,
+              env: branchEnv,
+              cwd: branchCwd,
+              signal: branchSignal,
+              ...(shouldForceKill ? { killSignal: 'SIGKILL' as NodeJS.Signals } : {}),
+            });
+            return { branchId: branch.id, result: { id: branch.id, stdout, json: parseJson(stdout) } };
+          }
+
+          if (branchExec.kind === 'pipeline') {
+            if (!ctx.registry) {
+              throw new Error(`Parallel branch ${branch.id} requires a command registry for pipeline execution`);
+            }
+            const pipelineText = resolveTemplate(branchExec.value, resolvedArgs, results);
+            const inputValue = resolveInputValue(branch.stdin, resolvedArgs, results);
+            const branchResult = await runPipelineStep({
+              stepId: branch.id,
+              pipelineText,
+              inputValue,
+              ctx: { ...ctx, signal: branchSignal },
+              env: branchEnv,
+              cwd: branchCwd,
+            });
+            return { branchId: branch.id, result: branchResult };
+          }
+
+          return { branchId: branch.id, result: { id: branch.id } };
+        };
+
+        let parallelTimeoutId: ReturnType<typeof setTimeout> | undefined;
+        const timeoutPromise = parallel.timeout_ms
+          ? new Promise<never>((_resolve, reject) => {
+            parallelTimeoutId = setTimeout(() => {
+              branchAbortController.abort();
+              reject(new Error(`Parallel step ${step.id} timed out after ${parallel.timeout_ms}ms`));
+            }, parallel.timeout_ms);
+          })
+          : null;
+
+        try {
+          if (wait === 'any') {
+            const branchPromises = parallel.branches.map((branch) => runBranch(branch));
+            const winner = await (timeoutPromise
+              ? Promise.race([...branchPromises, timeoutPromise])
+              : Promise.race(branchPromises)) as { branchId: string; result: WorkflowStepResult };
+            parallelBranchResults = { [winner.branchId]: winner.result };
+            branchAbortController.abort();
+          } else {
+            const branchPromises = parallel.branches.map((branch) => runBranch(branch));
+            const settled = await (timeoutPromise
+              ? Promise.race([Promise.allSettled(branchPromises), timeoutPromise])
+              : Promise.allSettled(branchPromises)) as PromiseSettledResult<{ branchId: string; result: WorkflowStepResult }>[];
+
+            parallelBranchResults = {};
+            for (const entry of settled) {
+              if (entry.status === 'rejected') {
+                throw new Error(`Parallel branch failed: ${entry.reason?.message ?? String(entry.reason)}`);
+              }
+              parallelBranchResults[entry.value.branchId] = entry.value.result;
+            }
+          }
+        } finally {
+          if (parallelTimeoutId !== undefined) clearTimeout(parallelTimeoutId);
+        }
+
+        const merged: Record<string, unknown> = {};
+        for (const [branchId, branchResult] of Object.entries(parallelBranchResults ?? {})) {
+          merged[branchId] = branchResult.json;
+        }
+        result = {
+          id: step.id,
+          json: merged,
+          stdout: wait === 'any'
+            ? (Object.values(parallelBranchResults ?? {})[0]?.stdout ?? '')
+            : JSON.stringify(merged),
+        };
+      } else if (execution.kind === 'workflow') {
         const workflowPath = resolveTemplate(execution.value, resolvedArgs, results);
         const resolvedWorkflowPath = path.isAbsolute(workflowPath)
           ? workflowPath
@@ -597,6 +787,12 @@ export async function runWorkflowFile({
       if (timeoutId !== undefined) clearTimeout(timeoutId);
     }
 
+    if (parallelBranchResults) {
+      for (const [branchId, branchResult] of Object.entries(parallelBranchResults)) {
+        results[branchId] = branchResult;
+        trackStepCost(costTracker, branchId, branchResult);
+      }
+    }
     results[step.id] = result;
     lastStepId = step.id;
 
@@ -773,7 +969,38 @@ function dryRunWorkflow({
     // rather than silently collapsing the reference to an empty string.
     const stdinNote = dryRunStdinNote(step.stdin);
 
-    if (execution.kind === 'workflow') {
+    if (execution.kind === 'parallel') {
+      lines.push(`  ${num}. ${step.id}  [parallel]`);
+      lines.push(`     wait: ${step.parallel?.wait ?? 'all'}`);
+      if (step.parallel?.timeout_ms) {
+        lines.push(`     timeout: ${step.parallel.timeout_ms}ms`);
+      }
+      for (const branch of step.parallel?.branches ?? []) {
+        const branchShell = typeof branch.run === 'string' ? branch.run : branch.command;
+        if (typeof branch.pipeline === 'string' && branch.pipeline.trim()) {
+          const pipelineText = resolveDryRunTemplate(branch.pipeline, resolvedArgs, results);
+          const pipelineNote = dryRunTemplateNote(pipelineText);
+          if (!ctx.registry) {
+            throw new Error(`Parallel branch ${branch.id} requires a command registry for pipeline execution`);
+          }
+          const stages = parsePipeline(pipelineText);
+          for (const stage of stages) {
+            if (hasDeferredDryRunStageName(stage.name)) continue;
+            if (!ctx.registry.get(stage.name)) {
+              throw new Error(`Parallel branch ${branch.id} pipeline references unknown command: ${stage.name}`);
+            }
+          }
+          lines.push(`     branch ${branch.id}: [pipeline] ${pipelineText}${pipelineNote ? `  ${pipelineNote}` : ''}`);
+        } else if (typeof branchShell === 'string' && branchShell.trim()) {
+          const command = resolveDryRunTemplate(branchShell, resolvedArgs, results);
+          const commandNote = dryRunTemplateNote(command);
+          lines.push(`     branch ${branch.id}: [shell] ${command}${commandNote ? `  ${commandNote}` : ''}`);
+        } else {
+          lines.push(`     branch ${branch.id}: [no-op]`);
+        }
+        results[branch.id] = { id: branch.id };
+      }
+    } else if (execution.kind === 'workflow') {
       const workflowPath = resolveDryRunTemplate(execution.value, resolvedArgs, results);
       const pathNote = dryRunTemplateNote(workflowPath);
       lines.push(`  ${num}. ${step.id}  [workflow]`);
@@ -1635,6 +1862,10 @@ async function runShellCommand({
 }
 
 function getStepExecution(step: WorkflowStep) {
+  if (step.parallel && typeof step.parallel === 'object' && !Array.isArray(step.parallel)) {
+    return { kind: 'parallel' as const, value: step.parallel };
+  }
+
   if (typeof step.workflow === 'string' && step.workflow.trim()) {
     return { kind: 'workflow' as const, value: step.workflow };
   }

test/parallel.test.ts

@@ -0,0 +1,175 @@
+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 { 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-parallel-'));
+  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('parallel wait=all runs all branches and merges output', async () => {
+  const result = await runWorkflow({
+    steps: [
+      {
+        id: 'fetch',
+        parallel: {
+          wait: 'all',
+          branches: [
+            { id: 'a', command: 'node -e "process.stdout.write(JSON.stringify({src:\\"a\\"}))"' },
+            { id: 'b', command: 'node -e "process.stdout.write(JSON.stringify({src:\\"b\\"}))"' },
+            { id: 'c', command: 'node -e "process.stdout.write(JSON.stringify({src:\\"c\\"}))"' },
+          ],
+        },
+      },
+    ],
+  });
+
+  assert.equal(result.status, 'ok');
+  const output = result.output as any[];
+  assert.equal(output[0].a.src, 'a');
+  assert.equal(output[0].b.src, 'b');
+  assert.equal(output[0].c.src, 'c');
+});
+
+test('parallel wait=any returns first branch result', async () => {
+  const result = await runWorkflow({
+    steps: [
+      {
+        id: 'race',
+        parallel: {
+          wait: 'any',
+          branches: [
+            { id: 'fast', command: 'node -e "process.stdout.write(JSON.stringify({winner:true}))"' },
+            { id: 'slow', command: 'node -e "setTimeout(() => process.stdout.write(JSON.stringify({winner:false})), 5000)"' },
+          ],
+        },
+      },
+    ],
+  });
+  assert.equal(result.status, 'ok');
+  const output = result.output as any[];
+  assert.equal(Object.keys(output[0]).length, 1);
+  assert.equal(output[0].fast.winner, true);
+});
+
+test('parallel branch results are available to later steps by branch id', async () => {
+  const result = await runWorkflow({
+    steps: [
+      {
+        id: 'fetch',
+        parallel: {
+          wait: 'all',
+          branches: [
+            { id: 'x', command: 'node -e "process.stdout.write(JSON.stringify({val:10}))"' },
+            { id: 'y', command: 'node -e "process.stdout.write(JSON.stringify({val:20}))"' },
+          ],
+        },
+      },
+      {
+        id: 'use',
+        command: 'node -e "process.stdout.write(JSON.stringify({x:$x.json.val,y:$y.json.val}))"',
+      },
+    ],
+  });
+  assert.equal(result.status, 'ok');
+  assert.deepEqual(result.output, [{ x: 10, y: 20 }]);
+});
+
+test('parallel wait=all propagates branch failure', async () => {
+  await assert.rejects(
+    () => runWorkflow({
+      steps: [
+        {
+          id: 'p',
+          parallel: {
+            wait: 'all',
+            branches: [
+              { id: 'ok', command: 'echo ok' },
+              { id: 'fail', command: 'node -e "process.exit(1)"' },
+            ],
+          },
+        },
+      ],
+    }),
+    /Parallel branch failed/,
+  );
+});
+
+test('parallel validation rejects empty branches', async () => {
+  const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-parallel-'));
+  const filePath = path.join(tmpDir, 'bad.lobster');
+  await fsp.writeFile(filePath, JSON.stringify({
+    steps: [{ id: 'p', parallel: { branches: [] } }],
+  }), 'utf8');
+  await assert.rejects(() => loadWorkflowFile(filePath), /non-empty branches/);
+});
+
+test('parallel validation rejects duplicate branch ids', async () => {
+  const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-parallel-'));
+  const filePath = path.join(tmpDir, 'bad.lobster');
+  await fsp.writeFile(filePath, JSON.stringify({
+    steps: [{
+      id: 'p',
+      parallel: {
+        branches: [
+          { id: 'dup', command: 'echo a' },
+          { id: 'dup', command: 'echo b' },
+        ],
+      },
+    }],
+  }), 'utf8');
+  await assert.rejects(() => loadWorkflowFile(filePath), /duplicate parallel branch id/);
+});
+
+test('parallel validation rejects branch without execution', async () => {
+  const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-parallel-'));
+  const filePath = path.join(tmpDir, 'bad.lobster');
+  await fsp.writeFile(filePath, JSON.stringify({
+    steps: [{
+      id: 'p',
+      parallel: {
+        branches: [{ id: 'empty' }],
+      },
+    }],
+  }), 'utf8');
+  await assert.rejects(() => loadWorkflowFile(filePath), /requires run, command, or pipeline/);
+});
+
+test('parallel timeout aborts block', async () => {
+  await assert.rejects(
+    () => runWorkflow({
+      steps: [
+        {
+          id: 'p',
+          parallel: {
+            wait: 'all',
+            timeout_ms: 100,
+            branches: [
+              { id: 'slow', command: 'node -e "setTimeout(() => process.stdout.write(\'ok\'), 5000)"' },
+            ],
+          },
+        },
+      ],
+    }),
+    /Parallel step p timed out after 100ms/,
+  );
+});

test/workflow_composition.test.ts

@@ -113,7 +113,7 @@ 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, or workflow/);
+  await assert.rejects(() => loadWorkflowFile(filePath), /can only define one of run, command, pipeline, workflow, or parallel/);
 });
 
 test('workflow validation rejects workflow combined with pipeline', async () => {
@@ -122,7 +122,7 @@ 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, or workflow/);
+  await assert.rejects(() => loadWorkflowFile(filePath), /can only define one of run, command, pipeline, workflow, or parallel/);
 });
 
 test('workflow validation rejects blank workflow path', async () => {