Fix dag processor callback cleanup for versioned bundle files

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM

[hkc-8010]

Fix dag-processor presence/orphan checks so versioned bundle files are not treated as removed or new solely because the queued/tracked DagFileInfo carries a bundle_version.

This is distinct from #66301 / #66474. Those changes addressed stale serialized DAG metadata when only the bundle version changed. This fix is later in the callback pipeline: scheduler-emitted DAG-level and task-level callback requests are fetched and queued correctly, but manager-side presence checks must not purge, orphan, or re-queue the same file just because the scanned DAG file is represented without bundle_version.

The change keeps DagFileInfo equality and hashing version-aware, but adds a presence_key on DagFileInfo and uses that for manager-side “is this file already present / already represented?” checks. This is intentionally narrower than changing global DagFileInfo equality with bundle_version=field(compare=False), because callback requests are tied to a specific bundle version and we do not want to collapse queue/process identity semantics outside these presence checks.

That is now applied in:

• purge_removed_files_from_queue
• terminate_orphan_processes
• remove_orphaned_file_stats
• _add_new_files_to_queue
• prepare_file_queue
• _sort_by_mtime / processed_recently

Tests added:

• preserve a versioned queued file, processor, and stats entry when the same unversioned file is still present
• purge/kill/remove those entries when the file is truly absent
• avoid re-adding a scanned unversioned file when the same path is already represented in queue/processors/stats under a versioned key
• avoid duplicate queue/stats behavior in prepare_file_queue() and modified-time mode when versioned entries already exist

Validation:

• pytest airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend postgres --python 3.10 --downgrade-pendulum --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q
• breeze testing core-tests --backend sqlite --python 3.10 --force-lowest-dependencies --db-reset -- airflow-core/tests/unit/dag_processing/test_manager.py -q

closes: #66483

---

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

• Yes (Codex)

Generated-by: Codex following the guidelines

---

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

[ephraimbuddy]

Diagnosis is right and the fix is in the right shape — separating queue identity from file presence is the correct move. Three things I'd want addressed before this lands:

1. The same identity mismatch lives at manager.py:1267 in _add_new_files_to_queue. Both _file_stats and _processors accumulate versioned entries (set at L1112 / L1254). The next scan's unversioned DagFileInfo doesn't match a versioned stat, so the file is re-queued as "new" and re-parsed every cycle until the version key happens to align. Same bug family, untouched here. If you don't want to fix it in this PR, please open a follow-up issue and link it in the body.

2. Why does bundle_version need to stay in DagFileInfo identity at all?_callback_to_execute[file_info] is a list — multiple callbacks for the same path already coexist there without needing distinct keys. The only place versioned identity matters is _processors, but L1248 (if file in self._processors: continue) currently lets two different versions each spawn a processor for the same path, which is arguably worse than collapsing them. A one-line bundle_version: str | None = field(compare=False) change would fix the bug without needing the _present_file_key machinery. Worth a sentence in the PR body explaining why the surgical fix was preferred.

3. Scope is broader than the title suggests. Scheduler-emitted TaskCallbackRequest for normally-versioned bundles flows through the same queue and hits the same orphan check. Worth saying "DAG-level and task-level callbacks" in the body so reviewers understand the actual blast radius.

---

Additional inline notes (anchoring failed via API; included here)

On manager.py line 1023 — the new deque(...) comprehension in purge_removed_files_from_queue:

All three orphan-check methods recompute the same present_keys set from the same input. handle_removed_files is the only caller — compute it once there and pass keys down:

defhandle_removed_files(self, known_files):
present_keys= {(f.bundle_name, f.rel_path) forvinknown_files.values() forfinv}
self.purge_removed_files_from_queue(present_keys)
self.terminate_orphan_processes(present_keys)
self.remove_orphaned_file_stats(present_keys)

Minor at typical sizes, but the signature change is free and removes the duplication.

On test_manager.py line 453 — processor.kill.assert_not_called():

The three new tests only cover the "preserved" direction. None assert that an entry is correctly purged when the file is genuinely gone — a regression that made _present_file_key collapse all keys to a constant would still pass the suite.

Please add the negative case for each of the three methods. For terminate_orphan_processes specifically:

deftest_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self):
manager=DagFileProcessorManager(max_runs=1)
versioned_file=DagFileInfo(
bundle_name="testing",
rel_path=Path("callbacks.py"),
bundle_path=TEST_DAGS_FOLDER,
bundle_version="v1",
)
processor=MagicMock()
manager._processors[versioned_file] =processormanager.terminate_orphan_processes(present=set())
assertmanager._processors== {}
processor.kill.assert_called_once()

Without this, assert_not_called() doesn't actually pin the behavior — it just confirms kill wasn't invoked, which is also true if kill is never reachable.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @ephraimbuddy before posting

[hkc-8010]

@ephraimbuddy Thanks, this was really helpful. I pushed a follow-up that addresses the points you called out:

• moved the dual-identity concept onto DagFileInfo as presence_key
• compute present_keys once in handle_removed_files() and pass it through the orphan-cleanup helpers
• fixed the same presence/equality mismatch in _add_new_files_to_queue()
• also made prepare_file_queue(), processed_recently(), and _sort_by_mtime() presence-aware so scanned unversioned files do not get re-queued or depend on duplicate unversioned stats when versioned entries already exist
• added negative tests for the orphan-cleanup methods, alongside the preserve-direction coverage

I kept bundle_version in DagFileInfo equality/hashing and updated the PR body to explain why. The goal here was to keep callback/process identity version-aware while narrowing the fix to manager-side “present/already represented” checks.

I also updated the PR description to call out the broader scope: DAG-level and task-level callbacks both flow through this path.

[hkc-8010]

By the way, the customer tested the PR as a patch on their current Airflow version and confirmed that the callbacks resumed functioning without any problems.

[ephraimbuddy]

LGTM