Durable execution

An agent task makes several model calls and tool calls. When it fails partway (a network error, a timeout, a provider outage), a plain Airflow retry runs every one of them again: the model calls are paid for twice and the tools’ side effects happen twice. With durable execution, each step the agent completes is recorded in a journal kept for the task instance, and the retry replays the recorded steps instead of running them again. Only the steps the failed attempt never finished run live.

Durable execution only helps a task that has retries. To decide whether a failure is worth retrying, see Retry policies; a retried LLMBatchOperator re-attaches to its running batch instead (Retries re-attach instead of re-submitting). To resume a Strands Agents agent on a retry, see Durable execution with Strands.

Turn it on

Set durable=True on AgentOperator:

airflow/providers/common/ai/example_dags/example_agent_durable.py[source]

if SQLToolset is not None:

    @dag(default_args={"retries": 3, "retry_delay": timedelta(seconds=30)}, tags=["example"])
    def example_agent_durable_operator():
        """Agent with durable execution: a retry replays the steps the failed attempt completed."""
        AgentOperator(
            task_id="durable_analyst",
            prompt="What are the top 5 customers by order count?",
            llm_conn_id="pydanticai_default",
            system_prompt=(
                "You are a SQL analyst. Use the available tools to explore "
                "the schema and answer the question with data."
            ),
            durable=True,
            toolsets=[
                SQLToolset(
                    db_conn_id="postgres_default",
                    allowed_tables=["customers", "orders"],
                    max_rows=20,
                )
            ],
        )

The same flag works on @task.agent:

airflow/providers/common/ai/example_dags/example_agent_durable.py[source]

if SQLToolset is not None:

    @dag(default_args={"retries": 3, "retry_delay": timedelta(seconds=30)}, tags=["example"])
    def example_agent_durable_decorator():
        @task.agent(
            llm_conn_id="pydanticai_default",
            system_prompt="You are a data analyst. Use tools to answer questions.",
            durable=True,
            toolsets=[
                SQLToolset(
                    db_conn_id="postgres_default",
                    allowed_tables=["orders"],
                )
            ],
        )
        def analyze(question: str):
            return f"Answer this question about our orders data: {question}"

        analyze("What was our total revenue last month?")

durable=True attaches the AirflowDurability capability to the agent. A pydantic-ai agent you build yourself in a plain @task takes the capability directly:

airflow/providers/common/ai/example_dags/example_agent_durable.py[source]

if SQLToolset is not None:

    @dag(default_args={"retries": 3, "retry_delay": timedelta(seconds=30)}, tags=["example"])
    def example_airflow_durability_capability():
        @task
        def analyze(question: str) -> str:
            agent = PydanticAIHook(llm_conn_id="pydanticai_default").create_agent(
                name="orders_analyst",
                instructions="You are a data analyst. Use tools to answer questions.",
                toolsets=[SQLToolset(db_conn_id="postgres_default", allowed_tables=["orders"])],
                capabilities=[AirflowDurability()],
            )
            return agent.run_sync(question).output

        analyze("What was our total revenue last month?")

The capability needs a name for the agent’s steps: the agent’s name, or AirflowDurability(name=...). AgentOperator uses the agent’s name when agent_params sets one, and the task id otherwise. Outside a running Airflow task the capability does nothing, so the same agent runs normally in a test or a notebook.

Give every toolset an id

pydantic-ai records each call of a FunctionToolset, MCPToolset or DynamicToolset under the toolset’s id, and refuses to run one without it. Set the id on the toolset itself:

orders = FunctionToolset(tools=[get_order], id="orders")

AgentOperator(durable=True) checks this when the Dag is parsed and fails with:

ValueError: durable=True needs a unique id on every FunctionToolset, such as FunctionToolset(..., id='orders'): durable replay records each tool call under it. Set it on the toolset itself, and pass a function that builds a toolset as DynamicToolset(function, id=...); the id of a Toolset capability around a toolset does not reach it.

