`pool`

Function, in @jr2/orchestrator/src/pool.ts. It is an export of @jr2/orchestrator, on the Authoring surface page, in Composition.

Overload 1

function pool<
  TWorker extends AnyStateMachine,
  TSchema extends z.ZodObject,
>(
  worker: TWorker,
  spec: PoolSpec<any, z.infer<TSchema>> & { input: TSchema },
): PoolMachine<TWorker, z.infer<TSchema>>

Make a Pool: a Machine that runs one worker Machine for each item that a Source gives, with at most cap workers at once. Use it as the root of a Workflow (export const machine = pool(…)), or compose it in another Machine.

The run ends when the Source has no item ready and no worker runs. Its output is { status, items, failures }: status is "drained" when no item is left, or "deadlocked" when items exist but none can start; items holds the output of each worker that got to a final state, by item id; failures holds the Failure of each worker that failed, by item id. A worker's Failure stops only that worker: the other workers and their Sandboxes stay. When the Pool ends with one or more failures, drained or deadlocked, it fails the run by default: the reason is "<n> of <m> items failed", and the detail is { status, items, failures }. With failures: "report", it ends done with that output. A worker's defect ends the whole run error. While workers wait on Gates, the run stays open.

customize() reaches through a Pool to its worker: customize(pool, { agents }) retunes the worker's Agents. The events that the Pool accepts are its wake event only; each worker keeps its own.

Parameters

worker: The Machine to run for each item. It gets the item, or what itemInput makes of it, as its input. It does not get the run input.
spec: The Source, the item id, the cap, and the worker's input. See PoolSpec.
- id (optional): string
  The Machine id: the name of the Pool in the Console. Default: "pool".
- input: z.ZodObject
  The run input of the Pool, as a zod object: what a caller sends to start a run. cap and itemInput read it, typed by it. A worker never gets it: each worker gets its item. Default: none, and then their input is unknown.
- source: SourceSpec<T>
  Where the items come from. Make it with source.
- itemId: (item: T) => string
  The id of an item. The Pool uses it as the id of the item's worker and in the active list that next gets. It is also the key of that worker's output in items. Give each item an id of its own, and the same id every time.
- cap (optional): number | ((args: { input: TInput; }) => number)
  The most workers that run at once: a number, or a function of the run input. Default: 1.
- itemInput (optional): (item: T, args: { input: TInput; }) => unknown
  Make a worker's input from its item and the run input. Default: the item itself.
- onDrained (optional): "final"
  What the Pool does when the Source has no more items and every worker has settled. Only "final": the run ends. Default: "final".
- failures (optional): "fail" | "report"
  What the Pool does at the end when one or more workers failed. The Pool catches each worker's Failure either way, so one failed item does not stop the other workers. "fail": when the Pool ends with failures, drained or deadlocked, it fails the run with the reason "<n> of <m> items failed" and the detail { status, items, failures }. "report": the Pool ends done, with the failures in its output. Default: "fail".

Returns: The Pool, as a Machine.

Example

import { fromPromise } from "xstate";
import { customize, pool, source } from "@jr2/orchestrator";
import { task } from "@jr2/machines";

type Issue = { id: string; title: string };

const backlog: Issue[] = [
  { id: "ISSUE-1", title: "Fix the greeting." },
  { id: "ISSUE-2", title: "Add a --name flag." },
];

const issues = source<Issue>({
  next: fromPromise(async ({ input }: { input: { active: string[] } }) => {
    const issue = backlog.find((i) => !input.active.includes(i.id)) ?? null;
    if (issue) backlog.splice(backlog.indexOf(issue), 1);
    return issue;
  }),
});

const fixer = customize(task, {
  repos: { app: { url: "https://github.com/you/app.git" } },
  agents: { coder: { model: "anthropic/claude-haiku-5-5" } },
});

export const machine = pool(fixer, {
  id: "issues",
  source: issues,
  itemId: (issue: Issue) => issue.id,
  cap: 2,
  itemInput: (issue: Issue) => ({ prompt: issue.title, branch: `fix/${issue.id}` }),
});

See also

Overload 2

function pool<
  TWorker extends AnyStateMachine,
  TInput = unknown,
>(
  worker: TWorker,
  spec: PoolSpec<any, TInput> & { input?: never },
): PoolMachine<TWorker, TInput>

Make a Pool with no input schema. A Pool is a Machine that runs one worker Machine for each item that a Source gives, with at most cap workers at once. See pool for the full description and an example.

Parameters

worker: The Machine to run for each item. It gets the item, or what itemInput makes of it, as its input. It does not get the run input.
spec: The Source, the item id, the cap, and the worker's input. See PoolSpec.
- id (optional): string
  The Machine id: the name of the Pool in the Console. Default: "pool".
- input (optional): z.ZodObject
  The run input of the Pool, as a zod object: what a caller sends to start a run. cap and itemInput read it, typed by it. A worker never gets it: each worker gets its item. Default: none, and then their input is unknown.
- source: SourceSpec<T>
  Where the items come from. Make it with source.
- itemId: (item: T) => string
  The id of an item. The Pool uses it as the id of the item's worker and in the active list that next gets. It is also the key of that worker's output in items. Give each item an id of its own, and the same id every time.
- cap (optional): number | ((args: { input: TInput; }) => number)
  The most workers that run at once: a number, or a function of the run input. Default: 1.
- itemInput (optional): (item: T, args: { input: TInput; }) => unknown
  Make a worker's input from its item and the run input. Default: the item itself.
- onDrained (optional): "final"
  What the Pool does when the Source has no more items and every worker has settled. Only "final": the run ends. Default: "final".
- failures (optional): "fail" | "report"
  What the Pool does at the end when one or more workers failed. The Pool catches each worker's Failure either way, so one failed item does not stop the other workers. "fail": when the Pool ends with failures, drained or deadlocked, it fails the run with the reason "<n> of <m> items failed" and the detail { status, items, failures }. "report": the Pool ends done, with the failures in its output. Default: "fail".

Returns: The Pool, as a Machine.

See also

pool ​

Overload 1 ​

Overload 2 ​

`pool`

Overload 1

Overload 2