workflow

js/workflow.ts

fino:workflow — local-first durable workflows for ordinary TypeScript code.

Workflows are async functions that can checkpoint named steps, call reusable activities, wait for timers, and wait for external signals. Use this module when an operation must survive process restarts or span request boundaries, without adopting a distributed worker platform.

Execution model

Workflow code may run more than once. Put external side effects, nondeterministic values, and expensive operations inside ctx.step() or ctx.call() so their results are saved and reused on resume. Ordinary TypeScript control flow (if, loops, Promise.all) remains the workflow structure.

import { SqliteWorkflowStore, activity, workflow } from 'fino:workflow';

const createTicket = activity({
  id: 'create-ticket',
  async run(input: { title: string }) {
    return { id: `ticket:${input.title}` };
  },
});

const routeTicket = workflow({
  id: 'route-ticket',
  async run(ctx, input: { title: string }) {
    const ticket = await ctx.call(createTicket, input);
    const approved = await ctx.waitForSignal<boolean>('approval');
    return { ticketId: ticket.id, approved };
  },
});

const store = await SqliteWorkflowStore.open('./workflow.db');
const run = await routeTicket.start({ title: 'refund request' }, { store });
await routeTicket.signal({ store, runId: run.runId, name: 'approval', payload: true });

Types

type WorkflowStatus = 'running' | 'waiting' | 'done' | 'error' | 'cancelled'

Status persisted for a workflow run.

running means the workflow can be driven immediately. waiting means it paused on a timer or signal and records that wait in WorkflowState.waitingOn. done, error, and cancelled are terminal states.

type WorkflowWait = { type: 'timer'; id: string; dueAt: number; stepIndex: number; } | { type: 'signal'; id: string; name: string; stepIndex: number; timeoutAt?: number; }

Durable wait recorded when a workflow pauses for time or an external signal.

Timer waits record an absolute dueAt timestamp. Signal waits record the signal name and optional absolute timeoutAt timestamp. stepIndex ties either wait back to the checkpoint position that created it.

Interfaces