Keep an id stable once tasks run with it: a retry finds its recorded tool calls by it. The provider’s own toolsets (SQLToolset, HookToolset and the rest) and tools passed as agent_params={"tools": [...]} need nothing. A capability that contributes @durable_operation methods needs an id for the same reason; pydantic-ai raises UserError naming it when it has none. pydantic-ai-harness SpendLimits has one by default.

Where the journal is kept

On Airflow 3.3 and later the journal is kept in the task state store, scoped to the task instance. No configuration is needed. Each step is written to the Airflow metadata database; for agents with large model responses or tool results, set [workers] state_store_backend to offload the values to external storage instead.

On Airflow before 3.3 the journal is a JSON file on ObjectStorage, and the location must be set:

[common.ai]
# Local filesystem, suitable for development
durable_cache_path = file:///tmp/airflow_durable_cache

Any ObjectStorage URI works. In production use a shared store, so a retry on another worker can read the journal:

[common.ai]
durable_cache_path = s3://my-bucket/airflow/durable-cache

Without the option, the task fails with:

ValueError: durable=True requires [common.ai] durable_cache_path to be set. Example: durable_cache_path = file:///tmp/airflow_durable_cache

How a retry replays

Each step the agent completes is recorded under its position in the run, with its name (which model or tool it was) and a fingerprint of what it was asked: the model, the message history, the settings and the tool definitions for a model request; the tool name, its arguments and the model-issued call id for a tool call.

The retry runs the agent from the start. At each position, a step with the same name and the same fingerprint as the recorded one is replayed: the recorded result is returned and the model or tool is not called. The first step that differs means the run has taken another path, because the prompt, the model, a setting, a tool or the conversation changed, so that step and every step after it run live and are recorded again, and the task log says so:

Durable: the run took a different path from the previous attempt; this step and every step after it run again

A changed agent costs a re-run; it never replays results that belong to another conversation. Settings that only affect transport, timeout and the prompt cache settings, are left out of the fingerprint, so changing them does not cause a re-run. Replay compares requests only: editing a tool’s implementation does not invalidate a recorded result for an identical call.

When a run fails, the journal notes where. The retry runs the step that failed again, and runs live everything the failed attempt did after it, such as the error handling of other capabilities. Tool calls already running alongside the failed one still replay. A step that raised and that the run recovered from, such as a tool error the model was told about, does not stop the steps after it from replaying.

A run can also fail on the model’s answers rather than in a step: pydantic-ai raises UnexpectedModelBehavior when the model keeps giving answers a tool rejects with ModelRetry, or output that fails validation, past its retry limit. Those answers are recorded, so replaying them would only reach the same error. The retry instead replays the steps before the model’s last request and asks the model again from there. The earlier rejected answers still count against the retry limit, so each Airflow retry gives the model one more answer, with the rejections it already got in its history. Tool calls the last answer made run again if the model asks for them again, including calls that succeeded alongside the one that was rejected. The task log names the step:

Durable: the run took a different path from the previous attempt; this step and every step after it run again position=2 step=analyst__model.request reason=the model's answers from this step on failed the previous attempt, so it is asked again

Some steps have no request to fingerprint: another capability’s @durable_operation, or tool discovery. Such a step replays on its name, and only when the step before it also replayed.

The journal records:

  • model requests, including streamed ones;

  • calls to tools from any toolset, including the provider’s own toolsets, MCP servers, toolsets built per run, and toolsets supplied by capabilities;

  • tool discovery for MCP and per-run toolsets;

  • @durable_operation methods of other capabilities, such as the spend accounting of pydantic-ai-harness SpendLimits and the summarization of SummarizingCompaction, so a retry neither charges the spend twice nor pays for a summary again.

