durable

js/task/durable.ts

fino:task/durable — durable tasks: fino:task executables whose runs execute inside a fino:workflow run.

A DurableTask is a Task in every consumer-facing way — it validates input, parses CLI argv, mounts as an AI tool, honors timeoutMs — but its handler runs inside a durable workflow execution: ctx.step() checkpoints results, ctx.sleep() and ctx.waitForSignal() park the run durably, and a resumed run replays from its last checkpoint instead of starting over.

Execution model

The handler re-executes from the top on every resume. Completed steps are replayed from the store by call order and id — they return their recorded results without re-running — so all side effects (including writer output that must not repeat) belong inside ctx.step(). Changing the sequence of step ids between runs of the same runId is a determinism error and the workflow runtime rejects it. ctx.env, ctx.cwd, ctx.prompt, and ctx.writer are per-drive values, not durable state.

run() drives a run to completion, waiting out timer parks in-process and honoring signalRun() deliveries. The start() / resume() / signalRun() surface performs exactly one drive per call and reports the parked state — schedulers (e.g. fino:jobs) own the timers in that mode.

Task-level suspend() is not available inside a durable handler: the AI-loop suspension signal would unwind through the workflow driver and persist the run as failed. Durable pauses use ctx.waitForSignal().

Without a store, runs checkpoint into a fresh in-memory store: replay semantics apply within the process, but nothing survives a restart.

import { durableTask } from 'fino:task/durable';
import { sqliteStore } from 'fino:store';

const ingest = durableTask({
  name: 'ingest',
  store: () => sqliteStore({ path: '/data/jobs.db' }),
  run: async (input: { url: string }, ctx) => {
    const doc = await ctx.step('fetch', () => fetch(input.url).then((r) => r.text()));
    await ctx.sleep('cooldown', '5s');
    return await ctx.step('summarize', () => doc.slice(0, 100));
  },
});

await ingest.run({ url: 'https://example.com' }, { runId: 'ingest-42' });

Interfaces

interface DurableTaskContext extends Omit<TaskContext, 'step' | 'suspend'> {

Context passed to a durable task handler.

Extends the task context with the workflow's durable operations. The numeric AI-loop step index moves to aiStep so step() can be the workflow checkpoint method, and suspend() always throws — durable pauses use waitForSignal().

Readonly Properties

readonly aiStep?: number

AI-loop step index (the plain TaskContext step number).

readonly workflowRunId: string

Durable workflow run id backing this execution.

readonly state: WorkflowStateBag

Durable key/value state persisted with every checkpoint.

Methods

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

Run fn once and checkpoint its result under id.

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

Invoke a workflow activity with validated input/output, checkpointed.

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

Park the run durably until duration elapses.

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

Park the run durably until signalRun() delivers name.

suspend(opts?: { reason?: string; payload?: unknown }): never

Always throws — durable pauses use waitForSignal().

interface DurableTaskOptions<Input = unknown, Output = unknown> extends Omit< TaskOptions<Input, Output>, 'run' > {

Options used to create a durable task.

Properties

store?: Store | (() => Store | Promise<Store>)

Workflow store backing this task's runs — a store instance or a lazy factory resolved once on first use. Defaults to a per-task memoryStore() (checkpointing without persistence).

run: DurableTaskHandler<Input, Output>

interface DurableTaskRunOptions extends TaskRunOptions {

Options for direct durable execution — TaskRunOptions plus per-run durability overrides.

Properties

store?: Store

Store override for this run (wins over the task-level store).

key?: string

Idempotency key recorded on the workflow run state.

interface DurableRunOptions {

Options for the one-drive control surface (start, resume, signalRun).

Properties

store?: Store
key?: string
signal?: AbortSignal
writer?: TaskOutputWriter

interface DurableRunHandle {

JSON-serializable snapshot of a durable run after one drive.

Properties

runId: string
status: WorkflowStatus
waitingOn?: WorkflowWait
result?: unknown
error?: { message: string; stack?: string }

Types

type DurableTaskHandler<Input, Output = unknown> = ( input: Input, ctx: DurableTaskContext, ) => Output | Promise<Output>

Function that performs a durable task.

Classes

class DurableTask<Input = unknown, Output = unknown> extends Task<Input, Output> {

A Task whose handler runs inside a durable fino:workflow execution.

Construct via durableTask(). All Task surfaces (direct run(), CLI parse(), AI invoke()) work unchanged; durability is added underneath by a synthetic task handler that drives the workflow run.

Constructors

constructor(opts: DurableTaskOptions<Input, Output>)

Create a durable task. Prefer the durableTask() factory.

Methods

override run(rawInput: Input, options: DurableTaskRunOptions = {}): Promise<Output>

Run to completion, waiting out durable parks in-process.

Timer parks self-resume at their due time; signal parks wait for a matching signalRun() (or the park's own timeout). The workflow state is checkpointed at every transition, so a run interrupted here can still be resumed later from its store.

import { durableTask } from 'fino:task/durable';

const t = durableTask({ name: 'noop', run: async () => 'ok' });
await t.run(undefined, { runId: 'once' });
async start( rawInput: Input, opts: DurableRunOptions & { runId?: string } = {}, ): Promise<DurableRunHandle>

Drive a run once: start it (or resume it when runId exists in the store) and return a serializable snapshot instead of waiting out parks.

import { durableTask } from 'fino:task/durable';

const t = durableTask({ name: 'nap', run: async (_i, ctx) => ctx.sleep('z', '1h') });
const handle = await t.start(undefined, {});
console.log(handle.status, handle.waitingOn?.type);
async resume(runId: string, opts: DurableRunOptions = {}): Promise<DurableRunHandle>

Resume a persisted run by one drive and return its snapshot.

import { durableTask } from 'fino:task/durable';

const t = durableTask({ name: 'nap', run: async (_i, ctx) => ctx.sleep('z', '1s') });
const handle = await t.resume('nap-run-1');
console.log(handle.status);
async signalRun( runId: string, name: string, payload?: unknown, opts: { store?: Store } = {}, ): Promise<void>

Deliver an external signal to a parked run and wake any in-process run() waiting on it. Does not drive the run — call resume() (or let the waiting run() continue) to process the delivery.

import { durableTask } from 'fino:task/durable';

const t = durableTask({ name: 'gate', run: async (_i, ctx) => ctx.waitForSignal('go') });
await t.signalRun('gate-run-1', 'go', { ok: true });

Functions

function durableTask<Input = unknown, Output = unknown>( opts: DurableTaskOptions<Input, Output>, ): DurableTask<Input, Output>

Create a durable task.

import { durableTask } from 'fino:task/durable';

const t = durableTask({
  name: 'compact',
  run: async (_input, ctx) => ctx.step('work', () => 42),
});