@@ -348,15 +348,17 @@ The ``JWTBearer`` security dependency validates the token once per request:
 2. Performs cryptographic signature validation via ``JWTValidator``.
 3. Verifies standard claims (``exp``, ``iat``, ``aud`` — ``nbf`` and ``iss`` if configured).
 4. Defaults the ``scope`` claim to ``"execution"`` if absent.
-5. **Validates the task identity claims against a typed Pydantic schema.** ``TIClaims``
-   (in ``airflow.api_fastapi.execution_api.datamodels.token``) enforces that ``scope`` is one
-   of the declared ``TokenScope`` literals (``"execution"`` or ``"workload"``); ``TIToken``
-   then parses the ``sub`` claim through a ``UUID`` field, which rejects non-UUID values.
-   A token whose ``scope`` is unknown, or whose ``sub`` is not a valid UUID, is rejected with
-   ``403 Forbidden`` even when the cryptographic signature checks pass. ``TIClaims`` keeps
-   ``extra="allow"`` so auth managers can attach additional, deployment-specific claims
-   without modifying the core schema; only the security-critical fields are typed.
-6. Creates a ``TIToken`` object with the task instance ID and validated claims.
+5. **Validates the identity and its scope-specific claims against typed Pydantic schemas.**
+   ``ExecutionToken`` (in ``airflow.api_fastapi.execution_api.datamodels.token``) parses ``sub``
+   as a UUID. Its claims form a discriminated union selected by ``scope``: ``TIClaims`` for
+   ``"execution"``, ``"workload"``, and ``"callback"``; ``DagProcessorSessionClaims`` for
+   provisioning; ``DagProcessorClaims`` for a registered Job; and ``DagParseClaims`` for a
+   parsing attempt. Each processor claim type requires expiry and its scope-specific grants
+   and identity fields. Unknown scopes, invalid subjects, or missing required claims return
+   ``403 Forbidden`` even when cryptographic validation succeeds. All claim types allow extra
+   fields so auth managers can attach deployment-specific claims.
+6. Constructs the corresponding token object: ``TIToken``, ``DagProcessorSessionToken``,
+   ``DagProcessorToken``, or an ``ExecutionToken`` carrying ``DagParseClaims``.
 7. Caches the validated token on the ASGI request scope for the duration of the request.
 
 Route-level enforcement is handled by ``require_auth``:
@@ -367,26 +369,32 @@ Route-level enforcement is handled by ``require_auth``:
   ``{task_instance_id}`` path parameter, preventing a worker from accessing another task's
   endpoints.
 
+The task-only, processor-session, and processor-Job dependencies additionally reject tokens of
+the wrong principal type. Endpoints shared by tasks and parsing attempts use the general
+``CurrentExecutionToken`` dependency with their declared route scopes.
+
 .. mermaid::
 
     flowchart TD
         REQ([Incoming request<br/>Authorization: Bearer ...])
         REQ --> CACHE{Cached on<br/>request.scope?}
-        CACHE -->|yes| RET([Return cached TIToken])
+        CACHE -->|yes| RET([Return cached token])
         CACHE -->|no| SIG[JWTValidator:<br/>verify signature]
         SIG -->|fail| F1([403 Forbidden])
         SIG -->|ok| STD[Verify exp / iat / nbf<br/>aud / iss]
         STD -->|fail| F1
         STD -->|ok| SCOPE[Default scope to<br/>'execution' if absent]
-        SCOPE --> SCHEMA[TIClaims:<br/>typed Pydantic schema]
+        SCOPE --> SCHEMA[ExecutionToken:<br/>UUID identity and<br/>scope-discriminated claims]
         SCHEMA -->|ValidationError| F1
         SCHEMA -->|ok| TYP{require_auth:<br/>scope in<br/>route.allowed_token_types?}
         TYP -->|no| F1
         TYP -->|yes| SELF{ti:self scope<br/>declared?}
-        SELF -->|no| OK([Return TIToken])
+        SELF -->|no| PRINCIPAL{Typed dependency:<br/>principal type matches?}
         SELF -->|yes| MATCH{token.sub ==<br/>task_instance_id?}
         MATCH -->|no| F1
-        MATCH -->|yes| OK
+        MATCH -->|yes| PRINCIPAL
+        PRINCIPAL -->|no| F1
+        PRINCIPAL -->|yes or general dependency| OK([Return validated token])
 
         classDef fail fill:#ffcdd2,stroke:#c62828,stroke-width:2px,color:#000
         classDef pass fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px,color:#000
