flow-engine

Getting started

Install#

npm i @dmytromykhailiuk/flow-engine

Quick start#

Register steps and predicates in code, describe the flow as data, run it. Steps are the only place where real code — API calls, side effects — lives; everything else is serialisable config.

import { createFlowRunner } from "@dmytromykhailiuk/flow-engine";

const runner = createFlowRunner((b) => b
  .registerStep("loadUser", async (input: { id: string }, _cfg: void, ctx) => ({
    user: await api.getUser(input.id, { signal: ctx.signal }),
  }))
  .registerStep("greet", (input: { name: string }, cfg: { greeting: string }) => ({
    message: `${cfg.greeting}, ${input.name}!`,
  }))
  .registerCondition("isPremiumUser", (ctx) =>
    (ctx.steps.loadUser as any)?.user?.plan === "premium",
  )
  .registerFlow("welcome", {
    nodes: [
      { step: "loadUser", input: { id: "input.userId" } },
      {
        if: [
          {
            when: "$isPremiumUser",
            flow: [{
              step: "greet",
              input: { name: "steps.loadUser.user.name" },
              config: { greeting: "Welcome back" },
            }],
          },
        ],
        else: [{
          step: "greet",
          input: { name: "steps.loadUser.user.name" },
          config: { greeting: "Hello" },
        }],
      },
    ],
    output: "steps.greet.message",
  }),
);

const handle = runner.run("welcome", { userId: "42" });

const result = await handle.getResult();   // never rejects
if (result.status === "completed") {
  console.log(result.output);         // "Welcome back, Ada!"
}

// external control, any time:
handle.pause();
handle.resume();
handle.cancel("user navigated away");

Core concepts#

API

Builder API#

createFlowRunner(configure, options?) hands your callback a builder; every register* call returns the builder with the new entry accumulated in its generic parameters. That is where the typing comes from: flow configs written in TypeScript autocomplete step ids and type-check each step's config against the registry.

const runner = createFlowRunner(
  (b) => b
    .registerStep(/* … */)
    .registerCondition(/* … */)
    .registerFlow(/* … */)
    .registerGuard(/* … */)
    .registerHook(/* … */)
    .registerSource(/* … */)
    .registerBackgroundFlows(/* … */),
  {
    onEvent: (e) => console.log(e.type),  // every run of this runner
    defaultStepTimeout: 30_000,           // ms, overridable per node
    queueAdapter: myBlocker,              // see Queues
    heartbeatMs: 10_000,                  // see Backend example
  },
);

After createFlowRunner() returns, the code registries (steps, predicates, sources) are frozen. The declarative layer — named flows, hooks, guards, background groups — can still be swapped atomically later with applyConfig(). Everything registered through the builder is validated eagerly: a broken flow config throws FlowValidationError at startup, not at the first run.

registerStep(id, handler)

The handler signature is (input, config, ctx) => output | Promise<output>. The input type is what flow configs must map into the step; config is the static per-node configuration; the output lands in context.steps[id].

b.registerStep(
  "chargeCard",
  async (input: { amount: number }, cfg: { currency: string }, ctx) => {
    // ctx.signal aborts on cancel() and on the step's timeout
    const receipt = await payments.charge(input.amount, cfg.currency, {
      signal: ctx.signal,
      idempotencyKey: ctx.executionKey,   // stable across retries AND restore
    });
    return { receiptId: receipt.id };
  },
)

The third argument, StepContext:

registerCondition(name, predicate)

A named JS predicate (ctx: FlowContextSnapshot) => boolean. Configs reference it as "$name" anywhere a condition is expected.

b.registerCondition("isPremiumUser", (ctx) => {
  const user = (ctx.steps.loadUser as { user?: { plan?: string } })?.user;
  return user?.plan === "premium";
})

registerFlow(name, config)

A named flow — runnable via run("name", …) and referenceable from subflow nodes, hooks, guards and background entries. Registered flows are validated at createFlowRunner() (forward references between named flows work at runtime; the type system only sees names registered earlier in the chain).

registerGuard, registerHook, registerSource, registerBackgroundFlows

