kaxil opened a new pull request, #76917:
URL: https://github.com/apache/airflow/pull/76917
Stacked on #76860 to #76882 and the HITL review PR below this one; that PR's
diff includes #75412 until it merges.
On Airflow 3.3 and later, `AgentOperator` and `@task.agent` now journal
their model responses and tool calls by default when the task can retry, so a
retry replays what the failed attempt completed instead of paying for every
model call again and repeating every tool's side effect. `durable` changes from
`bool = False` to `bool | None = None`:
| `durable` | What the task gets |
|---|---|
| `None` (default) | On for Airflow 3.3+ when `retries > 0`, or from 3.4
when a `retry_policy` is set (`RetryAction.RESCHEDULE` can rerun a task with no
retries). Off below 3.3, off with no retries, and off where it cannot apply,
with the reason in the task log |
| `True` | On everywhere, as before. Combinations it cannot apply to fail
the Dag at load time; a capability pydantic-ai refuses fails the task when it
runs |
| `False` | Off |
A retried task with the default, from a real run (gpt-5.4-mini; `send_note`
raised `ConnectionError` on try 1):
```
try 1 TOOL RAN: lookup_order(4411) in default_on try 1
TOOL RAN: send_note in default_on try 1
Durable (on by default): replayed 0 steps (0 model, 0 tool, 0 other),
recorded 3 new steps (2 model, 1 tool, 0 other)
try 2 TOOL RAN: send_note in default_on try 2
Durable (on by default): replayed 3 steps (2 model, 1 tool, 0 other),
recorded 2 new steps (1 model, 1 tool, 0 other)
```
A task whose capability pydantic-ai refuses to run durably (a
`@durable_operation` with no `id`) logged pydantic-ai's reason and ran without
durable execution:
```
Durable execution is off for this task: Capability 'SpendLedger' contributes
durable operations and needs an explicit `id` because persisted operation
identity and worker-side recovery must remain stable. ...
```
`lookup_order` ran once across both attempts. In the same Dag run, a
`durable=False` task ran both tools and all three model requests again on try
2, and a task with `retries=0` logged `Durable execution is off for this task:
the task has no retries, so no attempt would replay this one.`. Reading the
journal through `GET .../taskInstances/{task_id}/state-store` between the
attempts listed six `__commonai_durable__*` entries; after the retry succeeded,
the list was empty.
## Design rationale
**The effective mode is decided when the task runs, not in `__init__`.** The
task state store only exists in the task's context, and a reason logged in
`__init__` would land in the Dag processor's logs on every parse rather than in
the task log. `_decide_durable` runs at the start of `execute()` and of
`resume_hitl_review`. The summary line now says how it was turned on (`Durable
(on by default): ...` or `Durable (durable=True): ...`), and every agent span
carries `airflow.common_ai.durable`.
**One list of what durable execution cannot handle, used both ways.**
`_durable_problem()` is raised by `__init__` under `durable=True` and logged by
the default. That list now includes a toolset without an id inside a
`CombinedCapability` or wrapper capability, and a `Toolset(build_fn)`
capability. pydantic-ai refuses both with `UserError: Toolsets that are
'leaves' ... need to have a unique id` on every durable attempt. Under the
default they would otherwise turn a working task into one that always fails,
and under `durable=True` they now fail when the Dag loads instead of at run
time.
**pydantic-ai has the last word, so the build is a backstop.** pydantic-ai
checks a durable agent when it builds it and refuses shapes
`_durable_problem()` can't see from the operator: a capability function or
`DynamicCapability`, a custom `get_toolset`, a `Capability(toolsets=[...])`
toolset without an id, or a capability with `@durable_operation` methods and no
`id`. Under the default, a `UserError` from building the durable agent turns
durable execution off and the task log quotes pydantic-ai's reason;
`durable=True` keeps the error. The decision is settled by that build before
anything durable is written: in `execute`, and on a HITL review resume before
the reviewer's answer is stored.
**The default is lenient where `durable=True` is strict.** A model response
or tool result the journal cannot encode is logged and runs again on retry
under the default; `durable=True` keeps failing the task with
`DurableJournalError`. This uses the `fail_on_unrecordable` hook added lower in
the stack, so turning on the default never fails a task that worked without it.
**A `@durable_operation` called inside a tool's step now belongs to that
step.** The issue asked to confirm what happens to nested durable operations
when an Airflow toolset step replays. Agents a tool starts were already fine:
they are numbered under the step. A capability's `@durable_operation` called
from an Airflow toolset tool, such as a spend ledger's accrual, took a position
of its own in the outer run. On a retry the tool replayed without running it,
every later step moved by one, and the run diverged: the next tool ran twice.
Such an operation now runs inline whenever its step runs, which is what
pydantic-ai's own `durable_unit_scope` does for its toolsets on newer releases;
the 2.46 floor doesn't have it.
**A task that paused for tool approval before the upgrade resumes without a
journal.** `resume_after_tool_approval` (the non-durable transcript path) never
decides the mode, so it resumes the way it paused.
## Migration path
Most retried agent tasks need nothing. Three cases change on a retry:
- A tool call the failed attempt completed is not run again. A tool whose
effect doesn't outlive the attempt (a temp file, an in-memory session), or
whose answer has to be fresh, needs `metadata={"airflow": False}` (per tool) or
`durable=False` (per task).
- The agent's `run_id`, and its `conversation_id` when there is no message
history, stay the first attempt's across retries.
- Each completed step is one task state store write, and the journal sits in
the metadata database until the agent run succeeds. Model responses are stored
as returned; tool results are masked. Anyone who can read the task instance can
read it through the REST API. The durable docs page now covers what it stores,
who can read it and when it is deleted.
`default_args={"durable": False}` also opts out, but it turns off `durable`
on every resumable operator in the Dag too (KPO, Spark, Databricks, Glue, ...);
the docs say so.
The cost, from the measurement two PRs down (scripted model, `LocalExecutor`
+ Postgres on one machine): 21 to 24 ms added per journal step, 810 Execution
API calls instead of 7 for a 200-request run. That is small next to a model
request; a run that makes hundreds of millisecond tool calls feels it more.
## Gotchas
- **Release type.** This ships in common.ai 1.2.0, a minor release, with the
`.. warning::` changelog entry, as amazon 9.35.0 did for the Glue `durable`
flip. `docs/stability.rst` lists `durable` among stable features and says
changing a default needs a major release; shipping this in 1.2.0 is an
announced exception to that policy, which stays as written.
- The sandbox refusal looks at capabilities before pydantic-ai binds them to
the agent (`for_agent`). A custom capability that contributes a
`SandboxToolset` only once bound would run durably. No capability in the
provider does that.
- The changelog notes one narrow upgrade case: a task that used
`durable=True` on 1.0.0/1.1.0, whose failed attempt could not write its run
list, runs its steps live once after the upgrade.
The docs sweep touches every page that described durable execution as opt-in
(`durable=True`), and fixes `why_common_ai.rst`, `stability.rst` and
`toolsets/index.rst`, which still said tool approval doesn't work under durable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
