API Reference

createEngine

function createEngine(opts?: EngineOptions): Engine

Creates an engine instance. All options are optional; by default the engine uses an in-memory store with a concurrency of 10.

import { createEngine } from '@yellatech/engine'

const engine = createEngine()

The returned object exposes every method documented below.

Handler Registration

engine.register

// Without schema validation
function register(
  name: string,
  eventName: string,
  handler: Handler,
  opts?: { retry?: RetryPolicy; version?: string; concurrency?: number; emits?: string[] }
): void

// With schema validation (e.g. Zod)
function register<T>(
  name: string,
  eventName: string,
  schema: Schema<T>,
  handler: (ctx: HandlerContext<T>) => Promise<HandlerResult> | HandlerResult,
  opts?: { retry?: RetryPolicy; version?: string; concurrency?: number; emits?: string[] }
): void

Registers a named handler for an event. The name must be unique across all registrations.

When a schema is provided (any object with a .parse(data) method, such as a Zod schema), the payload is validated before the handler runs. If validation fails, the run transitions to errored with code SCHEMA_VALIDATION_FAILED.

import { z } from 'zod'

engine.register('send-welcome', 'user:signup', z.object({
  email: z.string().email(),
  name: z.string(),
}), async (ctx) => {
  // ctx.payload is typed as { email: string; name: string }
  await sendEmail(ctx.payload.email, `Welcome, ${ctx.payload.name}!`)
  return { success: true }
}, { retry: { maxRetries: 3, delay: 1000 } })

engine.on

function on(eventName: string, handler: Handler): string

Shorthand for register that auto-generates a unique process name. Returns the generated name.

const name = engine.on('user:signup', async (ctx) => {
  console.log('New user:', ctx.payload)
  return { success: true }
})

engine.registerMany

function registerMany(defs: {
  name: string
  event: string
  handler: Handler
  schema?: Schema
  retry?: RetryPolicy
}[]): void

Registers multiple handlers in a single call.

engine.registerMany([
  { name: 'log-signup', event: 'user:signup', handler: logHandler },
  { name: 'send-welcome', event: 'user:signup', handler: welcomeHandler,
    retry: { maxRetries: 3 } },
])

engine.process

function process<T>(config: ProcessDefinitionConfig<T>): void

Ergonomic alternative to register. Uses a single config object with named fields and provides a ProcessContext with ok(), fail(), and named-parameter effect() helpers.

engine.process({
  name: 'charge-order',
  on: 'order:paid',
  schema: z.object({ orderId: z.string(), total: z.number() }),
  retry: { maxRetries: 3, delay: 1000 },
  emits: ['order:charged'],
  run: async (ctx) => {
    const charge = await ctx.effect({
      key: 'stripe-charge',
      run: () => stripe.charges.create({ amount: ctx.payload.total }),
    })
    return ctx.ok({ chargeId: charge.id }, { emit: 'order:charged' })
  },
})

The process() handler receives a ProcessContext instead of HandlerContext:

Both register() and process() coexist. Use whichever style you prefer.

engine.unregister

function unregister(name: string): void

Removes a registered handler by name. Future events will no longer match this handler.

Emitting Events

engine.emit

function emit(
  eventName: string,
  payload: unknown,
  opts?: { idempotencyKey?: string }
): Run[]

Emits an event. Returns the created Run objects (one per matching handler). If the engine is stopped or no handlers match, returns an empty array.

The optional idempotencyKey prevents duplicate processing. If a run already exists for that key, no new runs are created and emit() returns an empty array.

const runs = engine.emit('order:placed', { orderId: '123' }, {
  idempotencyKey: 'order-123',
})

engine.emitAndWait

async function emitAndWait(
  eventName: string,
  payload: unknown,
  opts?: { idempotencyKey?: string; timeoutMs?: number }
): Promise<Run[]>

Emits an event and waits for the engine to go idle before returning. The returned runs contain final state and results. Times out after timeoutMs (default 30000ms). If the emit is skipped because of a duplicate idempotencyKey, or the engine is already stopped, this resolves to an empty array.

