Workflow Agents

Build durable multi-step agents with the workflow builder, useWorkflow(), and workflows.signal()

A workflow is Gencow's durable execution primitive for long-running AI jobs, human approval checkpoints, and multi-step work that must resume safely after a retry, sleep, or restart. New workflows use the fluent builder exported from your app's gencow/runtime.ts and are explicitly registered with defineApi().

When to use the workflow builder

Use a workflow when:

• the job can exceed normal mutation timeout
• the work must survive crash, sleep, or restart
• a human approval or external event should resume the flow later
• some steps are expensive and should not rerun after partial success
• you want operational visibility in React, CLI, and Dashboard Jobs

Use a normal procedure.mutation when the whole job is short and synchronous.

Set up the app-owned builder

Bind the workflow builder once in gencow/runtime.ts. This gives the workflow the same schema and optional user typing as the rest of the app.

// gencow/runtime.ts
import type { GencowSchema } from "./generated/db-schema.gen";
import {
    createGencowProcedureBuilders,
    createGencowWorkflowBuilders,
} from "@gencow/core";

export const procedure = createGencowProcedureBuilders<GencowSchema>();
export const workflow = createGencowWorkflowBuilders<GencowSchema>();

If your app has a custom user identity type, pass it as the second generic to both builders. Import workflow from ./runtime in application code; do not create a separate gencow/workflow.ts.

Scaffold the starter

gencow add Agent

This adds:

• gencow/agent.ts with createWorkflowAgent({ model }) and runAgent
• gencow/ai.ts when AI is not installed yet
• an explicit workflows: { runAgent } entry in gencow/index.ts

Register every new workflow explicitly:

// gencow/index.ts
import { defineApi } from "@gencow/core";
import { runAgent } from "./agent";

export default defineApi({
    workflows: { runAgent },
});

For each registered definition, codegen exposes one public start mutation such as api.agents.run. The internal resume action is scheduler-only and is never part of the client API.

Run gencow codegen after adding the registration so your frontend can use api.agents.run. The examples below assume your React tree is wrapped in `GencowProvider`.

Note — registration and names: Importing a fluent definition does not register it. Add it to the existing defineApi() export alongside your other entries. .name("agents.run") determines the generated workflow start api.agents.run; changing the workflows bag key does not rename that API.

Build a workflow

import { v } from "@gencow/core";
import { ai } from "./ai";
import { workflow } from "./runtime";

export const runAgent = workflow
    .name("agents.run")
    .input(v.object({
        goal: v.string(),
        requireApproval: v.optional(v.boolean()),
    }))
    .maxDuration("30m")
    .retries(3)
    .handler(async (wf, args) => {
        const plan = await wf.step("plan", async () => {
            const result = await ai.chat({
                model: "llm/economy",
                system: "Return a short actionable plan.",
                messages: [{ role: "user", content: args.goal }],
            });
            return result.text;
        });

        const [research, critique] = await wf.parallel([
            () => wf.step("research", async () => {
                const result = await ai.chat({
                    model: "llm/economy",
                    system: "Gather supporting facts and constraints.",
                    messages: [{ role: "user", content: `Goal: ${args.goal}\n\nPlan:\n${plan}` }],
                });
                return result.text;
            }),
            () => wf.step("critique", async () => {
                const result = await ai.chat({
                    model: "llm/economy",
                    system: "Find gaps and risks.",
                    messages: [{ role: "user", content: `Goal: ${args.goal}\n\nPlan:\n${plan}` }],
                });
                return result.text;
            }),
        ] as const);

        if (args.requireApproval) {
            const approval = await wf.waitForEvent<{ approved: boolean; note?: string }>("approval", "24h");
            if (approval.approved !== true) {
                return { outcome: "declined", plan, research, critique, final: null };
            }
        }

        const final = await wf.step("finalize", async () => {
            const result = await ai.chat({
                model: "llm/economy",
                system: "Synthesize research and critique.",
                messages: [{
                    role: "user",
                    content: `Goal: ${args.goal}\n\nPlan:\n${plan}\n\nResearch:\n${research}\n\nCritique:\n${critique}`,
                }],
            });
            return result.text;
        });

        return { outcome: "completed", plan, research, critique, final };
    });

.name(...) and .handler(...) are required; .handler(...) is terminal. Before it, you can compose .use(...), .input(...), .allowAnonymous(), .maxDuration(...), .lifecycleTimeout(...), .concurrency(...), and .retries(...).

