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

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

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", ... }):

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.

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

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

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

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.

// 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).

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.

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.

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 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).

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 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:

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). 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

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. The hand-wired form:

// 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.

Reference Files

01-stage-definitions.md - Complete stage API (defineStage, the stage context, result shapes, the legacy async-batch mode)
02-workflow-builder.md - defineWorkflow / WorkflowBuilder patterns, .version(), type inference
03-runtime-setup.md - Kernel & host configuration, job-lease tiers, degraded event sink
04-ai-integration.md - AI helper methods
05-persistence-setup.md - Database setup
06-async-batch-stages.md - Batch AI operations and the legacy async-batch stage mode
07-testing-patterns.md - createTestHarness, step mocks, conformance suites, testing with the kernel
08-common-patterns.md - Kernel patterns & best practices
09-troubleshooting.md - Debugging stuck runs, P2002 errors, ghost jobs
10-annotations.md - First-class provenance surface: ctx.annotate, kernel.annotations.*, conventions catalog
11-remote-activity-workers.md - Credential-free remote workers: defineRemoteStage, broker, worker SDK, HTTP transport, S3/R2 artifacts, ActivityExecutor port
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 - .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 - 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 - The claim check: automatic step-result spilling, opt-in job-payload spilling, spillThresholdBytes, and what is deliberately not spilled
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

Type Safety: All schemas are Zod - types flow through the entire pipeline
Command Kernel: All operations are typed commands dispatched through kernel.dispatch()
Environment-Agnostic: Kernel has no timers, no signals, no global state
Context Access: Use ctx.require() and ctx.optional() for type-safe stage output access
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
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
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
Self-Healing: Stage creation is idempotent (upsert), orchestration steps are isolated, stuck runs are automatically reaped
Cost Tracking: All AI calls automatically track tokens and costs
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
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.
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.
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.