From edcb26ef9c43ab17976bc8929ffe8065578515fb Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Thu, 29 Oct 2026 08:50:37 +0000
Subject: [PATCH 1/8] Send the Python task's execution_timeout to the
 supervisor

---
 .../language-sdks/index.rst                   |  8 +-
 airflow-core/docs/core-concepts/tasks.rst     |  9 ++
 .../execution_api/versions/__init__.py        |  9 +-
 .../execution_api/versions/v2026_10_30.py     | 14 ----
 .../execution_api/versions/v2027_02_28.py     | 18 +++-
 .../src/airflow/config_templates/config.yml   |  3 +-
 .../v2026_10_30/test_task_instances.py        | 34 --------
 .../v2027_02_28/test_task_instances.py        | 84 +++++++++++++++++++
 .../unit/dag_processing/test_processor.py     |  2 +-
 .../tests/unit/jobs/test_triggerer_job.py     |  2 +-
 go-sdk/README.md                              |  2 +-
 go-sdk/capabilities.yaml                      |  4 +-
 .../execution/genmodels/discriminators.gen.go |  8 +-
 go-sdk/pkg/execution/genmodels/models.gen.go  | 17 ++--
 go-sdk/pkg/execution/messages.go              |  2 +-
 go-sdk/schema/supervisor-schema.json          | 36 ++++----
 .../airflow/sdk/coordinators/_subprocess.py   |  2 +-
 .../src/airflow/sdk/execution_time/comms.py   |  9 +-
 .../sdk/execution_time/schema/schema.json     | 36 ++++----
 .../schema/versions/__init__.py               |  6 +-
 .../schema/versions/v2026_10_30.py            | 15 ----
 .../schema/versions/v2027_02_28.py            | 39 +++++++++
 .../airflow/sdk/execution_time/supervisor.py  | 19 +++--
 .../airflow/sdk/execution_time/task_runner.py |  4 +-
 .../execution_time/schema/test_migrator.py    | 10 +--
 .../execution_time/test_supervisor.py         | 36 ++++----
 .../execution_time/test_task_runner.py        |  4 +-
 ts-sdk/README.md                              |  2 +-
 ts-sdk/capabilities.yaml                      |  4 +-
 ts-sdk/schema/supervisor-schema.json          | 36 ++++----
 ts-sdk/src/generated/supervisor.ts            | 42 +++++-----
 31 files changed, 320 insertions(+), 196 deletions(-)
 create mode 100644 airflow-core/tests/unit/api_fastapi/execution_api/versions/v2027_02_28/test_task_instances.py
 create mode 100644 task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py

diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
index ea8d4b8448d10..f2ad01e921ca9 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
+++ b/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
diff --git a/airflow-core/docs/core-concepts/tasks.rst b/airflow-core/docs/core-concepts/tasks.rst
index d97c6410b41d6..dff6e52c022fe 100644
--- a/airflow-core/docs/core-concepts/tasks.rst
+++ b/airflow-core/docs/core-concepts/tasks.rst
@@ -145,6 +145,15 @@ if the task has not stopped within ``[core] killed_task_cleanup_time`` seconds a
 (for example because it is blocked in native code), the supervisor sends it SIGTERM, and SIGKILL after the same
 period again. Tasks run by a language SDK do not raise ``AirflowTaskTimeout`` and are stopped only this way.
 
+A task that does not set ``execution_timeout`` itself gets ``[core] default_task_execution_timeout``, which is read in
+different places depending on how the task runs:
+
+- A Python task reads it on the worker, which parses the Dag file again before running the task.
+- A task run by a language SDK reads it where the Dag processor runs, when the Dag is parsed and serialized.
+
+Set ``[core] default_task_execution_timeout`` to the same value on the Dag processor and the workers, so both kinds
+of task get the same default.
+
 In addition, sensors have a ``timeout`` parameter. This only matters for sensors in ``reschedule`` mode. ``timeout`` controls the maximum
 time allowed for the sensor to succeed. If ``timeout`` is breached, ``AirflowSensorTimeout`` will be raised and the sensor fails immediately
 without retrying.
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
index d66d483f2aacd..2c1127b0d5e81 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
+++ b/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,
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
index 5d23b073ec51c..1f7780ee32171 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
+++ b/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."""
 
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2027_02_28.py b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2027_02_28.py
index b54a50236284a..3b39aa4e83a64 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2027_02_28.py
+++ b/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)
diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml
index b87f27c509364..b18ea7091dd4b 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -398,7 +398,8 @@ core:
       description: |
         The default task execution_timeout value for the operators. Expected an integer value to
         be passed into timedelta as seconds. If not specified, then the value is considered as None,
-        meaning that the operators are never timed out by default.
+        meaning that the operators are never timed out by default. Python tasks read it on the worker
+        and tasks run by a language SDK read it on the Dag processor, so set the same value on both.
       version_added: 2.3.0
       type: integer
       example: ~
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py
index ae9029f3bea85..513d15f6673dc 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py
+++ b/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."""
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2027_02_28/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2027_02_28/test_task_instances.py
new file mode 100644
index 0000000000000..430bcf2c5a568
--- /dev/null
+++ b/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
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py b/airflow-core/tests/unit/dag_processing/test_processor.py
index c9c4ab19e089f..f56de3f3141d3 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/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",
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index ff0e03e8099e2..a614b30cc78e7 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -3359,7 +3359,7 @@ def get_type_names(union_type):
             "ResendLoggingFD",
             "CreateHITLDetailPayload",
             "SetRenderedMapIndex",
-            "StartExecutionTimeout",
+            "SetExecutionTimeout",
             "GetDag",
             # AIP-103 task store — triggerer has no task execution context.
             "GetTaskStateStore",
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 6da0ccba329aa..8ecef83ee845f 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -117,7 +117,7 @@ prek hook regenerate it.
 
 <!-- BEGIN AUTO-GENERATED LANG-SDK COMPAT MATRIX -->
 
-*Min. Airflow version: 3.4.0 · supervisor schema: 2026-10-30*
+*Min. Airflow version: 3.5.0 · supervisor schema: 2027-02-28*
 
 | Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes |
 |---|---|---|---|---|---|
diff --git a/go-sdk/capabilities.yaml b/go-sdk/capabilities.yaml
index 95b39727d4981..1fab164e58f36 100644
--- a/go-sdk/capabilities.yaml
+++ b/go-sdk/capabilities.yaml
@@ -17,11 +17,11 @@
 ---
 sdk: go
 
-min_airflow_version: "3.4.0"
+min_airflow_version: "3.5.0"
 
 # Keep in sync with SupervisorSchemaVersion in pkg/execution/messages.go, which is embedded in
 # bundle metadata. The render hook fails if the two disagree.
-supervisor_schema_version: "2026-10-30"
+supervisor_schema_version: "2027-02-28"
 
 # The task runner terminates a task with SucceedTask, RetryTask, or TaskState (failed/removed); only
 # an airflow.TriggerDagRun task emits TaskState skipped or DeferTask. It does not emit RescheduleTask
