Python

Python is a first-class durable runtime, not just a place to run a step. The durable-worker SDK (the same one the aviary/nestjs-durable ecosystem ships) speaks the wire your Adonis engine speaks byte-for-byte, so one Python worker can do either — or both:

• Implement steps an AdonisJS workflow calls — a ctx.step('payments.charge-card', …) reaches a handler written in Python.
• Author whole workflows in Python — the coordinator logic (ctx.step, ctx.sleep, ctx.gather_calls, …) lives in Python, and the Adonis engine drives it turn by turn.

Either way the engine stays the sole owner of durable state, recovery, and timers — the Python worker is a stateless consumer on the broker. Everything rides the same Redis queues the BullMQ transport uses.

pip install "durable-worker[redis]"

A Python step workerCopy Anchor Link

import asyncio
from durable_worker import Worker, FatalError
from durable_worker.redis_runner import run_redis_worker

worker = Worker()

@worker.step("payments.charge-card")
async def charge(data):
    res = await stripe.charge(data["orderId"], data["amountCents"])
    return {"chargeId": res.id}

async def main():
    await run_redis_worker(worker, connection="redis://localhost:6379")
    await asyncio.Event().wait()

asyncio.run(main())

The handler's argument is the step input (schema-validated by the engine); its return value is the output. Raise FatalError for a non-retryable failure (e.g. a declined card); any other exception is retryable and the engine applies the step's retry policy.

Connecting to the control planeCopy Anchor Link

There is no direct connection between the Python worker and your AdonisJS app — everything goes through the broker (Redis). Point the worker at the same Redis your engine's bullmq transport uses, and register a handler under the exact same name. Two things have to line up:

import { defineConfig, transports, stores } from '@adonis-agora/durable'

export default defineConfig({
  role: 'control-plane',
  transport: 'bullmq',
  transports: { bullmq: transports.bullmq({ connection: { host: '127.0.0.1', port: 6379 } }) },
  store: 'lucid',
  stores: { lucid: stores.lucid({ connection: 'pg' }) },
})

Each registered step name gets its own queue — no separate group to declare. Sharing one broker across tenants? The engine routes a step to <name>@<tenant>, where the tenant is the run's namespace (or the step's own partition, which wins); a default or unset namespace routes to the bare <name>. Start the Python worker with run_redis_worker(worker, partition="acme") to consume the acme tenant's queues. transports.bullmq({ partition }) on the Adonis side only decides which queues that Adonis process itself consumes; it does not change where the engine dispatches. See Tenancy.

The engine and the workers are decoupled producers and consumers on per-name queues:

  Adonis engine                              Redis (the broker)                    Python worker
  ─────────────                              ──────────────────                    ─────────────
  ctx.step(name)   ───dispatch───▶  durable-tasks-<name>[@<partition>]  ───consume──▶  @worker.step(name)
  resume ctx.step   ◀──consume────  durable-results                     ◀───add──────  return / raise

Match the connection, the prefix and the name, and a ctx.step('payments.charge-card', …) is delivered to whichever worker registered payments.charge-card — the queue name is the routing. A few properties follow from this being queue-based:

• At-least-once delivery. A task can be redelivered (a worker crash, a reconnect), so handlers should be idempotent — key external side effects on stepId (<runId>:<seq>) or your own dedupe key. The engine already dedupes a step that completed, but a handler can be re-run before its result is recorded.
• Workers are stateless and horizontal. Run as many copies as you want for a given step; Redis hands each task to one of them. Scale a slow step by adding workers, not by touching the workflow.
• The engine owns correlation. Every result carries the stepId it answers; the engine matches it to the suspended ctx.step and resumes the run. The worker never needs the run's state.
• A killed worker's job is reported. A process killed mid-step (an OOM kill) produces no result. The durable-worker runner, like the Adonis bullmq transport, listens for BullMQ's terminal failed event on its task workers, so when a peer's stalled-check fails the dead worker's job, a live replica publishes a synthetic retryable failed result and the engine's durable retry re-dispatches the step. If the only worker for a queue dies and never restarts, no stalled-check runs at all — that case needs the engine's remoteRedispatchMs self-heal (see lost remote dispatch).

AuthenticationCopy Anchor Link

The connection is a standard Redis URL, so credentials and TLS go in the URL:

