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`.
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`.
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.
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, defineAsyncBatchStage, WorkflowBuilder
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
Create sync stages. Curried form defineStage<TContext>()({...}) is recommended when you need typed ctx.require()/ctx.optional() — see 01-stage-definitions.md
defineAsyncBatchStage
Function
@bratsos/workflow-engine
Create async/batch stages
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. Its 5-positional-argument constructor (new WorkflowBuilder(id, name, description, input, output)) is @deprecated in favor of defineWorkflow() — the class itself (and .pipe()/.parallel()/.build()) is unaffected
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
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({...}) (recommended) — the 5-positional-argument new WorkflowBuilder(id, name, description, input, output) constructor is @deprecated (same-typed positional args are easy to transpose by accident); both return the same builder for .pipe()/.parallel()/.build().
const workflow = defineWorkflow({
id: "workflow-id",
name: "Workflow Name",
description: "Description",
input: InputSchema,
// output is optional -- decorative unless the workflow has zero piped stages
})
.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";
importtype { 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 - { getWorkflow(id) }// executor, // optional ActivityExecutor port - defaults to in-process; inject to run stages on remote workers (see 11-remote-activity-workers.md)// scheduler, // optional Scheduler port - @deprecated, unused by the kernel (zero schedule()/cancel() call sites); omit it, the kernel supplies its own no-op. Removal at 1.0// idempotencyStaleInProgressMs: 10 * 60 * 1000, // optional (v0.11+) - default 10 min; TTL before a stuck `in_progress` idempotency key can be reclaimed
});
// Dispatch typed commandsconst { 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)jobHeartbeatIntervalMs: 60_000, // v0.11+: heartbeat a job's lease while it executes
});
await host.start(); // Starts polling loops + signal handlersawait host.stop(); // Graceful shutdown
host.getStats(); // { workerId, jobsProcessed, orchestrationTicks, isRunning, uptimeMs }
Run one bounded maintenance cycle: claim pending, poll suspended, reap stale, flush outbox, reap stuck runs.
const tick = await host.runMaintenanceTick();
// { claimed, suspendedChecked, staleReleased, eventsFlushed, stuckReaped }// 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 workerconst 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).
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.
Copy the complete schema from the package README. This includes:
WorkflowRun, WorkflowStage, WorkflowLog, WorkflowArtifact, AICall, JobQueue, OutboxEvent, IdempotencyKey.
Implementing a custom WorkflowPersistence/JobQueue/AICallLogger adapter? Validate it with the exported conformance suites (v0.11+) instead of hand-rolling parity tests — see 07-testing-patterns.md.
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
Idempotency: run.create, job.execute, and run.rerunFrom (v0.11+) 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. Ghost jobs (running against non-RUNNING runs) are detected via ghost: true flag and not retried
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.rerunFrom cleans up artifacts by key prefix
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.
