Workflows & steps

WorkflowsCopy Anchor Link

A workflow is registered on the engine by name and version; its body is the deterministic async function (ctx, input) => … the engine executes and replays.

import engine from '@adonis-agora/durable/services/main'
import type { WorkflowCtx } from '@adonis-agora/durable'

engine.register('checkout', '1', async (ctx: WorkflowCtx, order: Order) => {
  /* ... deterministic orchestration ... */
})

register(name, version, fn, opts?) takes an options object:

Start runs with engine.start('checkout', input, runId) — pass the registered name and a stable runId. (Starting the same runId twice is a no-op, which is what makes scheduling and webhooks idempotent.)

1@Workflow({ name: 'checkout', version: '1' })2export default class CheckoutWorkflow {3  constructor(4    private inventory: InventorySteps,5    private payments: PaymentSteps,6    private shipping: ShippingSteps,7    private email: EmailSteps,8  ) {}9 10  async run(ctx: WorkflowCtx, order: Order) {11    const hold = await ctx.step(this.inventory.reserve, order)12    const charge = await ctx.step(this.payments.charge, { order, hold })13    await ctx.waitForSignal(`packed:${order.id}`)14    const label = await ctx.step(this.shipping.ship, order)15    await ctx.step(this.email.confirm, { order, label })16    return { chargeId: charge.id, tracking: label.tracking }17  }18}

StepsCopy Anchor Link

ctx.step(refOrName, input, opts?) is the one durable step primitive. It is always dispatched: the engine routes it to a handler registered under the step's name, suspends the run (zero compute) until the result lands, then resumes with it — checkpointed and replay-safe. The handler runs on whatever worker serves that name, in this process or another; the workflow code is the same either way.

There are three ways to author the handler a ctx.step routes to.

1// the workflow dispatches it by name and awaits the result:2await ctx.step(this.inventory.reserve, order)

1. By string nameCopy Anchor Link

The zero-decorator baseline — dispatch by the literal handler name. Nothing to import at the call site, and it works across runtimes (e.g. a Python worker serving the same name):

const charge = await ctx.step<{ chargeId: string }>(
  'payments:charge-card',
  { orderId: order.id, amountCents: order.total },
)

The handler is registered under that same name — see serving a handler.

2. @Step decorator (recommended)Copy Anchor Link

Mark a class method with @Step. Steps are discovered from app/steps (config stepsPath, which also takes several directories, plus the @adonis-agora/durable/hooks/steps assembler barrel) and served by name automatically. Pass the method reference to ctx.step for full input/output types:

import { Step } from '@adonis-agora/durable'
import { z } from 'zod'

export default class PaymentSteps {
  @Step({
    name: 'payments:charge-card',
    input: z.object({ orderId: z.number().int(), amountCents: z.number().int() }),
    output: z.object({ chargeId: z.string() }),
    retries: 3,
  })
  async chargeCard(input: { orderId: number; amountCents: number }) {
    return { chargeId: await stripe.charge(input) }
  }
}

// in the workflow (with the step class injected as `this.payments`):
const charge = await ctx.step(this.payments.chargeCard, { orderId: order.id, amountCents: order.total })

Bare @Step() derives the routing name as `${ClassName}.${method}` (refactor-safe); @Step('custom:name') overrides it; the object form adds opt-in input/output schemas (any Standard Schema — zod, valibot, arktype, …) and a def-level retry/liveness policy.

3. defineStepCopy Anchor Link

defineStep(name, fn, config?) builds a typed step handler without a class — the direct swap for a function-form workflow. It returns a ref you pass to ctx.step, and it is discoverable/servable by name:

import { defineStep } from '@adonis-agora/durable'
import { z } from 'zod'

export const chargeCard = defineStep(
  'payments:charge-card',
  async (input: { orderId: number; amountCents: number }) => ({ chargeId: await stripe.charge(input) }),
  {
    input: z.object({ orderId: z.number().int(), amountCents: z.number().int() }),
    output: z.object({ chargeId: z.string() }),
    retries: 3,
  },
)

