From b93cc1844ece8f0ce7bc5e57a8117931045f2810 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Sat, 12 Sep 2026 15:08:25 +0530 Subject: [PATCH 1/9] Persist retry_reason not just for retries, but even when a task fails --- .../execution_api/datamodels/taskinstance.py | 1 + .../execution_api/routes/task_instances.py | 6 +- .../execution_api/versions/__init__.py | 8 ++- .../execution_api/versions/v2026_10_30.py | 12 +++- .../versions/head/test_task_instances.py | 66 +++++++++++++++++++ task-sdk/src/airflow/sdk/api/client.py | 14 +++- .../airflow/sdk/api/datamodels/_generated.py | 1 + .../src/airflow/sdk/execution_time/comms.py | 1 + .../sdk/execution_time/schema/schema.json | 12 ++++ .../airflow/sdk/execution_time/supervisor.py | 3 + .../airflow/sdk/execution_time/task_runner.py | 9 ++- task-sdk/tests/task_sdk/api/test_client.py | 7 +- .../execution_time/test_supervisor.py | 22 +++++++ .../execution_time/test_task_runner.py | 59 +++++++++++++++++ 14 files changed, 214 insertions(+), 7 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 ddf31db971838..ac1cb781349d6 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 @@ -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): 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 a49d8fe30e546..d4b773b028365 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 @@ -670,7 +670,11 @@ def _create_ti_state_update_query_and_update_state( query = query.values(state=updated_state, next_method=None, next_kwargs=None) 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: + failed_retry_reason: str | None = ti_patch_payload.retry_reason[:500] + query = query.values(retry_reason=failed_retry_reason) + if ti is not None: + ti.retry_reason = failed_retry_reason 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): 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 70ea9be2be65c..22e762e39ae95 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 @@ -55,11 +55,17 @@ from airflow.api_fastapi.execution_api.versions.v2026_10_30 import ( AddArgBindingsToTIRunContext, AddCallbackRunEndpoint, + AddTerminalStateRetryReasonField, ) bundle = VersionBundle( HeadVersion(), - Version("2026-10-30", AddArgBindingsToTIRunContext, AddCallbackRunEndpoint), + Version( + "2026-10-30", + AddArgBindingsToTIRunContext, + AddCallbackRunEndpoint, + AddTerminalStateRetryReasonField, + ), Version( "2026-06-30", AddVariableKeysEndpoint, 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 0620053b3e2eb..ad4b8f55b31ca 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 @@ -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): @@ -52,3 +52,13 @@ class AddCallbackRunEndpoint(VersionChange): instructions_to_migrate_to_previous_version = ( endpoint("/callbacks/{callback_id}/run", ["PATCH"]).didnt_exist, ) + + +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, + ) 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 ccaecce822d6e..1acd717cc819f 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 @@ -2446,6 +2446,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( diff --git a/task-sdk/src/airflow/sdk/api/client.py b/task-sdk/src/airflow/sdk/api/client.py index ea220c68fde71..11c170e55021b 100644 --- a/task-sdk/src/airflow/sdk/api/client.py +++ b/task-sdk/src/airflow/sdk/api/client.py @@ -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()) diff --git a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py index 3635bad9628a0..a5f9df3d95dec 100644 --- a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py +++ b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py @@ -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): diff --git a/task-sdk/src/airflow/sdk/execution_time/comms.py b/task-sdk/src/airflow/sdk/execution_time/comms.py index 4a26f56e297dc..ee7aa11f331f0 100644 --- a/task-sdk/src/airflow/sdk/execution_time/comms.py +++ b/task-sdk/src/airflow/sdk/execution_time/comms.py @@ -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): 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 85f8b7cfa4e2b..23640ce3e8921 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json +++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json @@ -4203,6 +4203,18 @@ ], "default": null, "title": "Rendered Map Index" + }, + "retry_reason": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Retry Reason" } }, "required": [ diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py index 0a4808512856f..9c5feac20105b 100644 --- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py +++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py @@ -1407,6 +1407,7 @@ class ActivitySubprocess(WatchedSubprocess): _terminal_state: str | None = attrs.field(default=None, init=False) _final_state: str | None = attrs.field(default=None, init=False) + _retry_reason: str | None = attrs.field(default=None, init=False) # The terminal-state message currently being processed by `_handle_request`, # captured BEFORE the dedicated API call (succeed / retry / defer / # reschedule). If the API call raises (network blip, server 5xx, etc.), @@ -1567,6 +1568,7 @@ def update_task_state_if_needed(self): state=self.final_state, when=datetime.now(tz=timezone.utc), rendered_map_index=self._rendered_map_index, + retry_reason=self._retry_reason, ) def _send_terminal_state_msg( @@ -1803,6 +1805,7 @@ def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger, req_id: self._terminal_state = msg.state self._task_end_time_monotonic = time.monotonic() self._rendered_map_index = msg.rendered_map_index + self._retry_reason = msg.retry_reason elif isinstance(msg, SucceedTask): self._task_end_time_monotonic = time.monotonic() self._rendered_map_index = msg.rendered_map_index 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 eeb2788062248..da9e2a0813e7c 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -1887,6 +1887,7 @@ def _handle_current_task_failed( state=TaskInstanceState.FAILED, end_date=ti.end_date, rendered_map_index=ti.rendered_map_index, + retry_reason=decision.reason, ), TaskInstanceState.FAILED, ) @@ -1931,9 +1932,15 @@ 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 and ti._ti_context_from_server is not None: + max_tries = ti._ti_context_from_server.max_tries + retry_reason = f"{retry_reason}; retries exhausted ({ti.try_number} of {max_tries})" 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, ) diff --git a/task-sdk/tests/task_sdk/api/test_client.py b/task-sdk/tests/task_sdk/api/test_client.py index 684d4e7dcbb17..e31829797e4a6 100644 --- a/task-sdk/tests/task_sdk/api/test_client.py +++ b/task-sdk/tests/task_sdk/api/test_client.py @@ -432,6 +432,7 @@ def handle_request(request: httpx.Request) -> httpx.Response: assert actual_body["end_date"] == "2024-10-31T12:00:00Z" assert actual_body["state"] == state assert actual_body["rendered_map_index"] == "test" + assert actual_body["retry_reason"] == "auth error, do not retry" return httpx.Response( status_code=204, ) @@ -439,7 +440,11 @@ def handle_request(request: httpx.Request) -> httpx.Response: client = make_client(transport=httpx.MockTransport(handle_request)) client.task_instances.finish( - ti_id, state=state, when="2024-10-31T12:00:00Z", rendered_map_index="test" + ti_id, + state=state, + when="2024-10-31T12:00:00Z", + rendered_map_index="test", + retry_reason="auth error, do not retry", ) def test_task_instance_heartbeat(self): 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 93422f4426a00..8ae01f8ddba22 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py +++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py @@ -3705,6 +3705,28 @@ def test_update_task_state_no_recovery_without_pending_msg(self, watched_subproc watched_subprocess.client.task_instances.finish.assert_not_called() watched_subprocess.client.task_instances.succeed.assert_not_called() + def test_task_state_retry_reason_forwarded_to_finish(self, watched_subprocess, mocker): + """A TaskState message's retry_reason must reach the deferred finish() call.""" + watched_subprocess, _ = watched_subprocess + watched_subprocess._exit_code = 0 + + msg = TaskState( + state=TaskInstanceState.FAILED, + end_date=timezone.parse("2024-10-31T12:00:00Z"), + retry_reason="auth error, do not retry", + ) + watched_subprocess._handle_request(msg, mocker.Mock(), req_id=1) + + watched_subprocess.update_task_state_if_needed() + + watched_subprocess.client.task_instances.finish.assert_called_once_with( + id=watched_subprocess.id, + state=TaskInstanceState.FAILED, + when=mocker.ANY, + rendered_map_index=None, + retry_reason="auth error, do not retry", + ) + class TestSetSupervisorComms: class DummyComms: 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 cc9fb77e08921..5feac21c28f69 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 @@ -1196,6 +1196,65 @@ def execute(self, context): assert counted.count("operator_failures") == 1 +def test_retry_policy_fail_persists_reason(create_runtime_ti, mock_supervisor_comms): + class _AlwaysFails(BaseOperator): + def execute(self, context): + raise RuntimeError("boom") + + task = _AlwaysFails( + task_id="fail_with_reason", + retry_policy=ExceptionRetryPolicy( + rules=[RetryRule(exception=RuntimeError, action=RetryAction.FAIL, reason="do not retry")] + ), + ) + ti = create_runtime_ti(task=task, should_retry=True) + + state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) + + assert state == TaskInstanceState.FAILED + assert isinstance(msg, TaskState) + assert msg.retry_reason == "do not retry" + + +def test_retry_policy_retry_exhausted_persists_combined_reason(create_runtime_ti, mock_supervisor_comms): + """A policy-chosen RETRY that hits an exhausted budget still fails, with both reasons recorded.""" + + class _AlwaysFails(BaseOperator): + def execute(self, context): + raise RuntimeError("boom") + + task = _AlwaysFails( + task_id="retry_exhausted", + retry_policy=ExceptionRetryPolicy( + rules=[RetryRule(exception=RuntimeError, action=RetryAction.RETRY, reason="rate limit")] + ), + ) + ti = create_runtime_ti(task=task, try_number=2, max_tries=2, should_retry=False) + + state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) + + assert state == TaskInstanceState.FAILED + assert isinstance(msg, TaskState) + assert msg.retry_reason == "rate limit; retries exhausted (2 of 2)" + + +def test_plain_retries_exhausted_has_no_reason(create_runtime_ti, mock_supervisor_comms): + """Without a retry policy, exhausting the retry budget must not synthesize a reason.""" + + class _AlwaysFails(BaseOperator): + def execute(self, context): + raise RuntimeError("boom") + + task = _AlwaysFails(task_id="plain_exhausted") + ti = create_runtime_ti(task=task, try_number=2, max_tries=2, should_retry=False) + + state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) + + assert state == TaskInstanceState.FAILED + assert isinstance(msg, TaskState) + assert msg.retry_reason is None + + def test_run_downstream_skipped(mocked_parse, create_runtime_ti, mock_supervisor_comms, listener_manager): listener = TestTaskRunnerCallsListeners.CustomListener() listener_manager(listener) From 13f23e5fa11077db051ceb5c45c0e96d05f42185 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Sat, 12 Sep 2026 16:27:09 +0530 Subject: [PATCH 2/9] Surface the retry policy decision on task instances page --- .../datamodels/task_instance_history.py | 1 + .../core_api/datamodels/task_instances.py | 1 + .../core_api/openapi/_private_ui.yaml | 5 ++ .../openapi/v2-rest-api-generated.yaml | 10 ++++ .../ui/openapi-gen/requests/schemas.gen.ts | 22 +++++++++ .../ui/openapi-gen/requests/types.gen.ts | 2 + .../ui/public/i18n/locales/en/common.json | 1 + .../ui/src/pages/TaskInstance/Details.tsx | 18 ++++++- .../core_api/routes/public/test_hitl.py | 1 + .../routes/public/test_task_instances.py | 47 +++++++++++++++++++ .../airflowctl/api/datamodels/generated.py | 2 + 11 files changed, 109 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py index c913d91c1e607..7662b68bbd210 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py @@ -62,6 +62,7 @@ class TaskInstanceHistoryResponse(BaseModel): executor: str | None executor_config: Annotated[str, BeforeValidator(str)] dag_version: DagVersionResponse | None + retry_reason: str | None = None class TaskInstanceHistoryCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py index cd83cd192aa2a..3c893748440e8 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py @@ -89,6 +89,7 @@ class TaskInstanceResponse(BaseModel): queued_by_job: JobResponse | None = Field(alias="triggerer_job") dag_version: DagVersionResponse | None team_name: str | None = None + retry_reason: str | None = None class TaskInstanceCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index a51f805f09daa..e9284a9016a54 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml @@ -4796,6 +4796,11 @@ components: - type: string - type: 'null' title: Team Name + retry_reason: + anyOf: + - type: string + - type: 'null' + title: Retry Reason type: object required: - id diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index bea12275f6960..e39a3cbd5270c 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -15835,6 +15835,11 @@ components: anyOf: - $ref: '#/components/schemas/DagVersionResponse' - type: 'null' + retry_reason: + anyOf: + - type: string + - type: 'null' + title: Retry Reason type: object required: - task_id @@ -16018,6 +16023,11 @@ components: - type: string - type: 'null' title: Team Name + retry_reason: + anyOf: + - type: string + - type: 'null' + title: Retry Reason type: object required: - id diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 3198d4162e759..a38aa8a368ace 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -7189,6 +7189,17 @@ export const $TaskInstanceHistoryResponse = { type: 'null' } ] + }, + retry_reason: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Retry Reason' } }, type: 'object', @@ -7489,6 +7500,17 @@ export const $TaskInstanceResponse = { } ], title: 'Team Name' + }, + retry_reason: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Retry Reason' } }, type: 'object', diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index a4c17a6723d92..35cedc7999079 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1861,6 +1861,7 @@ export type TaskInstanceHistoryResponse = { executor: string | null; executor_config: string; dag_version: DagVersionResponse | null; + retry_reason?: string | null; }; /** @@ -1904,6 +1905,7 @@ export type TaskInstanceResponse = { triggerer_job: JobResponse | null; dag_version: DagVersionResponse | null; team_name?: string | null; + retry_reason?: string | null; }; /** diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json index 99e181ff9584c..9d673dcb51695 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json @@ -440,6 +440,7 @@ "queue": "Queue", "queuedWhen": "Queued At", "renderedMapIndex": "Rendered Map Index", + "retryReason": "Reason for state", "scheduledWhen": "Scheduled At", "trigger": "Trigger", "triggerer": { diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx index 017e24c54c7ae..46d2e0ba4c193 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx @@ -27,7 +27,7 @@ import { useTaskInstanceServiceGetTaskInstanceTryDetails, } from "openapi/queries"; -import { ClipboardRoot, ClipboardIconButton } from "src/system-components"; +import { Alert, ClipboardRoot, ClipboardIconButton } from "src/system-components"; import { DagVersionDetails } from "src/components/DagVersionDetails"; import RenderedJsonField from "src/components/RenderedJsonField"; @@ -131,6 +131,16 @@ export const Details = () => { return ( + {taskInstance?.retry_reason === null || taskInstance?.retry_reason === undefined ? undefined : ( + + {taskInstance.retry_reason} + + )} {taskInstance === undefined || tryNumber === undefined || taskInstance.try_number <= 1 ? (
) : ( @@ -162,6 +172,12 @@ export const Details = () => { + {tryInstance?.retry_reason === null || tryInstance?.retry_reason === undefined ? undefined : ( + + {translate("taskInstance.retryReason")} + {tryInstance.retry_reason} + + )} {translate("taskId")} diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py index ad7b069581ee3..8b746f3e93322 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py @@ -267,6 +267,7 @@ def expected_sample_hitl_detail_dict(sample_ti: TaskInstance) -> dict[str, Any]: "task_display_name": "sample_task_hitl", "task_id": TASK_ID, "team_name": None, + "retry_reason": None, "trigger": None, "triggerer_job": None, "try_number": 0, diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index 700ad0c53c70f..0952c6b13684f 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -244,8 +244,17 @@ def test_should_respond_200(self, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } + def test_should_include_retry_reason(self, test_client, session): + self.create_task_instances(session, task_instances=[{"retry_reason": "auth error, do not retry"}]) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context" + ) + assert response.status_code == 200 + assert response.json()["retry_reason"] == "auth error, do not retry" + @conf_vars({("core", "multi_team"): "True"}) def test_should_include_team_name(self, test_client, session): self.create_task_instances(session) @@ -329,6 +338,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, "dag_version": { "id": response_data["dag_version"]["id"], "version_number": expected_version_number, @@ -425,6 +435,7 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "unixname": getuser(), }, "team_name": None, + "retry_reason": None, } def test_should_respond_200_with_task_state_in_removed(self, test_client, session): @@ -479,6 +490,7 @@ def test_should_respond_200_with_task_state_in_removed(self, test_client, sessio "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } def test_should_respond_200_task_instance_with_rendered(self, test_client, session): @@ -536,6 +548,7 @@ def test_should_respond_200_task_instance_with_rendered(self, test_client, sessi "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } def test_raises_404_for_nonexistent_task_instance(self, test_client): @@ -657,6 +670,7 @@ def test_should_respond_200_mapped_task_instance_with_rtif(self, test_client, se "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } def test_should_respond_401(self, unauthenticated_test_client): @@ -2812,8 +2826,21 @@ def test_should_respond_200(self, test_client, session): "id": response_data["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, } + def test_should_include_retry_reason_from_history(self, test_client, session): + self.create_task_instances( + session, + task_instances=[{"state": State.SUCCESS, "retry_reason": "auth error, do not retry"}], + with_ti_history=True, + ) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries/1" + ) + assert response.status_code == 200 + assert response.json()["retry_reason"] == "auth error, do not retry" + @pytest.mark.parametrize("try_number", [1, 2]) def test_should_respond_200_with_different_try_numbers(self, test_client, try_number, session): self.create_task_instances(session, task_instances=[{"state": State.SUCCESS}], with_ti_history=True) @@ -2858,6 +2885,7 @@ def test_should_respond_200_with_different_try_numbers(self, test_client, try_nu "id": response_data["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, } @pytest.mark.parametrize("try_number", [1, 2]) @@ -2935,6 +2963,7 @@ def test_should_respond_200_with_mapped_task_at_different_try_numbers( "id": response_data["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, } def test_should_respond_200_with_task_state_in_deferred(self, test_client, session): @@ -3007,6 +3036,7 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "id": response_data["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, } def test_should_respond_200_with_task_state_in_removed(self, test_client, session): @@ -3054,6 +3084,7 @@ def test_should_respond_200_with_task_state_in_removed(self, test_client, sessio "id": response_data["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, } def test_should_respond_401(self, unauthenticated_test_client): @@ -3129,6 +3160,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "created_at": mock.ANY, "dag_display_name": "dag_with_multiple_versions", }, + "retry_reason": None, } def test_should_not_return_duplicate_runs(self, test_client, session): @@ -3866,6 +3898,7 @@ def test_should_respond_200_with_dag_run_id( "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, "try_number": 0, "unixname": getuser(), }, @@ -4329,6 +4362,7 @@ def test_should_respond_200(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, }, { "dag_id": "example_python_operator", @@ -4366,6 +4400,7 @@ def test_should_respond_200(self, test_client, session): "id": response_data["task_instances"][1]["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, }, ], "total_entries": 2, @@ -4437,6 +4472,7 @@ def test_ti_in_retry_state_not_returned(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, }, ], "total_entries": 1, @@ -4520,6 +4556,7 @@ def test_mapped_task_should_respond_200(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, }, { "dag_id": "example_python_operator", @@ -4557,6 +4594,7 @@ def test_mapped_task_should_respond_200(self, test_client, session): "id": response_data["task_instances"][1]["dag_version"]["id"], "version_number": 1, }, + "retry_reason": None, }, ], "total_entries": 2, @@ -4624,6 +4662,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "created_at": mock.ANY, "dag_display_name": "dag_with_multiple_versions", }, + "retry_reason": None, } @@ -4749,6 +4788,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, @@ -5027,6 +5067,7 @@ def test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, @@ -5167,6 +5208,7 @@ def test_update_mask_set_note_should_respond_200( "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, @@ -5232,6 +5274,7 @@ def test_set_note_should_respond_200(self, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, @@ -5329,6 +5372,7 @@ def test_set_note_should_respond_200_mapped_task_with_rtif(self, test_client, se "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, @@ -5414,6 +5458,7 @@ def test_set_note_should_respond_200_mapped_task_summary_with_rtif(self, test_cl "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } _check_task_instance_note( @@ -5610,6 +5655,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, @@ -5900,6 +5946,7 @@ def test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte "trigger": None, "triggerer_job": None, "team_name": None, + "retry_reason": None, } ], "total_entries": 1, diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index b7b97e5cd849e..8a09649397452 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -2158,6 +2158,7 @@ class TaskInstanceHistoryResponse(BaseModel): executor: Annotated[str | None, Field(title="Executor")] executor_config: Annotated[str, Field(title="Executor Config")] dag_version: DagVersionResponse | None + retry_reason: Annotated[str | None, Field(title="Retry Reason")] = None class TaskInstanceResponse(BaseModel): @@ -2200,6 +2201,7 @@ class TaskInstanceResponse(BaseModel): triggerer_job: JobResponse | None dag_version: DagVersionResponse | None team_name: Annotated[str | None, Field(title="Team Name")] = None + retry_reason: Annotated[str | None, Field(title="Retry Reason")] = None class TaskResponse(BaseModel): From a6f747cd5c9a5723f264951d076ec6d26630f394 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Tue, 15 Sep 2026 17:04:51 +0530 Subject: [PATCH 3/9] comments from kaxil --- .../api_fastapi/execution_api/routes/task_instances.py | 2 -- .../sdk/execution_time/schema/versions/__init__.py | 3 ++- .../sdk/execution_time/schema/versions/v2026_10_30.py | 9 +++++++++ task-sdk/src/airflow/sdk/execution_time/task_runner.py | 6 ++++-- task-sdk/tests/task_sdk/api/test_client.py | 7 ++++--- .../tests/task_sdk/execution_time/test_task_runner.py | 6 +++--- 6 files changed, 22 insertions(+), 11 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 d4b773b028365..4464e6fc35480 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 @@ -673,8 +673,6 @@ def _create_ti_state_update_query_and_update_state( if isinstance(ti_patch_payload, TITerminalStatePayload) and ti_patch_payload.retry_reason: failed_retry_reason: str | None = ti_patch_payload.retry_reason[:500] query = query.values(retry_reason=failed_retry_reason) - if ti is not None: - ti.retry_reason = failed_retry_reason 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): 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 7e5ce93f86bdc..06be64d346170 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 @@ -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"), ) 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 e6b93f5dea805..e6f6e920d4520 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 @@ -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,) 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 da9e2a0813e7c..74c9d30fc8b83 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -1933,8 +1933,10 @@ def _finalize_task_failure( retry_kwargs["retry_reason"] = retry_reason[:500] return RetryTask(**retry_kwargs), TaskInstanceState.UP_FOR_RETRY if retry_reason is not None and ti._ti_context_from_server is not None: - max_tries = ti._ti_context_from_server.max_tries - retry_reason = f"{retry_reason}; retries exhausted ({ti.try_number} of {max_tries})" + # max_tries is the retry count, not the attempt count -- total attempts is max_tries + 1. + total_attempts = ti._ti_context_from_server.max_tries + 1 + suffix = f"; retries exhausted ({ti.try_number} of {total_attempts})" + retry_reason = f"{retry_reason[: 500 - len(suffix)]}{suffix}" return ( TaskState( state=TaskInstanceState.FAILED, diff --git a/task-sdk/tests/task_sdk/api/test_client.py b/task-sdk/tests/task_sdk/api/test_client.py index e31829797e4a6..d3c47cebbf685 100644 --- a/task-sdk/tests/task_sdk/api/test_client.py +++ b/task-sdk/tests/task_sdk/api/test_client.py @@ -422,7 +422,8 @@ def handle_request(request: httpx.Request) -> httpx.Response: @pytest.mark.parametrize( "state", [state for state in TerminalTIState if state != TerminalTIState.SUCCESS] ) - def test_task_instance_finish(self, state): + @pytest.mark.parametrize("retry_reason", [None, "auth error, do not retry"]) + def test_task_instance_finish(self, state, retry_reason): # Simulate a successful response from the server that finishes (moved to terminal state) a task ti_id = uuid6.uuid7() @@ -432,7 +433,7 @@ def handle_request(request: httpx.Request) -> httpx.Response: assert actual_body["end_date"] == "2024-10-31T12:00:00Z" assert actual_body["state"] == state assert actual_body["rendered_map_index"] == "test" - assert actual_body["retry_reason"] == "auth error, do not retry" + assert actual_body["retry_reason"] == retry_reason return httpx.Response( status_code=204, ) @@ -444,7 +445,7 @@ def handle_request(request: httpx.Request) -> httpx.Response: state=state, when="2024-10-31T12:00:00Z", rendered_map_index="test", - retry_reason="auth error, do not retry", + retry_reason=retry_reason, ) def test_task_instance_heartbeat(self): 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 5feac21c28f69..4c4f50b46d243 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 @@ -1229,13 +1229,13 @@ def execute(self, context): rules=[RetryRule(exception=RuntimeError, action=RetryAction.RETRY, reason="rate limit")] ), ) - ti = create_runtime_ti(task=task, try_number=2, max_tries=2, should_retry=False) + ti = create_runtime_ti(task=task, try_number=3, max_tries=2, should_retry=False) state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) assert state == TaskInstanceState.FAILED assert isinstance(msg, TaskState) - assert msg.retry_reason == "rate limit; retries exhausted (2 of 2)" + assert msg.retry_reason == "rate limit; retries exhausted (3 of 3)" def test_plain_retries_exhausted_has_no_reason(create_runtime_ti, mock_supervisor_comms): @@ -1246,7 +1246,7 @@ def execute(self, context): raise RuntimeError("boom") task = _AlwaysFails(task_id="plain_exhausted") - ti = create_runtime_ti(task=task, try_number=2, max_tries=2, should_retry=False) + ti = create_runtime_ti(task=task, try_number=3, max_tries=2, should_retry=False) state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) From baa26759fd902267ecf4a8dc585a4337ebd61bdd Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Fri, 18 Sep 2026 10:16:58 +0530 Subject: [PATCH 4/9] review comments from kaxil --- .../execution_api/routes/task_instances.py | 4 +- .../airflow/sdk/definitions/retry_policy.py | 3 +- .../airflow/sdk/execution_time/task_runner.py | 17 +++++---- .../execution_time/schema/test_migrator.py | 34 +++++++++++++++++ .../execution_time/test_task_runner.py | 38 +++++++++++++++++-- 5 files changed, 82 insertions(+), 14 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 4464e6fc35480..ba3dc458eed7c 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 @@ -670,9 +670,9 @@ def _create_ti_state_update_query_and_update_state( query = query.values(state=updated_state, next_method=None, next_kwargs=None) 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: - failed_retry_reason: str | None = ti_patch_payload.retry_reason[:500] - query = query.values(retry_reason=failed_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): diff --git a/task-sdk/src/airflow/sdk/definitions/retry_policy.py b/task-sdk/src/airflow/sdk/definitions/retry_policy.py index 57d309edf87d0..d2a70a1e01ac4 100644 --- a/task-sdk/src/airflow/sdk/definitions/retry_policy.py +++ b/task-sdk/src/airflow/sdk/definitions/retry_policy.py @@ -53,7 +53,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" 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 74c9d30fc8b83..db2fd4fdbd825 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -1887,19 +1887,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, + 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]: @@ -1933,10 +1934,12 @@ def _finalize_task_failure( retry_kwargs["retry_reason"] = retry_reason[:500] return RetryTask(**retry_kwargs), TaskInstanceState.UP_FOR_RETRY if retry_reason is not None and ti._ti_context_from_server is not None: - # max_tries is the retry count, not the attempt count -- total attempts is max_tries + 1. - total_attempts = ti._ti_context_from_server.max_tries + 1 - suffix = f"; retries exhausted ({ti.try_number} of {total_attempts})" - retry_reason = f"{retry_reason[: 500 - len(suffix)]}{suffix}" + max_tries = ti._ti_context_from_server.max_tries + if max_tries > 0: + # max_tries is the retry count, not the attempt count -- total attempts is max_tries + 1. + suffix = f"; retries exhausted ({ti.try_number} of {max_tries + 1})" + retry_reason = f"{retry_reason[: 500 - len(suffix)]}{suffix}" + log.info("Retry policy decision", action="fail", reason=retry_reason) return ( TaskState( state=TaskInstanceState.FAILED, 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 cd5f5fff5fb56..19790f81cb5ac 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 @@ -470,3 +470,37 @@ def test_head_version_keeps_arg_bindings(self, real_migrator, startup_details): assert isinstance(defaulted, LiteralArgBinding) assert defaulted.from_default is True assert defaulted.value_schema.root == {"type": "integer", "format": "int64"} + + +class TestRealBundleRetryReasonUpgrade: + """ + Drive the *real* supervisor bundle through the ``retry_reason`` migration. + + ``TaskState`` flows foreign-runtime -> supervisor, the opposite direction from + ``arg_bindings`` above, so a runtime pinned to an older schema is exercised + through ``upgrade`` rather than ``downgrade``. + """ + + @pytest.fixture + def real_migrator(self) -> SchemaVersionMigrator: + return get_schema_version_migrator() + + def test_upgrade_fills_missing_retry_reason_with_none(self, real_migrator): + from airflow.sdk.execution_time.comms import TaskState + + body = {"type": "TaskState", "state": "failed", "end_date": None, "rendered_map_index": None} + out = real_migrator.upgrade(body, TaskState, "2026-06-16") + assert out["retry_reason"] is None + + def test_upgrade_keeps_retry_reason_at_head(self, real_migrator): + from airflow.sdk.execution_time.comms import TaskState + + body = { + "type": "TaskState", + "state": "failed", + "end_date": None, + "rendered_map_index": None, + "retry_reason": "auth error, do not retry", + } + out = real_migrator.upgrade(body, TaskState, "2026-10-30") + assert out["retry_reason"] == "auth error, do not retry" 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 4c4f50b46d243..5dc5b26432df2 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 @@ -1203,11 +1203,12 @@ def execute(self, context): task = _AlwaysFails( task_id="fail_with_reason", + retries=2, retry_policy=ExceptionRetryPolicy( rules=[RetryRule(exception=RuntimeError, action=RetryAction.FAIL, reason="do not retry")] ), ) - ti = create_runtime_ti(task=task, should_retry=True) + ti = create_runtime_ti(task=task) state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) @@ -1225,11 +1226,12 @@ def execute(self, context): task = _AlwaysFails( task_id="retry_exhausted", + retries=2, retry_policy=ExceptionRetryPolicy( rules=[RetryRule(exception=RuntimeError, action=RetryAction.RETRY, reason="rate limit")] ), ) - ti = create_runtime_ti(task=task, try_number=3, max_tries=2, should_retry=False) + ti = create_runtime_ti(task=task, try_number=3) state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) @@ -1238,6 +1240,34 @@ def execute(self, context): assert msg.retry_reason == "rate limit; retries exhausted (3 of 3)" +def test_retry_policy_retry_exhausted_reason_is_truncated_with_suffix_kept( + create_runtime_ti, mock_supervisor_comms +): + """A long reason is truncated to 500 chars total, with the exhausted-suffix always kept.""" + + class _AlwaysFails(BaseOperator): + def execute(self, context): + raise RuntimeError("boom") + + long_reason = "z" * 600 + task = _AlwaysFails( + task_id="retry_exhausted_long_reason", + retries=2, + retry_policy=ExceptionRetryPolicy( + rules=[RetryRule(exception=RuntimeError, action=RetryAction.RETRY, reason=long_reason)] + ), + ) + ti = create_runtime_ti(task=task, try_number=3) + + state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) + + assert state == TaskInstanceState.FAILED + assert isinstance(msg, TaskState) + assert msg.retry_reason is not None + assert len(msg.retry_reason) == 500 + assert msg.retry_reason.endswith("; retries exhausted (3 of 3)") + + def test_plain_retries_exhausted_has_no_reason(create_runtime_ti, mock_supervisor_comms): """Without a retry policy, exhausting the retry budget must not synthesize a reason.""" @@ -1245,8 +1275,8 @@ class _AlwaysFails(BaseOperator): def execute(self, context): raise RuntimeError("boom") - task = _AlwaysFails(task_id="plain_exhausted") - ti = create_runtime_ti(task=task, try_number=3, max_tries=2, should_retry=False) + task = _AlwaysFails(task_id="plain_exhausted", retries=2) + ti = create_runtime_ti(task=task, try_number=3) state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) From d9d3a00ac151923311f5a7296d7464d9df01e696 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Mon, 21 Sep 2026 12:46:33 +0530 Subject: [PATCH 5/9] after rebase comments --- providers/common/ai/docs/retry_policies.rst | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/providers/common/ai/docs/retry_policies.rst b/providers/common/ai/docs/retry_policies.rst index 8d1d5f14156b0..6d003e764e4c0 100644 --- a/providers/common/ai/docs/retry_policies.rst +++ b/providers/common/ai/docs/retry_policies.rst @@ -140,10 +140,12 @@ four fields: ``category``, ``should_retry``, ``suggested_delay_seconds``, and ``reasoning``. Only ``should_retry`` and ``suggested_delay_seconds`` affect the run. -``category`` and ``reasoning`` are only recorded on a RETRY. They are written +``category`` and ``reasoning`` are recorded on both outcomes. They are written to the task instance's ``retry_reason`` (truncated to 500 characters, see -below), then cleared once the next attempt starts running. On a FAIL they are -not written anywhere -- they only show up in the task log. +below). 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. When the model asked to retry but no attempts were left, the +stored reason ends with a ``; retries exhausted (N of M)`` note. Two limits are worth knowing about: @@ -208,7 +210,7 @@ When writing custom instructions: fields explicitly so the model fills them. - Be concrete with examples (``"'Warehouse suspended' -> transient"``) rather than vague rules ("treat warehouse issues as recoverable"). -- ``retry_reason`` is truncated to 500 chars in the audit log -- keep +- ``retry_reason`` is truncated to 500 chars on the task instance -- keep ``reasoning`` outputs concise. Parameters From d6bc665a89959d66bc2ad3f3b990bd69140905b3 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Mon, 21 Sep 2026 14:23:31 +0530 Subject: [PATCH 6/9] review comments --- .../datamodels/task_instance_history.py | 2 +- .../core_api/datamodels/task_instances.py | 2 +- .../core_api/openapi/_private_ui.yaml | 4 +- .../openapi/v2-rest-api-generated.yaml | 8 +- .../ui/openapi-gen/requests/schemas.gen.ts | 8 +- .../ui/openapi-gen/requests/types.gen.ts | 4 +- .../ui/public/i18n/locales/en/common.json | 6 +- .../src/pages/TaskInstance/Details.test.tsx | 157 ++++++++++++++++++ .../ui/src/pages/TaskInstance/Details.tsx | 46 ++++- .../core_api/routes/public/test_hitl.py | 2 +- .../routes/public/test_task_instances.py | 62 +++---- .../airflowctl/api/datamodels/generated.py | 4 +- 12 files changed, 248 insertions(+), 57 deletions(-) create mode 100644 airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py index 7662b68bbd210..256d3afc1aa86 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py @@ -62,7 +62,7 @@ class TaskInstanceHistoryResponse(BaseModel): executor: str | None executor_config: Annotated[str, BeforeValidator(str)] dag_version: DagVersionResponse | None - retry_reason: str | None = None + state_reason: str | None = Field(default=None, validation_alias="retry_reason") class TaskInstanceHistoryCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py index 3c893748440e8..38e8da0f577a3 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py @@ -89,7 +89,7 @@ class TaskInstanceResponse(BaseModel): queued_by_job: JobResponse | None = Field(alias="triggerer_job") dag_version: DagVersionResponse | None team_name: str | None = None - retry_reason: str | None = None + state_reason: str | None = Field(default=None, validation_alias="retry_reason") class TaskInstanceCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index 01c872ceebc6b..dbaeeda0cbe53 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml @@ -4837,11 +4837,11 @@ components: - type: string - type: 'null' title: Team Name - retry_reason: + state_reason: anyOf: - type: string - type: 'null' - title: Retry Reason + title: State Reason type: object required: - id diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 8e90323c72ab2..7151cdd004c63 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -16035,11 +16035,11 @@ components: anyOf: - $ref: '#/components/schemas/DagVersionResponse' - type: 'null' - retry_reason: + state_reason: anyOf: - type: string - type: 'null' - title: Retry Reason + title: State Reason type: object required: - task_id @@ -16223,11 +16223,11 @@ components: - type: string - type: 'null' title: Team Name - retry_reason: + state_reason: anyOf: - type: string - type: 'null' - title: Retry Reason + title: State Reason type: object required: - id diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 72648007fea27..83f671e5fa75c 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -7337,7 +7337,7 @@ export const $TaskInstanceHistoryResponse = { } ] }, - retry_reason: { + state_reason: { anyOf: [ { type: 'string' @@ -7346,7 +7346,7 @@ export const $TaskInstanceHistoryResponse = { type: 'null' } ], - title: 'Retry Reason' + title: 'State Reason' } }, type: 'object', @@ -7648,7 +7648,7 @@ export const $TaskInstanceResponse = { ], title: 'Team Name' }, - retry_reason: { + state_reason: { anyOf: [ { type: 'string' @@ -7657,7 +7657,7 @@ export const $TaskInstanceResponse = { type: 'null' } ], - title: 'Retry Reason' + title: 'State Reason' } }, type: 'object', diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 8240e3b611162..5698a38f93366 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1921,7 +1921,7 @@ export type TaskInstanceHistoryResponse = { executor: string | null; executor_config: string; dag_version: DagVersionResponse | null; - retry_reason?: string | null; + state_reason?: string | null; }; /** @@ -1965,7 +1965,7 @@ export type TaskInstanceResponse = { triggerer_job: JobResponse | null; dag_version: DagVersionResponse | null; team_name?: string | null; - retry_reason?: string | null; + state_reason?: string | null; }; /** diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json index f47d58514a87d..b91c94e437d3f 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json @@ -441,8 +441,12 @@ "queue": "Queue", "queuedWhen": "Queued At", "renderedMapIndex": "Rendered Map Index", - "retryReason": "Reason for state", "scheduledWhen": "Scheduled At", + "stateReason": "Reason for state", + "stateReasonSummary": { + "failed": "Stopped on try {{tryNumber}} of {{totalTries}}", + "upForRetry": "Retrying after try {{tryNumber}} of {{totalTries}}" + }, "trigger": "Trigger", "triggerer": { "assigned": "Assigned triggerer", diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx new file mode 100644 index 0000000000000..d28bd836f7415 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx @@ -0,0 +1,157 @@ +/*! + * 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. + */ +import "@testing-library/jest-dom"; +import { render, screen } from "@testing-library/react"; +import { describe, expect, it, vi } from "vitest"; + +import type { TaskInstanceHistoryResponse, TaskInstanceResponse } from "openapi/requests/types.gen"; + +import i18n from "src/i18n/config"; +import { Wrapper } from "src/utils/Wrapper"; + +import { Details } from "./Details"; + +// Sibling panels each fetch their own data and are unrelated to the state-reason +// banner and row under test. +vi.mock("./BlockingDeps", () => ({ BlockingDeps: () => undefined })); +vi.mock("./ExtraLinks", () => ({ ExtraLinks: () => undefined })); +vi.mock("./TriggererInfo", () => ({ TriggererInfo: () => undefined })); +vi.mock("src/components/DagVersionDetails", () => ({ DagVersionDetails: () => undefined })); +vi.mock("src/components/TaskTrySelect", () => ({ TaskTrySelect: () => undefined })); +vi.mock("src/components/TeamName", () => ({ TeamName: () => undefined })); +vi.mock("src/hooks/useShowTeam", () => ({ useShowTeam: () => false })); + +const mockTaskInstance = vi.fn<() => TaskInstanceResponse | undefined>(); +const mockTryInstance = vi.fn<() => TaskInstanceHistoryResponse | undefined>(); + +vi.mock("openapi/queries", async () => { + const actual = await vi.importActual("openapi/queries"); + + return { + ...actual, + useTaskInstanceServiceGetMappedTaskInstance: () => ({ data: mockTaskInstance() }), + useTaskInstanceServiceGetTaskInstanceTryDetails: () => ({ data: mockTryInstance() }), + }; +}); + +vi.mock("src/utils", async () => { + const actual = await vi.importActual("src/utils"); + + return { ...actual, useAutoRefresh: () => false }; +}); + +const buildTaskInstance = (overrides: Partial): TaskInstanceResponse => + ({ + dag_id: "test_dag", + dag_run_id: "run_1", + dag_version: null, + duration: null, + end_date: null, + id: "ti-id", + map_index: -1, + max_tries: 2, + note: null, + operator_name: "PythonOperator", + rendered_map_index: null, + start_date: null, + state: "failed", + state_reason: null, + task_display_name: "test_task", + task_id: "test_task", + trigger: null, + triggerer_job: null, + try_number: 3, + ...overrides, + }) as unknown as TaskInstanceResponse; + +const renderDetails = ( + taskInstance: TaskInstanceResponse, + tryInstance: Partial = {}, +) => { + mockTaskInstance.mockReturnValue(taskInstance); + mockTryInstance.mockReturnValue({ + ...taskInstance, + ...tryInstance, + }); + + return render(
, { wrapper: Wrapper }); +}; + +describe("Details state reason", () => { + it("does not render the banner when there is no reason", () => { + renderDetails(buildTaskInstance({ state_reason: null })); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + expect(screen.queryByText(i18n.t("common:taskInstance.stateReason"))).not.toBeInTheDocument(); + }); + + it.each([ + { state: "failed", titleKey: "failed" }, + { state: "up_for_retry", titleKey: "upForRetry" }, + ] as const)("titles the banner for a $state task", ({ state, titleKey }) => { + renderDetails(buildTaskInstance({ max_tries: 2, state, state_reason: "auth error", try_number: 3 })); + + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( + i18n.t(`common:taskInstance.stateReasonSummary.${titleKey}`, { totalTries: 3, tryNumber: 3 }), + ); + }); + + // Chakra encodes `status` in a generated class rather than a DOM attribute, so the error/warning + // distinction can only be pinned as "the two states do not render identically". + it("styles a failed banner differently from an up_for_retry one", () => { + const { unmount } = renderDetails(buildTaskInstance({ state: "failed", state_reason: "auth error" })); + const failedClass = screen.getByTestId("state-reason-alert").className; + + unmount(); + renderDetails(buildTaskInstance({ state: "up_for_retry", state_reason: "auth error" })); + + expect(screen.getByTestId("state-reason-alert").className).not.toBe(failedClass); + }); + + // The reason is only cleared once the task next reaches RUNNING, so a cleared task keeps a + // reason describing the previous attempt. Gating on state is what stops it being shown. + it.each(["queued", "running", "success", null] as const)( + "does not render the banner for a %s task that still carries a reason", + (state) => { + renderDetails(buildTaskInstance({ state, state_reason: "auth error, do not retry" })); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + }, + ); + + it("titles the banner with the try counts so it is distinct from the per-try row", () => { + renderDetails( + buildTaskInstance({ max_tries: 2, state: "failed", state_reason: "rate limit", try_number: 3 }), + ); + + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( + i18n.t("common:taskInstance.stateReasonSummary.failed", { totalTries: 3, tryNumber: 3 }), + ); + }); + + it("shows the selected try's reason in the table while the banner keeps the latest try's", () => { + renderDetails(buildTaskInstance({ state_reason: "latest try: rate limit" }), { + state_reason: "older try: auth error", + }); + + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent("latest try: rate limit"); + expect(screen.getByText("older try: auth error")).toBeInTheDocument(); + expect(screen.getByText(i18n.t("common:taskInstance.stateReason"))).toBeInTheDocument(); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx index 46d2e0ba4c193..09160ba2d3e8d 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx @@ -44,6 +44,8 @@ import { BlockingDeps } from "./BlockingDeps"; import { ExtraLinks } from "./ExtraLinks"; import { TriggererInfo } from "./TriggererInfo"; +type StateReasonSummary = { reason: string; status: "error" | "warning"; title: string }; + export const Details = () => { const { t: translate } = useTranslation(); const { renderDuration } = useDurationFormat(); @@ -115,6 +117,34 @@ export const Details = () => { return translate("common:none", { defaultValue: "None" }); }; + const stateReasonSummary = ((): StateReasonSummary | undefined => { + const reason = taskInstance?.state_reason; + + if (reason === null || reason === undefined || taskInstance === undefined) { + return undefined; + } + + const counts = { totalTries: taskInstance.max_tries + 1, tryNumber: taskInstance.try_number }; + + if (taskInstance.state === "failed") { + return { + reason, + status: "error", + title: translate("taskInstance.stateReasonSummary.failed", counts), + }; + } + + if (taskInstance.state === "up_for_retry") { + return { + reason, + status: "warning", + title: translate("taskInstance.stateReasonSummary.upForRetry", counts), + }; + } + + return undefined; + })(); + // omit kwargs from trigger const triggerWithoutKwargs = taskInstance?.trigger ? (({ kwargs, ...rest }) => rest)(taskInstance.trigger) @@ -131,14 +161,14 @@ export const Details = () => { return ( - {taskInstance?.retry_reason === null || taskInstance?.retry_reason === undefined ? undefined : ( + {stateReasonSummary === undefined ? undefined : ( - {taskInstance.retry_reason} + {stateReasonSummary.reason} )} {taskInstance === undefined || tryNumber === undefined || taskInstance.try_number <= 1 ? ( @@ -172,10 +202,10 @@ export const Details = () => { - {tryInstance?.retry_reason === null || tryInstance?.retry_reason === undefined ? undefined : ( + {tryInstance?.state_reason === null || tryInstance?.state_reason === undefined ? undefined : ( - {translate("taskInstance.retryReason")} - {tryInstance.retry_reason} + {translate("taskInstance.stateReason")} + {tryInstance.state_reason} )} diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py index 8b746f3e93322..1bb2c96d4f05d 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py @@ -267,7 +267,7 @@ def expected_sample_hitl_detail_dict(sample_ti: TaskInstance) -> dict[str, Any]: "task_display_name": "sample_task_hitl", "task_id": TASK_ID, "team_name": None, - "retry_reason": None, + "state_reason": None, "trigger": None, "triggerer_job": None, "try_number": 0, diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index 0952c6b13684f..70c4e62b01e05 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -244,16 +244,16 @@ def test_should_respond_200(self, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } - def test_should_include_retry_reason(self, test_client, session): + def test_should_include_state_reason(self, test_client, session): self.create_task_instances(session, task_instances=[{"retry_reason": "auth error, do not retry"}]) response = test_client.get( "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context" ) assert response.status_code == 200 - assert response.json()["retry_reason"] == "auth error, do not retry" + assert response.json()["state_reason"] == "auth error, do not retry" @conf_vars({("core", "multi_team"): "True"}) def test_should_include_team_name(self, test_client, session): @@ -338,7 +338,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, "dag_version": { "id": response_data["dag_version"]["id"], "version_number": expected_version_number, @@ -435,7 +435,7 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "unixname": getuser(), }, "team_name": None, - "retry_reason": None, + "state_reason": None, } def test_should_respond_200_with_task_state_in_removed(self, test_client, session): @@ -490,7 +490,7 @@ def test_should_respond_200_with_task_state_in_removed(self, test_client, sessio "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } def test_should_respond_200_task_instance_with_rendered(self, test_client, session): @@ -548,7 +548,7 @@ def test_should_respond_200_task_instance_with_rendered(self, test_client, sessi "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } def test_raises_404_for_nonexistent_task_instance(self, test_client): @@ -670,7 +670,7 @@ def test_should_respond_200_mapped_task_instance_with_rtif(self, test_client, se "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } def test_should_respond_401(self, unauthenticated_test_client): @@ -2826,10 +2826,10 @@ def test_should_respond_200(self, test_client, session): "id": response_data["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, } - def test_should_include_retry_reason_from_history(self, test_client, session): + def test_should_include_state_reason_from_history(self, test_client, session): self.create_task_instances( session, task_instances=[{"state": State.SUCCESS, "retry_reason": "auth error, do not retry"}], @@ -2839,7 +2839,7 @@ def test_should_include_retry_reason_from_history(self, test_client, session): "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries/1" ) assert response.status_code == 200 - assert response.json()["retry_reason"] == "auth error, do not retry" + assert response.json()["state_reason"] == "auth error, do not retry" @pytest.mark.parametrize("try_number", [1, 2]) def test_should_respond_200_with_different_try_numbers(self, test_client, try_number, session): @@ -2885,7 +2885,7 @@ def test_should_respond_200_with_different_try_numbers(self, test_client, try_nu "id": response_data["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, } @pytest.mark.parametrize("try_number", [1, 2]) @@ -2963,7 +2963,7 @@ def test_should_respond_200_with_mapped_task_at_different_try_numbers( "id": response_data["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, } def test_should_respond_200_with_task_state_in_deferred(self, test_client, session): @@ -3036,7 +3036,7 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "id": response_data["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, } def test_should_respond_200_with_task_state_in_removed(self, test_client, session): @@ -3084,7 +3084,7 @@ def test_should_respond_200_with_task_state_in_removed(self, test_client, sessio "id": response_data["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, } def test_should_respond_401(self, unauthenticated_test_client): @@ -3160,7 +3160,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "created_at": mock.ANY, "dag_display_name": "dag_with_multiple_versions", }, - "retry_reason": None, + "state_reason": None, } def test_should_not_return_duplicate_runs(self, test_client, session): @@ -3898,7 +3898,7 @@ def test_should_respond_200_with_dag_run_id( "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, "try_number": 0, "unixname": getuser(), }, @@ -4362,7 +4362,7 @@ def test_should_respond_200(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, }, { "dag_id": "example_python_operator", @@ -4400,7 +4400,7 @@ def test_should_respond_200(self, test_client, session): "id": response_data["task_instances"][1]["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, }, ], "total_entries": 2, @@ -4472,7 +4472,7 @@ def test_ti_in_retry_state_not_returned(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, }, ], "total_entries": 1, @@ -4556,7 +4556,7 @@ def test_mapped_task_should_respond_200(self, test_client, session): "id": response_data["task_instances"][0]["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, }, { "dag_id": "example_python_operator", @@ -4594,7 +4594,7 @@ def test_mapped_task_should_respond_200(self, test_client, session): "id": response_data["task_instances"][1]["dag_version"]["id"], "version_number": 1, }, - "retry_reason": None, + "state_reason": None, }, ], "total_entries": 2, @@ -4662,7 +4662,7 @@ def test_should_respond_200_with_versions(self, test_client, run_id, expected_ve "created_at": mock.ANY, "dag_display_name": "dag_with_multiple_versions", }, - "retry_reason": None, + "state_reason": None, } @@ -4788,7 +4788,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, @@ -5067,7 +5067,7 @@ def test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, @@ -5208,7 +5208,7 @@ def test_update_mask_set_note_should_respond_200( "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, @@ -5274,7 +5274,7 @@ def test_set_note_should_respond_200(self, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, @@ -5372,7 +5372,7 @@ def test_set_note_should_respond_200_mapped_task_with_rtif(self, test_client, se "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, @@ -5458,7 +5458,7 @@ def test_set_note_should_respond_200_mapped_task_summary_with_rtif(self, test_cl "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } _check_task_instance_note( @@ -5655,7 +5655,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session): "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, @@ -5946,7 +5946,7 @@ def test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte "trigger": None, "triggerer_job": None, "team_name": None, - "retry_reason": None, + "state_reason": None, } ], "total_entries": 1, diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index 3f75dd623e71e..89f71e6ddf9e0 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -2247,7 +2247,7 @@ class TaskInstanceHistoryResponse(BaseModel): executor: Annotated[str | None, Field(title="Executor")] executor_config: Annotated[str, Field(title="Executor Config")] dag_version: DagVersionResponse | None - retry_reason: Annotated[str | None, Field(title="Retry Reason")] = None + state_reason: Annotated[str | None, Field(title="State Reason")] = None class TaskInstanceResponse(BaseModel): @@ -2290,7 +2290,7 @@ class TaskInstanceResponse(BaseModel): triggerer_job: JobResponse | None dag_version: DagVersionResponse | None team_name: Annotated[str | None, Field(title="Team Name")] = None - retry_reason: Annotated[str | None, Field(title="Retry Reason")] = None + state_reason: Annotated[str | None, Field(title="State Reason")] = None class TaskResponse(BaseModel): From 6d467f3daa50857562124d8a4c6208d1009e3fe8 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Tue, 22 Sep 2026 15:17:18 +0530 Subject: [PATCH 7/9] review comments --- .../src/pages/TaskInstance/Details.test.tsx | 17 +++++++++++-- .../ui/src/pages/TaskInstance/Details.tsx | 25 ++++++++++++++++--- ts-sdk/src/generated/supervisor.ts | 2 ++ 3 files changed, 39 insertions(+), 5 deletions(-) diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx index d28bd836f7415..2f3166f42a45b 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx @@ -125,16 +125,29 @@ describe("Details state reason", () => { }); // The reason is only cleared once the task next reaches RUNNING, so a cleared task keeps a - // reason describing the previous attempt. Gating on state is what stops it being shown. + // reason describing the previous attempt. Gating on state is what stops it being shown, and it + // has to cover the row as well as the banner or the stale text just moves down the page. it.each(["queued", "running", "success", null] as const)( - "does not render the banner for a %s task that still carries a reason", + "renders neither the banner nor the row for a %s task that still carries a reason", (state) => { renderDetails(buildTaskInstance({ state, state_reason: "auth error, do not retry" })); expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + expect(screen.queryByText(i18n.t("common:taskInstance.stateReason"))).not.toBeInTheDocument(); + expect(screen.queryByText("auth error, do not retry")).not.toBeInTheDocument(); }, ); + it("keeps an earlier failed try's reason while the task is running again", () => { + renderDetails(buildTaskInstance({ state: "running", state_reason: null }), { + state: "failed", + state_reason: "try 1: auth error", + }); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + expect(screen.getByText("try 1: auth error")).toBeInTheDocument(); + }); + it("titles the banner with the try counts so it is distinct from the per-try row", () => { renderDetails( buildTaskInstance({ max_tries: 2, state: "failed", state_reason: "rate limit", try_number: 3 }), diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx index 09160ba2d3e8d..358ba47bf4fb4 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx @@ -46,6 +46,11 @@ import { TriggererInfo } from "./TriggererInfo"; type StateReasonSummary = { reason: string; status: "error" | "warning"; title: string }; +// The reason is only cleared once the task next reaches RUNNING, so a cleared task still carries +// the previous attempt's reason. Both the banner and the per-try row key off the state to avoid +// explaining a state the task is no longer in. +const STATES_WITH_REASON = ["failed", "up_for_retry"]; + export const Details = () => { const { t: translate } = useTranslation(); const { renderDuration } = useDurationFormat(); @@ -120,7 +125,12 @@ export const Details = () => { const stateReasonSummary = ((): StateReasonSummary | undefined => { const reason = taskInstance?.state_reason; - if (reason === null || reason === undefined || taskInstance === undefined) { + if ( + reason === null || + reason === undefined || + taskInstance === undefined || + !STATES_WITH_REASON.includes(taskInstance.state ?? "") + ) { return undefined; } @@ -145,6 +155,15 @@ export const Details = () => { return undefined; })(); + // Keyed off the selected try's own state, so an earlier failed try keeps its reason while the + // current one is running again. + const tryStateReason = + tryInstance?.state_reason !== null && + tryInstance?.state_reason !== undefined && + STATES_WITH_REASON.includes(tryInstance.state ?? "") + ? tryInstance.state_reason + : undefined; + // omit kwargs from trigger const triggerWithoutKwargs = taskInstance?.trigger ? (({ kwargs, ...rest }) => rest)(taskInstance.trigger) @@ -202,10 +221,10 @@ export const Details = () => { - {tryInstance?.state_reason === null || tryInstance?.state_reason === undefined ? undefined : ( + {tryStateReason === undefined ? undefined : ( {translate("taskInstance.stateReason")} - {tryInstance.state_reason} + {tryStateReason} )} diff --git a/ts-sdk/src/generated/supervisor.ts b/ts-sdk/src/generated/supervisor.ts index 8eb27a1d8152e..5a4612eab2eb2 100644 --- a/ts-sdk/src/generated/supervisor.ts +++ b/ts-sdk/src/generated/supervisor.ts @@ -597,6 +597,7 @@ export type State8 = "failed" | "skipped" | "removed"; export type EndDate8 = string | null; export type Type78 = "TaskState"; export type RenderedMapIndex5 = string | null; +export type RetryReason1 = string | null; export type Type79 = "TaskStateStoreResult"; export type Type80 = "TaskStatesResult"; export type LogicalDate6 = string | null; @@ -1825,6 +1826,7 @@ export interface TaskState { end_date?: EndDate8; type?: Type78; rendered_map_index?: RenderedMapIndex5; + retry_reason?: RetryReason1; } /** * Response to GetTaskStateStore; wraps the generated API response for supervisor to worker comms. From 74e7573a5606e4fa7bf83616ab1aef5106a1bb62 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Thu, 24 Sep 2026 16:27:06 +0530 Subject: [PATCH 8/9] after rebase comments on PR 2 --- .../datamodels/task_instance_history.py | 9 ++++ .../core_api/datamodels/task_instances.py | 11 ++++ .../src/pages/TaskInstance/Details.test.tsx | 39 ++++++++++---- .../ui/src/pages/TaskInstance/Details.tsx | 51 ++++++++----------- .../routes/public/test_task_instances.py | 28 ++++++++++ providers/common/ai/docs/retry_policies.rst | 15 +++--- 6 files changed, 108 insertions(+), 45 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py index 256d3afc1aa86..f997faf08d09d 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py @@ -23,8 +23,10 @@ AliasPath, BeforeValidator, Field, + field_validator, ) +from airflow._shared.secrets_masker import redact from airflow.api_fastapi.core_api.base import BaseModel from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse from airflow.utils.state import TaskInstanceState @@ -64,6 +66,13 @@ class TaskInstanceHistoryResponse(BaseModel): dag_version: DagVersionResponse | None state_reason: str | None = Field(default=None, validation_alias="retry_reason") + @field_validator("state_reason", mode="after") + @classmethod + def redact_state_reason(cls, v: str | None) -> str | None: + if v is None: + return None + return str(redact(v)) + class TaskInstanceHistoryCollectionResponse(BaseModel): """TaskInstanceHistory Collection serializer for responses.""" diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py index b70dfcd429b0b..3bf412a498d5e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py @@ -34,6 +34,7 @@ model_validator, ) +from airflow._shared.secrets_masker import redact from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse from airflow.api_fastapi.core_api.datamodels.job import JobResponse @@ -91,6 +92,16 @@ class TaskInstanceResponse(BaseModel): team_name: str | None = None state_reason: str | None = Field(default=None, validation_alias="retry_reason") + @field_validator("state_reason", mode="after") + @classmethod + def redact_state_reason(cls, v: str | None) -> str | None: + # A retry policy composes this from the exception text, and a policy may opt out of the + # worker-side redaction, so the same string that would be masked in a task log can reach + # here unmasked. + if v is None: + return None + return str(redact(v)) + class TaskInstanceCollectionResponse(BaseModel): """ diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx index 2f3166f42a45b..87d8b361b255d 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx @@ -18,13 +18,14 @@ */ import "@testing-library/jest-dom"; import { render, screen } from "@testing-library/react"; -import { describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import type { TaskInstanceHistoryResponse, TaskInstanceResponse } from "openapi/requests/types.gen"; import i18n from "src/i18n/config"; import { Wrapper } from "src/utils/Wrapper"; +import commonLocale from "../../../public/i18n/locales/en/common.json"; import { Details } from "./Details"; // Sibling panels each fetch their own data and are unrelated to the state-reason @@ -94,6 +95,12 @@ const renderDetails = ( }; describe("Details state reason", () => { + // Without the bundle, i18n.t() echoes the key back and every title assertion compares a key + // against itself, so interpolated counts are never checked. + beforeEach(() => { + i18n.addResourceBundle("en", "common", commonLocale, true, true); + }); + it("does not render the banner when there is no reason", () => { renderDetails(buildTaskInstance({ state_reason: null })); @@ -102,15 +109,27 @@ describe("Details state reason", () => { }); it.each([ - { state: "failed", titleKey: "failed" }, - { state: "up_for_retry", titleKey: "upForRetry" }, - ] as const)("titles the banner for a $state task", ({ state, titleKey }) => { - renderDetails(buildTaskInstance({ max_tries: 2, state, state_reason: "auth error", try_number: 3 })); - - expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( - i18n.t(`common:taskInstance.stateReasonSummary.${titleKey}`, { totalTries: 3, tryNumber: 3 }), - ); - }); + { maxTries: 2, state: "failed", titleKey: "failed", totalTries: 3, tryNumber: 3 }, + // try_number != max_tries + 1 is the norm mid-retry, and the differing numbers are what make + // a swapped or off-by-one interpolation visible. + { maxTries: 3, state: "up_for_retry", titleKey: "upForRetry", totalTries: 4, tryNumber: 2 }, + ] as const)( + "titles the banner for a $state task", + ({ maxTries, state, titleKey, totalTries, tryNumber }) => { + renderDetails( + buildTaskInstance({ + max_tries: maxTries, + state, + state_reason: "auth error", + try_number: tryNumber, + }), + ); + + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( + i18n.t(`common:taskInstance.stateReasonSummary.${titleKey}`, { totalTries, tryNumber }), + ); + }, + ); // Chakra encodes `status` in a generated class rather than a DOM attribute, so the error/warning // distinction can only be pinned as "the two states do not render identically". diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx index 358ba47bf4fb4..65bd65f928d41 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx @@ -47,9 +47,17 @@ import { TriggererInfo } from "./TriggererInfo"; type StateReasonSummary = { reason: string; status: "error" | "warning"; title: string }; // The reason is only cleared once the task next reaches RUNNING, so a cleared task still carries -// the previous attempt's reason. Both the banner and the per-try row key off the state to avoid -// explaining a state the task is no longer in. -const STATES_WITH_REASON = ["failed", "up_for_retry"]; +// the previous attempt's reason. Both the banner and the per-try row look the state up here, so a +// state added to one surface cannot be forgotten on the other: it has to bring a title with it. +const STATE_REASON_DISPLAY = { + failed: { status: "error", titleKey: "failed" }, + up_for_retry: { status: "warning", titleKey: "upForRetry" }, +} as const satisfies Record; + +const stateReasonDisplay = (state: string | null | undefined) => + state === null || state === undefined + ? undefined + : (STATE_REASON_DISPLAY as Record)[state]; export const Details = () => { const { t: translate } = useTranslation(); @@ -124,35 +132,20 @@ export const Details = () => { const stateReasonSummary = ((): StateReasonSummary | undefined => { const reason = taskInstance?.state_reason; + const display = stateReasonDisplay(taskInstance?.state); - if ( - reason === null || - reason === undefined || - taskInstance === undefined || - !STATES_WITH_REASON.includes(taskInstance.state ?? "") - ) { + if (reason === null || reason === undefined || taskInstance === undefined || display === undefined) { return undefined; } - const counts = { totalTries: taskInstance.max_tries + 1, tryNumber: taskInstance.try_number }; - - if (taskInstance.state === "failed") { - return { - reason, - status: "error", - title: translate("taskInstance.stateReasonSummary.failed", counts), - }; - } - - if (taskInstance.state === "up_for_retry") { - return { - reason, - status: "warning", - title: translate("taskInstance.stateReasonSummary.upForRetry", counts), - }; - } - - return undefined; + return { + reason, + status: display.status, + title: translate(`taskInstance.stateReasonSummary.${display.titleKey}`, { + totalTries: taskInstance.max_tries + 1, + tryNumber: taskInstance.try_number, + }), + }; })(); // Keyed off the selected try's own state, so an earlier failed try keeps its reason while the @@ -160,7 +153,7 @@ export const Details = () => { const tryStateReason = tryInstance?.state_reason !== null && tryInstance?.state_reason !== undefined && - STATES_WITH_REASON.includes(tryInstance.state ?? "") + stateReasonDisplay(tryInstance.state) !== undefined ? tryInstance.state_reason : undefined; diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index 2d642c88b68d1..9222414d82757 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -31,6 +31,7 @@ from sqlalchemy import delete, func, select, update from sqlalchemy.orm import joinedload +from airflow._shared.secrets_masker import mask_secret from airflow._shared.state import TaskScope from airflow._shared.timezones.timezone import datetime from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser @@ -255,6 +256,19 @@ def test_should_include_state_reason(self, test_client, session): assert response.status_code == 200 assert response.json()["state_reason"] == "auth error, do not retry" + @pytest.mark.enable_redact + def test_should_redact_secrets_in_state_reason(self, test_client, session): + """A policy may compose the reason from an unredacted exception, so mask on the way out.""" + mask_secret("hunter2") + self.create_task_instances( + session, task_instances=[{"retry_reason": "auth: the token hunter2 expired"}] + ) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context" + ) + assert response.status_code == 200 + assert "hunter2" not in response.json()["state_reason"] + @conf_vars({("core", "multi_team"): "True"}) def test_should_include_team_name(self, test_client, session): self.create_task_instances(session) @@ -2841,6 +2855,20 @@ def test_should_include_state_reason_from_history(self, test_client, session): assert response.status_code == 200 assert response.json()["state_reason"] == "auth error, do not retry" + @pytest.mark.enable_redact + def test_should_redact_secrets_in_state_reason_from_history(self, test_client, session): + mask_secret("hunter2") + self.create_task_instances( + session, + task_instances=[{"state": State.SUCCESS, "retry_reason": "auth: the token hunter2 expired"}], + with_ti_history=True, + ) + response = test_client.get( + "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries/1" + ) + assert response.status_code == 200 + assert "hunter2" not in response.json()["state_reason"] + @pytest.mark.parametrize("try_number", [1, 2]) def test_should_respond_200_with_different_try_numbers(self, test_client, try_number, session): self.create_task_instances(session, task_instances=[{"state": State.SUCCESS}], with_ti_history=True) diff --git a/providers/common/ai/docs/retry_policies.rst b/providers/common/ai/docs/retry_policies.rst index cd88962473f03..fa6439e3aa8e8 100644 --- a/providers/common/ai/docs/retry_policies.rst +++ b/providers/common/ai/docs/retry_policies.rst @@ -133,7 +133,9 @@ When a task fails, either policy: ``retry_reason``, on a FAIL as well as a RETRY: ``: `` from ``LLMRetryPolicy``, or one line such as ``category=network confidence=0.91 threshold=0.60 action=retry delay=10s`` - from ``ClassifierRetryPolicy``. + from ``ClassifierRetryPolicy``. The REST API exposes it as ``state_reason`` + on a task instance and on each try, and the Task Instance page shows it + under **Reason for state**. This classification call is a separate model request, made by the policy itself rather than by an operator -- it is not subject to an operator's @@ -406,11 +408,12 @@ 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`` (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. +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 From d9e102647b1517a58ca74e8a3712ff77339f58c4 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Fri, 25 Sep 2026 13:36:15 +0530 Subject: [PATCH 9/9] comments from brent and kaxil --- .../datamodels/task_instance_history.py | 12 +++- .../core_api/datamodels/task_instances.py | 12 +++- .../core_api/openapi/_private_ui.yaml | 4 ++ .../openapi/v2-rest-api-generated.yaml | 8 +++ .../ui/openapi-gen/requests/schemas.gen.ts | 6 +- .../ui/openapi-gen/requests/types.gen.ts | 6 ++ .../src/pages/TaskInstance/Details.test.tsx | 64 ++----------------- .../ui/src/pages/TaskInstance/Details.tsx | 46 +------------ .../ui/src/pages/TaskInstance/Header.test.tsx | 54 ++++++++++++++++ .../ui/src/pages/TaskInstance/Header.tsx | 20 ++++++ .../ui/src/pages/TaskInstance/stateReason.ts | 30 +++++++++ .../routes/public/test_task_instances.py | 31 +++++++-- .../airflowctl/api/datamodels/generated.py | 16 ++++- providers/common/ai/docs/retry_policies.rst | 7 +- .../airflow/sdk/execution_time/task_runner.py | 9 ++- .../execution_time/test_task_runner.py | 24 +++++++ 16 files changed, 227 insertions(+), 122 deletions(-) create mode 100644 airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py index f997faf08d09d..43e289a1cc49a 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py @@ -17,7 +17,7 @@ from __future__ import annotations from datetime import datetime -from typing import Annotated +from typing import Annotated, cast from pydantic import ( AliasPath, @@ -64,14 +64,20 @@ class TaskInstanceHistoryResponse(BaseModel): executor: str | None executor_config: Annotated[str, BeforeValidator(str)] dag_version: DagVersionResponse | None - state_reason: str | None = Field(default=None, validation_alias="retry_reason") + state_reason: str | None = Field( + default=None, + validation_alias="retry_reason", + description=( + "The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended." + ), + ) @field_validator("state_reason", mode="after") @classmethod def redact_state_reason(cls, v: str | None) -> str | None: if v is None: return None - return str(redact(v)) + return cast("str", redact(v)) class TaskInstanceHistoryCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py index 3bf412a498d5e..e2d84e9bb6bb0 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py @@ -18,7 +18,7 @@ from collections.abc import Iterable from datetime import datetime -from typing import Annotated, Any +from typing import Annotated, Any, cast from uuid import UUID from pydantic import ( @@ -90,7 +90,13 @@ class TaskInstanceResponse(BaseModel): queued_by_job: JobResponse | None = Field(alias="triggerer_job") dag_version: DagVersionResponse | None team_name: str | None = None - state_reason: str | None = Field(default=None, validation_alias="retry_reason") + state_reason: str | None = Field( + default=None, + validation_alias="retry_reason", + description=( + "The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended." + ), + ) @field_validator("state_reason", mode="after") @classmethod @@ -100,7 +106,7 @@ def redact_state_reason(cls, v: str | None) -> str | None: # here unmasked. if v is None: return None - return str(redact(v)) + return cast("str", redact(v)) class TaskInstanceCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index dbaeeda0cbe53..357517767f3b8 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml @@ -4842,6 +4842,10 @@ components: - type: string - type: 'null' title: State Reason + description: 'The reason the task instance reached its current state, as + recorded by a retry policy. May describe a previous attempt: it is cleared + only when the task next starts running, so a task waiting to be retried + or re-run can still carry the reason its last attempt ended.' type: object required: - id diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 77bff40fda763..aa9225d60e6ed 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -16327,6 +16327,10 @@ components: - type: string - type: 'null' title: State Reason + description: 'The reason the task instance reached its current state, as + recorded by a retry policy. May describe a previous attempt: it is cleared + only when the task next starts running, so a task waiting to be retried + or re-run can still carry the reason its last attempt ended.' type: object required: - task_id @@ -16515,6 +16519,10 @@ components: - type: string - type: 'null' title: State Reason + description: 'The reason the task instance reached its current state, as + recorded by a retry policy. May describe a previous attempt: it is cleared + only when the task next starts running, so a task waiting to be retried + or re-run can still carry the reason its last attempt ended.' type: object required: - id diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 2d1cdfcc919d0..2a08de114e68a 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -7530,7 +7530,8 @@ export const $TaskInstanceHistoryResponse = { type: 'null' } ], - title: 'State Reason' + title: 'State Reason', + description: 'The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.' } }, type: 'object', @@ -7841,7 +7842,8 @@ export const $TaskInstanceResponse = { type: 'null' } ], - title: 'State Reason' + title: 'State Reason', + description: 'The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.' } }, type: 'object', diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 1407a18f00d7a..cf76bdcf439a8 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1995,6 +1995,9 @@ export type TaskInstanceHistoryResponse = { executor: string | null; executor_config: string; dag_version: DagVersionResponse | null; + /** + * The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended. + */ state_reason?: string | null; }; @@ -2039,6 +2042,9 @@ export type TaskInstanceResponse = { triggerer_job: JobResponse | null; dag_version: DagVersionResponse | null; team_name?: string | null; + /** + * The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended. + */ state_reason?: string | null; }; diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx index 87d8b361b255d..cc7d063bbd744 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx @@ -28,8 +28,7 @@ import { Wrapper } from "src/utils/Wrapper"; import commonLocale from "../../../public/i18n/locales/en/common.json"; import { Details } from "./Details"; -// Sibling panels each fetch their own data and are unrelated to the state-reason -// banner and row under test. +// Sibling panels each fetch their own data and are unrelated to the row under test. vi.mock("./BlockingDeps", () => ({ BlockingDeps: () => undefined })); vi.mock("./ExtraLinks", () => ({ ExtraLinks: () => undefined })); vi.mock("./TriggererInfo", () => ({ TriggererInfo: () => undefined })); @@ -94,9 +93,8 @@ const renderDetails = ( return render(
, { wrapper: Wrapper }); }; -describe("Details state reason", () => { - // Without the bundle, i18n.t() echoes the key back and every title assertion compares a key - // against itself, so interpolated counts are never checked. +describe("Details state reason row", () => { + // Without the bundle i18n.t() echoes the key, so the label assertions below would pass blindly. beforeEach(() => { i18n.addResourceBundle("en", "common", commonLocale, true, true); }); @@ -104,54 +102,15 @@ describe("Details state reason", () => { it("does not render the banner when there is no reason", () => { renderDetails(buildTaskInstance({ state_reason: null })); - expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); expect(screen.queryByText(i18n.t("common:taskInstance.stateReason"))).not.toBeInTheDocument(); }); - it.each([ - { maxTries: 2, state: "failed", titleKey: "failed", totalTries: 3, tryNumber: 3 }, - // try_number != max_tries + 1 is the norm mid-retry, and the differing numbers are what make - // a swapped or off-by-one interpolation visible. - { maxTries: 3, state: "up_for_retry", titleKey: "upForRetry", totalTries: 4, tryNumber: 2 }, - ] as const)( - "titles the banner for a $state task", - ({ maxTries, state, titleKey, totalTries, tryNumber }) => { - renderDetails( - buildTaskInstance({ - max_tries: maxTries, - state, - state_reason: "auth error", - try_number: tryNumber, - }), - ); - - expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( - i18n.t(`common:taskInstance.stateReasonSummary.${titleKey}`, { totalTries, tryNumber }), - ); - }, - ); - - // Chakra encodes `status` in a generated class rather than a DOM attribute, so the error/warning - // distinction can only be pinned as "the two states do not render identically". - it("styles a failed banner differently from an up_for_retry one", () => { - const { unmount } = renderDetails(buildTaskInstance({ state: "failed", state_reason: "auth error" })); - const failedClass = screen.getByTestId("state-reason-alert").className; - - unmount(); - renderDetails(buildTaskInstance({ state: "up_for_retry", state_reason: "auth error" })); - - expect(screen.getByTestId("state-reason-alert").className).not.toBe(failedClass); - }); - - // The reason is only cleared once the task next reaches RUNNING, so a cleared task keeps a - // reason describing the previous attempt. Gating on state is what stops it being shown, and it - // has to cover the row as well as the banner or the stale text just moves down the page. + // Cleared only once the task next reaches RUNNING, so these states still carry a stale reason. it.each(["queued", "running", "success", null] as const)( "renders neither the banner nor the row for a %s task that still carries a reason", (state) => { renderDetails(buildTaskInstance({ state, state_reason: "auth error, do not retry" })); - expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); expect(screen.queryByText(i18n.t("common:taskInstance.stateReason"))).not.toBeInTheDocument(); expect(screen.queryByText("auth error, do not retry")).not.toBeInTheDocument(); }, @@ -163,27 +122,16 @@ describe("Details state reason", () => { state_reason: "try 1: auth error", }); - expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); expect(screen.getByText("try 1: auth error")).toBeInTheDocument(); }); - it("titles the banner with the try counts so it is distinct from the per-try row", () => { - renderDetails( - buildTaskInstance({ max_tries: 2, state: "failed", state_reason: "rate limit", try_number: 3 }), - ); - - expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( - i18n.t("common:taskInstance.stateReasonSummary.failed", { totalTries: 3, tryNumber: 3 }), - ); - }); - - it("shows the selected try's reason in the table while the banner keeps the latest try's", () => { + it("shows the selected try's reason in the table", () => { renderDetails(buildTaskInstance({ state_reason: "latest try: rate limit" }), { state_reason: "older try: auth error", }); - expect(screen.getByTestId("state-reason-alert")).toHaveTextContent("latest try: rate limit"); expect(screen.getByText("older try: auth error")).toBeInTheDocument(); + expect(screen.queryByText("latest try: rate limit")).not.toBeInTheDocument(); expect(screen.getByText(i18n.t("common:taskInstance.stateReason"))).toBeInTheDocument(); }); }); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx index 65bd65f928d41..c35a3462dd784 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx @@ -27,7 +27,7 @@ import { useTaskInstanceServiceGetTaskInstanceTryDetails, } from "openapi/queries"; -import { Alert, ClipboardRoot, ClipboardIconButton } from "src/system-components"; +import { ClipboardRoot, ClipboardIconButton } from "src/system-components"; import { DagVersionDetails } from "src/components/DagVersionDetails"; import RenderedJsonField from "src/components/RenderedJsonField"; @@ -43,21 +43,7 @@ import { isStatePending, useAutoRefresh, useDurationFormat } from "src/utils"; import { BlockingDeps } from "./BlockingDeps"; import { ExtraLinks } from "./ExtraLinks"; import { TriggererInfo } from "./TriggererInfo"; - -type StateReasonSummary = { reason: string; status: "error" | "warning"; title: string }; - -// The reason is only cleared once the task next reaches RUNNING, so a cleared task still carries -// the previous attempt's reason. Both the banner and the per-try row look the state up here, so a -// state added to one surface cannot be forgotten on the other: it has to bring a title with it. -const STATE_REASON_DISPLAY = { - failed: { status: "error", titleKey: "failed" }, - up_for_retry: { status: "warning", titleKey: "upForRetry" }, -} as const satisfies Record; - -const stateReasonDisplay = (state: string | null | undefined) => - state === null || state === undefined - ? undefined - : (STATE_REASON_DISPLAY as Record)[state]; +import { stateReasonDisplay } from "./stateReason"; export const Details = () => { const { t: translate } = useTranslation(); @@ -130,24 +116,6 @@ export const Details = () => { return translate("common:none", { defaultValue: "None" }); }; - const stateReasonSummary = ((): StateReasonSummary | undefined => { - const reason = taskInstance?.state_reason; - const display = stateReasonDisplay(taskInstance?.state); - - if (reason === null || reason === undefined || taskInstance === undefined || display === undefined) { - return undefined; - } - - return { - reason, - status: display.status, - title: translate(`taskInstance.stateReasonSummary.${display.titleKey}`, { - totalTries: taskInstance.max_tries + 1, - tryNumber: taskInstance.try_number, - }), - }; - })(); - // Keyed off the selected try's own state, so an earlier failed try keeps its reason while the // current one is running again. const tryStateReason = @@ -173,16 +141,6 @@ export const Details = () => { return ( - {stateReasonSummary === undefined ? undefined : ( - - {stateReasonSummary.reason} - - )} {taskInstance === undefined || tryNumber === undefined || taskInstance.try_number <= 1 ? (
) : ( diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx index 34ab9a45cb939..d805f3ed0520b 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.test.tsx @@ -25,6 +25,7 @@ import type { TaskInstanceResponse } from "openapi/requests/types.gen"; import i18n from "src/i18n/config"; import { Wrapper } from "src/utils/Wrapper"; +import commonLocale from "../../../public/i18n/locales/en/common.json"; import { Header } from "./Header"; // Action buttons and note preview pull in mutation/permission wiring that is @@ -77,3 +78,56 @@ describe("Header", () => { expect(screen.queryByText(i18n.t("common:dagDetails.team"))).not.toBeInTheDocument(); }); }); + +const renderHeader = (overrides: Partial) => + render(
, { wrapper: Wrapper }); + +describe("Header state reason banner", () => { + // Without the bundle i18n.t() echoes the key, so the titles below would assert nothing. + beforeEach(() => { + i18n.addResourceBundle("en", "common", commonLocale, true, true); + }); + + it("does not render when there is no reason", () => { + renderHeader({ state: "failed", state_reason: null }); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + }); + + // Cleared only once the task next reaches RUNNING, so these states still carry a stale reason. + it.each(["queued", "running", "success", null] as const)( + "does not render for a %s task that still carries a reason", + (state) => { + renderHeader({ state, state_reason: "auth error, do not retry" }); + + expect(screen.queryByTestId("state-reason-alert")).not.toBeInTheDocument(); + }, + ); + + it.each([ + { maxTries: 2, state: "failed", titleKey: "failed", totalTries: 3, tryNumber: 3 }, + // Differing numbers are what make a swapped or off-by-one interpolation visible. + { maxTries: 3, state: "up_for_retry", titleKey: "upForRetry", totalTries: 4, tryNumber: 2 }, + ] as const)( + "titles the banner for a $state task", + ({ maxTries, state, titleKey, totalTries, tryNumber }) => { + renderHeader({ max_tries: maxTries, state, state_reason: "auth error", try_number: tryNumber }); + + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent( + i18n.t(`common:taskInstance.stateReasonSummary.${titleKey}`, { totalTries, tryNumber }), + ); + expect(screen.getByTestId("state-reason-alert")).toHaveTextContent("auth error"); + }, + ); + + // Chakra puts `status` in a generated class, so "not identical" is all that can be asserted. + it("styles a failed banner differently from an up_for_retry one", () => { + const { unmount } = renderHeader({ state: "failed", state_reason: "auth error" }); + const failedClass = screen.getByTestId("state-reason-alert").className; + + unmount(); + renderHeader({ state: "up_for_retry", state_reason: "auth error" }); + + expect(screen.getByTestId("state-reason-alert").className).not.toBe(failedClass); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx index 850c3b42e30fa..26d79acd9256a 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx @@ -24,6 +24,8 @@ import { MdOutlineTask } from "react-icons/md"; import type { TaskInstanceResponse } from "openapi/requests/types.gen"; +import { Alert } from "src/system-components"; + import { ClearTaskInstanceButton } from "src/components/Clear"; import ClearTaskInstanceDialog from "src/components/Clear/TaskInstance/ClearTaskInstanceDialog"; import { DagVersion } from "src/components/DagVersion"; @@ -37,6 +39,8 @@ import { useShowTeam } from "src/hooks/useShowTeam"; import { useTaskInstanceNote } from "src/queries/useTaskInstanceNote"; import { useDurationFormat } from "src/utils"; +import { stateReasonDisplay } from "./stateReason"; + export const Header = ({ taskInstance }: { readonly taskInstance: TaskInstanceResponse }) => { const { t: translate } = useTranslation(); const { formatElapsed, renderDuration } = useDurationFormat(); @@ -80,8 +84,24 @@ export const Header = ({ taskInstance }: { readonly taskInstance: TaskInstanceRe // Stable dialog state at header/page level const [clearOpen, setClearOpen] = useState(false); + // On the header, not the details tab, so it shows on every tab without duplicating the row. + const stateReasonDisplayed = stateReasonDisplay(taskInstance.state); + const stateReason = taskInstance.state_reason; + return ( + {stateReasonDisplayed === undefined || stateReason === null || stateReason === undefined ? undefined : ( + + {stateReason} + + )} diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts b/airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts new file mode 100644 index 0000000000000..4ed28ac8df364 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/stateReason.ts @@ -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. + */ + +// The header banner and the per-try row both look the state up here, so a state added to one +// surface cannot be forgotten on the other: it has to bring a title with it. +const STATE_REASON_DISPLAY = { + failed: { status: "error", titleKey: "failed" }, + up_for_retry: { status: "warning", titleKey: "upForRetry" }, +} as const satisfies Record; + +export const stateReasonDisplay = (state: string | null | undefined) => + state === null || state === undefined + ? undefined + : (STATE_REASON_DISPLAY as Record)[state]; diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index 9222414d82757..5b8bb12b43cf8 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -256,10 +256,20 @@ def test_should_include_state_reason(self, test_client, session): assert response.status_code == 200 assert response.json()["state_reason"] == "auth error, do not retry" + @pytest.fixture + def masked_secret(self): + """The masker is a cached process global, so drop the pattern again for the next test.""" + from airflow._shared.secrets_masker import _secrets_masker + + masker = _secrets_masker() + patterns, replacer = set(masker.patterns), masker.replacer + mask_secret("hunter2") + yield + masker.patterns, masker.replacer = patterns, replacer + @pytest.mark.enable_redact - def test_should_redact_secrets_in_state_reason(self, test_client, session): + def test_should_redact_secrets_in_state_reason(self, test_client, session, masked_secret): """A policy may compose the reason from an unredacted exception, so mask on the way out.""" - mask_secret("hunter2") self.create_task_instances( session, task_instances=[{"retry_reason": "auth: the token hunter2 expired"}] ) @@ -267,7 +277,7 @@ def test_should_redact_secrets_in_state_reason(self, test_client, session): "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context" ) assert response.status_code == 200 - assert "hunter2" not in response.json()["state_reason"] + assert response.json()["state_reason"] == "auth: the token *** expired" @conf_vars({("core", "multi_team"): "True"}) def test_should_include_team_name(self, test_client, session): @@ -2855,9 +2865,18 @@ def test_should_include_state_reason_from_history(self, test_client, session): assert response.status_code == 200 assert response.json()["state_reason"] == "auth error, do not retry" - @pytest.mark.enable_redact - def test_should_redact_secrets_in_state_reason_from_history(self, test_client, session): + @pytest.fixture + def masked_secret(self): + from airflow._shared.secrets_masker import _secrets_masker + + masker = _secrets_masker() + patterns, replacer = set(masker.patterns), masker.replacer mask_secret("hunter2") + yield + masker.patterns, masker.replacer = patterns, replacer + + @pytest.mark.enable_redact + def test_should_redact_secrets_in_state_reason_from_history(self, test_client, session, masked_secret): self.create_task_instances( session, task_instances=[{"state": State.SUCCESS, "retry_reason": "auth: the token hunter2 expired"}], @@ -2867,7 +2886,7 @@ def test_should_redact_secrets_in_state_reason_from_history(self, test_client, s "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries/1" ) assert response.status_code == 200 - assert "hunter2" not in response.json()["state_reason"] + assert response.json()["state_reason"] == "auth: the token *** expired" @pytest.mark.parametrize("try_number", [1, 2]) def test_should_respond_200_with_different_try_numbers(self, test_client, try_number, session): diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index 82ac27af6352d..8ca5887850242 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -2359,7 +2359,13 @@ class TaskInstanceHistoryResponse(BaseModel): executor: Annotated[str | None, Field(title="Executor")] executor_config: Annotated[str, Field(title="Executor Config")] dag_version: DagVersionResponse | None - state_reason: Annotated[str | None, Field(title="State Reason")] = None + state_reason: Annotated[ + str | None, + Field( + description="The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.", + title="State Reason", + ), + ] = None class TaskInstanceResponse(BaseModel): @@ -2402,7 +2408,13 @@ class TaskInstanceResponse(BaseModel): triggerer_job: JobResponse | None dag_version: DagVersionResponse | None team_name: Annotated[str | None, Field(title="Team Name")] = None - state_reason: Annotated[str | None, Field(title="State Reason")] = None + state_reason: Annotated[ + str | None, + Field( + description="The reason the task instance reached its current state, as recorded by a retry policy. May describe a previous attempt: it is cleared only when the task next starts running, so a task waiting to be retried or re-run can still carry the reason its last attempt ended.", + title="State Reason", + ), + ] = None class TaskResponse(BaseModel): diff --git a/providers/common/ai/docs/retry_policies.rst b/providers/common/ai/docs/retry_policies.rst index fa6439e3aa8e8..62e1c721d2517 100644 --- a/providers/common/ai/docs/retry_policies.rst +++ b/providers/common/ai/docs/retry_policies.rst @@ -133,9 +133,10 @@ When a task fails, either policy: ``retry_reason``, on a FAIL as well as a RETRY: ``: `` from ``LLMRetryPolicy``, or one line such as ``category=network confidence=0.91 threshold=0.60 action=retry delay=10s`` - from ``ClassifierRetryPolicy``. The REST API exposes it as ``state_reason`` - on a task instance and on each try, and the Task Instance page shows it - under **Reason for state**. + from ``ClassifierRetryPolicy``. From Airflow 3.4 the REST API exposes it as + ``state_reason`` on a task instance and on each try, and the Task Instance + page shows it under **Reason for state** while the task is failed or up for + retry. This classification call is a separate model request, made by the policy itself rather than by an operator -- it is not subject to an operator's 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 46f2fdae41128..3ca875c651750 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -30,7 +30,7 @@ from datetime import datetime, timedelta, timezone from itertools import product from pathlib import Path -from typing import TYPE_CHECKING, Annotated, Any, Literal +from typing import TYPE_CHECKING, Annotated, Any, Literal, cast from urllib.parse import quote import attrs @@ -1816,6 +1816,10 @@ def _evaluate_retry_policy( Returns ``None`` when no policy is configured so the caller falls through to the standard retry logic. """ + from dataclasses import replace + + from airflow.sdk._shared.secrets_masker import redact + policy = getattr(ti.task, "retry_policy", None) if policy is None: return None @@ -1828,6 +1832,9 @@ def _evaluate_retry_policy( context=context, ) if decision.reason: + # Mask here, where mask_secret() registered the value: the API server rendering this + # later has its own masker and does not know the worker's secrets. + decision = replace(decision, reason=cast("str", redact(decision.reason))) # Close the group so the retry policy decision is not hidden inside "Post Execute". log.info("::endgroup::") log.info("Retry policy decision", action=decision.action.value, reason=decision.reason) 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 198b6ec070b9c..b14e3e1bb696d 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 @@ -64,6 +64,7 @@ timezone, ) from airflow.sdk._shared.observability.metrics.base_stats_logger import StatsLogger +from airflow.sdk._shared.secrets_masker import _secrets_masker from airflow.sdk._shared.state import AssetScope, TaskScope from airflow.sdk.api.datamodels._generated import ( AssetProfile, @@ -1274,6 +1275,29 @@ def execute(self, context): assert msg.retry_reason == "z" * 500 +@pytest.mark.enable_redact +def test_retry_policy_reason_is_redacted_in_the_worker(create_runtime_ti, mock_supervisor_comms): + """The reason is masked where mask_secret() registered the value, not in the API server.""" + _secrets_masker().add_mask("hunter2", None) + + class _AlwaysFails(BaseOperator): + def execute(self, context): + raise RuntimeError("403 Forbidden: token hunter2 expired") + + class _EchoPolicy(RetryPolicy): + def evaluate(self, exception, try_number, max_tries, context=None): + return RetryDecision(action=RetryAction.FAIL, reason=f"auth: {exception}") + + task = _AlwaysFails(task_id="redacted_reason", retries=2, retry_policy=_EchoPolicy()) + ti = create_runtime_ti(task=task) + + state, msg, error = run(ti, ti.get_template_context(), mock.MagicMock()) + + assert state == TaskInstanceState.FAILED + assert isinstance(msg, TaskState) + assert msg.retry_reason == "auth: 403 Forbidden: token *** expired" + + def test_plain_retries_exhausted_has_no_reason(create_runtime_ti, mock_supervisor_comms): """Without a retry policy, exhausting the retry budget must not synthesize a reason."""