* [main] CI: Upgrade important CI environment
* Ignore pydantic-ai's generic OpenAI-compatible provider in vendor drift checks
pydantic-ai-slim 2.63.0 adds a provider for any OpenAI-compatible endpoint. It needs a service name at construction and infer_model has no prefix for it, so a common.ai connection can never reach it and it is not a vendor the services page should list.
Generated-by: Claude Opus 5
---------
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
* [main] CI: Upgrade important CI environment
* Bump pypdf to 6.21.4 to fix high-severity advisories
The CI environment upgrade resolved pypdf 6.21.3, which the dependency
review gate rejects (GHSA-mfr5-77xh-9j7h, GHSA-cggf-277g-3q54,
GHSA-5849-834g-hvw5). The fixed 6.21.4 is still inside the 4-day
exclude-newer window, so add a temporary per-package cutoff for it.
Generated-by: Claude Opus 5
* Support pydantic-ai 2.60+ and opensandbox 1.2 in common.ai provider
pydantic-ai 2.60 records a durable streamed model request as the bare
response rather than a response-plus-events wrapper, so replaying or
tracking a streamed request in AirflowDurability failed to decode it.
The same release answers a tool retry with the call's own ToolReturnPart
(outcome "retried") instead of RetryPromptPart, which broke tests that
told retries from results by part type.
opensandbox 1.2 allows an egress rule with ports and no target. When the
enforced policy did not match the requested one, sorting the allowed
targets for the error message raised TypeError on that None, hiding the
real reason the sandbox was refused.
Both work with the older releases too, which the provider still supports.
Generated-by: Claude Opus 5
* Fix provider tests and type checks after the dependency upgrade
The tool-part predicates in the common.ai tests returned a plain bool, so
mypy could not narrow the message part they matched and the providers
type check failed.
On Python 3.14 the upgrade moves snowflake-connector-python to 5.1 and
snowflake-sqlalchemy to 2.1. connector 5.1's connect() no longer goes
through SnowflakeConnection, so patching that class stopped intercepting
the connection and the test tried a real login. snowflake-sqlalchemy
2.0.0 started percent-encoding "?" in the URL password, which the URI
test pinned to the older, unencoded form. Older Python versions keep the
earlier releases, so both tests have to accept either.
Generated-by: Claude Opus 5
---------
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
The Google provider installs a stub-only package for users because an earlier mypy workaround was placed in its runtime requirements. The development mypy extra already supplies the stubs, so retain the broken-release exclusion there.
* Allow FastAPI 0.137+ by requiring cadwyn 8.1
FastAPI was capped below 0.137 because its lazy router inclusion broke
cadwyn 7's versioned router generation. cadwyn 8.1 supports it, so lift
the cap and raise the cadwyn minimum. cadwyn 8.1 itself requires
fastapi>=0.137.1; raise our floor to 0.137.2, the first release with
fastapi.routing.iter_route_contexts, which the test helper relies on.
Fix what the newer FastAPI exposed: use pydantic's Discriminator for the
TIStateUpdate union, make Execution API trace-context injection reach
routes inside included routers (it was silently a no-op), and update
tests that relied on the removed get_flat_dependant or on flattened
router.routes.
related: #68562
* Stop FastAPI's native telemetry and lazy routers changing API behavior
FastAPI 0.142 added native OpenTelemetry that is on by default and runs
before routing and auth. With a tracer provider configured it extracted
caller trace context on unauthenticated requests and exported query
strings unredacted, bypassing [execution_api] otel_trace_propagation.
Exclude every request on the apps Airflow builds, which also stops the
apps mounted under them from tracing, and raise the FastAPI floor to
0.142.0, the first release with that option.
Lazy router inclusion also broke code reading app.routes: HTTP metric
route tags lost their router prefixes, the private UI OpenAPI spec
regenerated without its /ui routes, and the edge3 worker API route
tests saw no routes.
related: #68562
* Detect Execution API auth from enclosing routers and drop cadwyn changelog
Trace-context extraction in only-authenticated mode now checks each route's
merged dependencies, so authentication added by an enclosing router or an
include_router() call still counts. Walking routes through FastAPI's public
iter_route_contexts also avoids relying on its private included-router type.
Cadwyn 8.1 deprecates the changelog endpoint and warns whenever it is enabled;
the Execution API never used it.
0.3.16 stops pinning the docs' right-hand contents 263px from the top
(apache/airflow-site#1701), so it lines up with the content and the
left sidebar.
custom.css still padded the docs layout by 163px and offset anchor
targets by 161px, sized for a 40px banner above a 121px navbar. The site
now has no banner, a 90px navbar (80px at 1280px and below), and the
theme already pads the body for the navbar, so docs content started
253px down instead of 130px.
Drop the padding override so the theme's 40px applies, and size the
anchor offset to the navbar plus the same 40px.
0.3.15 spaces sidebar entries so wrapped titles read as one entry, and
lists only pages in the left sidebar, leaving page headings to the
right-hand TOC (apache/airflow-site#1700).
* [main] CI: Upgrade important CI environment
* Fix "prek run update-providers-build-files --all-files"
* Fix CI breakages from the dependency upgrade
- snowflake: snowflake-sqlalchemy (up to 1.11.1) cannot be imported on
SQLAlchemy 2.1, which made ORMSelectCompileState private. Cap
sqlalchemy<2.1 in the provider; the workspace lock resolves back to
SQLAlchemy 2.0.54. Removing the cap is tracked in #75118.
- airflow-core: type the dag_display_name and duration hybrid expressions
against the class (type[...] -> ColumnElement) and fix _get_dialect's
return type, so mypy holds on SQLAlchemy 2.0 and 2.1.
- common.ai: declare the OpenAI Decisions and System One vendors added in
pydantic-ai 2.53/2.55; reject sqlglot >= 30.22's exp.ListStage
(Snowflake LIST/LS @stage) on the allow_writes path; adapt tests to
pydantic-ai's streamed Anthropic requests and new request counting, and
to sqlglot's MySQL "--+" tokenization.
Generated-by: Claude Opus 5
* Type Backfill.duration expression against the class
#75078 dropped the type: ignores in this expression, which only held
while mypy did not check the body; typing cls as type[Backfill] keeps it
clean on SQLAlchemy 2.0 and 2.1, as dag_display_name and
DagRun.duration already are. Also annotate the Anthropic stream fixture
so mypy can index its events.
Generated-by: Claude Opus 5
---------
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Wei Lee <dev@wei-lee.me>
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
* Read XCom without object reconstruction during dependency evaluation
The scheduler evaluates task dependencies in-process, and two of those deps read
XCom: NotPreviouslySkippedDep reads skipmixin_key for every parent that inherits
from SkipMixin, and both it and TriggerRuleDep read past_depends_met when
wait_for_past_depends_before_skipping applies.
Both went through TaskInstance.xcom_pull -> XComModel.deserialize_value, which is
json.loads(value, cls=XComDecoder). XComDecoder's object_hook calls the Airflow
serialization module with full=True, so any nested dict carrying __classname__ is
imported and instantiated. The values these deps actually consume are primitives:
a mapping of lists of task ids, and a boolean. Nothing here needs an object
rebuilt, so the scheduler can read these values as plain JSON.
Add XComModel.deserialize_value_primitives, a plain json.loads with no
object_hook, and LazyXComPrimitivesSelectSequence alongside it. Thread full=True
through TaskInstance.xcom_pull and pass full=False at the four dep call sites.
All three of xcom_pull's return paths are covered -- the direct one, and both
lazy-sequence ones, which is the path a mapped task instance takes. The skip
decisions of SkipMixin tasks in the same mapped task group, which
NotPreviouslySkippedDep reads with one query per pass, are decoded the same way.
NotPreviouslySkippedDep also now checks that the skipmixin payload is a mapping
before testing membership in it. It reads whatever the parent task wrote, so a
scalar there previously raised TypeError inside dependency evaluation rather than
being ignored. Because the payload is now plain JSON, a set or tuple of task ids
arrives as its serde wrapper; its items are read from the wrapper without
constructing anything, and a decision whose task ids are not a list of strings is
ignored the same way.
Task code pulling another task's return value is unchanged and still gets the
original Python object; only the dependency-evaluation reads are narrowed.
Generated-by: Claude Opus 5
* Treat only a literal True as past-depends-met during dependency evaluation
Generated-by: Claude Opus 5 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Ignore skip decisions with a non-string classname
Generated-by: Claude Opus 5
Every iteration of a task inside a loop is its own task instance, but an XCom
backend's `serialize_value` only receives dag_id, run_id, task_id, map_index
and key, which is shared by different iterations of the same task. A backend
might naievly derives its storage path from those values would every iteration
to the same location, so every read would give back the last iteration's value
n times. It only shows when the XComs are read from outside the loop, which
makes it easy to miss.
To work around this, without breaking old XCom backends, I have introduced
a new method on the backend, `store_value`, and instead of 4 (or 5 or 6)
arguments it takes a single arg of type `XComCoordinate` to prevent arg sprawl
and to make this new fn future proof.
`XCom.set` on another task (i.e. passign a different `task_id`) cannot know
which task instance the server will resolve its coordinates to, so there the
id is unknown and the location has to be unique to the write. Leaving that
distinction to each backend would make the collision easy to reintroduce, so
`XComCoordinate.build_path` builds the location for both cases and shortens
the readable parts only when the path would exceed the backend-provided key
limit. `serialize_value` already combined converting a value with writing
it somewhere, and one more keyword would have kept growing its signature.
For each old and new hook fn, the override closest to the class wins, so a
subclass of a store_value backend that overrides serialize_value keeps its
override. Airflow can't tell whether a legacy backend's locations are unique,
so implementing store_value is what marks a backend as safe in a loop: a Dag
that uses a loop fails at startup under a serialize_value-only backend instead
of overwriting results silently.
Writes and deletes of another task's XCom did not follow the rule reads
use: they resolved to the caller's own task instance or the single live
one, and failed with 409 when several iterations of a loop task were live.
Resolving them like reads, from the calling task instance, lets a task in
a loop write its own iteration of a sibling and gives a caller outside the
loop the same 400 a read gets.
The backend's store and load talk to external storage, so the async paths,
which share an event loop with the triggerer's other triggers, run them in
a thread.
The trigger runner already carried its task instance id in a context
variable so the API could resolve a trigger's requests. It moved into the
Task SDK and holds the task instance, because the SDK has to recognise when
a trigger writes its own coordinates and then pass the id and region on.
The common.io ObjectStore backend now builds its object paths with build_path, so
they now sit under `ti-<id>` or `write-<rand_uuid>`. Rows already written keep
working because they store full paths in the metadata, and read just reads the
path it is given.
Overwritten and failed writes still leave their old objects behind, as they
do today; cleaning those up is separate work. The configured backend also
applies only to the XComs that Python tasks and triggers write and read.
Tasks written in other languages still store theirs directly in the
database, and fixing that needs the supervisor to apply the backend for
them. Again, this doesn't change from today.
airflow.apache.org is served with the ASF Content-Security-Policy, whose
script-src does not allow cdn.jsdelivr.net. sphinxcontrib-mermaid's default
output imports mermaid from there, so every mermaid diagram in the docs has
shown as raw source text on the published site since the first one was added.
Diagrams are now drawn into SVG files by mermaid-cli during the docs build.
mermaid-cli and Chromium are installed into the breeze container only when a
package being built contains a mermaid diagram, so the CI image is unchanged.
Models name each other in string relationships, which only resolve once
every model module is imported. The before_configured hook that imports
them all was registered in configure_orm(), so a process that skips it
(virtualenv-mode subprocesses, a task re-executed for run_as_user,
non-DB tests) or configures mappers before it runs (local settings)
only worked if it had already imported the right model modules. If it
had not, building a model failed, for example with Team.dag_bundles
unable to find DagBundleModel.
Register the hook in airflow.models.base, before any model class is
defined, and drop the test-only copy from the pytest plugin. Remove the
orm_event_handlers test: it guarded the line that moved, and now passes
whatever the code does.
* Depend on httpx2 instead of httpx in the Task SDK
httpx has stalled: fixes that landed in httpx2 (a fork of httpx 0.28.1)
have not shipped upstream, and starlette has already migrated, so FastAPI
following is a matter of time. Moving the Execution API client now keeps
the SDK on a maintained HTTP stack.
Objects do not cross the httpx/httpx2 boundary, so the in-process
Execution API transports move with the client that consumes them; the
rest of airflow-core stays on httpx and migrates separately.
related: https://github.com/apache/airflow/issues/70522
* Add newsfragment for the Task SDK httpx2 migration
* Match the in-process test client transport to the installed Task SDK
Compat tests pair the repo's devel-common helpers with a released Airflow
whose Task SDK is still on httpx. Pinning the helper to httpx2 hands those
clients a transport from the wrong package, and httpx rejects the response
stream it gets back.
* Rename newsfragment to the new PR number
* Pin the Task SDK ast-grep hook to a frozen revision
Every other third-party hook in the repo pins an immutable SHA; a mutable
tag lets the hook change under CI without a diff.
* Remove unneeded newsfragment
* Verify Task SDK HTTP logging and simplify test helpers
Both HTTP stacks can coexist while compatibility jobs use older SDK
distributions. The migration needs regression coverage for its logging
behavior without exposing development-only helpers as public APIs.
* Use httpx2 in the Dag run note client test
* Use httpx2 in the execution API shutdown tests
Tests added on main for in-process shutdown still referenced httpx, which this branch no longer imports in the module.
* Trust the system certificate store in the Task SDK API client
httpx2 verifies against the system trust store via truststore, so keeping certifi for the Execution API client would add a direct dependency only to maintain a second, different set of trusted roots in the same worker.
* Ban httpx imports in the Task SDK with ruff
Ruff already enforces banned imports, so a separate ast-grep hook and its extra pre-commit dependency are unnecessary. extend-banned-api adds to the root banned-api table instead of replacing it.
* Use httpx2 in the server termination acknowledgement test
The test arrived on main after the Task SDK moved to httpx2 and still referenced httpx, which the module no longer imports.
* Test the Task SDK SSL context against a real CA file
Mocking load_verify_locations while also writing a CA file to disk tested neither path convincingly. Keeping the certificate-generation imports inside the fixture limits them to the one place that needs them.
* Note when the in-process httpx resolver can be removed
* Apply suggestion from @ashb
Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>
* Use httpx2 in the supervisor launch error test helper
The helper arrived on main after the Task SDK moved to httpx2 and still referenced httpx, which the module no longer imports.
* Remove the duplicated in-process httpx resolver note
* Use httpx2 in the supervisor execution timeout test
The test arrived on main after the Task SDK moved to httpx2 and still referenced httpx, which the module no longer imports.
* Use httpx2 in the request handler tests
The tests arrived on main after the Task SDK moved to httpx2; the client rejects httpx transports.
---------
Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>
Previously to see the same error in docs build as you would get in CI,
you would have to run both the docs build and the spellchecking build.
The issue was that the spellcheck (using sphinx-contrib's spelling) was
defined as a new builder, meaning you could either build HTML, or you
can check spelling. On CI, this didn't result in any extra wall-clock
time as the jobs were parallelized (but this change does remove a job
from the overall matrix), but locally that was almost always not the
case, so this would mean full build to view the HTML, and spell checking
build to look for errors, resulting in it taking twice as long.
The upstream spellchecker was also slower than it should be mainly
because its importable-module filter calls find_spec on every distinct
identifier-like word before consulting the dictionary. Here it runs only
for words that are still misspelled. Word lists, filters and
`:spelling:ignore:` behave as before.
This also handles errors a lot more nicely (hence all the code deletions):
- They are written natively to a JSON file, not scraped from a stream
- They also show up as GH annotations (Where we can attribute it to a
real file) to make fixing them easier on PRs.
Please note: this is lightly breaking for Airflow contributors as
`breeze build-docs --docs-only` and `--spellcheck-only` are gone --
every build checks spelling now as it is essentially free (on the order
of milliseconds per module).
mask_secret() imported airflow.sdk.execution_time.task_runner only to check
whether SUPERVISOR_COMMS is set. settings.initialize() masks sensitive config
values at startup, so every CLI command and every scheduler, API server and
dag-processor start paid for importing the whole task runtime, including a
second provider discovery for the SDK configuration. Every process that sets
SUPERVISOR_COMMS has already imported task_runner, so it is now read from
sys.modules instead.
* Make non-DB tests independent of the task runner being imported at startup
Masking a config value at startup no longer imports the task runner, and
non-DB tests had quietly relied on that import:
- configure_orm() never succeeds against the bad connection string non-DB
tests use, so the hook that imports every model before SQLAlchemy
configures mappers was never installed. The task runner import was what
registered DagBundleModel, so Team.dag_bundles stopped resolving and the
first ORM object built in a worker failed every later test. Install the
same hook in the pytest plugin for non-DB runs.
- test_execute_deferrable patched builtins.open for the whole process. The
task runner is now first imported inside execute(), so the metrics
registry read the mocked file instead of its YAML. Patch open only in
the module that reads the kube config.
The plugin located a task instance by (task_id, map_index) and stored its review
state as XCom under that address. With loop passes sharing a task id and map
index that picks an arbitrary pass, so approving or rejecting could resolve
another pass's session and write the human action to it.
The plugin endpoints now take region_id and region_index and resolve them by
the same live-producer rules as the public API, so an ambiguous selection is
rejected instead of guessed. Reading, writing and checking completion all use
that exact attempt. The review UI forwards the selected task instance's region
coordinates when the host provides them. The plugin also runs on hosts that
predate regions, where an unknown region_id would break the request, so a host
without task instance context keeps the map-only request shape. A modern host
that has not resolved a selection renders no session rather than opening another
task's.
Tests that re-fetch a task instance to run it need the region too, otherwise
they run another pass's row.
The scheduler, the API and the UI must never see every task of a loop finished
while its next pass does not exist yet. If we recorded a continuing gate
as successful and created the next pass afterwards, a crash between the two
would leave recovery unable to tell an unfinished advancement from a completed
decision. Gate completion and next-pass creation therefore happen in one
transaction under the DagRun lock, which also stops a duplicate completion from
advancing the loop twice.
The continue-or-stop decision is a reserved XCom stored under the gate's attempt
UUID. A retried gate starts without a decision, and a retired attempt cannot
advance the loop. An invalid decision moves the gate attempt to retry or failed
within the same request; rejecting the request instead would leave the attempt
running with nobody to finish it.
Gates use normal task dependencies and an explicit loop context. A barrier
that waited for every body task would override trigger rules that allow early
progress. Workers and dag.test supply the same coordinates, so decisions and
previous-iteration reads reach the intended gate and data.
Skipping downstream of a task from outside a loop skips every live pass.
Skipping only one would let later passes run work the author meant to skip.
Mapped tasks were the one kind of dynamic work still outside the region
model: their instances sat in the sentinel region and were told apart only by
map_index. A loop has to be able to contain an expansion, one per pass, and that
only works if the expansion owns a region of its own.
The coordinate column is renamed in place so large deployments pay for metadata
changes, not a table rewrite. No table is rebuilt because SQLite's rebuild would
cascade-delete the rows that reference task_instance.id. legacy_task_data_owner,
xcom_v1 and rtif_v1 keep map_index because they record the coordinates their
rows had before regions existed. The ORM keeps map_index as an alias: Jinja
log-filename templates and provider code still read ti.map_index from ORM rows,
and removing it needs its own audit.
Mapping keeps its existing behaviour. A task's region is born with its
placeholder, re-expansion grows or shrinks it in place, and a map never forks.
Promoting the placeholder to index 0 stays in place too. Attempts retired before
the promotion stay at index -1: they own their XCom, rendered fields and log
files by UUID, and moving them could collide with retired attempts already
at index 0. A pre-region placeholder never ran, so it has no history to keep;
it moves into a freshly minted region and is promoted like any other, instead
of being retired as a phantom failed try. Pre-region expansions stay in the
sentinel region and keep revising in place, so the upgrade does not rewrite
existing rows.
Callers that address a task instance by public coordinates now find a task's own
region as well as the sentinel. The DagRun wait endpoint reads current producers
only and rejects loop results, because which pass's result to return would be
ambiguous.
* Require Python 3.11 in all distributions
Every pyproject.toml that declares requires-python now says >= 3.11 and
drops the Python 3.10 trove classifier, the tomli build requirement that
only existed for 3.10 and the per-version dependency markers for 3.10.
Lock files are regenerated.
With 3.10 gone, the tomli fallbacks in scripts, dev tooling and tests go
too, along with the tomli additional_dependencies of the prek hooks and
the blacken-docs py310 target. The tomllib imports that only sat inside
functions because of the fallback move to the top of their modules. The
four dev/registry and scripts/ci/prek modules left out of the earlier
UP017 pass also take their timezone.utc to datetime.UTC fixes here. The
minimum host Python checked by check-min-python-version, the default
Python of run_generate_constraints.py and the provider cut-off date
parsing (fromisoformat, which reads a trailing Z from 3.11) follow the
new baseline.
ruff infers py311 from the new requires-python, which enables the
PEP 646 star-index fix (UP044) applied here and makes two more rules
reportable: UP042 (str, Enum to StrEnum) alters what str() and
format() return for serialized enum values, and ASYNC109 means
dropping the public timeout parameter from async APIs. Both need
per-site judgement, so they go on the ignore list with that rationale.
The datamodel-codegen target moves to 3.11 too, with use-specialized-enum
turned off for the same reason: otherwise the generated Task SDK and
airflowctl enums would switch from (str, Enum) to StrEnum.
* Drop the Python 3.10 provider metadata shim
With provider distributions now requiring Python 3.11, breeze can derive
requires-python and classifiers from the default and allowed versions
again. The patch-exclusion test and the matching example in
MANAGING_PROVIDERS_LIFECYCLE.rst move from 3.10.0 to 3.11.0, the oldest
patch release CI still covers.
* Keep Breeze's locked dependency versions when dropping Python 3.10
Relocking Breeze for the Python 3.11 baseline ran while recently bumped versions were still inside the 4-day exclude-newer window, so it downgraded pyjwt (reverting #73969), prek, boto3/botocore and filelock. Relocking from the previous lock keeps them and only drops the packages Python 3.11 no longer needs.
Generated-by: Claude Opus 5
---------
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
Output of ruff --fix with the py311 target for airflow-core, task-sdk,
shared, airflow-ctl and devel-common: timezone.utc becomes
datetime.UTC and typing_extensions imports move to typing. No
behaviour change. Landing these before the requires-python bump keeps
lint green at every layer, because ruff infers its target from the
nearest pyproject.toml.
CI no longer runs Python 3.10, so the tomli fallbacks for tomllib, the
typing_extensions imports of Self, Unpack and assert_never, the
re-implementation of asyncio.Runner in the Task SDK and the version
gated test skips no longer have a version to serve.
This is a data-model change precursor for the Task Loop implementation.
The subject sounds simple, but unfortunately this change is far more complicated
than I'd like, but for a couple of good reasons.
First though, why do we want this: to support Task Loops (AIP-111) which lets
the number of loops of a task group be decided, and created just-in-time at
runtime. All good so far, and no changes needed far.
However things change when you get to clearing TIs. Task Loops are most useful
when the output of one loop feeds in to the next. So the only sensible default
is that when you clear mid-way through a loop, say you clear iterations 2, 3, 4,
and 5, we restart and run 2' onwards. Since it is a runtime decision is is no
guarantee how many it will run, and most crucially, we don't want to make the
previous TIs invisible.
Visually, this looks like (or [png][fig1])
```
+--------+ +--------+ + - - - -+ + - - - -+ + - - - -+
| iter 1 |------| iter 2 |- - - : iter 3 :- - - : iter 4 :- - - : iter 5 : old attempts
+--------+ +--------+ + - - - -+ + - - - -+ + - - - -+ keep them + their XComs
|
clear at 3 |
|
| +--------+ +--------+ +--------+
'----->| iter 3'|------| iter 4'|------| iter 5'| re-run
+--------+ +--------+ +--------+ new attempts, new outputs
```
[fig1]: https://github.com/user-attachments/assets/cbbf3ea6-499b-42fe-8cd8-955c3f5310ff
(where each box there is a set of tasks from a looped task group for AIP-111)
We could choose to just keep the retired/superseded TI in the existing
task_instance_history and delete/loose the XCom and other data.
But that struck me as very sub-optimal, and I'm using this to also fix a long term
pain point: that having TI and TIHistory tables separate causes some other odd quirks.
```
+---------------------------------+
| task_instance |
+------>| one row per task = latest try |<--------------+
| +---------------------------------+ |
| FK on |
| copied on retry (dag_id, task_id, |
| run_id, map_index) |
+- - - - - - - - - - - - - -+ +------------------------+
: task_instance_history : | xcom · rtif |
: older tries, copied out : | no try_number column |
+- - - - - - - - - - - - - -+ +------------------------+
```
> XCom points at "whoever is current at these coordinates". When the task retries or is cleared, the old attempt's outputs have nowhere to live.
[fig2]: https://github.com/user-attachments/assets/018d0611-8bef-48d0-8467-8568d7832622
So why the `is_workingset` column? This is to keep the scheduler performant. The
current main reason to have TI and TIHistory separate is so that the scheduler
can
Now to the gnarliest point of this change: xcom_v2 et al. Why do this? Simply
put, it's because of the time to migrate the data. With the new FK on TI, we
would need to update _every single tuple_ on XCom table. I ran some tests on a
copy of a production database with 309 million XCom rows on how long it would
have taken to migrate XCom to store the creating task_instance_id on it: I gave
up and cancelled it after 14.5 hours. 309M is larger that most people will have,
but more than 14.5 hours is simply not worth it.
So instead I've got for the "generational table" approach, where we write to a
new table, and read from both. It's slightly more complicate than that due to a
few shadowing cases etc, but that's the gist of the approach.
Now, I hear you asking: "Doesn't this mean we have to carry this legacy code
path until Airflow 4 at an unknown time in the future?" Nope, I've got this
covered, just not included in this PR. In a future PR (likely landing for 3.4.x
or 3.5.0) I plan on having the scheduler "slowly" (but not too slowly) migrate
chunks of the data from the v1 table to v2, and then in 3.6 we add a migration
that does the final "move all data form xcom_v1 to xcom_v2" and document "You
need to upgrade to 3.4.x/3.5.0, and wait for X time, before upgrading to 3.6,
else expect a very slow migration" (some status in the admin screens to be added
to make this visible.)
A few notes:
- On Postgres, the migration creates a new unique constraint in the "NOT
VALID" state. This is safe because that migration was already covered
by an existing constraint we dropped. NOT VALID means that Postgres
hasn't checked all the current rows, but that it will apply it do an
insert or update going forward. This is safe for us, and saves ~20mins
on the large DB I was testing on.
- The "legacy-owner" table is created mostly so that rows in xcom_v1 et
al. still have a FK that will be caught by ON DELETE CASCADE.
```
+----------------------------+
| task_instance |
+--------->| one UUID per attempt |<--------------+
| +----------------------------+ |
| |
| task_instance_id | task_instance_id
| |
+---------------------------------+ +------------------------------------+
| legacy_task_data_owner | | xcom_v2 · rtif_v2 |
| task coords → UUID, 8.2M row | | all new writes, empty at upgrade |
+---------------------------------+ +------------------------------------+
^
| FK on task coords, NOT VALID
|
+---------------------------------+
| xcom_v1 · rtif_v1 |
| ~300M old rows, untouched |
+---------------------------------+
```
[fig3]: https://github.com/user-attachments/assets/164861a5-cf6b-44f4-a850-1eadcbc06203
* Keep builtin annotations from resolving to project attributes in Sphinx 9
Sphinx 9 falls back from a class lookup to a fuzzy data/attribute search
when resolving annotation cross-references. A builtin such as type or
object in an annotation then matches every documented attribute with
that name, and the docs build fails with "more than one target found"
(sphinx-doc/sphinx#14223). Python 3.11 and newer resolve Sphinx 9 from
the lock file, so the docs build breaks as soon as it moves off 3.10.
Skipping that fallback for builtin names restores the Sphinx 8 behaviour
of linking them to the Python documentation.
Claude-Session: https://claude.ai/code/session_01SkLWWaTT1cnFqTT1jhFGxe
* Type the Sphinx domain override for both Sphinx 8 and Sphinx 9
The docs build still type-checks against Sphinx 8 on Python 3.10, where
PythonDomain.resolve_xref returns Element | None, while Sphinx 9 narrows
it to reference | None. No single precise annotation is a valid override
for both.
A delayed executor event must identify the attempt that produced it after
clearing or retrying replaces the live task instance. Reconcile scheduler
events by UUID and retain the retiring identity through restart completion
and heartbeat cleanup.
Core upgrades can retain older executor providers, so preserve their
coordinate-keyed interfaces. Capture UUIDs before submission or adoption
and translate legacy events from those snapshots without looking up the
current row by coordinates. Retain accepted adoptions across empty event
drains, including providers that track adopted work only in private maps.
LocalExecutor opts into UUID keys end to end. Legacy providers cannot
distinguish attempts that reuse the entire coordinate key; features that
require that isolation must require the UUID capability until providers
migrate their submission, adoption, result decoding and cleanup paths.
* Add dag_bundle_name option to subprocess coordinators
Language-SDK coordinators previously required a hand-managed filesystem root (jars_root/executables_root/bundles_root) for compiled artifacts, kept outside Airflow's Dag-bundle machinery and prone to drifting from the Dags they serve. Allowing artifacts to be located through a configured Dag bundle -- or the task's own bundle when none is named -- lets them be delivered and versioned the same way as everything else.
* Document dag_bundle_name option for language-SDK coordinators
* Harden subprocess coordinator artifact bundle resolution
Language-SDK subprocess coordinators materialized artifact bundles without
the version-lock protection the Python task path holds, so bundle cleanup
could delete a version a running JVM was still reading from. Resolution now
holds the version lock for the subprocess lifetime, mirroring the Python task
path, and materialization failures surface in the task log rather than the
worker log.
Classification moves to the base coordinator so a subclass that adds no
explicit-root kwarg falls back to the task-bundle default instead of failing
every task, the Java coordinator now requires an explicit entrypoint when it
has to scan a whole bundle, and configured-bundle lookup checks names only so
an unrelated unimportable bundle no longer breaks coordinator construction.
* Make JavaCoordinator main_class optional in co-located mode
When a subprocess coordinator locates artifacts through a Dag bundle rather than an explicit root, JavaCoordinator previously rejected a missing main_class outright. That made the co-located default unusable without extra configuration and left the entrypoint no more determinate than an auto-detected scan would be. Auto-detect over the resolved bundle now handles that case, with the remaining cross-bundle ambiguity deferred to an IMPORT_ERROR-stage check under AIP-85.
A named Dag bundle resolves to its latest version, so its version was unset and the version lock held around the subprocess silently did nothing, letting bundle cleanup delete artifacts a running task was still reading. The concrete current version is now pinned before locking.
* Resolve coordinator Dag bundles without importing the task runner
A language-SDK task start needs a bundle materialized on disk before the
subprocess can be launched, but reaching that through the task runner made
every such start pay for the operator tree, pydantic, and OTel plumbing that
the supervisor side never uses. The bundle lifecycle now lives on its own, so
the coordinator depends on the Dag-bundle machinery alone.
The bundle-name check a coordinator runs at construction and the class-importing
parse the manager runs at startup previously read the configuration
independently, so the two could disagree about what is configured. Keying the
parsed configuration on the configuration itself lets both share one snapshot
while a configuration change is still picked up.
Subclasses had to declare the name of their explicit-root kwarg separately from
its value, which is two things to keep in step for one piece of information.
* Read the Dag bundle configuration through a single entry point
Splitting the read across a keyed builder and a wrapper meant two names for one
idea, and the empty case leaked out as a module-level constant that existed only
to be returned. One function now answers "what is configured", parsed once per
process so both readers of it see the same answer.
Parsing once per process means a later configuration change is not observed.
That only arises in tests, which switch configuration freely, so overriding
configuration there drops the memo rather than silently serving a stale one.
* Preserve custom subprocess coordinator compatibility
External language SDK coordinators can implement the documented command hook. Keeping that contract stable avoids unexpected-keyword failures when upgrading while allowing Dag-bundle-backed artifact discovery.
* Make task bundle access dependencies explicit
Module-level imports keep dependency analysis reliable and follow Airflow's import-placement contract without changing bundle access behavior.
* Reject concurrent subprocess coordinator execution consistently
Bundle resolution does not need mutable instance state, while overwriting scan roots can make a command build against another task's artifacts.
* Reject empty subprocess coordinator artifact roots
An explicitly empty root is likely a configuration or templating error. Treating it as omitted can silently scan the task's Dag bundle and run unintended artifacts.
* Reject blank subprocess coordinator artifact roots
An empty string normalizes to the current directory, so it bypassed the empty-roots check and scanned the worker's working directory.
A team-scoped plugin's timetables, partition mappers, windows, deadline
references and priority weight strategies were usable by any Dag. Dags are
now checked when their parsing results are stored: a Dag that uses another
team's class gets an import error instead of being stored.
The SQLAlchemy 2.1.0 sdist ships a pyproject.toml with a duplicate normalized extra (mssql-pymssql), which uv refuses to parse, so the constraints jobs fail building it whenever highest resolution selects it.
Generated-by: Claude Opus 5
* [main] CI: Upgrade important CI environment
* Cap azure-datalake-store below 1 in the Microsoft Azure provider
azure-datalake-store 1.x removed the lib.auth API that AzureDataLakeHook
uses, so the floor bump to 1.0.1 in the CI upgrade broke the hook and
every TestAzureDataLakeHook test. Supporting 1.x is tracked in #73795.
Generated-by: Claude Opus 5
* Regenerate provider requirement tables after the floor bumps
The floor bumps changed provider pyproject.toml files without regenerating
the README.rst and docs/index.rst requirement tables derived from them,
so update-providers-build-files and sync-provider-readme failed static checks.
Generated-by: Claude Opus 5
---------
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
An attempt has a UUID A and try number N: (A,N). Previously, we assigned
a new UUID B when retrying or clearing, but left the scheduler to
increment the number. That left (B,N) in the database: the replacement's
UUID paired with the previous attempt's number.
We want to always allocate both fields together, archiving (A,N)
before creating (B,N+1). The scheduler then uses that allocation:
| Retry lifecycle | Previously | Now |
|-----------------|------------|---------|
| Running | (A,N) | (A,N) |
| UP_FOR_RETRY | (B,N) | (B,N+1) |
| Scheduled | (B,N+1) | (B,N+1) |
When clearing a running task, we must also wait for its process to stop
before allowing the replacement to run. We reject heartbeats with 409
when the TI is RESTARTING, which tells the worker to stop. But we cannot
tell from that response when the worker has actually stopped. Its child
may still be handling SIGTERM, running cleanup, or changing external
state. Starting the replacement then could let both attempts act on
the same external system.
We therefore leave both fields unchanged throughout shutdown:
| Running clear | Identity | State |
|----------------------|----------|------------|
| Before clear | (A,N) | RUNNING |
| Clear requested | (A,N) | RESTARTING |
| Cleared again | (A,N) | RESTARTING |
| Termination reported | (B,N+1) | NULL |
By keeping the UUID unchanged as A, we let the worker report against
the attempt it owns. We also reject start requests during RESTARTING
so another workload cannot start that attempt while shutdown continues.
The supervisor knows the child has stopped once it has reaped it.
We need the supervisor to report that fact to the API before we can
release the replacement without relying on an executor event. It sends
SERVER_TERMINATED through the existing finish request to acknowledge
the shutdown. We do not store SERVER_TERMINATED as a TI state: the
worker is confirming that it stopped as instructed, rather than
reporting a task failure.
Once the supervisor confirms termination, we make the replacement
eligible immediately and restore its retry budget. A clear must allow
an immediate rerun even when the task has exhausted its retries,
without introducing an ordinary retry delay or failure callback.
We still need to support workers back to Task SDK 1.0.0, which cannot
send this acknowledgement. The scheduler therefore accepts terminal
executor events as a fallback while the matching attempt is RESTARTING.
Executor SUCCESS only means the workload finished, so either SUCCESS
or FAILED can complete the clear. The API and scheduler both lock and
recheck the row before allocating a replacement. If the API completes
the clear first, the old executor event no longer matches the live
attempt. We do not need to track the worker version.
Moving allocation earlier must not change retry timing or which logs
users see. During UP_FOR_RETRY, we still calculate backoff and select
logs using the failed attempt N, although the live row now identifies
N+1. We preserve both fields across deferral and rescheduling because
they continue the same attempt. Try zero still means we have not
allocated an attempt.
We also need to bring existing pending rows into the new convention.
Migration 0134 advances UP_FOR_RETRY rows and NULL-state rows with
positive try numbers: the old code had already replaced their UUIDs
but left scheduling to advance their numbers. Rows at try zero with
NULL state remain unallocated.
We use Cadwyn to preserve older API contracts. For current clients,
the API returns 410 for an archived attempt and 404 for an unknown UUID;
older clients retain 404 for both. If a worker loses a response, it can
use 410 to establish that its attempt has ended, but cannot infer that
the server accepted the particular outcome it reported.
We retain the existing heartbeat-timeout and orphan recovery behavior.
Those recovery paths cannot prove that the old worker has stopped.
* Show pending task tries consistently across task tabs
Retries now allocate the next try before it starts, so selecting the
last executed try only in Logs makes tab changes misleading. Keep the
current try as the default and preserve explicit selections across tabs.
Explain why a pending try has no logs, don't try to fetch any logs from
the API server, and offer the previous try instead.
* Fix (and test) RESTARTING processing via executor events
New clients will usually ack the 409 to by reporting `server_terminated`
state, but old clients (and crashed tasks) won't, so we need to make
sure that executor events still correctly progress a RESTARTING TI into
a new try in UP_FOR_RETRY/None.
A Job's team scope was a single nullable column, which cannot describe a
Dag processor that parses bundles belonging to more than one team. Scoping
such a processor to no team at all loses the information that its work is
relevant to each of those teams, so team coverage in the health endpoint
and the team column in the UI cannot be derived from it.
The task_map table was created (by me and TP) years ago to store the
number of results (length) that a task upstream of a mapped task
created. This is then used by the scheduler to know how many mapped
tasks to expand to.
The `keys` column has never been read from (and only got written to in
2.x, but not in any 3.x), it was probably to let people iterate over
tasks that returned dictionaries, but well, that turned out to not be
required.
This is a pre-cursor tidy up to make the PRs for task loops (AIP-111) and
after that dynamic sub-graphs (AIP-113) simpler.
The wire protocol for both public API and Exec API remain unchanged, this
storage on the DB table was purely a server-side storage decision.
This reverts commit ecfa20322b.
Main is broken, so reverting this for now.
Errors:
```
azure-ai-projects 2.7.0 has requirement openai>=3.0.0, but you have openai 2.54.0.
```
A team-scoped plugin's macros were reachable from every task's templates,
so any team could call macros written to talk to another team's systems.
The worker decides this from a new ``multi_team`` flag on the run context
rather than from its own config, which is not guaranteed to match the
scheduler's (misconfigured or intentional attempt to read macros from
other teams): reading it as disabled while it is in fact enabled would drop
scoping and apply a team's macros everywhere. The existing ``team_name``
cannot carry that signal, being ``None`` both for a teamless task and for
every task when multi-team is off, so a plugin declaring a team would
otherwise lose its macros in single-team deployments.
Scoping is applied at attribute access rather than when macros are loaded,
because the loader mutates a process-wide module and a worker process can
serve tasks from more than one team.
Run Sphinx in-process inside each docs build worker so the fixed per-package
startup (autoapi's astroid parsing of the airflow modules, provider.yaml
parsing, the airflow import, inventory loading) is paid once per worker
instead of once per package. A full 115-package build drops from 645 s to
214 s locally with 4 workers (41.6 to 14.2 CPU-minutes); a small provider
now takes about 2 s instead of 17 s.
Make the spelling builder write an inventory so a --spellcheck-only run can
resolve references to pages added in the same change, and retry only the
packages that failed instead of rebuilding everything with html and spelling
again. That rebuild-all pass is what turned a 28-minute spellcheck job into a
93-minute one.
Also schedule the heaviest packages first with one-at-a-time dispatch, parse
each provider.yaml once with a compiled validator and the C YAML loader, and
skip the runner disk cleanup for the docs job (93 GB were free before it).
Runners are not uniform, so the docs job now passes a 40 GB threshold to the prepare action instead of skipping the cleanup unconditionally. Also document why a spelling build that fails on misspellings still publishes its inventory: under -W Sphinx counts warnings and fails at the end rather than aborting.
Refactors `BaseExecutor` workload queue management so that adding a new workload type is a small.
## Problem
Adding a new workload type (like ExecuteCallback or TestConnection) required touching ~6 places in BaseExecutor: a new queue dict, a new `supports_*` flag,slots calculation, an isinstance branch in queue_workload, a dedicated scheduling method, and isinstance branches in dequeue/trigger logic. Each provider executor that overrode `queue_workload` also needed updating. This made extending the executor interface unnecessarily painful.
## What this does
Replaces the per-type queue dicts and boolean capability flags with three primitives:
- `executor_queues`: one `defaultdict(dict)` keyed by `WorkloadType`, replacing `queued_tasks`, `queued_callbacks` and `queued_connection_tests`.
- `supported_workload_types`: a `frozenset[WorkloadType]` replacing `supports_callbacks` and `supports_connection_test`.
- `WORKLOAD_TYPE_PRIORITY` plus a per-schema `sort_key`: the type registry decides ordering across types, the schema decides ordering within a type.
`queue_workload` in the base class is now generic: validate the type, store by key. `trigger_tasks` and `trigger_connection_tests` are merged into `trigger_workloads`, which drains all queue types through a single per-heartbeat budget.
## Adding a new workload type after this refactor
1. Declare the `WorkloadType` member and place it in `_workload_type_priority_order`.
2. Add the schema to the `All` and `ExecutorWorkload` discriminated unions, and override `sort_key` if it needs ordering within its group.
3. Add a branch to `BaseExecutor.run_workload` and to `state_class_for_key`.
4. Add the type to `supported_workload_types` on supporting executors and handle it in `_process_workloads`.
A team-scoped plugin's operator extra links were rendered on every task
instance, so one team's links appeared on other teams' Dags and pointed
users at systems they may have no access to.
Ownership is tracked per link class rather than per link name: link
instances are unhashable and compare equal to each other, and a plugin
link may deliberately share a name with an operator's own link, so a
name key would make the operator's link look team-owned. A class
registered by both a global and a team-scoped plugin stays global.
The ERD reference pages embedded the schema as an SVG image, so table and
column names weren't indexed by search engines or full-text search. The
generate_erd Sphinx extension already regenerates the diagram from the live
SQLAlchemy models on every doc build, so switching its output format doesn't
introduce any new staleness risk. eralchemy (already a doc-build dependency)
can emit Mermaid ER-diagram markup directly, removing the need for the
graphviz system package during doc generation.
Deployments often need to localize the Airflow UI into a language Airflow does not ship, or
override specific strings, without forking the frontend. Plugins can now contribute UI
translations (a <language>/<namespace>.json directory tree or an inline mapping) that are
deep-merged over the built-in ones and served, pre-merged at startup, only for the files a
plugin changes; everything else keeps being served from the static mount. Translations are not
versioned in lockstep with Airflow, so a broken source is skipped rather than failing startup,
and keys absent from the English reference are logged. Right-to-left languages render correctly.
* [main] CI: Upgrade important CI environment
* Keep workspace-relative action references in CI workflows
zizmor 1.30.0 adds the self-repository audit, which wants every in-repo
`uses: ./...` rewritten to GitHub's `uses: $/...` form. Converting the
workflows proved that two things outside this repository's control stop the
new syntax from working here.
`$/` makes the runner fetch apache/airflow as an action repository, and that
download cannot materialise the symlinks this repository commits: it aborts on
`.claude/skills/airflow-translations`, whose target is not tracked, before any
job starts. Separately, ASF infrastructure's allowlist check does not
recognise the syntax and reports every `$/...` reference as an action missing
from the allowlist.
Nor does the audit's threat model reach us in the meantime -- it guards a
privileged workflow against loading a local action out of a checkout untrusted
code could have altered first, and no Airflow workflow runs on
`pull_request_target` or `workflow_run`.
Generated-by: Claude Code (Opus 5)
Claude-Session: https://claude.ai/code/session_01EbprJE5EYQ2Ng7DwAyQdgB
---------
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
github-actions[bot]andJarek Potiuk
Gore Williams
Hussein Awala
Jed Cunningham
Kaxil Naik
Tzu-ping Chung
Ash Berlin-Taylor
ZachandAsh Berlin-Taylor
Wei Lee
Shahar EpsteinandJarek Potiuk
Divyansh Yadav
Dilnaz Amanzholova
Jason(Zhe-You) Liu
Niko Oliveira
Vincent
Amogh Desai
Anish Giri
Jyun-An Chen
Pierre Jeambrun
