Flows

import {
  flow,
  signalFlow,
  cancelFlow,
  listFlows,
  createFlowId,
  FlowSuspendedError,
  FlowCancelledError,
  FlowExpiredError,
} from "@use-crux/core/flow";

Overview

Flows group generate() calls into structured pipelines with named steps, automatic devtools tracing, suspend/resume, and retry/fallback. A run opens a canonical flow.run span, and each executed flow.step() opens a canonical flow.step child. Generated spans inside a flow inherit the active observability context for structured pipeline grouping in devtools. Flows can nest; calling flow().run() inside another flow preserves the parent-child relationship in the graph.

Flows support suspend/resume for human-in-the-loop workflows. When flow.suspend() is called, the flow persists its state to the configured RecordStore and returns { status: 'suspended' }. The observability graph records a flow.suspension marker linked to the causing step, and RunDetail presents that marker beside the other flow steps so successful generations and completed steps do not remain visually stuck. Pass the flow ID to .resume(flowId) to resume a previously suspended flow.

flow(name, handler)

Define a named flow and return a frozen FlowHandle. Separates definition from execution: the handler is captured once, then .run() can be called repeatedly with different inputs and options. Input-bearing flows infer their input type from the handler's second parameter.

Returns: FlowHandle<T, TInput> -- frozen handle with .run(), .resume(), .signal(), .cancel(), and .name

import { flow } from "@use-crux/core/flow";
import { generate } from "@use-crux/ai";

const researchFlow = flow(
  "research",
  async (flow, input: { topic: string }) => {
    const plan = await flow.step("plan", () =>
      generate(planner, { model, input: { topic: input.topic } }),
    );
    return flow.step("search", () =>
      generate(searcher, { model, input: { plan } }),
    );
  },
);

const result = await researchFlow.run({ topic: "TypeScript monorepos" });
if (result.status === "completed") {
  console.log(result.output);
}

FlowHandle<T, TInput>

The frozen handle returned by flow().

FlowRunOptions

Options passed to handle.run().

FlowResumeOptions

Options passed to handle.resume().

FlowScope<TInput>

The scope object passed to the flow handler. Generic over TInput (defaults to void).

step(label, fn, options?)

Execute a named step within the flow. Opens a canonical flow.step span and restores observability context so generated child spans nest under the step. On resume, completed steps are replayed from cache (skip-replay) and are not re-emitted.

For the Flows beta, label is the durable step identity. Each flow.step() label must be unique within one flow execution; duplicate labels throw before the runtime reads from the replay cache. Prefer stable literal labels. Dynamic labels are advanced and should only come from deterministic values that will be the same across retries and resumes.

fn accepts both plain functions () => T and flow-aware functions (flow: FlowScope) => T. Flow-aware functions receive the scope automatically, enabling external step definitions that access flow.input, flow.results, etc.

suspend(name, options?)

Suspend the flow and wait for an external signal. Throws internally to unwind the call stack -- no code after suspend() executes in the current call. The current flow emits a flow.suspension observability marker and rolls the flow/run presentation status to suspended; the causing step can still end ok. On resume, the signal payload is returned (typed if schema is provided).

waitUntil(name, conditionFn, options?)

Suspend the flow until a condition function returns true. Core evaluates the condition during the initial run and again only when your app explicitly resumes the flow. If the condition is true, execution continues; if false, the flow re-suspends. Core does not poll, schedule, or wake flows in the background.

Runtime-backed flow APIs

These APIs require a configured Runtime Engine. Without config({ runtime }), they throw RUNTIME_REQUIRED. Object-bound suspend(), signal(), and resume() remain available without runtime config.

waitFor(event, options?)

Suspend until a durable event arrives. On first execution Crux registers a waiter in the runtime store; on replay the same await resolves with the event payload.

const approval = await flow.waitFor(
  { name: "document.approved", schema: approvalSchema },
  { match: { documentId }, timeout: "24h" },
);

match compares top-level JSON payload fields. timeout schedules a durable timeout through the runtime.

defer(task, input)

Buffer an executable durable task to run independently. The task becomes durable at the next suspension or completion barrier, so replay cannot enqueue it twice.