b
  // named guard — referenced from FlowConfig.guards (see Guards)
  .registerGuard("hasSession", {
    validate: "input.sessionId != null",
    onFailure: "redirectToLogin",
  })
  // named entry point — see Hooks
  .registerHook("onUserLogin", [
    { when: "input.user.plan === 'premium'", flow: "premiumOnboarding" },
    { flow: "defaultOnboarding" },                 // unconditional fallback
  ])
  // reactive data source — see Background flows
  .registerSource("cart", {
    getSnapshot: () => cartSignal.value,
    subscribe: (cb) => effect(cb),
  })
  .registerBackgroundFlows("cartWatchers", [
    { when: "sources.cart.items.length > 0 && !sources.cart.reserved", flow: "reserveStock" },
  ])

The language

Expression language#

Configs are JSON, and JSON cannot carry callbacks — so all dynamic reads are string expressions. They are never compiled to JavaScript: the engine tokenizes them, parses a small AST and interprets it. What the grammar can express is the entire security boundary.

Grammar

Scopes — which roots exist where

Using a root that does not exist at a site is a config error, caught statically by validation and loudly at runtime — a typo like stpes.a never silently becomes undefined.

Security

• No eval, no new Function — expressions from a backend cannot become remote code execution.
• No assignment, no new, no reaching globals — globalThis, fetch, window are unreachable whatever the expression says.
• __proto__, constructor and prototype are blocked on read.
• Methods resolve from the built-in prototypes and are identity-checked against the receiver — a function smuggled into your data is not callable.
• Amplifier methods (repeat, padStart, padEnd) are deliberately absent: with untrusted input they are a denial-of-service primitive.
• Parsed ASTs are cached by source string, with a bounded cache.

$predicates

Anywhere a condition is expected, a string starting with $ references a predicate registered with registerCondition. Expressions cannot start with $; unknown names are a validation error (unknown-condition).

{ "when": "$isPremiumUser" }

The config

Flow nodes#

A flow is { nodes, output?, tag?, guards?, queue? }. Nodes execute in order; between any two nodes the engine checks pause/cancel and commits a checkpoint. output is an expression producing the flow's result (default: the whole context snapshot); tag labels the flow for finish unwinding; queue serialises runs (see Queues).

step — run a registered step

{
  "step": "loadUser",
  "id": "owner",
  "input": { "id": "input.userId", "source": "'mobile-app'", "limit": 10 },
  "config": { "withProfile": true },
  "timeout": 5000,
  "retry": { "attempts": 2, "delay": 300, "backoff": "exponential" },
  "onError": "fail"
}

if — first matching branch

{
  "if": [
    { "when": "$isPremiumUser", "flow": [{ "step": "showPremiumOffer" }] },
    { "when": "steps.loadUser.user.age >= 18", "flow": [{ "step": "showAdultOffer" }] }
  ],
  "else": [{ "step": "showDefaultOffer" }]
}

Branches are checked top-down; the first truthy when wins. No match → else; no else → no-op. The branch array is the switch — there is no separate switch node. A branch event reports the matched index.

loop — times, while, forEach

{ "loop": { "times": 3 }, "flow": [{ "step": "ping" }] }

{
  "loop": { "while": "steps.poll.status !== 'done'", "max": 50 },
  "flow": [{ "step": "poll" }]
}

{
  "loop": { "forEach": "steps.loadOrders.orders", "as": "order" },
  "flow": [
    { "step": "processOrder", "input": { "id": "loop.order.id", "position": "loop.index" } }
  ]
}

• times — a fixed number of iterations.
• while — the condition is checked before each iteration. max is the runaway brake: exceeding it fails the run with kind: "loop". The validator warns (while-without-max) when it is missing.
• forEach — an expression producing an array. null/undefined → zero iterations (consistent with safe member access); any other non-array → kind: "loop" failure. The array is captured once before the first iteration.

Inside the body the loop root exposes loop.index, loop.count (times / forEach length; absent for while) and loop.item (forEach). as gives the item an alias — loop.order above. In nested loops index/count/item always belong to the nearest loop, while the as-aliases of outer loops stay visible (inner wins on a name clash):