// in the workflow:
const charge = await ctx.step(chargeCard, { orderId: order.id, amountCents: order.total })

Step schemas are optional — a bare @Step() or defineStep(name, fn) carries none, and the engine skips validation when they're absent. When present, input is validated before the handler runs and output before its result is returned (at the serve boundary). The handler receives the input schema's output value (parsed / coerced / defaulted), and the caller receives the output schema's output value.

Validation with any Standard Schema libraryCopy Anchor Link

Every schema slot — @Step({ input, output }), defineStep, a workflow's inputSchema and its typed searchAttributes — takes any Standard Schema: zod (3.24+ / 4), valibot, arktype, effect schema, and every other library that implements the ~standard interface. The package depends on none of them; bring the one your app already uses.

import { defineStep } from '@adonis-agora/durable'
import * as v from 'valibot'

export const chargeCard = defineStep(
  'payments:charge-card',
  // `input` is inferred from the schema: { orderId: number; amountCents: number }
  async (input) => ({ chargeId: await stripe.charge(input) }),
  {
    input: v.object({ orderId: v.pipe(v.number(), v.integer()), amountCents: v.number() }),
    retries: 3,
  },
)

• Types are inferred from the schema when the handler is unannotated: the handler's argument is the schema's output type, and the returned ref takes the schema's input type (so ctx.step(chargeCard, …) checks the call site). An annotated handler keeps its own types, exactly as before.
• Sync and async validators both work (a schema whose validate returns a Promise is awaited).
• A failure is not retried. It throws a SchemaValidationError — a FatalError with code invalid_input or invalid_output and the issues in details.issues ([{ message, path }]) — so the step fails once and the run fails (or routes via onError: [{ code: 'invalid_input', workflow }], or code: SchemaValidationError for both). Retrying cannot fix a payload that doesn't match its schema.

Dispatch optionsCopy Anchor Link

ctx.step accepts a third StepDispatchOpts argument to route the call and override the step's declared policy field-by-field: { queue, priority, fairnessKey, transport, requires, retries, backoff, backoffMs, backoffMaxMs, jitter, timeoutMs, pickupTimeoutMs }.

await ctx.step(chargeCard, input, { queue: 'payments', priority: 10, retries: 5 })

See Retries & timeouts for the durable-retry, timeoutMs and pickupTimeoutMs liveness semantics, Flow control for queues, and Transports for pinning transport (to the selected transport or one listed in transportPool).

requires — refuse to dispatch into a queue nobody servesCopy Anchor Link

requires names the capabilities a live worker must advertise to run this call:

await ctx.step('billing.charge', input, { requires: ['saga'] })

If no live worker advertises every name — or none is protocol-compatible — the run parks as blocked with a precise reason (which capability is missing, how many live workers were considered) instead of dispatching into a queue that will never be drained. Nothing is thrown into the workflow body; the run waits for a capable worker to appear and is visible as blocked in the dashboard.

Omit it and the step runs anywhere, which is the default. A per-call requires replaces the one declared on the @Step — it does not merge with it. See capability-aware routing.

Serving a step handlerCopy Anchor Link

Steps under app/steps are discovered and served by name automatically (the convention mirrors app/workflows). The low-level escape hatch is transport.handle(name, fn) — register a handler on a worker-side transport under the same name a ctx.step dispatches to:

// worker side — the transport serving this name
transport.handle('payments:charge-card', async (input: { orderId: number; amountCents: number }) => {
  return { chargeId: await stripe.charge(input) }
})

Routing is by name: a step's queue/wire token is derived from its (sanitized) name, suffixed <name>@<partition> when the run belongs to a non-default namespace or the transport sets a partition (see Tenancy). The transports' old group option is deprecated: it is accepted and ignored, so use partition for isolation.

Local (in-process) stepsCopy Anchor Link

