Focketplow

A small, type-safe node-graph / workflow orchestration engine for TypeScript.

Inspired by PocketFlow, Focketplow gives you one durable primitive — a prep → exec → post Node wired into a directed graph — and gets out of your way. It is async-first, ships a hooks layer for tracing and error recovery, and has built-in cycle protection and bounded parallelism.

What this is — and isn't. Focketplow is the graph engine. It deliberately does not ship an LLM client, tool-calling, streaming, or memory. You bring those and compose them inside nodes. If you want a "batteries-included agent framework," this isn't it — but if you want a thin, honest, fully-typed spine to build one on (or to orchestrate any async pipeline), it is.

Install

npm install focketplow

Examples

Runnable, API-key-free examples live in examples/ — each runs with npx tsx examples/<name>.ts and marks the ← SWAP lines that go live:

| Example | Demonstrates | |---------|--------------| | simple-agents | conditional routing + the hooks layer | | customer-support-agent | resilient pipeline: retry, timeout→fallback, onError→escalation, HITL gate | | batch-enrichment | BatchNode concurrency + failFast + per-item signal | | timeout-graceful-degradation | timeoutMs → FlowTimeoutError → cached fallback routing | | retry-and-circuit-breaker | maxRetries/backoff + an onError-driven circuit breaker | | human-in-the-loop | onNodeStart gate as an approval seam | | deep-research | ForkJoin + Subflow — parallel isolated sub-agents in a tool-calling loop | | search-agent, project-manager | routing, parallel synthesis, dependency scheduling |

The 30-second tour

import { Flow, Node } from "focketplow";

// A node: prepare → execute → decide next step.
class Greet extends Node<{ name: string }, { loud?: boolean }, string, string> {
  prep(shared) { return shared.name; }      // `shared` is the obj passed to run() below
  exec(name) { return `hello, ${name}`; }
  post(shared, _p, out) {
    return this.params.loud ? 'shout' : undefined; // undefined = default edge
  }
}

const greet = new Greet();
const done = new Node<{ name: string }>(); // terminal node — Node's prep/exec/post are no-ops by default

greet.next(done);            // default edge
greet.on('shout').to(done);  // conditional edge

const flow = new Flow<{ name: string }>(greet).setParams({ loud: false });
// The object you pass to run() IS `shared` — that same mutable object is handed to
// every node's prep()/post(), so nodes communicate by reading/writing it.
await flow.run({ name: 'world' }); // async-first: run() returns a Promise

Core concepts

Nodes & lifecycle

Every node has three overridable phases. The engine always awaits them, so bodies may be sync or async — no separate async base classes.

| Phase | Receives | Returns | Purpose | |-------|----------|---------|---------| | prep(shared) | shared context | data for exec | Read/setup | | exec(prepRes) | result of prep | result for post | Do the work (retried) | | post(shared, prepRes, execRes) | all of the above | Action | Write back & choose the next edge |

Actions & routing

post() returns an action (string | undefined):

• undefined → follow the default edge (.next(node)).
• a string → follow the edge registered with .on(action).to(node).

Clean termination (no matching edge) just ends the flow. A named action with no matching edge emits a soft warning — typo detection without false alarms.

Shared context & params

• shared is the object you pass to flow.run(shared). That exact same mutable object is handed to every node's prep() and post() and threaded through the whole flow, so nodes communicate by reading and writing it. Nothing magic: you create it (the run({...}) argument), the engine just forwards it to each node. (It is not immutable — design it accordingly.)
• params is node config. A Flow propagates its own params to every node as defaults; a node's own params (set via node.setParams()) are merged on top and are never clobbered. So node-specific config is safe inside a flow, and BatchFlow bundles layer in below a node's own params too.

The seven classes

| Class | Role | |-------|------| | BaseNode | Lifecycle + graph wiring. No retry. | | Node | BaseNode + retry (maxRetries, waitMs, execFallback). | | BatchNode | Processes prep()'s array item-by-item with a concurrency option + failFast. | | Flow | Orchestrates a graph: hooks, cycle guard, can be nested as a node. | | BatchFlow | Runs a sub-flow once per bundle (concurrency + failFast). | | Subflow | Runs a nested Flow over an isolated context, then folds results back. | | ForkJoin | Fans out to N distinct branch nodes in parallel (isolated shareds), then joins. |