A tool that raises ModelRetry, ToolFailed, ApprovalRequired or CallDeferred replays as the same signal. AgentOperator does not support tool approval or deferred tools with durable=True, so there such a call fails the task with UnsupportedToolDeferralError, as it would without durable. A tool that raises any other exception runs again on retry.

A task can run more than one agent, and an agent’s tool can run another agent; each agent run is replayed on its own, and one a tool started is replayed under that tool call. The journal is deleted when the task succeeds, including any steps an earlier attempt recorded beyond where the last one went, so a task cleared later starts fresh. In a plain @task, which has no such moment, the capability deletes a run’s journal as soon as that run succeeds: after a failure, only the run that failed replays, and the agents that ran before it run again.

Values that are not plain JSON

Plain JSON tool arguments and settings are hashed as they are. Anything else is fingerprinted from pydantic’s JSON rendering, so ordinary types that are not JSON, such as a datetime or Decimal tool argument, a dataclass in tool_choice, the bytes in a BinaryContent or a dict keyed by date, fingerprint normally. Bytes in tool arguments and settings are rendered as base64, so binary data that is not valid UTF-8 fingerprints too; bytes inside a pydantic model follow that model’s ser_json_bytes setting instead, which renders them as UTF-8 text by default. Because a tool call is fingerprinted from how its arguments render, a field excluded from serialization and a secret value, which renders masked, take no part in it.

The members of a set are ordered before hashing. A set of strings iterates in an order that changes from one process to the next, and every attempt is a new process, so without this a step holding one would never match on retry. A list that a serializer produces is hashed as the serializer produced it, so a set that a serializer turns into a list may not match on retry. The one list that is re-sorted is the output of a JSON-only serializer (when_used="json") that returns the members of a set: it cannot be told from the set itself, so ["z-first", "a"] hashes as ["a", "z-first"].

A request cannot be fingerprinted when:

  • two distinct dict keys render alike, such as 1 and "1", since the fingerprint could then not tell those payloads apart;

  • a set whose order changes between processes cannot be found in the rendering, because its members have different shapes, or because a JSON-only serializer renamed or reshaped it, even one that also sorts it;

  • a value would render through an iterator, since rendering would consume it. A parameter annotated Iterable[...] is the usual case: pydantic validates it lazily, and reading it to hash it would consume input the tool has not read yet. Tool arguments are rendered from copies, so fingerprinting never changes what the tool receives, and an argument that cannot be copied is not fingerprinted either;

  • pydantic cannot render a value at all.

Such a step is handled like a step with no request to fingerprint: it replays on its name, and only when the step before it replayed. The task log names it with a could not fingerprint model request or could not fingerprint tool call warning. On the model path the cause is rarely confined to one step. model_settings and the tool definitions travel with every request, and a value that enters the message history, for example in a tool return, is carried into every later request, so every model step from that point on is replayed without verification.

A set of models or dataclass instances, and from pydantic 2.13 a dict keyed by a model or dataclass instance, is a special case. In a tool argument or a setting it cannot be fingerprinted. In the message history it is rendered, and the history is then hashed from its JSON rendering alone: nothing in it is ordered and nothing is refused, so a set of strings elsewhere in that history can make the run diverge at that step on retry.

Code mode

With pydantic-ai-harness’s CodeMode (Code mode), the model sends Python to one run_code tool, and the snippet calls the agent’s tools as functions. run_code is not a step of its own: the retry runs each snippet again, and the tool calls the snippet makes replay from the journal one by one, so a snippet that called three tools is three recorded steps. When the snippet returns what it returned before, the model request after it replays too.

A call the snippet makes replays only when the call at that position in the journal has the same tool, arguments and call id. Code mode numbers a snippet’s calls in the order the snippet makes them, so when calls under asyncio.gather reach the journal in another order on the retry, the first call out of place and every step after it run live. A call never gets the result recorded for another one.

The snippet’s own code is not recorded. Whatever it does outside a tool call happens again on the retry, such as writing to a directory shared through mount or calling an os_access handler, and if what it reads that way has changed, the tool calls it makes next differ and run live. Keep side effects in tools, whose results are recorded.