const { workId } = await flow.defer(embedDocument, { documentId });

The target must be a durableTask() from @use-crux/core/runtime, not a plan-ledger task.

after(task, delay, input)

Buffer a durable child task timer. The timer is independent of the parent flow and becomes durable at the next barrier.

await flow.after(sendReminder, "2d", { userId });

untilIdle({ scope: 'current-flow' })

Suspend until child work created by the current flow reaches terminal state.

await flow.untilIdle({ scope: "current-flow" });

Global idle is not a v1 runtime primitive.

SuspendOptions<T>

StepOptions

Retry and fallback options for a flow step.

FlowResult<T>

Discriminated union on status:

Every variant also carries _meta: OperationResultMeta for the current flow.run invocation. Resume preserves the validated W3C trace and opens a fresh span, so a resumed result has the same traceId and a fresh spanId. Invocation-local result IDs are removed from persisted input, step output, signals, and scheduled work; continuation trace carriers remain intact. See Operation Result Correlation.

FlowSnapshot

Persisted flow state stored in RecordStore under the crux:flow:{flowId} key.

Lifecycle Functions

signalFlow(flowId, name, payload?)

Write a signal payload to the store at crux:signal:{flowId}:{name}. The signal is picked up on the next resume.

cancelFlow(flowId, reason?)

Cancel a suspended flow by updating its stored snapshot status to 'cancelled'.

handle.cancel(flowId)

Cancel a flow through the handle that defined it. Without a Runtime Engine, this delegates to cancelFlow(). With a Runtime Engine, it atomically marks the durable work and flow snapshot cancelled and cleans up the flow's waiter/timer registrations. Repeated calls and unknown flow ids are idempotent no-ops.

Cancelling a parent flow does not cancel independent child work created through flow.defer() or flow.after().

listFlows(options?)

Query the store for flow snapshots.

Returns: Promise<FlowSummary[]> -- array of { flowId, name, status, suspendedAt, createdAt, updatedAt, timeoutAt? }

createFlowId()

Generates a unique flow ID for cross-boundary correlation. Returns string.

Error Classes

These errors are used internally for control flow -- they are caught by the flow runtime and converted to FlowResult variants. You should not normally need to catch them.

Suspend/Resume Example

import { flow, signalFlow } from "@use-crux/core/flow";
import { z } from "zod";

const approvalFlow = flow(
  "content-approval",
  async (flow, input: { topic: string }) => {
    const draft = await flow.step("draft", () =>
      generate(writer, { model, input }),
    );

    // Suspend and wait for human approval
    const approval = await flow.suspend("review", {
      schema: z.object({
        approved: z.boolean(),
        feedback: z.string().optional(),
      }),
      timeout: "24h",
      onExpired: ({ flowId }) => console.log(`Flow ${flowId} expired`),
    });

    if (!approval.approved) {
      flow.cancel("Rejected by reviewer");
    }

    return flow.step("publish", () =>
      generate(publisher, {
        model,
        input: { draft, feedback: approval.feedback },
      }),
    );
  },
);

// Start the flow
const result = await approvalFlow.run({ topic: "AI safety" });
// result.status === 'suspended'

// Later, signal the flow to resume
await approvalFlow.signal(result.flowId, "review", {
  approved: true,
  feedback: "Great draft, just fix the intro.",
});

// Resume the flow
const final = await approvalFlow.resume(result.flowId);
// final.status === 'completed'

Types

import type {
  FlowHandle,
  FlowResumeOptions,
  FlowRunOptions,
  FlowScope,
  FlowResult,
  FlowSnapshot,
  FlowSummary,
  ListFlowsOptions,
  SuspendOptions,
  StepOptions,
} from "@use-crux/core/flow";

import {
  FlowSuspendedError,
  FlowCancelledError,
  FlowExpiredError,
} from "@use-crux/core/flow";

Related

• Guide: Flows overview
• Guide: Suspend and resume
• Guide: Typed input and step composition
• Guide: Agent primitives in flows
• Guide: Convex
• Cookbook: Plan with approval
• Cookbook: Long-running flow