Fix dynamic dag_id resolution in TriggerDagRunOperator links

[hwang-cadent]

Fixes #46402 - Dynamic dag_id resolution in TriggerDagRunOperator links

Description

This PR addresses the issue where TriggerDagRunOperator links fail to work correctly when the dag_id is dynamically determined at runtime (e.g., from XCom values or complex templates). The current implementation only uses the static trigger_dag_id attribute, which doesn't reflect the actual resolved dag_id when it's determined dynamically.

Changes

• Added XCOM_DAG_ID constant to store the resolved dag_id in XCom during task execution
• Modified TriggerDagRunLink.get_link() to prioritize XCom values over static operator attributes
• Updated both Airflow 2.x (_trigger_dag_af_2) and 3.x (_trigger_dag_af_3) execution paths to push the resolved dag_id to XCom
• Added comprehensive test coverage for dynamic dag_id scenarios in both Airflow versions
• Maintained full backward compatibility with existing static dag_id usage
• Added proper error handling for test environments where supervisor communication isn't available

Testing

• All existing tests continue to pass (10 tests passed, 26 skipped for Airflow 2.x compatibility)
• Added new test cases for dynamic dag_id resolution in both Airflow 2.x and 3.x
• Verified backward compatibility with static dag_id usage
• Tested error handling for environments without supervisor communication
• Ensured XCom fallback behavior works correctly

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

[hwang-cadent]

@XD-DENG @ashb Could you please review this PR?

This PR fixes dynamic dag_id resolution in TriggerDagRunOperator links (Fixes #46402). The changes involve:

• XCom usage for storing resolved dag_id
• Link generation updates
• Core operator functionality

Thank you!

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[pierrejeambrun]

Something probably went wrong in the rebase process. The change set is huge.

Do you mind fixing the branch please so we can move forward.

[hwang-cadent]

Something probably went wrong in the rebase process. The change set is huge.

Do you mind fixing the branch please so we can move forward.

Rebased on the latest apache/main and cleaned up. The change set now includes only the 4 PR-related files). Ready for review. Thank you for your patience.

[potiuk]

@dabla @uranusjr ?

[dabla]

Looking good to me reviewed it from mobile though so I might have missed something.

[dabla]

Looking good to me apart from question if this change should also not be directly tested in test_serialized_objects instead of relying on indirect tests?

[hwang-cadent]

Looking good to me apart from question if this change should also not be directly tested in test_serialized_objects instead of relying on indirect tests?

Good point. Currently tested indirectly through TriggerDagRunOperator tests. A direct unit test in test_serialized_objects.py for _deserialize_field_value() handling ARG_NOT_SET for date fields would be clearer. Should I add this test, or would you prefer to handle it in a follow-up?

[dabla]

Looking good to me apart from question if this change should also not be directly tested in test_serialized_objects instead of relying on indirect tests?

Good point. Currently tested indirectly through TriggerDagRunOperator tests. A direct unit test in test_serialized_objects.py for _deserialize_field_value() handling ARG_NOT_SET for date fields would be clearer. Should I add this test, or would you prefer to handle it in a follow-up?

It think it would be better even though it's indirectly tested through TriggerDagRunOperator tests. So yes would add case in test_serialized_objects.

[hwang-cadent]

Looking good to me apart from question if this change should also not be directly tested in test_serialized_objects instead of relying on indirect tests?

Good point. Currently tested indirectly through TriggerDagRunOperator tests. A direct unit test in test_serialized_objects.py for _deserialize_field_value() handling ARG_NOT_SET for date fields would be clearer. Should I add this test, or would you prefer to handle it in a follow-up?

It think it would be better even though it's indirectly tested through TriggerDagRunOperator tests. So yes would add case in test_serialized_objects.

Thanks for the feedback. I've added a direct test case test_deserialize_field_value_with_arg_not_set_for_date_fields in test_serialized_objects.py to verify that _deserialize_field_value returns NOTSET for ARG_NOT_SET date fields. This complements the indirect tests through TriggerDagRunOperator and improves test coverage.
The test verifies the behavior for date fields like logical_date, start_date, and end_date when they contain ARG_NOT_SET values.

[potiuk]

Bug: if not trigger_dag_id: uses a truthiness check. If XCom.get_value returns an empty string "" (unlikely but possible), this would incorrectly fall through to the operator attribute fallback.

