@clemrig/flow

Write agent workflows in TypeScript and run them with Claude Code, Codex or Pi.

A workflow is an explicit graph of CODE stages (your code), AGENT stages (an agent run by Claude Code, Codex or Pi) and GATE stages that accept a result or send it back for correction, a bounded number of times. This package is the engine: the API to write workflows, their execution, the agent runtimes and the runs. @clemrig/factory builds on it a command that takes a repository's issues from spec to pull request, beside the repository's own workflows; @clemrig/flow-ui shows the runs in a local dashboard.

Requirements

macOS or Linux, Node.js 24 or later. For agent stages, the claude CLI (2.1.296 or later), the codex CLI (0.160.1 or later, in a Git repository) or the pi CLI (1.1.0 or later) in the PATH, connected to its provider. Each agent runs with the machine's own configuration of its CLI, plus its profile's settings (a Claude Code profile can leave out the user's settings files with settingSources and the machine's MCP servers with mcp: "profile"); a Claude Code call also waits for the agent's sub-agents, however long they stay idle, within the profile's timeout (CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS=0), where print mode would stop them after 10 minutes and drop their results. The agent's answer ends the call: Claude Code is stopped there, with any command the agent left running in the background, such as a dev server, and the stage's journal counts them. Each command of a Claude Code agent registers the process group it runs in, so that whatever it leaves running, detached with & or nohup included, ends with the call, whatever ends it: this needs a ps that lists processes with their group and start time (macOS's, or procps on Linux), checked before the run. The engine never reads or stores credentials.

Run a workflow

npm install @clemrig/flow

import { defineWorkflow, END, node, startRun, step, z } from "@clemrig/flow";

const hello = defineWorkflow({
  name: "hello",
  input: z.object({ name: z.string().min(1) }),
  initial: "greet",
  nodes: {
    greet: node(
      step(async ({ state }) => ({ greeting: `Hello ${String(state.input.name)}` })),
      { ok: END },
    ),
  },
});

const run = await startRun(
  {
    workflow: hello,
    input: { name: "flow" },
    prompts: async () => ({}), // prompt texts by name, for agent stages
    dir: process.cwd(), // where stages and agents work, until a stage moves the run; given to stages as `dir`
    configDir: process.cwd(), // where relative skill, extension and MCP paths of profiles resolve
  },
  { onEvent: (event) => console.log(event.type, event.stage) },
);
const state = await run.done;
console.log(state.status, state.data);

A workflow may declare steps: ordered groups of its stages that show how far a run went, for instance steps: [{ name: "greet", stages: ["greet"] }]. defineWorkflow refuses steps unless they name each step once and hold each stage exactly once; the engine itself ignores them, and the factory's Kanban shows them.

Before it creates the run, startRun checks, in this order, dir (an existing directory), the budget (maxCostUsd, in USD, replaces the workflow's; refused when an agent stage uses Codex, which announces no cost), the input against the workflow's input (an InvalidInputError lists every problem), the prompts (a non-empty text for each agent stage's prompt) and each runtime the workflow uses, its version and connection checked in dir; the first failing check rejects it with its reasons. It then records the launch (a launch event: flow's version, the caller labels given, such as a caller's version and where its prompts come from, each runtime as checked, the budget in force and whether it was given, the run's own or the workflow's, the machine, dir, and the names of the variables set that the runtimes read, never their values) and returns the run, its end (done) and stop(reason), which fails the run before its next stage. Runs are kept in runsDir when it is given. The engine never changes the process's current directory: stage code reaches the working directory through dir, and the run's own folder through runDir, where it can keep files that belong to the run. A stage moves the run by returning dir, the absolute path of an existing directory: its next stages and agents work there, and so does the run when it is resumed (state.dir, absent while the run works in the directory it was started or resumed in; the stage's event keeps it too). A stage that gives a relative path or no directory fails, its data dropped. An agent's conversation continues in the directory where it started, since a runtime such as Claude Code keeps it per directory: move the run between conversations, not during one. While child processes run (agents, commands), a SIGINT or SIGTERM to the program interrupts them, each agent with the signal on which it ends its own commands, so that none outlives it, and stops each run before its next stage, even if a stage catches the interruption. runCommand(argv, { cwd, timeoutMs, signal, env, onLine, stdin, output }) is runProcess's full form: signal stops the program, its process group included, without failing once it aborts, such as a server its caller no longer needs, which then needs no timeoutMs; output appends what the program writes, errors included, to a file instead of capturing it. A program about to quit at once can call killChildProcesses(), which kills the child processes of every run, but not the commands Codex runs in their own process groups.

resumeRun({ workflow, runId, maxCostUsd, data, prompts, dir, configDir, runsDir, caller }, hooks) resumes a failed run at the stage where it stopped, with the same ID, input, data, corrections, measures and agent conversations: that stage runs again, journals why the run had stopped (Resumed after: …) and passes it on to its agent, which is asked to continue rather than start over. It refuses, changing nothing, a run that did not fail (a run left running after its process was killed included), a run of another workflow, a stage the workflow no longer has, a run that already spent its budget, a run a stage moved to a directory that no longer exists, and whatever the workflow's optional resumeProblem({ state, dir }) returns, dir being where the run works, for instance a working directory that is no longer as the stage left it; then it checks prompts and runtimes like startRun, takes the run over (two resumptions never both run it: the state is reloaded and marked running under a run.lock file created atomically in the run's folder), records the launch with the values given, and returns the same handle. The budget is the maxCostUsd given, else the one the run was started or last resumed with (state.maxCostUsd), else the workflow's. The stage starts over: what it did before it failed must be safe to do again, or resumeProblem must refuse. data gives the run values decided by the operator, merged into its data before the stage runs again, seen by resumeProblem and named in the cause it journals. A run stopped by a retry edge past its maximum is allowed one more correction on that edge by each resumption (state.allowances; exceededEdge(workflow, state) names that edge, <stage>.<outcome>: state.exceeded, or for a run stopped before the engine kept it, the edge of its stage whose message its error is, if a single one is): the stage runs again and, if it asks for the correction again, takes it; the correction it could not take is kept in its failed stage event's feedback. Each finished stage event also carries its agent's answer (output) and the correction it sent back (feedback). Calls that announced no cost before the resumption (a call stopped by its timeout) no longer fail the budget; later ones do. The time a run spent stopped is not part of its duration (state.pausedMs, left out by elapsedMs).

The stages, gates, corrections, agent profiles, sessions and prompts are described in the "Write a workflow" section of @clemrig/factory's README; import them from @clemrig/flow here. What it says of .factory/, prompt files and factory run options belongs to the factory: with the engine alone, the program gives the input, the prompt texts and the directories to startRun.

Runs

Each run keeps state.json, events.jsonl and, for each agent call, a record in calls/<n>.json (the workflow's prompt, the request as the runtime read it, the command with the prompt, the sub-agents' definitions and the MCP servers' configuration replaced by what names them and the names of the variables the request added (env), never their values, what the request named instead of sending (omitted: a Claude Code conversation that already holds a long value unchanged gets its name under # Already received, not the value again), the configuration the session confirmed when its runtime announces one, the end of a successful call's error output, and the agent's final text) in its own folder of runsDir, by default .flow/runs/ of dir. When the engine creates .flow/, it writes a .gitignore that ignores everything in it: runs hold inputs, agent outputs and the journal, and stay private. listRuns(runsDir), loadRun(runsDir, id), loadRunEvents(runsDir, id), loadAgentCall(runsDir, id, n), summarizeRun(state), summarizeCorrections(state, events) (each retry edge taken, with the maximum its events recorded) and presentResult(workflow, state) (any object with a presentResult in place of the workflow; without one, the validated data as JSON) read them; flow-ui shows them.

License

MIT