Flows

Some AI work cannot finish inside one Convex action: a plan needs human approval, an external system must call back, or a pipeline outlives a single worker. flow() from @use-crux/convex/server makes that work durable: a flow can cross Convex action boundaries, suspend at a named point, receive a signal, and resume where it left off.

import { action, flow } from "@use-crux/convex/server";

Define A Flow

"use node";

import { action, flow } from "@use-crux/convex/server";
import { internal } from "../_generated/api";
import { v } from "convex/values";
import { z } from "zod";

export const writerFlow = flow({
  name: "writer",
  args: {
    draftId: v.id("drafts"),
    brief: v.string(),
  },
  signals: {
    approval: z.object({ approved: z.boolean() }),
  },
  handler: async (scope, args, ctx) => {
    const plan = await scope.step("plan", () =>
      ctx.crux.runAction("plan draft", internal.writer.plan, {
        draftId: args.draftId,
        brief: args.brief,
      }),
    );

    const approval = await scope.suspend("approval");
    if (!approval.approved) return { status: "rejected" as const };

    return scope.step("write", () =>
      ctx.crux.runAction("write draft", internal.writer.write, {
        draftId: args.draftId,
        plan,
      }),
    );
  },
});

The Flow Handle

flow() returns a handle with everything needed to start, wrap, and signal the flow:

Public Wrapper

Do not expose .action publicly when you need auth or tenant policy. Wrap .handler:

export const startWriter = action({
  args: writerFlow.args,
  handler: async (ctx, args) => {
    await requireDraftAccess(ctx, args.draftId);
    return writerFlow.handler(ctx, args);
  },
});

Export .action for internal start/resume when policy is already enforced elsewhere:

export const writer = writerFlow.action;

Signal And Resume

When approval arrives, signal the flow and schedule the resume action:

export const approvePlan = action({
  args: {
    flowId: v.string(),
    approved: v.boolean(),
  },
  handler: async (ctx, args) => {
    await writerFlow.signal(
      ctx,
      internal.workflows.writerFlow.writer,
      args.flowId,
      "approval",
      {
        approved: args.approved,
      },
    );
  },
});

If the flow declares signals, .signal() uses the same core schema contract as flow.suspend(): invalid payloads throw before the pending signal is written or the resume action is scheduled.

If a mutation-only module cannot import the flow handle or action reference safely, keep a tiny app-local helper that calls signalFlow() and ctx.scheduler.runAfter().

Immediate Compositions

Immediate compositions such as parallel(), consensus(), and short swarm() runs can execute inside a Crux-aware Convex action or flow step. They preserve the active Crux context. They are not durable by themselves.

Use flow() when you need lifecycle helpers. Do not hide durability inside an immediate composition API.

Best Practices

• Use flow for long-running approval, resume, or external-signal workflows.
• Use readable step names: plan, approval, write, not function IDs.
• Use ctx.crux.runAction() for child steps that cross worker boundaries.
• Wrap .handler in public actions for auth and tenant checks.
• Keep application lifecycle status in app metadata; Crux flow state models execution, not your product workflow status.