{
  "loop": { "forEach": "steps.loadUsers.users", "as": "user" },
  "flow": [{
    "loop": { "forEach": "loop.user.orders", "as": "order" },
    "flow": [{
      "step": "sync",
      "input": { "userId": "loop.user.id", "orderId": "loop.order.id" }
    }]
  }]
}

runner.run("searchAll", { query }, { vars: { items: [], page: 0 } });

{
  "loop": { "while": "vars.page < steps.search.totalPages", "max": 100 },
  "flow": [
    { "step": "search", "input": { "query": "input.query", "page": "vars.page" } },
    { "assign": {
        "items": "vars.items.concat(steps.search.items)",
        "page": "vars.page + 1"
    } }
  ]
}

parallel — concurrent branches

{
  "parallel": [
    [{ "step": "loadUser", "input": { "id": "input.userId" } }],
    [{ "step": "loadSettings" }],
    [{ "step": "loadNotifications" }]
  ],
  "settle": "all"
}

• Each branch runs on a child context: a shared readonly view of input/steps/vars as of node start; its writes stay local until the join, then merge into the parent.
• Branches may not write the same step ids or vars — the validator rejects the config (duplicate-parallel-write) instead of leaving you a data race.
• settle: "failFast" (default): the first branch error aborts the remaining branches' signals; nothing merges; the run fails with kind: "parallel" and branchErrors.
• settle: "all": every branch runs to completion. Writes of successful branches merge; if anything failed the node then throws with the complete branchErrors. Wrap it in try for "do what you can, report the rest": the catch sees error.branchErrors while the successful outputs are already in steps.
• cancel() beats both modes — all branches abort immediately.
• finish without a tag ends only its branch; with an outer flow's tag the branch completes, the others are awaited, and the unwind continues after the join.

subflow — a flow as a function

{ "subflow": "checkout", "input": { "cartId": "steps.loadCart.id" }, "as": "checkout" }

A subflow is a function, not a macro: it gets its own context — input is the resolved mapping (default: the parent's input), steps and vars start empty — and sees nothing else of the parent. Its output (the config's output expression, or its whole context) is stored under steps[as]. A failure inside surfaces as kind: "subflow" with the inner error as cause. subflow accepts a registered flow name or an inline config.

assign — transform data safely

{
  "assign": {
    "total": "steps.calcA.value + steps.calcB.value",
    "user.fullName": "steps.loadUser.user.firstName + ' ' + steps.loadUser.user.lastName"
  }
}

Writes only under vars — keys are dot-paths relative to it. input and steps are read-only for configs: the provenance of step outputs cannot be forged (reserved-root validation error). Assignments run in key order — later ones see earlier results. Writes are copy-on-write, which is what keeps already-committed checkpoints intact.

finish — early exit

{ "finish": true }

{ "finish": { "tag": "checkout", "output": "vars.total" } }

• Without a tag — completes the current flow (the nearest FlowConfig) with status completed.
• With a tag — unwinds the subflow stack to the nearest flow whose tag matches and completes that flow; its result carries finishedBy: tag. Labeled-break semantics. A tag that matches nothing completes the root run.
• output overrides the finished flow's output; the expression is evaluated in the context of the flow where the finish node stood.

break / continue

{
  "loop": { "forEach": "steps.loadOrders.orders", "as": "order" },
  "flow": [
    { "step": "processOrder", "input": { "id": "loop.order.id" } },
    { "if": [
        { "when": "steps.processOrder.status === 'rate-limited'", "flow": [{ "break": true }] },
        { "when": "steps.processOrder.status === 'skipped'",      "flow": [{ "continue": true }] }
    ] }
  ]
}

break exits the nearest loop; continue skips to the next item / iteration / condition re-check. Target an outer loop by label:

{
  "loop": { "forEach": "steps.loadUsers.users", "as": "user" }, "label": "users",
  "flow": [{
    "loop": { "forEach": "loop.user.orders" },
    "flow": [
      { "if": [{ "when": "loop.item.blocked", "flow": [{ "break": { "label": "users" } }] }] }
    ]
  }]
}

