apache · amoghrajesh · Sep 24, 2026 · Sep 12, 2026 · Sep 15, 2026 · Sep 18, 2026
diff --git a/airflow-core/docs/core-concepts/tasks.rst b/airflow-core/docs/core-concepts/tasks.rst
@@ -296,8 +296,8 @@ A policy that raises an ordinary exception, or returns something other than a
 ``RetryDecision``, is logged and treated as DEFAULT, so one broken policy does not take the
 rules after it down with it. The winning decision's reason names the policy that decided and
 then what the earlier ones said (``HTTPStatusRetryPolicy: HTTP 503 (after ExceptionRetryPolicy:
-no decision)``). On a RETRY that string is the task's ``retry_reason``; on FAIL, or when no
-policy decided, it appears in the task log as the ``Retry policy decision`` line.
+no decision)``). That string is the task's ``retry_reason`` on a FAIL as well as a RETRY; when
+no policy decided, it appears only in the task log as the ``Retry policy decision`` line.
 
 Custom retry policies
 ~~~~~~~~~~~~~~~~~~~~~

@@ -85,6 +85,7 @@ class TITerminalStatePayload(StrictBaseModel):
     end_date: UtcDateTime
     """When the task completed executing"""
     rendered_map_index: str | None = None
+    retry_reason: str | None = None
 
 
 class TISuccessStatePayload(StrictBaseModel):

@@ -672,6 +672,8 @@ def _create_ti_state_update_query_and_update_state(
 
         if updated_state == TaskInstanceState.FAILED:
             # This is the only case needs extra handling for TITerminalStatePayload
+            if isinstance(ti_patch_payload, TITerminalStatePayload) and ti_patch_payload.retry_reason:
+                query = query.values(retry_reason=ti_patch_payload.retry_reason[:500])
             if ti is not None:
                 _handle_fail_fast_for_dag(ti=ti, dag_id=dag_id, session=session, dag_bag=dag_bag)
         elif isinstance(ti_patch_payload, TIRetryStatePayload):

@@ -56,6 +56,7 @@
     AddArgBindingsToTIRunContext,
     AddCallbackRunEndpoint,
     AddMultiTeamToTIRunContext,
+    AddTerminalStateRetryReasonField,
 )
 
 bundle = VersionBundle(
@@ -64,6 +65,7 @@
         "2026-10-30",
         AddArgBindingsToTIRunContext,
         AddCallbackRunEndpoint,
+        AddTerminalStateRetryReasonField,
         AddMultiTeamToTIRunContext,
     ),
     Version(

@@ -26,7 +26,7 @@
     schema,
 )
 
-from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext, TITerminalStatePayload
 
 
 class AddArgBindingsToTIRunContext(VersionChangeWithSideEffects):
@@ -54,6 +54,16 @@ class AddCallbackRunEndpoint(VersionChange):
     )
 
 
