[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

What

#74222 made the Execution API reject writes unless the token's sub is a live task attempt or a callback. Variables are global: no task attempt owns them, so that check protects no attempt data there. It does break every caller that writes Variables without a task identity and runs outside the API server, such as Dag processors and asset watchers on remote hosted agents.

Current behaviour

With a client on API version 2026-10-30 or newer, PUT and DELETE /execution/variables/{key} return 404 when the token's sub is not a live TaskInstance or Callback. Variable.set and Variable.delete from a remote Dag processor (parse time or Dag/task callbacks) fail. In 3.3 they worked. In-process callers are unaffected because they are already exempt.

Proposed change

Add an allow_non_ti_writer route marker in security.py and apply it to put_variable and delete_variable. require_auth skips the automatic live attempt check on marked routes. The token type check and has_variable_access still run.

Changes of Note

• I added a separate marker rather than reusing skip_auto_ti_attempt_live, because that one means "this route checks liveness itself", which Variables don't.
• XCom, RTIF, task state and Dag run writes keep the check. test_unknown_execution_identity_cannot_write now uses an XCom write to show non-task IDs are still blocked there.

User implications / backcompat

• Restores the 3.3 behaviour for Variable writes.
• An archived (retried or deleted) attempt can write Variables again instead of getting 410. That is the deliberate tradeoff: Variables carry no attempt data a stale token could corrupt. So I think it's a tradeoff that can be lived with.

Testing

End to end against a running API server (breeze start-airflow), using real signed tokens shaped like a remote Dag processor (sub = agent uid, no scope) and a remote asset watcher (sub = null UUID), sent over HTTP with the Task SDK `Client -

Using this script:

from __future__ import annotations

from uuid import UUID, uuid4

from airflow.api_fastapi.execution_api.app import create_jwt_generator
from airflow.models.variable import Variable
from airflow.sdk.api.client import Client, ServerResponseError

BASE_URL = "http://localhost:8080/execution/"

CALLERS = {
    "agent Dag processor (sub=agent uid, no scope)": {"sub": str(uuid4())},
    "remote asset watcher (sub=null UUID)": {"sub": str(UUID(int=0)), "scope": "execution"},
}


def main() -> None:
    generator = create_jwt_generator()
    for name, claims in CALLERS.items():
        client = Client(base_url=BASE_URL, token=generator.generate(extras=claims))
        key = f"e2e-non-ti-{uuid4().hex[:8]}"
        print(f"\n== {name}")
        try:
            client.variables.set(key, "written-over-http")
            print(f"  set    OK, DB value = {Variable.get(key)!r}")
            client.variables.delete(key)
            print("  delete OK")
        except ServerResponseError as e:
            print(f"  REJECTED {e.response.status_code}: {e.response.text}")
        try:
            client.xcoms.set("some_dag", "some_run", "some_task", "k", "v")
            print("  xcom   OK (unexpected)")
        except ServerResponseError as e:
            print(f"  xcom   rejected {e.response.status_code} (expected, still guarded)")


if __name__ == "__main__":
    main()

Before (main):

== agent Dag processor (sub=agent uid, no scope)
  REJECTED 404: {"detail":{"reason":"not_found","message":"Task Instance not found"}}
  xcom   rejected 404 (expected, still guarded)

== remote asset watcher (sub=null UUID)
  REJECTED 404: {"detail":{"reason":"not_found","message":"Task Instance not found"}}
  xcom   rejected 404 (expected, still guarded)

After (this branch):

== agent Dag processor (sub=agent uid, no scope)
  set    OK, DB value = 'written-over-http'
  delete OK
  xcom   rejected 404 (expected, still guarded)

== remote asset watcher (sub=null UUID)
  set    OK, DB value = 'written-over-http'
  delete OK
  xcom   rejected 404 (expected, still guarded)

Unit tests: test_non_ti_identity_can_write_variables covers set and delete for an unknown ID and an archived attempt, and fails without the change.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[pierrejeambrun]

LGTM