@@ -520,7 +520,7 @@ not yet use this client. Provisioning a token file alone does not switch that co
 HTTP or remove its need for database access.
 
 In-memory sessions from workload identity
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 Processor integrations can use ``WorkloadIdentitySessionTokenProvider`` to obtain and renew
 ``dag_processor_session`` credentials through ``POST /execution/dag-processor/session-token``.
@@ -545,14 +545,20 @@ Use your identity provider's exact issuer, HTTPS JWKS URL, and signing algorithm
 ``ES256``, or ``EdDSA``). Use a dedicated audience, distinct from ``[execution_api] jwt_audience``.
 The API server verifies the signature, issuer, audience, issued-at time, expiry, and any not-before
 claim, using ``[api_auth] jwt_leeway`` for clock skew. Only exact subjects listed in ``grants`` can
-obtain a session. Every granted bundle must be configured on the API server, which otherwise refuses
-to start. This option is empty by default, disabling the exchange.
+obtain a session. If a grant names a Dag bundle that the API server does not have configured, the
+API server logs an error at startup and refuses that subject's exchanges with HTTP 403. This option
+is empty by default, disabling the exchange.
 Apply the same trust configuration to all API-server replicas and restart them after changes.
 
+The option trusts a single issuer. Dag processors whose identities come from different issuers, such
+as separate Kubernetes clusters, cannot all obtain sessions from the same API server.
+
 The JWKS endpoint must accept unauthenticated HTTPS GET requests from the Airflow API server.
-The fetcher sends no authentication credentials and uses HTTPX's default certificate trust,
-including ``SSL_CERT_FILE`` or ``SSL_CERT_DIR`` when configured. These environment settings affect
-other HTTPX clients in the API-server process; there is no per-issuer CA or authentication option.
+The fetcher sends no authentication credentials and uses HTTPX's default certificate trust.
+``SSL_CERT_FILE`` or ``SSL_CERT_DIR``, when set, replace the default CA bundle rather than adding
+to it, and they apply to every HTTPX client in the API-server process. A bundle that holds only a
+private CA cuts those clients off from public roots, so include the roots they still need. There is
+no per-issuer CA or authentication option.
 Authenticated JWKS endpoints, including Kubernetes API-server discovery protected by service-account
 authentication, are not supported by this implementation. For those clusters, serve the public JWKS
 through a suitable HTTPS endpoint before enabling the exchange. See Kubernetes'
@@ -586,7 +592,8 @@ The client does not wait automatically. This prevents two processes sharing an i
 active Jobs at the same time.
 
 The processor integration supplies a callable that reads the current platform identity on each
-exchange. For example:
+exchange. Any exception the callable raises is re-raised as ``WorkloadIdentityUnavailableError``, an
+``OSError``, so a failed optional renewal leaves the valid Job credential in use. For example:
 
 .. code-block:: python
 
@@ -613,8 +620,9 @@ session JWT to disk and does not fall back to a session-token file. ``token_prov
 ``token_file`` are mutually exclusive. HTTPS is required for the exchange, and redirects are rejected.
 The exchange uses the same API TLS configuration and request timeout as the Task SDK client,
 including private certificate authorities and client certificates. Registration retries transient
-exchange failures using the configured Execution API retry policy; optional early renewal makes
-one bounded attempt so it does not delay requests using a valid Job credential.
+exchange failures using the configured Execution API retry policy, waiting at least the
+``Retry-After`` interval of an HTTP 503 before the next attempt. Optional early renewal makes one
+bounded attempt so it does not delay requests using a valid Job credential.
 
 Session expiry is capped by both the external identity's remaining lifetime and
 ``[execution_api] jwt_expiration_time``. Clock-skew leeway never extends this cap, and an identity
@@ -631,12 +639,12 @@ unknown signing key. Fetches, including failed attempts, are limited to one per
 API-server process. Publish new keys before signing with them and retain old keys while their tokens
 remain valid.
 
-If a required signing key cannot be fetched, the exchange returns HTTP 503 with ``Retry-After``,
-allowing registration to retry. A cached matching key remains usable during an outage. Invalid
-identities and keys absent from a successfully fetched keyset return HTTP 401.
+If a required signing key cannot be fetched, the exchange returns HTTP 503 with ``Retry-After``, and
+registration waits at least that long before retrying. A cached matching key remains usable during
+an outage. Invalid identities and keys absent from a successfully fetched key set return HTTP 401.
 
 File-based provisioning
