---
question: How do I write a workflow whose steps survive a crash or a retry?
kind: task
via_web: na
via_cli: na
via_python: available
via_api: na
via_config: na
ref_python: "@workflow"
prereq: You need ABI 3.x, a NATS project (the default of `abi new project` on 3.x) and a module listed under `modules:` in `config.yaml`. ABI 2.x has no durable workflows.
next:
  - /v2/capabilities/build/workflows/run-a-workflow
  - /v2/capabilities/build/workflows/schedule-a-workflow
  - /v2/capabilities/build/workflows/monitor-and-cancel-a-run
---

> **You are reading the v2.x docs (engine modules). New projects use v3.** [Read this page in v3](/capabilities/build/workflows/write-a-durable-workflow) · [Migrate from 2.x to 3.0](/capabilities/upgrade/migrate-to-3)

# Write a durable workflow

A durable workflow is a method of the module, decorated with `@workflow`. It is a job with a typed input and output whose steps are checkpointed: each `ctx.step` stores its result, so when the run is retried (a crashed host, a deploy, a failed step) the steps that finished return their stored results and only the rest run. Inside a run it can fan out over items and start child runs of other modules' workflows.

Use it when the work has several steps that are slow or costly to repeat, or when it spreads over other modules. For one call that returns data to an agent, a [workflow class](/v2/capabilities/build/agents/creating-a-workflow) is enough: in a NATS project it is published as the durable workflow `workflow.<slug>` anyway.

Durable workflows are ABI 3.x with NATS. On ABI 2.x, and on 3.x without NATS, there is no `@workflow`: write a workflow class, and schedule it with a [Dagster](/v2/capabilities/build/applications/dagster) orchestration (legacy on 3.x).

## Web UI

Not applicable. A durable workflow is a Python method of the module. Nexus starts it and shows its runs: see [Run a workflow](/v2/capabilities/build/workflows/run-a-workflow?via=web).

## CLI

Not applicable. No `abi` command writes a `@workflow` method. Write the method as in the Python steps.

## Python

### Declare the input and the output

Input and output are Pydantic models. Their JSON Schemas are published with the workflow: Nexus builds its launch form from the input schema, and checks a start against it.

```python title="src/reports/__init__.py"
from datetime import datetime, timedelta, timezone

from pydantic import BaseModel, Field


class ReportInput(BaseModel):
    day: str | None = Field(
        default=None, description="The day to report on, as YYYY-MM-DD; empty: yesterday (UTC)"
    )
    regions: list[str] = Field(default_factory=lambda: ["emea", "amer"])


class ReportOutput(BaseModel):
    key: str
    rows: int
```

A `dict` works too, with `input_schema=` and `output_schema=` on the decorator to describe it.

A scheduled run sends the same input on every tick, so `day` is optional: the run works out which day it reports on itself (next step).

### Write the method

Import `workflow` and `SyncFlowContext` from `naas_abi_core.module.workflows`. A plain method runs in a worker thread and gets a `SyncFlowContext`, whose `step`, `map`, `call` and `map_call` block until they return. This one drops a daily report in object storage:

```python title="src/reports/__init__.py"
from naas_abi_core.module.Module import BaseModule, ModuleDependencies
from naas_abi_core.module.workflows import SyncFlowContext, workflow
from naas_abi_core.services.object_storage.ObjectStorageService import ObjectStorageService


class ABIModule(BaseModule):
    dependencies: ModuleDependencies = ModuleDependencies(
        modules=[],
        services=[ObjectStorageService],
    )

    @workflow(max_attempts=3)
    def daily_report(self, ctx: SyncFlowContext, inp: ReportInput) -> ReportOutput:
        """Builds the day's report and drops it in object storage."""
        day = inp.day or ctx.step("day", yesterday)
        rows = ctx.map("fetch", [{"region": r, "day": day} for r in inp.regions], self.fetch_region)
        csv = ctx.step("render", self.render_csv, day, rows)
        key = ctx.step("store", self.store, f"daily/{day}.csv", csv)
        return ReportOutput(key=key, rows=sum(len(r) for r in rows))

    def fetch_region(self, item: dict) -> list[dict]:
        ...  # the rows of item["region"] for item["day"]

    def store(self, key: str, csv: str) -> str:
        self.engine.services.object_storage.put_object("reports", key, csv.encode())
        return key


def yesterday() -> str:
    return (datetime.now(timezone.utc) - timedelta(days=1)).date().isoformat()
```

The day is a step of its own: a retry of the run, even after midnight, reads the stored day and writes the same key.

The first line of the method's docstring is the workflow's description. Its name is the workflow's name (`daily_report`); pass `name=` to the decorator to choose another.

`ctx.cancelled` is an `asyncio.Event`: test it with `ctx.cancelled.is_set()` in a long step.

The engine needs `naas-abi-core[nats]` (part of `[all]`). Without it, `@workflow` still imports, but nothing runs.

### Or write it async

