Persist retry_reason not just for retries but even when a task fails

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Motivation

Today, when a task goes into "FAILED" it alone tells you nothing about why it stopped and that's something a retry policy in the first place should surface best. A few examples:

1. Auth error, policy said don't bother retrying.
   Your task hits an API and gets a 403 because the key expired. The policy is smart enough to know retrying won't help (it'll just fail 3 more times the same way), so it fails immediately after try 1. Without a reason, you just see "FAILED after 1 try" and think something's broken with retries. With a reason, you see: "auth error, policy chose not to retry" now you know exactly what to go fix (the API key), not the code.

2. Rate limit, retries all used up.
   Your task keeps hitting a rate limit. The policy says "retry" each time, so it retries 3 times, but the budget (retries=3) runs out. Without a reason, you just see 3 red X's and no explanation. With a reason: "rate limit" and you immediately know it's not a bug; it's just that the external service was too slow to respond, and maybe you should bump retries or add a delay. The attempt counts are already on the same row as try_number and max_tries, so whatever displays the reason can say "3 of 3" itself.

3. Different failures on different tries.
   Try 1 failed for one reason, try 2 for a totally different reason. Right now all you see is a row of identical red icons with no way to tell if its the same recurring problem or three unrelated ones. Reasons attached to each try let someone debugging a flaky task actually see the story, instead of guessing.

The underlying idea: today, all that classification work the policy does (LLM or exception-based) happens, gets logged once in task logs. Only the "it retried" case kept the reason. This change makes sure the reason survives for the FAILED case too, so a future screen can show a plain sentence like "Stopped at try 2: auth error, no retry" instead of just a bare failure with no story behind it.

What

If you've set a retry_policy on a task, it can decide things like "this looks like an auth error, dont bother retrying" or "retries are exhausted, this was a rate limit," but today that explanation is only ever logged to a log line and thrown away otherwise. It never reaches the database, so no API response or UI screen(I am proposing we build it to provide a better UX to users and its more "in the face") can ever show it, no matter how much we build on top later.

This PR is the first, necessary step toward fixing that: make sure the reason actually gets saved whenever a task fails, not just when it retries. Once its reliably in the database, a future PR can expose it through the API and the UI, so a Dag author looking at a failed task instance can see a plain reason instead of just "FAILED" with no explanation.

Current behaviour

retry_reason is already written to the database, but only when a retry policy chooses to retry. The two failure outcomes people most want explained never save anything: a policy deciding FAIL outright, and a policy deciding RETRY but the retry budget being exhausted. Both currently build a bare TaskState(state=FAILED) with no reason, so the classification text is logged and discarded.

• Policy says FAIL: TaskState(state=FAILED), reason dropped.
• Policy says RETRY but the budget is exhausted: same, reason dropped.
• Policy says RETRY with budget remaining: RetryTask(retry_reason=...), already persisted (unchanged).
• No retry policy at all: unaffected either way.

Proposed change

Thread retry_reason through both FAILED paths, end to end:

• _handle_current_task_failed's FAIL branch and _finalize_task_failure's exhausted budget branch now attach a reason to the TaskState they build. Only the policy's own words are stored; attempt counts are left to whatever displays the reason, which has try_number and max_tries alongside it.
• When the budget is what stopped a policy-chosen RETRY, that is logged as its own event (Retry policy requested a retry but no attempts remain, with try_number and max_tries) rather than a second Retry policy decision line contradicting the first.
• TaskState (task-sdk message) gains a retry_reason field.
• FAILED does not go through the same "send immediately" path as RETRY; it's deferred until the subprocess exits (the safety net that also covers a hard crash with no message at all). So the supervisor now captures retry_reason off the TaskState message and forwards it to the deferred finish() call.
• TITerminalStatePayload (execution API request schema) gains retry_reason, gated behind a new Cadwyn version change added to the existing unreleased 2026-10-30 version (tentative date)
• The FAILED branch of the state update route now truncates to 500 chars (matching the existing RETRY branch) and persists the reason to task_instance.retry_reason.
• Docs updated where they said the reason is only recorded on a RETRY: providers/common/ai/docs/retry_policies.rst and airflow-core/docs/core-concepts/tasks.rst.

Only tasks with a configured retry policy are affected; a plain retries=N task with no policy gets no reason and no behaviour change, since there's nothing meaningful to attach.

Testing

After running the example_llm_retry_policy dag, earlier if I ran:

SELECT task_instance.state, task_instance.retry_reason from task_instance;

I would get:

failed,
failed,
up_for_retry,"rate_limit: The error message explicitly indicates a 429 HTTP status code (""Too Many Requests"") with a rate limit exceeded message. The error also provides guidance to retry after 60 seconds. Despite this being attempt 3 of 3, rate limit errors are transient API throttling issues that should be retried with appropriate backoff."

Now we see:

up_for_retry,category=resource confidence=n/a threshold=n/a action=retry delay=300s
up_for_retry,"rate_limit: The error explicitly indicates a 429 HTTP status code with ""Too Many Requests"" message, which is a clear signal of API rate limiting. The error message also helpfully specifies a 60-second retry delay, which is appropriate for rate limit scenarios."
failed,"auth: The error indicates an authentication/authorization failure with a 403 Forbidden status code and explicitly states ""API key expired for service account"". This is a credential issue that requires manual intervention to renew or rotate the API key. Retrying without fixing the underlying credential problem will not resolve the issue."
failed,"data: This is a schema validation error indicating a type mismatch in the input data. The column 'user_id' was expected to be INT but received a STRING value in row 42. This is a data quality issue with the input, not a transient problem. Retrying will not fix this - the underlying data needs to be corrected or the schema needs to be adjusted. This requires intervention to either clean the data or modify the validation logic."