-~~~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^^^
 
 On a trusted host with the API signing key, provision a session token for the bundles
 the client may access::
@@ -652,7 +660,7 @@ provisioner with the same token file and bundles keeps that session, so running
 their Jobs.
 
 Credential permissions and renewal
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 The three credentials have different issuers and permissions:
 
 
@@ -163,6 +163,9 @@ async def fetch_jwks(self, *, force: bool = False) -> None:
                 return
 
             try:
+                # PyJWKSet.from_dict raises AttributeError for a JSON array or scalar.
+                if not isinstance(data, dict):
+                    raise TypeError(f"JWKS must be a JSON object, not {type(data).__name__}")
                 keyset = jwt.PyJWKSet.from_dict(data)
             except (jwt.PyJWKError, jwt.PyJWKSetError, KeyError, TypeError, ValueError):
                 if not self.raise_on_fetch_error:
 
@@ -112,12 +112,12 @@ async def lifespan(app: FastAPI, registry: svcs.Registry):
 
     # InProcessExecutionAPI stubs out the identity verifier: don't create one in that case.
     if DagProcessorIdentityVerifier in registry:
-        app.state.dag_processor_identity_verifier = None
         yield
         return
 
     async with create_dag_processor_identity_verifier() as verifier:
-        app.state.dag_processor_identity_verifier = verifier
+        # None when the exchange is disabled; the route answers 404 then.
+        registry.register_value(DagProcessorIdentityVerifier, verifier)
         yield
 
 
 
@@ -116,7 +116,7 @@ def app(self):
             # Auth dependency overrides bypass validation, so the stub skips validator setup in lifespan().
             registry.register_value(JWTValidator, None)
             # No external Dag processor can reach this app, so it must not load trust settings meant for
-            # API servers, whose Dag bundle checks can fail in other components.
+            # API servers, which can fail to load in components configured differently.
             registry.register_value(DagProcessorIdentityVerifier, None)
             _configure_in_process_auth(self._app, triggerer_id=self.triggerer_id)
 
 
@@ -24,7 +24,7 @@
 import jwt
 import svcs
 from cadwyn import VersionedAPIRouter
-from fastapi import Depends, HTTPException, Request, Response, status
+from fastapi import Depends, HTTPException, Response, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import ValidationError
 
@@ -48,10 +48,10 @@
 
 
 async def require_workload_identity(
-    request: Request,
     credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(identity_bearer)],
+    services: svcs.Container = DepContainer,
 ) -> DagProcessorIdentity:
-    verifier: DagProcessorIdentityVerifier | None = request.app.state.dag_processor_identity_verifier
+    verifier: DagProcessorIdentityVerifier | None = await services.aget(DagProcessorIdentityVerifier)
     if verifier is None:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "Workload identity provisioning is not enabled")
     if credentials is None:
@@ -66,22 +66,20 @@ async def require_workload_identity(
             "Workload identity signing keys are unavailable",
             headers={"Retry-After": str(verifier.jwks.refresh_retry_interval_secs)},
         ) from None
-    except (jwt.PyJWTError, ValidationError, TypeError, OverflowError):
+    except (jwt.PyJWTError, ValidationError):
         raise HTTPException(
             status.HTTP_401_UNAUTHORIZED, "Invalid workload identity", headers={"WWW-Authenticate": "Bearer"}
         ) from None
-    except WorkloadIdentityNotAllowed:
-        raise HTTPException(
-            status.HTTP_403_FORBIDDEN, "Workload identity is not granted any Dag bundles"
-        ) from None
+    except WorkloadIdentityNotAllowed as error:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, str(error)) from None
 
 
 @router.post(
     "/dag-processor/session-token",
     responses=create_openapi_http_exception_doc(
         [
             (status.HTTP_401_UNAUTHORIZED, "The workload identity is missing, invalid, or expired"),
-            (status.HTTP_403_FORBIDDEN, "The identity is not authorized as a Dag processor"),
+            (status.HTTP_403_FORBIDDEN, "The identity has no usable Dag bundle grant"),
             (status.HTTP_404_NOT_FOUND, "Workload identity provisioning is disabled"),
             (status.HTTP_503_SERVICE_UNAVAILABLE, "The identity provider's signing keys are unavailable"),
         ]
@@ -92,11 +90,21 @@ def exchange_session_token(
     identity: Annotated[DagProcessorIdentity, Depends(require_workload_identity)],
     services: svcs.Container = DepContainer,
 ) -> DagProcessorSessionTokenResponse:
+    """
+    Exchange an external workload identity for a Dag processor session credential.
+
+    The bearer token is an OIDC identity issued by the deployment platform, not an Airflow token. The
+    API server picks the Dag bundle grants from its own configuration, and the session cannot outlive
+    the identity. While the identity provider's signing keys cannot be fetched, the exchange returns
+    503 with ``Retry-After``.
+    """
     generator: JWTGenerator = services.get(JWTGenerator)
     try:
         lifetime = get_token_lifetime(generator, identity.expires_at, "Workload identity has expired")
     except ExpiredDagProcessorToken as error:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(error)) from None
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED, str(error), headers={"WWW-Authenticate": "Bearer"}
+        ) from None
     token = generate_dag_processor_session_token(
         generator, session_id=identity.session_id, bundle_names=identity.bundle_names, valid_for=lifetime
     )
 
