@graphorin/workflow

Step-graph workflow engine for the Graphorin framework.

@graphorin/workflow is the durable workflow layer of the Graphorin framework. It owns the synchronous-step execution loop, the Graphorin-named primitive set (Directive, Dispatch, pause, LatestValue, AnyValue, Reducer, ListAggregate, Stream, Barrier, Ephemeral), the per-channel atomic merge resolver, the HITL pause(...) / resume(directive) lifecycle, and the AbortSignal-aware cancellation contract.

The package is library-mode-first: every primitive you need to write a small workflow (createWorkflow, createNode, Directive, Dispatch, pause, InMemoryCheckpointStore) ships from the npm package without the optional standalone server.

Dependencies

• @graphorin/core - typed contracts (WorkflowEvent, Channel, Directive, Dispatch, pause, CheckpointStore, …).
• @graphorin/observability - the Tracer interface used to record workflow.run / workflow.step / workflow.task / workflow.checkpoint spans.

For production deployments, plug in @graphorin/store-sqlite's SqliteCheckpointStore to get durable-by-default checkpoint persistence.

Quick start

import {
  createNode,
  createWorkflow,
  Directive,
  InMemoryCheckpointStore,
  latestValue,
  listAggregate,
  pause,
} from '@graphorin/workflow';

interface OrderState {
  status: 'pending' | 'validated' | 'approved' | 'shipped';
  notes: ReadonlyArray<string>;
  decision?: 'approved' | 'rejected';
}

const checkpointStore = new InMemoryCheckpointStore();

const orderProcessing = createWorkflow<OrderState>({
  name: 'order-processing',
  channels: {
    status: latestValue<OrderState['status']>({ default: 'pending' }),
    notes: listAggregate<string>({ default: [] }),
    decision: latestValue<OrderState['decision']>(),
  },
  nodes: {
    validate: createNode({
      name: 'validate',
      run: async () => ({ status: 'validated', notes: ['validated'] }),
    }),
    awaitApproval: createNode({
      name: 'awaitApproval',
      run: async () => {
        const decision = pause<{ kind: 'approval' }, 'approved' | 'rejected'>({
          kind: 'approval',
        });
        return { decision, status: decision === 'approved' ? 'approved' : 'pending' };
      },
    }),
    ship: createNode({
      name: 'ship',
      run: async () => ({ status: 'shipped', notes: ['shipped'] }),
    }),
  },
  edges: [
    { from: '__start__', to: 'validate' },
    { from: 'validate', to: 'awaitApproval' },
    { from: 'awaitApproval', to: 'ship', when: (s) => s.decision === 'approved' },
    { from: 'awaitApproval', to: '__end__', when: (s) => s.decision !== 'approved' },
    { from: 'ship', to: '__end__' },
  ],
  checkpointStore,
});

const stream = orderProcessing.execute({}, { threadId: 'order-42' });
for await (const event of stream) {
  if (event.type === 'workflow.suspended') {
    // ... show approval UI to operator ...
    const resumed = orderProcessing.resume(
      'order-42',
      new Directive({ resume: 'approved' }),
    );
    for await (const next of resumed) console.log(next);
  }
}

Highlights