diff --git a/go-sdk/pkg/execution/genmodels/discriminators.gen.go b/go-sdk/pkg/execution/genmodels/discriminators.gen.go
index 3a81b6ce7a88a..8736a6e2c9730 100644
--- a/go-sdk/pkg/execution/genmodels/discriminators.gen.go
+++ b/go-sdk/pkg/execution/genmodels/discriminators.gen.go
@@ -90,12 +90,12 @@ const (
 	TypeSentFDs                     = "SentFDs"
 	TypeSetAssetStateStoreByName    = "SetAssetStateStoreByName"
 	TypeSetAssetStateStoreByURI     = "SetAssetStateStoreByUri"
+	TypeSetExecutionTimeout         = "SetExecutionTimeout"
 	TypeSetRenderedFields           = "SetRenderedFields"
 	TypeSetRenderedMapIndex         = "SetRenderedMapIndex"
 	TypeSetTaskStateStore           = "SetTaskStateStore"
 	TypeSetXCom                     = "SetXCom"
 	TypeSkipDownstreamTasks         = "SkipDownstreamTasks"
-	TypeStartExecutionTimeout       = "StartExecutionTimeout"
 	TypeStartupDetails              = "StartupDetails"
 	TypeSucceedTask                 = "SucceedTask"
 	TypeTICount                     = "TICount"
@@ -321,6 +321,9 @@ func EnsureType(m any) any {
 	case SetAssetStateStoreByURI:
 		b.Type = TypeSetAssetStateStoreByURI
 		return b
+	case SetExecutionTimeout:
+		b.Type = TypeSetExecutionTimeout
+		return b
 	case SetRenderedFields:
 		b.Type = TypeSetRenderedFields
 		return b
@@ -336,9 +339,6 @@ func EnsureType(m any) any {
 	case SkipDownstreamTasks:
 		b.Type = TypeSkipDownstreamTasks
 		return b
-	case StartExecutionTimeout:
-		b.Type = TypeStartExecutionTimeout
-		return b
 	case StartupDetails:
 		b.Type = TypeStartupDetails
 		return b
diff --git a/go-sdk/pkg/execution/genmodels/models.gen.go b/go-sdk/pkg/execution/genmodels/models.gen.go
index f95ceb53d629a..9d94e14942365 100644
--- a/go-sdk/pkg/execution/genmodels/models.gen.go
+++ b/go-sdk/pkg/execution/genmodels/models.gen.go
@@ -1555,6 +1555,16 @@ type SetAssetStateStoreByURI struct {
 	Value JsonValue `msgpack:"value"`
 }
 
+// Tell the supervisor the task's “execution_timeout“ so it can enforce it from
+// outside the task process.
+type SetExecutionTimeout struct {
+	// TimeoutSeconds corresponds to the JSON schema field "timeout_seconds".
+	TimeoutSeconds float64 `msgpack:"timeout_seconds"`
+
+	// Type corresponds to the JSON schema field "type".
+	Type string `msgpack:"type,omitempty"`
+}
+
 // Payload for setting RTIF for a task instance.
 type SetRenderedFields struct {
 	// RenderedFields corresponds to the JSON schema field "rendered_fields".
@@ -1629,13 +1639,6 @@ type SkipDownstreamTasks struct {
 	Type string `msgpack:"type,omitempty"`
 }
 
-// Tell the supervisor to start enforcing the run context's “execution_timeout“
-// from outside the task process.
-type StartExecutionTimeout struct {
-	// Type corresponds to the JSON schema field "type".
-	Type string `msgpack:"type,omitempty"`
-}
-
 type StartupDetails struct {
 	// BundleInfo corresponds to the JSON schema field "bundle_info".
 	BundleInfo BundleInfo `msgpack:"bundle_info"`
diff --git a/go-sdk/pkg/execution/messages.go b/go-sdk/pkg/execution/messages.go
index 1dd596201258a..3f45c1c3fc185 100644
--- a/go-sdk/pkg/execution/messages.go
+++ b/go-sdk/pkg/execution/messages.go
@@ -32,7 +32,7 @@ import (
 // reported in a bundle's airflow-metadata manifest as
 // sdk.supervisor_schema_version so the supervisor can down/upgrade messages to
 // a shape the bundle understands.
-const SupervisorSchemaVersion = "2026-10-30"
+const SupervisorSchemaVersion = "2027-02-28"
 
 // The message-type discriminator strings (genmodels.Type*) are generated from the
 // schema's "type" consts in discriminators.gen.go; outbound messages stamp the
diff --git a/go-sdk/schema/supervisor-schema.json b/go-sdk/schema/supervisor-schema.json
index ebe8c2ab65ce8..f48c5ee097d18 100644
--- a/go-sdk/schema/supervisor-schema.json
+++ b/go-sdk/schema/supervisor-schema.json
@@ -1,6 +1,6 @@
 {
   "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "api_version": "2026-10-30",
+  "api_version": "2027-02-28",
   "description": "Apache Airflow SDK Supervisor Schema",
   "$defs": {
     "AssetAliasReferenceAssetEventDagRun": {
@@ -3791,6 +3791,27 @@
       "title": "SetAssetStateStoreByUri",
       "type": "object"
     },
+    "SetExecutionTimeout": {
+      "description": "Tell the supervisor the task's ``execution_timeout`` so it can enforce it from outside the task process.",
+      "properties": {
+        "timeout_seconds": {
+          "exclusiveMinimum": 0,
+          "title": "Timeout Seconds",
+          "type": "number"
+        },
+        "type": {
+          "const": "SetExecutionTimeout",
+          "default": "SetExecutionTimeout",
+          "title": "Type",
+          "type": "string"
+        }
+      },
+      "required": [
+        "timeout_seconds"
+      ],
+      "title": "SetExecutionTimeout",
+      "type": "object"
+    },
     "SetRenderedFields": {
       "description": "Payload for setting RTIF for a task instance.",
       "properties": {
@@ -3984,19 +4005,6 @@
       "title": "SkipDownstreamTasks",
       "type": "object"
     },
-    "StartExecutionTimeout": {
-      "description": "Tell the supervisor to start enforcing the run context's ``execution_timeout`` from outside the task process.",
-      "properties": {
-        "type": {
-          "const": "StartExecutionTimeout",
-          "default": "StartExecutionTimeout",
-          "title": "Type",
-          "type": "string"
-        }
-      },
-      "title": "StartExecutionTimeout",
-      "type": "object"
-    },
     "StartupDetails": {
       "properties": {
         "ti": {
diff --git a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
index aa30b834ef64d..135477c41103e 100644
--- a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
+++ b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
@@ -546,7 +546,7 @@ def _on_child_started(
             bundle_info=bundle_info,
             sentry_integration=sentry_integration,
         )
-        # Language runtimes do not send StartExecutionTimeout, so the clock starts once they have the task.
+        # Language runtimes do not send SetExecutionTimeout, so the clock starts once they have the task.
         self._arm_execution_timeout()
 
     def _prepare_ti_context(self, ti: TaskInstance, ti_context: TIRunContext) -> TIRunContext | None:
diff --git a/task-sdk/src/airflow/sdk/execution_time/comms.py b/task-sdk/src/airflow/sdk/execution_time/comms.py
index 45af5b757a3fb..f0e8336426c34 100644
--- a/task-sdk/src/airflow/sdk/execution_time/comms.py
+++ b/task-sdk/src/airflow/sdk/execution_time/comms.py
@@ -1136,10 +1136,11 @@ class SetRenderedMapIndex(BaseModel):
     type: Literal["SetRenderedMapIndex"] = "SetRenderedMapIndex"
 
 
-class StartExecutionTimeout(BaseModel):
-    """Tell the supervisor to start enforcing the run context's ``execution_timeout`` from outside the task process."""
+class SetExecutionTimeout(BaseModel):
+    """Tell the supervisor the task's ``execution_timeout`` so it can enforce it from outside the task process."""
 
-    type: Literal["StartExecutionTimeout"] = "StartExecutionTimeout"
+    timeout_seconds: Annotated[float, Field(gt=0)]
+    type: Literal["SetExecutionTimeout"] = "SetExecutionTimeout"
 
 
 class TriggerDagRun(TriggerDAGRunPayload):
@@ -1351,7 +1352,7 @@ class GetDag(BaseModel):
     | SetTaskStateStore
     | SetXCom
     | SkipDownstreamTasks
-    | StartExecutionTimeout
+    | SetExecutionTimeout
     | SucceedTask
     | ValidateInletsAndOutlets
     | TaskState
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
index ebe8c2ab65ce8..f48c5ee097d18 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
@@ -1,6 +1,6 @@
 {
   "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "api_version": "2026-10-30",
+  "api_version": "2027-02-28",
   "description": "Apache Airflow SDK Supervisor Schema",
   "$defs": {
     "AssetAliasReferenceAssetEventDagRun": {
@@ -3791,6 +3791,27 @@
       "title": "SetAssetStateStoreByUri",
       "type": "object"
     },
+    "SetExecutionTimeout": {
+      "description": "Tell the supervisor the task's ``execution_timeout`` so it can enforce it from outside the task process.",
+      "properties": {
+        "timeout_seconds": {
+          "exclusiveMinimum": 0,
+          "title": "Timeout Seconds",
+          "type": "number"
+        },
+        "type": {
+          "const": "SetExecutionTimeout",
+          "default": "SetExecutionTimeout",
+          "title": "Type",
+          "type": "string"
+        }
+      },
+      "required": [
+        "timeout_seconds"
+      ],
+      "title": "SetExecutionTimeout",
+      "type": "object"
+    },
     "SetRenderedFields": {
       "description": "Payload for setting RTIF for a task instance.",
       "properties": {
@@ -3984,19 +4005,6 @@
       "title": "SkipDownstreamTasks",
       "type": "object"
     },
-    "StartExecutionTimeout": {
-      "description": "Tell the supervisor to start enforcing the run context's ``execution_timeout`` from outside the task process.",
-      "properties": {
-        "type": {
-          "const": "StartExecutionTimeout",
-          "default": "StartExecutionTimeout",
-          "title": "Type",
-          "type": "string"
-        }
-      },
-      "title": "StartExecutionTimeout",
-      "type": "object"
-    },
     "StartupDetails": {
       "properties": {
         "ti": {
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
index 27ff074bb237b..83446d7b776c8 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
@@ -40,16 +40,19 @@ def get_bundle() -> VersionBundle:
     from airflow.sdk.execution_time.schema.versions.v2026_10_30 import (
         AddArgBindingsToSupervisorTIRunContext,
         AddDagDefinitionsToDagFileParsingResult,
-        AddExecutionTimeoutToSupervisorTIRunContext,
         AddExpandedTICountToSupervisorTIRunContext,
         AddFirstTaskRescheduleStartDateToSupervisorTIRunContext,
         AddLoopRuntime,
         AddRegionCoordinatesToSupervisorTaskInstance,
         AddRetryReasonToTaskState,
     )
+    from airflow.sdk.execution_time.schema.versions.v2027_02_28 import (
+        AddExecutionTimeoutToSupervisorTIRunContext,
+    )
 
     return VersionBundle(
         HeadVersion(),
+        Version("2027-02-28", AddExecutionTimeoutToSupervisorTIRunContext),
         Version(
             "2026-10-30",
             AddArgBindingsToSupervisorTIRunContext,
@@ -59,7 +62,6 @@ def get_bundle() -> VersionBundle:
             AddRegionCoordinatesToSupervisorTaskInstance,
             AddFirstTaskRescheduleStartDateToSupervisorTIRunContext,
             AddExpandedTICountToSupervisorTIRunContext,
-            AddExecutionTimeoutToSupervisorTIRunContext,
         ),
         Version("2026-06-16"),
     )
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py
index ddc99b274b987..d69abf63b5f34 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py
@@ -133,18 +133,3 @@ class AddExpandedTICountToSupervisorTIRunContext(VersionChange):
     instructions_to_migrate_to_previous_version = (
         schema(TIRunContext).field("expanded_ti_count").didnt_exist,
     )
-
-
-class AddExecutionTimeoutToSupervisorTIRunContext(VersionChange):
-    """
-    Add the ``execution_timeout`` field carrying the task's timeout, which the supervisor enforces.
-
-    The supervisor-schema mirror of the execution API's ``AddExecutionTimeoutToTIRunContext``,
-    named apart so the two migrations are not confused.
-    """
-
-    description = __doc__
-
-    instructions_to_migrate_to_previous_version = (
-        schema(TIRunContext).field("execution_timeout").didnt_exist,
-    )
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py
new file mode 100644
index 0000000000000..8a0e320a1e8bb
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py
@@ -0,0 +1,39 @@
+# 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.
+
+"""Supervisor schema version 2027-02-28."""
+
+from __future__ import annotations
+
+from cadwyn import VersionChange, schema
+
+from airflow.sdk.api.datamodels._generated import TIRunContext
+
+
+class AddExecutionTimeoutToSupervisorTIRunContext(VersionChange):
+    """
+    Add the ``execution_timeout`` field carrying the task's timeout, which the supervisor enforces.
+
+    The supervisor-schema mirror of the execution API's ``AddExecutionTimeoutToTIRunContext``,
+    named apart so the two migrations are not confused.
+    """
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        schema(TIRunContext).field("execution_timeout").didnt_exist,
+    )
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 05810f0e45326..80026cf5b0317 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -130,12 +130,12 @@
     SentFDs,
     SetAssetStateStoreByName,
     SetAssetStateStoreByUri,
+    SetExecutionTimeout,
     SetRenderedFields,
     SetRenderedMapIndex,
     SetTaskStateStore,
     SetXCom,
     SkipDownstreamTasks,
-    StartExecutionTimeout,
     StartupDetails,
     SucceedTask,
     TaskBreadcrumbsResult,
@@ -2020,10 +2020,12 @@ def _handle_execution_timeout_if_needed(self):
         can still run Python signal handlers; one that has not reported a terminal state within the grace
         period is stuck (native code, SIGSEGV) and is sent SIGTERM, then SIGKILL.
 
-        For a Python task, ``StartExecutionTimeout`` arms the deadline right before ``execute()``, so bundle
-        loading and Dag parsing are not covered; Dag parsing has its own ``dagbag_import_timeout``. For a
-        task a language SDK runs, it is armed once the runtime has been sent ``StartupDetails``. Either
-        way, the process is sent SIGTERM ``killed_task_cleanup_time`` after the timeout elapses.
+        For a Python task, ``SetExecutionTimeout`` arms the deadline right before ``execute()`` with the
+        timeout the task process enforces itself, so both use the same value; bundle loading and Dag
+        parsing are not covered, as Dag parsing has its own ``dagbag_import_timeout``. For a task a language
+        SDK runs, it is armed with the run context's ``execution_timeout`` once the runtime has been sent
+        ``StartupDetails``. Either way, the process is sent SIGTERM ``killed_task_cleanup_time`` after the
+        timeout elapses.
         """
         if self._execution_timeout_due_in() > 0:
             return
@@ -2167,9 +2169,10 @@ def _handle_skip_downstream_tasks(
         self.client.task_instances.skip_downstream_tasks(self.id, msg)
         return None, {}
 
-    def _handle_start_execution_timeout(
-        self, msg: StartExecutionTimeout, log: FilteringBoundLogger, req_id: int
+    def _handle_set_execution_timeout(
+        self, msg: SetExecutionTimeout, log: FilteringBoundLogger, req_id: int
     ) -> RequestResult:
+        self._execution_timeout_seconds = msg.timeout_seconds
         self._arm_execution_timeout()
         return None, {}
 
@@ -2478,7 +2481,7 @@ def _handle_clear_asset_state_store_by_uri(
                 register_request_method(SetRenderedMapIndex, _handle_set_rendered_map_index),
                 register_request_method(SetTaskStateStore, _handle_set_task_state_store),
                 register_request_method(SkipDownstreamTasks, _handle_skip_downstream_tasks),
-                register_request_method(StartExecutionTimeout, _handle_start_execution_timeout),
+                register_request_method(SetExecutionTimeout, _handle_set_execution_timeout),
                 register_request_method(SucceedTask, _handle_finished_task),
                 register_request_method(TaskState, _handle_task_state),
                 register_request_method(TriggerDagRun, _handle_trigger_dag_run),
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index caccb6d737c37..ee6f50295c47e 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -119,10 +119,10 @@
     ResendLoggingFD,
     RetryTask,
     SentFDs,
+    SetExecutionTimeout,
     SetRenderedFields,
     SetRenderedMapIndex,
     SkipDownstreamTasks,
-    StartExecutionTimeout,
     StartupDetails,
     SucceedTask,
     TaskBreadcrumbsResult,
@@ -2352,7 +2352,7 @@ def _run_execute_callable(
             # It's possible we're already timed out, so fast-fail if true
             if timeout_seconds <= 0:
                 raise AirflowTaskTimeout()
-            SUPERVISOR_COMMS.send(StartExecutionTimeout())
+            SUPERVISOR_COMMS.send(SetExecutionTimeout(timeout_seconds=timeout_seconds))
             # Run task in timeout wrapper
             with timeout(timeout_seconds):
                 result = ctx.run(execute, context=context)
diff --git a/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py b/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
index f89e6b87e81f7..2e6db689c8fc0 100644
--- a/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
+++ b/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
@@ -550,20 +550,20 @@ def test_task_coordinates_follow_supervisor_version(self, real_migrator, startup
             assert ti["region_id"] == startup_details.ti.region_id
             assert ti["region_index"] == 7
 
-    @pytest.mark.parametrize("version", ["2026-06-16", "2026-10-30"])
+    @pytest.mark.parametrize("version", ["2026-06-16", "2026-10-30", "2027-02-28"])
     def test_execution_timeout_follows_supervisor_version(self, real_migrator, startup_details, version):
         startup_details.ti_context.execution_timeout = 90.0
         context = real_migrator.downgrade(startup_details, version).model_dump()["ti_context"]
 
-        if version == "2026-06-16":
-            assert "execution_timeout" not in context
-        else:
+        if version == "2027-02-28":
             assert context["execution_timeout"] == 90.0
+        else:
+            assert "execution_timeout" not in context
 
     def test_head_version_keeps_arg_bindings(self, real_migrator, startup_details):
         from airflow.sdk.api.datamodels._generated import LiteralArgBinding, XComArgBinding
 
-        out = real_migrator.downgrade(startup_details, "2026-10-30")
+        out = real_migrator.downgrade(startup_details, "2027-02-28")
         assert out.ti_context.arg_bindings is not None
         literal, xcom, defaulted = (a.root for a in out.ti_context.arg_bindings)
         assert isinstance(literal, LiteralArgBinding)
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 72a9e29c7e38b..1c2eba302a569 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -141,12 +141,12 @@
     SentFDs,
     SetAssetStateStoreByName,
     SetAssetStateStoreByUri,
+    SetExecutionTimeout,
     SetRenderedFields,
     SetRenderedMapIndex,
     SetTaskStateStore,
     SetXCom,
     SkipDownstreamTasks,
-    StartExecutionTimeout,
     StartupDetails,
     SucceedTask,
     TaskBreadcrumbsResult,
@@ -1711,7 +1711,7 @@ def test_execution_timeout_kills_task_that_does_not_stop(self, mocker):
         monotonic = mocker.patch("time.monotonic", autospec=True, return_value=1.0)
         proc = self._start_activity_subprocess(mocker, execution_timeout=30.0)
 
-        proc._handle_request(StartExecutionTimeout(), log=mocker.Mock(), req_id=1)
+        proc._handle_request(SetExecutionTimeout(timeout_seconds=30.0), log=mocker.Mock(), req_id=1)
 
         monotonic.return_value += 34
         proc._handle_execution_timeout_if_needed()
@@ -1763,16 +1763,24 @@ def test_python_task_does_not_arm_execution_timeout_at_startup(self, mocker):
         proc._handle_execution_timeout_if_needed()
         mock_kill.assert_not_called()
 
-    def test_start_execution_timeout_without_timeout_never_kills_task(self, mocker):
+    @pytest.mark.parametrize(
+        "context_timeout",
+        [
+            pytest.param(None, id="no_run_context_timeout"),
+            pytest.param(3600.0, id="longer_run_context_timeout"),
+        ],
+    )
+    def test_set_execution_timeout_uses_task_process_value(self, mocker, context_timeout):
+        mocker.patch("airflow.sdk.execution_time.supervisor.KILLED_TASK_CLEANUP_TIME", 5.0)
         mock_kill = mocker.patch("airflow.sdk.execution_time.supervisor.WatchedSubprocess.kill")
         monotonic = mocker.patch("time.monotonic", autospec=True, return_value=1.0)
-        proc = self._start_activity_subprocess(mocker, execution_timeout=None)
+        proc = self._start_activity_subprocess(mocker, execution_timeout=context_timeout)
 
-        proc._handle_request(StartExecutionTimeout(), log=mocker.Mock(), req_id=1)
+        proc._handle_request(SetExecutionTimeout(timeout_seconds=30.0), log=mocker.Mock(), req_id=1)
 
-        monotonic.return_value += 3600
+        monotonic.return_value += 35
         proc._handle_execution_timeout_if_needed()
-        mock_kill.assert_not_called()
+        mock_kill.assert_called_once_with(signal.SIGTERM, force=True, escalation_delay=5.0)
 
     @pytest.mark.parametrize(
         ("terminal_state", "exit_code"),
@@ -1786,7 +1794,7 @@ def test_execution_timeout_not_enforced_once_task_is_over(self, mocker, terminal
         mock_kill = mocker.patch("airflow.sdk.execution_time.supervisor.WatchedSubprocess.kill")
         monotonic = mocker.patch("time.monotonic", autospec=True, return_value=1.0)
         proc = self._start_activity_subprocess(mocker, execution_timeout=30.0)
-        proc._handle_request(StartExecutionTimeout(), log=mocker.Mock(), req_id=1)
+        proc._handle_request(SetExecutionTimeout(timeout_seconds=30.0), log=mocker.Mock(), req_id=1)
         proc._terminal_state = terminal_state
         proc._exit_code = exit_code
 
@@ -1810,12 +1818,6 @@ def test_execution_timeout_enforced_by_supervisor(
         # Far longer than the test may take: the monitor loop has to wake up for the timeout on its own
         mocker.patch("airflow.sdk.execution_time.supervisor.MIN_HEARTBEAT_INTERVAL", 30)
 
-        client_with_ti_start.task_instances.start.return_value = (
-            client_with_ti_start.task_instances.start.return_value.model_copy(
-                update={"execution_timeout": 0.1}
-            )
-        )
-
         def subprocess_main():
             import signal
 
@@ -1829,7 +1831,7 @@ def _on_term(signum, frame):
                 print("Ignoring SIGTERM", file=sys.stderr)
 
             signal.signal(signal.SIGTERM, _on_term)
-            comms.send(StartExecutionTimeout())
+            comms.send(SetExecutionTimeout(timeout_seconds=0.1))
             sleep(30)
             exit(5)
 
@@ -2771,8 +2773,8 @@ class RequestTestCase:
         test_id="set_rtif",
     ),
     RequestTestCase(
-        message=StartExecutionTimeout(),
-        test_id="start_execution_timeout",
+        message=SetExecutionTimeout(timeout_seconds=30.0),
+        test_id="set_execution_timeout",
     ),
     RequestTestCase(
         message=SetRenderedMapIndex(rendered_map_index="Label: task_1"),
diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
index ea1b4afd964b0..311b1f64407f0 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
@@ -151,11 +151,11 @@
     RetryTask,
     SetAssetStateStoreByName,
     SetAssetStateStoreByUri,
+    SetExecutionTimeout,
     SetRenderedFields,
     SetTaskStateStore,
     SetXCom,
     SkipDownstreamTasks,
-    StartExecutionTimeout,
     StartupDetails,
     SucceedTask,
     TaskRescheduleStartDate,
@@ -6951,7 +6951,7 @@ def execute(context):
             _run_execute_callable(context={}, execute=execute, task=task)
 
         task.on_kill.assert_called_once()
-        mock_supervisor_comms.send.assert_called_once_with(StartExecutionTimeout())
+        mock_supervisor_comms.send.assert_called_once_with(SetExecutionTimeout(timeout_seconds=0.01))
 
     def test_fast_fails_when_timeout_already_elapsed(self, mock_supervisor_comms):
         """A non-positive timeout fast-fails before running the callable and still calls on_kill."""
diff --git a/ts-sdk/README.md b/ts-sdk/README.md
index 490c031bfe6a6..39adbe38f23e1 100644
--- a/ts-sdk/README.md
+++ b/ts-sdk/README.md
@@ -100,7 +100,7 @@ Do not edit the table by hand. Update the manifest and run the `update-ts-sdk-re
 
 <!-- BEGIN AUTO-GENERATED LANG-SDK COMPAT MATRIX -->
 
-*Min. Airflow version: 3.4.0 · supervisor schema: 2026-10-30*
+*Min. Airflow version: 3.5.0 · supervisor schema: 2027-02-28*
 
 | Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes |
 |---|---|---|---|---|---|
diff --git a/ts-sdk/capabilities.yaml b/ts-sdk/capabilities.yaml
index 6fef910ad011d..a049b7f9fb79a 100644
--- a/ts-sdk/capabilities.yaml
+++ b/ts-sdk/capabilities.yaml
@@ -17,11 +17,11 @@
 ---
 sdk: ts
 
-min_airflow_version: "3.4.0"
+min_airflow_version: "3.5.0"
 
 # Keep in sync with SUPERVISOR_API_VERSION in src/generated/supervisor.ts. The render hook fails
 # if the two disagree.
-supervisor_schema_version: "2026-10-30"
+supervisor_schema_version: "2027-02-28"
 
 # The runtime terminates a task with SucceedTask, RetryTask, TaskState (failed/removed), or, for a
 # human-in-the-loop task, AwaitInputTask; only a triggerDagRun task emits TaskState skipped or
diff --git a/ts-sdk/schema/supervisor-schema.json b/ts-sdk/schema/supervisor-schema.json
index ebe8c2ab65ce8..f48c5ee097d18 100644
--- a/ts-sdk/schema/supervisor-schema.json
+++ b/ts-sdk/schema/supervisor-schema.json
@@ -1,6 +1,6 @@
 {
   "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "api_version": "2026-10-30",
+  "api_version": "2027-02-28",
   "description": "Apache Airflow SDK Supervisor Schema",
   "$defs": {
     "AssetAliasReferenceAssetEventDagRun": {
@@ -3791,6 +3791,27 @@
       "title": "SetAssetStateStoreByUri",
       "type": "object"
     },
+    "SetExecutionTimeout": {
+      "description": "Tell the supervisor the task's ``execution_timeout`` so it can enforce it from outside the task process.",
+      "properties": {
+        "timeout_seconds": {
+          "exclusiveMinimum": 0,
+          "title": "Timeout Seconds",
+          "type": "number"
+        },
+        "type": {
+          "const": "SetExecutionTimeout",
+          "default": "SetExecutionTimeout",
+          "title": "Type",
+          "type": "string"
+        }
+      },
+      "required": [
+        "timeout_seconds"
+      ],
+      "title": "SetExecutionTimeout",
+      "type": "object"
+    },
     "SetRenderedFields": {
       "description": "Payload for setting RTIF for a task instance.",
       "properties": {
@@ -3984,19 +4005,6 @@
       "title": "SkipDownstreamTasks",
       "type": "object"
     },
-    "StartExecutionTimeout": {
-      "description": "Tell the supervisor to start enforcing the run context's ``execution_timeout`` from outside the task process.",
-      "properties": {
-        "type": {
-          "const": "StartExecutionTimeout",
-          "default": "StartExecutionTimeout",
-          "title": "Type",
-          "type": "string"
-        }
-      },
-      "title": "StartExecutionTimeout",
-      "type": "object"
-    },
     "StartupDetails": {
       "properties": {
         "ti": {
diff --git a/ts-sdk/src/generated/supervisor.ts b/ts-sdk/src/generated/supervisor.ts
index 69285544734c5..0b6e4d7f4f171 100644
--- a/ts-sdk/src/generated/supervisor.ts
+++ b/ts-sdk/src/generated/supervisor.ts
@@ -580,13 +580,15 @@ export type Type66 = "SetAssetStateStoreByName";
 export type Uri9 = string;
 export type Key16 = string;
 export type Type67 = "SetAssetStateStoreByUri";
-export type Type68 = "SetRenderedFields";
+export type TimeoutSeconds = number;
+export type Type68 = "SetExecutionTimeout";
+export type Type69 = "SetRenderedFields";
 export type RenderedMapIndex3 = string;
-export type Type69 = "SetRenderedMapIndex";
+export type Type70 = "SetRenderedMapIndex";
 export type TiId8 = string;
 export type Key17 = string;
 export type ExpiresAt = string | null;
-export type Type70 = "SetTaskStateStore";
+export type Type71 = "SetTaskStateStore";
 export type Key18 = string;
 export type DagId21 = string;
 export type RunId14 = string;
@@ -594,10 +596,9 @@ export type TaskId9 = string;
 export type MapIndex7 = number | null;
 export type DagResult1 = boolean;
 export type MappedLength = number | null;
-export type Type71 = "SetXCom";
+export type Type72 = "SetXCom";
 export type Tasks = (string | [unknown, unknown])[];
-export type Type72 = "SkipDownstreamTasks";
-export type Type73 = "StartExecutionTimeout";
+export type Type73 = "SkipDownstreamTasks";
 export type DagRelPath = string;
 export type StartDate6 = string;
 export type SentryIntegration = string;
@@ -1782,6 +1783,16 @@ export interface SetAssetStateStoreByUri {
   value: JsonValue;
   type?: Type67;
 }
+/**
+ * Tell the supervisor the task's ``execution_timeout`` so it can enforce it from outside the task process.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "SetExecutionTimeout".
+ */
+export interface SetExecutionTimeout {
+  timeout_seconds: TimeoutSeconds;
+  type?: Type68;
+}
 /**
  * Payload for setting RTIF for a task instance.
  *
@@ -1790,7 +1801,7 @@ export interface SetAssetStateStoreByUri {
  */
 export interface SetRenderedFields {
   rendered_fields: RenderedFields;
-  type?: Type68;
+  type?: Type69;
 }
 export interface RenderedFields {
   [k: string]: JsonValue;
@@ -1803,7 +1814,7 @@ export interface RenderedFields {
  */
 export interface SetRenderedMapIndex {
   rendered_map_index: RenderedMapIndex3;
-  type?: Type69;
+  type?: Type70;
 }
 /**
  * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1814,7 +1825,7 @@ export interface SetTaskStateStore {
   key: Key17;
   value: JsonValue;
   expires_at: ExpiresAt;
-  type?: Type70;
+  type?: Type71;
 }
 /**
  * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1829,7 +1840,7 @@ export interface SetXCom {
   map_index?: MapIndex7;
   dag_result?: DagResult1;
   mapped_length?: MappedLength;
-  type?: Type71;
+  type?: Type72;
 }
 /**
  * Update state of downstream tasks within a task instance to 'skipped', while updating current task to success state.
@@ -1839,15 +1850,6 @@ export interface SetXCom {
  */
 export interface SkipDownstreamTasks {
   tasks: Tasks;
-  type?: Type72;
-}
-/**
- * Tell the supervisor to start enforcing the run context's ``execution_timeout`` from outside the task process.
- *
- * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
- * via the `definition` "StartExecutionTimeout".
- */
-export interface StartExecutionTimeout {
   type?: Type73;
 }
 /**
@@ -2048,4 +2050,4 @@ export interface XComSequenceSliceResult {
  *  (e.g. bundle metadata) and runs the migrator accordingly.
  *  Exposed so the SDK author / operator can confirm which schema
  *  version their build is pinned to. */
-export const SUPERVISOR_API_VERSION = "2026-10-30" as const;
+export const SUPERVISOR_API_VERSION = "2027-02-28" as const;

From 39d3ddb13d0c50d6d8b7d090918c5031d6ca95f0 Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Thu, 29 Oct 2026 09:33:08 +0000
Subject: [PATCH 2/8] Load the stub task only inside the is_stub branch of
 ti_run

Only a task a language SDK runs needs its definition from the serialized Dag, so keeping every use of it under that branch makes it clear a Python task never loads one.
---
 .../execution_api/routes/task_instances.py    | 55 +++++++++----------
 .../versions/head/test_task_instances.py      |  1 +
 2 files changed, 27 insertions(+), 29 deletions(-)

diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 9473accb162a8..50aae71ef7481 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -384,36 +384,33 @@ 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.
+        if 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 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.",
+                        },
+                    )
 
-        # 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:
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 2c8dc7e95a9c0..28cbc57ad124a 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -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(

From ac3325a35a57ef68bcfc0b8ed587195ade9cf1ee Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Thu, 29 Oct 2026 09:37:21 +0000
Subject: [PATCH 3/8] Let a stub task ask for no execution_timeout

A language SDK runtime has to tell a task that sets no timeout, which gets the worker's default, from one that asks for none. Sending zero for the latter needs a negative timeout to stay negative, so the supervisor, not the server, turns it into the grace period alone.
---
 .../execution_api/datamodels/taskinstance.py        |  7 ++++---
 .../execution_api/services/task_instances.py        | 10 +++++-----
 .../versions/head/test_task_instances.py            |  4 ++--
 .../src/airflow/sdk/execution_time/supervisor.py    |  6 ++++--
 .../tests/task_sdk/coordinators/test_subprocess.py  | 13 ++++++++-----
 5 files changed, 23 insertions(+), 17 deletions(-)

diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
index c90d64da1ec3f..2230b9744d138 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
@@ -558,10 +558,11 @@ class TIRunContext(BaseModel):
 
     execution_timeout: float | None = None
     """
-    The ``execution_timeout`` in seconds of a task a language SDK runs, from its serialized Dag.
+    The ``execution_timeout`` in seconds set on 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.
+    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
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
index 15e0d6fcf8cc8..0535b63fcd8ad 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
@@ -69,11 +69,11 @@ def get_arg_bindings(task: SerializedOperator | None) -> list | None:
 
 def get_execution_timeout(task: SerializedOperator | None) -> 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 is None or task.execution_timeout is None:
         return None
-    return max(task.execution_timeout.total_seconds(), 0.0)
+    return task.execution_timeout.total_seconds()
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 28cbc57ad124a..caa0266fc731c 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++ b/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.
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 80026cf5b0317..939ab29f98e01 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -2177,10 +2177,12 @@ def _handle_set_execution_timeout(
         return None, {}
 
     def _arm_execution_timeout(self) -> None:
-        if self._execution_timeout_seconds is None:
+        # Zero means no timeout, as a zero timedelta does for a Python task.
+        if not self._execution_timeout_seconds:
             return
+        # A negative timeout has already elapsed, so the task gets only the grace period.
         self._execution_timeout_enforce_at = (
-            time.monotonic() + self._execution_timeout_seconds + KILLED_TASK_CLEANUP_TIME
+            time.monotonic() + max(self._execution_timeout_seconds, 0.0) + KILLED_TASK_CLEANUP_TIME
         )
 
     def _handle_set_rendered_fields(
diff --git a/task-sdk/tests/task_sdk/coordinators/test_subprocess.py b/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
index b9f16aeeb9a84..a9b7ab4736408 100644
--- a/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
+++ b/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
@@ -1324,10 +1324,13 @@ def test_stopping_the_task_also_stops_what_the_runtime_spawned(self, tmp_path):
 
 class TestPopenActivitySubprocessExecutionTimeout:
     @pytest.mark.parametrize(
-        ("execution_timeout", "killed"),
+        ("execution_timeout", "elapsed", "killed"),
         [
-            pytest.param(30.0, True, id="with-timeout"),
-            pytest.param(None, False, id="without-timeout"),
+            pytest.param(30.0, 35, True, id="with-timeout"),
+            pytest.param(None, 35, False, id="without-timeout"),
+            pytest.param(0.0, 35, False, id="zero-means-no-timeout"),
+            pytest.param(-30.0, 3, False, id="negative-keeps-grace-period"),
+            pytest.param(-30.0, 6, True, id="negative-has-elapsed"),
         ],
     )
     @patch("airflow.sdk.execution_time.supervisor.KILLED_TASK_CLEANUP_TIME", 5.0)
@@ -1335,7 +1338,7 @@ class TestPopenActivitySubprocessExecutionTimeout:
     @patch.object(_PopenActivitySubprocess, "send_msg", autospec=True)
     @patch("time.monotonic", autospec=True, return_value=1.0)
     def test_arms_execution_timeout_from_run_context(
-        self, mock_monotonic, mock_send_msg, mock_kill, execution_timeout, killed
+        self, mock_monotonic, mock_send_msg, mock_kill, execution_timeout, elapsed, killed
     ):
         client = MagicMock()
         client.task_instances.start.return_value = TIRunContext.model_construct(
@@ -1359,7 +1362,7 @@ def test_arms_execution_timeout_from_run_context(
             sentry_integration="",
         )
 
-        mock_monotonic.return_value += 35
+        mock_monotonic.return_value += elapsed
         proc._handle_execution_timeout_if_needed()
         assert mock_kill.called is killed
 

From e6a0f919dc2941b187127659e949ca7c764cf40e Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Thu, 29 Oct 2026 11:54:13 +0000
Subject: [PATCH 4/8] Drop the note on where the default execution_timeout is
 read

---
 airflow-core/docs/core-concepts/tasks.rst            | 9 ---------
 airflow-core/src/airflow/config_templates/config.yml | 3 +--
 2 files changed, 1 insertion(+), 11 deletions(-)

diff --git a/airflow-core/docs/core-concepts/tasks.rst b/airflow-core/docs/core-concepts/tasks.rst
index dff6e52c022fe..d97c6410b41d6 100644
--- a/airflow-core/docs/core-concepts/tasks.rst
+++ b/airflow-core/docs/core-concepts/tasks.rst
@@ -145,15 +145,6 @@ if the task has not stopped within ``[core] killed_task_cleanup_time`` seconds a
 (for example because it is blocked in native code), the supervisor sends it SIGTERM, and SIGKILL after the same
 period again. Tasks run by a language SDK do not raise ``AirflowTaskTimeout`` and are stopped only this way.
 
-A task that does not set ``execution_timeout`` itself gets ``[core] default_task_execution_timeout``, which is read in
-different places depending on how the task runs:
-
-- A Python task reads it on the worker, which parses the Dag file again before running the task.
-- A task run by a language SDK reads it where the Dag processor runs, when the Dag is parsed and serialized.
-
-Set ``[core] default_task_execution_timeout`` to the same value on the Dag processor and the workers, so both kinds
-of task get the same default.
-
 In addition, sensors have a ``timeout`` parameter. This only matters for sensors in ``reschedule`` mode. ``timeout`` controls the maximum
 time allowed for the sensor to succeed. If ``timeout`` is breached, ``AirflowSensorTimeout`` will be raised and the sensor fails immediately
 without retrying.
diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml
index b18ea7091dd4b..b87f27c509364 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -398,8 +398,7 @@ core:
       description: |
         The default task execution_timeout value for the operators. Expected an integer value to
         be passed into timedelta as seconds. If not specified, then the value is considered as None,
-        meaning that the operators are never timed out by default. Python tasks read it on the worker
-        and tasks run by a language SDK read it on the Dag processor, so set the same value on both.
+        meaning that the operators are never timed out by default.
       version_added: 2.3.0
       type: integer
       example: ~

From f0e6354b63b0db5ee66238e26642360b284f6f15 Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Thu, 29 Oct 2026 21:09:09 +0800
Subject: [PATCH 5/8] Set the run context's execution_timeout only in the
 coordinator

---
 task-sdk/src/airflow/sdk/coordinators/_subprocess.py  |  4 +++-
 task-sdk/src/airflow/sdk/execution_time/supervisor.py |  1 -
 .../tests/task_sdk/execution_time/test_supervisor.py  | 11 ++---------
 3 files changed, 5 insertions(+), 11 deletions(-)

diff --git a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
index 135477c41103e..d0997db4e6e93 100644
--- a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
+++ b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
@@ -560,7 +560,8 @@ def _prepare_ti_context(self, ti: TaskInstance, ti_context: TIRunContext) -> TIR
         this task instance through the backend, deleting that XCom here leaves the stored object in the
         backend's storage.
 
-        If an XCom cannot be deleted, this try fails and the task does not run.
+        If an XCom cannot be deleted, this try fails and the task does not run. Otherwise, it also takes the
+        run context's ``execution_timeout``, which the coordinator enforces once the runtime has the task.
         """
         for key in ti_context.xcom_keys_to_clear or ():
             self.process_log.debug("Clearing XCom with key", key=key)
@@ -579,6 +580,7 @@ def _prepare_ti_context(self, ti: TaskInstance, ti_context: TIRunContext) -> TIR
                 )
                 self.kill(signal.SIGKILL)
                 return None
+        self._execution_timeout_seconds = ti_context.execution_timeout
         # The supervisor has already deleted the listed XComs, so it sends the runtime an empty list.
         # If a runtime also deletes these XComs itself, the empty list keeps the runtime from deleting
         # them a second time.
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 939ab29f98e01..ba12f3876b4e5 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -1744,7 +1744,6 @@ def _on_child_started(
             # tell us "no, stop!" for any reason)
             ti_context = self.client.task_instances.start(ti.id, self.pid, datetime.now(tz=UTC))
             self._should_retry = ti_context.should_retry
-            self._execution_timeout_seconds = ti_context.execution_timeout
             self._last_successful_heartbeat = time.monotonic()
         except Exception as e:
             # On any error kill that subprocess!
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 1c2eba302a569..bf20ba7e111b2 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -1763,18 +1763,11 @@ def test_python_task_does_not_arm_execution_timeout_at_startup(self, mocker):
         proc._handle_execution_timeout_if_needed()
         mock_kill.assert_not_called()
 
-    @pytest.mark.parametrize(
-        "context_timeout",
-        [
-            pytest.param(None, id="no_run_context_timeout"),
-            pytest.param(3600.0, id="longer_run_context_timeout"),
-        ],
-    )
-    def test_set_execution_timeout_uses_task_process_value(self, mocker, context_timeout):
+    def test_set_execution_timeout_uses_task_process_value(self, mocker):
         mocker.patch("airflow.sdk.execution_time.supervisor.KILLED_TASK_CLEANUP_TIME", 5.0)
         mock_kill = mocker.patch("airflow.sdk.execution_time.supervisor.WatchedSubprocess.kill")
         monotonic = mocker.patch("time.monotonic", autospec=True, return_value=1.0)
-        proc = self._start_activity_subprocess(mocker, execution_timeout=context_timeout)
+        proc = self._start_activity_subprocess(mocker, execution_timeout=3600.0)
 
         proc._handle_request(SetExecutionTimeout(timeout_seconds=30.0), log=mocker.Mock(), req_id=1)
 

From 9936a0461b5db0c5419e0b719d031c7fceb7a2db Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Fri, 30 Oct 2026 18:31:34 +0800
Subject: [PATCH 6/8] Address review nits on execution_timeout docs and schema
 description

---
 .../execution_api/datamodels/taskinstance.py    | 17 +++++++++--------
 go-sdk/pkg/execution/genmodels/models.gen.go    |  5 ++++-
 go-sdk/schema/supervisor-schema.json            |  1 +
 .../airflow/sdk/api/datamodels/_generated.py    |  8 +++++++-
 .../src/airflow/sdk/coordinators/_subprocess.py |  5 ++---
 .../sdk/execution_time/schema/schema.json       |  1 +
 .../airflow/sdk/execution_time/supervisor.py    | 10 ++++------
 ts-sdk/schema/supervisor-schema.json            |  1 +
 ts-sdk/src/generated/supervisor.ts              |  3 +++
 9 files changed, 32 insertions(+), 19 deletions(-)

diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
index 2230b9744d138..4a98f3f205e70 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
@@ -556,14 +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 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.
-    """
+    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
     """
diff --git a/go-sdk/pkg/execution/genmodels/models.gen.go b/go-sdk/pkg/execution/genmodels/models.gen.go
index 9d94e14942365..3cefba3e3f9e0 100644
--- a/go-sdk/pkg/execution/genmodels/models.gen.go
+++ b/go-sdk/pkg/execution/genmodels/models.gen.go
@@ -1709,7 +1709,10 @@ type TIRunContext struct {
 	// DagRun corresponds to the JSON schema field "dag_run".
 	DagRun DagRun `msgpack:"dag_run"`
 
-	// ExecutionTimeout corresponds to the JSON schema field "execution_timeout".
+	// 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.
 	ExecutionTimeout interface{} `msgpack:"execution_timeout,omitempty"`
 
 	// ExpandedTICount corresponds to the JSON schema field "expanded_ti_count".
diff --git a/go-sdk/schema/supervisor-schema.json b/go-sdk/schema/supervisor-schema.json
index f48c5ee097d18..1db287cb17db5 100644
--- a/go-sdk/schema/supervisor-schema.json
+++ b/go-sdk/schema/supervisor-schema.json
@@ -5409,6 +5409,7 @@
             }
           ],
           "default": null,
+          "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.",
           "title": "Execution Timeout"
         },
         "multi_team": {
diff --git a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
index 648aa3bda2833..701be36e39513 100644
--- a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
+++ b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
@@ -1017,5 +1017,11 @@ class TIRunContext(BaseModel):
     should_retry: Annotated[bool | None, Field(title="Should Retry")] = False
     start_date: Annotated[AwareDatetime | None, Field(title="Start Date")] = None
     arg_bindings: Annotated[list[TaskArgBinding] | None, Field(title="Arg Bindings")] = None
-    execution_timeout: Annotated[float | None, Field(title="Execution Timeout")] = None
+    execution_timeout: Annotated[
+        float | None,
+        Field(
+            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.",
+            title="Execution Timeout",
+        ),
+    ] = None
     multi_team: Annotated[bool | None, Field(title="Multi Team")] = False
diff --git a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
index d0997db4e6e93..426a8f5b5e27e 100644
--- a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
+++ b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
@@ -560,9 +560,9 @@ def _prepare_ti_context(self, ti: TaskInstance, ti_context: TIRunContext) -> TIR
         this task instance through the backend, deleting that XCom here leaves the stored object in the
         backend's storage.
 
-        If an XCom cannot be deleted, this try fails and the task does not run. Otherwise, it also takes the
-        run context's ``execution_timeout``, which the coordinator enforces once the runtime has the task.
+        If an XCom cannot be deleted, this try fails and the task does not run.
         """
+        self._execution_timeout_seconds = ti_context.execution_timeout
         for key in ti_context.xcom_keys_to_clear or ():
             self.process_log.debug("Clearing XCom with key", key=key)
             try:
@@ -580,7 +580,6 @@ def _prepare_ti_context(self, ti: TaskInstance, ti_context: TIRunContext) -> TIR
                 )
                 self.kill(signal.SIGKILL)
                 return None
-        self._execution_timeout_seconds = ti_context.execution_timeout
         # The supervisor has already deleted the listed XComs, so it sends the runtime an empty list.
         # If a runtime also deletes these XComs itself, the empty list keeps the runtime from deleting
         # them a second time.
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
index f48c5ee097d18..1db287cb17db5 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
@@ -5409,6 +5409,7 @@
             }
           ],
           "default": null,
+          "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.",
           "title": "Execution Timeout"
         },
         "multi_team": {
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index ba12f3876b4e5..4f9622e34fa33 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -2019,12 +2019,10 @@ def _handle_execution_timeout_if_needed(self):
         can still run Python signal handlers; one that has not reported a terminal state within the grace
         period is stuck (native code, SIGSEGV) and is sent SIGTERM, then SIGKILL.
 
-        For a Python task, ``SetExecutionTimeout`` arms the deadline right before ``execute()`` with the
-        timeout the task process enforces itself, so both use the same value; bundle loading and Dag
-        parsing are not covered, as Dag parsing has its own ``dagbag_import_timeout``. For a task a language
-        SDK runs, it is armed with the run context's ``execution_timeout`` once the runtime has been sent
-        ``StartupDetails``. Either way, the process is sent SIGTERM ``killed_task_cleanup_time`` after the
-        timeout elapses.
+        ``SetExecutionTimeout`` arms the deadline right before ``execute()`` with the timeout the task
+        process enforces itself, so both use the same value; bundle loading and Dag parsing are not covered,
+        as Dag parsing has its own ``dagbag_import_timeout``. The process is sent SIGTERM
+        ``killed_task_cleanup_time`` after the timeout elapses.
         """
         if self._execution_timeout_due_in() > 0:
             return
diff --git a/ts-sdk/schema/supervisor-schema.json b/ts-sdk/schema/supervisor-schema.json
index f48c5ee097d18..1db287cb17db5 100644
--- a/ts-sdk/schema/supervisor-schema.json
+++ b/ts-sdk/schema/supervisor-schema.json
@@ -5409,6 +5409,7 @@
             }
           ],
           "default": null,
+          "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.",
           "title": "Execution Timeout"
         },
         "multi_team": {
diff --git a/ts-sdk/src/generated/supervisor.ts b/ts-sdk/src/generated/supervisor.ts
index 0b6e4d7f4f171..8cc93eb7bedf4 100644
--- a/ts-sdk/src/generated/supervisor.ts
+++ b/ts-sdk/src/generated/supervisor.ts
@@ -270,6 +270,9 @@ export type TaskId1 = string;
 export type Name9 = string;
 export type Kind1 = "literal";
 export type FromDefault = boolean;
+/**
+ * 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.
+ */
 export type ExecutionTimeout = number | null;
 export type MultiTeam = boolean;
 export type Type13 = "TaskCallbackRequest";

From 22a00beab2c4f406a6a1691767bee4280a0b1059 Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Sat, 31 Oct 2026 11:55:16 +0800
Subject: [PATCH 7/8] Reword execution_timeout comment for runtimes that send
 SetExecutionTimeout

---
 task-sdk/src/airflow/sdk/coordinators/_subprocess.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
index 426a8f5b5e27e..241f8733aea11 100644
--- a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
+++ b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
@@ -546,7 +546,8 @@ def _on_child_started(
             bundle_info=bundle_info,
             sentry_integration=sentry_integration,
         )
-        # Language runtimes do not send SetExecutionTimeout, so the clock starts once they have the task.
+        # A language runtime may not send SetExecutionTimeout, so the clock starts once it has the task.
+        # If the runtime sends it, the clock restarts from that message.
         self._arm_execution_timeout()
 
     def _prepare_ti_context(self, ti: TaskInstance, ti_context: TIRunContext) -> TIRunContext | None:

From 27a5bcafdefe5275eb94003ae9b74add0774a986 Mon Sep 17 00:00:00 2001
From: ChickenBenny <zxc123benny14159@gmail.com>
Date: Sun, 1 Nov 2026 16:15:33 +0800
Subject: [PATCH 8/8] Address review nits on ti_run stub task lookup

---
 .../execution_api/routes/task_instances.py            | 11 ++++++++---
 .../execution_api/services/task_instances.py          |  8 ++++----
 2 files changed, 12 insertions(+), 7 deletions(-)

diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 50aae71ef7481..05fbe062d5aef 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -386,10 +386,15 @@ def ti_run(
 
         # 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.
-        if ti.is_stub and (serialized_task := get_task(dag_bag, ti, session=session)) is not None:
+        # 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 client_supports_arg_bindings() and (arg_bindings := get_arg_bindings(serialized_task)):
+            if arg_bindings := get_arg_bindings(serialized_task):
                 try:
                     context.arg_bindings = get_arg_bindings_adapter().validate_python(arg_bindings)
                 except ValidationError:
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
index 0535b63fcd8ad..9f33341144e98 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
@@ -54,7 +54,7 @@ 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.
 
@@ -62,18 +62,18 @@ def get_arg_bindings(task: SerializedOperator | None) -> list | None:
     ``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, or ``None`` if it sets none.
 
     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 task.execution_timeout is None:
+    if task.execution_timeout is None:
         return None
     return task.execution_timeout.total_seconds()