const runs = await engine.emitAndWait('user:signup', { name: 'Alice' })
console.log(runs[0].result) // { success: true }

HandlerContext

Every handler receives a HandlerContext object.

type HandlerContext<T = unknown> = {
  event: string
  payload: T
  runId: string
  correlationId: string
  context: Record<string, unknown>
  setContext(key: string, value: unknown): void
  effect<R>(effectKey: string, fn: () => Promise<R> | R): Promise<R>
}

ctx.effect

ctx.effect is the primary mechanism for safe side effects. Wrap any external call (API requests, emails, database writes) in an effect:

engine.register('charge', 'order:placed', async (ctx) => {
  const payload = ctx.payload as { total: number }
  const charge = await ctx.effect('stripe-charge', () =>
    stripe.charges.create({ amount: payload.total })
  )
  return { success: true, payload: { chargeId: charge.id } }
})

If the handler retries after a crash, the charge is not re-executed. The stored result is returned instead.

HandlerResult

Handlers must return a HandlerResult.

type HandlerResult = {
  success: boolean
  payload?: unknown
  triggerEvent?: string
  deferred?: boolean
  error?: string
  errorCode?: ErrorCode
}

Chaining Events

Return triggerEvent to create parent-child chains:

engine.register('validate', 'order:placed', async (ctx) => {
  const payload = ctx.payload as { orderId: string }
  return {
    success: true,
    triggerEvent: 'order:validated',
    payload: { orderId: payload.orderId },
  }
})

engine.register('ship', 'order:validated', async (ctx) => {
  return { success: true }
})

Child runs share the parent’s correlationId and inherit its context.

Query Methods

engine.getRun

function getRun(id: string): Run | null

Returns a single run by ID, or null if not found.

engine.getRunning

function getRunning(): Run[]

Returns all runs currently in the running state.

engine.getCompleted

function getCompleted(): Run[]

Returns all runs in the completed state.

This method is deprecated for large result sets. This is a raw state query, so it includes runs whose derived status is deferred. Prefer getRunsByStatusPaginated('completed', ...) when you want operator-facing completed-only results without deferred runs.

engine.getErrored

function getErrored(): Run[]

Returns all runs in the errored state (including dead-lettered runs that exhausted retries).

engine.getIdle

function getIdle(): Run[]

Returns all runs in the idle state (queued but not yet claimed by the dispatcher).

engine.getChain

function getChain(runId: string): Run[]

Returns the given run and all of its descendants. To retrieve the full causal chain from the top, pass the root run ID first.

engine.getProcesses

function getProcesses(): ProcessDefinition[]

Returns all registered process definitions.

engine.getEffects

function getEffects(runId: string): EffectRecord[]

Returns all durable effect records for a given run. Useful for inspecting which effects completed, failed, or are still in-flight.

engine.countByState

function countByState(state: ProcessState): number

Returns the number of runs in a given raw engine state without loading every run into memory.

console.log(engine.countByState('running'))
console.log(engine.countByState('completed'))

engine.getRunsPaginated

function getRunsPaginated(
  state: ProcessState | null,
  limit: number,
  offset: number,
  opts?: RunQueryOptions
): { runs: Run[]; total: number }

Returns a page of runs filtered by raw engine state. Pass null to page across all runs. Use opts.root to restrict results to root runs, and opts.order to sort by startedAt.

const page = engine.getRunsPaginated('running', 25, 0, {
  root: true,
  order: 'desc',
})

console.log(page.total)
console.log(page.runs.map((run) => run.id))

engine.getRunsByStatusPaginated

function getRunsByStatusPaginated(
  status: RunStatus,
  limit: number,
  offset: number,
  opts?: RunQueryOptions
): { runs: Run[]; total: number }

Returns a page of runs filtered by operator-facing status, including derived statuses like deferred and dead-letter.

const page = engine.getRunsByStatusPaginated('dead-letter', 25, 0, {
  order: 'desc',
})

console.log(page.total)
console.log(page.runs[0]?.result?.error)

