Move ExecutorCallback execution into a supervised process

[ferruzzi]

Move ExecutorCallback execution into a supervised process and make the entire flow more generic to account for future workload types

• Generalized base workload class which both tasks and callbacks inherit (Added key, display_name, success_state, and failure_state fields)
• Created a callback supervisor
• Simplified/unified the shared workflow between tasks and callbacks
• Aligned logging messages

Changes applied to both the LocalExecutor and the Celery Executor (the only executors that currently support synchronous callbacks in the worker)

Promised follow-up to #61153
promised tags for @jason810496@ashb@amoghrajesh

---

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

• Yes (please specify the tool below)

---

• 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 or {issue_number}.significant.rst, in airflow-core/newsfragments.

[ferruzzi]

Failures don't feel related and don't give any obvious error message, I'll try rebasing tomorrow and see if that clears them

[jscheffl]

in-deed, these three tests are broken on main. Hope they are fixed soon, else check #ci-cd channel in Slack.

[jscheffl]

Okay for Edge (just a doc change), would prefer the core SDK experts making the review.

[Copilot]

Pull request overview

This PR moves synchronous ExecutorCallback execution into a supervised subprocess (similar to task execution), and generalizes executor workload handling so tasks and callbacks share a unified flow across LocalExecutor and CeleryExecutor.

Changes:

• Introduces a callback supervisor (supervise_callback) and migrates callback execution to run in a supervised subprocess.
• Adds generic workload metadata (key, display_name, success_state, failure_state) to unify executor handling for tasks and callbacks.
• Aligns “workload finished” logging and updates executor/provider integrations and tests accordingly.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
task-sdk/tests/task_sdk/execution_time/test_supervisor.py	Updates expected supervisor log fields to the new “Workload finished” format.
task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py	Adds unit tests for callback importing/execution behavior.
task-sdk/src/airflow/sdk/execution_time/supervisor.py	Renames “Task finished” log event to “Workload finished” and adds workload identifiers.
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py	Adds supervised callback execution implementation and subprocess wrapper.
providers/edge3/src/airflow/providers/edge3/cli/worker.py	Updates a reference comment to the renamed LocalExecutor helper.
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py	Switches callback workload execution to supervise_callback.
airflow-core/tests/unit/executors/test_local_executor.py	Updates tests for renamed LocalExecutor helper and adds callback supervisor invocation tests.
airflow-core/tests/unit/executors/test_base_executor.py	Migrates callback execution tests to use the new execute_callback helper.
airflow-core/src/airflow/executors/workloads/task.py	Adds generic workload properties to ExecuteTask.
airflow-core/src/airflow/executors/workloads/callback.py	Adds generic workload properties to ExecuteCallback and removes the old inline callback executor.
airflow-core/src/airflow/executors/workloads/base.py	Introduces abstract workload interface (key, display_name, success/failure states).
airflow-core/src/airflow/executors/workloads/init.py	Adds ExecutorWorkload union for executor-supported workload types.
airflow-core/src/airflow/executors/local_executor.py	Unifies task/callback execution into _execute_workload and runs callbacks via supervise_callback.
airflow-core/src/airflow/executors/base_executor.py	Updates typing to use ExecutorWorkload for workload queuing/scheduling.

[o-nikolas]

Left a few comments, on the whole it looks reasonable to me. But @ashb or @amoghrajesh should definitely look at the supervisor portion.

[Copilot]

Pull request overview

Copilot reviewed 22 out of 22 changed files in this pull request and generated 2 comments.

[ashb]

Re-triggering with full tests just incase they weren't all running

[ashb]

LGTM now, but please investigate the Kube test