• Durable execution. Every execution step ends with a checkpoint written through the pluggable CheckpointStore from @graphorin/core, carrying the full resumable frontier (pending pauses, Dispatch tasks, completed-but-unwalked nodes). A new process can pick up exactly where the previous one left off via workflow.resume(threadId, directive) - including crashed threads whose latest checkpoint is running - and retry(threadId) restarts a failed thread re-running only the failed work (successful siblings replay from persisted writes). Checkpoint writes are CAS-guarded: a racing resume loses with checkpoint-version-conflict instead of forking the timeline. Resume re-executes the paused node body from the top (snapshot resume, not deterministic replay) - side effects before a pause() must be idempotent. Checkpoint state must be JSON-safe; Map/Set/Date fail fast with state-not-serializable on every store.
• Synchronous-step semantics. Tasks planned for an execution step run in parallel; their writes merge atomically per channel; the merged state is persisted; the next step plans against the new state. The semantics are documented for predictability under concurrent writes.
• Channel descriptors as merge strategies. LatestValue (overwrite, error on collision), AnyValue (last-writer-wins), Reducer((prev, next) => merged), ListAggregate (append), Stream (append-only queue, optional uniqueness), Barrier(['a', 'b']) (keyed join - a node fed by 2+ of the barrier's writers waits for every writer and runs exactly once), Ephemeral (per-step value).
• HITL via pause(value). A node calls pause(...); the engine catches the signal, persists state, and yields a workflow.suspended event. workflow.resume(threadId, new Directive({ resume })) re-enters the paused node with the supplied value.
• Static pauseAt. Declare pauseAt: { before: ['shipOrder'], after: ['chargeCard'] } to suspend without hand-rolling pause(...) inside the node body.
• Durable primitives. sleepFor(ms) / sleepUntil(iso) suspend a thread on a durable timer; awaitExternal(name) parks the thread on an awakeable that workflow.resolveAwakeable(threadId, name, value) completes; requestApproval(name, payload?) + workflow.approve( threadId, name, decision) persist a typed approval round-trip resolved by name.
• Firing durable timers (W-032). createTimerDriver({ workflows: [{ workflow, checkpointStore }] }) polls each store's listSuspended(namespace, { dueBefore }) (the engine stamps CheckpointMetadata.wakeAt on timer suspends) and calls workflow.tick(threadId) on due threads - start()/stop() in library mode, or hand the driver to createServer({ workflowTimers: { driver } }) for lifecycle + /v1/health integration. A store without listSuspended fails fast with TimerDriverStoreUnsupportedError; manual workflow.tick() remains available for one-off firing. Per-node timeoutMs / retry (via nodeDefaults or per node) bound flaky steps with node-timeout; WorkflowConfig.version pins a definition to its checkpoints (workflow-version-mismatch / workflow-divergence on drift), and journalSteps journals each completed task's channel writes so crash recovery replays them exactly once and re-runs only unfinished tasks. The task's SIDE EFFECTS stay at-least-once: the journal entry lands after the task finishes, so a crash inside that window re-runs the task - keep effects idempotent for strict once-semantics.
• Dynamic parallelism via Dispatch(node, args). A node returns one or more Dispatch('processOrder', { orderId }) directives; the engine schedules each as a parallel task in the next execution step.
• Cancellation. workflow.execute(input, { signal }) accepts an AbortSignal; aborting it stops the run within the configurable grace window (default 100 ms) and produces a structured WorkflowAbortedError. Pending tasks see the same signal via ctx.signal.
• Stream modes. Pass { stream: 'updates' } (or 'values', 'tasks', 'checkpoints', 'debug', 'custom'; 'messages' is reserved) to choose the granularity of the emitted events.
• fork(threadId, fromCheckpointId). Create a parallel timeline branched off a previous checkpoint without touching the original thread.
• Pluggable observability. Pass the tracer from @graphorin/observability to record workflow.run, workflow.step, workflow.task, workflow.checkpoint spans.
• Typed error surface. WorkflowError base class with the stable code discriminator covers invalid-config, invalid-channel-write, multi-write-into-latest-value, unknown-node, thread-not-found, checkpoint-not-found, checkpoint-version-conflict, resume-without-suspension, concurrent-resume-rejected, pause-not-found, workflow-aborted, workflow-cancel-timeout, max-steps-exceeded, node-execution-failed, node-timeout, reducer-failed, state-validation-failed, workflow-version-mismatch, workflow-divergence, dead-end, state-not-serializable.

Composition with @graphorin/agent

@graphorin/workflow does not depend on @graphorin/agent. The two compose orthogonally - a workflow node may invoke agent.run(...) directly from its run(state, ctx) body, but no import edge ever crosses between the two packages. The agent's in-loop fan-out helper (Agent.fanOut(...)) and the workflow's node-level Dispatch(...) solve different problems and live in different lifecycles:

| Primitive | Lives in | Lifecycle | Durability | | -------------------- | ----------------------- | ---------------------------------- | ------------------- | | Dispatch(...) | @graphorin/workflow | per workflow execution step | checkpointed | | Agent.fanOut(...) | @graphorin/agent | per agent run (single agent step) | inline (no per-child checkpoint) |

Use Dispatch(...) when the parallel work needs to survive process restart, when the parallel tasks are durable graph nodes with their own edges, OR when the parallel work spans multiple workflow execution steps. Use Agent.fanOut(...) when the parallel work is inline within an agent's reasoning loop, when the children are sub-agents (not arbitrary node functions), AND when the result is consumed by the parent agent's continuing loop without checkpointing.

Documentation

The full architecture lives in the framework docs (the workflow engine reference). The package's CHANGELOG.md records every change.

License

MIT © Oleksiy Stepurenko. See LICENSE for the full text.

Project Graphorin · v0.8.0 · MIT License · © 2026 Oleksiy Stepurenko · https://github.com/o-stepper/graphorin