engine.getMetrics

function getMetrics(): EngineMetrics

Returns a current queue snapshot plus in-process totals since engine creation.

const metrics = engine.getMetrics()
console.log(metrics.queue.running)
console.log(metrics.totals.deadLetters)

engine.getObservability

function getObservability(query?: EngineObservabilityQuery): EngineObservabilityReport

Returns bucketed rollups for runs, effects, and system-level errors over a time window. This is the same aggregated data used by the dashboard charts.

const report = engine.getObservability({
  from: Date.now() - 15 * 60 * 1000,
  to: Date.now(),
})

console.log(report.summary.runs.completed)
console.log(report.recentErrors)

engine.subscribeEvents

function subscribeEvents(listener: (event: EngineEvent) => void): () => void

Registers a low-latency lifecycle listener and returns an unsubscribe function. This complements onEvent when you need to attach listeners after engine creation.

const unsubscribe = engine.subscribeEvents((event) => {
  if (event.type === 'run:retry') {
    console.warn(`retrying ${event.run.processName}: ${event.error}`)
  }
})

unsubscribe()

engine.getGraph

function getGraph(): EventGraph

Returns the static event flow graph built from all registered processes. Nodes represent processes, edges represent event connections between them (process A emits an event that process B listens for).

The graph is built from the emits declarations on your processes. If no processes declare emits, the graph will have nodes but no edges.

const graph = engine.getGraph()
console.log(graph.nodes)
// [{ name: 'validate', on: 'order:new', emits: ['order:validated'] }, ...]
console.log(graph.edges)
// [{ from: 'validate', event: 'order:validated', to: 'charge' }, ...]

Repair APIs

engine.retryRun

function retryRun(runId: string): Run

Manually retries a run that is in the errored state. Resets the run to idle with retryAfter = 0 (immediate) and kicks the dispatcher. Throws EngineError with code RUN_NOT_FOUND if the run does not exist, or INVALID_RUN_STATE if not in the errored state.

engine.requeueDead

function requeueDead(runId: string): Run

Requeues a dead-lettered run (one that exhausted its retry budget). Unlike retryRun, this also resets the attempt counter, giving the run a fresh set of retries. Throws EngineError with code RUN_NOT_FOUND if the run does not exist, or INVALID_RUN_STATE if the run is not in dead-letter status.

engine.cancelRun

function cancelRun(runId: string): Run

Cancels a run in the idle or running state by transitioning it to errored with the error 'cancelled'. Throws EngineError with code RUN_NOT_FOUND if the run does not exist, or INVALID_RUN_STATE if the run is not in idle or running state.

Deferred Events

engine.resume

function resume(runId: string, payload?: unknown): Run[]

Resumes a deferred run. When a handler returns { triggerEvent, deferred: true }, the event is stored but not emitted. Calling resume() emits the stored event, continuing the chain with the original correlation ID and context.

If both the stored result payload and the provided payload are objects, they are shallow-merged. Otherwise, the provided payload replaces the stored payload.

// A handler that defers the next step
engine.register('validate', 'order:new', async (ctx) => {
  const payload = ctx.payload as { orderId: string }
  return {
    success: true,
    triggerEvent: 'order:approved',
    deferred: true,
    payload: { orderId: payload.orderId },
  }
})

// Later, resume it
const runs = engine.resume(completedRunId, { approvedBy: 'alice' })
// 'order:approved' is now emitted with merged payload

Throws EngineError with:

• RUN_NOT_FOUND if the run does not exist.
• INVALID_RUN_STATE if the run is not completed, has no triggerEvent, or is not deferred.

See Concepts: Deferred Events for the full pattern.

Lifecycle

engine.stop

async function stop(opts?: {
  graceful?: boolean
  timeoutMs?: number
}): Promise<void>

Stops the engine. No new events are accepted after calling stop.

• Immediate stop (default): Stops accepting new work and tears down runtime resources without waiting for active handlers to finish cleanly.
• Graceful stop: Set graceful: true to wait for all in-flight runs to finish before resolving. Times out after timeoutMs (default 30000ms).

