[shahar1]

In parallel test runs, a failed Python Dag import could leave stale state that broke later imports in the same worker; the fix always clears it.

AI Summary

`PythonDagImporter` sets `DagContext.current_autoregister_module_name` before importing a Dag file, and drops that module from `sys.modules` if the import fails. The name was only reset in `_process_modules`, which is skipped when `_load_modules` raises: the `AirflowConfigException` re-raise for an invalid `get_dagbag_import_timeout` value, or the generic `except Exception` early return (for example, a timeout policy that raises). The stale name stayed on the class. After that, every `with DAG(...)` in the same process ran `DagContext.pop()`, which does `sys.modules[stale_name]` and raised `KeyError`. Dags built by other importers in the same process, such as the YAML importer, failed to build.

This PR moves the reset into a finally around _load_modules.

How it surfaced

Three YAML importer tests failed in the Task SDK tests / task-sdk:P3.11 tests job of #74889 (run 39004464506). That PR moves CI onto 32-CPU runners, so xdist started 32 workers and grouped tests differently. All three failures were on worker gw12:

test_import_file_ref_outside_dag_dir_is_rejected[/etc/passwd]  - assert False
test_import_yaml_inside_zip_bundle                             - KeyError: 'unusual_prefix_8ffeae589d38742da930da86c044dccc96dd545f_dag'
test_import_template_file_reads_then_templates                 - KeyError: 'unusual_prefix_8ffeae589d38742da930da86c044dccc96dd545f_dag'

The log runs at verbosity 0, so it does not list which tests ran on each worker. The hash identifies the predecessor anyway: 8ffeae58… is the sha1 of /tmp/pytest-of-root/pytest-0/popen-gw12/test_invalid_dagbag_import_tim0/bundle/dag.py. That is the temp path of TestPythonDagImporter::test_invalid_dagbag_import_timeout_raises_custom_exception on gw12. The /etc/passwd case has the same cause: the KeyError raised by DAG.__exit__ replaces the expected "resolves outside" error.

Minimal deterministic reproduction (Breeze, one process, before this fix gives 3 failed / 1 passed):

breeze run pytest -p no:randomly \
  task-sdk/tests/task_sdk/importers/test_python_importer.py::TestPythonDagImporter::test_invalid_dagbag_import_timeout_raises_custom_exception \
  "task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py::test_import_file_ref_outside_dag_dir_is_rejected[/etc/passwd]" \
  task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py::test_import_yaml_inside_zip_bundle \
  task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py::test_import_template_file_reads_then_templates

Testing

• The new test_failed_load_does_not_break_later_dags covers both error paths. Without the fix, both cases fail with the KeyError.
• The reproduction above passes with the fix (4 passed).
• task-sdk/tests/task_sdk/importers and shared/module_loading/tests pass (305 tests).
• The full task-sdk suite passed twice in Breeze with -n 32 --skip-db-tests (4057 passed, 6 skipped each time).

---

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

• Yes — Claude Code (Opus 5.5)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[potiuk]

LGTM. I traced the paths that skipped the reset:

• the AirflowConfigException re-raise for a non-numeric get_dagbag_import_timeout;
• a timeout policy that raises something other than ImportError/AttributeError, which hits the except Exception early return;
• KeyboardInterrupt from parse().

In each case the stale name was left on the class, and every later DagContext.pop() hit KeyError on sys.modules[...]. The finally covers all of these. Dropping the reset from _process_modules is safe because nothing there depends on the name. The new test fails without the fix, since the second with DAG(...) raises.

CI for the last push shows a cancelled run, so it needs a full re-run before merge.

---

Drafted-by: Claude Opus 5.5; reviewed by potiuk before posting

[github-actions]

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test