ctx.localStep(name, fn, opts?) runs a unit of work in-process: the body executes once right here in the engine, then its result is checkpointed and replayed — never dispatched to a worker. Reach for it for cheap in-process work, or as an escape hatch when a dispatched step is overkill.

const quote = await ctx.localStep('quote', () => pricing.fetch(order), { retries: 3 })

The body receives a step logger as its argument, so you can emit annotations from inside it. It supports the same StepOptions as a dispatched step (retries, backoff, timeout, and compensate for in-process sagas) — see Retries & backoff.

What gets persistedCopy Anchor Link

A local step is as durable as a dispatched one: when the body settles, the engine writes a checkpoint to the store — with stores.lucid(), a row in your database — holding the step's name, position, output (or error), attempts, timings and logger events. On every later pass, a completed checkpoint at that position is returned without running the body again, so a run that resumes after a ctx.sleep('3d') or a deploy reads the result from the database.

Three consequences follow:

• A crash mid-body re-runs it. Nothing is recorded until the body returns, so a process that dies inside it replays the step from scratch. Delivery is at-least-once: make side effects idempotent, e.g. key a charge on `${ctx.runId}:charge` so a re-run is deduplicated by the provider.
• The output must survive JSON. The Lucid store serializes it with JSON.stringify. On the first pass the step hands back the value the body returned; on a replay it hands back the parsed copy — a Date comes back as an ISO string, a Map/Set as {}, a class instance as a plain object, undefined fields are dropped. Return plain data (convert dates with toISOString() and parse them where you use them) so both passes see the same shape. Payloads can be encrypted at rest with a CodecStateStore.
• Position and name identify the checkpoint. Renaming or reordering local steps while runs are in flight raises a NonDeterminismError on replay — see Versioning.

In-flight visibilityCopy Anchor Link

When a local step's body begins, the engine emits a step.started lifecycle event and (by default) writes a running checkpoint, so a long step shows up in the dashboard the moment it starts — not only once it finishes. The running checkpoint is a placeholder overwritten by the step's completed/failed result; it never short-circuits replay (only a completed checkpoint does), so a crash mid-body simply re-runs the step.

Toggle it in config/durable.ts:

export default defineConfig({
  trackStepStart: false, // default true
})

The step.started event fires either way; the flag only gates the extra checkpoint write. Set it to false on hot paths with many short local steps to halve their checkpoint writes, at the cost of in-flight visibility that survives a page reload.

Singleton workflows: one run per keyCopy Anchor Link

Some workflows must not run concurrently for the same subject. Two syncs for one shop racing each other produce interleaved writes; two "recompute this tenant's billing" runs at once double-charge. A singleton workflow serializes runs by a key derived from the input:

export default class SyncShopWorkflow extends BaseWorkflow {
  static workflow = {
    name: 'sync-shop',
    version: '1',
    singleton: {
      key: (input) => (input as { shopId: string }).shopId,
      limit: 1,
      maxQueueDepth: 20,
    },
  }

  async run(ctx: WorkflowCtx, input: { shopId: string }) {
    /* ... */
  }
}

A run whose key is already at its limit (default 1) is created, parked as suspended, and admitted in creation order the moment a slot frees. maxQueueDepth caps the backlog: a start beyond limit + maxQueueDepth is rejected with SingletonQueueFullError before any run is created. See Singleton workflows for admission, notify-on-release, back-pressure and how it differs from a deterministic runId.

Fan-out (parallel steps)Copy Anchor Link

Run steps concurrently with Promise.all — checkpoints stay deterministic because each step's position is taken in the synchronous prefix before any await:

const [a, b] = await Promise.all([
  ctx.localStep('a', () => doA()),
  ctx.localStep('b', () => doB()),
])

Promise.all is right for a handful of different steps. For the same dispatched step over many inputs, pass the step reference to ctx.all instead. It runs one durable step per input and returns the outputs in input order:

const results = await ctx.all(processLine, lines, { concurrency: 500, mode: 'failFast' })

