@lunora/workflow

@lunora/workflow gives Lunora durable workflows: chain steps into a single unit that survives Worker restarts and redeploys and replays deterministically on failure. Each step's result is persisted and memoized: when a later step throws, only that step retries; the completed ones don't run again.

It wraps Cloudflare Workflows (a first-party, GA durable-execution engine) with type-safe authoring: you write a defineWorkflow and codegen emits the WorkflowEntrypoint class, wires a typed ctx.workflows handle, and reconciles the [[workflows]] binding in wrangler.jsonc for you.

Use it for sagas, multi-stage AI pipelines, human-in-the-loop approvals, and any long-running orchestration where "re-run the whole job from scratch" isn't good enough.

Define a workflow

Workflows live in lunora/workflows.ts as named exports. Scaffold one with vis generate lunora-workflow --name=orderPipeline, or write it by hand:

// lunora/workflows.ts
import { defineWorkflow } from "@lunora/workflow";

import { api } from "./_generated/api";
import type { Id } from "./_generated/server";

export const channelWelcome = defineWorkflow<
    { channelId: Id<"channels"> }, // params (input)
    { channelId: Id<"channels">; posted: number } // output
>({
    handler: async (context) => {
        const { channelId } = context.params;

        // A durable, memoized, retried step.
        await context.step.do("greet", () => context.run(api.messages.send, { channelId, text: "👋 Welcome!" }, { shardKey: channelId }));

        // The workflow hibernates here and resumes after a minute —
        // the delay survives Worker evictions and redeploys.
        await context.step.sleep("settle", "1 minute");

        await context.step.do("tips", () => context.run(api.messages.send, { channelId, text: "Tip: messages stream live." }, { shardKey: channelId }));

        context.log.info("welcome sequence complete", { channelId });

        return { channelId, posted: 2 };
    },
});

defineWorkflow<Params, Output> takes a single config object:

• handler: the workflow body; receives the run context (below).
• name (optional): override the deployed workflow name. Defaults to the kebab-cased export name (orderPipeline → order-pipeline).

The run context

The handler receives one argument bundling the durable-step API and a typed function caller:

Durable steps

context.step is the Cloudflare Workflows step API:

// Memoized + retried. On replay, a completed step returns its stored
// result without re-running the callback.
const order = await context.step.do("load", () => context.run(api.orders.get, { id: orderId }, { shardKey: orderId }));

// Per-step retry policy.
await context.step.do("charge", { retries: { limit: 3, delay: "2 seconds", backoff: "exponential" }, timeout: "30 seconds" }, () =>
    context.run(api.payments.charge, { orderId }),
);

// Durable delays — minutes to weeks, not held in memory.
await context.step.sleep("cool-off", "1 minute");
await context.step.sleepUntil("run-at", new Date("2026-07-01T00:00:00Z"));

// Pause until an external event (approvals, webhooks, human-in-the-loop).
// Prefer `context.waitForEvent` with a declared event — see below.
const approval = await context.step.waitForEvent<{ approved: boolean }>("await-approval", {
    type: "approval",
    timeout: "7 days",
});

Declared events (defineWorkflowEvent)

The native step.waitForEvent(name, { type }) matches instance.sendEvent({ type }) by a bare string, and the payload crosses as unknown. Both ends drift silently: a typo'd type hibernates the instance until its timeout (24 hours by default) with no error anywhere, and a payload whose shape changed resumes the workflow on garbage. Declare the event once instead, so there is no literal to typo and no second place to update on a rename, and both ends parse the payload:

// lunora/events.ts
import { defineWorkflowEvent } from "@lunora/workflow";
import { v } from "@lunora/values";

export const orderApproved = defineWorkflowEvent("order-approved", v.object({ approvedBy: v.string() }));

Wait on it inside the workflow — the payload is typed and parsed:

const { approvedBy } = await context.waitForEvent(orderApproved, { name: "await approval", timeout: "7 days" });

Send it from a mutation or action with the same definition:

export const approve = mutation.input({ approvedBy: v.string(), instanceId: v.string() }).mutation(async ({ ctx, args: { approvedBy, instanceId } }) => {
    await ctx.workflows.get("orderPipeline").sendEvent(instanceId, orderApproved, { approvedBy });
});

