task_instance_mutation_hook registered but never called in Airflow 2.11

[eladm-scytale]

Apache Airflow version

2.11.X

If "Other Airflow 3 version" selected, which one?

No response

What happened?

Apache Airflow version

2.11.x

What happened

The task_instance_mutation_hook defined in airflow_local_settings.py is registered correctly with POLICY_PLUGIN_MANAGER (confirmed via startup logs), but the hook function is never invoked when tasks are scheduled.

What you think should happen instead?

The hook should be called for each task instance before scheduling, allowing modifications to pool and other properties.

How to reproduce

1. Create $AIRFLOW_HOME/config/airflow_local_settings.py:

def task_instance_mutation_hook(task_instance):
    print(f"HOOK CALLED: task={task_instance.task_id}")
    # Modify pool for manual runs
    run_id = getattr(task_instance, 'run_id', '')
    if run_id.startswith('manual__'):
        task_instance.pool = "high_priority_pool"

print("airflow_local_settings loaded")

2.Trigger any DAG manually: airflow dags trigger <dag_id>

3.Expected: Logs show "HOOK CALLED: task=..."
Actual: Hook is never called, no log output

Additional attempts (all failed)
Plugin with @hookimpl decorator and explicit POLICY_PLUGIN_MANAGER.register()
Monkey-patching airflow.settings and airflow.models.taskinstance
All approaches show successful registration but the hook is never invoked.

Operating System

macOS/Linux

Versions of Apache Airflow Providers

N/A - core issue

Deployment

Docker-Compose

Deployment details

Docker Compose with CeleryExecutor

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[eladkal]

Fix for main branch #63049
Fix for 2.11.2 #62999