Every result of a Promise.all replays the workflow, so a fan of N steps costs O(N²) checkpoint reads. ctx.all puts a join barrier in front of the workflow instead: a result resumes it only once no step of the fan is still pending, or when one fails, so the workflow replays once per settled wave. concurrency sends the steps in waves of that size, and mode is waitAll (the default: every step settles, then a GatherError lists the failed indexes) or failFast (throws on the first failed step). ctx.allSettled(step, inputs, opts?) is the same fan but never throws for a failed item: it resolves one Settled per input. A step is much cheaper than a child workflow per item: a checkpoint and a dispatch instead of a run row, a turn, a lease and a parent signal. See fanning out over a step for the full rules.

Long-running loops — ctx.continueAsNewCopy Anchor Link

A workflow that loops forever — polling for new work, or a long-lived accumulator parked on ctx.waitForSignal — writes checkpoints on every turn. Left unbounded, that history grows forever and every replay gets slower re-reading it. ctx.continueAsNew(input?) is the reset: it completes the current run and hands off to a fresh execution of the same workflow with a clean history, under a traceable id <runId> → <runId>~1 → <runId>~2 …

export default class PollInboxWorkflow extends BaseWorkflow {
  static workflow = { name: 'poll-inbox', version: '1' }

  async run(ctx: WorkflowCtx, input: { mailbox: string; cursor?: string }) {
    const { messages, nextCursor } = await ctx.step(fetchSince, input)
    for (const message of messages) await ProcessMessageWorkflow.dispatch(message)
    await ctx.sleep('30s')
    // End this run and start fresh with the new cursor — history resets instead of growing forever.
    return ctx.continueAsNew({ mailbox: input.mailbox, cursor: nextCursor })
  }
}

It is terminal — it always throws internally, so code after it never runs. The current run completes with no output, and whatever the next iteration needs must be carried forward explicitly through input. Call it outside any try/catch of your own, or rethrow it with isWorkflowControlFlowSignal.

See Continue-as-new for the crash-safe handoff and namespace inheritance.

Fatal errorsCopy Anchor Link

Any thrown error is retried up to the step's limit. To stop retrying a business failure that a retry can't fix, throw FatalError — it fails the run immediately:

import { FatalError } from '@adonis-agora/durable'

await ctx.localStep('charge', async () => {
  const res = await stripe.charge(order)
  if (res.declined) throw new FatalError('card declined', 'declined')
  return res
})

The optional second argument is a machine-readable code that ends up on the run's structured error.

Catching errors — rethrow control-flow signalsCopy Anchor Link

If you wrap a step in your own try/catch inside the workflow body — to run a cleanup or compensating path on failure — you must rethrow control-flow signals untouched. Suspend-the-run operations (ctx.sleep, ctx.waitForSignal, ctx.continueAsNew) unwind the current turn by throwing: they are not real failures, and running a failure path on one records extra commands into history that a later replay never produces — which surfaces as a NonDeterminismError when the run resumes.

Use isWorkflowControlFlowSignal(error) to let those through before treating an error as a genuine failure:

import { isWorkflowControlFlowSignal } from '@adonis-agora/durable'

try {
  await ctx.step('chargeCard', input)
} catch (error) {
  if (isWorkflowControlFlowSignal(error)) throw error // suspend / continue-as-new: rethrow as-is
  await ctx.step('refund', input) // a REAL failure — safe to compensate here
  throw error
}

The predicate is true for the engine's control-flow signals (currently a suspend and a continue-as-new) regardless of which module instance threw them — it checks a stamped marker (CONTROL_FLOW_SIGNAL), not instanceof, so it stays correct across a duplicated module copy. It is deliberately false for a cancelled run (a terminal outcome you may legitimately want to observe) and for a FatalError or a step rejection (real failures your catch is meant to handle) — so this guard never swallows a genuine error. Most of the time you don't need it: prefer the built-in saga compensation over hand-rolled try/catch, and reach for isWorkflowControlFlowSignal only when you truly must catch inside the workflow body.