That's the entire surface. (Pre-0.1 there were 11 classes — AsyncNode, AsyncBatchNode, AsyncParallelBatchNode, AsyncFlow, AsyncBatchFlow, AsyncParallelBatchFlow are all gone; async-first execution and the concurrency option replace them.)

Retries (fixed semantics)

maxRetries is the number of retries after the first attempt — so total attempts are 1 + maxRetries, and the default 0 means "run once, don't retry." On final failure, execFallback(prepRes, error) is called (default: rethrow).

class Flaky extends Node<C, P, Req, Res> {
  constructor() { super({ maxRetries: 3, waitMs: 1000 }); }
  exec(req: Req): Res { /* may throw; retried up to 3 times */ }
  execFallback(req: Req, err: unknown): Res { /* final-attempt fallback */ }
}

Hooks — tracing, metrics & error recovery

Register one or more hook sets on a Flow. They're your cross-cutting layer for logging, OpenTelemetry/Langfuse spans, timeouts, and turning failures into routes.

flow.use({
  onStart: ({ shared }) => log.info('flow start'),
  onNodeStart: ({ node, path }) => log.debug({ node: node.name, path }),
  onNodeEnd: ({ node, durationMs, skipped, error }) =>
    metrics.timing({ node: node.name, ms: durationMs, skipped, error: !!error }),
  onRetry: ({ node, attempt, error }) => log.warn({ node: node.name, attempt, error }),
  onError: ({ node, error }) => {
    // Return a string action to ROUTE to an error node (post() is skipped),
    // or return nothing / 'throw' to propagate.
    return 'error_path';
  },
  onEnd: ({ action }) => log.info({ action }),
});

Because onError can return an action, a failing exec no longer kills the whole flow — you can route to a recovery node, exactly like any other edge.

Human-in-the-loop gates

onNodeStart can return an action to gate a node: its prep/exec/post are skipped and the flow routes via that action's edge. This is the HITL / approval seam.

flow.use({
  async onNodeStart({ node, shared }) {
    if (node.name !== 'DeleteNode') return;      // run normally
    if (!(await askUser(`Delete ${shared.path}?`))) return 'kept'; // SKIP + route
    // returning undefined lets the node run
  },
});

A gated node still fires onNodeEnd (with skipped: true) for tracing. See examples/human-in-the-loop.ts for a full create→confirm→delete workflow.

Parallelism: concurrency + failFast

BatchNode and BatchFlow take { concurrency, failFast }:

// Up to 8 items in flight; one failure doesn't abort the rest (yields `undefined`).
class Enrich extends BatchNode<C, P, Item, Result>({
  // ...
}) {}

new Enrich({ concurrency: 8, failFast: false });

• concurrency: 1 (default) = sequential, order preserved.
• concurrency > 1 = bounded parallel, still order-preserved.
• failFast: false (default) = Promise.allSettled-style: failed items become undefined and fire onError; the rest finish. failFast: true aborts on the first failure.

BatchFlow isolation note. In BatchFlow, concurrency > 1 is now isolated: each bundle runs over its own cloned shared and a freshly cloned node graph, so concurrent bundles neither stomp each other's shared nor race on node params. Fold results back by overriding merge(parentShared, bundleShareds). (concurrency === 1 keeps the classic behaviour — bundles accumulate into the single real shared, bundle params flow into this.params.)

Parallel delegation: ForkJoin

For the canonical "delegate to N parallel sub-agents, then synthesize" pattern (the shape of a deep-research agent), ForkJoin fans out to distinct branch nodes concurrently, each over its own isolated copy of the shared, then calls join() to merge the results:

import { Flow, Node, ForkJoin } from "focketplow";

class ResearchAgent extends Node<Ctx, { topic: string }> {
  /* …does the research for one topic, writes findings to shared… */
}

class FanOut extends ForkJoin<Ctx> {
  join(shared: Ctx, branchShareds: (Ctx | undefined)[]) {
    shared.findings = branchShareds
      .filter((b): b is Ctx => !!b)
      .flatMap((b) => b.findings);
  }
}

const fan = new FanOut(
  [
    new ResearchAgent().setParams({ topic: "RAG" }),
    new ResearchAgent().setParams({ topic: "fine-tuning" }),
    new ResearchAgent().setParams({ topic: "hybrid" }),
  ],
  { concurrency: 3 }
);
await new Flow<Ctx>(fan).run({ /* … */ });

Each branch runs its own full lifecycle (hooks fire per branch, HITL onNodeStart gates work per branch, retry applies) over a private shared clone, so branches can't observe or clobber one another. Each branch must be a distinct instance — do not pass the same node twice, since concurrent lifecycles on one instance race on its params.