@@ -489,9 +497,10 @@ guidance, and the planned strategic and tactical improvements.
 Dag processor HTTP client credentials
 -------------------------------------
 
-The core ``DagProcessorAPIClient`` supports authenticated access to the Execution API
-without holding its signing key. This is the client and authentication foundation;
-``airflow dag-processor`` still uses its existing database-backed lifecycle.
+The core ``DagProcessorAPIClient`` lets processor integrations authenticate to the Execution
+API without holding its signing key. The standard ``airflow dag-processor`` command does
+not yet use this client. Provisioning a token file alone does not switch that command to
+HTTP or remove its need for database access.
 
 On a trusted host with the API signing key, provision a session token for the bundles
 the client may access::
@@ -503,17 +512,24 @@ The command writes the token atomically with owner-only permissions. Keep the pr
 running and mount its directory rather than a single file so token replacement stays
 visible. Each processor process needs its own session.
 
-The ``dag_processor_session`` token registers a Job through ``POST /jobs``. The returned
-``dag_processor`` token can heartbeat and complete that Job, read Connections and Variables
-for a granted bundle, and exchange a parsing credential through
-``POST /jobs/{job_id}/parse-token``. Registration retries keep the same registration UUID.
-A new process uses a new UUID; it cannot replace a session's Job while that Job is alive.
-Ending or replacing the Job invalidates its runtime credentials.
+The three credentials have different issuers and permissions:
+
+- ``dag_processor_session`` is issued by trusted 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
+  credential through ``POST /jobs/{job_id}/parse-token``. It cannot outlive the session
+  token used to obtain it.
+- ``dag_parse`` is returned by the parsing-token exchange for one attempt, bundle, and
+  relative file location. It permits parse-time requests, including callback-context reads,
+  but cannot manage Jobs or exchange tokens. It cannot outlive its parent Job credential.
+
+Registration retries keep the same registration UUID. A new process uses a new UUID;
+it cannot replace a session's Job while that Job is alive. Ending or replacing the Job
+invalidates its runtime credentials.
 
 The client selects the bundle with ``use_bundle`` and a parsing attempt with ``use_parse``.
-A ``dag_parse`` token binds the attempt UUID, bundle, and relative file location to the Job.
-It permits parse-time requests for that bundle, including callback-context reads, but cannot
-manage Jobs or exchange tokens. A request header cannot override the signed bundle.
+A request header cannot override the parsing token's signed bundle.
 Connection and Variable access uses the bundle's current team.
 
 The client attempts renewal at 80% of token lifetime. Transient early-renewal failures leave
 
@@ -532,10 +532,11 @@ potentially still executes with direct database access in the Dag File Processor
    * **Tightened deserialization allowlist.** The serializer's class-name allowlist regex is
      anchored to require a full-string match, so attacker-controlled values cannot smuggle
      untrusted class names by appending an allowed suffix to a non-allowed name.
-   * **Typed JWT claims schema.** Even after cryptographic validation, claims are run through a
-     typed Pydantic schema (``TIClaims``) that enforces the ``scope`` literal, and then through
-     ``TIToken`` which parses ``sub`` as a UUID. A token whose ``scope`` is unknown or whose
-     ``sub`` is not a valid UUID is rejected before any route handler runs.
+   * **Typed JWT claims schemas.** Even after cryptographic validation, ``ExecutionToken`` validates
+     ``sub`` as a UUID and selects a claims schema by ``scope``: ``TIClaims`` for task, workload,
+     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.
 
 **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
 
@@ -56,6 +56,7 @@
 
 __all__ = [
     "create_task_execution_api_app",
+    "create_jwt_generator",
     "lifespan",
     "CorrelationIdMiddleware",
 ]
@@ -77,7 +78,8 @@ def _jwt_validator():
     return validator
 
 