Steps vs. sub-process eventsCopy Anchor Link

These look similar in the dashboard but are fundamentally different — the distinction is durability:

• A step (ctx.step or ctx.localStep) is a durable checkpoint. The engine records its result, so on a crash, retry, or replay it is not re-executed — the saved result is replayed. It's a first-class node in the run graph and the unit of recovery.

• A sub-process event is a log annotation emitted inside a step via the step logger's log.sub(name, status) — e.g. one entry per item in a fan-out the step performs internally. It is not durable on its own: it's metadata attached to the parent step's checkpoint. If the parent step retries, all of its sub-process events are produced again. Use it for visibility, not recovery.

// One durable step that internally processes N items and records each outcome for visibility.
await ctx.localStep('process-batch', async (log) => {
  for (const item of batch) {
    try {
      await handle(item)
      log.sub(item.id, 'ok') // a sub-process event — shown under the step, not a checkpoint
    } catch (e) {
      log.sub(item.id, 'failed', String(e))
    }
  }
})

The step logger also offers log.debug/info/warn/error(message, data?) for leveled logs, log.subProcess(name, body) for nested, phased sub-work, and log.subEvent({ ... }) for a fully-specified sub-process entry (its own id, group, phase, status, message and data) when log.sub's three arguments are not enough. Rule of thumb: if you want each unit to retry or replay independently, make it a step (or a child workflow). If you only want to see what happened inside one durable unit, emit sub-process events.

Emitting from deep inside a stepCopy Anchor Link

The logger argument only reaches the step body itself. When the code that knows the progress lives a few calls down (a shared batch inserter, an HTTP client, a parser), currentStep() returns the logger of the step running on the current async path, so you don't have to thread it through every signature:

import { currentStep } from '@adonis-agora/durable'

// A generic utility: it takes no logger, and its callers were not edited.
export async function insertRows(rows: Row[]) {
  for (const chunk of chunks(rows, 100)) {
    await db.table('rows').multiInsert(chunk)
    currentStep()?.info('inserted batch', { rows: chunk.length })
  }
}

It is an AsyncLocalStorage, installed around every step body, local and dispatched alike, with the same logger instance the body receives, so ambient and explicit events land in the same list. Concurrent steps each see their own. Outside a step it returns undefined, so the optional chain keeps the helper usable on a non-durable path and in unit tests.

Reporting liveness from inside a long stepCopy Anchor Link

A step that runs for minutes has a problem the logger cannot solve: from the engine's side, "working hard" and "the worker died" look identical. log.heartbeat(progress?) is the signal that separates them:

@Step()
async reindexCatalog(input: { shopId: string }, log: StepLogger) {
  const pages = await catalog.pageCount(input.shopId)

  for (let page = 1; page <= pages; page += 1) {
    await catalog.reindexPage(input.shopId, page)
    log.heartbeat({ page, pages })
  }

  return { pages }
}

Each beat does three things:

• It rearms the step's timeoutMs. That window means "maximum silence", not "maximum duration" — so a step that beats every few seconds can run for an hour under a timeoutMs: 30_000 and never time out, while a worker that dies is caught within the window. See Retries & timeouts.
• It persists the progress payload on the step's checkpoint, so durable:runs and the dashboard can show where a long step is, not just that it is running.
• It costs almost nothing. Beats are throttled — at most one every 5 seconds leaves the worker, and the engine persists at most one every 10 seconds per step, with an exemption so the first beat carrying a progress payload always lands. Call it as often as is convenient; the throttles do the rationing.

Beat from a dispatched step, where there is a worker whose silence is meaningful. Inside ctx.localStep the body runs in the engine itself, so there is nothing to report liveness to and log.heartbeat is a no-op.

Step interceptorsCopy Anchor Link