await run_redis_worker(
    worker,
    connection="rediss://default:my-password@my-redis.example.com:6379",
)

Use redis:// for plaintext and rediss:// for TLS. This is the same Redis the engine's transports.bullmq({ connection }) points at — the worker uses the same credentials. The engine side accepts the same URL form: transports.bullmq({ connection: process.env.REDIS_URL! }) parses it into ioredis options (rediss: turns on TLS, the path's first segment selects the database).

run_redis_worker optionsCopy Anchor Link

Concurrency and memoryCopy Anchor Link

concurrency takes a fixed int, the string "adaptive", or a dict of overrides. An adaptive Python worker tunes its own limit from a latency gradient and, since durable-worker 0.25, protects itself from OOM with a memory admission gate: before each step starts, the worker asks whether there is room for one more right now and waits rather than starting a step it cannot fit. (This is the Python worker's own controller; the Adonis-side worker controller is transports.bullmq({ concurrency }).)

worker = Worker(concurrency="adaptive")                  # gate on, all defaults
worker = Worker(concurrency={"min": 1, "max": 16})       # adaptive, with bounds
worker = Worker(concurrency={"ramAdmission": False})     # gate off (pre-0.25 behaviour)

Memory usage is read as current usage: cgroup v2 memory.current, then cgroup v1 memory.usage_in_bytes, then /proc/self/status VmRSS, and only as a last resort ru_maxrss. Reclaimable page cache (memory.stat inactive_file) is subtracted, giving the same working set kubelet evicts on.

• A deferred step is not a failed step — it keeps its job (BullMQ renews the lock) and starts as soon as there is headroom.
• With nothing in flight the gate always admits, so a worker can never deadlock waiting for memory that only it could free.
• Crossing ramCeilingPct drops the limit to min in one move; growing back requires growHeadroomTicks consecutive ticks of real headroom.
• Per-step-name costs are learned from steps that ran alone and reserved for the next one, because usage at admission time doesn't yet include what an admitted step is about to allocate. The first task of a process runs alone so there is always one real measurement.

A fixed concurrency=N worker is never gated — N is your explicit promise — but it still publishes the same memory numbers on its heartbeat, which the dashboard's worker cards show. Adaptive protects that pod: a shared dependency still needs a global cap (see flow control).

The wire protocolCopy Anchor Link

The contract is plain JSON, identical across languages. The engine dispatches a task:

{
  "runId": "wrun_8Kb2", "seq": 1, "name": "payments.charge-card",
  "stepId": "wrun_8Kb2:1", "group": "payments",
  "input": { "orderId": "o1", "amountCents": 4200 }, "attempt": 1
}

The worker replies with a result:

{ "runId": "wrun_8Kb2", "seq": 1, "stepId": "wrun_8Kb2:1",
  "status": "completed", "output": { "chargeId": "ch_1" },
  "startedAt": 1718900000123 }

A task can also carry transport (the id of the transport it was dispatched on), priority, traceparent, and context (the originating request's context snapshot). A failed result carries error: { message, code?, details?, retryable? } instead of output; retryable: false (what FatalError sends) skips the engine's retries.

startedAt is the epoch-ms moment the worker picked the task up. The engine subtracts the dispatch time from it to report queue-wait — how long the step sat in the queue before a worker was free — so a Python step's queue time shows up in the dashboard just like a Node one. The SDK stamps it for you.

Worker.process_task(task) -> result is the pure, transport-agnostic core (no broker needed — ideal for unit tests). Because the contract is language-neutral, a Go or Rust worker can implement the same thing.

Replying on the right brokerCopy Anchor Link

task["transport"] is the id of the transport the engine dispatched on, and a result must go back on that same broker. A run_redis_worker per broker already replies on the broker it consumes, so there is nothing to route. A single Python loop that multiplexes several brokers picks the reply target with durable_worker.routing.reply_target(task, targets), which looks the id up in a dict and falls back to the lone entry when there is one. An Adonis engine built from config/durable.ts uses one transport, so its tasks carry a single id; this matters only for an engine you construct yourself with a transports pool.

Cooperative cancellationCopy Anchor Link

When a run is cancelled — engine.cancel(runId) on the Adonis side — a step already running in a Python worker can't be force-killed from outside the process. Instead the engine broadcasts a control message and the handler bails out on its own at a safe point.

• The engine publishes {"kind": "cancel", "runId": ...} on its control plane. With controlPlanes.redis that is the Redis channel <prefix>-control — the same channel and payload the NestJS transport uses, which is the one durable-worker subscribes to. The bullmq transport does not publish this channel itself, so configure controlPlanes.redis on the same Redis, with the prefix the Python worker uses (durable-<namespace> when the engine sets a namespace), or Python handlers never see the cancel.
• run_redis_worker subscribes to that channel and feeds a CancellationRegistry — a thread-safe set of cancelled run ids. One is created for you if you don't pass your own.
• The registry is surfaced to your handler through the step context as ctx.cancelled (a bool) and ctx.raise_if_cancelled() (raises Cancelled).

import { defineConfig, transports, controlPlanes } from '@adonis-agora/durable'

export default defineConfig({
  role: 'control-plane',
  transport: 'bullmq',
  transports: { bullmq: transports.bullmq({ connection: { host: '127.0.0.1', port: 6379 } }) },
  controlPlane: controlPlanes.redis({ connection: 'main' }), // channel `durable-control`
  // ...
})

Opt in by declaring the second ctx parameter on your handler, then poll it inside the loop:

from durable_worker import Worker, Cancelled

worker = Worker()

@worker.step("pipeline.transcode")
def transcode(data, ctx):
    out = []
    for chunk in data["chunks"]:
        if ctx.cancelled:                         # the run was cancelled — stop early
            ctx.warn("cancelled — stopping early", {"done": len(out)})
            return {"partial": out, "cancelled": True}
        out.append(encode_chunk(chunk))
    return {"chunks": out}

@worker.step("pipeline.crawl")
async def crawl(data, ctx):
    results = []
    for url in data["urls"]:
        ctx.raise_if_cancelled()                  # raises Cancelled the moment the run is cancelled
        results.append(await fetch(url))
    return {"results": results}

Cancelled is a subclass of Exception, so it surfaces as a failed step. To drive a registry yourself (e.g. share one across runners), construct it and pass it in:

from durable_worker.cancellation import CancellationRegistry

cancellation = CancellationRegistry()
await run_redis_worker(worker, connection="redis://localhost:6379", cancellation=cancellation)

Cancellation is cooperative: a handler that never checks ctx.cancelled runs to completion and the engine discards its result. The control-channel subscription is best-effort — it needs the redis async client for pub/sub; if it isn't available the runner logs and continues, steps still run, and cancellation just isn't observed.

Authoring workflows in PythonCopy Anchor Link

A Python worker can own the whole workflow function, not just a step. The Adonis engine stays the sole owner of durable state, recovery, and timers: it advances a run one turn at a time by sending the run's history as a workflow task, the Python runtime replays the workflow function locally, and returns the commands the replay produced (a step dispatch, a sleep, a signal wait) for the engine to persist and act on — a pure, store-free coordinator.

One unified worker (recommended)Copy Anchor Link

The simplest deployment registers the workflow and its step handlers on one Worker. Each registered @worker.workflow/@worker.step name gets its own queue (route-by-handler), so a single process replays the workflow turns and runs the steps it dispatches. run_workers connects it and drains cleanly on SIGTERM/SIGINT:

from durable_worker import Worker, WorkflowContext, redis_url_from_env, run_workers

worker = Worker()

@worker.step("ingestion")
async def ingestion(data):
    return {"rows": load(data["key"])}

@worker.workflow("pipeline")
def pipeline(ctx: WorkflowContext, data):
    started_at = ctx.now()                               # replay-stable capture
    results = ctx.gather_calls([                          # fan out parallel remote steps in this run
        {"name": "ingestion", "input": {"key": f"/{data['id']}/a.csv"}},
        {"name": "ingestion", "input": {"key": f"/{data['id']}/b.csv"}},
    ])
    ctx.sleep(60_000)                                    # durable timer
    return {"rows": sum(r["rows"] for r in results), "startedAt": started_at}

run_workers([worker], redis=redis_url_from_env())        # one loop, one Redis pool, graceful shutdown

A failure the workflow should turn into an outcome (a line that fails validation) is routed to a handler: .on_error(code) on the workflow function registers the handler as its own workflow (pipeline.onError.<function>) and announces the route in the worker's handshake descriptor; the Adonis engine starts it when a run fails with that code. Raise FatalError(message, code, details=...) in the step so it skips the retries and the handler gets failure["error"]["details"]. See Routing a run's failure.

@pipeline.on_error("invalid_line")
def reject_line(ctx: WorkflowContext, failure):  # {"runId", "workflow", "input", "error"}
    ctx.step("record_rejected_line", failure)
    return {"rows": 0, "rejected": True}

Workflows as classesCopy Anchor Link

A workflow can be a class: run(self, ctx, input) is the body, and each method marked @on_error(code) is a handler (<name>.onError.<method>). Decorate it with @worker.workflow(name). A registered class is a workflow reference: pass it to ctx.start_child, ctx.gather_children or ctx.all_settled instead of the name.

from durable_worker import FatalError, WorkflowContext, on_error

class InvalidLine(FatalError):
    code = "invalid_line"

@worker.workflow("line")
class LineWorkflow:
    def run(self, ctx: WorkflowContext, line):
        ctx.step("validate_line", line)
        return {"ok": 1, "failed": 0}

    @on_error(InvalidLine)
    def reject_line(self, ctx: WorkflowContext, failure):
        ctx.step("record_rejected_line", failure)  # side effects go in steps
        return {"ok": 0, "failed": 1}

@worker.workflow("file")
def file(ctx: WorkflowContext, lines):
    return ctx.all_settled(LineWorkflow, lines)

The body and the handlers are replayed, so their side effects belong in steps. A class that needs something at construction (configuration, a client) is declared with @workflow(name, ...) instead and handed to the worker built: worker.register(LineWorkflow(cfg)). The instance serves every run and every replay, so keep only what it was built with on it, never run state.

The WorkflowContext opsCopy Anchor Link

ctx mirrors the TypeScript WorkflowCtx — each op is keyed by a deterministic seq, so a completed op replays its recorded result instead of re-running. Every op compiles to one of the commands the Adonis engine applies (call, recordStep, sleep, waitSignal, startChild):

Changing the op sequence under a run already in flight raises NondeterminismError rather than silently diverging. The three gather* fan-outs default to wait_all — raising an aggregate GatherFailed if anything fails — with an opt-in fail_fast. A single step that fails raises StepFailed in the workflow; catch it to compensate, or let it propagate to fail the run.

Dedicated workflow workerCopy Anchor Link

To keep workflow replay and step execution in separate processes, author on a dedicated WorkflowWorker and run it with run_redis_workflow_worker (the workflow-shaped sibling of run_redis_worker):

from durable_worker import WorkflowWorker, redis_url_from_env
from durable_worker.redis_runner import run_redis_workflow_worker

workflows = WorkflowWorker(group="py-workflows")

@workflows.workflow("pipeline")
def pipeline(ctx, base_id):
    rows = ctx.step("ingestion", {"key": f"/{base_id}/data.csv"}, group="pipeline")
    return {"rows": rows}

# await run_redis_workflow_worker(workflows, connection=redis_url_from_env())

You can also mix step and workflow workers in one process with run_workers([step_worker, workflows], redis=...).

Dispatching a Python workflow from AdonisCopy Anchor Link

Start the workflow by its name — whichever live Python worker is serving that name picks up the turns. From an AdonisJS service:

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

await engine.start('pipeline', { id: baseId }, `pipeline-${baseId}`)

or, from inside another workflow, await ctx.child('pipeline', { id: baseId }). The queue name is the routing: the worker serving pipeline runs it. If none is live, the start throws workflow pipeline is not registered, so a typo'd name never silently hangs. The live-queue check reads worker heartbeats, so it works on any transport that lists live worker groups, but only bullmq carries Python workers. The group is resolved per start for the run's namespace: an operator starting a tenant's run ({ namespace: 'acme' }) reaches the tenant's pipeline@acme worker, the same token the tenant's steps route to.

Each turn's decision comes back on <P>-decisions, and with several Adonis replicas any of them may take it; it is applied by run id either way (see Remote workflow turns). The Python worker beats for the run every 5 s while it replays a turn, so set remoteAdvanceSilenceMs in config/durable.ts (say, two minutes) to have a turn whose worker died re-driven instead of waiting forever.

The workflow-task/decision wire is language-neutral, so the same Python workflow runs unchanged against an Adonis, a NestJS, or any other conformant control plane.