The plan, research, critique, and finalize checkpoints make progress reusable. For example, if finalization fails, a retry can reuse the saved research rather than paying for the same AI calls again. A declined approval returns a business outcome without running finalization.

Advanced — parallel checkpoints: Use distinct, stable wf.step() names inside each parallel branch, as above. This makes branch results durable across supported workflow runtimes; wf.parallel() alone is not a portable checkpoint guarantee. Keep branches independent, and place sleeps or human approval waits in the sequential handler after the group.

Protect and configure workflow starts

.use(...) is start middleware. Use it for authorization, tenancy checks, rate limits, tracing, and safe input normalization when a run is admitted. It does not wrap handler execution, individual steps, or resumes; rerunning request middleware during recovery could duplicate side effects.

export const runTenantAgent = workflow
    .name("agents.tenantRun")
    .input(v.object({ goal: v.string() }))
    .use(async ({ context, input, next }) => {
        await assertTenantMayStartAgent(context.auth, input.goal);
        return next();
    })
    .lifecycleTimeout("24h")
    .concurrency(5)
    .handler(async (wf, input) => {
        return wf.step("run", () => runForTenant(input.goal));
    });

Input validation and middleware run in the order they are declared. The admitted input and identity are durable; put execution-wide policy inside a handler or named step instead of middleware.

assertTenantMayStartAgent() and runForTenant() above are app-owned helpers: implement them with your tenant membership and business rules.

Note — three different deadlines: .maxDuration("10m") does not give a reviewer only ten minutes to respond. Use wf.waitForEvent("approval", "24h") for that checkpoint's wait limit, and .lifecycleTimeout("48h") if the whole run must finish within two days. A durable pause releases the worker rather than holding a request open.

Advanced — middleware order: Declare .input(...) before middleware that needs validated, typed input. Middleware placed before .input(...) sees the earlier input shape. Put resource checks that must reflect current permissions at execution time inside the relevant step as well.

Mental model and runtime primitives

A workflow is not a background thread. Its handler re-enters from the top, while completed checkpoints are memoized.

start mutation
  -> _gencow_workflows row created
  -> scheduler queues resume action
  -> runner claims the row
  -> handler executes
  -> completed wf.step() outputs are memoized in _gencow_workflow_steps
  -> retry/sleep/wait resumes the same workflow later

• wf.step(name, fn) — memoized checkpoint
• wf.parallel([...]) — run independent branches concurrently; checkpoint each branch with wf.step()
• wf.waitForEvent(name, timeout?) — wait for an external signal with optional timeout
• wf.sleep(duration) — pause and resume without consuming retries

Authoring rules

