Load the correct DAG version when a task starts from a trigger

[seanghaeli]

If the run and the task point at different DAG versions, the trigger should load the run's version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish @hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_run
def get_serialized_dag_model_for_run(
    self, dag_run: DagRun, *, session: Session
) -> SerializedDagModel | None:
    """Return the SerializedDagModel for the version a run executes against."""
    if version_id := self._version_from_dag_run(dag_run=dag_run, session=session):
        return self.get_serialized_dag_model(version_id=version_id, session=session)
    return None

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	def get_dag_for_run(self, dag_run: DagRun, session: Session) -> SerializedDAG | None:
	if version_id := self._version_from_dag_run(dag_run=dag_run, session=session):
	return self._get_dag(version_id=version_id, session=session)
	return None

[jason810496]

Suggested change

	dag_run = trigger.task_instance.get_dagrun(session=session)
	serialized_dag_model = dag_bag.get_serialized_dag_model(
	version_id=trigger.task_instance.dag_version_id,
	version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
	or trigger.task_instance.dag_version_id,
	session=session,
	)
	dag_run = trigger.task_instance.get_dagrun(session=session)
	serialized_dag_model = dag_bag.get_serialized_dag_model_for_run(
	dag_run, session=session
	) or dag_bag.get_serialized_dag_model(
	version_id=trigger.task_instance.dag_version_id, session=session
	)