The payload is validated on send (a bad value fails the caller's request, before the workflow is woken) and again on receive. A payload that fails the validator on receive is a non-retryable failure — the event has already been consumed, so replaying the wait could only hibernate the instance again.

What this does not buy is a compile error for every mismatch: two definitions with the same payload shape are mutually assignable, so sending orderRejected where the workflow awaits orderApproved still type-checks. What it removes is the hand-matched string.

Event types beginning with lunora: are reserved for the framework's own instance-to-instance protocol (the context.parallel branch join). defineWorkflowEvent and the typed sendEvent above reject them; the native untyped instance.sendEvent({ type, payload }) does not go through that check.

Reusable steps (defineStep)

An inline context.step.do("name", () => …) is fine for one-offs, but defineStep lets you define a step once, schema-validated and reusable across workflows. Args are validated (with @lunora/values) before the body runs, and the return value is validated after (when you declare returns), so a bad payload fails fast instead of corrupting later steps. Scaffold one with vis generate lunora-step --name=chargeOrder (appends to lunora/steps.ts), or write it by hand:

// lunora/steps.ts
import { defineStep } from "@lunora/workflow";
import { v } from "@lunora/values";

import { api } from "./_generated/api";

export const charge = defineStep("charge", {
    args: { orderId: v.string(), amount: v.number() },
    returns: v.object({ receiptId: v.string() }),
    config: { retries: { limit: 3, backoff: "exponential", delay: "10 seconds" } },
    handler: async (context, { orderId, amount }) => {
        if (context.attempt > 1) context.log.warn(`retrying charge for ${orderId}`);

        return context.run(api.payments.charge, { orderId, amount });
    },
    // Compensation — runs if a *later* step fails after this one committed.
    rollback: async (context) => {
        await context.run(api.payments.refund, { orderId: context.args.orderId });
    },
});

Run it from a workflow body; context.runStep wraps it in a durable step.do(...) for you:

export const orderPipeline = defineWorkflow<{ orderId: string }>({
    handler: async (context) => {
        const { receiptId } = await context.runStep(charge, { orderId: context.params.orderId, amount: 4200 });

        return { receiptId };
    },
});

The step handler context gives you context.attempt (1-based retry counter), context.config, context.env, context.run, context.log, and context.step ({ name, count }). Pass context.runStep(step, args, { config }) to override the step's durability config for a single call.

Fan-out with child-DO isolation (ctx.parallel / ctx.spawn)

A workflow instance is one Durable Object: ~128 MB memory and a 5-minute CPU budget, shared across everything it runs. Fanning out heavy work with Promise.all(branches.map((b) => ctx.runStep(b, …))) therefore crowds every branch onto that one budget: one OOM or timeout takes the whole batch down.

ctx.parallel runs each branch as its own child workflow instance, with its own DO and its own memory / CPU / retry budget, and the parent hibernates (zero cost) until the branches report back. Branches reference declared child workflows by their lunora/workflows.ts export name; branch<Output>(name, params) carries the result type into the returned tuple, in declaration order:

import { branch, defineWorkflow } from "@lunora/workflow";

export const mediaPipeline = defineWorkflow<{ key: string }>({
    handler: async (ctx) => {
        // Each branch is a separate declared workflow running in its own DO.
        const [labels, thumb, transcript] = await ctx.parallel([
            branch<{ tags: string[] }>("imageTag", { key: ctx.params.key }),
            branch<{ url: string }>("thumbnail", { key: ctx.params.key }),
            branch<{ text: string }>("transcribe", { key: ctx.params.key }),
        ]);

        return { labels, thumb, transcript };
    },
});

• Isolated resources: each branch gets a full DO budget; a heavy branch can't starve its siblings.
• Hibernating join: the parent consumes nothing while branches run; each child signals its result back when it finishes.
• Fail-fast: if any branch fails, ctx.parallel rejects with that branch's error (non-retryable, because retrying the join can't re-run an already-failed child). Still-running siblings are left to finish (Cloudflare can't cleanly cancel a running instance).
• Replay-safe: child instance ids are derived deterministically from the parent, so a parent replay re-attaches to the existing children instead of double-spawning. There is a cap of MAX_BRANCHES (100) per call.
• Bounded branch output: a branch reports back over Cloudflare's event channel, which caps a payload at 1 MiB. A larger return value can never reach the parent, so the branch is failed immediately with BranchOutputTooLarge naming its byte count — rather than leaving the parent hibernating on its join until the branch timeout (Cloudflare's default is 24 hours) and then compensating a branch that had succeeded. Return a reference the parent can dereference (an R2 key, a row id) instead of the payload itself.

For fire-and-forget (start a child pipeline without awaiting it), use ctx.spawn(name, params), which returns a live instance handle:

await ctx.spawn("sendReceipt", { orderId: ctx.params.orderId });

Both reach child workflows through the standard WORKFLOW_* bindings codegen already wires, so there is nothing extra to configure. A branch/spawn name with no matching declared workflow throws an error.

Rollback (saga compensation)

A step's optional rollback handler is forwarded to Cloudflare's native step rollback: it runs when a later step in the same instance fails, so you can undo a committed side effect (refund a charge, delete an uploaded object). The rollback context carries the original args, the error that triggered it, the step's output (if it completed), and env / run / log. Cloudflare owns rollback ordering and execution; defineStep just wires the handler and an optional rollbackConfig.

That native rollback is intra-instance: it undoes steps within one workflow. To compensate across a ctx.parallel(...) group (each branch is its own instance), give a branch a compensateWith, the export name of a workflow to run if a sibling branch fails after this one already completed:

const [charge, reserve] = await ctx.parallel([
    branch("chargeCard", { orderId }, { compensateWith: "refundCard" }),
    branch("reserveStock", { orderId }, { compensateWith: "releaseStock" }),
]);

