Standardize task run state conflict error response

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

@@ -29,6 +29,7 @@
 import structlog
 from cadwyn import VersionedAPIRouter
 from fastapi import Body, HTTPException, Query, Response, Security, status
+from fastapi.responses import JSONResponse
 from opentelemetry import trace
 from opentelemetry.trace import StatusCode
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
@@ -214,16 +215,15 @@ def ti_run(
             previous_state=previous_state,
         )
 
-        # TODO: Pass a RFC 9457 compliant error message in "detail" field
-        # https://datatracker.ietf.org/doc/html/rfc9457
-        # to provide more information about the error
-        # FastAPI will automatically convert this to a JSON response
-        # This might be added in FastAPI in https://github.com/fastapi/fastapi/issues/10370
-        raise HTTPException(
+        return JSONResponse(
             status_code=status.HTTP_409_CONFLICT,
-            detail={
+            media_type="application/problem+json",
+            content={
+                "type": "about:blank",
+                "title": "Conflict",
+                "status": status.HTTP_409_CONFLICT,
+                "detail": "TI was not in a state where it could be marked as running",
                 "reason": "invalid_state",
-                "message": "TI was not in a state where it could be marked as running",
                 "previous_state": previous_state,
             },
         )

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

@@ -51,9 +51,13 @@
     AddTeamNameField,
     AddVariableKeysEndpoint,
 )
+from airflow.api_fastapi.execution_api.versions.v2026_08_10 import (
+    UseProblemDetailsForTaskRunStateConflict,
+)
 
 bundle = VersionBundle(
     HeadVersion(),
+    Version("2026-08-10", UseProblemDetailsForTaskRunStateConflict),
     Version(
         "2026-06-30",
         AddVariableKeysEndpoint,

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

@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for
+
+
+class UseProblemDetailsForTaskRunStateConflict(VersionChange):
+    """Use RFC 9457 problem details for task run state conflict responses."""
+
+    description = __doc__
+    instructions_to_migrate_to_previous_version = ()
+
+    @convert_response_to_previous_version_for(
+        "/task-instances/{task_instance_id}/run",
+        ["PATCH"],
+        migrate_http_errors=True,
+    )
+    def restore_legacy_task_run_state_conflict_error(response: ResponseInfo) -> None:
+        """Restore the legacy FastAPI HTTPException error shape for older clients."""
+        if response.status_code != 409 or not isinstance(response.body, dict):
+            return
+        if response.body.get("reason") != "invalid_state" or response.body.get("previous_state") is None:
+            return
+
+        response.body = {
+            "detail": {
+                "reason": response.body["reason"],
+                "message": response.body.get("detail"),
+                "previous_state": response.body["previous_state"],
+            }
+        }
+        response.headers["content-type"] = "application/json"

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

@@ -855,12 +855,14 @@ def test_ti_run_state_conflict_if_not_queued(
         )
 
         assert response.status_code == 409
+        assert response.headers["content-type"] == "application/problem+json"
         assert response.json() == {
-            "detail": {
-                "message": "TI was not in a state where it could be marked as running",
-                "previous_state": initial_ti_state,
-                "reason": "invalid_state",
-            }
+            "type": "about:blank",
+            "title": "Conflict",
+            "status": 409,
+            "detail": "TI was not in a state where it could be marked as running",
+            "previous_state": initial_ti_state,
+            "reason": "invalid_state",
         }
 
         assert session.scalar(select(TaskInstance.state).where(TaskInstance.id == ti.id)) == initial_ti_state

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

@@ -0,0 +1,76 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow.utils.state import State
+
+from tests_common.test_utils.db import (
+    clear_db_assets,
+    clear_db_dags,
+    clear_db_logs,
+    clear_db_runs,
+    clear_db_serialized_dags,
+)
+
+pytestmark = pytest.mark.db_test
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+    clear_db_logs()
+    clear_db_runs()
+    clear_db_serialized_dags()
+    clear_db_dags()
+    clear_db_assets()
+    yield
+    clear_db_logs()
+    clear_db_runs()
+    clear_db_serialized_dags()
+    clear_db_dags()
+    clear_db_assets()
+
+
+def test_task_run_state_conflict_error_uses_legacy_shape_for_older_versions(
+    client, session, create_task_instance
+):
+    client.headers["Airflow-API-Version"] = "2026-06-30"
+    ti = create_task_instance(task_id="test_ti_run_state_conflict_legacy_error", state=State.SUCCESS)
+    session.commit()
+
+    response = client.patch(
+        f"/execution/task-instances/{ti.id}/run",
+        json={
+            "state": "running",
+            "hostname": "random-hostname",
+            "unixname": "random-unixname",
+            "pid": 100,
+            "start_date": "2024-10-31T12:00:00Z",
+        },
+    )
+
+    assert response.status_code == 409
+    assert response.headers["content-type"] == "application/json"
+    assert response.json() == {
+        "detail": {
+            "message": "TI was not in a state where it could be marked as running",
+            "previous_state": State.SUCCESS,
+            "reason": "invalid_state",
+        }
+    }

task-sdk/src/airflow/sdk/api/client.py

@@ -1325,21 +1325,28 @@ def from_response(cls, response: httpx.Response) -> ServerResponseError | None:
         if not (400 <= response.status_code < 600):
             return None
 
-        if response.headers.get("content-type") != "application/json":
+        content_type = response.headers.get("content-type", "").split(";")[0]
+        if content_type != "application/json" and not content_type.endswith("+json"):
             return None
 
         detail: list[RemoteValidationError] | dict[str, Any] | None = None
         try:
-            body = _ErrorBody.model_validate_json(response.read())
+            body = msgspec.json.decode(response.read())
 
-            if isinstance(body.detail, list):
-                detail = body.detail
-                msg = "Remote server returned validation error"
-            elif isinstance(body.detail, dict):
-                detail = body.detail
+            if isinstance(body, dict) and {"type", "title", "status"}.issubset(body):
+                detail = body
                 msg = "Server returned error"
             else:
-                msg = body.detail or "Un-parseable error"
+                body = _ErrorBody.model_validate(body)
+
+                if isinstance(body.detail, list):
+                    detail = body.detail
+                    msg = "Remote server returned validation error"
+                elif isinstance(body.detail, dict):
+                    detail = body.detail
+                    msg = "Server returned error"
+                else:
+                    msg = body.detail or "Un-parseable error"
         except Exception:
             try:
                 detail = msgspec.json.decode(response.content)

task-sdk/tests/task_sdk/api/test_client.py

@@ -363,12 +363,14 @@ def handle_request(request: httpx.Request) -> httpx.Response:
             if request.url.path == f"/task-instances/{ti_id}/run":
                 return httpx.Response(
                     409,
+                    headers={"content-type": "application/problem+json"},
                     json={
-                        "detail": {
-                            "reason": "invalid_state",
-                            "message": "TI was not in a state where it could be marked as running",
-                            "previous_state": "running",
-                        }
+                        "type": "about:blank",
+                        "title": "Conflict",
+                        "status": 409,
+                        "detail": "TI was not in a state where it could be marked as running",
+                        "reason": "invalid_state",
+                        "previous_state": "running",
                     },
                 )
             return httpx.Response(status_code=204)
@@ -387,12 +389,14 @@ def handle_request(request: httpx.Request) -> httpx.Response:
             if request.url.path == f"/task-instances/{ti_id}/run":
                 return httpx.Response(
                     409,
+                    headers={"content-type": "application/problem+json"},
                     json={
-                        "detail": {
-                            "reason": "invalid_state",
-                            "message": "TI was not in a state where it could be marked as running",
-                            "previous_state": previous_state,
-                        }
+                        "type": "about:blank",
+                        "title": "Conflict",
+                        "status": 409,
+                        "detail": "TI was not in a state where it could be marked as running",
+                        "reason": "invalid_state",
+                        "previous_state": previous_state,
                     },
                 )
             return httpx.Response(status_code=204)

task-sdk/tests/task_sdk/execution_time/test_supervisor.py

@@ -959,12 +959,14 @@ def handle_request(request: httpx.Request) -> httpx.Response:
             if request.url.path == f"/task-instances/{ti_id}/run":
                 return httpx.Response(
                     409,
+                    headers={"content-type": "application/problem+json"},
                     json={
-                        "detail": {
-                            "reason": "invalid_state",
-                            "message": "TI was not in a state where it could be marked as running",
-                            "previous_state": "running",
-                        }
+                        "type": "about:blank",
+                        "title": "Conflict",
+                        "status": 409,
+                        "detail": "TI was not in a state where it could be marked as running",
+                        "reason": "invalid_state",
+                        "previous_state": "running",
                     },
                 )
             return httpx.Response(status_code=204)