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 { sqliteStore } from 'fino:store';
import { 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 sqliteStore({ path: './application.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 stored 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

Store-compatible structured-cloneable 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 when it is not undefined.
  • 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(). Absent when the input was undefined.

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 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: Store

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: Store

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: Store

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

async function loadWorkflowRun(store: Store, runId: string): Promise<WorkflowState | null>

Load one detached workflow snapshot from a generic store.

function saveWorkflowRun(store: Store, state: WorkflowState): Promise<void>

Save one detached workflow snapshot to a generic store.

async function listWorkflowRuns( store: Store, filter: { workflowId?: string; status?: WorkflowStatus } = {}, ): Promise<WorkflowState[]>

List workflow snapshots from a generic store, newest first.

async function deleteWorkflowRun(store: Store, runId: string): Promise<void>

Delete one workflow snapshot and notify local watchers.

function watchRun(store: Store, 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 this runtime saves the same run 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 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.