Should be:
if trigger_dag_id is None:

[potiuk]

The core idea of storing the resolved trigger_dag_id in XCom for dynamic link resolution is sound and addresses a real user-facing bug. However, there are several issues that should be addressed before merging:

1. Architecture inconsistency in AF3: The trigger_dag_id XCom is pushed inside the operator's _trigger_dag_af_3() (before DagRunTriggerException is raised), while trigger_run_id is pushed in task_runner._handle_trigger_dag_run() (after successful trigger). This creates inconsistent state on trigger failure.
2. Bug in get_link fallback check: Uses truthiness (if not trigger_dag_id) instead of identity (if trigger_dag_id is None).
3. Separate concerns bundled together: The serialized_objects.py deserialization fix for ARG_NOT_SET on _date fields is a distinct bug and should ideally be a separate PR for cleaner review and bisectability.
4. Test assertions weakened: Ordered call verification was replaced with unordered assert_any_call, losing important ordering guarantees.

[dabla]

This should become as @potiuk pointed out:

if trigger_dag_id is not None:

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[potiuk]

@hwang-cadent — This PR has new commits since the last review requesting changes from @potiuk. If you believe you've addressed the feedback, please ping the reviewer(s) to request a re-review. 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.

[hwang-cadent]

@potiuk @dabla — addressed the review feedback in 93beb9697f and split out the serialization concern as you suggested:

Comments 1, 2 and 4 (in 93beb9697f on this PR):

1. AF3 architecture inconsistency — Moved XCOM_DAG_ID push out of _trigger_dag_af_3() and into task_runner._handle_trigger_dag_run() next to trigger_run_id. Both XComs are now written together only after a successful trigger, so XCom state stays consistent if the trigger fails.
2. get_link fallback check — Changed if not trigger_dag_id: to if trigger_dag_id is None: so any explicit value pushed to XCom is honored.
3. Ordered call assertions — Restored assert_has_calls with the full ordered call list in test_handle_trigger_dag_run_wait_for_completion (and added the new SetXCom("trigger_dag_id") to test_handle_trigger_dag_run).

Comment 3 (separate PR):

3. Bundled serialization fix — Agreed. Split out as a standalone PR: Fix ARG_NOT_SET deserialization for _date fields in OperatorSerialization #66564 (Fix ARG_NOT_SET deserialization for _date fields in OperatorSerialization). Once that lands I'll rebase this PR on main and drop the serialized_objects.py / test_serialized_objects.py hunks from here.

Could you take another look when you get a chance? Thank you!

[hwang-cadent]

Rebased on the latest apache/main (which had a meaningful refactor adding _validate_datetime_param and run_after support to TriggerDagRunOperator) and squashed the branch into a single clean commit (d1a826d9f8).

Conflict resolution highlights:

• Took apache/main's __init__ / execute / _trigger_dag_af_3 / _trigger_dag_af_2 structure verbatim, including _validate_datetime_param and run_after handling.
• Dropped my earlier isinstance(logical_date, ArgNotSet) defensive checks — those were a workaround for the deserialization bug now being fixed in Fix ARG_NOT_SET deserialization for _date fields in OperatorSerialization #66564, so they're no longer needed once that lands.
• Removed the serialized_objects.py / test_serialized_objects.py hunks from this PR (now tracked in Fix ARG_NOT_SET deserialization for _date fields in OperatorSerialization #66564 as discussed with @potiuk and @dabla).

Final scope of this PR (4 files, +138 / -22):

• providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py: add XCOM_DAG_ID constant, push it from _trigger_dag_af_2, and prefer it in TriggerDagRunLink.get_link() (with is None identity check).
• task-sdk/src/airflow/sdk/execution_time/task_runner.py: push trigger_dag_id next to trigger_run_id in _handle_trigger_dag_run after a successful trigger.
• providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py: switch existing AF3 link test to mock XCom.get_value, add test_extra_operator_link_with_dynamic_dag_id for both AF2 and AF3.
• task-sdk/tests/task_sdk/execution_time/test_task_runner.py: extend ordered assert_has_calls in both test_handle_trigger_dag_run and test_handle_trigger_dag_run_wait_for_completion with the new SetXCom("trigger_dag_id") step.

PR is now mergeable: true against main. Could you take another pass when you have a chance, @potiuk @dabla? Thank you!