// Immediate
await engine.stop()

// Graceful with 10s timeout
await engine.stop({ graceful: true, timeoutMs: 10_000 })

If the engine was created with a SQLite store, stop also closes the database connection.

engine.drain

async function drain(timeoutMs?: number): Promise<void>

Waits until all idle and running runs have finished. Resolves immediately if the queue is already empty. Rejects with DRAIN_TIMEOUT if the timeout (default 30000ms) is exceeded.

engine.getServer

function getServer(): Promise<DevServer> | null

Returns a promise that resolves to the dev server handle, or null if no server option was provided to createEngine. The promise resolves once the port is bound.

const engine = createEngine({ server: { port: 3000 } })
const server = await engine.getServer()!
console.log(server.address) // { host: '127.0.0.1', port: 3000 }

// Add custom routes via the Hono app
server.app.get('/custom', (c) => c.json({ ok: true }))

// Stop the server independently
await server.stop()

The DevServer handle also exposes serve(), but when it comes from engine.getServer() the server is already listening.

See Dashboard for the full dashboard route list and panel walkthrough.

Types

Run

The core data structure representing a single handler execution.

type Run = {
  id: string
  correlationId: string
  processName: string
  eventName: string
  state: ProcessState          // 'idle' | 'running' | 'completed' | 'errored'
  context: Record<string, unknown>
  payload: unknown
  result: HandlerResult | null
  timeline: TimelineEntry[]
  parentRunId: string | null
  childRunIds: string[]
  startedAt: number
  completedAt: number | null
  depth: number
  idempotencyKey: string | null
  attempt: number
  retryAfter: number | null
  leaseOwner: string | null
  leaseExpiresAt: number | null
  heartbeatAt: number | null
  handlerVersion: string | null
}

ProcessDefinition

type ProcessDefinition = {
  name: string
  eventName: string
  handler: Handler
  schema?: Schema
  retry?: RetryPolicy
  version?: string
  concurrency?: number
  emits?: string[]
}

ProcessDefinitionConfig

type ProcessDefinitionConfig<T = unknown> = {
  name: string
  on: string
  schema?: Schema<T>
  retry?: RetryPolicy
  version?: string
  concurrency?: number
  emits?: string[]
  run: (ctx: ProcessContext<T>) => Promise<HandlerResult> | HandlerResult
}

Configuration object for engine.process(). The run function receives a ProcessContext.

ProcessContext

type ProcessContext<T = unknown> = {
  event: string
  payload: T
  runId: string
  correlationId: string
  context: Record<string, unknown>
  setContext(key: string, value: unknown): void
  effect: EffectFn
  ok<P>(payload?: P, opts?: { emit?: string }): HandlerResult
  fail(error: string, code?: ErrorCode): HandlerResult
}

Ergonomic context passed to process() handlers. Wraps HandlerContext with:

• ok(payload?, opts?), Returns a success result. Pass { emit: 'event:name' } to chain.
• fail(error, code?), Returns a failure result.
• effect({ key, run }), Named-parameter effect. key accepts string or string[].

RetryPolicy

type RetryPolicy = {
  maxRetries: number
  delay?: number | ((attempt: number) => number)
}

The delay field accepts a fixed number of milliseconds or a function that computes delay from the current attempt (0-based). See Configuration for examples.

RunStatus

type RunStatus = ProcessState | 'deferred' | 'dead-letter'

Operator-facing status exposed by the API and dashboard. This distinguishes raw engine lifecycle state from derived states like deferred and dead-letter.

RunQueryOptions

type RunQueryOptions = {
  root?: boolean
  order?: 'asc' | 'desc'
}

Shared paging options for getRunsPaginated() and getRunsByStatusPaginated().

EngineOptions

See Configuration for the full option breakdown, and Dashboard for the built-in UI and route surface.

EffectRecord

type EffectRecord = {
  runId: string
  effectKey: string
  state: EffectState             // 'started' | 'completed' | 'failed'
  output: unknown
  error: string | null
  startedAt: number
  completedAt: number | null
}

