Fix cleared tasks getting stuck when a Dag run has no version

[jedcunningham]

A Dag run can have no version of its own: runs carried over from Airflow 2 predate the version columns, and airflow db clean can remove the version a run was created with. The latest is then the only version such a run can use, but clearing left it without one, and a task instance with no version is never enqueued — so the task sat in queued until it timed out, while its revived run held up new ones.

Clearing a run with no version of its own now moves it and its task instances onto the latest version together. A task instance with no version whose run does have one joins its run's version instead — the two must not drift apart. Such a run was also rejected outright with a 404 when its version had been removed but its bundle version kept, because version resolution then returned nothing at all; it now falls back to the latest.

The clear dialogs also show the "run with latest bundle version" option for these runs, checked and disabled, since previously they only offered it when there was a version to compare against — leaving no workaround. That decision follows the run, not the individual task instance, for the same reason as above.

Verified

Reproduced end to end rather than simulated: real Airflow 2.11.2 against Postgres, 13 scheduled runs executed by the Airflow 2 scheduler, then a real airflow db migrate to Airflow 3. That leaves dag_run.created_dag_version_id, dag_run.bundle_version and task_instance.dag_version_id all NULL, which is the state this fixes.

Before the fix, clearing a task on a migrated run:

[warning] TaskInstance legacy_dag.second scheduled__2026-08-05... does not have a dag_version_id set, cannot be enqueued
second | queued     <- never handed to the executor

After, on the same database:

second | success | dag_version_id=01a00205-... | try_number=2
dag_run | success
enqueue refusals: 0

The airflow db clean route was confirmed the same way: a run whose task instances the scheduler had moved to a newer version, leaving the original referenced only by the run, has its version deleted and created_dag_version_id nulled.

Re-run from scratch after review, on a fresh Airflow 2.11.2 → airflow db migrate → Airflow 3 database (104 migrations; run and both task instances land with every version column NULL), driving the REST API the way the UI does:

	with the fix	with only the first commit
clear a task on a version-less run	re-ran to success on try 2, run and task instance pinned together	same
new Dag version, then clear the version-less sibling	took its run's version	took the newest, leaving the run behind
version removed, bundle version kept	clear accepted, both tasks re-ran	404 Dag with id ... was not found

Worth noting for the second row: on a non-versioned bundle the scheduler then pulls the task instance to the latest anyway, since _verify_integrity_if_dag_changed is gated on not dag_run.bundle_version. That is existing behaviour and is not changed here; the invariant this PR fixes is what the clear itself writes, which was verified with the Dag paused so nothing else could touch the row.

UI verified in a browser against that migrated database — the option renders checked and disabled on version-less runs (both the run and task instance dialogs) and is correctly absent on runs that have a version.

---

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

• Yes — Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

[dstandish]

this change looks ok

HOWEVER

this function is getting insane

there's 100+ lines of nested if statements in clear_task_instances that are completely indecipherable to a human. we are completely dependent on AI+tests to know what the hell it is doing and whether it's doing what it's supposed to. it's getting increasingly fragile with every conditional that we add and i'm sure eventually it will bite us

additionally, the column created_dag_version_id(which this PR is using and extending) was supposed to be an immutable stamp of the version the dag run was created with. now it's updated and means "whatever the version is for the dag run at this particular moment" and it is therefore misnamed now. it is also possible that there is functionality that is dependent upon the old meaning / contract that is now broken.

[jedcunningham]

All fair points.

This does add one more consumer of created_dag_version_id, but I don't want to tackle that bigger problem here and will leave it to #71453.

[github-actions]

Backport successfully created: v3-3-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-3-test