> ## Documentation Index
> Fetch the complete documentation index at: https://claritylabs.inc/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Connecting CL Pipelines to Convex Storage and Scheduler

> Step-by-step guide to connecting cl-pipelines to Convex using pipelineFields, createConvexStorageAdapter, and createConvexSchedulerAdapter.

The `@claritylabs/cl-pipelines/convex` sub-path export ships pre-built adapters for Convex's storage and scheduler APIs. You define the five pipeline mutations in your Convex backend, then pass them to `createConvexStorageAdapter` — the adapters handle serialization, scheduling, and the `advancePhase` wiring automatically.

## Installation

```bash theme={"system"}
npm install @claritylabs/cl-pipelines zod ai
```

Import the Convex-specific helpers from the sub-path:

```typescript theme={"system"}
import {
  pipelineFields,
  createConvexStorageAdapter,
  createConvexSchedulerAdapter,
} from "@claritylabs/cl-pipelines/convex";
```

## Setup

### Add pipeline fields to your schema

Use `pipelineFields()` to add the four required columns to any Convex table. Spread it inside `defineTable`:

```typescript theme={"system"}
// convex/schema.ts
import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";
import { pipelineFields } from "@claritylabs/cl-pipelines/convex";

export default defineSchema({
  applications: defineTable({
    applicantName: v.string(),
    submittedAt: v.number(),
    ...pipelineFields(),
    // Adds: pipelineStatus, pipelineError, pipelineCheckpoint, pipelineLog
  }),
});
```

### Implement the five pipeline mutations

Create an internal mutation file (e.g. `convex/applications.ts`) with the five methods your `StorageAdapter` needs. Each mutation maps to one `StorageAdapter` method.

```typescript getJob theme={"system"}
export const getJob = internalQuery({
  args: { jobId: v.string() },
  handler: async (ctx, { jobId }) => {
    const doc = await ctx.db.get(jobId as any);
    if (!doc) return null;
    return {
      status: doc.pipelineStatus,
      checkpoint: doc.pipelineCheckpoint ?? null,
      error: doc.pipelineError,
    };
  },
});
```

```typescript setStatus theme={"system"}
export const setStatus = internalMutation({
  args: {
    jobId: v.string(),
    status: v.union(
      v.literal("idle"),
      v.literal("running"),
      v.literal("paused"),
      v.literal("complete"),
      v.literal("error")
    ),
    error: v.union(v.string(), v.null()),
  },
  handler: async (ctx, { jobId, status, error }) => {
    await ctx.db.patch(jobId as any, {
      pipelineStatus: status,
      pipelineError: error ?? undefined, // clears on null — REQUIRED
    });
  },
});
```

```typescript setCheckpoint theme={"system"}
export const setCheckpoint = internalMutation({
  args: { jobId: v.string(), checkpoint: v.union(v.any(), v.null()) },
  handler: async (ctx, { jobId, checkpoint }) => {
    await ctx.db.patch(jobId as any, {
      pipelineCheckpoint: checkpoint ?? undefined,
    });
  },
});
```

```typescript appendLog theme={"system"}
export const appendLog = internalMutation({
  args: {
    jobId: v.string(),
    timestamp: v.number(),
    message: v.string(),
    phase: v.optional(v.string()),
    level: v.optional(v.string()),
  },
  handler: async (ctx, { jobId, timestamp, message, phase, level }) => {
    const doc = await ctx.db.get(jobId as any);
    if (!doc) return;
    const log = doc.pipelineLog ?? [];
    await ctx.db.patch(jobId as any, {
      pipelineLog: [...log, { timestamp, message, phase, level }],
    });
  },
});
```

```typescript clearLog theme={"system"}
export const clearLog = internalMutation({
  args: { jobId: v.string() },
  handler: async (ctx, { jobId }) => {
    await ctx.db.patch(jobId as any, { pipelineLog: [] });
  },
});
```

> [!WARNING]
>
> The `setStatus` mutation **must** always patch `pipelineError: error ?? undefined`. When `error` is `null`, you must clear the field. If you skip this patch, stale error messages will survive retries and appear in your UI even after a successful run.

### Create the advance action

Create an `internalAction` that constructs both adapters and calls `advancePhase`. This is the function Convex's scheduler will call after each phase:

