Engine methods

Methods on the Engine returned by createEngine.

engine.start(pollIntervalMs?)

Initializes storage and starts the polling loop, running tick() every pollIntervalMs (default 1000). Call once at startup. Returns Promise<void>.

engine.stop()

Stops the polling loop, ends all open streams, aborts in-flight steps, and waits for any in-flight tick to finish. In-flight runs are left running so they can be reclaimed. Registered schedules are deliberately left in place — they belong to the storage, not to this instance. Returns Promise<void>.

engine.tick()

Claims up to concurrency pending or stale runs and executes them in parallel. Useful for CLI tools and tests. If you use tick() without start(), call storage.initialize() first. Returns Promise<void>.

engine.enqueue(name, input, options?)

Submits a run. Type-safe — only accepts registered workflow names with matching input. Returns the created WorkflowRun (use run.id to track it).

Option	Type	Description
idempotencyKey	string	Same key + same input returns the existing run; same key + different input throws IdempotencyConflictError

Throws WorkflowNotFoundError for an unknown name and ValidationError if input fails the schema.

engine.stream(options?)

Returns a pull-based ResultStream — an AsyncIterableIterator<EngineEvent> (and AsyncDisposable) of execution events. Each call returns an independent stream.

Option	Type	Default	Description
bufferSize	number	Infinity	Max events buffered before the engine pauses (backpressure). 0 = strict rendezvous; any non-negative integer allows that many buffered events.

Invalid bufferSize (negative, or a non-integer that isn't Infinity) throws ConfigError. Breaking out of the loop, await using, or engine.stop() unsubscribes automatically. See Streaming Results.

engine.cancel(runId)

Cancels a pending or running workflow. Returns true if cancelled, false if it already completed / failed / cancelled. Aborts the current step's AbortSignal immediately. See Cancellation.

engine.sendEvent(runId, eventName, payload)

Delivers an external event to a run that is (or will be) waiting on waitForEvent(eventName). The payload — validated against that wait's schema, if any — becomes the wait's result and the next step's prev.

await engine.sendEvent(run.id, 'approved', { approver: 'alice' })

Delivery is durable and order-independent: an event sent before the run reaches the wait is buffered and consumed when it gets there. Returns false if the run does not exist or has already finished (completed / failed / cancelled); throws ConfigError if the workflow has no such event step, or ValidationError if the payload fails the wait's schema. See Waiting for Events.

engine.schedule(name, input, recurrence, options?)

Registers a durable recurring schedule. Returns Promise<string> resolving to its key. Validates name, input, and the recurrence immediately.

recurrence is a number of milliseconds, { every } (milliseconds or a duration string like '1h'), or { cron } (a five-field expression, evaluated in UTC). A malformed or unsatisfiable cron expression throws ConfigError here rather than on the first tick.

The schedule is persisted rather than held as an in-process timer, so it survives restarts; registering the same key again updates it in place, preserving the cadence unless the interval changed. Due firings are claimed atomically, so N instances sharing a schedule produce one run per interval. options.key sets the identity explicitly (it otherwise defaults to a hash of the name, interval, and input). See Scheduled Workflows.

engine.unschedule(key)

Removes a durable schedule. Returns Promise<boolean> — false if no such schedule existed. Because schedules are shared, this stops it for every instance.

engine.listSchedules()

Returns Promise<readonly WorkflowSchedule[]> — every registered schedule, ordered by key, each carrying its nextRunAt.

engine.getRunStatus(runId)

Returns { run, steps } — the run's current status and all its step results — or null if the run is not found.

const info = await engine.getRunStatus(run.id)
info?.run.status // 'pending' | 'running' | 'sleeping' | 'waiting' | 'completed' | 'failed' | 'cancelled'
info?.steps      // StepResult[]

engine.listRuns(filter?)

Lists runs ordered by createdAt descending, then id descending (a total, stable order even when runs share a createdAt). For inspection or dead-letter visibility. All filter fields are optional:

await engine.listRuns({
  status: 'failed',   // only this RunStatus
  workflow: 'order',  // only this workflow
  limit: 50,          // max rows (default 100; must be a positive integer)
  before: cursor,     // keyset cursor (createdAt), paired with beforeId
  beforeId: cursorId, // keyset cursor tie-break (id)
})

Paginate with a keyset cursor — pass the last row's createdAt and id. Using both is exact even when timestamps collide; before alone is a coarse "created before T" filter that can drop runs tied on that millisecond:

const page1 = await engine.listRuns({ limit: 50 })
const last = page1.at(-1)
const page2 = last
  ? await engine.listRuns({ limit: 50, before: last.createdAt, beforeId: last.id })
  : []

engine.resume(runId)

Re-queues a failed or cancelled run so the engine picks it up again. Completed steps are preserved and skipped on replay, so execution resumes at the step that failed; the failed step's result is discarded and re-run. Returns true if the run was resumable, false if it does not exist or is in any other state (pending / running / completed).

const failed = await engine.listRuns({ status: 'failed' })
for (const run of failed) {
  await engine.resume(run.id)
}