Flows

A single generate() call handles one prompt. Real applications chain multiple calls into pipelines: research then synthesize, plan then execute, draft then validate. flow makes those pipelines a first-class primitive: every step has a name, every retry is declarative, and every trace is grouped automatically.

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

const researchFlow = flow("research", async (flow) => {
  const plan = await flow.step("plan", () =>
    generate(planner, { model, input: { query } }),
  );
  const results = await flow.step("search", () =>
    generate(searcher, { model, input: { plan: plan.object } }),
  );
  return flow.step("synthesize", () =>
    generate(synthesizer, { model, input: { results: results.object } }),
  );
});

await researchFlow.run();

When You Need This

Use flow when you see these signals in your code:

• Chaining outputs: the result of one generate() feeds into the next as input
• Manual try/catch for retries: you're wrapping generate() in retry loops yourself
• Human-in-the-loop gates: your pipeline needs to pause for approval, review, or external input before continuing
• Cross-action tracing: your pipeline spans multiple Convex actions, Lambda invocations, or API routes
• Eval↔runtime comparison: you want to compare production step performance against eval baselines in devtools

If you're making a single generate() call, you don't need a flow, just call generate() directly. Flows are for multi-step pipelines where the steps have meaningful names and data dependencies.

What Flows Give You

• Named steps: label each stage of your pipeline; thread data between steps naturally
• Automatic retries: retry failed steps with configurable backoff, no manual try/catch
• Fallbacks: switch to a cheaper model or alternative strategy when retries are exhausted
• Suspend/resume: pause for human approval, external events, or async dependencies; resume later, even in a different process
• Structured tracing: every generate() inside a step is tagged with flowId, stepId, and stepLabel automatically
• Devtools grouping: the Timeline groups traces by flow and step instead of guessing from timing
• Eval↔runtime comparison: production tasks and Eval Cases share canonical run and trace identities
• Cross-boundary: works across Convex actions, microservices, or any multi-process setup by passing IDs
• Define once, run many: flow() returns a frozen handle; call .run() repeatedly with different inputs

Basic Usage

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

const editPipeline = flow("edit-pipeline", async (flow) => {
  const analysis = await flow.step("analyze", () =>
    generate(analyzer, { model, input: { document } }),
  );
  return flow.step("edit", () =>
    generate(editor, { model, input: { analysis: analysis.object, document } }),
  );
});

const result = await editPipeline.run();

flow returns a FlowHandle with a .run() method that returns whatever the handler returns (wrapped in a FlowResult). Each flow.step() returns the result of its function: use it to thread data between steps. Step labels are durable replay identities for the Flows beta, so keep each label unique and stable. Dynamic labels are advanced; use them only when the generated label is deterministic across retries and resumes.

Sessions

withSession groups traces by user interaction. flow structures them into pipelines within that session.

import { withSession, flow, createSessionId } from "@use-crux/core";

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

await withSession(createSessionId(), () => researchFlow.run());

Sessions and flows compose in any order. Multiple flows can run within one session:

await withSession(sessionId, async () => {
  await researchFlow.run();
  await writeFlow.run();
});

Retry and Fallback

LLM calls fail: rate limits, timeouts, malformed output. Instead of wrapping every step in try/catch, declare retry and fallback behavior:

await flow.step(
  "search",
  () => generate(searcher, { model: primaryModel, input }),
  {
    retry: { attempts: 3, delay: 1000, backoff: "exponential" },
    fallback: () => generate(searcher, { model: cheapModel, input }),
  },
);

This retries up to 3 times with exponential backoff (1s, 2s delays), then falls back to a cheaper model if all attempts fail. The step still returns its result transparently: callers don't need to know about the retry logic.

You can use retry alone, fallback alone, or both together. Without either, errors propagate normally.

Crux policy-terminal errors are not retried by default and do not run execution fallback: GuardrailBlockedError, ConstraintViolationError, and ValidationExhaustedError. This keeps safety, constraint, and validation decisions from being bypassed by a generic fallback. Use shouldRetry only when you intentionally want to override that policy.

// Retry only, fail hard after 3 attempts
await flow.step("validate", () => generate(validator, { model, input }), {
  retry: { attempts: 3, delay: 500 },
});

// Fallback only, no retry, just switch strategy on first failure
await flow.step(
  "summarize",
  () => generate(summarizer, { model: premiumModel, input }),
  {
    fallback: () => generate(summarizer, { model: cheapModel, input }),
  },
);

Parallel Flows

Run independent pipelines concurrently:

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

await withSession(sessionId, () =>
  Promise.all(queries.map((q) => topicResearch.run({ query: q }))),
);

Each .run() call gets its own flowId and groups independently in devtools.

Testing flow-backed model behavior

Keep focused flow control tests beside the flow. For model behavior, expose the production generate.task() or stream.task() used by a flow step and point an Eval at that managed task. This preserves the task identity and host contract required for exact reuse. See the Evals guide.

Signaling Handles

Prefer flowHandle.signal(flowId, name, payload) on the flow handle. The bare signalFlow() export is a low-level primitive that only writes the signal payload; see Suspend and resume for the full picture.

Where to Next