+class AddTerminalStateRetryReasonField(VersionChange):
+    """Add the `retry_reason` field to TITerminalStatePayload for failed retry-policy decisions."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        schema(TITerminalStatePayload).field("retry_reason").didnt_exist,
+    )
+
+
 class AddMultiTeamToTIRunContext(VersionChange):
     """Add ``multi_team`` so a worker can determine multi-team (e.g. for plugin scoping) without needing to trust its own config."""
 

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
@@ -2490,6 +2490,72 @@ def test_ti_update_state_to_failed_table_check(self, client, session, create_tas
         assert ti.next_kwargs is None
         assert ti.duration == 3600.00
 
+    def test_ti_update_state_to_failed_persists_retry_reason(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_ti_update_state_to_failed_persists_retry_reason",
+            state=State.RUNNING,
+        )
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/state",
+            json={
+                "state": TerminalTIState.FAILED,
+                "end_date": DEFAULT_END_DATE.isoformat(),
+                "retry_reason": "auth error, do not retry",
+            },
+        )
+
+        assert response.status_code == 204
+
+        session.expire_all()
+        ti = session.get(TaskInstance, ti.id)
+        assert ti.state == State.FAILED
+        assert ti.retry_reason == "auth error, do not retry"
+
+    def test_ti_update_state_to_failed_truncates_retry_reason(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_ti_update_state_to_failed_truncates_retry_reason",
+            state=State.RUNNING,
+        )
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/state",
+            json={
+                "state": TerminalTIState.FAILED,
+                "end_date": DEFAULT_END_DATE.isoformat(),
+                "retry_reason": "x" * 600,
+            },
+        )
+
+        assert response.status_code == 204
+
+        session.expire_all()
+        ti = session.get(TaskInstance, ti.id)
+        assert ti.retry_reason == "x" * 500
+
+    def test_ti_update_state_to_failed_without_retry_reason(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_ti_update_state_to_failed_without_retry_reason",
+            state=State.RUNNING,
+        )
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/state",
+            json={
+                "state": TerminalTIState.FAILED,
+                "end_date": DEFAULT_END_DATE.isoformat(),
+            },
+        )
+
+        assert response.status_code == 204
+
+        session.expire_all()
+        ti = session.get(TaskInstance, ti.id)
+        assert ti.retry_reason is None
+
     def test_ti_update_state_not_running(self, client, session, create_task_instance):
         """Test that a 409 error is returned when attempting to update a TI that is not in RUNNING state."""
         ti = create_task_instance(

@@ -129,9 +129,9 @@ When a task fails, either policy:
    it is the picked category's ``retry`` and ``delay``, unless the policy has a
    confidence bar and the answer is under it, in which case the answer is
    discarded (see `Confidence`_ below).
-4. The decision is logged in the task logs and, on a RETRY, written to the task
-   instance's ``retry_reason``: ``<category>: <reasoning>`` from
-   ``LLMRetryPolicy``, or one line such as
+4. The decision is logged in the task logs and written to the task instance's
+   ``retry_reason``, on a FAIL as well as a RETRY: ``<category>: <reasoning>``
+   from ``LLMRetryPolicy``, or one line such as
    ``category=network confidence=0.91 threshold=0.60 action=retry delay=10s``
    from ``ClassifierRetryPolicy``.
 
@@ -403,8 +403,14 @@ Under ``LLMRetryPolicy`` it answers four fields: ``category``, ``should_retry``,
 ``suggested_delay_seconds`` and ``reasoning``, and the first two after
 ``category`` decide the run. A positive delay is used as returned, with no
 upper limit; zero or negative means no override, so the task's own
-``retry_delay`` and backoff apply. ``category`` and ``reasoning`` become the
-``retry_reason``.
+``retry_delay`` and backoff apply.
+
+``category`` and ``reasoning`` become the ``retry_reason`` (truncated to 500
+characters), recorded on both outcomes. On a RETRY the value is cleared once the next attempt starts running;
+a FAIL is terminal, so there is no next attempt to clear it and the reason stays
+on the row. Only the model's own words are stored -- attempt counts are left to
+whatever displays the reason. Recording on a FAIL requires Airflow 3.4.0; on
+earlier versions only the RETRY outcome is recorded.
 
 Under ``ClassifierRetryPolicy`` it answers the category name and nothing else. It does not
 decide whether to retry, it does not choose the delay, and it does not explain
@@ -414,11 +420,6 @@ the confidence, the bar, the action). A model cannot return a category the
 policy does not recognize, and it cannot return a category paired with an
 action that contradicts it.
 
-The ``retry_reason`` is only recorded on a RETRY. It is written to the task
-instance (truncated to 500 characters), then cleared once the next attempt
-starts running. On a FAIL it is not written anywhere -- it only shows up in the
-task log.
-
 RETRY cannot give a task more attempts than ``retries`` allows. FAIL ends the
 task straight away even when attempts were left, so a wrong classification into
 a failing category costs the task the retries it would otherwise have had;

diff --git a/providers/common/compat/src/airflow/providers/common/compat/_retry_policy.py b/providers/common/compat/src/airflow/providers/common/compat/_retry_policy.py
@@ -74,7 +74,8 @@ class ChainRetryPolicy(RetryPolicy):
 
     The winning decision's reason names the policy that decided, then what every earlier policy
     said: ``HTTPStatusRetryPolicy: HTTP 404 (after ExceptionRetryPolicy: no decision)``. The
-    worker stores it as ``retry_reason`` on a RETRY and logs it otherwise.
+    worker stores it as ``retry_reason`` on a RETRY, and on a FAIL from Airflow 3.4; when no
+    policy decided it is only logged.
 
     :param policies: The policies to consult, in order. At least one.
     """
