Fix in-process Execution API secrets routing in client contexts

[henry3260]

Why

Execution API requests can run inside in-process paths where SUPERVISOR_COMMS is present, which may incorrectly classify server-side code as client-side Task SDK execution.
When thathappens, Variable or Connectionreads can route back into Task SDK paths instead of staying server-side, creating a recursive chain and potential hangs.
The previous context signal was primarily environment-based, which is not always safe for request-scoped behavior in threaded in-process execution.

Because of that, Variable and Connection operations triggered while serving in-process Execution API
requests could still be routed back through Task SDK APIs instead of using the server-side secrets/backend
path, leading to recursive self-calls and hanging lookups.
closes: #65482

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

• Yes (please specify the tool below)

---

• 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.

---

Important

🛠️ Maintainer triage note for @henry3260 · by @potiuk · 2026-06-17 18:53 UTC

Some review feedback from jason810496, jscheffl is waiting on you:

• Please respond to the open review thread(s) — reply or push a fix in each, then mark them resolved.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Reply or push a fix in each thread, then mark them resolved.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[henry3260]

Could someone please take a look at this PR when they have a chance? Thanks!

[jason810496]

Nice! Thanks for the fix.

Additionally, could we add test for the scenario mentioned in the issue (#65482) to make sure the expected behavior?

Thanks.

[jason810496]

Please correct me if my understanding is wrong.

Another direction: How about adding lifespan to ensure SUPERVISOR_COMMS will be initialized for in-process Execution API?

Additionally, it would be nice to add airflow-e2e test for airflow dag test command with common scenarios as follow-up.

[henry3260]

How about adding lifespan to ensure SUPERVISOR_COMMS will be initialized for in-process Execution API?

I think lifespan is the wrong scope for this. It initializes app-scoped state once at startup, but the issue here is request-scoped
we need to isolate an in-process Execution API request from the outer client context. Initializing SUPERVISOR_COMMS in lifespan would make that client marker app-wide and could leak it into server-side request handling wdyt?

[jason810496]

How about adding lifespan to ensure SUPERVISOR_COMMS will be initialized for in-process Execution API?

I think lifespan is the wrong scope for this. It initializes app-scoped state once at startup, but the issue here is request-scoped we need to isolate an in-process Execution API request from the outer client context. Initializing SUPERVISOR_COMMS in lifespan would make that client marker app-wide and could leak it into server-side request handling wdyt?

For dag.test() command, it will use InProcessTestSupervisor under the hook, which means all the request will be processed in process. I'm not sure will there be race condition for override_process_context (when the _PROCESS_CONTEXT_OVERRIDE.reset teardown).

IMO, it's no harm to set the lifespan for InProcessTestSupervisor only app (override the app instance in _api_client factory method)

[henry3260]

How about adding lifespan to ensure SUPERVISOR_COMMS will be initialized for in-process Execution API?

I think lifespan is the wrong scope for this. It initializes app-scoped state once at startup, but the issue here is request-scoped we need to isolate an in-process Execution API request from the outer client context. Initializing SUPERVISOR_COMMS in lifespan would make that client marker app-wide and could leak it into server-side request handling wdyt?

For dag.test() command, it will use InProcessTestSupervisor under the hook, which means all the request will be processed in process. I'm not sure will there be race condition for override_process_context (when the _PROCESS_CONTEXT_OVERRIDE.reset teardown).

IMO, it's no harm to set the lifespan for InProcessTestSupervisor only app (override the app instance in _api_client factory method)

I understand your concern! Since ContextVar is independent per thread task, there shouldn't be any race conditions during the reset. That being said, I completely agree with your idea to minimize the influence area. I will update the PR to apply the override specifically within InProcessTestSupervisor.

[jason810496]

LGTM overall, thanks.

[vatsrahul1001]

Moving to 3.3.1 and planning only to include critical/high priority bug fixes to 3.3.0 found in 3.3.0b2.

[henry3260]

@kaxil Would you mind taking a look when you have a moment? Thanks!

[vatsrahul1001]

Added maintainer review label

[kaxil]

Have you been able to run the repro from #65482 against this branch? I applied it locally and the reported scenario still hangs, so I don't think this closes that issue.

What I ran: the script from the issue, with system_site_packages=True and no PyPI requirements so the venv resolves to local sources, the variable stored only in the metastore, under dag.test().

On main the run hangs, and the log shows Using Variable.get from 'airflow.models' is deprecated coming from airflow-core/src/airflow/models/variable.py:170 with a correlation_id attached, so the Execution API handler did take the Task SDK path. That part matches your diagnosis.

With this branch applied the warning is gone and the handler resolves through the metastore, so should_use_task_sdk_api_path() does what you intend. The run still hangs. A faulthandler dump inside the venv child shows it parked here:

File ".../sdk/execution_time/comms.py", line 332 in _read_frame
File ".../sdk/execution_time/comms.py", line 371 in _get_response
File ".../sdk/execution_time/comms.py", line 275 in send
File ".../sdk/execution_time/secrets/execution_api.py", line 119 in get_variable

The child never gets a response, because InProcessTestSupervisor.send_msg (task-sdk/src/airflow/sdk/execution_time/supervisor.py, line 2133 on main) appends to the in-process deque rather than writing a response frame to the socket that _setup_subprocess_socket() handed the child:

def send_msg(self, msg, request_id, error=None, **dump_opts):
    """Override to use in-process comms."""
    self.comms.messages.append(msg)

_handle_request answers every request through send_msg (supervisor.py:1918), and the only thing writing to that socket directly is _send_new_log_fd. That's why the logging-FD handshake added in #57212 works from a venv task while GetVariable and GetConnection never come back.

If I route responses for socket-delivered requests back onto the socket, the repro passes on plain main, without this branch. Applying this branch on top produces the same value, minus one round trip through the SDK client and the deprecation warning. In the in-process path the nested call is already covered by set_supervisor_comms(None) in InProcessSupervisorComms.send, which is why the detour still resolves to the right value today.

So as far as I can tell this is a robustness fix on that path rather than the fix for #65482.

[henry3260]

Have you been able to run the repro from #65482 against this branch? I applied it locally and the reported scenario still hangs, so I don't think this closes that issue.

What I ran: the script from the issue, with system_site_packages=True and no PyPI requirements so the venv resolves to local sources, the variable stored only in the metastore, under dag.test().

On main the run hangs, and the log shows Using Variable.get from 'airflow.models' is deprecated coming from airflow-core/src/airflow/models/variable.py:170 with a correlation_id attached, so the Execution API handler did take the Task SDK path. That part matches your diagnosis.

With this branch applied the warning is gone and the handler resolves through the metastore, so should_use_task_sdk_api_path() does what you intend. The run still hangs. A faulthandler dump inside the venv child shows it parked here:

File ".../sdk/execution_time/comms.py", line 332 in _read_frame
File ".../sdk/execution_time/comms.py", line 371 in _get_response
File ".../sdk/execution_time/comms.py", line 275 in send
File ".../sdk/execution_time/secrets/execution_api.py", line 119 in get_variable

The child never gets a response, because InProcessTestSupervisor.send_msg (task-sdk/src/airflow/sdk/execution_time/supervisor.py, line 2133 on main) appends to the in-process deque rather than writing a response frame to the socket that _setup_subprocess_socket() handed the child:

def send_msg(self, msg, request_id, error=None, **dump_opts):
    """Override to use in-process comms."""
    self.comms.messages.append(msg)

_handle_request answers every request through send_msg (supervisor.py:1918), and the only thing writing to that socket directly is _send_new_log_fd. That's why the logging-FD handshake added in #57212 works from a venv task while GetVariable and GetConnection never come back.

If I route responses for socket-delivered requests back onto the socket, the repro passes on plain main, without this branch. Applying this branch on top produces the same value, minus one round trip through the SDK client and the deprecation warning. In the in-process path the nested call is already covered by set_supervisor_comms(None) in InProcessSupervisorComms.send, which is why the detour still resolves to the right value today.

So as far as I can tell this is a robustness fix on that path rather than the fix for #65482.

Thanks for the detailed diagnosis, @kaxil — you're right on all counts. I spent time reading through the code to confirm the root cause you pointed out, and I've reworked the PR around it. Summary of what changed in 0a6a3ad:

• Root cause fix: the real hang was InProcessTestSupervisor.send_msg appending responses to the in-process deque instead of writing a response frame back to the socket the subprocess reads from. The subprocess-spawning operators (PythonVirtualenvOperator / ExternalPythonOperator) now get their responses routed back onto the socket, so GetVariable/GetConnection actually return.
• Env var precedence: dropped the env-var arm so the server short-circuit reads _PROCESS_CONTEXT_OVERRIDE only, leaving SUPERVISOR_COMMS winning over the env var. This keeps parity with ensure_secrets_backend_loaded() and avoids the dag.test() metastore fallback / lost AirflowSecretsBackendAccessDenied enforcement.
• request_scoped_server_context: made it the default so the parameter goes away, which also covers the triggerer_job_runner.in_process_api_server() instance.
• lru_cache singleton: no longer keying the cache on the new kwarg, so the singleton stays a singleton.
• Test coverage: added a test that drives a request through the in-process path with SUPERVISOR_COMMS present and asserts the SDK path isn't re-entered, covering the actual PythonVirtualenvOperator hangs indefinitely on Variable.get() (Task SDK) under airflow dags test and dag.test() — no error, no timeout #65482 scenario.

PTAL when you have a moment, thanks!