Isolated sub-contexts: Subflow

Subflow runs a nested Flow over a context derived from the parent shared and folds only what you choose back via reduce() — the "subagent with isolated context" primitive:

import { Flow, Node, Subflow } from "focketplow";

class MySub extends Subflow<ParentCtx, {}, SubCtx> {
  deriveShared(parent: ParentCtx): SubCtx {
    return { query: parent.question, findings: [] }; // fresh, isolated
  }
  reduce(sub: SubCtx, parent: ParentCtx) {
    parent.answer = sub.findings.join("\n"); // copy back only the result
  }
}

const sub = new MySub(new Flow<SubCtx>(new ResearchNode()));
await new Flow<ParentCtx>(sub).run({ question: "…" });

The sub-flow inherits the parent's hooks (tracing continuity) and is retried per the Subflow's maxRetries/waitMs. Default deriveShared() clones the parent shared.

Cycle guard

Every Flow has a maxSteps budget (default 1000). If a run exceeds it — e.g. an LLM keeps returning the same routing action — it throws FlowCycleError instead of hanging.

new Flow<C>(start, { maxSteps: 50 });

Timeouts

maxSteps bounds step count, not wall-clock. For a node that can hang — an LLM call that never returns, a deadlocked fetch — give it a timeoutMs. The whole prep → exec → post lifecycle is raced against it; on expiry FlowTimeoutError goes through onError like any failure, so you can route to a recovery node:

class Fetch extends Node<C>({ /* ... */ }) {
  constructor() { super({ timeoutMs: 5000 }); }   // 5s budget for the whole node
}
fetch.on("timeout").to(fallback);
flow.use({ onError: ({ error }) => (error instanceof FlowTimeoutError ? "timeout" : undefined) });

Cancellation

Pass an AbortSignal to make a run cancellable. Aborting rejects with FlowAbortError, propagated from wherever execution was — useful for request-scoped teardown or a "stop" button:

const ctrl = new AbortController();
flow.run(ctx, { signal: ctrl.signal }).catch((e) => { /* FlowAbortError on cancel */ });
// elsewhere: ctrl.abort();

failFast: true now actually cancels in-flight siblings (previously a failure left them running). It works by aborting an internal signal shared by the parallel items / branches / bundles. For cooperative cancellation — aborting a fetch the instant a sibling fails — read the node's signal (or the signal passed to execItem) and hand it to your I/O:

class Enrich extends BatchNode<C, P, Item, Result>({ concurrency: 8, failFast: true }) {
  async execItem(item: Item, signal?: AbortSignal) {
    return await fetch(url(item), { signal });   // aborts promptly on sibling failure
  }
}

Three error types, all FlowError subclasses: FlowCycleError (maxSteps), FlowTimeoutError (timeoutMs — routable via onError), and FlowAbortError (cancellation — terminal, never routed; never retried).

Nesting

A Flow is itself a Node, so you can drop one inside another. The nested flow's hooks merge with the parent's, and each has its own maxSteps budget.

Migration from 0.0.x (breaking)

This is the 0.1 rewrite. The headline changes:

1. One async-first engine. AsyncNode/AsyncFlow/etc. are removed. Use Node/Flow, override prep/exec/post (optionally async), and call await flow.run(shared). The missing runAsync() is no longer an issue — run() is the async entry point.
2. maxRetries now means retries. maxRetries: 1 previously ran exec once; it now runs it twice (1 retry). Default is 0.
3. Batching collapsed. BatchNode/BatchFlow take { concurrency, failFast } instead of three separate parallel/async classes. Override execItem(item) (not exec) on BatchNode.
4. Hooks added (flow.use({...})) for tracing and error-as-routing.
5. Cycle guard (maxSteps) and failFast added; onNodeStart gates (return an action to skip+route) added for human-in-the-loop.
6. Flow params no longer clobber node params — they merge, with the node's own params winning.

What's intentionally not here

Streaming, tool/structured-output helpers, message memory, checkpointing/resume, and human-in-the-loop interrupts. These belong to the layer you build on top of Focketplow (or to a heavier framework). The engine's job is to route execution reliably and observably. What it does now provide for agent-style work are the graph primitives ForkJoin (parallel delegation) and Subflow (isolated sub-contexts) — the structural shape of a deep-research / multi-agent system, without the LLM/tooling.

License

ISC