---
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 a NATS project (the default of `abi new project`) and a module. A remote module also needs discovery and the `document` service in its `ModuleDependencies`, as `abi new module` writes it.
next:
  - /capabilities/build/workflows/run-a-workflow
  - /capabilities/build/workflows/schedule-a-workflow
  - /capabilities/build/workflows/monitor-and-cancel-a-run
---

# 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](/capabilities/build/agents/creating-a-workflow) is enough: it is published as the durable workflow `workflow.<slug>` anyway.

## 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](/capabilities/build/workflows/run-a-workflow?via=web).

## CLI

Not applicable. No `abi` command writes a `@workflow` method. `abi new module` writes a remote module whose `jobs.py` explains the pattern; 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

In a remote module, the method is `async` and gets a `FlowContext`. This one drops a daily report in object storage:

```python title="src/reports/__init__.py"
from naas_abi_sdk import BaseModule, ModuleDependencies
from naas_abi_sdk.workflows import FlowContext, workflow


class ABIModule(BaseModule):
    module_id = "acme.reports"
    dependencies = ModuleDependencies(services=("document", "object_storage"))

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

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

    async def store(self, key: str, csv: str) -> str:
        await 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 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.

### Or write it in an engine module

An engine module imports from `naas_abi_core.module.workflows`. An `async` method works as above. A plain method runs in a worker thread and gets a `SyncFlowContext`, whose `step`, `map`, `call` and `map_call` block until they return:

```python title="src/my_module/__init__.py"
from naas_abi_core.module.Module import BaseModule
from naas_abi_core.module.workflows import SyncFlowContext, workflow


class ABIModule(BaseModule):
    @workflow(max_attempts=3)
    def daily_report(self, ctx: SyncFlowContext, inp: ReportInput) -> ReportOutput:
        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))
```

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

### 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 = await 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. Item `i` is the step `key[i]`, so a retry runs only the items that did not finish. Results keep the order of `items`.

### 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 = await ctx.call("acme.mailer", "send_digest", {"key": key})
enriched = await ctx.map_call("enrich", "acme.items", "enrich_item", [{"id": i} for i in ids])
```

A class of another module is called by its published name: `ctx.call("acme.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](/capabilities/build/workflows/schedule-a-workflow). |

### Start the module

Restart the module (`abi dev up`, or its container). It declares the workflow when it starts, 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](/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.