@@ -123,5 +124,6 @@ def evaluate(
             if trail:
                 reason = f"{reason} (after {'; '.join(trail)})"
             return RetryDecision(action=decision.action, retry_delay=decision.retry_delay, reason=reason)
-        # The worker logs this reason as the policy decision; it is not stored, since nothing is retried by it.
+        # The worker logs this reason as the policy decision; it is not stored, since no policy
+        # took a position.
         return RetryDecision(action=RetryAction.DEFAULT, reason=f"no policy decided ({'; '.join(trail)})")
@@ -271,13 +271,23 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
             raise
         return TIRunContext.model_validate_json(resp.read())
 
-    def finish(self, id: uuid.UUID, state: TerminalStateNonSuccess, when: datetime, rendered_map_index):
+    def finish(
+        self,
+        id: uuid.UUID,
+        state: TerminalStateNonSuccess,
+        when: datetime,
+        rendered_map_index,
+        retry_reason: str | None = None,
+    ):
         """Tell the API server that this TI has reached a terminal state."""
         if state == TaskInstanceState.SUCCESS:
             raise ValueError("Logic error. SUCCESS state should call the `succeed` function instead")
         # TODO: handle the naming better. finish sounds wrong as "even" deferred is essentially finishing.
         body = TITerminalStatePayload(
-            end_date=when, state=TerminalStateNonSuccess(state), rendered_map_index=rendered_map_index
+            end_date=when,
+            state=TerminalStateNonSuccess(state),
+            rendered_map_index=rendered_map_index,
+            retry_reason=retry_reason,
         )
         self.client.patch(f"task-instances/{id}/state", content=body.model_dump_json())
 

@@ -724,6 +724,7 @@ class TITerminalStatePayload(BaseModel):
     state: TerminalStateNonSuccess
     end_date: Annotated[AwareDatetime, Field(title="End Date")]
     rendered_map_index: Annotated[str | None, Field(title="Rendered Map Index")] = None
+    retry_reason: Annotated[str | None, Field(title="Retry Reason")] = None
 
 
 class XComArgBinding(BaseModel):

@@ -56,7 +56,8 @@ class RetryAction(Enum):
 
     The retry is still subject to the task's ``retries`` count -- the policy
     can fail a task earlier but cannot extend past the configured maximum.
-    When all retries are exhausted, RETRY behaves identically to DEFAULT.
+    When all retries are exhausted, RETRY fails the task like DEFAULT does,
+    but still records the policy's reason; DEFAULT records none.
     """
 
     FAIL = "fail"
@@ -378,7 +379,8 @@ class ChainRetryPolicy(RetryPolicy):
 
     The winning decision's reason names the policy that decided, then what every earlier policy
     said: ``HTTPStatusRetryPolicy: HTTP 404 (after ExceptionRetryPolicy: no decision)``. The
-    worker stores it as ``retry_reason`` on a RETRY and logs it otherwise.
+    worker stores it as ``retry_reason`` on a RETRY or a FAIL, and logs it when no policy
+    decided.
 
     :param policies: The policies to consult, in order. At least one.
     """
@@ -427,5 +429,6 @@ def evaluate(
             if trail:
                 reason = f"{reason} (after {'; '.join(trail)})"
             return RetryDecision(action=decision.action, retry_delay=decision.retry_delay, reason=reason)
-        # The worker logs this reason as the policy decision; it is not stored, since nothing is retried by it.
+        # The worker logs this reason as the policy decision; it is not stored, since no policy
+        # took a position.
         return RetryDecision(action=RetryAction.DEFAULT, reason=f"no policy decided ({'; '.join(trail)})")
@@ -869,6 +869,7 @@ class TaskState(BaseModel):
     end_date: datetime | None = None
     type: Literal["TaskState"] = "TaskState"
     rendered_map_index: str | None = None
+    retry_reason: str | None = None
 
 
 class SucceedTask(TISuccessStatePayload):

@@ -4213,6 +4213,18 @@
           ],
           "default": null,
           "title": "Rendered Map Index"
+        },
+        "retry_reason": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Retry Reason"
         }
       },
       "required": [

@@ -39,11 +39,12 @@ def get_bundle() -> VersionBundle:
 
     from airflow.sdk.execution_time.schema.versions.v2026_10_30 import (
         AddArgBindingsToSupervisorTIRunContext,
+        AddRetryReasonToTaskState,
     )
 
     return VersionBundle(
         HeadVersion(),
-        Version("2026-10-30", AddArgBindingsToSupervisorTIRunContext),
+        Version("2026-10-30", AddArgBindingsToSupervisorTIRunContext, AddRetryReasonToTaskState),
         Version("2026-06-16"),
     )
 

@@ -20,6 +20,7 @@
 from cadwyn import VersionChange, schema
 
 from airflow.sdk.api.datamodels._generated import TIRunContext
+from airflow.sdk.execution_time.comms import TaskState
 
 
 class AddArgBindingsToSupervisorTIRunContext(VersionChange):
@@ -34,3 +35,11 @@ class AddArgBindingsToSupervisorTIRunContext(VersionChange):
     description = __doc__
 
     instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("arg_bindings").didnt_exist,)
