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 { SqliteWorkflowStore } from 'fino:workflow';
const ingest = durableTask({
name: 'ingest',
store: () => SqliteWorkflowStore.open('/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?: WorkflowStore | (() => WorkflowStore | Promise<WorkflowStore>)
Workflow store backing this task's runs — a store instance or a lazy
factory resolved once on first use. Defaults to a per-task
InMemoryWorkflowStore (checkpointing without persistence).
run: DurableTaskHandler<Input, Output>
interface DurableTaskRunOptions extends TaskRunOptions {
Options for direct durable execution — TaskRunOptions plus per-run
durability overrides.
Properties
store?: WorkflowStore
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?: WorkflowStore
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?: WorkflowStore;
} = {}): 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),
});