Run one worker per item

Make a Workflow that runs one worker Machine for each item of work, with pool() and source().

In this example, the items are the open issues of a GitHub Repo that have the label jr2. The worker is the shipped Machine task. Each worker works on one issue in its own Sandbox. Then it waits at a Gate until a person approves its work.

The Workflow

The example uses @jr2/machines. Add it to the Instance at the Kit version. <version> is the version of @jr2/orchestrator in your package.json:

npm install --save-exact @jr2/machines@<version>

Save this file as workflows/issues.ts. Replace you/app with your Repo, in the two places where it occurs.

The imports:

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

The model of the tab:

const model = "anthropic/claude-haiku-5-5";

The Source, the worker and the Pool:

// One work item: an open issue of the Repo.
type Issue = { number: number; title: string; body: string };

// The open issues with the label `jr2`: the first page only, at most 100. The Repo must be public,
// because the request has no token.
const ISSUES = "https://api.github.com/repos/you/app/issues?labels=jr2&state=open&per_page=100";

// The issues that this Source gave to a worker. A worker has each one now, or it finished it.
// The set is in the memory of the Orchestrator, and a restart empties it.
const claimed = new Set<number>();

// The Source: give the next open issue that no worker has had.
const issues = source<Issue>({
  next: fromPromise<Issue | null, { active: string[] }>(async ({ input }) => {
    const res = await fetch(ISSUES, { headers: { accept: "application/vnd.github+json" } }).catch(() => undefined);
    if (!res?.ok) {
      // While workers run, look again later. With no worker, stop the run with the error.
      if (input.active.length > 0) return null;
      throw new Error(`GET ${ISSUES} failed: ${res ? res.status : "no answer"}`);
    }
    const open = (await res.json()) as Array<{
      number: number;
      title: string;
      body: string | null;
      pull_request?: unknown;
    }>;
    // The issues API also lists pull requests. Skip them, and skip each issue that a worker has now.
    const issue = open.find(
      (i) => i.pull_request === undefined && !claimed.has(i.number) && !input.active.includes(`issue-${i.number}`),
    );
    if (!issue) return null;
    claimed.add(issue.number);
    return { number: issue.number, title: issue.title, body: issue.body ?? "" };
  }),
  // Send this event to the `source` Gate when you add the label to an issue.
  wake: defineEvent({
    name: "issues_changed",
    description: "An issue got the label jr2. Look for work again.",
    audience: "external",
    input: z.object({}),
  }),
  // While workers run and no issue is ready, look again every 5 minutes.
  pollEvery: 300_000,
});

// The worker: `task`, with its two Open parts bound.
const fixer = customize(task, {
  repos: { app: { url: "https://github.com/you/app.git" } },
  agents: { coder: { model } },
});

// The run input of the Pool. A worker does not get it.
const door = z.object({
  workers: z.number().int().min(1).max(4).default(2).describe("How many issues to work on at once."),
});

// The Pool: one `fixer` for each issue, at most `workers` at once.
export const machine = pool(fixer, {
  id: "issues",
  input: door,
  source: issues,
  itemId: (issue: Issue) => `issue-${issue.number}`,
  cap: ({ input }) => input.workers,
  // A `task` inside a Pool needs a branch. Make one from the issue number.
  itemInput: (issue: Issue) => ({
    prompt: `Fix issue #${issue.number} of this Repo: ${issue.title}\n\n${issue.body}`,
    branch: `jr2/issue-${issue.number}`,
  }),
});

The file has five parts:

• The Source. next gets the open issues from the GitHub REST API. It gives the first issue that no worker has had. When no issue is ready, it gives null. The request has no token, so the Repo must be public. Without a token, GitHub allows 60 requests in each hour. The example reads only the first page of issues: at most 100. When the request fails while workers run, next gives null in place of the error, so that the failure does not stop the run.
• The record of the Source. next keeps each issue that it gives in the Set claimed, and skips the issues in active. The Set is in the memory of the Orchestrator, and each run of the Workflow uses the same Set. A restart of the Orchestrator empties it. Then next can give a finished issue again, but never an issue that a worker of this run has now.
• The worker. fixer is task with its two Open parts bound: the Repo and the model of the coder. Bind a shipped Machine's Open parts tells more.
• The run input. door is the run input of the Pool. cap reads workers from it. A worker never gets it.
• The Pool. itemId gives the id of each item. itemInput makes the input of the worker from the item: the prompt and the branch.

Run jr2 up to converge the Instance with the new Workflow:

jr2 up

…
jr2 up — instance "my-instance" → context kind-jr2 / namespace my-instance
typecheck: tsc --noEmit (the instance's own compiler)
failures: workflow "issues" — the states that fail the run (ADR-0073)
  issues.failing: invokes fail
  task.lost: invokes fail
…
seats: workflow "issues" — who may send what, per state (ADR-0015)
  task.working agent:coder: [finish]
  task.review gate: [approve, request_changes]
…
agents: coder, triager (carried by 4 workflow machine(s))
…
sandbox nodes: jr2-control-plane
…
converged — `jr2 run <workflow>` when ready

The first line, …, is the line → .env:, with the keys that the CLI took from .env. Each jr2 command prints it first. The output on this page shows … in its place.

Start the run

Give the label jr2 to one open issue or more of your Repo. With no such issue, the run ends at once, as drained.

Start a run in one terminal. workers is the largest number of issues that the run works on at the same time:

jr2 run issues --input '{"workers":2}'

…
→ context kind-jr2 / namespace my-instance
run e1effd33-9bf6-4137-ae70-3aa504910784 (issues) — attached
  → active "discovering"
  → active "discovering"
  → active "discovering"
…

The examples on this page use the Instance my-instance in the kubectl context kind-jr2. Your run id, instance id and assembly hash are different. Give each jr2 command in the Instance folder. Give the commands in the next steps in a second terminal. In them, <run id> is the run id, or a unique prefix of it, 4 characters or more.

jr2 run stays attached until the run ends. The run ends when no issue is ready and no worker runs.

Find the Gates of the workers

Run jr2 status:

jr2 status <run id>

…
→ context kind-jr2 / namespace my-instance
{"runId":"e1effd33-9bf6-4137-ae70-3aa504910784","workflow":"issues",…,"gates":[{"gate":"source",…},{"gate":"issue-3.body.review",…},{"gate":"issue-2.body.review",…}]}

In the JSON that it prints, gates holds each open Gate of the run:

• source: the Gate of the wake event of the Source. It is open for the whole run.
• issue-<number>.body.review: the review Gate of the worker for that issue. It opens when the coder finishes, or when its Turn fails. The id starts with the item id. body is the slot of the Machine that task wraps.

The meta of a review Gate gives the branch and the worktree of the worker. The name of the Sandbox of a worker holds its item id: ws-<first 8 characters of the run id>-<item id>-<hash>. Look inside a Sandbox tells how to read the work before you approve it.

Approve the work of one worker

Push the branch before you approve. approve ends that worker and removes its Sandbox, and the commits that you did not push are lost. Look inside a Sandbox tells how to push.

Then send approve to the Gate of the worker. This example approves issue 2:

jr2 send <run id> --gate issue-2.body.review --event approve

…
→ context kind-jr2 / namespace my-instance
delivered approve to gate "issue-2.body.review" on e1effd33-9bf6-4137-ae70-3aa504910784

The Pool then calls next again, and starts a worker for the next issue.

The review Gate also takes request_changes, with --input '{"notes":"<text>"}'. Then the coder continues with the notes, and the Gate opens again when it finishes. Answer a Gate gives the other ways to answer a Gate.

Tell the Pool about new work

While workers run and no issue is ready, the Pool waits. When you add the label to an issue, send the wake event to the source Gate. The Pool then calls next at once:

jr2 send <run id> --gate source --event issues_changed

…
→ context kind-jr2 / namespace my-instance
delivered issues_changed to gate "source" on e1effd33-9bf6-4137-ae70-3aa504910784

Without the event, the Pool calls next again when a worker finishes, or after pollEvery. A webhook can send the same event with POST /runs/:runId/gates/:gate/events.

The end of the run

When the last worker finishes and no issue is ready, the run ends. In the first terminal, jr2 run prints the last state of the Pool, then the status of the run as JSON:

…
  → active "discovering"
  → done "drained"
{"runId":"e1effd33-9bf6-4137-ae70-3aa504910784","workflow":"issues","instanceId":"db9712c8-7c19-4d18-8fb1-8c01d326808f","status":"done","value":"drained","context":{"runInput":{"workers":2,"instanceId":"db9712c8-7c19-4d18-8fb1-8c01d326808f"},"cap":2,"active":[],"items":{"issue-2":{"branch":"jr2/issue-2"},"issue-3":{"branch":"jr2/issue-3"},"issue-1":{"branch":"jr2/issue-1"}},"failures":{},"status":"drained"},"children":[],"waiting":[],"held":[],"assembly":"ca3b0e760c85","output":{"status":"drained","items":{"issue-2":{"branch":"jr2/issue-2"},"issue-3":{"branch":"jr2/issue-3"},"issue-1":{"branch":"jr2/issue-1"}},"failures":{}}}

In the JSON, context.items holds the output of each worker that finished, by item id. context.failures holds the Failure of each worker that failed, by item id: here, none. For this Source, context.status is always drained, because next never gives open. Pools and Sources tells what each status means.

Rules

• itemId must give each item an id of its own, and the same id each time. It is the id of the worker, the key in items, and the start of the id of each Gate of the worker.
• next must not give an item that a worker has now. jr2 does not refuse the item when next gives it. A second worker starts in the same Sandbox and the same Worktree as the first. When the two workers register the same Gate or Agent, the run stops with the status error, and its fault is registration address "<address>" is already live (two concurrent registrations share an id). For task, <address> is gate/<run id>/<item id>.body.review or agent/<run id>/<item id>.body/coder.
• next gets only active: the ids of the items that workers have now. To not give a finished item again, the Source must keep the items that it gave.
• When a worker fails, for example when the Sandbox of a task is lost, the Pool keeps the Failure in failures and goes on. Only the Sandbox of that worker is deleted. The other workers, and the work in their Sandboxes, stay. A defect in a worker is different: the run stops with the status error.
• When the Pool ends with one or more failures, drained or deadlocked, the run ends failed, with the reason <n> of <m> items failed, and jr2 run exits 1. The detail of the Failure is { status, items, failures }. With the option failures: "report", the run ends done, and the failures are in its output.
• When next rejects, the run stops. Its status is error, and its fault is the message of the error.
• pool() does not carry the item type of the Source to itemId and itemInput. Write the type on their parameter, as in (issue: Issue).
• A worker gets its item, or what itemInput makes of it. It never gets the run input. cap and itemInput get the run input as { input }, typed by the input schema of the Pool.
• The Orchestrator refuses a run input that does not match the input schema of the Pool. jr2 run prints error: invalid input for workflow "<workflow>": <reason>.
• A task inside a Pool needs branch in its input. Without it, the run stops with the status error, and its fault starts with task: no `branch` on the door and no Instance ID to derive one from.
• Without cap, one worker runs at a time. Each task worker has its own Workspace. Thus the run has cap Sandboxes or fewer at one time.
• The Pool takes the wake event only while it waits: workers run, and no item is ready. At other times, jr2 send refuses it: error: gate "source" is open, and the run does not take "<event>" now: no writing of it would move the run with this payload — not delivered (takes now: nothing).
• onDrained takes only "final": the run ends.

Read next

• Pools and Sources
• Composition
• Gates
• Reference: pool(), source(), PoolSpec, SourceSpec, task
• ADR-0017: the run loop is a Pool over a Source