app/workflows & make:workflow

The way to author a workflow is a class under app/workflows/ that extends BaseWorkflow and declares its identity with a static workflow = { name, version }, mirroring how @adonisjs/queue authors a job per file under app/jobs/. The durable provider scans app/workflows/ at boot and registers every exported workflow class on the engine automatically — you never call engine.register(...) by hand.

Scaffold a workflowCopy Anchor Link

node ace make:workflow order

This emits app/workflows/order_workflow.ts. The registered name is the snake-cased argument with any workflow suffix removed (order), and a path argument nests the file (node ace make:workflow billing/refund writes app/workflows/billing/refund_workflow.ts):

import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'

interface OrderInput {
  // Define your workflow input here
}

export default class OrderWorkflow extends BaseWorkflow {
  static workflow = { name: 'order', version: '1' }

  // Optional: run this workflow on a recurring schedule (colocated with the class), fired by the
  // durable:work worker. Cron needs the optional cron-parser peer dependency.
  // static schedule = { cron: '0 4 * * *', timezone: 'America/Sao_Paulo' }

  async run(ctx: WorkflowCtx, input: OrderInput) {
    // Your workflow logic here — compose ctx.step / ctx.localStep / ctx.child / ctx.sleep …
    return input
  }
}

The class's run(ctx, input) is the workflow body — the same deterministic (ctx, input) => … that engine.register takes, just attached to a class so a class ref (OrderWorkflow.dispatch(input), ctx.child(OrderWorkflow)) carries its input/output types.

The static workflow optionsCopy Anchor Link

static workflow is a WorkflowOptions object. Only name is required:

Discovery forwards each of these to engine.register. Two names differ between the two forms: searchAttributes here is searchAttributesSchema on engine.register. And three engine.register options have no static workflow equivalent: validateInput (a function-form validator), eventBatch (debounce/batch coalescing for onEvent) and requires (worker capabilities). A workflow that needs one of those is registered by hand with engine.register.

Validating the inputCopy Anchor Link

Declare an inputSchema — any Standard Schema (zod, valibot, arktype, …) — and every start is validated before a run is created:

import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'
import { type } from 'arktype'

const OrderInput = type({ orderId: 'string', amount: 'number > 0' })

export default class OrderWorkflow extends BaseWorkflow {
  static workflow = { name: 'order', version: '1', inputSchema: OrderInput }

  async run(ctx: WorkflowCtx, input: typeof OrderInput.infer) {
    // …
  }
}

A bad payload makes start / dispatch throw a SchemaValidationError (code invalid_input, issues in details.issues) and nothing is persisted; an onEvent subscriber whose schema rejects the payload simply doesn't start. The same option exists on engine.register(name, version, fn, { inputSchema }) and engine.registerRemote(...), next to the function-form validateInput — when both are set the schema runs first and validateInput sees the parsed value.

The run stores what was sent; run receives what was parsed. The run's stored input is the payload exactly as the caller sent it. run(ctx, input) receives the schema's output value (parsed / coerced / defaulted), re-parsed from the stored input before every turn, replays included. That is what makes a value JSON can't carry — a Date from z.coerce.date() — reach run as a Date on every turn: the SQL stores keep the input in a JSON column, so a stored parsed value would come back as a string. Two rules follow:

• The schema must be deterministic. No default(() => Date.now()), no random ids: a replay must parse the same value the first execution did.
• Changing the schema is a breaking change to the workflow — bump version, as for any change to the body. An in-flight run whose stored input no longer parses fails with a SchemaValidationError (invalid_input).

A remote workflow's body gets the parsed value too: the engine re-parses the stored input each turn and puts the result in the task. The task travels as JSON, so there coercions and defaults apply but a non-JSON value arrives as its JSON form.

Singleton and concurrency keys (singleton.key(input), concurrency.key(input)) are computed from the stored input — the value the singleton gate re-reads when it re-checks admission later.

Auto-registrationCopy Anchor Link

At boot the provider scans the configured directory (default app/workflows), imports each module, and registers every exported workflow class (default and named exports) from its static workflow config, with every option in the table above, plus any colocated schedule and any colocated event trigger. The scan is recursive, so sub-directories of app/workflows are picked up too. A class without a static workflow (or whose name is not a string) is skipped. When the codegen barrel exists, the provider imports it instead of scanning the directory at runtime.

Dependency injection in the constructorCopy Anchor Link

Workflow classes are instantiated through the AdonisJS container, exactly like an @adonisjs/queue job. Decorate the class with @inject() and declare what it needs as constructor parameters — the container resolves them:

import { inject } from '@adonisjs/core'
import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'
import BillingSteps from '#steps/billing_steps'
import NotificationSteps from '#steps/notification_steps'

interface OrderInput {
  orderId: string
  amount: number
}

@inject()
export default class OrderWorkflow extends BaseWorkflow {
  static workflow = { name: 'order', version: '1' }

  constructor(
    private billing: BillingSteps,
    private notifications: NotificationSteps,
  ) {
    super()
  }

  async run(ctx: WorkflowCtx, input: OrderInput) {
    const charge = await ctx.step(this.billing.charge, {
      orderId: input.orderId,
      amount: input.amount,
    })

    await ctx.step(this.notifications.orderConfirmed, {
      orderId: input.orderId,
      receiptUrl: charge.receiptUrl,
    })

    return { charged: charge.id }
  }
}

Two things follow from this, and they are the reason to prefer it over reaching for the container inside run:

