[hussein-awala]

Summary

Fix stale Dag-version associations on waiting task instances when an unversioned Dag changes during an active run.

For a local, unversioned bundle, a downstream task can start after its upstream task but still display an older Dag version. The upstream task's version reference is updated while the waiting downstream task's reference is left unchanged.

Root cause

For runs without a pinned bundle version, _verify_integrity_if_dag_changed() updates unfinished task instances using:

TI.state.in_(State.unfinished)

Although State.unfinished includes Python None, SQL IN (..., NULL) does not match rows whose state is NULL. Task instances waiting for their upstream dependencies can therefore be excluded from the version update, while running or deferred tasks receive the latest version.

Subsequent integrity checks can return early once the latest version is associated with any task instance or task-instance history record in the run, allowing the stale association to persist.

Generic reproduction

1. Create an unversioned Dag containing upstream >> downstream and start a run.
2. Leave upstream running or deferred, with downstream still in state=None.
3. Update the Dag and persist a new serialized Dag version.
4. Run the scheduler's scheduling pass.
5. Observe that upstream references the new version but downstream retains the old one.

Expected: both unfinished task instances reference the new Dag version, including the downstream task before it becomes schedulable.

Changes

• Explicitly include TI.state.is_(None) in the existing unfinished-task update filter.
• Add a database-backed, parametrized regression test for running and deferred upstream tasks.
• Verify that the downstream task receives the new version while waiting and retains it when scheduled after upstream success.

The fix preserves the existing Dag/run scope and versioned-bundle guard. It changes task-instance version metadata; it does not add source-file versioning to local bundles or retroactively repair previously inconsistent runs. A displayed Dag version alone is not proof of the source revision executed by a worker using an unversioned bundle.

Validation

Reproduced on a local main checkout at ef9feefd83:

• Before the fix: both new regression cases failed at the downstream version assertion.
• After the fix: all five version-integrity tests passed through Breeze:

breeze run pytest airflow-core/tests/unit/jobs/test_scheduler_job.py -k verify_integrity -vs

The publication branch was then based on newer upstream main. Fast pre-commit checks, including the airflow-core mypy hook, and applicable manual checks passed for the two changed files. Ruff formatting/linting and git diff --check also passed.

---

Was generative AI tooling used to co-author this PR?

• Yes — pi (OpenAI GPT-6 Astra)

Generated-by: pi (OpenAI GPT-6 Astra) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.

[Copilot]

🟢 Approval recommended

The targeted SQL fix is correct and has regression coverage for the reported scheduling scenarios.

0 open findings

What changed in this PR

Fixes stale Dag-version associations for waiting task instances in unversioned Dag runs.

Changes:

• Includes NULL task states in unfinished-task version updates.
• Adds regression coverage for running and deferred upstream tasks.

File	Description
scheduler_job_runner.py	Updates waiting task instances to the latest Dag version.
test_scheduler_job.py	Verifies waiting and subsequently scheduled tasks retain the latest version.

🧠 Review effort: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.