Send the Python task's execution_timeout to the supervisor

airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst

@@ -224,13 +224,13 @@ release that shipped it. See the legend below the tables for what each mark mean
      - Java
      - TypeScript
    * - Min. Airflow version
+     - 3.5.0
      - 3.4.0
-     - 3.4.0
-     - 3.4.0
+     - 3.5.0
    * - Current Supervisor Schema Version
+     - 2027-02-28
      - 2026-10-30
-     - 2026-10-30
-     - 2026-10-30
+     - 2027-02-28
 
 .. list-table:: Common to both authoring modes
    :header-rows: 1

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -556,13 +556,15 @@ class TIRunContext(BaseModel):
     ``None`` for regular tasks and for stub tasks that declare no parameters.
     """
 
-    execution_timeout: float | None = None
-    """
-    The ``execution_timeout`` in seconds of a task a language SDK runs, from its serialized Dag.
-
-    The supervisor enforces the timeout with this value. A Python task runs its own timer in the task
-    runner instead, and gets no value here. ``None`` when no timeout is set.
-    """
+    execution_timeout: float | None = Field(
+        default=None,
+        description=(
+            "The execution_timeout in seconds set on a task a language SDK runs, from its serialized Dag. "
+            "Unset when the task sets none. Zero means no timeout and a negative value has already "
+            "elapsed, as for a Python task. A Python task runs its own timer in the task runner instead, "
+            "and gets no value here."
+        ),
+    )
 
     multi_team: bool = False
     """

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -384,36 +384,38 @@ def ti_run(
         if first_task_reschedule_start_date is not None:
             context.first_task_reschedule_start_date = first_task_reschedule_start_date
 
-        # Avoid loading the serialized Dag on the normal Python task path: only a lang-SDK
-        # (foreign-runtime) task needs its definition here, and ti.is_stub carries the row's own copy
-        # of the serialized flag.
-        serialized_task = get_task(dag_bag, ti, session=session) if ti.is_stub else None
-
-        # Only set for lang-SDK (foreign-runtime) tasks with a captured TaskFlow arg
-        # spec; the route excludes unset fields, keeping regular responses lean.
-        if client_supports_arg_bindings() and (arg_bindings := get_arg_bindings(serialized_task)):
-            try:
-                context.arg_bindings = get_arg_bindings_adapter().validate_python(arg_bindings)
-            except ValidationError:
-                log.exception(
-                    "Serialized arg_bindings spec failed validation",
-                    dag_id=ti.dag_id,
-                    task_id=ti.task_id,
-                    dag_version_id=ti.dag_version_id,
-                )
-                raise HTTPException(
-                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                    detail={
-                        "reason": "invalid_arg_bindings",
-                        "message": "The serialized TaskFlow arg spec for this stub task is not valid.",
-                    },
-                )
+        # Only a lang-SDK (foreign-runtime) task needs its definition here: a Python task parses its
+        # Dag file itself. ti.is_stub carries the row's own copy of the serialized flag, so a Python
+        # task never loads the serialized Dag. The arg_bindings version check also covers
+        # execution_timeout, which only exists in a later API version.
+        if (
+            client_supports_arg_bindings()
+            and ti.is_stub
+            and (serialized_task := get_task(dag_bag, ti, session=session)) is not None
+        ):
+            # The route excludes unset fields, keeping responses without a captured TaskFlow arg spec lean.
+            if arg_bindings := get_arg_bindings(serialized_task):
+                try:
+                    context.arg_bindings = get_arg_bindings_adapter().validate_python(arg_bindings)
+                except ValidationError:
+                    log.exception(
+                        "Serialized arg_bindings spec failed validation",
+                        dag_id=ti.dag_id,
+                        task_id=ti.task_id,
+                        dag_version_id=ti.dag_version_id,
+                    )
+                    raise HTTPException(
+                        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                        detail={
+                            "reason": "invalid_arg_bindings",
+                            "message": "The serialized TaskFlow arg spec for this stub task is not valid.",
+                        },
+                    )
 
-        # Resolved here for a lang-SDK task only, since a regular task no longer loads its
-        # definition. #75387 has a Python task send its own execution_timeout to the supervisor,
-        # which is where the value for that kind of task belongs.
-        if (execution_timeout := get_execution_timeout(serialized_task)) is not None:
-            context.execution_timeout = execution_timeout
+            # The runtime cannot read the execution_timeout set on a Python stub, so it gets it here.
+            # A Python task enforces its own and sends that to the supervisor.
+            if (execution_timeout := get_execution_timeout(serialized_task)) is not None:
+                context.execution_timeout = execution_timeout
 
         # Only set if they are non-null
         if ti.next_method:

airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py

@@ -54,26 +54,26 @@ def get_task(dag_bag: DBDagBag, ti: Any, *, session: Session) -> SerializedOpera
     return dag.task_dict.get(ti.task_id)
 
 
-def get_arg_bindings(task: SerializedOperator | None) -> list | None:
+def get_arg_bindings(task: SerializedOperator) -> list | None:
     """
     Return the TaskFlow arg spec of *task* if it is a stub task.
 
     Mapped (``.expand()``) stubs never capture a parse-time spec, so they resolve to
     ``None`` here and keep the legacy ignored-args behavior; per-map-index delivery
     lands in a follow-up.
     """
-    if task is None or not task.is_stub:
+    if not task.is_stub:
         return None
     return task.arg_bindings
 
 
-def get_execution_timeout(task: SerializedOperator | None) -> float | None:
+def get_execution_timeout(task: SerializedOperator) -> float | None:
     """
-    Return the ``execution_timeout`` of *task* in seconds.
+    Return the ``execution_timeout`` of *task* in seconds, or ``None`` if it sets none.
 
-    Like the Python task runner, a zero timeout means no timeout, and a negative one has
-    already elapsed.
+    The value is sent as it is, so a runtime handles it the way the Python task runner does: zero
+    means no timeout, and a negative one has already elapsed.
     """
-    if task is None or not task.execution_timeout:
+    if task.execution_timeout is None:
         return None
-    return max(task.execution_timeout.total_seconds(), 0.0)
+    return task.execution_timeout.total_seconds()

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -56,7 +56,6 @@
     AddArgBindingsToTIRunContext,
     AddCallbackRunEndpoint,
     AddDagRunNoteUpdateEndpoint,
-    AddExecutionTimeoutToTIRunContext,
     AddExpandedTICountToTIRunContext,
     AddFirstTaskRescheduleStartDateField,
     AddLoopContext,
@@ -70,17 +69,19 @@
 from airflow.api_fastapi.execution_api.versions.v2026_10_30_xcom_params import (
     AddPreviousIterationToXComFilterParams,
 )
-from airflow.api_fastapi.execution_api.versions.v2027_02_28 import AddDagProcessorJobEndpoints
+from airflow.api_fastapi.execution_api.versions.v2027_02_28 import (
+    AddDagProcessorJobEndpoints,
+    AddExecutionTimeoutToTIRunContext,
+)
 
 bundle = VersionBundle(
     HeadVersion(),
-    Version("2027-02-28", AddDagProcessorJobEndpoints),
+    Version("2027-02-28", AddDagProcessorJobEndpoints, AddExecutionTimeoutToTIRunContext),
     Version(
         "2026-10-30",
         AddArgBindingsToTIRunContext,
         AddCallbackRunEndpoint,
         AddDagRunNoteUpdateEndpoint,
-        AddExecutionTimeoutToTIRunContext,
         AddExpandedTICountToTIRunContext,
         AddLoopContext,
         AddLoopDecisionField,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py

@@ -93,20 +93,6 @@ def remove_arg_bindings_field(response: ResponseInfo) -> None:
         response.body.pop("arg_bindings", None)
 
 
-class AddExecutionTimeoutToTIRunContext(VersionChange):
-    """Add ``execution_timeout`` so the supervisor can enforce it."""
-
-    description = __doc__
-    instructions_to_migrate_to_previous_version = (
-        schema(TIRunContext).field("execution_timeout").didnt_exist,
-    )
-
-    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
-    def remove_execution_timeout_field(response: ResponseInfo) -> None:  # type: ignore[misc]
-        """Strip ``execution_timeout`` from the run context for older clients."""
-        response.body.pop("execution_timeout", None)
-
-
 class AddCallbackRunEndpoint(VersionChange):
     """Add the callbacks/{callback_id}/run endpoint a worker uses to exchange its single-use callback token."""
 

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2027_02_28.py

@@ -17,7 +17,9 @@
 
 from __future__ import annotations
 
-from cadwyn import VersionChange, endpoint
+from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, endpoint, schema
+
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
 
 
 class AddDagProcessorJobEndpoints(VersionChange):
@@ -31,3 +33,17 @@ class AddDagProcessorJobEndpoints(VersionChange):
         endpoint("/jobs/{job_id}/complete", ["POST"]).didnt_exist,
         endpoint("/jobs/{job_id}/parse-token", ["POST"]).didnt_exist,
     )
+
+
+class AddExecutionTimeoutToTIRunContext(VersionChange):
+    """Add ``execution_timeout`` so the supervisor can enforce it."""
+
+    description = __doc__
+    instructions_to_migrate_to_previous_version = (
+        schema(TIRunContext).field("execution_timeout").didnt_exist,
+    )
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def remove_execution_timeout_field(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Strip ``execution_timeout`` from the run context for older clients."""
+        response.body.pop("execution_timeout", None)

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -1199,9 +1199,9 @@ def transform(country: str): ...
         [
             pytest.param(task.stub(execution_timeout=timedelta(seconds=90)), 90.0, id="stub-with-timeout"),
             pytest.param(task.stub, None, id="stub-without-timeout"),
-            pytest.param(task.stub(execution_timeout=timedelta(0)), None, id="stub-zero-timeout"),
+            pytest.param(task.stub(execution_timeout=timedelta(0)), 0.0, id="stub-zero-timeout"),
             pytest.param(
-                task.stub(execution_timeout=timedelta(seconds=-30)), 0.0, id="stub-negative-timeout"
+                task.stub(execution_timeout=timedelta(seconds=-30)), -30.0, id="stub-negative-timeout"
             ),
             # A Python task's timeout stays with the task runner, which sends it to the supervisor,
             # so the server resolves one only for a task whose definition it loads anyway.
@@ -2153,6 +2153,7 @@ def test_ti_run_skips_queued_duration_metric_when_the_response_fails(
             dag_id=str(uuid4()),
         )
         ti.queued_dttm = queued_at
+        ti.is_stub = True
         session.commit()
 
         with mock.patch(

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py

@@ -17,7 +17,6 @@
 
 from __future__ import annotations
 
-from datetime import timedelta
 from uuid import uuid4
 
 import pytest
@@ -54,7 +53,6 @@ def test_task_coordinate_schema_by_version(client, version):
     assert ("region_index" in previous) == (version == "2026-10-30")
     context = response.json()["components"]["schemas"]["TIRunContext"]["properties"]
     assert ("loop" in context) == (version == "2026-10-30")
-    assert ("execution_timeout" in context) == (version == "2026-10-30")
 
 
 @pytest.mark.parametrize(
@@ -289,38 +287,6 @@ def test_head_version_includes_arg_bindings(self, client, stub_ti):
         ]
 
 
-class TestExecutionTimeoutBackwardCompat:
-    def setup_method(self):
-        clear_db_runs()
-
-    def teardown_method(self):
-        clear_db_runs()
-
-    @pytest.fixture
-    def stub_ti(self, dag_maker):
-        with dag_maker("test_execution_timeout_compat_dag", serialized=True):
-
-            @task.stub(execution_timeout=timedelta(seconds=90))
-            def sleeper(): ...
-
-            sleeper()
-
-        (ti,) = dag_maker.create_dagrun().get_task_instances()
-        ti.set_state(State.QUEUED)
-        dag_maker.session.flush()
-        return ti
-
-    def test_old_version_strips_execution_timeout(self, old_ver_client, stub_ti):
-        response = old_ver_client.patch(f"/execution/task-instances/{stub_ti.id}/run", json=RUN_PATCH_BODY)
-        assert response.status_code == 200
-        assert "execution_timeout" not in response.json()
-
-    def test_head_version_includes_execution_timeout(self, client, stub_ti):
-        response = client.patch(f"/execution/task-instances/{stub_ti.id}/run", json=RUN_PATCH_BODY)
-        assert response.status_code == 200
-        assert response.json()["execution_timeout"] == 90.0
-
-
 @pytest.fixture
 def archived_attempt(client, exec_app, monkeypatch, create_task_instance, session):
     """A running attempt, authenticated with a signed token, that has since been retried."""

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2027_02_28/test_task_instances.py

@@ -0,0 +1,84 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+import pytest
+
+from airflow.sdk import task
+from airflow.utils.state import State
+
+from tests_common.test_utils.db import clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+RUN_PATCH_BODY = {
+    "state": "running",
+    "hostname": "h",
+    "unixname": "u",
+    "pid": 1,
+    "start_date": "2024-09-30T12:00:00Z",
+}
+
+
+@pytest.fixture
+def old_ver_client(client):
+    """Execution API version immediately before ``execution_timeout`` was added."""
+    client.headers["Airflow-API-Version"] = "2026-10-30"
+    return client
+
+
+@pytest.mark.parametrize("version", ["2026-10-30", "2027-02-28"])
+def test_execution_timeout_schema_by_version(client, version):
+    response = client.get(f"/execution/openapi.json?version={version}")
+    assert response.status_code == 200
+    context = response.json()["components"]["schemas"]["TIRunContext"]["properties"]
+    assert ("execution_timeout" in context) == (version == "2027-02-28")
+
+
+class TestExecutionTimeoutBackwardCompat:
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    @pytest.fixture
+    def stub_ti(self, dag_maker):
+        with dag_maker("test_execution_timeout_compat_dag", serialized=True):
+
+            @task.stub(execution_timeout=timedelta(seconds=90))
+            def sleeper(): ...
+
+            sleeper()
+
+        (ti,) = dag_maker.create_dagrun().get_task_instances()
+        ti.set_state(State.QUEUED)
+        dag_maker.session.flush()
+        return ti
+
+    def test_old_version_strips_execution_timeout(self, old_ver_client, stub_ti):
+        response = old_ver_client.patch(f"/execution/task-instances/{stub_ti.id}/run", json=RUN_PATCH_BODY)
+        assert response.status_code == 200
+        assert "execution_timeout" not in response.json()
+
+    def test_head_version_includes_execution_timeout(self, client, stub_ti):
+        response = client.patch(f"/execution/task-instances/{stub_ti.id}/run", json=RUN_PATCH_BODY)
+        assert response.status_code == 200
+        assert response.json()["execution_timeout"] == 90.0

airflow-core/tests/unit/dag_processing/test_processor.py

@@ -2380,7 +2380,7 @@ def get_type_names(union_type):
             "UpdateHITLDetail",
             "GetHITLDetailResponse",
             "SetRenderedMapIndex",
-            "StartExecutionTimeout",
+            "SetExecutionTimeout",
             # AIP-103 task/asset store — Dag processor has no task execution context.
             "GetTaskStateStore",
             "SetTaskStateStore",