From 882658cec5f31d39e132675e876ab53758051432 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
 <41898282+github-actions[bot]@users.noreply.github.com>
Date: Tue, 14 Apr 2026 20:09:30 +0200
Subject: [PATCH 1/4] [v3-2-test] Bump actions/github-script in the
 github-actions-updates group (#65150) (#65160)

Bumps the github-actions-updates group with 1 update: [actions/github-script](https://github.com/actions/github-script).

Updates `actions/github-script` from 8.0.0 to 9.0.0
- [Release notes](https://github.com/actions/github-script/releases)
- [Commits](https://github.com/actions/github-script/compare/ed597411d8f924073f98dfc5c65a23a2325f34cd...3a2844b7e9c422d3c10d287c895573f7108da1b3)
(cherry picked from commit e5a047ca8d23614cefeaad89e08eb1e3b0482e60)



---
updated-dependencies:
- dependency-name: actions/github-script
  dependency-version: 9.0.0
  dependency-type: direct:production
  update-type: version-update:semver-major
  dependency-group: github-actions-updates
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

From e72366139cc85863ba43313f44228f79e6d72997 Mon Sep 17 00:00:00 2001
From: Jarek Potiuk <jarek@potiuk.com>
Date: Tue, 14 Apr 2026 20:20:55 +0200
Subject: [PATCH 2/4] [v3-2-test] Added breeze generate issue content for
 airflow-ctl (#65042) (#65241)

* Add breeze generate issue content for airflow-ctl

* add new command to doc
(cherry picked from commit b24538b0bc3a53d8c666b62e52e023a17d102467)

Co-authored-by: Justin Pakzad <114518232+justinpakzad@users.noreply.github.com>

From a50c1230f3d02a60c44f8c2a24c099e62e74024b Mon Sep 17 00:00:00 2001
From: Jarek Potiuk <jarek@potiuk.com>
Date: Tue, 14 Apr 2026 20:28:34 +0200
Subject: [PATCH 3/4] [v3-2-test] Run release calendar verification on its own
 schedule (#65118) (#65242)

* Move release calendar verification to its own scheduled workflow

Run dev/verify_release_calendar.py from a dedicated daily scheduled
workflow instead of as a canary job in the main CI pipeline, and
notify the #release-management Slack channel when the check fails so
the issue is surfaced to release managers directly.

* Include wiki and calendar links in release calendar Slack alert
(cherry picked from commit 048e9a191b4a6625bb4bbad23fad537b256f4062)

From 4448dfd8a0f09339123a72f4cf755e3a3dc28f97 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Sun, 19 Apr 2026 01:44:07 +0100
Subject: [PATCH 4/4] [v3-2-test] Include TI UUID in scheduler, DAG processor,
 triggerer, and worker logs (#65458)

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

- Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
- Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
- Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
- DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

* Move ti_id into TaskInstance.__repr__; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.__repr__ once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:
- TaskInstance.__repr__ now appends `ti_id={self.id}` before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
- Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing `%s` format arg was a TaskInstance; repr now supplies ti_id.
- Kept the explicit `ti_id=%s` in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
- Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not __repr__.
- Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

* Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.__repr__, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

* Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com>
---
 .../src/airflow/dag_processing/processor.py   | 24 ++++++++++++++++---
 .../src/airflow/jobs/scheduler_job_runner.py  | 12 +++++-----
 .../src/airflow/jobs/triggerer_job_runner.py  | 11 ++++++++-
 .../src/airflow/models/taskinstance.py        |  2 +-
 .../tests/unit/jobs/test_scheduler_job.py     |  8 ++++---
 .../airflow/sdk/execution_time/task_runner.py | 14 +++++++++--
 6 files changed, 55 insertions(+), 16 deletions(-)

diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py
index 6cbef2fef67bb..148d978084dd8 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -304,7 +304,14 @@ def _execute_callbacks(
     dagbag: DagBag, callback_requests: list[CallbackRequest], log: FilteringBoundLogger
 ) -> None:
     for request in callback_requests:
-        log.debug("Processing Callback Request", request=request.to_json())
+        if isinstance(request, (TaskCallbackRequest, EmailRequest)):
+            log.debug(
+                "Processing Callback Request",
+                request=request.to_json(),
+                ti_id=str(request.ti.id),
+            )
+        else:
+            log.debug("Processing Callback Request", request=request.to_json())
         with BundleVersionLock(
             bundle_name=request.bundle_name,
             bundle_version=request.bundle_version,
@@ -369,6 +376,7 @@ def _execute_task_callbacks(dagbag: DagBag, request: TaskCallbackRequest, log: F
             dag_id=request.ti.dag_id,
             task_id=request.ti.task_id,
             run_id=request.ti.run_id,
+            ti_id=str(request.ti.id),
         )
         return
 
@@ -418,11 +426,21 @@ def get_callback_representation(callback):
 
     for idx, callback in enumerate(callbacks):
         callback_repr = get_callback_representation(callback)
-        log.info("Executing Task callback at index %d: %s", idx, callback_repr)
+        log.info(
+            "Executing Task callback at index %d: %s (ti_id=%s)",
+            idx,
+            callback_repr,
+            request.ti.id,
+        )
         try:
             callback(context)
         except Exception:
-            log.exception("Error in callback at index %d: %s", idx, callback_repr)
+            log.exception(
+                "Error in callback at index %d: %s (ti_id=%s)",
+                idx,
+                callback_repr,
+                request.ti.id,
+            )
 
 
 def _execute_email_callbacks(dagbag: DagBag, request: EmailRequest, log: FilteringBoundLogger) -> None:
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index bda95bbf4b06a..44fa6e2141a23 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -971,9 +971,8 @@ def _get_sentry_integration(executor: BaseExecutor) -> str:
                 continue
 
             self.log.debug(
-                "Queueing workload for TI: %s id=%s try_number=%d state=%s scheduler_job_id=%s executor=%s",
+                "Queueing workload for TI: %s try_number=%d state=%s scheduler_job_id=%s executor=%s",
                 ti,
-                ti.id,
                 ti.try_number,
                 ti.state,
                 self.job.id,
@@ -1250,12 +1249,11 @@ def process_executor_events(
             if ti.try_number != try_number:
                 cls.logger().warning(
                     "TI try_number mismatch: db_try_number=%d event_try_number=%d "
-                    "ti=%s ti_id=%s state=%s job_id=%s. "
+                    "ti=%s state=%s job_id=%s. "
                     "Another scheduler may have already modified this TI.",
                     ti.try_number,
                     try_number,
                     ti,
-                    ti.id,
                     ti.state,
                     job_id,
                 )
@@ -1267,7 +1265,7 @@ def process_executor_events(
                 continue
 
             msg = (
-                "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s, map_index=%s, "
+                "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s, map_index=%s, ti_id=%s, "
                 "run_start_date=%s, run_end_date=%s, "
                 "run_duration=%s, state=%s, executor=%s, executor_state=%s, try_number=%s, max_tries=%s, "
                 "pool=%s, queue=%s, priority_weight=%d, operator=%s, queued_dttm=%s, scheduled_dttm=%s,"
@@ -1279,6 +1277,7 @@ def process_executor_events(
                 ti.task_id,
                 ti.run_id,
                 ti.map_index,
+                ti.id,
                 ti.start_date,
                 ti.end_date,
                 ti.duration,
@@ -1387,7 +1386,8 @@ def process_executor_events(
                 # Handle cleared tasks that were successfully terminated by executor
                 if ti.state == TaskInstanceState.RESTARTING and state == TaskInstanceState.SUCCESS:
                     cls.logger().info(
-                        "Task %s was cleared and successfully terminated. Setting to scheduled for retry.", ti
+                        "Task %s was cleared and successfully terminated. Setting to scheduled for retry.",
+                        ti,
                     )
                     # Adjust max_tries to allow retry beyond normal limits (like clearing does)
                     ti.max_tries = ti.try_number + ti.task.retries
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 4cfe262b71ec3..a25632acac701 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -1209,7 +1209,16 @@ async def run_trigger(self, trigger_id: int, trigger: BaseTrigger, timeout_after
 
             await greenback.ensure_portal()
 
-        bind_log_contextvars(trigger_id=trigger_id)
+        ti = trigger.task_instance
+        bind_log_contextvars(
+            trigger_id=trigger_id,
+            ti_id=str(ti.id) if ti else None,
+            dag_id=ti.dag_id if ti else None,
+            task_id=ti.task_id if ti else None,
+            run_id=ti.run_id if ti else None,
+            try_number=ti.try_number if ti else None,
+            map_index=ti.map_index if ti else None,
+        )
 
         name = self.triggers[trigger_id]["name"]
         self.log.info("trigger %s starting", name)
diff --git a/airflow-core/src/airflow/models/taskinstance.py b/airflow-core/src/airflow/models/taskinstance.py
index 742bc36372474..393ca23e21593 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -1095,7 +1095,7 @@ def __repr__(self) -> str:
         prefix = f"<TaskInstance: {self.dag_id}.{self.task_id} {self.run_id} "
         if self.map_index != -1:
             prefix += f"map_index={self.map_index} "
-        return prefix + f"[{self.state}]>"
+        return prefix + f"[{self.state}] ti_id={self.id}>"
 
     def next_retry_datetime(self):
         """
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index fd77644bc44b5..1fd8c91f3cbcc 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -691,7 +691,7 @@ def test_process_executor_events_with_callback(
             bundle_name="dag_maker",
             bundle_version=None,
             msg=f"Executor {executor} reported that the task instance "
-            "<TaskInstance: test_process_executor_events_with_callback.dummy_task test [queued]> "
+            f"<TaskInstance: test_process_executor_events_with_callback.dummy_task test [queued] ti_id={ti1.id}> "
             "finished with state failed, but the task instance's state attribute is queued. "
             "Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally",
             context_from_server=mock.ANY,
@@ -1903,11 +1903,13 @@ def test_not_enough_pool_slots(self, caplog, dag_maker):
         some_pool = Pool(pool="some_pool", slots=2, description="my pool", include_deferred=False)
         session.add(some_pool)
         session.commit()
+        cannot_run_ti_id = next(t for t in dr.task_instances if t.task_id == "cannot_run").id
         with caplog.at_level(logging.WARNING):
             self.job_runner._executable_task_instances_to_queued(max_tis=32, session=session)
             assert (
-                "Not executing <TaskInstance: "
-                "SchedulerJobTest.test_test_not_enough_pool_slots.cannot_run test [scheduled]>. "
+                f"Not executing <TaskInstance: "
+                f"SchedulerJobTest.test_test_not_enough_pool_slots.cannot_run test [scheduled] "
+                f"ti_id={cannot_run_ti_id}>. "
                 "Requested pool slots (4) are greater than total pool slots: '2' for pool: some_pool"
                 in caplog.text
             )
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 6e0b0766be380..1c251e6668327 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -38,6 +38,7 @@
 from opentelemetry import trace
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 from pydantic import AwareDatetime, ConfigDict, Field, JsonValue, TypeAdapter
+from structlog.contextvars import bind_contextvars
 
 from airflow.dag_processing.bundles.base import BaseDagBundle, BundleVersionLock
 from airflow.dag_processing.bundles.manager import DagBundlesManager
@@ -914,6 +915,17 @@ def startup(msg: StartupDetails) -> tuple[RuntimeTaskInstance, Context, Logger]:
 
         setproctitle(f"airflow worker -- {msg.ti.id}")
 
+    # Bind TI identifiers so every subsequent log line from this worker process
+    # carries ti_id, enabling single-grep lifecycle reconstruction across components.
+    bind_contextvars(
+        ti_id=str(msg.ti.id),
+        dag_id=msg.ti.dag_id,
+        task_id=msg.ti.task_id,
+        run_id=msg.ti.run_id,
+        try_number=msg.ti.try_number,
+        map_index=msg.ti.map_index,
+    )
+
     try:
         get_listener_manager().hook.on_starting(component=TaskRunnerMarker())
     except Exception:
@@ -1160,8 +1172,6 @@ def _validate_task_inlets_and_outlets(*, ti: RuntimeTaskInstance, log: Logger) -
 def _defer_task(
     defer: TaskDeferred, ti: RuntimeTaskInstance, log: Logger
 ) -> tuple[ToSupervisor, TaskInstanceState]:
-    # TODO: Should we use structlog.bind_contextvars here for dag_id, task_id & run_id?
-
     log.info("Pausing task as DEFERRED. ", dag_id=ti.dag_id, task_id=ti.task_id, run_id=ti.run_id)
     classpath, trigger_kwargs = defer.trigger.serialize()
     queue: str | None = None
