Durable Tasks

Use durable tasks when you don’t know the shape of work ahead of time. For example, an AI agent that picks its next action based on a model response, a fan-out where N is determined by the input data, or a pipeline that branches and spawns sub-workflows based on intermediate results. In all of these cases, the “graph” of work doesn’t exist when the task starts; it emerges at runtime as the task makes decisions and spawns children.

A durable task is a single long-running function that acts as an orchestrator: it spawns child tasks, waits for their results, makes decisions, and spawns more. Hatchet checkpoints its progress so it can recover from crashes, survive long waits, and resume on any worker without re-executing completed work.

When to Use Durable Tasks

How It Works

A durable task builds the workflow at runtime through child spawning. The task function runs, inspects data, and decides what to do next by spawning child tasks. The parent is evicted while children execute, freeing its worker slot. When children complete, the parent resumes from its checkpoint and continues.

This is fundamentally different from a DAG, where every task and dependency is declared before execution begins. With durable tasks, the number of children, which branches to take, and whether to loop or stop are all determined by your code at runtime.

The Durable Context

Declare a task as durable (using durable_task instead of task) and it receives a DurableContext instead of a normal Context. The DurableContext extends Context with methods for checkpointing and waiting:

Example Task

durable_workflow = hatchet.workflow(name="DurableWorkflow")

Now add tasks to the workflow. The first is a regular task; the second is a durable task that sleeps and waits for an event:

EVENT_KEY = "durable-example:event"
SLEEP_TIME = 5


@durable_workflow.task()
async def ephemeral_task(input: EmptyModel, ctx: Context) -> None:
    print("Running non-durable task")


@durable_workflow.durable_task()
async def durable_task(input: EmptyModel, ctx: DurableContext) -> dict[str, str]:
    print("Waiting for sleep")
    await ctx.aio_sleep_for(duration=timedelta(seconds=SLEEP_TIME))
    print("Sleep finished")

    print("Waiting for event")
    await ctx.aio_wait_for(
        "event",
        UserEventCondition(event_key=EVENT_KEY, expression="true"),
    )
    print("Event received")

    return {
        "status": "success",
    }

If this task is interrupted at any time, it will continue from where it left off. If the task calls ctx.aio_sleep_for for 24 hours and is interrupted after 23 hours, it will only sleep for 1 more hour on restart.

Or Groups

Durable tasks can combine multiple wait conditions using or groups. For example, you could wait for either an event or a sleep (whichever comes first):

@durable_workflow.durable_task()
async def wait_for_or_group_1(
    _i: EmptyModel, ctx: DurableContext
) -> dict[str, str | int]:
    start = time.time()
    wait_result = await ctx.aio_wait_for(
        uuid4().hex,
        or_(
            SleepCondition(timedelta(seconds=SLEEP_TIME)),
            UserEventCondition(event_key=EVENT_KEY),
        ),
    )

    key = list(wait_result.keys())[0]
    event_id = list(wait_result[key].keys())[0]

    return {
        "runtime": int(time.time() - start),
        "key": key,
        "event_id": event_id,
    }

Spawning Child Tasks

Child spawning is the primary way durable tasks build workflows at runtime. A durable task can spawn any runnable (regular tasks, other durable tasks, or entire DAG workflows), wait for results, and decide what to do next.

The parent is evicted while children execute, so it consumes no resources. The number and type of children can be determined dynamically based on input, intermediate results, or model outputs.

See Child Spawning for patterns and full examples.