[sandeep-paluru]

A deferred task was given a fresh execution_timeout every time it resumed. Time already spent, including the deferral, did not count, so the task could run longer than the limit configured for it.

Since #74869 the supervisor arms the duration on TIRunContext.execution_timeout, and every SDK uses that number. When the task is resuming (next_method is set), the run route now sends the time still left: the configured timeout minus the time since the original start_date, clamped at 0. A task that is not resuming still gets the full timeout. If no timeout is set, the field stays unset. A value of 0 means the limit has already elapsed, so the task fails as a timeout promptly. A start_date ahead of the clock does not extend the timeout.

The Python runner applies those seconds to its in-process timeout, which is what raises AirflowTaskTimeout and runs on_kill. It uses the task's own execution_timeout only when the server omits the field, and it no longer subtracts elapsed time itself. How long a trigger may run while the task is still deferred is unchanged.

No new execution API version. execution_timeout is already on TIRunContext from #74869 with the same type, and older clients still have the field stripped.

Tests in airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py use time_machine: a fresh run keeps the full timeout, a resume with time left gets the remainder, a clock ahead of start_date does not extend it, an already elapsed resume returns 0 with the field present, and a task with no timeout omits the field. Task SDK tests check that the runner uses the server seconds, falls back when the field is omitted, and raises AirflowTaskTimeout before the callable when the server sends 0.

prek pre-commit checks pass, including airflow-core and task-sdk mypy, ruff, and the execution API version check.

closes: #73475

---

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

• Yes (AI coding assistant)

Generated-by: AI coding assistant following the guidelines

[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
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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Andrushika]

Note for maintainers: I think this should wait until #74869 is merged, since the design around execution_timeout will likely look quite different afterwards. That PR moves the timeout value into TIRunContext so the supervisor arms the deadline itself.

Under that shape, I'd argue the remaining-time-after-deferral computation belongs outside the Python task runner.
So this PR would need a rework rather than a rebase once #74869 lands.

[sandeep-paluru]

Thanks, that makes sense. I read through #74869 and agree the runner-side math won't fit once the supervisor owns the deadline.

Once it lands, I think the natural spot is get_execution_timeout on the server side: when the TI is resuming from deferral (next_method is set), send execution_timeout - (now - start_date) in TIRunContext instead of the full value, clamped at 0 so the existing "negative means already elapsed" handling fails it straight away. That way Python and language SDK tasks both get the right remaining time and the task runner doesn't need to know about deferral at all.

I'll hold this one until #74869 merges and then rework it into that shape, with the tests moved to the execution API side. Happy to hear if a maintainer would rather see it go a different way.

[sandeep-paluru]

Reworked this on top of #74869 now that it's merged. The run route sends the remaining timeout on resume through TIRunContext.execution_timeout, so every SDK gets it, and the task runner no longer does its own elapsed-time math. Description updated.

[sandeep-paluru]

@jason810496 since you reviewed and merged #74869, would you mind taking a look at this when you get a chance? It builds directly on the TIRunContext.execution_timeout change from that PR. No rush, thanks.