The two failed rows are the change: before this PR both were blank.

What's next

retry_reason still isn't exposed anywhere outside the database. Adding it to TaskInstanceResponse (API) and rendering it in the Task Instance UI panel are the next two steps, so this data can actually reach a Dag author or surface it on the UI.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[amoghrajesh]

@kaxil can I get a round of review on this?

[kaxil]

Round 3, verified at 524f00d against apache/main. Nothing here blocks. The code paths (FAIL branch, exhausted-budget branch, supervisor capture, route write, both version gates) do what the description says, and the round 2 fixes landed as described. The Tests workflow has not run on the latest push (only Mergeable and WIP show on 524f00d), so I ran the affected task-sdk and core modules locally: 153 task-sdk and 57 core tests pass. Four things:

providers/common/ai/docs/retry_policies.rst now says the value "is exposed by the REST API as state_reason ... and the Task Instance page in the UI shows it under Reason for state". Neither exists on main or on this branch; both come from #73030, and #73030's own copy of this page does not carry the paragraph (it still has the ; retries exhausted (N of M) sentence). If this merges first, the published page points at a field that does not exist. Moving that paragraph to #73030 keeps each PR's docs matching its own code.

The same file is the one conflict git merge-tree reports, because main reorganized the page in #73523 and #73526. After resolving, the sentences to change are now at :132 ("on a RETRY, written to the task instance's retry_reason") and :420-423 ("The retry_reason is only recorded on a RETRY ... On a FAIL it is not written anywhere"). One more outside this diff: airflow-core/docs/core-concepts/tasks.rst:299-300 from #73508 (merged today) says "on FAIL, or when no policy decided, it appears in the task log as the Retry policy decision line", which this PR makes false for the FAIL half.

task_runner.py:1942 logs a second Retry policy decision line with the opposite action. _evaluate_retry_policy has already logged Retry policy decision action=retry reason=<same text> a moment earlier, so the exhausted-budget path now emits the same event name twice for one decision, once as retry and once as fail. What actually happened is "policy asked to retry, no attempts remain", and saying that (with try_number and max_tries) is what an on-call reader needs. tasks.rst:300 on main points them at "the Retry policy decision line", so two of them disagreeing is what they would hit.

test_migrator.py:57 imports TaskInstanceState from airflow.utils.state. It is the only task-sdk test importing from core's utils; the other six use from airflow.sdk import TaskInstanceState, and task-sdk/src never imports airflow.utils.state.

Two housekeeping notes: the PR description still describes the ; retries exhausted (3 of 3) suffix and shows it in the Testing sample, which round 3 removed; and #73030 still carries both that suffix and the duplicate log line, so it will want the same two changes when it rebases onto this.

[amoghrajesh]

@kaxil thanks.

Docs: That paragraph described #73030's work, not this PR's so I removed it, it'll go there instead. Merge conflict resolved by taking main's text and keeping only what this PR changes: the reason is recorded on a FAIL too, and stays on the row since a FAIL is terminal. Also fixed the two other spots you found: step 4 of "How it works" and tasks.rst:300.

The log line. Agreed, two decisions disagreeing is worse than none. Now a separate event with the numbers an reader needs:

Retry policy requested a retry but no attempts remain
  reason=... try_number=3 max_tries=2

Ironic, given we just took counts out of the stored reason, but this is where they belong. Test asserts Retry policy decision appears once; checked it fails if the old line returns.

The import: Now from airflow.sdk import TaskInstanceState.

I'd like to land this first, then rebase #73030 on top and handle comments there.

The PR desc has been edited too.

[kaxil]

Round 4, verified at 21033cb. The round 3 items landed: the state_reason paragraph is gone, step 4 of "How it works" and tasks.rst:299-300 now match the code, the exhausted-budget path logs its own Retry policy requested a retry but no attempts remain event with try_number and max_tries, and the migrator test imports from airflow.sdk. Task SDK and core tests pass on this push; the three red checks are not from this change (constraints and compat 2.11.1 hit a PyPI 503 during the distribution build, and the docs build failed in the ibm-mq provider's docs). Nothing here blocks. Two stale sentences are worth fixing before merge, plus two notes for the rebase and #73030:

providers/common/ai/docs/retry_policies.rst:423-426 says "The retry_reason is only recorded on a RETRY ... On a FAIL it is not written anywhere" again. You removed it in d9d3a00, and the latest merge from main brought it back, since main's ClassifierRetryPolicy rewrite of the page carries the same paragraph. It now contradicts the paragraph this PR adds at :408-413 on the same page. Dropping it, or folding the "truncated to 500 characters" detail into :408, fixes it.

The same claim is in the ChainRetryPolicy docstring at task-sdk/src/airflow/sdk/definitions/retry_policy.py:381-382: "The worker stores it as retry_reason on a RETRY and logs it otherwise." That is the rendered API reference, and after this PR the worker stores it on a FAIL too. The comment at :431 ("it is not stored, since nothing is retried by it") now gives the wrong reason for the same thing; "since no policy took a position" would be accurate.

For the rebase: git merge-tree against current main reports a conflict in supervisor.py, because #73290 moved TaskState handling into _handle_task_state. The self._retry_reason = msg.retry_reason capture needs to move into that handler, and test_task_state_retry_reason_forwarded_to_finish will fail if it gets dropped in the resolution, so that part is covered.

For #73030: a FAIL reason now outlives the FAIL. Only ti_run clears retry_reason, so mark-success leaves a SUCCESS row carrying it, and a cleared TI keeps it until the next run starts. Nothing reads the column yet, so this belongs with the UI work there (render it only for FAILED or UP_FOR_RETRY, or clear it in clear_task_instances).