interface WorkflowRetryOptions {

Retry policy used by activity() and ctx.step().

Fields
  • maxAttempts caps the total number of attempts. Missing or invalid values below 1 behave as 1.

Readonly Properties

readonly maxAttempts?: number

Maximum number of times to run the activity or step body before surfacing the last error. Values below 1 are treated as 1.

A thrown NonRetryableWorkflowError stops retrying immediately.

interface WorkflowStepState {

One completed checkpointed call.

Each record stores the checkpoint id, the completed status, and the optional serialized result that replay returns instead of rerunning the step.

Readonly Properties

readonly id: string

Stable step id supplied to ctx.step(), the called activity id, the sleep id, or the signal name.

readonly status: 'done'

Step completion marker. Only completed steps are persisted.

readonly result?: unknown

JSON-serializable result reused when workflow code replays.

interface WorkflowState {

Persisted workflow run state.

Fields
  • runId uniquely identifies the run.
  • workflowId identifies the workflow definition that owns the run.
  • status tracks lifecycle state.
  • cursor is the first contiguous checkpoint index not yet complete.
  • input is the original start() input.
  • steps stores completed checkpoints in call order.
  • state stores the durable ctx.state bag.
  • signals queues delivered signals until ctx.waitForSignal() consumes them.
  • createdAt and updatedAt are Unix timestamps in milliseconds.
  • key is an optional caller-supplied lookup or idempotency key.
  • waitingOn describes the current timer or signal wait.
  • result holds the final output for done runs.
  • error holds serialized failure details for error runs.

Properties

runId: string

Unique id for this workflow run.

workflowId: string

Workflow.id for the definition that owns the run.

status: WorkflowStatus

Current lifecycle state. waiting runs have waitingOn; done runs have result; error runs have error.

cursor: number

First checkpoint index that has not completed contiguously.

input: unknown

Original input passed to Workflow.start().

steps: WorkflowStepState[]

Completed checkpoint records in call order. Sparse positions can appear while parallel steps finish out of order.

state: Record<string, unknown>

Durable key/value bag behind ctx.state.

signals: WorkflowSignal[]

Queued external signals that have been delivered but not yet consumed by a matching ctx.waitForSignal() call.

createdAt: number

Unix timestamp in milliseconds when the run was first saved.

updatedAt: number

Unix timestamp in milliseconds for the last saved update.

key?: string

Optional caller-supplied idempotency or lookup key.

waitingOn?: WorkflowWait

Current timer or signal wait when status is waiting.

result?: unknown

Final validated output when status is done.

error?: { message: string; stack?: string; }

Serialized failure details when status is error.

interface WorkflowSignal {

Signal delivered to a waiting workflow.

name is matched against ctx.waitForSignal(name), payload is returned from that wait, and receivedAt records when Workflow.signal() stored it.

Properties

name: string

Signal name matched against ctx.waitForSignal(name).

payload: unknown

Payload returned from ctx.waitForSignal().

receivedAt: number

Unix timestamp in milliseconds when Workflow.signal() stored the signal.

interface WorkflowStore {

Store contract for durable workflow state.

Implementations persist full WorkflowState snapshots. load() returns null for unknown run ids, list() may filter by workflowId and status, and delete() may treat missing runs as a successful no-op.

Methods

save(state: WorkflowState): Promise<void>

Persist a full workflow state snapshot.

load(runId: string): Promise<WorkflowState | null>

Load one workflow run by id, or null when the store has no matching run.

list(filter?: { workflowId?: string; status?: WorkflowStatus; }): Promise<WorkflowState[]>

Return persisted runs, optionally filtered by workflow id or lifecycle status.

delete(runId: string): Promise<void>

Delete one run by id. Stores may treat unknown ids as a successful no-op.

interface ActivityDef< In, Out > {

Activity definition passed to activity().

Fields
  • id is the stable checkpoint id used by ctx.call().
  • inputSchema and outputSchema optionally validate activity boundaries.
  • retry supplies the default retry policy.
  • run performs the side effect or expensive work that should not repeat on workflow replay.

Readonly Properties

readonly id: string

Stable activity id used as the checkpoint id for ctx.call().

readonly inputSchema?: unknown

Optional validation schema for activity inputs.

readonly outputSchema?: unknown

Optional validation schema for activity outputs.

readonly retry?: WorkflowRetryOptions

Default retry policy for calls to this activity.

Methods

run(input: In, ctx: WorkflowContext): Promise<Out> | Out

Activity body. Put external side effects here so workflow replay reuses the persisted result instead of repeating the effect.

interface WorkflowStateBag {

Mutable durable state helper exposed on WorkflowContext.

get(), set(), and clear() update the run's persisted state object. entries() returns a shallow copy so callers can inspect the bag without mutating it directly.

Methods

get<T = unknown>(key: string): T | undefined

Read a durable value by key, returning undefined when the key is absent.

set(key: string, value: unknown): void

Set or replace a durable value by key.

clear(key: string): void

Remove a durable value by key.

entries(): Record<string, unknown>

Return a shallow copy of all durable state entries.

interface WorkflowContext {

Context passed to a workflow body and activity.

Properties
  • runId and workflowId identify the current execution.
  • signal is the optional abort signal passed to start() or resume().
  • state is the run-scoped durable key/value bag.
Checkpoints

step() and call() persist results and reuse them on replay. sleep() and waitForSignal() persist waits and return a waiting run result until the wait can complete.

Readonly Properties

readonly runId: string

Id of the run currently being executed.

readonly workflowId: string

Id of the workflow definition currently being executed.

readonly signal?: AbortSignal

Optional abort signal passed through start or resume options.

readonly state: WorkflowStateBag

Durable key/value state scoped to this run.

Methods

step<T>(id: string, fn: () => Promise<T> | T, opts?: { retry?: WorkflowRetryOptions; }): Promise<T>

Execute a named checkpoint. On replay, a completed checkpoint with the same id returns its persisted result without calling fn.

call< In, Out >(activity: Activity<In, Out>, input: In, opts?: { retry?: WorkflowRetryOptions; }): Promise<Out>

Call a reusable activity as a checkpointed operation.

sleep(id: string, duration: number | string | Date): Promise<void>

Pause until a relative duration, duration string, or absolute date is due.

Numeric durations are milliseconds. String durations accept ms, s, m, h, and d units.

waitForSignal<T = unknown>(name: string, opts?: { timeout?: number | string | Date; }): Promise<T>

Pause until a matching external signal is delivered.

The resolved value is the signal payload. When timeout is provided and is already expired during resume, the workflow throws a timeout error.

interface WorkflowDef< In, Out > {

Workflow definition passed to workflow().

Fields
  • id is the stable workflow definition id persisted with every run.
  • inputSchema and outputSchema optionally validate run boundaries.
  • run contains the workflow body. Use ctx.step() and ctx.call() around nondeterministic work so resume can replay deterministically.

Readonly Properties

readonly id: string

Stable workflow definition id recorded in each run.

readonly inputSchema?: unknown

Optional validation schema checked before the run is first saved.

readonly outputSchema?: unknown

Optional validation schema checked before a run completes as done.

Methods

run(ctx: WorkflowContext, input: In): Promise<Out> | Out

Workflow body. Keep nondeterministic work inside ctx.step() or ctx.call() so replay can reuse durable checkpoints.

interface WorkflowStartOptions {

Options for starting a workflow run.

store is required. runId and key let callers supply identifiers, signal is exposed on ctx.signal, and onCheckpoint runs after each checkpoint snapshot is persisted.

Readonly Properties

readonly store: WorkflowStore

Durable store that owns the run state.

readonly runId?: string

Optional caller-supplied run id. When omitted, Fino generates one.

readonly key?: string

Optional idempotency or lookup key persisted on WorkflowState.key.

readonly signal?: AbortSignal

Abort signal observed by workflow code through ctx.signal.

readonly onCheckpoint?: (state: WorkflowState) => void

Called after each checkpoint is persisted.

interface WorkflowResumeOptions {

Options for resuming a workflow run.

store and runId select the persisted run. signal is exposed on ctx.signal, and onCheckpoint runs after each new checkpoint snapshot is persisted.

Readonly Properties

readonly store: WorkflowStore

Durable store that owns the run state.

readonly runId: string

Existing run id to load and resume.

readonly signal?: AbortSignal

Abort signal observed by workflow code through ctx.signal.

readonly onCheckpoint?: (state: WorkflowState) => void

Called after each checkpoint is persisted.

interface WorkflowSignalOptions {

Signal delivery options.

store and runId select a waiting run. name must match the run's current signal wait, and payload becomes the value returned by ctx.waitForSignal().

Readonly Properties

readonly store: WorkflowStore

Durable store that owns the target run state.

readonly runId: string

Existing waiting run id.

readonly name: string

Signal name that must match the run's current waitingOn.name.

readonly payload?: unknown

Optional payload returned by ctx.waitForSignal().

interface WorkflowResult<Out = unknown> {

Result returned from start and resume.

runId, status, and state are always present. waitingOn is present for waits, and result is populated when the workflow reaches done.

Properties

runId: string

Run id that was started or resumed.

status: WorkflowStatus

Lifecycle state after driving the workflow.

state: WorkflowState

Full durable state snapshot after driving the workflow.

waitingOn?: WorkflowWait

Current wait details when status is waiting.

result?: Out

Final workflow output when status is done.

Functions

function observableWorkflowStore(inner: WorkflowStore): WorkflowStore

Wrap a workflow store so every save and delete publishes a run update.

The wrapped store remains the durable source of truth. Notifications only tell local watchers to reload the run state.

function watchRun(store: WorkflowStore, runId: string): ReadonlySignal<WorkflowState | null>

Watch one workflow run in a store.

The returned signal starts as null, loads the current state asynchronously, and refreshes whenever an observableWorkflowStore() wrapper publishes a run update for the same id.

function activity< In = unknown, Out = unknown >(def: ActivityDef<In, Out>): Activity<In, Out>

Create a reusable workflow activity.

function workflow< In = unknown, Out = unknown >(def: WorkflowDef<In, Out>): Workflow<In, Out>

Create a durable workflow.

Classes

class InMemoryWorkflowStore implements WorkflowStore {

In-memory store for tests and single-process prototypes.

Methods

async save(state: WorkflowState): Promise<void>

Save or replace one workflow run state.

async load(runId: string): Promise<WorkflowState | null>

Load a workflow run by id, or null when it is unknown.

async list(filter: { workflowId?: string; status?: WorkflowStatus; } = {}): Promise<WorkflowState[]>

List runs, newest first, optionally filtered by workflow id or status.

async delete(runId: string): Promise<void>

Delete one workflow run if it exists.

class SqliteWorkflowStore implements WorkflowStore {

Database-backed workflow store.

The class name is retained for compatibility. Pass a sqlite:// path or a postgres:// URL to choose the storage engine through fino:database.

Static Methods

static async open(path: string, opts?: { fs?: object; }): Promise<SqliteWorkflowStore>

Open a workflow store and create the required tables.

Methods

async save(state: WorkflowState): Promise<void>

Save or replace one workflow run state.

async load(runId: string): Promise<WorkflowState | null>

Load a workflow run by id, or null when it is unknown.

async list(filter: { workflowId?: string; status?: WorkflowStatus; } = {}): Promise<WorkflowState[]>

List persisted runs, newest first, optionally filtered by workflow id or status.

async delete(runId: string): Promise<void>

Delete one workflow run if it exists.

async close(): Promise<void>

Close the underlying SQLite database.

class NonRetryableWorkflowError extends Error {

Error that prevents retries for a failed activity or step.

Constructors

constructor(message: string)

class WorkflowError extends Error {

Error used when workflow execution fails.

Readonly Properties

readonly state: WorkflowState

Durable state associated with the failed workflow.

Constructors

constructor(message: string, state: WorkflowState)

class Activity< In = unknown, Out = unknown > {

Reusable effectful operation called from a workflow.

Readonly Properties

readonly id: string

Stable activity id used as the checkpoint id for ctx.call().

readonly inputSchema?: unknown

Optional validation schema checked before run() is called.

readonly outputSchema?: unknown

Optional validation schema checked after run() resolves.

readonly retry?: WorkflowRetryOptions

Default retry policy used when ctx.call() does not provide one.

readonly run: (input: In, ctx: WorkflowContext) => Promise<Out> | Out

Activity body supplied to activity().

Constructors

constructor(def: ActivityDef<In, Out>)

class Workflow< In = unknown, Out = unknown > {

Durable workflow definition.

Readonly Properties

readonly id: string

Stable workflow definition id recorded in every run.

readonly inputSchema?: unknown

Optional schema used to validate start() input before the run is saved.

readonly outputSchema?: unknown

Optional schema used to validate the final workflow result.

Constructors

constructor(def: WorkflowDef<In, Out>)

Create a workflow definition from WorkflowDef.

Methods

start(input: In, opts: WorkflowStartOptions): Promise<WorkflowResult<Out>>

Start a new durable run and drive it until it completes, fails, or waits.

resume(opts: WorkflowResumeOptions): Promise<WorkflowResult<Out>>

Resume a persisted run from its last checkpoint.

async signal(opts: WorkflowSignalOptions): Promise<void>

Deliver an external signal to a run waiting in ctx.waitForSignal().

class WorkflowRun< In = unknown, Out = unknown > {

Stateful workflow run handle.

Constructors

constructor(workflow: Workflow<In, Out>, opts: WorkflowStartOptions | WorkflowResumeOptions)

Create a handle for starting or resuming one workflow run.

Getters

get state(): WorkflowState | undefined

Last loaded state for this handle, if it has started or resumed a run.

Methods

async start(input: In): Promise<WorkflowResult<Out>>

Start this run with input.

async resume(): Promise<WorkflowResult<Out>>

Resume this run from durable state.

async cancel(): Promise<void>

Mark this run as cancelled and clear any pending wait.