@@ -23,18 +23,21 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
-from typing import Annotated, Any, Literal
+from typing import Annotated, Literal
 from uuid import NAMESPACE_URL, UUID, uuid5
 
 import httpx
 import jwt
+import structlog
 from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator
 
 from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
 from airflow.api_fastapi.execution_api.datamodels.token import DagBundleGrant
 from airflow.configuration import conf
 from airflow.dag_processing.bundles.manager import get_configured_bundle_names
 
+log = structlog.get_logger(logger_name=__name__)
+
 NonemptyString = Annotated[str, Field(min_length=1)]
 
 
@@ -84,10 +87,20 @@ class DagProcessorIdentityVerifier:
     def __init__(self, config: WorkloadIdentityConfig, *, client: httpx.AsyncClient):
         if config.audience in conf.getlist("execution_api", "jwt_audience"):
             raise ValueError("Workload identity must use a dedicated audience")
-        if unknown := frozenset().union(*config.grants.values()) - get_configured_bundle_names():
-            raise ValueError(
+        configured = get_configured_bundle_names()
+        # Not fatal: a renamed bundle must not stop the whole API server, only the subjects granted it.
+        self._unconfigured_grants = {
+            subject: missing
+            for subject, bundles in config.grants.items()
+            if (missing := bundles - configured)
+        }
+        if self._unconfigured_grants:
+            log.error(
                 "[execution_api] dag_processor_workload_identity grants Dag bundles missing from "
-                f"[dag_processor] dag_bundle_config_list: {sorted(unknown)}"
+                "[dag_processor] dag_bundle_config_list; these subjects cannot obtain a session",
+                unconfigured_grants={
+                    subject: sorted(missing) for subject, missing in self._unconfigured_grants.items()
+                },
             )
         self.config = config
         self.jwks = JWKS(
@@ -102,11 +115,20 @@ def __init__(self, config: WorkloadIdentityConfig, *, client: httpx.AsyncClient)
         )
 
     async def authenticate(self, token: str) -> DagProcessorIdentity:
-        claims = await self.validator.avalidated_claims(token)
+        try:
+            claims = await self.validator.avalidated_claims(token)
+        except (TypeError, OverflowError) as error:
+            # PyJWT before 2.15 raises these for non-numeric or infinite time claims instead of DecodeError.
+            log.warning("Rejecting a workload identity with malformed time claims", exc_info=True)
+            raise jwt.InvalidTokenError("Workload identity has malformed time claims") from error
         identity = _IdentityClaims.model_validate(claims)
         if not (grants := self.config.grants.get(identity.sub)):
-            raise WorkloadIdentityNotAllowed("Workload identity is not authorized as a Dag processor")
-        instance_id: Any = claims
+            raise WorkloadIdentityNotAllowed("Workload identity is not granted any Dag bundles")
+        if missing := self._unconfigured_grants.get(identity.sub):
+            raise WorkloadIdentityNotAllowed(
+                f"Workload identity is granted Dag bundles this API server does not have: {sorted(missing)}"
+            )
+        instance_id: object = claims
         for key in self.config.instance_id_claim:
             if not isinstance(instance_id, dict) or key not in instance_id:
                 raise jwt.InvalidTokenError("Workload identity lacks its configured instance claim")
 
@@ -2431,10 +2431,14 @@ execution_api:
 
         Requires issuer, HTTPS jwks_url, a dedicated audience, an instance_id_claim path
         (a list of keys into the signed identity claims), and grants mapping exact subjects
-        to allowed Dag bundle names. The API server refuses to start if a grant names a Dag
-        bundle it does not have configured. The optional algorithm defaults to RS256; ES256 and
-        EdDSA are also supported. Use a signed, per-replica instance claim: a service
-        account shared by replicas is not sufficient on its own.
+        to allowed Dag bundle names. If a grant names a Dag bundle the API server does not have
+        configured, the API server logs an error at startup and refuses that subject's exchanges.
+        The optional algorithm defaults to RS256; ES256 and EdDSA are also supported. Use a
+        signed, per-replica instance claim: a service account shared by replicas is not
+        sufficient on its own.
+
+        Only one issuer is supported. Dag processors whose identities come from different
+        issuers, such as separate Kubernetes clusters, cannot all use the same API server.
         The JWKS endpoint must accept unauthenticated HTTPS requests and have a certificate
         trusted by the API server's HTTPX environment. Per-issuer CA and JWKS authentication
         settings are not supported.
 
@@ -25,10 +25,37 @@
 from typing import Protocol
 
 import httpx2
+from tenacity import RetryCallState, retry, retry_if_exception, stop_after_attempt, wait_random_exponential
 
 from airflow.api_fastapi.execution_api.datamodels.job import DagProcessorSessionTokenResponse
 from airflow.api_fastapi.execution_api.versions import bundle
-from airflow.sdk.api.client import Client  # noqa: SDK001 - reuse Execution API TLS and retry settings
+from airflow.sdk.api.client import (  # noqa: SDK001 - reuse Execution API TLS and retry settings
+    API_RETRIES,
+    API_RETRY_WAIT_MAX,
+    API_RETRY_WAIT_MIN,
+    Client,
+    _log_and_trace_retry,
+    _should_retry_api_request,
+)
+
+_backoff = wait_random_exponential(min=API_RETRY_WAIT_MIN, max=API_RETRY_WAIT_MAX)
+
+
+class WorkloadIdentityUnavailableError(OSError):
+    """The identity supplier failed; the client handles it like an unreadable credential file."""
+
+
+def _wait_for_retry_after(retry_state: RetryCallState) -> float:
+    """Back off like the SDK client, but never retry a 503 before its ``Retry-After`` elapses."""
+    wait = _backoff(retry_state)
+    error = retry_state.outcome.exception() if retry_state.outcome else None
+    if not isinstance(error, httpx2.HTTPStatusError) or error.response.status_code != 503:
+        return wait
+    try:
+        retry_after = float(error.response.headers["Retry-After"])
+    except (KeyError, ValueError):  # Missing, or an HTTP-date rather than seconds.
+        return wait
+    return max(wait, min(retry_after, API_RETRY_WAIT_MAX))
 
 
 class SessionTokenProvider(Protocol):
@@ -64,14 +91,28 @@ def _update_auth(self, response: httpx2.Response) -> None:
         # Runtime refresh headers must not replace the external workload identity.
         pass
 
+    # The server refetches signing keys at most every 10 seconds; the SDK's backoff alone often retries
+    # every attempt inside that cooldown.
+    @retry(
+        retry=retry_if_exception(_should_retry_api_request),
+        stop=stop_after_attempt(API_RETRIES),
+        wait=_wait_for_retry_after,
+        before_sleep=_log_and_trace_retry,
+        reraise=True,
+    )
+    def _request_with_retry(self, *args, **kwargs):
+        return self._request_without_retry(*args, **kwargs)
+
 
 class WorkloadIdentitySessionTokenProvider:
     """
     Exchange deployment identity for a session JWT, retaining the JWT only in memory.
 
     The identity supplier must return a current, renewable deployment credential. For example, it may
-    read a projected service-account token that the platform rotates. The owning Dag processor client
-    closes this provider; do not share it between clients.
+    read a projected service-account token that the platform rotates. Any exception it raises is
+    re-raised as ``WorkloadIdentityUnavailableError``, so a failed early renewal keeps the valid Job
+    credential in use. The owning Dag processor client closes this provider; do not share it between
+    clients.
     """
 
     def __init__(
@@ -93,7 +134,10 @@ def get_token(
     ) -> str:
         if not force_refresh and self._token is not None and monotonic() < self._renew_at:
             return self._token
-        identity = self._identity_token().strip()
+        try:
+            identity = self._identity_token().strip()
+        except Exception as error:
+            raise WorkloadIdentityUnavailableError("Unable to obtain the workload identity") from error
         if not identity:
             raise ValueError("Workload identity credential is empty")
         started_at = monotonic()