Airflow попускает задачу, выбранную через BranchPythonOperator

Имеется вот такой вот граф

(start_batch_task >> start_session_download_task >> get_save_files_task >> end_session_download_task
 >> need_to_load_new_files_task >> [start_session_stg_task, end_batch_task])

start_session_stg_task >> load_stg_task >> end_session_stg_task >> end_batch_task

need_to_load_new_files_task представлен оператором BranchPythonOperator, который возвращает соотвественно один из тасков [start_session_stg_task, end_batch_task]

Проблема в том, что если он выбирает end_batch_task, то он почему-то его скипает. И происходит это только если end_batch_task есть после end_session_stg_task. Если его оттуда убрать, все работает прекрасно. Почему он его скипает, если ветка успешно выбирает его в качестве цели?

Если поставить у end_batch_task trigger_rule=TriggerRule.ONE_SUCCESS, то тогда он выполняется вообще не обращая внимания на выбор ветки (что логично).

Вот пример логов когда выбран end_batch_task, но по сути он его всё равно скипнул (скрин)

[2024-12-12, 16:21:12 MSK] {python.py:177} INFO - Done. Returned value was: end_batch_task
[2024-12-12, 16:21:12 MSK] {python.py:211} INFO - Branch callable return end_batch_task
[2024-12-12, 16:21:12 MSK] {skipmixin.py:155} INFO - Following branch end_batch_task
[2024-12-12, 16:21:12 MSK] {skipmixin.py:211} INFO - Skipping tasks ['start_session_stg_task']