-def _jwt_generator():
+def create_jwt_generator() -> JWTGenerator:
+    """Create a signer using the Execution API's configured key, audience, and lifetime."""
     from airflow.configuration import conf
 
     generator = JWTGenerator(
@@ -99,7 +101,7 @@ async def lifespan(app: FastAPI, registry: svcs.Registry):
     # record this here
     app.state.svcs_registry = registry
 
-    registry.register_factory(JWTGenerator, _jwt_generator)
+    registry.register_factory(JWTGenerator, create_jwt_generator)
 
     # InProcessExecutionAPI stubs out JWTValidator: don't re-register in that case.
     if JWTValidator not in registry:
 
@@ -15,32 +15,34 @@
 # specific language governing permissions and limitations
 # under the License.
 """
-Issue Dag processor session tokens.
+Issue Dag processor session, Job, and parsing tokens.
 
-Only trusted provisioning runs this: it needs the Execution API signing key, which a Dag processor
-must never hold, because the processor runs the code it parses.
+Only trusted provisioning and the API server issue these tokens. A Dag processor must never hold
+the signing key, because the processor runs the code it parses.
 """
 
 from __future__ import annotations
 
-import contextlib
-import os
-import tempfile
 from typing import TYPE_CHECKING
 
-from airflow.api_fastapi.execution_api.app import _jwt_generator
+from airflow._shared.timezones import timezone
 
 if TYPE_CHECKING:
     from collections.abc import Collection
-    from pathlib import Path
     from uuid import UUID
 
+    from airflow.api_fastapi.auth.tokens import JWTGenerator
+
+
+class ExpiredDagProcessorToken(ValueError):
+    """The parent credential expired before a Job or parsing credential could be issued."""
+
 
 def generate_dag_processor_session_token(
-    *, session_id: UUID, bundle_names: Collection[str], valid_for: float
+    generator: JWTGenerator, *, session_id: UUID, bundle_names: Collection[str], valid_for: float
 ) -> str:
     """Return a ``dag_processor_session`` token, which a Dag processor exchanges for a Job token."""
-    return _jwt_generator().generate(
+    return generator.generate(
         extras={
             "sub": str(session_id),
             "scope": "dag_processor_session",
@@ -50,16 +52,51 @@ def generate_dag_processor_session_token(
     )
 
 
-def write_token_file(path: Path, token: str) -> None:
-    """Replace the token file in one step, so a processor rereading it never sees a partial token."""
-    fd, tmp_path = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.")
-    try:
-        with os.fdopen(fd, "w") as tmp:
-            tmp.write(token)
-            tmp.flush()
-            os.fsync(tmp.fileno())
-        os.replace(tmp_path, path)
-    except BaseException:
-        with contextlib.suppress(FileNotFoundError):
-            os.unlink(tmp_path)
-        raise
+def generate_dag_processor_token(
+    generator: JWTGenerator,
+    *,
+    session_id: UUID,
+    job_id: int,
+    bundle_names: Collection[str],
+    session_expiry: float,
+) -> str:
+    """Issue a Job credential that cannot outlive its provisioned session credential."""
+    remaining = session_expiry - timezone.utcnow().timestamp()
+    if remaining <= 0:
+        raise ExpiredDagProcessorToken("Session credential has expired")
+    return generator.generate(
+        extras={
+            "sub": str(session_id),
+            "scope": "dag_processor",
+            "dag_bundles": sorted(bundle_names),
+            "job_id": job_id,
+        },
+        valid_for=min(generator.valid_for, remaining),
+    )
+
+
+def generate_dag_parse_token(
+    generator: JWTGenerator,
+    *,
+    session_id: UUID,
+    job_id: int,
+    attempt_id: UUID,
+    bundle_name: str,
+    relative_fileloc: str,
+    processor_expiry: float,
+) -> str:
+    """Issue a parsing credential that cannot outlive its parent Job credential."""
+    remaining = processor_expiry - timezone.utcnow().timestamp()
+    if remaining <= 0:
+        raise ExpiredDagProcessorToken("Processor credential has expired")
+    return generator.generate(
+        extras={
+            "sub": str(attempt_id),
+            "scope": "dag_parse",
+            "session_id": str(session_id),
+            "job_id": job_id,
+            "dag_bundles": [bundle_name],
+            "relative_fileloc": relative_fileloc,
+        },
+        valid_for=min(generator.valid_for, remaining),
+    )
@@ -20,48 +20,70 @@
 from typing import Annotated, Literal
 from uuid import UUID
 
-from pydantic import ConfigDict, Field, model_validator
+from pydantic import ConfigDict, Field
 
 from airflow.api_fastapi.core_api.base import BaseModel
-from airflow.typing_compat import Self
 
-TokenScope = Literal[
-    "execution", "workload", "callback", "dag_processor_session", "dag_processor", "dag_parse"
-]
+TaskTokenScope = Literal["execution", "workload", "callback"]
+DagProcessorSessionScope = Literal["dag_processor_session"]
+DagProcessorScope = Literal["dag_processor"]
+DagParseScope = Literal["dag_parse"]
+TokenScope = Literal[TaskTokenScope, DagProcessorSessionScope, DagProcessorScope, DagParseScope]
 
 
-class ExecutionClaims(BaseModel):
+class _JWTClaims(BaseModel):
     """
     Validated JWT claims for an Execution API principal.
 
-    JWTValidator validates exp/iat/nbf/aud before these claims are constructed.
+    JWTValidator checks standard JWT claims before constructing claims for HTTP requests.
     Extra claims are allowed for compatibility with existing task tokens.
     """
 
     model_config = ConfigDict(extra="allow")
 
-    scope: TokenScope = "execution"
+
+class TIClaims(_JWTClaims):
+    """Claims for task execution, workload exchange, or callback execution."""
+
+    scope: TaskTokenScope = "execution"
+    # Trusted in-process callers construct task identities without a JWT or expiry.
     exp: float | None = None
-    dag_bundles: frozenset[Annotated[str, Field(min_length=1)]] | None = None
-    """Dag bundles a Dag processor token may act for."""
-    job_id: int | None = None
-    """Job a ``dag_processor`` token was issued for when that Job registered."""
-    session_id: UUID | None = None
-    """Processor session that owns a parsing attempt, whose own identity is the token subject."""
-    relative_fileloc: str | None = Field(default=None, min_length=1, max_length=2000)
-    """Bundle-relative file being parsed; an archive is one file under the current processor model."""
-
-    @model_validator(mode="after")
-    def validate_dag_processor_claims(self) -> Self:
-        if self.scope in ("dag_processor_session", "dag_processor", "dag_parse") and not self.dag_bundles:
-            raise ValueError(f"A {self.scope} token must grant at least one Dag bundle")
-        if self.scope in ("dag_processor", "dag_parse") and self.job_id is None:
-            raise ValueError(f"A {self.scope} token must name the Job it was issued for")
-        if self.scope == "dag_parse" and (
-            len(self.dag_bundles or ()) != 1 or self.session_id is None or self.relative_fileloc is None
-        ):
-            raise ValueError("A dag_parse token must name one bundle, its file, and its processor session")
-        return self
+
+
+DagBundleGrant = Annotated[frozenset[Annotated[str, Field(min_length=1)]], Field(min_length=1)]
+
+
+class DagProcessorSessionClaims(_JWTClaims):
+    """Provisioned bundle grants used only to register a Job or renew its credential."""
+
+    scope: DagProcessorSessionScope = "dag_processor_session"
+    exp: float
+    dag_bundles: DagBundleGrant
+
+
+class DagProcessorClaims(_JWTClaims):
+    """Claims for managing one registered Job and exchanging its parsing credentials."""
+
+    scope: DagProcessorScope = "dag_processor"
+    exp: float
+    dag_bundles: DagBundleGrant
+    job_id: int
+
+
+class DagParseClaims(_JWTClaims):
+    """Claims for one file-parsing attempt within a registered Job's bundle."""
+
+    scope: DagParseScope = "dag_parse"
+    exp: float
+    dag_bundles: DagBundleGrant = Field(max_length=1)
+    job_id: int
+    session_id: UUID
+    relative_fileloc: str = Field(min_length=1, max_length=2000)
+
+
+ExecutionClaims = Annotated[
+    TIClaims | DagProcessorSessionClaims | DagProcessorClaims | DagParseClaims, Field(discriminator="scope")
+]
 
 
 class ExecutionToken(BaseModel):
@@ -71,6 +93,19 @@ class ExecutionToken(BaseModel):
     claims: ExecutionClaims
 
 
-# Preserve imports for task-specific consumers while shared endpoints use the general principal.
-TIClaims = ExecutionClaims
-TIToken = ExecutionToken
+class TIToken(ExecutionToken):
+    """Authenticated task, workload, or callback identity."""
+
+    claims: TIClaims
+
+
+class DagProcessorSessionToken(ExecutionToken):
+    """Authenticated provisioned Dag processor session."""
+
+    claims: DagProcessorSessionClaims
+
+
+class DagProcessorToken(ExecutionToken):
+    """Authenticated Dag processor Job."""
+
+    claims: DagProcessorClaims