CodeMode’s eager and speculate options start tool calls while the model is still writing the snippet. Both are off while the run is durable, so a retry runs the finished snippet the way the previous attempt did.

What the task log shows

AgentOperator logs a summary at the end of every attempt, including the one that fails. Here an agent asked for two tools in one step: charge_card succeeded and flaky_lookup raised, which failed the attempt. The failing attempt logged:

LLM run failed: requests=1, tool_calls=1, input_tokens=58, output_tokens=9, total_tokens=67
Durable: replayed 0 steps (0 model, 0 tool, 0 other), recorded 2 new steps (1 model, 1 tool, 0 other)

The retry replayed the model response and charge_card, and ran only flaky_lookup and the final model request:

Durable: replayed 2 steps (1 model, 1 tool, 0 other), recorded 2 new steps (1 model, 1 tool, 0 other)
LLM run complete: model=test, requests=1, tool_calls=1, input_tokens=65, output_tokens=20, total_tokens=85

These lines come from a test agent with two tools, not from the examples above. charge_card ran once across both attempts, and the retry’s usage counts only its own two live calls. A step that ran live but could not be recorded is named in a WARNING when it happens; the summary names such tool steps again and counts such model steps. Per-step detail is logged at DEBUG.

The agent’s run_id and conversation_id

Without durable, each attempt’s agent run gets a key for the attempt as its pydantic-ai run_id: the task instance id on Airflow 3, which changes on every retry. With durable=True every attempt keeps the first attempt’s id, and so does the run’s conversation_id when there is no message_history. Capabilities that key their own state on these ids, such as pydantic-ai-harness SpendLimits and StepPersistence, then find the records the replayed steps refer to: with a per-attempt id, a retry that replays SpendLimits’ accounting for a run window fails with a KeyError. The run_id XCom, from a failed attempt or a successful one, and the gen_ai.agent.call.id span attribute carry the same id; the spans of each attempt still carry that attempt’s own airflow.task_instance.id and try_number. The id is stored with the journal and deleted with it, so a task cleared after it succeeded starts a new one.

Usage and limits

A replayed step adds nothing to the usage the run counts against usage_limits or reports in the usage XCom: not its request, tokens, cost or tool calls, nor the usage a replayed @durable_operation brings back. Each attempt counts only the calls it actually makes. This holds after clearing a failed task instance too: the clear starts a fresh budget but keeps the journal, and what the rerun replays from it is free.

Side effects and idempotency

Durable execution records what a tool returned. A replayed tool’s code does not run, so a side effect it had (a file written, a message sent) is not repeated. A tool that fails before its result is recorded runs again on retry, so a tool that sent an email and then raised sends it again. The same holds when the worker dies between a tool’s side effect and the write of its result, and for the tool calls of a model answer that the retry asks for again after the model ran out of retries. Make tools with side effects idempotent: check whether the work already happened, or let a database constraint reject the duplicate.

Tool results are masked by Airflow’s secret masker before they are recorded, so a connection password in a result reaches the journal as ***.

A toolset can opt out of replay. ManagedAgentToolset does by default (replayable=False), because a managed agent acts on a system Airflow cannot observe: its calls run again on every attempt, while the steps around them still replay.

A result has to survive a JSON round trip to be recorded. One that cannot be encoded fails the task with DurableJournalError and no retry, since it would fail the same way every time; the model could not have read it either. A write the task state store refuses does not fail the task: the task log names the step, and a retry runs it again.

Limitations

  • durable=True cannot be combined with enable_hitl_review, a SandboxToolset or tool approval. AgentOperator raises ValueError for the first two when the Dag is parsed.

  • A task instance that is retried across an upgrade of this provider from a version without AirflowDurability runs its steps live once: the journal keys changed.

See also

Was this entry helpful?