@@ -519,6 +519,125 @@ API without holding its signing key. The standard ``airflow dag-processor`` comm
 not yet use this client. Provisioning a token file alone does not switch that command to
 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``.
+The provider retains these credentials only in memory. It requires a renewable, signed workload
+identity from the deployment platform; Airflow does not create that identity or require the
+processor to hold an Airflow signing key.
+
+Configure the API server's ``[execution_api] dag_processor_workload_identity`` option with JSON::
+
+    {
+      "issuer": "https://identity.example/",
+      "jwks_url": "https://identity.example/openid/v1/jwks",
+      "audience": "airflow-dag-processor-bootstrap",
+      "algorithm": "RS256",
+      "instance_id_claim": ["kubernetes.io", "pod", "uid"],
+      "grants": {
+        "system:serviceaccount:airflow:processor": ["dags-folder"]
+      }
+    }
+
+Use your identity provider's exact issuer, HTTPS JWKS URL, and signing algorithm (``RS256``,
+``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, and every granted bundle must be configured on the API server. 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 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.
+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'
+`service account issuer discovery documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/#service-account-issuer-discovery>`_.
+The processor-to-Airflow TLS settings described below do not configure this outbound JWKS connection.
+
+``instance_id_claim`` is a path through signed claims to a nonempty string identifying one
+processor replica. A shared service-account subject alone is insufficient: replicas must have
+different instance identities. Run one processor per identity. The session ID stays stable across
+identity-token renewal and API-server restarts. Changes to the issuer, audience, subject, instance
+claim, instance value, or bundle grants produce a different session. Request bodies and caller
+hostnames cannot select the session or its grants.
+
+Changing grants while API-server replicas serve different configurations can repeatedly retire
+processor registrations, because those replicas issue different session IDs for the same identity.
+Stop the affected processors, finish deploying identical trust configuration to every API-server
+replica, and then restart the processors.
+
+For Kubernetes, the example uses the Pod UID in a
+`bound service-account token <https://kubernetes.io/docs/reference/access-authn-authz/service-accounts-admin/#bound-service-account-token-volume-mechanism>`_.
+Project a token with the dedicated audience and mount the projection directory read-only in the
+processor container. Configure the projection's ``expirationSeconds`` and allow Kubernetes to
+rotate it. See `serviceAccountToken projected volumes
+<https://kubernetes.io/docs/concepts/storage/projected-volumes/#serviceaccounttoken>`_.
+
+A container restart within the same Pod keeps the Pod UID and therefore the session ID.
+A replacement process cannot register while the previous Job is still alive. The client raises
+``DagProcessorJobAlreadyRunning``; the integration must retry registration with backoff until the
+previous Job completes or its heartbeat becomes stale under ``[dag_processor] health_check_threshold``.
+The client does not wait automatically. This prevents two processes sharing an identity from owning
+active Jobs at the same time.
+
+The processor integration supplies a callable that reads the current platform identity on each
+exchange. For example:
+
+.. code-block:: python
+
+    from pathlib import Path
+
+    from airflow.dag_processing.api_client import DagProcessorAPIClient
+    from airflow.dag_processing.credentials import WorkloadIdentitySessionTokenProvider
+
+    identity_path = Path("/var/run/airflow-identity/token")
+    provider = WorkloadIdentitySessionTokenProvider(
+        base_url="https://airflow.example/execution/",
+        identity_token=identity_path.read_text,
+    )
+    processor = DagProcessorAPIClient(
+        base_url="https://airflow.example/execution/",
+        token_provider=provider,
+        hostname="processor-1",
+    )
+
+Use this client for the integration's Job lifecycle and close it during shutdown; closing it also
+closes the provider and clears its session cache. The file in this example contains the platform's
+projected identity, not an Airflow session credential. The provider never writes the exchanged
+session JWT to disk and does not fall back to a session-token file. ``token_provider`` and
+``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.
+
+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
+with less than one second remaining cannot obtain a session. Renewal rereads the platform identity
+and retains the same session and Job. During an identity-provider or API outage, the client can continue using
+an already-issued Job credential until it expires; it cannot renew without a valid identity.
+Removing a grant stops future exchanges after API-server configuration is reloaded. Credentials
+already issued remain subject to their expiry and Job ownership checks.
+
+Verification uses the issuer's JWKS and does not perform online token introspection. In particular,
+it does not check whether a Kubernetes Pod still exists. A previously issued identity may remain
+usable until expiry. The JWKS cache refreshes hourly and also refreshes when an identity names an
+unknown signing key. Fetches, including failed attempts, are limited to one per ten seconds 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.
+
+File-based provisioning
+~~~~~~~~~~~~~~~~~~~~~~~
+
 On a trusted host with the API signing key, provision a session token for the bundles
 the client may access::
 
