Deliver the per-run log ID template to workers via TIRunContext

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

@@ -435,6 +435,12 @@ class TIRunContext(BaseModel):
     always reflects when the task *first* started, not when it was rescheduled/resumed.
     """
 
+    log_id_template: str | None = None
+    """
+    Elasticsearch/OpenSearch log id template pinned to this Dag run (``LogTemplate.elasticsearch_id``),
+    so remote log writers use the template that was in effect when the run was created.
+    """
+
 
 class PrevSuccessfulDagRunResponse(BaseModel):
     """Schema for response with previous successful DagRun information for Task Template Context."""

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

@@ -76,7 +76,7 @@
     require_auth,
 )
 from airflow.configuration import conf
-from airflow.exceptions import InvalidPartitionKeyError, TaskNotFound
+from airflow.exceptions import AirflowException, InvalidPartitionKeyError, TaskNotFound
 from airflow.models.asset import AssetActive
 from airflow.models.base import ID_LEN
 from airflow.models.dag import DagModel
@@ -310,6 +310,9 @@ def ti_run(
             should_retry=_is_eligible_to_retry(previous_state, ti.try_number, ti.max_tries),
         )
 
+        with contextlib.suppress(AirflowException):  # no LogTemplate row: worker falls back to conf
+            context.log_id_template = dr.get_log_template(session=session).elasticsearch_id
+
         # Only set if they are non-null
         if ti.next_method:
             context.next_method = ti.next_method

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

@@ -51,9 +51,14 @@
     AddTeamNameField,
     AddVariableKeysEndpoint,
 )
+from airflow.api_fastapi.execution_api.versions.v2026_09_30 import AddLogIdTemplateField
 
 bundle = VersionBundle(
     HeadVersion(),
+    Version(
+        "2026-09-30",
+        AddLogIdTemplateField,
+    ),
     Version(
         "2026-06-30",
         AddVariableKeysEndpoint,

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

@@ -0,0 +1,40 @@
+# 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 cadwyn import (
+    ResponseInfo,
+    VersionChange,
+    convert_response_to_previous_version_for,
+    schema,
+)
+
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
+
+
+class AddLogIdTemplateField(VersionChange):
+    """Add the Dag-run-pinned `log_id_template` field to TIRunContext."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("log_id_template").didnt_exist,)
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def remove_log_id_template_field(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Remove log_id_template field for older API versions."""
+        response.body.pop("log_id_template", None)

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

@@ -278,6 +278,7 @@ def test_ti_run_state_to_running(
             "variables": [],
             "connections": [],
             "xcom_keys_to_clear": [],
+            "log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}",
         }
         # upstream_map_indexes is now computed by Task SDK, not returned by the server in HEAD version
         assert "upstream_map_indexes" not in result
@@ -692,6 +693,7 @@ def test_next_kwargs_still_encoded(self, client, session, create_task_instance,
             "next_method": "execute_complete",
             "next_kwargs": expected_next_kwargs,
             "start_date": None,
+            "log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}",
         }
 
     @pytest.mark.parametrize("resume", [True, False])
@@ -766,6 +768,7 @@ def test_next_kwargs_determines_start_date_update(self, client, session, create_
             "xcom_keys_to_clear": [],
             "next_method": "execute_complete",
             "next_kwargs": expected_next_kwargs,
+            "log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}",
         }
         session.expunge_all()
         ti = session.get(TaskInstance, ti.id)
@@ -1030,6 +1033,76 @@ def test_ti_run_populates_team_name(
         assert response.status_code == 200
         assert response.json()["dag_run"]["team_name"] == (team_name if expect_team else None)
 
+    def test_ti_run_populates_log_id_template(
+        self, client, session, create_task_instance, create_log_template, time_machine
+    ):
+        """``log_id_template`` echoes the LogTemplate row pinned to the Dag run."""
+        instant = timezone.parse("2024-09-30T12:00:00Z")
+        time_machine.move_to(instant, tick=False)
+
+        custom_elasticsearch_id = "{dag_id}-{task_id}-{logical_date}-{try_number}"
+        create_log_template("{{ ti.dag_id }}.log", custom_elasticsearch_id)
+
+        # The Dag run pins the newest LogTemplate row at creation time.
+        ti = create_task_instance(
+            task_id="test_ti_run_populates_log_id_template",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=instant,
+            dag_id=str(uuid4()),
+        )
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": instant.isoformat(),
+            },
+        )
+
+        assert response.status_code == 200
+        assert response.json()["log_id_template"] == custom_elasticsearch_id
+
+    def test_ti_run_populates_latest_log_id_template(
+        self, client, session, create_task_instance, create_log_template, time_machine
+    ):
+        """Each Dag run pins whichever LogTemplate row is newest when it is created."""
+        instant = timezone.parse("2024-09-30T12:00:00Z")
+        time_machine.move_to(instant, tick=False)
+
+        for i in range(3):
+            elasticsearch_id = f"{{dag_id}}-{{task_id}}-{{logical_date}}-{{try_number}}-v{i}"
+            create_log_template(f"{{{{ ti.dag_id }}}}-v{i}.log", elasticsearch_id)
+
+            ti = create_task_instance(
+                task_id="test_ti_run_populates_latest_log_id_template",
+                state=State.QUEUED,
+                dagrun_state=DagRunState.RUNNING,
+                session=session,
+                start_date=instant,
+                dag_id=str(uuid4()),
+            )
+            session.commit()
+
+            response = client.patch(
+                f"/execution/task-instances/{ti.id}/run",
+                json={
+                    "state": "running",
+                    "hostname": "random-hostname",
+                    "unixname": "random-unixname",
+                    "pid": 100,
+                    "start_date": instant.isoformat(),
+                },
+            )
+
+            assert response.status_code == 200
+            assert response.json()["log_id_template"] == elasticsearch_id
+
     def test_ti_run_creates_audit_log(self, client, session, create_task_instance, time_machine):
         """Test that transitioning to RUNNING creates an audit log record."""
         instant_str = "2024-09-30T12:00:00Z"

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_09_30/__init__.py

@@ -0,0 +1,16 @@
+# 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.

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

@@ -0,0 +1,85 @@
+# 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
+
+import pytest
+
+from airflow._shared.timezones import timezone
+from airflow.utils.state import DagRunState, State
+
+from tests_common.test_utils.db import clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+TIMESTAMP_STR = "2024-09-30T12:00:00Z"
+TIMESTAMP = timezone.parse(TIMESTAMP_STR)
+
+RUN_PATCH_BODY = {
+    "state": "running",
+    "hostname": "h",
+    "unixname": "u",
+    "pid": 1,
+    "start_date": TIMESTAMP_STR,
+}
+
+
+@pytest.fixture
+def old_ver_client(client):
+    """Execution API version immediately before ``log_id_template`` was added."""
+    client.headers["Airflow-API-Version"] = "2026-06-30"
+    return client
+
+
+class TestLogIdTemplateFieldBackwardCompat:
+    @pytest.fixture(autouse=True)
+    def _freeze_time(self, time_machine):
+        time_machine.move_to(TIMESTAMP_STR, tick=False)
+
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    def test_old_version_strips_log_id_template(self, old_ver_client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_log_id_template_downgrade",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        session.commit()
+
+        response = old_ver_client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        assert response.status_code == 200
+        assert "log_id_template" not in response.json()
+
+    def test_head_version_includes_log_id_template(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_log_id_template_head",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        session.commit()
+
+        response = client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        assert response.status_code == 200
+        assert response.json()["log_id_template"] == "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}"

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

@@ -27,7 +27,7 @@
 
 from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue, RootModel
 
-API_VERSION: Final[str] = "2026-06-30"
+API_VERSION: Final[str] = "2026-09-30"
 
 
 class AssetAliasReferenceAssetEventDagRun(BaseModel):
@@ -797,3 +797,4 @@ class TIRunContext(BaseModel):
     xcom_keys_to_clear: Annotated[list[str] | None, Field(title="Xcom Keys To Clear")] = None
     should_retry: Annotated[bool | None, Field(title="Should Retry")] = False
     start_date: Annotated[AwareDatetime | None, Field(title="Start Date")] = None
+    log_id_template: Annotated[str | None, Field(title="Log Id Template")] = None

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-06-16",
+  "api_version": "2026-09-30",
   "description": "Apache Airflow SDK Supervisor Schema",
   "$defs": {
     "AssetAliasReferenceAssetEventDagRun": {
@@ -4878,6 +4878,18 @@
           ],
           "default": null,
           "title": "Start Date"
+        },
+        "log_id_template": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Log Id Template"
         }
       },
       "required": [

task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py

@@ -37,8 +37,11 @@ def get_bundle() -> VersionBundle:
     """
     from cadwyn import HeadVersion, Version, VersionBundle
 
+    from airflow.sdk.execution_time.schema.versions.v2026_09_30 import AddLogIdTemplateField
+
     return VersionBundle(
         HeadVersion(),
+        Version("2026-09-30", AddLogIdTemplateField),
         Version("2026-06-16"),
     )
 

task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_09_30.py

@@ -0,0 +1,30 @@
+# 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 cadwyn import VersionChange, schema
+
+from airflow.sdk.api.datamodels._generated import TIRunContext
+
+
+class AddLogIdTemplateField(VersionChange):
+    """Add the Dag-run-pinned `log_id_template` field to TIRunContext (nested in StartupDetails)."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("log_id_template").didnt_exist,)