Both are loop-scoped: they cannot cross a subflow or a parallel branch boundary (that is finish's job). Misplacement is a static error: orphan-break, unknown-loop-label.

try / catch / finally

{
  "try": [
    { "step": "reserveStock", "input": { "items": "vars.items" } },
    { "step": "chargeCard",   "input": { "amount": "steps.reserveStock.total" } }
  ],
  "catch": [
    { "step": "reportError", "input": { "message": "error.message", "failedStep": "error.stepId" } },
    { "step": "releaseStock" }
  ],
  "finally": [
    { "step": "closePaymentSession" }
  ]
}

• What is caught: any error that would otherwise fail the run — a step error after retries, a timeout, a broken expression, a subflow/parallel/loop failure. Order for a step: retry → onError (a skipped step throws nothing) → nearest try → the run fails.
• What is not: cancel and pause (not errors), guard rejection (happens before nodes), and control flow — finish/break/continue pass through catch untouched, but do run finally.
• catch scope: the error root — the intercepted FlowError without cause: error.kind, error.message, error.stepId, error.path, error.branchErrors.
• A catch that completes cleanly swallows the error; execution continues after the try node. Re-throwing = throwing from a step inside catch. An error in catch flies upward.
• finally always runs — after success, after catch, and during any unwind through the node. An error inside finally replaces the pending outcome, exactly like the language.
• Both sections are optional but at least one is required (bad-node otherwise). try + finally is the "cleanup without swallowing" pattern.
• The cancel exception: on cancel() the finally does not run — cancellation is immediate and terminal; cleanup on cancel belongs to step handlers via their AbortSignal.

At runtime

Running flows#

// by name, inline config, or a name that arrived later via applyConfig
const handle = runner.run("welcome", { userId: "42" }, {
  onEvent: (e) => log(e),        // this run only
  signal: abortController.signal, // external abort, linked to the run
  vars: { items: [] },            // seed vars before the first node
  queue: "cart",                  // overrides FlowConfig.queue
});

run() validates synchronously (throwing FlowValidationError on a broken config) and returns the handle immediately; execution proceeds in the background.

The handle

FlowResult

type FlowResult<Output> =
  | { status: "completed"; output: Output; context: FlowContextSnapshot; finishedBy?: string }
  | { status: "rejected";  guard: string;  context: FlowContextSnapshot }
  | { status: "cancelled"; reason?: unknown; context: FlowContextSnapshot }
  | { status: "failed";    error: FlowError; context: FlowContextSnapshot }
  | { status: "suspended"; reason?: string; snapshot: FlowRunSnapshot; context: FlowContextSnapshot };

• completed — output is the flow's output expression (or the whole context). finishedBy is set when a tagged finish ended the flow early.
• rejected — a guard said no. Not an error, not a completion.
• suspended — the run left this process via ctx.suspend(); persist snapshot and restore() it later.

FlowError

interface FlowError {
  kind: "step" | "timeout" | "expression" | "subflow" | "parallel" | "loop";
  message: string;
  path: FlowPath;          // where in the config it happened
  stepId?: string;
  cause?: unknown;         // the original thrown value (not serialized)
  branchErrors?: FlowError[];  // kind "parallel": every failed branch
}

Queues#

By default different runs execute in parallel — every run() is independent. Sequential execution is opt-in through named FIFO queues, at two levels (the run option overrides the config):

// declaratively, in the flow config (serialisable — a backend can decide):
b.registerFlow("syncCart", { queue: "cart", nodes: [/* … */] });

// or per run:
runner.run("syncCart", input, { queue: "cart" });

• Runs sharing a queue id execute strictly one at a time, in run() call order. Different queues — and runs without one — never block each other.
• While waiting, a run has status: "queued" and a queued event fires. Guards are checked after the queue is acquired (conditions may have changed while waiting).
• cancel() while queued leaves the queue instantly without executing; pause() keeps the position and parks the run before its first node once its turn comes.
• subflow nodes and parallel branches never enter queues — only top-level runs are serialised.
• Do not await runner.run() into the same queue from inside a step — that is a self-deadlock. Compose flows with subflow instead.

The backend is pluggable. The engine needs one method — FlowQueueAdapter:

interface FlowQueueAdapter {
  run<T>(id: string, fn: () => T | Promise<T>): Promise<T>;
}

// default: a private in-process FIFO (createExecutionBlocker()).
// share queues with non-flow code:
import { createExecutionBlocker } from "@dmytromykhailiuk/execution-blocker";
const blocker = createExecutionBlocker();
const runner = createFlowRunner(setup, { queueAdapter: blocker });
await blocker.run("cart", () => updateCartOutsideFlows());  // same queue as queue:"cart" runs

// or pass a Redis-backed adapter to make queues cluster-wide — see Backend example

Guards#

Declarative preconditions, checked over { input, vars } before the first node of a run. The old engine called them flow validators.

b.registerGuard("hasSession", {
  validate: "input.sessionId != null",   // the guard passes when true
  skipWhen: "input.isGuestCheckout",     // when true, the guard is skipped entirely
  onSuccess: "trackAccess",              // fire-and-forget side runs
  onFailure: "redirectToLogin",
});

b.registerFlow("checkout", {
  guards: ["hasSession", { validate: "vars.featureEnabled === true" }],  // names or inline
  nodes: [/* … */],
});

• Guards run in order. The first failing one launches its onFailure (as a separate run) and the main run resolves { status: "rejected", guard: name }.
• Only when all pass do their onSuccess flows launch, and the flow proceeds.
• Guards are re-checked on a fresh run only — restore() skips them (the run had already started).

Hooks#

Named entry points: an array of { when?, flow, input? } entries, checked top-down against { input }. Exactly one — the first match — runs; an entry without when is the unconditional fallback.

b.registerHook("onUserLogin", [
  { when: "input.user.plan === 'premium'", flow: "premiumOnboarding" },
  { when: "$isReturningUser",              flow: "welcomeBack" },
  { flow: "defaultOnboarding" },
]);

const handle = runner.runHook("onUserLogin", { user });
// undefined when nothing matched (or the hook is not registered) — nothing ran
if (handle) await handle.getResult();

Background flows#

Reactive triggers: "when this condition over external data becomes true — run that flow". The engine knows nothing about reactive libraries; you hand it a minimal source and it subscribes:

interface ExternalSource<T> {
  getSnapshot(): T;
  subscribe(onChange: () => void): () => void;
}

// signals:
b.registerSource("cart", { getSnapshot: () => cartSignal.value, subscribe: (cb) => effect(cb) });
// redux:
b.registerSource("session", { getSnapshot: () => store.getState().session, subscribe: (cb) => store.subscribe(cb) });

b.registerBackgroundFlows("cartWatchers", [
  {
    when: "sources.cart.items.length > 0 && !sources.cart.reserved",
    flow: "reserveStock",
    input: { "items": "sources.cart.items" },
    mode: "edge",          // default
    concurrent: "skip",    // default
  },
]);

runner.startBackgroundFlows("cartWatchers");
runner.stopBackgroundFlows("cartWatchers");
runner.stopAllBackgroundFlows();

startBackgroundFlows(groupId) is idempotent (restarting re-subscribes); stop unsubscribes and cancels the group's active triggered runs. After applyConfig(), running groups whose entries changed are restarted on the new version.

Durability

Persistence#

A run's state is plain JSON. After every completed node the engine commits a checkpoint snapshot; your app stores it wherever it likes — localStorage, Redis, a database — and after a refresh, a crash or on another instance, restore(snapshot) continues the run from that exact position.

The engine deliberately ships no storage of its own — it hands you snapshots and takes them back, and where they live in between is a three-method decision your app makes once. The same loop backs every environment: localStorage in a browser (see Frontend example), Redis or a database on a server (see Backend example), a file, an S3 bucket, or an in-memory map in tests.

Checkpoints

// Any storage works — three operations is the whole contract:
interface SnapshotStore {
  save(runId: string, snapshot: FlowRunSnapshot): Promise<void>;
  delete(runId: string): Promise<void>;
  loadAll(): Promise<FlowRunSnapshot[]>;
}

const runner = createFlowRunner(setup, {
  onEvent: (e) => {
    if (e.type === "checkpoint") void store.save(e.runId, e.snapshot);
    if (e.type === "flowEnd") void store.delete(e.runId);   // terminal — snapshot no longer needed
    if (e.type === "suspended") void store.save(e.runId, e.snapshot); // NOT terminal — keep it
  },
});

// on process start: continue everything unfinished
for (const snapshot of await store.loadAll()) {
  runner.restore(snapshot);
}

• A checkpoint commits after every completed node — a step's output written, an assign applied, a branch chosen, a loop iteration finished, a parallel branch done. Between two checkpoints there is exactly one node of progress to lose.
• The checkpoint event carries the full snapshot, so saving is a plain overwrite by runId — no diffing, no ordering concerns; the last write wins and is always a consistent state.
• handle.getSnapshot() returns the same state on demand (as of the last committed checkpoint) — for save-on-shutdown hooks: beforeunload in a browser, SIGTERM in a worker.
• flowEnd fires only on terminal endings (completed / failed / cancelled / rejected) — that is your signal to delete the stored snapshot. A suspended run emits its own event instead, because its snapshot is still needed.
• Checkpoints are synchronous in-memory work (a shallow context clone plus a cursor clone); the I/O of persisting them is yours and can be throttled — skipping intermediate saves only widens the at-least-once replay window, it never corrupts anything.
• Every run is persistable, however it was started — run(), runHook(), a guard side-flow or a background trigger.

The snapshot

You never build or edit one — treat it as a JSON blob keyed by runId. But it is readable, and understanding it helps. Here a process died — a page refresh, a worker crash, a deploy — on iteration 2 of a paging loop, right after fetchPage committed:

{
  "version": 1,
  "runId": "run_mf83k_1_a9x2",
  "flowName": "syncOrders",
  "createdAt": 1786309484512,
  "status": "running",
  "queue": "orders",

  "context": {
    "input": { "userId": "42" },
    "steps": {
      "loadOrders": { "totalPages": 5, "total": 112 },
      "fetchPage": { "items": [{ "id": "o-51" }, { "id": "o-52" }] }
    },
    "vars": { "items": [{ "id": "o-1" }, { "id": "o-2" }], "page": 2 }
  },

  "cursor": [
    { "next": 1 },
    { "next": 1, "enter": { "kind": "loop", "index": 2 } }
  ],

  "config": { "queue": "orders", "nodes": [ /* the full syncOrders config */ ] },
  "flows": {}
}

• cursor — a stack of frames, root first: "at root node 1 (the loop); inside its body, iteration 2, next body node 1". Frames also record chosen if-branches, try sections, subflow contexts and completed parallel branches. The shape is internal but stable within a major version.
• config + flows — the flow config and every named subflow it references, embedded. The snapshot is self-contained on the declarative side: restore does not care what applyConfig has done since; only the code registries (steps, predicates, sources) must still provide the referenced names.
• Serialisability is your side of the contract: step outputs, input and vars must be JSON-serialisable if you use persistence. The engine does not enforce it — without persistence any values are fine.

restore()

const handle = runner.restore(snapshot, options?);  // FlowRunOptions

• Validates the snapshot version and that every step / predicate / source its configs reference is registered — FlowValidationError otherwise.
• Continues from cursor on the stored context, with the same runId; a flowRestore event fires.
• A paused snapshot restores paused — the app decides when to handle.resume(). queued/running re-enter their queue (if any) or start immediately. Guards are not re-checked.
• At-least-once: a node that started but had not committed its checkpoint (the process died mid-step) runs again after restore. Steps with side effects should be idempotent — that is what ctx.executionKey is for: it is identical across retries and across restores, so an idempotency header, a SET NX key or a unique DB constraint dedupes the replay.
• parallel is branch-atomic: a checkpoint commits when each branch completes; on restore, done branches (including branches settled with an error under settle: "all") do not re-run — unfinished ones restart from their beginning.
• Two processes restoring the same snapshot will both run it. The engine is environment-agnostic (browser and server), so double-restore locking is deliberately the application's job — Web Locks in a browser, a distributed lock on a backend.

suspend() — durable waits

For waits measured in minutes or hours — a message-queue reply, a human approval, a webhook — holding a process on await wastes memory, a queue slot and (on a backend) a distributed lock. ctx.suspend(reason?) freezes the run and releases the process:

• The run ends in this process with status: "suspended"; getResult() resolves with a final snapshot; a suspended event fires (flowEnd does not — do not delete the stored snapshot).
• Suspend is neither an error nor an unwind: try/catch does not intercept it and finally does not run — the run is frozen at this node, and the finally will run after restore, on the normal path.
• The queue slot is released; restore() re-enters the queue.
• Restore re-executes the suspending step (at-least-once). The canonical pattern: check an external store by ctx.executionKey first — return the result if it arrived, suspend again if not.

b.registerStep("awaitPayment", async (input: { orderId: string }, _cfg: void, ctx) => {
  const reply = await redis.get(`reply:${ctx.executionKey}`);
  if (reply) return JSON.parse(reply);

  // messageId = executionKey → the broker/consumer dedupes the re-publish after restore
  await mq.publish("payments", { orderId: input.orderId, replyTo: ctx.executionKey });
  ctx.suspend("waiting for payment reply");
});

// the reply consumer, possibly on another instance:
//   await redis.set(`reply:${msg.replyTo}`, JSON.stringify(payload));
//   const snapshot = JSON.parse(await redis.get(`flow:${msg.runId}`));
//   runner.restore(snapshot);   // the step re-runs and finds the reply

applyConfig#

The declarative layer — everything that is JSON — can be loaded or replaced at any point of the runner's life. The typical scenario: flow configs arrive from a backend after the app has started.

interface EngineConfig {
  flows?: Record<string, FlowConfig>;
  hooks?: Record<string, HookEntry[]>;
  guards?: Record<string, FlowGuardConfig>;
  backgroundFlows?: Record<string, BackgroundFlowEntry[]>;
}

const json = await fetch("/api/flow-config").then((r) => r.json());

runner.applyConfig(json);              // atomic: all or nothing
runner.run("welcome", input);          // new runs see the new version

• Atomic. The whole document is validated against the step/predicate/source registries first; any error → FlowValidationError and no change is applied.
• mode: "merge" (default) upserts by name — new names appear, existing ones (including builder-registered) are overridden. "replace" resets the declarative layer to "builder registrations + this document" — the way to delete names.
• Run versioning. Every run captures the registry as of its start; active and paused runs finish on their version. applyConfig affects new runs only — a flow never changes under a run's feet.
• Started background groups whose entries changed are restarted on the new version; stopped ones are simply updated.
• Steps, predicates and sources are code — they cannot appear in the document and cannot change after createFlowRunner().
• Names added at runtime are not in the type-level union — run() accepts them via its string fallback, and an unknown name is a synchronous FlowValidationError.

Events#

Two subscription levels: runner.subscribe(listener) / FlowRunnerOptions.onEvent see every run of the runner; FlowRunOptions.onEvent sees one run. A listener that throws is contained — it can never break a run.

One runId per run() / runHook() call; nested subflows and parallel branches emit into the same stream — nesting is visible in path.

Validation#

Everything TypeScript cannot see is checked before a run starts: registerFlow configs at createFlowRunner() (fail fast at startup), inline configs synchronously inside run() (throwing FlowValidationError), and backend documents explicitly:

const result = runner.validateFlow(config);  // never throws
if (!result.valid) {
  for (const issue of result.issues) {
    console.warn(issue.severity, issue.code, issue.path.join("."), issue.message);
  }
}

Validation results are cached per config object and revalidated only when the declarative registry version moves — repeated run() calls with the same config pay nothing.

Recipes

Frontend example#

In a browser the snapshot store is localStorage and "process start" is app init: persist on every checkpoint, delete on flowEnd, and after a page refresh continue everything unfinished. This is the whole integration:

const runner = createFlowRunner(setup, {
  onEvent: (e) => {
    if (e.type === "checkpoint") {
      localStorage.setItem(`flow:${e.runId}`, JSON.stringify(e.snapshot));
    }
    if (e.type === "flowEnd") {
      localStorage.removeItem(`flow:${e.runId}`);  // terminal — snapshot no longer needed
    }
  },
});

// on app init: continue everything unfinished
for (const key of Object.keys(localStorage)) {
  if (!key.startsWith("flow:")) continue;
  runner.restore(JSON.parse(localStorage.getItem(key)!));
}

• A run interrupted by a refresh resumes from its last committed node; the step that was in flight re-runs (at-least-once).
• Suspended runs are covered too: their final snapshot arrives through the suspended event's snapshot field (or the result), and the same init loop picks them up — the waiting step re-checks its external store and suspends again if the reply has not arrived.
• For an extra safety net against the tab closing mid-node, save handle.getSnapshot() in beforeunload.
• Two tabs restoring the same snapshot will both run it — if that matters, take a Web Lock keyed by runId before calling restore().

Operations

Backend example#

The engine is environment-agnostic; a clustered deployment with long-running steps is a composition of the primitives above. The reference shape, with RabbitMQ triggering runs and Redis holding snapshots and locks:

const runner = createFlowRunner(setup, {
  heartbeatMs: 10_000,
  queueAdapter: redisQueueAdapter,   // optional: cluster-wide FIFO per queue id
  onEvent: async (e) => {
    if (e.type === "checkpoint") await redis.set(`flow:${e.runId}`, JSON.stringify(e.snapshot));
    if (e.type === "suspended")  await redis.set(`flow:${e.runId}`, JSON.stringify(e.snapshot));
    if (e.type === "flowEnd")    await redis.del(`flow:${e.runId}`, `lock:${e.runId}`);
    if (e.type === "heartbeat")  await redis.pexpire(`lock:${e.runId}`, 30_000);  // renew the lease
  },
});

// RabbitMQ triggers runs; ack after a terminal (or suspended) result
channel.consume("flows", async (msg) => {
  const { flowName, input, runId } = JSON.parse(msg.content.toString());
  if (!(await redis.set(`lock:${runId}`, INSTANCE_ID, "NX", "PX", 30_000))) {
    return channel.ack(msg);          // someone else owns this run
  }
  const result = await runner.run(flowName, input).getResult();
  channel.ack(msg);                   // "suspended" is also an ack — restore continues it later
});

// a reaper finds dead runs: a stored snapshot whose lock expired
for (const key of await redis.keys("flow:*")) {
  const runId = key.slice(5);
  if (await redis.set(`lock:${runId}`, INSTANCE_ID, "NX", "PX", 30_000)) {
    runner.restore(JSON.parse(await redis.get(key)));
  }
}

• Long steps: heartbeat fires even mid-step, so the lock lease outlives a 5-minute handler; a stale lock + an existing snapshot marks a run another instance may pick up. The interrupted node re-runs there — at-least-once, deduped by ctx.executionKey.
• Long waits: ctx.suspend() frees the process entirely; the reply consumer restores the run wherever it lands.
• Cross-instance queues: implement FlowQueueAdapter over a Redis FIFO lock and queue: "cart" becomes cluster-wide with zero config changes.
• Rolling deploys: snapshots embed their flow configs, but steps are code — keep step ids stable between versions, and remove a step only after the snapshots referencing it have drained.

TypeScript#

An honest boundary: the compiler checks what it can see, the validator checks the rest.

The compiler catches

• Unknown step ids in inline configs — step is a union of registry keys, and each node's config type is looked up by id.
• Unknown flow names in run() / subflow (names registered earlier in the chain; runtime also resolves forward references and applyConfig names through the string fallback).
• Step handler signatures, predicate signatures, source snapshot types.
• Registered predicate names autocomplete as "$name" in condition slots.

Runtime only

• String expressions and input mappings — the content of strings is invisible to the type system; whether a mapping's result matches a step's input type cannot be checked statically. This is the honest limit of JSON configs; the compensation is validation plus errors that carry an exact path.
• run<Input, Output>() type parameters are the caller's assertion.
• Existence of "$name" predicates and of names added via applyConfig.