Add run_after to TriggerDagRunOperator

[arnoldmr01]

• related: AF3 TriggerDagRunOperator / DagRunTriggerException should support run_after #60443

Why

To let TriggerDagRunOperator support run_after parameter in order to trigger multiple DagRuns of the same DAG in parallel without being constrained by logical_date.

What does the PR change

• Add run_after parameter to TriggerDauRunOperator, TriggerDagRun, TriggerDAGRunPayload
• Changed execute() logic. If run_after is NOTSET, the func will follow the original logic; otherwise, logical_date will be None

---

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

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

---

[jason810496]

Thanks for the PR! LGTM overall.

[potiuk]

@arnoldmr01 This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Other failing CI checks: Failing: Additional PROD image tests / Airflow CTL integration tests with PROD image. Run prek run --from-ref main locally to reproduce. See static checks docs.
• ⚠️ Unresolved review comments: This PR has 5 unresolved review threads from maintainers. Please review and resolve all inline review comments before requesting another review. You can resolve a conversation by clicking 'Resolve conversation' on each thread after addressing the feedback. See pull request guidelines.

Note: Your branch is 544 commits behind main. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.

What to do next:

• The comment informs you what you need to do.
• Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed.
• Maintainers will then proceed with a normal review.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. If you have questions, feel free to ask on the Airflow Slack.

[jason810496]

LGTM and it will be good to go after resolving conflict of uv.lock. Thanks!

[jason810496]

Mark as request changes in case of further regression on main CI.

[jason810496]

Even though the CI passed, I'm still concerned about the uv.lock change. Would it be possible to revert the uv.lock change before we merge?

[Copilot]

Pull request overview

Adds run_after support to TriggerDagRunOperator and wires it through the Task SDK client/message flow and the Execution API request schema so operators can trigger parallel DagRuns without relying on logical_date.

Changes:

• Add run_after parameter handling to TriggerDagRunOperator (including Airflow 3 exception payload wiring).
• Extend Task SDK TriggerDagRun message, API client payload, and supervisor dispatch to include run_after.
• Extend Execution API TriggerDAGRunPayload (and versioning migrations) to accept run_after, and pass it into trigger_dag.

Reviewed changes

Copilot reviewed 13 out of 14 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py	Add run_after parameter + parsing/dispatch; adjust AF2 warning behavior.
providers/standard/src/airflow/providers/standard/version_compat.py	Add NOTSET import + is_arg_set compatibility helper.
providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py	Add unit test coverage for run_after; adjust AF2 warning assertion.
task-sdk/src/airflow/sdk/api/client.py	Add run_after to dag_runs.trigger() and payload.
task-sdk/src/airflow/sdk/api/datamodels/_generated.py	Add run_after field to generated TriggerDAGRunPayload.
task-sdk/src/airflow/sdk/exceptions.py	Extend DagRunTriggerException to carry run_after.
task-sdk/src/airflow/sdk/execution_time/supervisor.py	Pass run_after through supervisor -> client trigger call.
task-sdk/src/airflow/sdk/execution_time/task_runner.py	Include run_after when translating trigger exception into supervisor message.
task-sdk/tests/task_sdk/execution_time/test_supervisor.py	Update request-case expectations for new trigger() signature and add run_after case.
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/dagrun.py	Add run_after field to Execution API request model.
airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py	Forward payload.run_after into trigger_dag().
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_04_06.py	Add Cadwyn version change entry for run_after field (plus migration instructions).
airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py	Register the new version change.
uv.lock	Updates aiobotocore requirement metadata in the lockfile.

Comments suppressed due to low confidence (1)

task-sdk/tests/task_sdk/execution_time/test_supervisor.py:2159

• REQUEST_TEST_CASES now contains two entries with the same test_id (both dag_run_trigger). Since these IDs are used by pytest parametrization (ids=lambda tc: tc.test_id), please make them unique (e.g. differentiate the “with note” vs “without note” case) to keep test output/debugging unambiguous.

            response=OKResponse(ok=True),
        ),
        test_id="dag_run_trigger",
    ),
    RequestTestCase(
        message=TriggerDagRun(
            dag_id="test_dag",
            run_id="test_run",
            conf={"key": "value"},
            logical_date=timezone.datetime(2025, 1, 1),
            reset_dag_run=True,
        ),
        expected_body={"ok": True, "type": "OKResponse"},
        client_mock=ClientMock(
            method_path="dag_runs.trigger",
            args=("test_dag", "test_run", {"key": "value"}, timezone.datetime(2025, 1, 1), None, True, None),
            response=OKResponse(ok=True),
        ),
        test_id="dag_run_trigger",
    ),

[arnoldmr01]

Modified attributes_not_supported_in_airflow_2 as a set. e.g.
attributes_not_supported_in_airflow_2 = { "run_after": NOTSET, "note": None, }
The keys the not supported attributes and the values are the default values.

The logic to check if those not supported attributes are set by the user is

unsupported_parameters = []
            for attr, default_value in self.attributes_not_suppported_in_airflow_2.items():
                value = getattr(self, attr, default_value)
                if value is not default_value:
                    unsupported_parameters.append(attr)

[arnoldmr01]

Modified attributes_not_supported_in_airflow_2 to

attributes_not_supported_in_airflow_2 = { "run_after": NOTSET, "note": None, }

and Modified the typo

[arnoldmr01]

Changed AddRunAfterFiled to AddRunAfterField