execution_timeout on @task.stub tasks is not enforced #74716

Under which category would you file this issue?

Task SDK

Apache Airflow version

main

What happened and how to reproduce it?

A @task.stub task is declared in a Python Dag and runs in a language SDK. Its execution_timeout is set on the Python stub and is serialized with the Dag. But the runtime never gets it. StartupDetails, TIRunContext and TaskInstance in the supervisor schema have no timeout field. So the runtime cannot start a timer or send SetExecutionTimeout, and the task runs past the timeout.

A Python task does not have this problem because the task runner parses the Dag file and reads task.execution_timeout itself (task_runner.py).

Steps to reproduce (Go)

1. Pack a Go bundle with one handler that sleeps 90 seconds and logs every 5 seconds.
2. Add a Dag with this task as @task.stub(queue="golang", execution_timeout=timedelta(seconds=10)), and a Python task with the same timeout as a control.
3. In breeze shell --backend postgres, start the api-server, dag-processor and scheduler with LocalExecutor, using the same coordinator config as the Go SDK e2e test. Trigger the Dag.

@task.stub(queue="golang", execution_timeout=timedelta(seconds=10))
def go_sleep(): ...


@task(execution_timeout=timedelta(seconds=10))
def py_sleep():
    for i in range(18):
        time.sleep(5)
        print(f"still running {(i + 1) * 5}s")


@dag(dag_id="timeout_repro", schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def timeout_repro():
    go_sleep()
    py_sleep()

func goSleep(actx airflow.Context) error {
	start := time.Now()
	for i := 1; i <= 18; i++ {
		select {
		case <-actx.Done():
			actx.Logger().InfoContext(actx, "context cancelled", "elapsed_s", time.Since(start).Seconds())
			return actx.Err()
		case <-time.After(5 * time.Second):
			actx.Logger().InfoContext(actx, "still running", "elapsed_s", time.Since(start).Seconds())
		}
	}
	actx.Logger().InfoContext(actx, "finished after full sleep", "elapsed_s", time.Since(start).Seconds())
	return nil
}

// registered as airflow.TaskHandler("timeout_repro", "go_sleep", goSleep)

Result

I got the same result in two runs. I only ran Go. The other language SDKs get the same StartupDetails, so I think they behave the same, but I did not run them.

What you think should happen instead?

The task should fail when it runs past execution_timeout, like a Python task, with its retries applied.

Operating System

No response

Deployment

None

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

The runtime would need the timeout from the server, for example as a new field in TIRunContext. That would be an Execution API change, so it is bigger than the native Dag case.

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