• The step reference is typed. this.billing.charge is a @Step-decorated method, so ctx.step infers both its input and its output — no generic to write, and renaming the method is a compile error rather than a runtime "no handler for …".
• The body stays a pure composition. Resolving services is construction, not workflow logic, so the deterministic run reads as the sequence of steps it is.

The instance is built once, at registration, and reused for every run — so treat the constructor as wiring and keep per-run state in the workflow's input and step outputs, never on this.

A class with no constructor dependencies is unaffected: the container simply builds it with no arguments.

Injecting a step class only gives the workflow a typed reference to its methods. The step bodies are served by whatever registers them, which is stepsPath discovery (default app/steps) on the process running the worker, so a @Step class has to live there to run.

That discovery resolves each @Step class through the container too, so a step class can declare its own @inject() constructor dependencies and its step bodies get them:

import { inject } from '@adonisjs/core'
import { Step } from '@adonis-agora/durable'
import StripeService from '#services/stripe_service'

@inject()
export default class BillingSteps {
  constructor(private stripe: StripeService) {}

  @Step('billing:charge')
  async charge(input: { orderId: string; amount: number }) {
    return this.stripe.charge(input)
  }
}

The provider (embedded worker) and node ace durable:worker (a tenant pod) both serve it this way. Like an @adonisjs/queue job, the instance is resolved on every step execution, not once at boot: a service the container doesn't bind as a singleton is fresh per execution, and an app.container.swap(StripeService, () => fake) made in a test after boot is what the next execution receives. It's never the instance injected into the workflow.

Running a workflowCopy Anchor Link

The ergonomic way to run or compose a registered class is its context-aware statics, Wf.dispatch and Wf.start:

// fire-and-forget a top-level run from a controller/service — returns { runId }
const { runId } = await OrderWorkflow.dispatch(input)

// block until the run settles and return its result (scripts / short runs)
const result = await OrderWorkflow.start(input)

Both are context-aware: called inside a running workflow body they route to a child instead of the engine. See Child workflows for the full semantics table.

The second argument (WorkflowDispatchOptions) is an optional runId plus the engine's StartOptions:

await OrderWorkflow.dispatch(input, {
  runId: `order:${input.orderId}`, // idempotency key; omitted → a random UUID
  version: '2',                    // pin a registered version; omitted → the newest
  startAt: tomorrow,               // create the run now, start it later (epoch ms or Date)
  tags: ['vip'],                   // merged with static workflow.tags
  searchAttributes: { amount: input.amount },
  priority: 10,                    // dispatch priority for a remote run; higher wins
  namespace: 'tenant-a',           // worker-pool partition for this run
})

origin and a per-start concurrency quota are accepted too. A version that is not registered throws before any run is created; it never falls back to the newest.

Configure or disable discovery in config/durable.ts:

export default defineConfig({
  workflowsPath: 'app/workflows', // default; set false to disable and register by hand
})

Several directories (modular apps)Copy Anchor Link

An app split by bounded context (app/orders/flows, app/billing/flows, …) lists every directory. stepsPath takes an array the same way:

export default defineConfig({
  workflowsPath: ['app/orders/flows', 'app/billing/flows'],
  stepsPath: ['app/orders/steps', 'app/billing/steps'],
})

Every directory is scanned, in order. An empty array is the same as false.

Names must be unique across the directories. If two directories define the same workflow name and version, or the same step name, with different classes or handlers, boot fails with a DuplicateDiscoveryError that names both files. The same class reached from two directories (a re-export, or overlapping directories) is not a clash and registers once. Inside a single directory nothing changed: a duplicate there is not checked, and the later registration wins.

If you use the codegen barrels, give the hooks the same directories through their source option. See Several directories.

Registering an extra directory by handCopy Anchor Link

To register a directory from a preload, call registerWorkflowsFromDir for workflows and registerStepsFromDir for steps. Pass the engine to both. For steps, the engine forwards each handler to its transport through engine.handle(name, fn), so you don't need the transport the provider keeps private:

import app from '@adonisjs/core/services/app'
import engine from '@adonis-agora/durable/services/main'
import { registerStepsFromDir, registerWorkflowsFromDir } from '@adonis-agora/durable'

await registerWorkflowsFromDir(engine, app.makePath('app/engine/flows'), (Ctor) =>
  app.container.make(Ctor),
)
await registerStepsFromDir(engine, app.makePath('app/engine/steps'), (Ctor) =>
  app.container.make(Ctor),
)

Both accept one directory or an array, and apply the same cross-directory duplicate check. Both also take an optional third argument, the factory that builds each class (for steps, each class with @Step methods, called on every step execution; a defineStep handler is served as-is). Without it the class is built with new Ctor(), once, so pass (Ctor) => app.container.make(Ctor) as above when the classes use constructor injection. It returns the metadata of every workflow it registered. For a single class, registerWorkflowClass(engine, Cls, factory?) does the same for one class. engine.handle throws if the engine's transport can't serve handlers. A tenant pod has no engine: node ace durable:worker serves stepsPath there.

From a queue job to a durable workflowCopy Anchor Link

The convention is a deliberate parallel, so switching a @adonisjs/queue job to a durable workflow is a mechanical move — app/jobs/<name>.ts → app/workflows/<name>_workflow.ts, make:job → make:workflow, Job.execute() → Workflow.run(ctx, input), Job.dispatch() → Workflow.dispatch(input) — and you gain durable checkpoints, replay, timers, signals and saga compensation on top.