• Keep step names stable across retries and deploys. wf.step("plan") is safe; wf.step(\plan-${Date.now()}`)` is not.
• Return only JSON-serializable values from inputs, step outputs, signal payloads, and final results.
• Durable payloads are bounded. Store large arrays, embeddings, documents, media, and provider responses in app tables or storage, then pass only IDs, counts, checksums, or status summaries.
• Put side effects inside named steps and make them idempotent when possible.
• Use wf.waitForEvent(..., timeout) when the business process needs a deadline.
• Keep normal synchronous work in procedure.mutation; use a workflow when resume semantics matter.

Advanced — retries and external side effects: A process can stop after an external service accepts a request but before the step result is saved. A retry may execute that request again. Use a stable idempotency key derived from wf.workflowId and the step name when the external API supports it; checkpointing alone does not make an external write exactly once. Pass wf.signal to requests that support cancellation (signal for fetch, abortSignal for AI SDK generateText) and call wf.throwIfAborted() between long operations so cancellation can stop work promptly.

Track from React

import { useState } from "react";
import { api } from "../gencow/api";
import { useMutation, useWorkflow } from "@gencow/react";

export function AgentRunner() {
    const [runId, setRunId] = useState<string | null>(null);
    const { mutate: start, isPending, error } = useMutation(api.agents.run);
    const state = useWorkflow(api.workflows.get, runId);

    async function handleStart() {
        try {
            const run = await start({ goal: "Launch brief", requireApproval: true });
            setRunId(run.id);
        } catch {
            // useMutation exposes the failed start through `error` below.
        }
    }

    return (
        <section>
            <button onClick={handleStart} disabled={isPending || state.isActive}>
                Start agent
            </button>
            <p>{state.derivedStatus ?? "Ready"}</p>
            {state.currentStep && <p>Step: {state.currentStep}</p>}
            {error && <p role="alert">{error.message}</p>}
            {state.error && <p role="alert">{state.error.message}</p>}
            {state.data?.error && <p role="alert">{state.data.error}</p>}
            {state.isTerminal && <pre>{JSON.stringify(state.data?.result, null, 2)}</pre>}
        </section>
    );
}

useWorkflow() prefers exact-ID realtime updates and falls back to safe polling while the run is active. A start returns WorkflowStartResult; the handler result is available in the durable workflow state, not the start response.

Useful snapshot fields:

• status — canonical engine status (pending, running, completed, failed)
• derivedStatus — display state (queued, waiting, sleeping, running, completed, failed)
• currentStep — current checkpoint name
• steps — step timeline with output and error snapshots

Note — accepted is not completed: start() returns a run ID after admission, not the final AI result. isPending describes the start request; state.isActive describes the durable run. Read state.data?.result after completion, and use state.data?.error to inspect a workflow execution failure. state.error instead describes a failure to fetch the snapshot.

Approval and external events

Use wf.waitForEvent() for human approval, webhook continuation, or another out-of-band signal.

const approval = await wf.waitForEvent<{ approved: boolean; note?: string }>(
    "approval",
    "30m",
);
if (approval.approved !== true) {
    return { outcome: "declined" };
}

Send the matching event from an authorized React component:

export function ApprovalButtons({ runId }: { runId: string }) {
    const { mutate: signal, isPending, error } = useMutation(api.workflows.signal);

    async function decide(approved: boolean) {
        try {
            await signal({
                id: runId,
                event: "approval",
                payload: { approved },
            });
        } catch {
            // Render the signal request error below.
        }
    }

    return (
        <div>
            <button disabled={isPending} onClick={() => decide(true)}>Approve</button>
            <button disabled={isPending} onClick={() => decide(false)}>Decline</button>
            {error && <p role="alert">{error.message}</p>}
        </div>
    );
}

This component uses the useMutation and api imports from the preceding example. Render it when your UI offers approval for the selected run. If the wait timeout elapses first, the workflow fails terminally without consuming retry backoff.

Note — signal payloads: The type parameter on waitForEvent<T>() helps TypeScript inside the handler; it does not validate an incoming signal at runtime. Validate business-sensitive fields before acting on them. The example requires approved === true before proceeding, and the event name must match "approval" exactly.

Ownership and operations

Workflow runtime rows live in platform-owned _gencow_workflows, _gencow_workflow_steps, and _gencow_workflow_events tables. Do not declare them in schema.ts, alter them in app migrations, or expose an unfiltered workflow list. Prefer useWorkflow(api.workflows.get, run.id); custom list queries must require auth and scope rows by user_id.

Inspect runs in React, gencow jobs, or the Cloud Dashboard Jobs tab.

• Retry continues the same run and preserves completed checkpoints.
• Replay creates a fresh run from the original input.
• Cancel is best effort: it marks the run terminal and removes queued work when possible. Running work stops cooperatively when it observes the abort signal; cancellation cannot undo a completed external write.
• Delete removes a terminal run and its stored snapshots.

In Gencow Cloud, the DB-backed scheduler survives app sleep and restart. Local development uses an in-memory scheduler, so restart durability is best effort.

Deprecated: legacy workflow() API

workflow(name, options) imported from @gencow/core remains available only for existing applications during the migration window. It globally registers a legacy mutation wire, so do not use it in new code, templates, or examples.

Existing definitions can remain temporarily and can be migrated incrementally:

// Legacy only — deprecated.
import { workflow, v } from "@gencow/core";

export const runImport = workflow("imports.run", {
    args: { sourceId: v.string() },
    handler: async (wf, args) => {
        return wf.step("import", () => importSource(args.sourceId));
    },
});

Move it to workflow.name("imports.run").input(...).handler(...) using the app-owned builder, then export the new definition through defineApi({ workflows: { runImport } }).

Keep .name("imports.run") when replacing the definition so the generated start remains api.imports.run. Add the new definition to workflows, remove the old constructor for that name, and run gencow codegen.

Note — incremental migration: Other legacy workflows can remain while you migrate one definition. Omitted legacy definitions stay on their legacy wire, even with workflows: {}. Avoid constructing a legacy workflow and a new fluent workflow with the same name in the same app.