Flows

import {
  flow,
  withFlow,
  signalFlow,
  cancelFlow,
  listFlows,
  createFlowId,
  FlowSuspendedError,
  FlowCancelledError,
  FlowExpiredError,
} from '@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 CruxStore 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 { resume: flowId } to .run() 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.

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

import { flow } from '@crux/core/flow'
import { generate } from '@crux/ai'

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

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

withFlow(name, fn, options?)

Execute a function directly within a flow scope. Lower-level alternative to flow -- useful for one-off flows or when you need full control over options at the call site.

Returns: Promise<FlowResult<T>>

FlowHandle<T, TInput>

The frozen handle returned by flow().

FlowRunOptions<TInput>

Options passed to handle.run().

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.

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. On each resume, re-evaluates the condition. If true, continues; if false, re-suspends.

SuspendOptions<T>

StepOptions

Retry and fallback options for a flow step.

FlowResult<T>

Discriminated union on status:

FlowSnapshot

Persisted flow state stored in CruxStore 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'.

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 withFlow() and converted to FlowResult variants. You should not normally need to catch them.

Suspend/Resume Example

import { flow, signalFlow } from '@crux/core/flow'
import { z } from 'zod'

const approvalFlow = flow('content-approval', async (flow) => {
  const draft = await flow.step('draft', () => generate(writer, { model, input: flow.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({ input: { 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.run({ resume: result.flowId })
// final.status === 'completed'

Types

import type {
  FlowHandle,
  FlowRunOptions,
  FlowScope,
  FlowResult,
  FlowSnapshot,
  FlowSummary,
  ListFlowsOptions,
  SuspendOptions,
  StepOptions,
  WithFlowOptions,
} from '@crux/core/flow'

import { FlowSuspendedError, FlowCancelledError, FlowExpiredError } from '@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