For cross-cutting concerns around step execution — timing, structured logging, tracing, error enrichment — register an onion-style interceptor instead of repeating the same wrapper in every step body. engine.use(interceptor) takes (invocation, next) => Promise<unknown>: call next() to run the step body (or the next interceptor) and return — or transform — its result; throw to fail the step. First registered runs outermost, and use returns a function that removes the interceptor.

import engine from '@adonis-agora/durable/services/main'
import logger from '@adonisjs/core/services/logger'

engine.use(async (invocation, next) => {
  const startedAt = performance.now()
  try {
    return await next()
  } finally {
    logger.info(
      { workflow: invocation.workflow, step: invocation.stepName, attempt: invocation.attempt },
      `step took ${performance.now() - startedAt}ms`,
    )
  }
})

invocation carries runId, workflow, stepName, seq and the 1-based attempt.

Deterministic sources — ctx.now & ctx.sideEffectCopy Anchor Link

Because the body replays, never read a value that changes on each run. The context exposes deterministic, checkpointed sources — each records its value the first time and replays it afterwards:

const issuedAt = await ctx.now()                              // epoch ms — captured once, replayed verbatim
const nonce = await ctx.sideEffect(() => crypto.randomUUID()) // any generated value, captured once
const sampled = (await ctx.sideEffect(() => Math.random())) < 0.1

ctx.now() is the deterministic wall-clock. ctx.sideEffect(fn) is the general deterministic capture: it runs fn once, checkpoints the result, and returns that same value on every replay without re-running fn — wrap any non-determinism you control (a UUID/ULID, Math.random(), a config/env read) in it. For real work with side effects — a DB write, an API call — use a step instead, since sideEffect's fn must be effectively pure (it runs only once and is not re-executed on replay).

See Versioning & determinism for why and how the runtime guard works.

TagsCopy Anchor Link

Label runs with tags to find them later. Static tags on register apply to every run; per-run tags are added at start. Both are merged onto the run and are searchable in the dashboard.

engine.register('pipeline', '1', async (ctx) => { /* ... */ }, { tags: ['etl', 'critical'] })

// per-run tags merge with the static ones → run.tags = ['etl', 'critical', 'nightly']
await engine.start('pipeline', input, runId, { tags: ['nightly'] })

The dashboard shows a run's tags and a tag filter. Programmatically, query by tag with RunQuery.tag (engine.listRuns({ tag: 'etl' })) or the dashboard API's ?tag= param.

Search attributesCopy Anchor Link

Tags are a flat label; search attributes are typed, queryable key/value metadata on a run — an amount, a tier, a customerId — for the range and equality filters a tag can't express. Set them from inside the workflow with ctx.upsertSearchAttributes(attrs), where attrs is a Record<string, string | number | boolean>:

await ctx.upsertSearchAttributes({ amount: order.total, tier: order.customer.tier })

It is a shallow merge into the run's searchAttributes (keys you don't pass are kept), and it is durable and exactly-once: recorded at this position on the first run and skipped on replay. Use it instead of reaching for store.updateRun(ctx.runId, …) from a workflow body. Query them with engine.listRuns({ attributes: [{ key: 'amount', op: 'gte', value: 20000 }] }) — eq, ne, gt, gte, lt, lte, and in with a values set, ANDed together. See Search attributes for start-time attributes and the full query surface.

Typed search attributesCopy Anchor Link

Declare the run's attribute shape with a Standard Schema whose output is flat (string / number / boolean values; optional keys allowed) and every ctx.upsertSearchAttributes is checked — at compile time through WorkflowCtx<A>, and at runtime against the merged attributes:

import { BaseWorkflow, attributesOf, type InferSearchAttributes, type WorkflowCtx } from '@adonis-agora/durable'
import { z } from 'zod'

const checkoutAttrs = z.object({ tier: z.enum(['free', 'pro']), amount: z.number().optional() })
type CheckoutAttrs = InferSearchAttributes<typeof checkoutAttrs>

export default class CheckoutWorkflow extends BaseWorkflow {
  static workflow = { name: 'checkout', searchAttributes: checkoutAttrs }