+
+
+class AddRetryReasonToTaskState(VersionChange):
+    """Add `retry_reason` to `TaskState`."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(TaskState).field("retry_reason").didnt_exist,)
@@ -1760,6 +1760,7 @@ def _send_terminal_state_msg(
                 state=msg.state,
                 when=msg.end_date or datetime.now(tz=timezone.utc),
                 rendered_map_index=self._rendered_map_index,
+                retry_reason=msg.retry_reason,
             )
         elif isinstance(msg, SucceedTask):
             self.client.task_instances.succeed(

@@ -1902,18 +1902,20 @@ def _handle_current_task_failed(
                 state=TaskInstanceState.FAILED,
                 end_date=ti.end_date,
                 rendered_map_index=ti.rendered_map_index,
+                retry_reason=decision.reason[:500] if decision.reason is not None else None,
             ),
             TaskInstanceState.FAILED,
         )
     if decision is not None and decision.action == RetryAction.RETRY:
         return _finalize_task_failure(
-            ti, retry_delay_override=decision.retry_delay, retry_reason=decision.reason
+            ti, log, retry_delay_override=decision.retry_delay, retry_reason=decision.reason
         )
-    return _finalize_task_failure(ti)
+    return _finalize_task_failure(ti, log)
 
 
 def _finalize_task_failure(
     ti: RuntimeTaskInstance,
+    log: Logger,
     retry_delay_override: timedelta | None = None,
     retry_reason: str | None = None,
 ) -> tuple[RetryTask, TaskInstanceState] | tuple[TaskState, TaskInstanceState]:
@@ -1946,9 +1948,22 @@ def _finalize_task_failure(
         if retry_reason is not None:
             retry_kwargs["retry_reason"] = retry_reason[:500]
         return RetryTask(**retry_kwargs), TaskInstanceState.UP_FOR_RETRY
+    if retry_reason is not None:
+        # Policy's own words only: attempt counts belong to whoever renders this, which has
+        # try_number and max_tries alongside and need not guess when retries was never set.
+        retry_reason = retry_reason[:500]
+        log.info(
+            "Retry policy requested a retry but no attempts remain",
+            reason=retry_reason,
+            try_number=ti.try_number,
+            max_tries=ti._ti_context_from_server.max_tries if ti._ti_context_from_server else None,
+        )
     return (
         TaskState(
-            state=TaskInstanceState.FAILED, end_date=end_date, rendered_map_index=ti.rendered_map_index
+            state=TaskInstanceState.FAILED,
+            end_date=end_date,
+            rendered_map_index=ti.rendered_map_index,
+            retry_reason=retry_reason,
         ),
         TaskInstanceState.FAILED,
     )