Decouple workload JWT token lifetime from task_queued_timeout

[Kunal8954]

What

Adds a new, optional [execution_api] workload_token_expiration_time
config option that controls how long a workload's JWT token (minted
when a task instance or callback is queued for an executor) stays
valid, independently of [scheduler] task_queued_timeout.

Why

Today, BaseWorkloadSchema.generate_token() hardcodes the workload
token's lifetime to [scheduler] task_queued_timeout. That value has
a second, unrelated job: it's also how long a task can sit in the
queued state before the scheduler's orphan-detection sweep reaps it.

This forces operators into an uncomfortable tradeoff: raising
task_queued_timeout to tolerate a longer queue backlog (e.g. under
heavy load, or with a slow-starting KubernetesExecutor) also silently
extends how long a genuinely stuck task goes undetected - the exact
scenario in #72469, where a scheduler crash mid-dispatch left tasks
stuck in queued for up to 10 hours before being reaped, because
task_queued_timeout had been raised for legitimate queue-wait
reasons.

Design

• New option: [execution_api] workload_token_expiration_time
  (float, seconds, default unset).
• When unset, _workload_token_valid_for() falls back to
  [scheduler] task_queued_timeout - existing deployments see no
  behavior change unless they opt in.
• When set, it's used directly and task_queued_timeout is no longer
  consulted for token lifetime (only for its original orphan-detection
  purpose).
• I deliberately did not give the new option its own independent
  hardcoded default (e.g. 600): if it defaulted to something shorter
  than an operator's already-customized task_queued_timeout, a
  legitimately-queued task's auth token could expire before the
  scheduler's own orphan sweep fires, breaking task execution for
  people who'd raised task_queued_timeout without knowing about this
  new option. Falling back preserves current behavior by construction.

Changes

• airflow-core/src/airflow/executors/workloads/base.py: extracted
  _workload_token_valid_for() implementing the fallback logic used
  by generate_token() (inherited by ExecuteTask, ExecuteCallback,
  and the connection-test workload).
• airflow-core/src/airflow/config_templates/config.yml: added the
  new option under [execution_api], and cross-referenced it from
  [scheduler] task_queued_timeout's description.
• Tests in airflow-core/tests/unit/executors/test_workloads.py
  covering both the fallback and the override path, plus an end-to-end
  token-decoding test.
• Newsfragment.

Fixes #72469

Gen-AI disclosure: used Claude Code (Anthropic) to investigate the
existing coupling, implement the change, write tests, and draft this
description.

🤖 Generated with Claude Code

https://claude.ai/code/session_015kAvbZ6SeKgp6jbGcvpXSh