Workflow API

The workflow builder, run handle, and queries.

queue.workflows
  .define(name: string)
  // ... steps ...
  .submit(): WorkflowHandle;

Builder steps

Method	Description
`step(name, task, opts?)`	A task node. `opts`: `after`, `args`, `queue`, `maxRetries`, `timeoutMs`, `priority`, `condition`, `compensate`, `cache`.
`fanOut(name, { after, task, itemsFrom })`	Expand into one child per item.
`fanIn(name, { after, task })`	Collect children's results into an array.
`gate(name, { after, timeoutMs?, onTimeout?, message? })`	Pause for approval.
`subWorkflow(name, { after, workflow })`	Run a child workflow as a node.
`chain(steps[], { after? })`	Canvas: wire `steps` sequentially.
`group(steps[], { after? })`	Canvas: run `steps` in parallel.
`chord(steps[], callback, { after? })`	Canvas: a parallel group joined by `callback`.
`submit()`	Pre-enqueue the DAG → `WorkflowHandle`.
`build()`	Build (don't submit) — for use as a sub-workflow.

Canvas steps are { name, task, ...stepOptions } (after is managed by the helper).

condition: "on_success" (default) · "on_failure" · "always".

The chained submit() takes no arguments. To set submit-time queueDefault/params, stop before the final .submit() and pass the builder to queue.workflows.submit(builder, { queueDefault?, params? }) instead:

const workflow = queue.workflows.define("etl").step("extract", "extract");
const run = queue.workflows.submit(workflow, { queueDefault: "io", params: { runId } });

`WorkflowHandle`

Member	Description
`runId`	The run's id.
`status()`	The current `WorkflowRun` snapshot, or `undefined`.
`wait()`	Resolves with the `WorkflowRun` when terminal.
`nodes()`	Per-step status (fan-out parents carry `fanOutCount`).

Queries

queue.workflows.list({ state: "running" });
queue.workflows.run(runId); // a single run snapshot
queue.workflows.nodes(runId); // per-step status
queue.workflows.dag(runId); // the DAG graph as a JSON string (nodes + edges); JSON.parse it, or use analyze()
queue.workflows.children(runId); // spawned sub-workflow runs
queue.workflows.approveGate(runId, "review");
queue.workflows.rejectGate(runId, "review", reason);
queue.workflows.resolveGate(runId, "review", approved);
queue.workflows.clearCache(); // drop all cached step results

Every query reads storage on each call — nothing is cached, so polling run() in a loop is a query in a loop. All of them work from any process: a run is identified by its id, not by the handle that submitted it.

Query	Declared as	Notes
`list({ definitionName?, state?, limit?, offset? })` → `WorkflowRun[]`	`listWorkflowRuns`	Offset-paginated. Omitted filters mean "any".
`run(runId)` → `WorkflowRun \| undefined`	`getWorkflowRun`	`undefined` for an unknown run — it does not throw.
`nodes(runId)` → `WorkflowNode[]`	`getWorkflowNodes`	Per-step records, including a fan-out parent's expansion count.
`dag(runId)` → `string \| undefined`	`getWorkflowDag`	The stored definition, as a JSON string of `{ nodes, edges }`. `JSON.parse` it, or let `analyze()` do that.
`children(runId)` → `WorkflowRun[]`	`getWorkflowChildren`	Sub-workflow runs spawned by `runId`. Empty unless something in the deployment submits sub-workflows.
`submit(builder, opts?)` → `WorkflowHandle`	`submitWorkflow`	Pre-enqueues the whole DAG in one call, so a run exists before any step does. Throws at construction if the addon was built without the `workflows` cargo feature.

analyze() is dag() plus nodes(), so it costs two queries and returns undefined whenever either is missing or the DAG will not parse.

Analysis

analyze(runId) returns a WorkflowAnalysis over the run's DAG + node statuses (or undefined if the run is unknown), for graph introspection:

const a = queue.workflows.analyze(runId);
a?.node("load"); // one node's status record, or undefined
a?.roots(); // entry nodes
a?.leaves(); // exit nodes
a?.ancestors("load"); // transitive upstream deps
a?.descendants("extract"); // transitive downstream
a?.topologicalOrder(); // a valid run order (throws on a cycle)
a?.topologicalLevels(); // nodes grouped by dependency depth
a?.criticalPath(); // longest dependency chain (root → leaf)
a?.stats(); // { total, byStatus, completed, failed, running, pending }

See the Workflows guides for fan-out, conditions, gates, sub-workflows, and saga compensation. Requires the addon built with the workflows cargo feature.