[Task SDK] `BaseOperator.post_execute` is invoked without the `result` argument in the new task runner

[vishalmore90]

Under which category would you file this issue?

Task SDK

Apache Airflow version

main (development)

What happened and how to reproduce it?

In the new Airflow 3 Task SDK (task-sdk), the _execute_task function correctly invokes post_execute hooks after a task finishes executing. However, while it correctly passes the task execution result to the kwarg-based _post_execute_hook, it completely omits the result argument when calling an overridden class-level post_execute method.

As a consequence, any custom operator migrating from Airflow 2 to Airflow 3 that overrides BaseOperator.post_execute(self, context, result=None) to process the returned result will silently receive None instead of the actual execution result. Furthermore, if a custom operator defines result as a required positional argument, it will cause a hard crash during the task lifecycle.

Steps to reproduce:

1. Create a custom operator that overrides post_execute and expects the result argument:

   from airflow.sdk import BaseOperator
   
   class MyCustomOperator(BaseOperator):
       def execute(self, context):
           return "successful_payload"
   
       def post_execute(self, context, result):
           print(f"Task result was: {result}")

2. Run the task using the new Airflow 3 Task SDK runner.
3. The task runner will throw a TypeError: post_execute() missing 1 required positional argument: 'result' (if no default was provided in the signature). If a default result=None was provided, the operator will silently process None despite execute() successfully returning "successful_payload".

What you think should happen instead?

The Task SDK runner should pass the result argument when calling the overridden post_execute class method, matching both the behavior of the hook invocation above it and Airflow 2 backward compatibility.

Updating task-sdk/src/airflow/sdk/execution_time/task_runner.py around line 2262 fixes the issue:

-        create_executable_runner(post_execute_hook, outlet_events, logger=log).run(context)
+        create_executable_runner(post_execute_hook, outlet_events, logger=log).run(context, result)

Deployment

Other

Anything else?

Code Pointer:
In task-sdk/src/airflow/sdk/execution_time/task_runner.py around line 2259-2262:

    if (post_execute_hook := task._post_execute_hook) is not None:
        create_executable_runner(post_execute_hook, outlet_events, logger=log).run(context, result)
    if getattr(post_execute_hook := task.post_execute, "__func__", None) is not BaseOperator.post_execute:
        create_executable_runner(post_execute_hook, outlet_events, logger=log).run(context) # <--- BUG: `result` is omitted!

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

[vishalmore90]

/assign