# Workflow Engine

> Guide for @bratsos/workflow-engine - a type-safe workflow engine with AI integration, stage pipelines, and persistence. Use when building multi-stage workflows, AI-powered pipelines, implementing workflow persistence, defining stages, or working with batch AI operations. For upgrades between published versions, route through `migrations/README.md`.

- Skill: `bratsos/workflow-engine` (Agent Skill, multi-file: 25 files)
- Install (CLI): `npx skillmds@latest add bratsos/workflow-engine`
- Raw SKILL.md: https://api.skillmd.com/api/skills/bratsos/workflow-engine/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- License: MIT
- Author: bratsos (https://skillmd.com/u/bratsos)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/bratsos/workflow-engine

---


## Upgrading between versions

When the user wants to upgrade `@bratsos/workflow-engine` in their project (e.g., "upgrade my project to the latest workflow-engine version", "I just bumped workflow-engine, walk me through the migration"), route to `migrations/README.md` — the upgrade router. It explains how to detect the installed and previous versions, find the relevant migration guides, and apply them in order. Multi-version upgrades (e.g., 0.6 → 0.8) load and apply multiple migration files sequentially.

# @bratsos/workflow-engine Skill

Type-safe workflow engine for building AI-powered, multi-stage pipelines with persistence and batch processing support. Uses a **command kernel** architecture with environment-agnostic design.

## Architecture Overview

The engine follows a **kernel + host** pattern:

- **Core library** (`@bratsos/workflow-engine`) - Command kernel, stage/workflow definitions, persistence adapters
- **Node Host** (`@bratsos/workflow-engine-host-node`) - Long-running worker with polling loops and signal handling
- **Serverless Host** (`@bratsos/workflow-engine-host-serverless`) - Stateless single-invocation for edge/lambda/workers
- **Remote Host** (`@bratsos/workflow-engine-host-remote`) - Credential-free remote activity workers: run a stage's `execute()` on a separate, disposable machine (no DB connection, no root object-store credentials)

The **kernel** is a pure command dispatcher. All workflow operations are expressed as typed commands dispatched via `kernel.dispatch()`. Hosts wrap the kernel with environment-specific process management.

## When to Apply

- User wants to create workflow stages or pipelines
- User mentions `defineStage`, `defineWorkflow`, `WorkflowBuilder`, `ctx.step`
- User is implementing workflow persistence with Prisma
- User needs AI integration (generateText, generateObject, embeddings, batch)
- User is building multi-stage data processing pipelines
- User mentions kernel, command dispatch, or job execution
- User wants to set up a Node.js worker or serverless worker
- User wants to run a stage on a separate / remote / disposable machine, or mentions credential-free workers, `defineRemoteStage`, the `ActivityExecutor` port, or offloading heavy stages (transcoding, ffmpeg, batch inference)
- User wants to rerun a workflow from a specific stage
- User needs to test workflows with in-memory adapters

## Quick Start

```typescript
import { defineStage, defineWorkflow } from "@bratsos/workflow-engine";
import { createKernel, createWorkflowRegistry } from "@bratsos/workflow-engine/kernel";
import { createNodeHost } from "@bratsos/workflow-engine-host-node";
import {
  createPrismaWorkflowPersistence,
  createPrismaJobQueue,
  createPrismaStepLedger,
  createPrismaAICallLogger,
} from "@bratsos/workflow-engine";
import { z } from "zod";

// 1. Define a stage
const processStage = defineStage({
  id: "process",
  name: "Process Data",
  schemas: {
    input: z.object({ data: z.string() }),
    output: z.object({ result: z.string() }),
    config: z.object({ verbose: z.boolean().default(false) }),
  },
  async execute(ctx) {
    return { output: { result: ctx.input.data.toUpperCase() } };
  },
});

// 2. Build a workflow
const workflow = defineWorkflow({
  id: "my-workflow",
  name: "My Workflow",
  description: "Processes data",
  input: z.object({ data: z.string() }),
})
  .pipe(processStage)
  .build();

// 3. Create kernel
const kernel = createKernel({
  persistence: createPrismaWorkflowPersistence(prisma),
  blobStore: myBlobStore,
  jobTransport: createPrismaJobQueue(prisma),
  eventSink: myEventSink,
  clock: { now: () => new Date() },
  registry: createWorkflowRegistry([workflow]), // enumerable, so definition-version filtering works (see 13-definition-versioning.md)
  stepLedger: createPrismaStepLedger(prisma), // needed for ctx.step.* (durable steps)
  services: { aiLogger: createPrismaAICallLogger(prisma) }, // ctx.ai / ctx.step.ai
});

// 4. Start a Node host
const host = createNodeHost({
  kernel,
  jobTransport: createPrismaJobQueue(prisma),
  workerId: "worker-1",
});
await host.start();

// 5. Dispatch a command
await kernel.dispatch({
  type: "run.create",
  idempotencyKey: crypto.randomUUID(),
  workflowId: "my-workflow",
  input: { data: "hello" },
});
```

## Core Exports Reference

| Export | Type | Import Path | Purpose |
|--------|------|-------------|---------|
| `defineStage` | Function | `@bratsos/workflow-engine` | Create sync stages. Curried form `defineStage<TContext>()({...})` is recommended when you need typed `ctx.require()`/`ctx.optional()` — see 01-stage-definitions.md |
| `createWorkflowRegistry` | Function | `@bratsos/workflow-engine/kernel` | Build an enumerable `WorkflowRegistry` from a list of workflows; required for definition-version filtering (a hand-written `{ getWorkflow }` cannot enumerate) |
| `defineWorkflow` | Function | `@bratsos/workflow-engine` | **Recommended** way to build a workflow (options-object API, v0.11+); returns a `WorkflowBuilder` to `.pipe()`/`.parallel()`/`.build()` |
| `WorkflowBuilder` | Class | `@bratsos/workflow-engine` | Chain stages into workflows: `.stage(id, def)` (typed `ctx.require`, dependency ids checked), `.stage(prebuilt)`, `.pipe()`, `.parallel([...])` / `.parallel((group) => ...)`, `.build()`. Create it with `defineWorkflow(id, options?)` or `defineWorkflow({...})` |
| `createKernel` | Function | `@bratsos/workflow-engine/kernel` | Create command kernel |
| `createNodeHost` | Function | `@bratsos/workflow-engine-host-node` | Create Node.js host |
| `createServerlessHost` | Function | `@bratsos/workflow-engine-host-serverless` | Create serverless host |
| `defineRemoteStage` / `createActivityWorker` | Function | `@bratsos/workflow-engine-host-remote` | Run a stage on a credential-free remote worker (see 11-remote-activity-workers.md) |
| `createRoutingExecutor` / `createLocalExecutor` | Function | `@bratsos/workflow-engine/kernel` | `ActivityExecutor` port: route specific stages to a remote executor / default in-process executor |
| `createAIHelper` | Function | `@bratsos/workflow-engine` | AI operations (text, object, embed, batch) with AI SDK & OpenRouter batch support |
| `registerEmbeddingProvider` | Function | `@bratsos/workflow-engine` | Register custom embedding providers (Voyage, Cohere, etc.) |
| `createStageIds` | Function | `@bratsos/workflow-engine` | Create stage ID constants from a workflow |
| `defineStageIds` | Function | `@bratsos/workflow-engine` | Define stage ID constants from a tuple |
| `isValidStageId` | Function | `@bratsos/workflow-engine` | Runtime stage ID validation |
| `assertValidStageId` | Function | `@bratsos/workflow-engine` | Assert stage ID validity (throws) |
| `definePlugin` | Function | `@bratsos/workflow-engine/kernel` | Define kernel plugins |
| `createPluginRunner` | Function | `@bratsos/workflow-engine/kernel` | Create plugin event processor |
| `typedKey` | Function | `@bratsos/workflow-engine/conventions` | Define a well-known annotation key with linked value type |
| `Trigger` / `Decision` / `Approval` / `Revision` | Constants | `@bratsos/workflow-engine/conventions` | Well-known annotation key namespaces (v0.8+) |
| `RunReapStuckCommand` / `RunReapStuckResult` | Types | `@bratsos/workflow-engine` | `run.reapStuck` command/result shapes (export-drift fix, v0.11+) |
| `ModelFilter` | Type | `@bratsos/workflow-engine` | Filter shape for `listModels({ filter })` (export-drift fix, v0.11+) |
| `persistenceConformanceSuite` / `jobQueueConformanceSuite` / `aiCallLoggerConformanceSuite` / `stepLedgerConformanceSuite` | Function | `@bratsos/workflow-engine/testing` | Vitest conformance suites for validating custom adapters (v0.11+; `stepLedgerConformanceSuite` v1.0) |

## Kernel Commands

All operations go through `kernel.dispatch(command)`:

| Command | Description |
|---------|-------------|
| `run.create` | Create a new workflow run |
| `run.claimPending` | Claim pending runs, enqueue first-stage jobs |
| `run.transition` | Advance to next stage group or complete |
| `run.cancel` | Cancel a running workflow (authoritative: cascades to stages + jobs) |
| `run.redrive` | Retry (`from: { kind: "lastFailure" }`, default), restart (`"start"`) or rerun from a stage (`"stage"`) on the same run id; optionally re-pin with `definitionVersion: "latest"`. The superseded attempt is archived under `run.supersededAttempt` (see 14-redrive.md) |
| `run.rerunFrom` | **Deprecated** — use `run.redrive`. Rerun from a specific stage; delegates to `run.redrive` (accepts an optional `idempotencyKey`) |
| `run.listVersions` | Per-definition-version run counts: has a version drained, and which versions have active runs nobody here serves (see 13-definition-versioning.md) |
| `run.purge` | Retention: delete terminal runs finished before `olderThan` (`{ olderThan, statuses?, limit? }` → `{ purged, workflowRunIds }`), clearing the step ledger, job rows and blobs with them |
| `job.execute` | Execute a single stage (uses multi-phase transactions; see 08-common-patterns.md). Takes the job's `attempt`/`maxAttempts` and an optional `abortSignal` |
| `job.heartbeat` | One beat of a host's job-lease heartbeat: renews the lease, reports `{ runStatus, leaseHeld }`, and is what aborts `ctx.abortSignal` |
| `stage.pollSuspended` | Poll suspended stages for readiness (claims each stage first; skips cancelled runs; per-stage transactions) |
| `step.signal` | Complete a durable `ctx.step.waitForSignal` step with a payload (idempotent) |
| `lease.reapStale` | Release stale job leases (`{ staleThresholdMs, absoluteTimeoutMs? }` → `{ released, expired }`) |
| `run.reapStuck` | Detect and fail RUNNING runs with no recent activity (heals wedged runs whose stages all finished) |
| `outbox.flush` | Publish pending outbox events; reports `eventSinkStatus: "healthy" \| "degraded"` |
| `plugin.replayDLQ` | Replay dead-letter queue events |

## Stage Definition

### Sync Stage

```typescript
const myStage = defineStage({
  id: "my-stage",
  name: "My Stage",
  description: "Optional",
  dependencies: ["prev"],

  schemas: {
    input: InputSchema,     // Zod schema or "none"
    output: OutputSchema,
    config: ConfigSchema,
  },

  async execute(ctx) {
    const { input, config, workflowContext } = ctx;
    const prevOutput = ctx.require("prev");
    const optOutput = ctx.optional("other");

    ctx.log("INFO", "Processing..."); // returns void

    // Durable side effects and AI calls: ctx.step.run / ctx.step.ai.* (see 12-durable-steps.md)
    const summary = await ctx.step.ai.generateText("summary", "gemini-2.5-flash", prompt);

    return {
      output: { ... },
      customMetrics: { itemsProcessed: 10 },
    };
  },
});
```

`ctx.step`, `ctx.ai`, `ctx.aiLogger` and `ctx.abortSignal` are always present on the context (1.0). `ctx.step.*` needs `createKernel({ stepLedger })`; `ctx.ai` needs `createKernel({ services: { aiLogger } })`.

### Async Batch Stage (legacy mode)

Prefer `ctx.step.ai.map(id, items, { policy: "batch" })` inside a plain `defineStage` for new code — the batch bookkeeping then lives in the step ledger (12-durable-steps.md). The `mode: "async-batch"` / `checkCompletion` shape still runs; `defineAsyncBatchStage` was removed from the root entry at 1.0, so write it as `defineStage({ mode: "async-batch", ... })`:

```typescript
const batchStage = defineStage({
  id: "batch-process",
  name: "Batch Process",
  mode: "async-batch",
  schemas: { input: "none", output: OutputSchema, config: ConfigSchema },

  async execute(ctx) {
    if (ctx.resumeState) {
      return { output: await ctx.storage.load("batch-result") };
    }

    const batch = ctx.ai.batch("gemini-2.5-flash", "google"); // ctx.ai is scoped to workflow.<runId>.stage.<stageId>
    const handle = await batch.submit(requests);

    return {
      suspended: true,
      state: {
        batchId: handle.id,
        submittedAt: new Date().toISOString(),
        pollInterval: 60000,
        maxWaitTime: 3600000,
        metadata: { batchRefs: handle.refs },
      },
      pollConfig: { pollInterval: 60000, maxWaitTime: 3600000, nextPollAt: new Date(Date.now() + 60000) },
    };
  },

  async checkCompletion(suspendedState, ctx) {
    const batch = ctx.ai.batch("gemini-2.5-flash", "google");
    const status = await batch.getStatus(suspendedState.batchId, suspendedState.metadata);
    if (status.status === "completed") {
      const results = await batch.getResults(suspendedState.batchId, suspendedState.metadata);
      return { ready: true, output: { results } };
    }
    if (status.status === "failed") return { ready: false, error: "Batch failed" };
    return { ready: false, nextCheckIn: 60000 };
  },
});
```

## WorkflowBuilder / defineWorkflow

Workflows are linear pipelines of **execution groups**. `.pipe()` creates single-stage groups; `.parallel()` creates multi-stage groups. Parallel group outputs are keyed by stage ID in the workflow context.

Build with `defineWorkflow(id, { input })` or `defineWorkflow({ id, name, input })`. `.stage(id, definition)` defines and adds a stage whose `ctx.require()` is typed from the stages before it and whose `dependencies` must name earlier stage ids; `.pipe(stage)` / `.stage(stage)` add a `defineStage()` result. See [references/12-durable-steps.md](references/12-durable-steps.md#the-builder).

```typescript
const workflow = defineWorkflow({
  id: "workflow-id",
  name: "Workflow Name",
  description: "Description",
  input: InputSchema,
  // no `output` option (removed at 1.0): the workflow output schema is always the last stage's
})
  .pipe(stage1)                          // Group 0
  .pipe(stage2)                          // Group 1
  .parallel([stage3a, stage3b])          // Group 2 (concurrent, output: { "stage3a-id": ..., "stage3b-id": ... })
  .pipe(stage4)                          // Group 3
  .build();

// In stage4, access parallel outputs by stage ID:
ctx.require("stage3a-id")  // output of stage3a
ctx.require("stage3b-id")  // output of stage3b

workflow.getStageIds();
workflow.getExecutionPlan();
workflow.getDefaultConfig();
workflow.validateConfig(config);
```

When a workflow completes, the final execution group's output is persisted in `WorkflowRun.output` and included in the `workflow:completed` event.

## Kernel Setup

```typescript
import { createKernel } from "@bratsos/workflow-engine/kernel";
import type { Kernel, KernelConfig, Persistence, BlobStore, JobTransport, EventSink, Clock } from "@bratsos/workflow-engine/kernel";

const kernel = createKernel({
  persistence,   // Persistence port - runs, stages, logs, outbox, idempotency
  blobStore,     // BlobStore port - large payload storage
  jobTransport,  // JobTransport port - job queue
  eventSink,     // EventSink port - async event publishing
  clock,         // Clock port - injectable time source
  registry,      // WorkflowRegistry - createWorkflowRegistry(workflows), or a hand-written { getWorkflow(id) } (no version filtering)
  // stepLedger,  // optional StepLedger port - required for ctx.step.* (InMemoryStepLedger / createPrismaStepLedger)
  // services: { aiLogger, ai? }, // optional - backs ctx.ai / ctx.aiLogger and the run's cost roll-up
  // executor,   // optional ActivityExecutor port - defaults to in-process; inject to run stages on remote workers (see 11-remote-activity-workers.md)
  // idempotencyStaleInProgressMs: 10 * 60 * 1000, // optional (v0.11+) - default 10 min; TTL before a stuck `in_progress` idempotency key can be reclaimed
  // spillThresholdBytes: 65_536, // optional (1.0) - step results above it go to blobStore (see 15-large-payloads.md)
});

// Dispatch typed commands
const { workflowRunId } = await kernel.dispatch({
  type: "run.create",
  idempotencyKey: "unique-key",
  workflowId: "my-workflow",
  input: { data: "hello" },
});
```

### Node Host

```typescript
import { createNodeHost } from "@bratsos/workflow-engine-host-node";

const host = createNodeHost({
  kernel,
  jobTransport,
  workerId: "worker-1",
  orchestrationIntervalMs: 10_000,
  jobPollIntervalMs: 1_000,
  staleLeaseThresholdMs: 300_000,   // default as of v0.11 (was 60_000)
  jobAbsoluteTimeoutMs: 3_600_000,  // 1.0: absolute cap on one job claim (default 1h, 0 disables)
  jobHeartbeatIntervalMs: 60_000,   // v0.11+: heartbeat a job's lease while it executes; 1.0: also feeds ctx.abortSignal
  // retention: { olderThanMs: 30 * 24 * 60 * 60 * 1000 }, // 1.0: run.purge on every tick (off by default)
  // serves: "all",                 // 1.0: disable definition-version filtering (default: derived from the registry)
});

await host.start();   // Starts polling loops + signal handlers
await host.stop();    // Graceful shutdown (waits for the in-flight job, then a final outbox.flush)
host.getStats();      // { workerId, jobsProcessed, orchestrationTicks, isRunning, uptimeMs, eventSink }
```

### Serverless Host

```typescript
import {
  createServerlessHost,
  type ServerlessHost,
  type ServerlessHostConfig,
  type JobMessage,
  type JobResult,
  type ProcessJobsResult,
  type MaintenanceTickResult,
} from "@bratsos/workflow-engine-host-serverless";

const host = createServerlessHost({
  kernel,
  jobTransport,
  workerId: "my-worker",
  // Optional tuning (same defaults as Node host)
  staleLeaseThresholdMs: 300_000,   // default as of v0.11 (was 60_000)
  jobAbsoluteTimeoutMs: 3_600_000,  // 1.0: absolute cap on one job claim (default 1h, 0 disables)
  jobHeartbeatIntervalMs: 60_000,   // v0.11+
  maxClaimsPerTick: 10,
  maxSuspendedChecksPerTick: 10,
  maxOutboxFlushPerTick: 100,
  flushOutboxAfterJob: true,        // 1.0: publish outbox events right after handleJob settles (bounded by outboxFlushTimeoutMs, 5_000)
  // retention: { olderThanMs: ... }, serves: "all" — as on the Node host
});
```

#### `handleJob(msg: JobMessage): Promise<JobResult>`

Execute a single pre-dequeued job. Consumers wire platform-specific ack/retry around the result.

```typescript
// JobMessage shape (matches queue message body)
interface JobMessage {
  jobId: string;
  workflowRunId: string;
  workflowId: string;
  stageId: string;
  attempt: number;
  maxAttempts?: number;
  payload: Record<string, unknown>;
}

// JobResult
interface JobResult {
  outcome: "completed" | "suspended" | "failed";
  error?: string;
  dead?: boolean;          // orphan or malformed message: failed and acknowledged, nothing ran
  willRetry?: boolean;     // stage left PENDING with attempts remaining — the job must run again
  attempt?: number;
  maxAttempts?: number;
  retryDelayMs?: number;   // backoff before the retry (2^attempt seconds) when willRetry
}

const result = await host.handleJob(msg);
// Built-in transports re-enqueue a retry from fail(jobId, error, true) themselves, so ack;
// a push transport whose fail() cannot re-enqueue retries the message itself after retryDelayMs.
if (result.willRetry) msg.retry({ delaySeconds: result.retryDelayMs! / 1000 });
else msg.ack();
```

#### `processAvailableJobs(opts?): Promise<ProcessJobsResult>`

Dequeue and process jobs from the job transport. Defaults to 1 job (safe for edge runtimes with CPU limits).

```typescript
const result = await host.processAvailableJobs({ maxJobs: 5 });
// { processed: number, succeeded: number, failed: number }
```

#### `runMaintenanceTick(): Promise<MaintenanceTickResult>`

Run one bounded maintenance cycle: claim pending, poll suspended, reap stale (both lease tiers), flush outbox, reap stuck runs, and purge when `retention` is set.

```typescript
const tick = await host.runMaintenanceTick();
// { claimed, suspendedChecked, staleReleased, staleExpired, eventsFlushed, stuckReaped, purged,
//   eventsFailed, eventsDeadLettered, eventSinkStatus, eventSinkError? }
// Note: resumed suspended stages are automatically followed by run.transition.
```

## Remote Activity Workers

Run a stage's `execute()` on a **separate, credential-free machine** (no database connection, no root object-store credentials) via the **`@bratsos/workflow-engine-host-remote`** package. The orchestrator owns all state; a remote worker leases the task, runs the real stage code, writes large artifacts directly to object storage by reference, and reports back — all driven through the engine's existing suspend/resume machinery (no new DB table).

Two wiring models:

- **Proxy stage** (recommended for long stages): `defineRemoteStage(realStage, transport, opts?)` suspends immediately (releasing the kernel job lease) and resumes when the worker reports.
- **`ActivityExecutor` port** (short stages / in-core routing): inject `createRemoteExecutor(transport)` — or `createRoutingExecutor({ remote, remoteStageIds })` to route only specific stages — via `createKernel({ executor })`. Backward-compatible: the default `createLocalExecutor()` is byte-for-byte the in-process behavior.

```typescript
import { defineRemoteStage } from "@bratsos/workflow-engine-host-remote";

// Orchestrator: wrap a heavy stage so it runs on a remote worker
const workflow = defineWorkflow({ ... })
  .pipe(defineRemoteStage(heavyStage, oTransport, { maxWaitMs: 3_600_000, stageCodeVersion: "v1" }))
  .pipe(coreStage)
  .build();
```

The worker runs in a separate process/machine with **zero standing credentials** (`createActivityWorker` + `createHttpWorkerTransport`), receiving a presigned URL per artifact. See [`references/11-remote-activity-workers.md`](references/11-remote-activity-workers.md) for the worker, broker, HTTP transport, S3/R2 artifacts, durability, and limitations.

## Annotations (Provenance)

Attach typed key-value facts to runs and stages for queryable provenance — trigger context, decisions, approvals, anything else you'd want to ask back later. Writes are buffered during a stage and flushed atomically with the stage outcome (durable, not fire-and-forget).

```typescript
import { Decision, Trigger } from "@bratsos/workflow-engine/conventions";

// Inside a stage's execute()
ctx.annotate(Decision.outcome, "low");                       // typed
ctx.annotate(Decision.confidence, 0.42);                     // typed
ctx.annotate("acme.compliance.signoff", "alice@acme.com");   // custom key
ctx.annotate({
  actor: { kind: "agent", id: "triage-v3" },
  attributes: {
    "decision.outcome": "low",
    "decision.rationale": "below threshold",
    "decision.used_fallback": true,
  },
});

// At run creation — captures trigger context
await kernel.dispatch({
  type: "run.create",
  workflowId: "ticket-triage",
  input: { ticket },
  annotations: [{
    actor: { kind: "system", id: "zendesk" },
    attributes: {
      "trigger.source": "webhook:zendesk",
      "trigger.parent_run_id": previousRunId,
    },
  }],
});

// External attach (plugins, post-hoc reviews)
await kernel.annotations.attach(runId, {
  actor: { kind: "user", id: "alice@acme.com" },
  attributes: { "review.disposition": "approved-anyway" },
  idempotencyKey: "review-2026-05-24-alice",
});

// Query — flat key namespace, indexed
await kernel.annotations.list(runId);
await kernel.annotations.list(runId, { keyPrefix: "decision." });
await kernel.annotations.list(runId, { actorId: "triage-v3" });
```

Annotations replace the deprecated `WorkflowRun.metadata` column. Existing `metadata` is automatically surfaced as `legacy.metadata.*` virtual rows when consumers call `kernel.annotations.list()` (no dual-write, lazy synthesis). See [`references/10-annotations.md`](references/10-annotations.md) for the full API and conventions catalog.

## AI Integration & Cost Tracking

Inside a stage use `ctx.ai` (an `AIHelper` scoped to `workflow.<runId>.stage.<stageId>`, built from `createKernel({ services: { aiLogger } })`) or the durable `ctx.step.ai.*`. Outside a stage, build one yourself:

```typescript
const ai = createAIHelper(
  `workflow.${ctx.workflowRunId}.stage.${ctx.stageId}`,
  aiCallLogger,
);

const { text, cost } = await ai.generateText("gemini-2.5-flash", prompt);
const { object } = await ai.generateObject("gemini-2.5-flash", prompt, schema);
const { embedding } = await ai.embed("text-embedding-004", ["text1"], { dimensions: 768 });
// generateText/generateObject/streamText options also accept maxRetries / abortSignal (v0.11+)
// OpenRouter embedding models (OpenAI, Cohere, etc.)
const { embedding } = await ai.embed("openai/text-embedding-3-small", ["text1"]);

// Provider-specific options passthrough (Voyage, Cohere, etc.)
const { embedding } = await ai.embed("voyage-4-large", ["text1"], {
  providerOptions: { voyage: { outputDimension: 512, inputType: "document" } },
});

// Custom embedding providers (Voyage, Cohere, Jina, etc.)
import { registerEmbeddingProvider } from "@bratsos/workflow-engine";
import { voyage } from "voyage-ai-provider";
registerEmbeddingProvider("voyage", (modelId) => voyage.embeddingModel(modelId));
// Then register models with provider: "voyage" and use ai.embed() as usual

// Batch operations (Google, Anthropic, OpenAI, OpenRouter)
const batch = ai.batch("gemini-2.5-flash", "google");
const handle = await batch.submit([{ id: "1", prompt: "Summarize..." }]);
```

## Persistence Setup

### Required Prisma Models (ALL are required)

Copy the complete schema from the package's `prisma/schema.prisma` (also in the [package README](../../README.md#1-database-setup)). This includes:
WorkflowRun, WorkflowDefinition, WorkflowStage, WorkflowStep, WorkflowLog, WorkflowArtifact, AICall, WorkflowAnnotation, JobQueue, OutboxEvent, IdempotencyKey. `WorkflowBlob` is optional (only for `createPrismaBlobStore`). `sql/enqueue.sql` (the `workflow_engine_enqueue` function) is optional and applied by your own migration.

### Create Persistence

```typescript
import {
  createPrismaWorkflowPersistence,
  createPrismaJobQueue,
  createPrismaAICallLogger,
} from "@bratsos/workflow-engine/persistence/prisma";

const persistence = createPrismaWorkflowPersistence(prisma);
const jobQueue = createPrismaJobQueue(prisma);
const aiCallLogger = createPrismaAICallLogger(prisma);
const stepLedger = createPrismaStepLedger(prisma); // durable steps (workflow_steps)

// SQLite - MUST pass databaseType option
const persistence = createPrismaWorkflowPersistence(prisma, { databaseType: "sqlite" });
const jobQueue = createPrismaJobQueue(prisma, { databaseType: "sqlite" });
```

## Testing

The quickest path is `createTestHarness` (1.0): an in-memory kernel plus the drive loop, with `harness.run(workflowId, input)`, `harness.mockAi`, `harness.steps.mockResult/mockError/mockTimeout/skipSleeps` and `harness.cancel` — see [07-testing-patterns.md](references/07-testing-patterns.md). The hand-wired form:

```typescript
// In-memory persistence and job queue
import {
  InMemoryWorkflowPersistence,
  InMemoryJobQueue,
  InMemoryAICallLogger,
} from "@bratsos/workflow-engine/testing";

// Kernel-specific test adapters
import {
  FakeClock,
  InMemoryBlobStore,
  CollectingEventSink,
} from "@bratsos/workflow-engine/kernel/testing";

// Create kernel with all in-memory adapters
const persistence = new InMemoryWorkflowPersistence();
const jobQueue = new InMemoryJobQueue();
const kernel = createKernel({
  persistence,
  blobStore: new InMemoryBlobStore(),
  jobTransport: jobQueue,
  eventSink: new CollectingEventSink(),
  clock: new FakeClock(),
  registry: { getWorkflow: (id) => workflows.get(id) },
});

// Test a full workflow lifecycle
await kernel.dispatch({ type: "run.create", idempotencyKey: "test", workflowId: "my-wf", input: {} });
await kernel.dispatch({ type: "run.claimPending", workerId: "test-worker" });
const job = await jobQueue.dequeue();
await kernel.dispatch({ type: "job.execute", workflowRunId: job.workflowRunId, workflowId: job.workflowId, stageId: job.stageId, config: {} });
await kernel.dispatch({ type: "run.transition", workflowRunId: job.workflowRunId });
```

Implementing a custom `WorkflowPersistence`/`JobQueue`/`AICallLogger`/`StepLedger` adapter? Validate it with the exported conformance suites (v0.11+) instead of hand-rolling parity tests — see [07-testing-patterns.md](references/07-testing-patterns.md#conformance-suites-for-custom-adapters-v011).

## Reference Files

- [01-stage-definitions.md](references/01-stage-definitions.md) - Complete stage API (`defineStage`, the stage context, result shapes, the legacy async-batch mode)
- [02-workflow-builder.md](references/02-workflow-builder.md) - `defineWorkflow` / WorkflowBuilder patterns, `.version()`, type inference
- [03-runtime-setup.md](references/03-runtime-setup.md) - Kernel & host configuration, job-lease tiers, degraded event sink
- [04-ai-integration.md](references/04-ai-integration.md) - AI helper methods
- [05-persistence-setup.md](references/05-persistence-setup.md) - Database setup
- [06-async-batch-stages.md](references/06-async-batch-stages.md) - Batch AI operations and the legacy async-batch stage mode
- [07-testing-patterns.md](references/07-testing-patterns.md) - `createTestHarness`, step mocks, conformance suites, testing with the kernel
- [08-common-patterns.md](references/08-common-patterns.md) - Kernel patterns & best practices
- [09-troubleshooting.md](references/09-troubleshooting.md) - Debugging stuck runs, P2002 errors, ghost jobs
- [10-annotations.md](references/10-annotations.md) - First-class provenance surface: `ctx.annotate`, `kernel.annotations.*`, conventions catalog
- [11-remote-activity-workers.md](references/11-remote-activity-workers.md) - Credential-free remote workers: `defineRemoteStage`, broker, worker SDK, HTTP transport, S3/R2 artifacts, `ActivityExecutor` port
- [12-durable-steps.md](references/12-durable-steps.md) - Durable steps (`ctx.step.run/waitFor/waitForSignal/sleep`), determinism rules, `DuplicateStepKeyError`, `step.externalKey`/`onReclaim`, `ctx.step.ai.*` and `ai.map` policies, `ctx.ai` injection, adapter seam and timeouts, the builder-first `defineWorkflow().stage()` API, migrating async-batch stages to steps
- [13-definition-versioning.md](references/13-definition-versioning.md) - `.version()` and the derived structural hash, the `workflow_definitions` snapshot, version-filtered claiming (`serves`, `ghostReason: "version"`), `run.listVersions`, and shadowing a candidate build against live runs
- [14-redrive.md](references/14-redrive.md) - `run.redrive`'s retry / restart / rerun modes, re-pinning onto another definition version, the preserved `run.supersededAttempt`, and migrating off the deprecated `run.rerunFrom`
- [15-large-payloads.md](references/15-large-payloads.md) - The claim check: automatic step-result spilling, opt-in job-payload spilling, `spillThresholdBytes`, and what is deliberately not spilled
- [16-operational-console.md](references/16-operational-console.md) - `@bratsos/workflow-engine-console`: mounting the handler, the deny-by-default action vocabulary, `ConsoleReadPort`, query timeouts, and the dev CLI

## Key Principles

1. **Type Safety**: All schemas are Zod - types flow through the entire pipeline
2. **Command Kernel**: All operations are typed commands dispatched through `kernel.dispatch()`
3. **Environment-Agnostic**: Kernel has no timers, no signals, no global state
4. **Context Access**: Use `ctx.require()` and `ctx.optional()` for type-safe stage output access
5. **Transactional Outbox**: Events written to outbox, published via `outbox.flush` command. `job.execute` and `stage.pollSuspended` use multi-phase transactions to avoid holding connections during external I/O; `stage.pollSuspended` claims each suspended stage (version-guarded `nextPollAt` lease) before polling it, so several orchestrating processes replay a stage once
6. **Idempotency**: `run.create`, `job.execute`, `run.redrive` and `run.rerunFrom` replay cached results by key; concurrent same-key dispatch throws `IdempotencyInProgressError`; a key stuck `in_progress` past `KernelConfig.idempotencyStaleInProgressMs` (default 10 min, v0.11+) can be reclaimed
7. **Authoritative Cancellation**: `run.cancel` cascades to stages + jobs and aborts `ctx.abortSignal` in the executing body (via the host's `job.heartbeat`). Ghost jobs (running against non-RUNNING runs) are reported with `ghost: true` and a `ghostReason`: `"orphan"` is discarded, `"race"` and `"version"` are re-delivered
8. **Self-Healing**: Stage creation is idempotent (upsert), orchestration steps are isolated, stuck runs are automatically reaped
9. **Cost Tracking**: All AI calls automatically track tokens and costs
10. **BlobStore-Only Artifacts**: All artifact storage goes through the BlobStore port. `run.redrive` cleans up the artifacts of the stage records it deletes by key prefix after commit (a reopened stage keeps its storage). A durable step result over `spillThresholdBytes` (64 KiB) is written there too, with the ledger row keeping a reference — see [15-large-payloads.md](references/15-large-payloads.md)
11. **Durable Provenance**: `ctx.annotate(...)` writes are buffered and flushed inside the stage-completion transaction. Annotations are atomic with the stage outcome — a stage's annotations either all persist or all roll back together with the stage update and outbox events.
12. **Pluggable Execution**: stage execution goes through an injectable `ActivityExecutor` port (default in-process `LocalExecutor`). Inject a remote executor — or wrap a stage with `defineRemoteStage` — to run `execute()` on a separate credential-free machine without changing kernel internals.
13. **Definition Pinning**: a run records the definition version it was created under, and a host built with `createWorkflowRegistry` claims only runs pinned to a version it serves. A run at a version this build does not serve is left alone — `PENDING` runs stay pending, a `RUNNING` job comes back with `ghostReason: "version"` — never failed, because a host cannot tell a decommissioned fleet from a peer mid-deploy. See [13-definition-versioning.md](references/13-definition-versioning.md).