If any branch fails, every already-completed sibling with a compensateWith is rolled back in reverse declaration order: its compensation workflow is spawned (durable, replay-safe) with { branch, error, index, output } as its ctx.params, so refundCard receives what chargeCard returned plus the failing sibling's error, and can ctx.runStep(...) its own undo. A group where no branch sets compensateWith behaves exactly as a plain fail-fast fan-out, with zero overhead until you opt in. The failing branch itself is not group-compensated (its own per-step rollbacks already ran inside its instance).

Failing without retries (NonRetryableError)

Throw NonRetryableError from a step or the handler to fail the instance immediately, skipping retries. It is the portable, Node-importable mirror of cloudflare:workflows' native error (so your workflow code stays unit-testable). The runtime converts it to the native error at the workflow boundary.

import { NonRetryableError } from "@lunora/workflow";

export const charge = defineStep("charge", {
    args: { orderId: v.string() },
    handler: async (context, { orderId }) => {
        const order = await context.run(api.orders.get, { id: orderId });

        if (order.status === "cancelled") {
            throw new NonRetryableError("order cancelled — retrying will never succeed");
        }

        return context.run(api.payments.charge, { orderId });
    },
});

Start and manage instances

Codegen wires a typed ctx.workflows handle onto mutations and actions. Start an instance from a function:

// lunora/channels.ts
import { mutation, v } from "./_generated/server";

export const create = mutation.input({ channelId: v.id("channels") }).mutation(async ({ ctx, args: { channelId } }) => {
    const instance = await ctx.workflows.get("channelWelcome").create({ params: { channelId } });

    return { instanceId: instance.id };
});

ctx.workflows.get("<exportName>") returns a handle with:

• create({ id?, params?, retention? }): start an instance. create() resolves once the instance is queued (not when it finishes); the workflow runs on Cloudflare's infrastructure.

  The queue is immediate and not transactional: create() is an RPC that leaves the shard as it is called, and nothing defers it to the surrounding mutation's commit. A mutation that starts a workflow and then throws has already queued an instance, and that instance runs against state the mutation never wrote. Start workflows from an action that calls the mutation first (await ctx.runMutation(...), then create()), so the write is committed before the instance exists.

• createBatch([...]): start many in one batch.

• get(id): a handle to an existing instance.

• sendEvent(instanceId, event, payload): deliver a declared event to one instance, to satisfy its context.waitForEvent. The payload is validated before the send.

create/get hand back the native Cloudflare instance, which exposes its own lifecycle: status() (returns status + output/error), pause(), resume(), restart(), terminate(), and the untyped sendEvent({ type, payload }).

Listing instances and step timelines

The Workflow binding can only create/get and read a single instance's status: it has no instance list and no per-step detail. To list instances or read a step timeline (what the Studio renders), use createWorkflowsRestClient, which talks to Cloudflare's account-scoped Workflows REST API. The API token is a secret, so keep this server-side:

import { createWorkflowsRestClient } from "@lunora/workflow";

const client = createWorkflowsRestClient({
    accountId: env.CLOUDFLARE_ACCOUNT_ID,
    apiToken: env.CLOUDFLARE_API_TOKEN, // scope: Workflows Read (Edit for setInstanceStatus)
});

const page = await client.listInstances({ workflowName: "order-pipeline", status: "running" });
const detail = await client.getInstance({ workflowName: "order-pipeline", instanceId: page.instances[0].id });

getInstance returns the instance plus a normalized steps[] timeline (each step's type, attempts, timing, output, and error). setInstanceStatus({ action }) pauses, resumes, or terminates an instance and requires an Edit-scoped token.

Wiring (handled by codegen)

After adding lunora/workflows.ts, re-run codegen (lunora dev does this on save). It:

1. Emits a WorkflowEntrypoint subclass per export into lunora/_generated/workflows.ts (channelWelcome → ChannelWelcomeWorkflow).
2. Wires the typed ctx.workflows handle.
3. Reconciles the binding, class_name, and name entries of wrangler.jsonc's workflows[] array from the same definition.

Your worker entry must re-export the generated classes so wrangler can find them:

// src/server/index.ts
export { ChannelWelcomeWorkflow } from "../../lunora/_generated/workflows";

The resulting binding:

// wrangler.jsonc (codegen-reconciled)
{
    "workflows": [{ "binding": "WORKFLOW_CHANNEL_WELCOME", "class_name": "ChannelWelcomeWorkflow", "name": "channel-welcome" }],
}

Advisor lints

Two static advisors keep workflow wiring honest:

• workflow_unknown_target (error): ctx.workflows.get("name") references a workflow that doesn't exist (typo or removed export).
• workflow_unused (info): a declared workflow is never started by any function (dead code that's still billed as a WorkflowEntrypoint), unless it's triggered externally.

Studio

The Studio's Workflows panel (under the Functions domain) lists every declared workflow with its export name, generated class, binding, and deployed name; lets you start an instance from a JSON-params form; and shows a table of instances with their live status (queued / running / complete / errored) and output. See the Studio guide.

See also

• Scheduler: runAfter / runAt / cron, for single-shot deferred work that doesn't need replay.
• Studio: the Workflows inspector.
• Advisors: the workflow lints.