@@ -532,9 +651,12 @@ token replacement stays visible. Each processor process needs its own session. R
 provisioner with the same token file and bundles keeps that session, so running processors keep
 their Jobs.
 
+Credential permissions and renewal
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
 The three credentials have different issuers and permissions:
 
-- ``dag_processor_session`` is issued by trusted provisioning. It can only register a
+- ``dag_processor_session`` is issued by the identity exchange or trusted file provisioning. It can only register a
   Job or renew that Job's credential through ``POST /jobs``.
 - ``dag_processor`` is returned by Job registration. It can heartbeat and complete that
   Job, read Connections and Variables for a granted bundle, and exchange a parsing
@@ -625,6 +747,9 @@ All JWT-related configuration parameters:
    * - ``[execution_api] jwt_audience``
      - ``urn:airflow.apache.org:task``
      - Audience claim for Execution API tokens.
+   * - ``[execution_api] dag_processor_workload_identity``
+     - Empty (disabled)
+     - JSON trust settings and subject-to-bundle grants for workload identity session provisioning.
 
 .. important::
 
 
@@ -537,6 +537,12 @@ potentially still executes with direct database access in the Dag File Processor
      and callback credentials, or ``DagProcessorSessionClaims``, ``DagProcessorClaims``, or
      ``DagParseClaims`` for processor credentials. Unknown scopes, invalid subjects, and missing
      required scope-specific claims are rejected before any route handler runs.
+   * **Workload identity provisioning.** The optional Dag processor session exchange validates
+     deployment-issued JWTs against separately configured issuer, audience, and JWKS settings.
+     Bundle grants come from the API server configuration. Exchanged sessions stay in the client's
+     memory, but workload identity and runtime credentials remain accessible to code with access
+     to that process. This does not add isolation between the processor and user code; see
+     :doc:`jwt_token_authentication` for configuration and expiry limitations.
 
 **Token signing key might be a shared secret**
    In symmetric key mode (``[api_auth] jwt_secret``), the same secret key is used to both generate and
 
@@ -16,12 +16,14 @@
 # under the License.
 from __future__ import annotations
 
+import asyncio
 import json
 import os
 import time
 import uuid
 from base64 import urlsafe_b64encode
 from collections.abc import Callable, Sequence
+from contextlib import nullcontext
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Literal, overload
 
@@ -45,6 +47,7 @@
 __all__ = [
     "InvalidClaimError",
     "JWKS",
+    "JWKSUnavailableError",
     "JWTGenerator",
     "JWTValidator",
     "generate_private_key",
@@ -99,9 +102,19 @@ def _guess_best_algorithm(key: AllowedPrivateKeys):
     raise ValueError(f"Unknown key object {type(key)}")
 
 
+class JWKSUnavailableError(RuntimeError):
+    """The required signing key could not be loaded because a JWKS fetch failed."""
+
+
 @attrs.define(repr=False)
 class JWKS:
-    """A class to fetch and sync a set of JSON Web Keys."""
+    """
+    Fetch and cache a set of JSON Web Keys.
+
+    With ``raise_on_fetch_error``, missing keys raise ``JWKSUnavailableError`` after a failed
+    fetch. Cached matching keys remain usable, and a successfully fetched set with no matching key
+    still raises ``KeyError``.
+    """
 
     url: str
     fetched_at: float = 0
@@ -113,6 +126,10 @@ class JWKS:
     refresh_jwks: bool = True
     refresh_interval_secs: int = 3600
     refresh_retry_interval_secs: int = 10
+    refresh_on_unknown_kid: bool = False
+    raise_on_fetch_error: bool = False
+    _fetch_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, init=False)
+    _fetch_failed: bool = attrs.field(default=False, init=False)
 
     def __repr__(self) -> str:
         return f"JWKS(url={self.url}, fetched_at={self.fetched_at})"