```typescript theme={"system"}
// convex/applications.ts
import { internalAction } from "./_generated/server";
import { internal } from "./_generated/api";
import { v } from "convex/values";
import { advancePhase } from "@claritylabs/cl-pipelines";
import { createConvexStorageAdapter, createConvexSchedulerAdapter } from "@claritylabs/cl-pipelines/convex";
import type { AgentCheckpoint } from "@claritylabs/cl-pipelines";
import { phases } from "./pipeline"; // your phase definitions

export const advance = internalAction({
  args: { jobId: v.string() },
  handler: async (ctx, { jobId }) => {
    const mutations = {
      getJob: internal.applications.getJob,
      setStatus: internal.applications.setStatus,
      setCheckpoint: internal.applications.setCheckpoint,
      appendLog: internal.applications.appendLog,
      clearLog: internal.applications.clearLog,
    };

    const storage = createConvexStorageAdapter<AgentCheckpoint>({ ctx, mutations });
    const scheduler = createConvexSchedulerAdapter({
      ctx,
      advanceAction: internal.applications.advance,
    });

    await advancePhase({ jobId, phases, storage, scheduler });
  },
});
```

### Start a job from a public action

Call `runPipeline` from a Convex `action` (not a mutation — `runPipeline` calls `scheduleAdvance`, which requires action context):

```typescript theme={"system"}
import { action } from "./_generated/server";
import { internal } from "./_generated/api";
import { v } from "convex/values";
import { runPipeline } from "@claritylabs/cl-pipelines";
import { createConvexStorageAdapter, createConvexSchedulerAdapter } from "@claritylabs/cl-pipelines/convex";
import type { AgentCheckpoint } from "@claritylabs/cl-pipelines";

export const startJob = action({
  args: { jobId: v.string(), userMessage: v.string() },
  handler: async (ctx, { jobId, userMessage }) => {
    const mutations = {
      getJob: internal.applications.getJob,
      setStatus: internal.applications.setStatus,
      setCheckpoint: internal.applications.setCheckpoint,
      appendLog: internal.applications.appendLog,
      clearLog: internal.applications.clearLog,
    };

    const storage = createConvexStorageAdapter<AgentCheckpoint>({ ctx, mutations });
    const scheduler = createConvexSchedulerAdapter({
      ctx,
      advanceAction: internal.applications.advance,
    });

    await runPipeline<AgentCheckpoint>({
      jobId,
      phases,
      storage,
      scheduler,
      initialState: {
        messages: [{ role: "user", content: userMessage }],
        pendingToolCalls: [],
        turn: 0,
      },
    });
  },
});
```

### Add a retry action

Expose a retry action that accepts a `mode` argument so your UI can trigger both `"resume"` and `"full"` retries:

```typescript theme={"system"}
export const retryJob = action({
  args: {
    jobId: v.string(),
    mode: v.union(v.literal("resume"), v.literal("full")),
    userMessage: v.string(),
  },
  handler: async (ctx, args) => {
    const mutations = {
      getJob: internal.applications.getJob,
      setStatus: internal.applications.setStatus,
      setCheckpoint: internal.applications.setCheckpoint,
      appendLog: internal.applications.appendLog,
      clearLog: internal.applications.clearLog,
    };

    const storage = createConvexStorageAdapter<AgentCheckpoint>({ ctx, mutations });
    const scheduler = createConvexSchedulerAdapter({
      ctx,
      advanceAction: internal.applications.advance,
    });

    await runPipeline<AgentCheckpoint>({
      jobId: args.jobId,
      phases,
      storage,
      scheduler,
      retryMode: args.mode,
      initialState: {
        messages: [{ role: "user", content: args.userMessage }],
        pendingToolCalls: [],
        turn: 0,
      },
    });
  },
});
```

## createConvexStorageAdapter

```typescript theme={"system"}
createConvexStorageAdapter<TState>({
  ctx: ActionCtx,
  mutations: ConvexPipelineMutations,
}): StorageAdapter<TState>
```

**ctx** `ActionCtx` *required*

The Convex action context (`ctx`) from your `internalAction` or `action` handler.

**mutations** `ConvexPipelineMutations` *required*

A record mapping the five pipeline mutation names to their `internal.*` references. Import `ConvexPipelineMutations` type from `@claritylabs/cl-pipelines/convex` for type-safe wiring.

## createConvexSchedulerAdapter

```typescript theme={"system"}
createConvexSchedulerAdapter({
  ctx: ActionCtx,
  advanceAction: FunctionReference<"action">,
  jobIdArgName?: string,  // default: "jobId"
}): SchedulerAdapter
```

**ctx** `ActionCtx` *required*

The Convex action context.

**advanceAction** `FunctionReference\<'action'\>` *required*

The `internal.*` reference to your `advance` action (e.g. `internal.applications.advance`).

**jobIdArgName** `string`

The argument name your advance action uses for the job ID. Defaults to `"jobId"`.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.