An `async` method gets a `FlowContext` (also in `naas_abi_core.module.workflows`) and awaits each call: `day = inp.day or await ctx.step("day", yesterday)`. A step's `fn` that is a plain function runs in a worker thread, so a step that calls a service does not block the loop. An `async def` step function that calls a service directly does: the engine's services are synchronous. Give such a call a plain step function, or write the workflow as a plain method.

### Checkpoint each step

`ctx.step(key, fn, *args, **kwargs)` runs `fn` once for the run and stores its result. `fn` may be sync (it runs in a thread) or async. On a retried run, a step that already succeeded returns the stored result without calling `fn`.

```python
items = ctx.step("collect", self.collect, inp.day, retries=2, backoff=1.0)
```

`retries` adds attempts of that step alone, after `backoff` seconds, doubling each time. When they are spent, the exception reaches the method and the step is recorded `FAILED`; it runs again on the next attempt of the run.

### Fan out

`ctx.map(key, items, fn, concurrency=8)` runs `fn` on each item in this process, `concurrency` at a time, each on a worker thread of its own. Item `i` is the step `key[i]`, so a retry runs only the items that did not finish. Results keep the order of `items`. An item function in a long `map` should check `ctx.cancelled`.

### Call other workflows

`ctx.call(module_id, name, input)` starts a child run of a workflow of any module, waits for it and returns its output. `ctx.map_call(key, module_id, name, inputs, concurrency=8)` starts one child per input:

```python
summary = ctx.call("mailer", "send_digest", {"key": key})
enriched = ctx.map_call("enrich", "items", "enrich_item", [{"id": i} for i in ids])
```

A module's id is its name under `modules:` in `config.yaml`. A class of another module is called by its published name: `ctx.call("items", "workflow.lookup_workflow", {...})`.

- The child's run id is stored before the parent waits. A retried parent waits for the same child instead of starting another.
- A child that does not end `SUCCEEDED` (or `SKIPPED`) fails the step with `ChildRunFailed`.
- The parent keeps its concurrency slot while it waits. A workflow that calls itself is refused with `SelfCall`: its child would wait for the slot its parent holds. A cycle through two workflows can still deadlock.
- A parent that is cancelled, times out, or fails with no attempt left cancels the children it still waits for.
- Without a `key`, each call is keyed by a counter. Pass `key=` when calls depend on a condition, so the keys stay the same on every attempt.

### Set the limits

| Option | Default | Meaning |
|---|---|---|
| `max_attempts` | 1 | Attempts of the whole run. Steps that finished are not run again. |
| `max_concurrency` | 1 | Runs of this workflow at once, across every replica. |
| `timeout` | none | A `timedelta`. The run is cancelled and ends `TIMED_OUT`. |
| `triggers` | none | `Cron`, `Every` or `OnEvent`: see [Schedule a workflow](/v2/capabilities/build/workflows/schedule-a-workflow). |

### Start the module

Restart the engine (`abi dev up`, or its container). The module declares the workflow when it loads, and the workflow shows in Nexus under System › Jobs, with the **Workflows** filter.

## REST API

Not applicable. No route writes a workflow. Once the module runs, `POST /api/workflows/{module_id}/{name}/runs` starts it: see [Run a workflow](/v2/capabilities/build/workflows/run-a-workflow?via=api).

## Config

Not applicable. A workflow is declared in code. It needs a `nats:` block in `config.yaml`: without NATS, durable workflows are not hosted.

## Rules for steps

- **Keys are stable.** A key names a stored result, so it must be the same on every attempt of the run and unique within it. Build it from the input or from an earlier step's result, never from the time or a random value. A key is at most 200 characters, with no control characters.
- **Order is stable.** `map` and `map_call` key items by position: take the list from an earlier step, so it comes in the same order on a retry.
- **Results are small.** A step result is JSON (a Pydantic model is dumped), at most 256 KiB, and a run record keeps 64 KiB of output. Put large data in object storage and return its key, as `store` does above.
- **Fan-outs are bounded.** One `map` or `map_call` takes at most 10,000 items. Split larger lists into batches, one child workflow per batch.

## Failures and retries

- **An invalid input fails the run at once.** The input is checked against the input model before the method runs. A refused input fails the run with `InvalidWorkflowInput`, without a retry. An output that does not match the output model, a step result over 256 KiB and an invalid step key fail it the same way: another attempt could not fix them.
- **Other errors retry the whole run.** While `max_attempts` allows, the run starts again after a backoff, and the steps that succeeded return their stored results.
- **`ctx.skip(reason)`** ends the run `SKIPPED`, with no output, when there is nothing to do.

## Make a run safe to repeat

Delivery is at least once. A child run can also start twice: the parent sends the child's trigger, then stores the child's run id. If the parent's host stops between the two, the retried parent sends the trigger again with the same idempotency key, which JetStream drops as a duplicate only within two minutes; a retry that comes later starts a second child. Write each step so a second run changes nothing: write to a key derived from the input (`daily/<day>.csv`), not to the end of a list, and record a side effect that cannot be undone (an email) under its key before you do it. Long waits (sleeping for a day, waiting for an approval) are not supported inside a run: end the run and start another later.