In-place bundle_version refresh leaves old runs showing the wrong commit and breaks triggering at an earlier commit

[potiuk]

Apache Airflow version

main (reproduced at 5428b3b)

What happened

Since #68336 (fix for #67657), when a bundle advances to a new commit but the Dag's serialized
content is unchanged, SerializedDagModel.write_dag repoints the latest DagVersion.bundle_version
to the new commit in place instead of creating a new DagVersion — even when that version already
has task instances
(serialized_dag.py#L704-L722).

A code-only change — e.g. editing the body of a @task function, which is not part of the
serialized Dag — therefore rewrites history of an existing DagVersion. Two user-visible side
effects follow:

1. Old runs show the wrong commit. A run created on commit A keeps dag_run.bundle_version = A
   and correctly executes A (workloads use dag_run.bundle_version), but GET /dagRuns/{run_id}
   reports its dag_versions[].bundle_version and bundle_url as commit B. The UI "view source"
   link for that run points at code it did not run.
2. Triggering at an earlier commit fails. POST /dags/{dag_id}/dagRuns with bundle_version
   (Add the option to select bundle version parameter on dag run trigger endpoint #61550) looks up a DagVersion row by bundle_version
   (dag_version.py#L176-L178),
   so for any commit that was overwritten in place it returns
   404 ... does not have a version for bundle_version '<A>', even though A was a real, parsed,
   previously run version.

What you think should happen instead

• A run's reported version/commit/source link should be the commit it actually executed
  (dag_run.bundle_version), not whatever the DagVersion row points at now.
• Triggering at a commit Airflow has already parsed and run should work, or fail with a message
  explaining that the commit was superseded in place.

This should not undo #67657's fix. Possible directions (for discussion): derive per-run
version/URL from dag_run.bundle_version; record superseded commits for a DagVersion
(e.g. a small dag_version_bundle_version history table) so lookups by commit still resolve;
or create a new DagVersion when the current one is already referenced by runs.

How to reproduce

Git-backed bundle, a Dag with a @task:

1. Commit A (return "GOOD"), let it parse, trigger a run.
2. Commit B changing only the task body (return "BUGGY"), let it parse.
3. dag_version still has one row, now with bundle_version = B (same dag_hash).
4. POST /dags/{dag_id}/dagRuns {"bundle_version": "<A>"} → 404.
5. GET /dags/{dag_id}/dagRuns/<run from step 1> → bundle_version: A but
   dag_versions[0].bundle_version: B.

Log from a local reproduction using real git commits, the Dag processor's sync_bag_to_db
path and the real REST endpoint:

commit A = 04f74a7423  (compute() returns GOOD)
  v1  bundle_version=04f74a7423  dag_hash=95fd125570
POST .../dagRuns bundle_version=04f74a7423 -> HTTP 200
commit B = 607ad282da  (compute() returns BUGGY; Dag structure unchanged)
  v1  bundle_version=607ad282da  dag_hash=95fd125570
POST .../dagRuns bundle_version=04f74a7423 -> HTTP 404
existing run from A:
  dag_run.bundle_version                     = 04f74a7423
  workload.bundle_info.version               = 04f74a7423
  GET dagRun dag_versions[v1].bundle_version = 607ad282da  bundle_url=.../tree/607ad282da

Note: with the default [core] min_serialized_dag_update_interval (30s), commits parsed within
30s of each other are collapsed as well.

Anything else

Also relevant to AIP-109 (Dag version pinning), where rollback targets DagVersion rows.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Juwon-Ogunseye]

i will work on this

[Juwon-Ogunseye]

Here's my proposed approach, which keeps #67657 fixed:

When the Dag hash is unchanged but the bundle moved to a new commit, write_dag currently updates the latest DagVersion.bundle_version in place, even if that version already has task instances.

write_dag already has a rule for the changed-hash case: update the latest version in place only if it has no task instances, otherwise create a new DagVersion. I'd apply the same rule to the unchanged-hash case:

• No task instances on the latest version → refresh bundle_version in place (same as today).
• Task instances exist → create a new DagVersion for the new commit.

This way:

• Old DAG version used even when selecting Run with latest bundle version on task instance "clear" #67657 stays fixed, since new tasks still resolve the new commit.
• Versions that already ran keep their commit, so old runs show the right commit and triggering at that commit works.

Trade-off: #68336 intentionally avoided new versions on commit-only changes. With this approach, a new version is only created for a commit that follows a run of the current version, not on every commit.

Does this direction sound OK? If so, I'll open a PR with tests covering both cases. @potiuk