TimelineEntry

type TimelineEntry = {
  state: ProcessState
  timestamp: number
  event?: string
  payload?: unknown
  error?: string
}

EventGraph

The static event flow graph returned by engine.getGraph().

type EventGraphNode = {
  name: string        // process name
  on: string          // event this process listens for
  emits: string[]     // events this process may emit
}

type EventGraphEdge = {
  from: string        // source process name
  event: string       // the connecting event
  to: string          // target process name
}

type EventGraph = {
  nodes: EventGraphNode[]
  edges: EventGraphEdge[]
}

EngineMetrics

type EngineMetrics = {
  queue: { idle: number; running: number; completed: number; errored: number }
  totals: { retries: number; deadLetters: number; resumes: number; leaseReclaims: number; internalErrors: number }
}

Current queue counts plus in-process totals accumulated since the engine instance was created.

EngineEvent

type EngineEvent =
  | { type: 'run:start'; run: Run }
  | { type: 'run:complete'; run: Run; durationMs: number }
  | { type: 'run:error'; run: Run; error: string; durationMs: number }
  | { type: 'run:retry'; run: Run; error: string; attempt: number }
  | { type: 'run:dead'; run: Run; error: string }
  | { type: 'run:resume'; resumedRun: Run; childRuns: Run[] }
  | { type: 'effect:complete'; runId: string; effectKey: string; durationMs: number }
  | { type: 'effect:error'; runId: string; effectKey: string; error: string; durationMs: number }
  | { type: 'effect:replay'; runId: string; effectKey: string }
  | { type: 'lease:reclaim'; run: Run }
  | { type: 'internal:error'; error: unknown; context: string }

Lifecycle event union used by onEvent() and subscribeEvents().

EngineObservabilityQuery

type EngineObservabilityQuery = {
  from?: number
  to?: number
  bucketMs?: number
  errorLimit?: number
}

Optional time-window settings passed to getObservability().

EngineObservabilityReport

type EngineObservabilityReport = {
  from: number
  to: number
  bucketSizeMs: number
  summary: EngineObservabilitySummary
  buckets: EngineObservabilityBucket[]
  recentErrors: EngineObservabilityError[]
}

Bucketed observability rollups returned by getObservability().

ErrorCode

const ErrorCode = {
  PAYLOAD_TOO_LARGE: 'PAYLOAD_TOO_LARGE',
  CHAIN_DEPTH_EXCEEDED: 'CHAIN_DEPTH_EXCEEDED',
  HANDLER_TIMEOUT: 'HANDLER_TIMEOUT',
  HANDLER_NOT_FOUND: 'HANDLER_NOT_FOUND',
  SCHEMA_VALIDATION_FAILED: 'SCHEMA_VALIDATION_FAILED',
  PROCESS_ALREADY_REGISTERED: 'PROCESS_ALREADY_REGISTERED',
  DRAIN_TIMEOUT: 'DRAIN_TIMEOUT',
  GRACEFUL_STOP_TIMEOUT: 'GRACEFUL_STOP_TIMEOUT',
  LEASE_EXPIRED: 'LEASE_EXPIRED',
  RUN_NOT_FOUND: 'RUN_NOT_FOUND',
  INVALID_RUN_STATE: 'INVALID_RUN_STATE',
  INVALID_CONFIG: 'INVALID_CONFIG',
}

DevServer

type DevServer = {
  address: { host: string; port: number }
  app: Hono
  serve(opts?: { host?: string; port?: number }): Promise<{ host: string; port: number }>
  stop(): Promise<void>
}

Handle returned by engine.getServer(). The app field is the Hono instance, exposed for adding custom routes.

DevServerOptions

type DevServerOptions = {
  host?: string   // default: '127.0.0.1'
  port?: number   // default: 3000
}

Options for the dev dashboard server, passed via EngineOptions.server.

Schema

type Schema<T = unknown> = { parse(data: unknown): T }

Any object with a .parse() method. Zod schemas satisfy this interface out of the box.