Add duration, queued_dttm, and start_date to RuntimeTaskInstance for listener hooks

[dandanseo123]

Description:

In AF2, listener hooks (on_task_instance_success/failed) received the full ORM TaskInstance model, which includes duration, queued_dttm, and start_date. In AF3, these hooks receive RuntimeTaskInstance — a minimal runtime object missing these fields. Plugins relying on them for metrics break when migrating to AF3.

Additionally, DAG-level hooks (on_dag_run_*) already expose queue time via DagRun.queued_at, but task-level hooks do not expose the equivalent queued_dttm. This is inconsistent across the listener API.

Changes

• duration : added as a computed property on RuntimeTaskInstance: (end_date - start_date).total_seconds(). This matches the ORM's set_duration() formula.
• queued_dttm: Flows from DB → TIRunContext → RuntimeTaskInstance via the Execution API ti_run endpoint.
• start_date: Flows from DB → TIRunContext → RuntimeTaskInstance. Preserves the original start time across deferrals so that duration is computed correctly for deferred tasks (previously used the supervisor process start time, under-reporting duration).

All three fields added to RuntimeTaskInstanceProtocol.

Use case

Listener plugins that publish task execution metrics need duration (how long the task ran) and queued_dttm (how long it waited in queue before starting). Without these fields, plugin authors must either query the DB directly from the listener hook or accept missing metrics, neither is ideal.

---

• Yes (please specify the tool below)

Generated-by: Claude Code following the guidelines

[Copilot]

Pull request overview

This PR aims to restore AF2-era listener-hook compatibility in AF3 by exposing task-level runtime timing fields on RuntimeTaskInstance, specifically duration (computed) and queued_dttm (propagated from the Execution API).

Changes:

• Add RuntimeTaskInstance.duration as a computed property based on start_date/end_date and extend RuntimeTaskInstanceProtocol accordingly.
• Add queued_dttm to RuntimeTaskInstance and flow it from the Execution API ti_run response (TaskInstance.queued_dttm → TIRunContext → StartupDetails → parse()).
• Add/extend unit tests for both Task SDK runtime objects and the Execution API response payload.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
task-sdk/src/airflow/sdk/execution_time/task_runner.py	Adds queued_dttm, adds computed duration, and threads queued_dttm into parse(); refactors template-context caching.
task-sdk/src/airflow/sdk/types.py	Extends RuntimeTaskInstanceProtocol with queued_dttm and duration.
task-sdk/src/airflow/sdk/api/datamodels/_generated.py	Adds queued_dttm to generated TIRunContext SDK model.
task-sdk/tests/task_sdk/execution_time/test_task_runner.py	Adds Task SDK tests for duration and queued_dttm.
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py	Includes queued_dttm in the ti_run query and conditionally sets it on TIRunContext.
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py	Adds queued_dttm field to server-side TIRunContext.
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py	Adds tests asserting queued_dttm inclusion/omission in ti_run responses.

Comments suppressed due to low confidence (1)

task-sdk/src/airflow/sdk/api/datamodels/_generated.py:666

• task-sdk/src/airflow/sdk/api/datamodels/_generated.py is generated (see header). Adding queued_dttm here should come from regenerating the SDK models from the updated Execution API OpenAPI schema (after introducing the schema change via a new Cadwyn version). Otherwise it’s easy for this file to drift from the actual server schema/version.

Suggested fix: regenerate the Task SDK datamodels via the repo’s OpenAPI/datamodel-codegen workflow, rather than hand-editing this generated module, and ensure the SDK’s API_VERSION aligns with the new server version that introduces queued_dttm.

    next_kwargs: Annotated[dict[str, Any] | str | None, Field(title="Next Kwargs")] = None
    xcom_keys_to_clear: Annotated[list[str] | None, Field(title="Xcom Keys To Clear")] = None
    should_retry: Annotated[bool | None, Field(title="Should Retry")] = False
    queued_dttm: Annotated[AwareDatetime | None, Field(title="Queued Dttm")] = None

[dandanseo123]

@ashb @kaxil @amoghrajesh would appreciate a review when you get a chance. Thanks!

[potiuk]

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

Issues found:

• ❌ Merge conflicts: This PR has merge conflicts with the main branch. Your branch is 47 commits behind main. Please rebase your branch (git fetch origin && git rebase origin/main), resolve the conflicts, and push again. See contributing quick start.

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.
• There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates.
• 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. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

[Copilot]

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.