Concurrent batch mode

Use batch mode when you have N independent work items and each deserves its own session document — its own history, token accounting, run status and error log. Extracting fifty invoices, scoring a queue of documents, running the same interview against a list of subjects.

Selecting the coordinator

Batch mode is chosen by the first request’s config, not by a different flow class:

{
  "flowName": "InvoiceFlow",
  "message": "start",
  "config": { "_concurrent": true }
}

Flow.run() reads that flag from flow context and takes the other branch:

const isConcurrent = this.getContext<boolean>("config._concurrent");
const resp = isConcurrent
  ? await this.spawnSteps()
  : await this.requireCurrentStep().run(message);

The same flow class is therefore both the coordinator and the worker. A worker is simply an ordinary run of that flow, started with a config that does not contain _concurrent.

spawnSteps()

Override spawnSteps() to describe the fan-out. Its return value becomes the coordinator’s response message.

protected async spawnSteps(): Promise<string> {
  const step = await this.goto(PresidentStep);
  const nths = ["10th", "11th", "12th", "13th", "14th", "15th", "16th"];

  await this.concurrentSteps<string>({
    items: nths,
    batchSize: 3,
    onConfig: (item) => ({ nth: item, isPresident: true }),
    onBotResponse(item, response) {
      step.saveState({ [item]: response["message"] });
    },
  });

  const msg = `Finished concurrent flow: ${this.id}`;
  new SessionLogger(this.getSessionDoc()).log(msg);
  step.sessionCompleted();
  return msg;
}

Note the last two lines. Logging and completion are explicit; nothing does them for you.

concurrentSteps()

public async concurrentSteps<T>({
  items,
  batchSize,
  onConfig,
  onBotResponse,
}: {
  items: T[];
  batchSize: number;
  onConfig: (item: T) => object;
  onBotResponse: (item: T, response: any) => void;
}): Promise<void>

Progress is written to the coordinator’s session log:

Processing batch 1/3 (batch size:3)
Batch 1 completed.

There is a second wrinkle. When a worker step sets a non-plain content type, the demo controller sends the raw payload instead of the standard envelope. InvoiceFlow workers return HttpContentType.Json, so response.data is the invoice JSON itself, not { success, completed, message, session }.

SELF_URL and worker sessions

Each item becomes an HTTP POST built by SelfClient, whose base URL is SELF_URL:

const body = {
  flowName: this.id,
  config: { ...onConfig(item) },
};
await selfCaller.post("", body);

SELF_URL=http://localhost:8000/ai/run

Because the path is empty, SELF_URL must be the complete run endpoint, not the service root. Three properties follow from this design:

1. No session ID is sent. Every worker creates a brand-new session document, bound to the same registered flow name and starting from that flow’s initial step.
2. Nothing is shared. The coordinator’s context, state and memory are invisible to a worker. Everything a worker needs must go through onConfig.
3. It is a real HTTP request. It crosses the load balancer, so in a multi-instance deployment the worker may run in a different process from the coordinator.

What the coordinator must own

concurrentSteps() provides batching and error logging. Everything else is yours.

A failed item never reaches onBotResponse. If you only count callbacks, a batch where every worker failed looks exactly like a batch that was never started.

InvoiceFlow.spawnSteps() illustrates the gap: it fans out correctly, but never marks the coordinator session complete, so that session remains running forever.

isBatch()

public isBatch(): boolean {
  return false;
}

Despite the name, this does not select spawnSteps(). Returning true makes the engine call saveSession() once before run() begins, so a long-running coordinator has a persisted document that its SessionLogger writes can land in. Only config._concurrent chooses the batch dispatch path.

Contrast with nested execution

The runStep() column describes a sequential call outside a parallel worker. Inside a runSteps() worker, runStep() uses an isolated, atomic one-child barrier and propagates a child failure to its caller.

Choose nested execution when child work belongs to the caller’s decision within one session. Choose batch mode when each item needs its own independently saved flow session.

A small runSteps() example

Suppose a registered InventoryLookupStep reads a SKU from this.getParallelInvocation().params. Its tool returns JSON through directResult(...). It returns each lookup as branch output, so repeated workers do not need to replace the same Step state field. The calling Step can collect available results and failures:

const batch = await this.runSteps(
  [
    { step: InventoryLookupStep, key: "shirt", params: { sku: "SHIRT-01" } },
    { step: InventoryLookupStep, key: "hat", params: { sku: "HAT-02" } },
  ],
  { maxConcurrency: 2 },
);

this.saveState({
  inventoryChecks: {
    results: batch.fulfilled.map((branch) => ({
      key: branch.key,
      output: branch.output,
    })),
    failures: batch.rejected.map((branch) => ({
      key: branch.key,
      error: branch.error,
    })),
  },
});

Distinct keys are required when a Step class appears more than once. Results remain in request order. This example records partial success; the caller must decide whether the available inventory checks are sufficient for its next action.

The default retain-successes policy publishes successful worker state at the join. { failurePolicy: "atomic" } suppresses that batch’s application-state publication if any branch fails; it still returns branch outcomes. Invalid shared mutations, conflicting writes, and other framework failures throw. Neither policy rolls back external side effects.

For nested calls, publication reaches the immediate caller’s private scope first; canonical session state changes only when the enclosing outer join accepts those updates. A join is also distinct from a durable save: checkpoint: "root-join" requests a session save after a root join applies state, but does not provide automatic per-child crash recovery.

Continue with Nested execution for state isolation, explicit reducer contributions, nested joins, and cancellation; the Step reference for API contracts; and Parallelism and fan-out for the contrast with LangGraph’s scheduling and recovery boundaries.

Failure modes

Related: Sessions, migration, batch mode, Raw JSON and batch fan-out, and Error handling and completion.