Stop the secrets backend lookup from importing the task runner outside tasks

task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py

@@ -52,6 +52,7 @@
     SOCKET_CLEANUP_TIMEOUT,
     WatchedSubprocess,
     _ensure_client,
+    _import_task_runner_before_fork,
     _make_process_nondumpable,
 )
 
@@ -255,6 +256,7 @@ def _target():
                 _log.error("Callback failed", error=error_msg)
                 sys.exit(1)
 
+        _import_task_runner_before_fork()
         return super().start(
             id=UUID(id) if not isinstance(id, UUID) else id,
             client=client,

task-sdk/src/airflow/sdk/execution_time/supervisor.py

@@ -172,6 +172,7 @@
     handle_update_dag_run_note,
 )
 from airflow.sdk.execution_time.schema import get_schema_version_migrator, resolve_body_class
+from airflow.sdk.log import _supervisor_comms
 
 try:
     from socket import send_fds
@@ -364,6 +365,19 @@ def _subprocess_main():
     main()
 
 
+def _import_task_runner_before_fork() -> None:
+    """
+    Import the task runner so that a child forked from this process inherits it.
+
+    Task and callback processes both import it, so importing it once here saves every forked child
+    from importing it again. It is not imported at module level because secrets lookups outside a
+    task import this module, and they must not load the task runtime.
+    """
+    # If it fails to import, the child imports it again and reports the error in its own log.
+    with suppress(ImportError):
+        import airflow.sdk.execution_time.task_runner  # noqa: F401
+
+
 def _reset_signals():
     # Uninstall the rich etc. exception handler
     sys.excepthook = sys.__excepthook__
@@ -1669,6 +1683,8 @@ def start(  # type: ignore[override]
         # Tests override `target` with a local stub to exercise the base
         # infrastructure; keep bare fork for those.
         use_exec = target is _subprocess_main and _task_process_uses_exec()
+        if target is _subprocess_main and not use_exec:
+            _import_task_runner_before_fork()
         proc: Self = super().start(
             id=what.id,
             client=client,
@@ -2919,14 +2935,9 @@ def ensure_secrets_backend_loaded() -> list[BaseSecretsBackend]:
     from airflow.sdk.execution_time.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
 
     # 1. Check for client context (SUPERVISOR_COMMS)
-    try:
-        from airflow.sdk.execution_time import task_runner
-
-        if hasattr(task_runner, "SUPERVISOR_COMMS") and task_runner.SUPERVISOR_COMMS is not None:
-            # Client context: task runner with SUPERVISOR_COMMS
-            return ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
-    except (ImportError, AttributeError):
-        pass
+    if _supervisor_comms() is not None:
+        # Client context: task runner with SUPERVISOR_COMMS
+        return ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
 
     # 2. Check for explicit server context
     if os.environ.get("_AIRFLOW_PROCESS_CONTEXT") == "server":

task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py

@@ -566,6 +566,20 @@ def test_start_keeps_bare_fork_regardless_of_exec_option(self, base_start_kwargs
 
         assert self.mock_super_start.call_args.kwargs.get("use_exec", False) is False
 
+    def test_start_imports_task_runner_before_forking(self, base_start_kwargs):
+        """The callback child is forked from this process, so it inherits a task runner imported here."""
+        calls = []
+        self.mock_super_start.side_effect = lambda **kwargs: calls.append("fork")
+
+        with patch(
+            "airflow.sdk.execution_time.callback_supervisor._import_task_runner_before_fork",
+            autospec=True,
+            side_effect=lambda: calls.append("import task runner"),
+        ):
+            CallbackSubprocess.start(**base_start_kwargs)
+
+        assert calls == ["import task runner", "fork"]
+
 
 class TestSuperviseCallbackExchangesTokenFirst:
     """No callback code may run until the single-use token has been exchanged."""

task-sdk/tests/task_sdk/execution_time/test_secrets.py

@@ -17,6 +17,10 @@
 
 from __future__ import annotations
 
+import subprocess
+import sys
+import textwrap
+
 import pytest
 
 from airflow.sdk.api.datamodels._generated import ConnectionResponse
@@ -308,6 +312,22 @@ def test_fallback_context_no_markers(self, monkeypatch):
         assert "MetastoreBackend" not in backend_classes
         assert "ExecutionAPISecretsBackend" not in backend_classes
 
+    def test_does_not_import_task_runner_outside_task_execution(self):
+        code = textwrap.dedent(
+            """
+            import sys
+            from airflow.sdk.execution_time.supervisor import ensure_secrets_backend_loaded
+
+            ensure_secrets_backend_loaded()
+            print("airflow.sdk.execution_time.task_runner" in sys.modules)
+            """
+        )
+
+        result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, check=False)
+
+        assert result.returncode == 0, result.stderr
+        assert result.stdout.strip().splitlines()[-1] == "False"
+
 
 class TestSecretsBackendMemoisation:
     BACKEND = "airflow.secrets.environment_variables.EnvironmentVariablesBackend"

task-sdk/tests/task_sdk/execution_time/test_supervisor.py

@@ -5818,6 +5818,60 @@ def test_activity_start_opts_into_fork_exec(monkeypatch, mocker, task_uses_exec,
     assert base_start.call_args.kwargs["target"] is target
 
 
+@pytest.mark.parametrize(
+    ("task_uses_exec", "target", "expected_calls"),
+    [
+        pytest.param(False, supervisor._subprocess_main, ["import task runner", "fork"], id="bare-fork"),
+        pytest.param(True, supervisor._subprocess_main, ["fork"], id="exec"),
+        pytest.param(False, lambda: None, ["fork"], id="test-target"),
+    ],
+)
+def test_activity_start_imports_task_runner_before_bare_fork(
+    monkeypatch, mocker, task_uses_exec, target, expected_calls
+):
+    """A forked task process inherits the task runner from the supervisor; a spawned one imports its own."""
+    calls = []
+
+    def fork(*args, **kwargs):
+        calls.append("fork")
+        return MagicMock(spec=ActivitySubprocess)
+
+    monkeypatch.setattr(supervisor, "_task_process_uses_exec", lambda: task_uses_exec)
+    mocker.patch.object(
+        supervisor,
+        "_import_task_runner_before_fork",
+        autospec=True,
+        side_effect=lambda: calls.append("import task runner"),
+    )
+    mocker.patch.object(supervisor.WatchedSubprocess, "start", autospec=True, side_effect=fork)
+
+    ActivitySubprocess.start(
+        dag_rel_path=os.devnull,
+        bundle_info=FAKE_BUNDLE,
+        what=TaskInstance(
+            id="4d828a62-a417-4936-a7a6-2b3fabacecab",
+            task_id="b",
+            dag_id="c",
+            run_id="d",
+            try_number=1,
+            dag_version_id=uuid7(),
+            queue="default",
+        ),
+        client=MagicMock(spec=sdk_client.Client),
+        target=target,
+        logger=MagicMock(),
+    )
+
+    assert calls == expected_calls
+
+
+def test_import_task_runner_before_fork_leaves_import_errors_to_the_child(monkeypatch):
+    """A task runner that fails to import is reported by the child, which imports it again."""
+    monkeypatch.setitem(sys.modules, "airflow.sdk.execution_time.task_runner", None)
+
+    supervisor._import_task_runner_before_fork()
+
+
 class TestTaskProcessUsesExec:
     """The config opt-in for fork+exec of the task process where the platform does not force it."""