@@ -128,19 +145,30 @@ def from_private_key(cls, *keys: AllowedPrivateKeys | tuple[AllowedPrivateKeys,
         obj._jwks = jwt.PyJWKSet(keyset)
         return obj
 
-    async def fetch_jwks(self) -> None:
-        if not self._should_fetch_jwks():
-            return
-        if self.url.startswith("http"):
-            data = await self._fetch_remote_jwks()
-        else:
-            data = self._fetch_local_jwks()
-
-        if not data:
-            return
-
-        self._jwks = jwt.PyJWKSet.from_dict(data)
-        log.debug("Fetched JWKS", url=self.url, keys=len(self._jwks.keys))
+    async def fetch_jwks(self, *, force: bool = False) -> None:
+        lock = self._fetch_lock if self.refresh_on_unknown_kid else nullcontext()
+        async with lock:
+            if not self._should_fetch_jwks(force=force):
+                return
+            self._fetch_failed = True
+            if self.url.startswith("http"):
+                data = await self._fetch_remote_jwks()
+            else:
+                data = self._fetch_local_jwks()
+
+            if not data:
+                return
+
+            try:
+                keyset = jwt.PyJWKSet.from_dict(data)
+            except (jwt.PyJWKError, jwt.PyJWKSetError, KeyError, TypeError, ValueError):
+                if not self.raise_on_fetch_error:
+                    raise
+                log.warning("Unable to load JWKS response", url=self.url, exc_info=True)
+                return
+            self._jwks = keyset
+            self._fetch_failed = False
+            log.debug("Fetched JWKS", url=self.url, keys=len(self._jwks.keys))
 
     async def _fetch_remote_jwks(self) -> dict[str, Any] | None:
         try:
@@ -151,7 +179,7 @@ async def _fetch_remote_jwks(self) -> dict[str, Any] | None:
             )
             if TYPE_CHECKING:
                 assert self.url
-            self.last_fetch_attempt_at = int(time.monotonic())
+            self.last_fetch_attempt_at = time.monotonic()
             response = await self.client.get(self.url)
             response.raise_for_status()
             self.fetched_at = int(time.monotonic())
@@ -172,7 +200,7 @@ def _fetch_local_jwks(self) -> dict[str, Any] | None:
             log.exception("Failed to read local JWKS", url=self.url)
             return None
 
-    def _should_fetch_jwks(self) -> bool:
+    def _should_fetch_jwks(self, *, force: bool = False) -> bool:
         """
         Check if we need to fetch the JWKS based on the last fetch time and the refresh interval.
 
@@ -188,8 +216,16 @@ def _should_fetch_jwks(self) -> bool:
         # For remote fetches we check if the JWKS is not loaded (fetched_at = 0) or if the last fetch was more than
         # refresh_interval_secs ago and the last fetch attempt was more than refresh_retry_interval_secs ago
         now = time.monotonic()
+        # Unknown keys are untrusted input; bound fetches even for an empty cache or a failed refresh.
+        if (
+            self.refresh_on_unknown_kid
+            and self.last_fetch_attempt_at
+            and now - self.last_fetch_attempt_at < self.refresh_retry_interval_secs
+        ):
+            return False
         return self.refresh_jwks and (
-            not self._jwks
+            force
+            or not self._jwks
             or (
                 self.fetched_at == 0
                 or (
@@ -204,9 +240,19 @@ async def get_key(self, kid: str) -> jwt.PyJWK:
         await self.fetch_jwks()
 
         if self._jwks:
-            return self._jwks[kid]
+            try:
+                return self._jwks[kid]
+            except KeyError:
+                if self.refresh_on_unknown_kid:
+                    await self.fetch_jwks(force=True)
+                    try:
+                        return self._jwks[kid]
+                    except KeyError:
+                        pass
+
+        if self.raise_on_fetch_error and self._fetch_failed:
+            raise JWKSUnavailableError("Unable to load the required signing key")
 
-        # It didn't load!
         raise KeyError(f"Key ID {kid} not found in keyset")
 
     def status(self):
 
@@ -41,6 +41,7 @@
     get_signing_args,
 )
 from airflow.api_fastapi.execution_api.in_process import InProcessExecutionAPI
+from airflow.api_fastapi.execution_api.workload_identity import create_dag_processor_identity_verifier
 
 if TYPE_CHECKING:
     from airflow.models.dagbag import DBDagBag
@@ -105,7 +106,9 @@ async def lifespan(app: FastAPI, registry: svcs.Registry):
         # Create an app scoped validator, so that we don't have to fetch it every time
         registry.register_value(JWTValidator, _jwt_validator(), ping=JWTValidator.status)
 
-    yield
+    async with create_dag_processor_identity_verifier() as verifier:
+        app.state.dag_processor_identity_verifier = verifier
+        yield
 
 
 class CorrelationIdMiddleware(BaseHTTPMiddleware):
@@ -139,6 +142,9 @@ class JWTReissueMiddleware(BaseHTTPMiddleware):
     async def dispatch(self, request: Request, call_next):
         response: Response = await call_next(request)
 
+        if getattr(request.state, "skip_execution_token_reissue", False):
+            return response
+
         refreshed_token: str | None = None
         auth_header = request.headers.get("authorization")
         if auth_header and auth_header.lower().startswith("bearer "):
 
@@ -43,10 +43,10 @@ class IssuedToken(NamedTuple):
 
 
 class ExpiredDagProcessorToken(ValueError):
-    """The parent credential expired before a Job or parsing credential could be issued."""
+    """The parent credential expired before a session, Job, or parsing credential could be issued."""
 
 
-def _get_lifetime(generator: JWTGenerator, parent_expiry: float, message: str) -> int:
+def get_token_lifetime(generator: JWTGenerator, parent_expiry: float, message: str) -> int:
     """Return whole seconds a child token may live: the signer's lifetime, capped by its parent's expiry."""
     lifetime = math.floor(min(generator.valid_for, parent_expiry - timezone.utcnow().timestamp()))
     if lifetime < 1:
@@ -77,7 +77,7 @@ def generate_dag_processor_token(
     session_expiry: float,
 ) -> IssuedToken:
     """Issue a Job credential that cannot outlive its provisioned session credential."""
-    lifetime = _get_lifetime(generator, session_expiry, "Session credential has expired")
+    lifetime = get_token_lifetime(generator, session_expiry, "Session credential has expired")
     token = generator.generate(
         extras={
             "sub": str(session_id),
@@ -101,7 +101,7 @@ def generate_dag_parse_token(
     processor_expiry: float,
 ) -> IssuedToken:
     """Issue a parsing credential that cannot outlive its parent Job credential."""
-    lifetime = _get_lifetime(generator, processor_expiry, "Processor credential has expired")
+    lifetime = get_token_lifetime(generator, processor_expiry, "Processor credential has expired")
     token = generator.generate(
         extras={
             "sub": str(attempt_id),
 
@@ -60,6 +60,13 @@ class JobRegisterResponse(StrictBaseModel):
     expires_in: int = Field(gt=0, description="Seconds until the token expires.")
 
 
+class DagProcessorSessionTokenResponse(StrictBaseModel):
+    """A session credential issued after workload identity verification."""
+
+    token: str = Field(min_length=1, repr=False)
+    expires_in: int = Field(gt=0, description="Seconds until the session credential expires.")
+
+
 class JobHeartbeatResponse(StrictBaseModel):
     """Current state of the Job; ``restarting`` asks the processor to stop."""