  async run(ctx: WorkflowCtx<CheckoutAttrs>, input: Order) {
    await ctx.upsertSearchAttributes({ tier: input.tier }) // { tier: 'enterprise' } is a type error
  }
}

// Read them back typed — the schema is resolved from the class:
const attrs = attributesOf(CheckoutWorkflow, run) // CheckoutAttrs, or {} for history that predates the schema

A write whose merged result fails the schema throws a SchemaValidationError in the body: nothing is written and nothing is recorded, so the run fails (it is not retried). The schema must validate synchronously — an async refinement could make replay nondeterministic — and readSearchAttributes(schema, run) / attributesOf are deliberately lenient (old history that doesn't match reads as {}). With engine.register, pass the schema as searchAttributesSchema. A bare WorkflowCtx (no type argument) accepts any attribute bag, exactly as before.

Delayed starts — startAtCopy Anchor Link

Start a run now but have it execute later by passing startAt (epoch ms or a Date) to engine.start:

// Created durably right away; the body first executes at 09:00 tomorrow.
await engine.start('send-reminder', { orderId }, runId, { startAt: tomorrowAt9am })

The run is created immediately but parked suspended with its wake timer set to startAt; the ordinary durable-timer poller starts it when due. That makes the delay crash-safe with no ctx.sleep polluting the workflow body or its history — the body's first checkpoint is its first real step, and a fix-and-replay of the run doesn't re-wait the delay. A startAt in the past (or absent) starts immediately. One event-lifecycle nuance: a delayed run skips the run.started pending → running hop and goes suspended → … when it fires.

Run origin — package attributionCopy Anchor Link

When several packages start workflows on one engine — your app plus a couple of @adonis-agora/* libraries — "whose code produced this run?" is worth answering without archaeology. Every run carries an origin: the npm package that declared its workflow. You normally don't set it. The engine derives it from the nearest package.json name above a file that belongs to the workflow, using the first of these that is available:

1. The class's declaring file, when the class carries @Scheduled, @OnEvent or @OnDiagnostic. The decorator records where the class was defined, so the origin follows the class no matter who registers it.
2. The discovered module's file. app/workflows/charge_workflow.ts resolves to your app's name. A built app resolves the same way, because build/ carries a copy of the app's package.json. The generated barrel resolves from the app root.
3. The file calling registerWorkflowClass(engine, Cls) or engine.register(...). A library that registers its own workflows is attributed to the library. In a pnpm layout, node_modules/.pnpm/…/node_modules/@scope/lib/dist/x.js resolves to @scope/lib.

An explicit origin always wins over the derived one. It is accepted at three levels, and the most specific wins:

// On the class — every run of this workflow:
static workflow = { name: 'sync-catalog', version: '1', origin: '@acme/catalog-pipeline' }

// On a manual registration:
engine.register('sync-catalog', '1', fn, { origin: '@acme/catalog-pipeline' })

// On one start — overriding the registration, for a shared workflow started on another package's behalf:
await engine.start('sync-catalog', input, runId, { origin: '@acme/billing' })

Set it explicitly when the derived answer would be wrong. The usual case is an app that hand-registers an undecorated class shipped by a library: signal 3 credits the app with it.

Derivation never guesses. When nothing resolves, the run has no origin. That happens when the runtime gives no usable stack frame (a non-V8 runtime, Error.stackTraceLimit = 0), when no package.json with a name sits above the file, and for registerRemote registrations. A single-file bundle that folds several packages into one file is attributed to the bundle's package; that one case can't be detected. At boot the provider logs, once, the discovered workflows it couldn't attribute.

The origin lands on run.origin, is filterable (engine.listRuns({ origin: '@acme/catalog-pipeline' }), the dashboard API's ?origin= param), and powers the console's origin facet. Runs with no origin are grouped under app there, and the facet only appears once some run has a named origin. With derivation, those are mostly runs whose package couldn't be derived, plus runs created before the field existed. It's attribution, not routing: use namespace to pin where a run executes.