[ChickenBenny]

• related: Enforce execution_timeout for tasks run by a language SDK #74869
• related: execution_timeout on @task.stub tasks is not enforced #74716
• related: Language SDK runtimes do not handle execution_timeout in the task process #74717
• related: Load the serialized Dag in ti_run only for stub tasks #75795
• related: Child processes of language SDK tasks survive when the task is stopped #75533

---

Follow-up to #74869, reworked after review.

Since #75795, ti_run only loads the serialized task for stub tasks, so a Python task no longer gets TIRunContext.execution_timeout. The supervisor arms its backstop from that value, so on main the backstop is never armed for Python tasks. The runner's SIGALRM still works for normal timeouts, but if the task is stuck in native code (or segfaults), nothing kills it.

Going back to reading the run context for Python tasks isn't a good fix either. Before #75795 that's what we did, and the supervisor's value could differ from the task.execution_timeout the runner uses (different [core] default_task_execution_timeout on Dag processor and worker, Dag file changed after serialization, pre_execute, mapped expand(execution_timeout=...)). If the supervisor's was shorter, it sent SIGTERM before the runner raised AirflowTaskTimeout, and on_kill, retries, callbacks and listeners were skipped.

So this PR takes the timeout from wherever the task is actually defined:

• Python tasks: the runner uses task.execution_timeout from the parsed Dag file (same as 3.3) and sends it to the supervisor with SetExecutionTimeout(timeout_seconds). This is the message from Enforce execution_timeout from the task supervisor side #73806 that shipped in 3.4.0, replacing StartExecutionTimeout. Supervisor and runner now use the same value.
• Language SDK tasks: ti_run still returns execution_timeout only for is_stub tasks, as on main. That path already loads the serialized task for arg_bindings, so no extra cost. The runtime has no way to read the timeout set on a Python @task.stub (execution_timeout on @task.stub tasks is not enforced #74716), so it has to come from the server.

TIRunContext.execution_timeout is now sent as-is, with the same meaning the Python runner gives task.execution_timeout:

Value	Meaning
unset	task doesn't set one
0	no timeout
negative	already elapsed, the supervisor only gives the grace period

The server no longer clamps negative values to 0, since runtimes need 0 to mean "no timeout" rather than "not set".

#74869 was reverted from v3-4-test (#75469), so 3.4.0 doesn't have TIRunContext.execution_timeout. The schema changes move from 2026-10-30 to 2027-02-28:

• Execution API: AddExecutionTimeoutToTIRunContext moved to v2027_02_28.py.
• Supervisor schema: new v2027_02_28.py with AddExecutionTimeoutToSupervisorTIRunContext, because StartupDetails carries the same TIRunContext and the runtimes will read the field in the follow-ups. SetExecutionTimeout is unchanged from 3.4.0, so it needs no entry.
• Go and TypeScript SDKs built from main need Airflow 3.5.

Follow-ups

With this PR, a stub without execution_timeout still gets the Dag processor's [core] default_task_execution_timeout, same as main today. The follow-ups move that default to the worker, like for Python tasks:

1. Standard provider / core: @task.stub stores no execution_timeout when the Dag doesn't set one, and 0 when it's set to None (default_args count as set). client_defaults in the serialized Dag stop carrying execution_timeout, otherwise the Dag processor's default comes back on read.
2. Go SDK: the coordinator passes the worker's default to the runtime as AIRFLOW__CORE__DEFAULT_TASK_EXECUTION_TIMEOUT. The runtime uses the run context value, falls back to that default, and sends SetExecutionTimeout.
3. TypeScript SDK: same as 2.

1 drops the default that Go/TS tasks get today and 2/3 bring it back from the worker, so 1 should go in with or after 2 and 3. All three should land in 3.5.0.

Known limitations

• Language SDK tasks mapped with expand(execution_timeout=...) get no value. No language SDK supports mapped tasks yet anyway.
• The Java SDK doesn't handle execution_timeout yet. After follow-up 1, a Java task without a timeout won't get the Dag processor's default anymore.

[jason810496]

I agreed we should make the getter of execution time on both side consistent.

A [core] default_task_execution_timeout set only on the worker no longer applies to Python tasks. It now comes from where the Dag is serialized, same as for language SDK tasks.

Would it be better to still respect the config if we set then fallback to the field set on the TIRunContext?

Since this will introduce breaking change for user IIUC.

We intentionally avoid the value being passed from runner to supervisor in previous PR, but it's fine for the supervisor to respect the config set at the worker.

[ChickenBenny]

Thanks! Agreed that a worker-only [core] default_task_execution_timeout silently stopping is the breaking part worth avoiding.

I didn't make the worker config take precedence over TIRunContext, because the worker cannot tell whether the run-context value is the default or set by the Dag: the Dag processor writes the resolved default into the serialized task itself, so it looks the same as an explicit execution_timeout. "Worker config first" would override explicit values, and avoiding that would need a new TIRunContext field plus a Cadwyn version change.

Instead, the worker config is used only when the run context has no timeout:

• Runner: uses TIRunContext.execution_timeout. If it is None and the re-parsed task carries the worker's default, it uses the worker's default and logs a deprecation warning.
• Supervisor: on StartExecutionTimeout, if the run context has no timeout, it reads its own [core] default_task_execution_timeout. Both sides compute max(default.total_seconds(), 0.0) from the same config, so they always agree, and StartExecutionTimeout still carries no value.

The "equals the worker's default" check matters: the re-parsed value can also come from a Dag file changed since it was serialized, or from pre_execute, which the supervisor cannot see. In those cases there is no fallback, so runner and supervisor stay consistent.

[jason810496]

Thanks for tracing this. I agree #74869 has a bug when the Dag processor and the worker resolve execution_timeout differently. It is already on v3-4-test (#75376) but not in any tag yet, so we should fix it before the next 3.4 pre-release.

On v3-4-test, the runner arms SIGALRM from task.execution_timeout, while the supervisor arms from TIRunContext.execution_timeout. When the server value is smaller (a different default_task_execution_timeout, a Dag file changed after serialization, pre_execute, or a mapped expand(execution_timeout=...)), the supervisor sends SIGTERM before the runner raises AirflowTaskTimeout, so on_kill, the retry policy, callbacks and listeners are skipped.

Making the run context the only source for Python tasks fixes that mismatch, but a few cases still end up with no timeout or two different ones:

• expand(execution_timeout=...): the server only reads partial_kwargs, so it sends None. Values from .map() or from upstream XComs can't be resolved on the server at all.
• A core that doesn't send the field: tasks that set execution_timeout themselves get no timeout. The worker-default fallback only covers tasks whose value equals the worker default.
• BaseAsyncOperator still uses task.execution_timeout in asyncio.wait_for, so it can disagree with SIGALRM.

How about splitting the two paths instead?

• Python tasks: task.execution_timeout is the only source. The runner arms SIGALRM with it and sends the same value to the supervisor.
• Language SDK tasks: keep using TIRunContext.execution_timeout, armed by the coordinator as today.

# comms.py
class StartExecutionTimeout(BaseModel):
    timeout_seconds: Annotated[float, Field(gt=0)]
    type: Literal["StartExecutionTimeout"] = "StartExecutionTimeout"


# supervisor.py
def _handle_start_execution_timeout(self, msg, log, req_id):
    self._execution_timeout_seconds = msg.timeout_seconds
    self._arm_execution_timeout()
    return None, {}


# task_runner.py
SUPERVISOR_COMMS.send(StartExecutionTimeout(timeout_seconds=timeout_seconds))

This is close to what #73806 did with SetExecutionTimeout. Both timers then use the same value, and mapped tasks, pre_execute and a worker-only default keep working as before. _resolve_execution_timeout, the supervisor fallback, the create_runtime_ti change and the significant newsfragment can then be dropped, and the TIRunContext.execution_timeout docstring would say it is for tasks run by a language SDK. A language SDK task mapped with expand(execution_timeout=...) would still get None, which we can note as a known limitation.

[jason810496]

Mark as request change in case of accidental merge.

[Andrushika]

Agree with Jason, and sorry for my incorrect suggestion in #74869.

I think we can keep this PR to only "undo" the changes on the Python Task SDK path. Maybe we should also rename StartExecutionTimeout back to SetExecutionTimeout, since it carries the timeout again. That name was approved in #73806 and already shipped in 3.4.0b1 and b2.

One thing we may need to document: with Jason's approach, Python tasks and language SDK tasks read [core] default_task_execution_timeout from different places. For a task that does not set execution_timeout itself:

Config set on	Python task	Language SDK task
Dag processor only (A)	no timeout	A
Workers only (B)	B	no timeout
Both, different values	B	A

For language SDK tasks, the default is filled from the Dag processor's config when the Dag is parsed:

airflow/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py

Line 430 in 99bdeab

DagSerialization.fill_config_defaults(dag.data)

(WOW it's kinda weird to the users isn't it)

[ChickenBenny]

Thanks both! Reworked along Jason's suggestion, so my earlier reply about the worker-default fallback is outdated.

• Python tasks: the runner sends task.execution_timeout to the supervisor with SetExecutionTimeout, renamed back as Andrushika suggested, so both timers use the same value. The fallback and the newsfragment are dropped.
• Language SDK tasks: unchanged.
• The default_task_execution_timeout table is documented in tasks.rst and config.yml.
• Since Enforce execution_timeout for tasks run by a language SDK #74869 was reverted from v3-4-test, its schema changes move to 2027-02-28.

Two things worth a look:

1. Go and TS SDKs built from main now need Airflow 3.5, since a 3.4 supervisor rejects schema 2027-02-28. Is that OK?
2. A language SDK task mapped with expand(execution_timeout=...) still gets None. I listed it as a known limitation.

[jason810496]

After revisit the end to end path, I think the best way is:

1. Make this PR revert the loading serialized Dag part and add the AIRFLOW__CORE__DEFAULT_TASK_EXECUTION_TIMEOUT via coordinator to pass the env to SDK runtime.

airflow/task-sdk/src/airflow/sdk/coordinators/_subprocess.py

Lines 193 to 206 in b19bc48

	def _build_runtime_env() -> dict[str, str]:
	"""Return the environment for a language SDK runtime, with the Airflow config it reads resolved."""
	# A language SDK runtime cannot read Airflow's config, so the options it needs are resolved here.
	# An option without a config default gets the fallback Python uses. StartupDetails arrives too
	# late: logs may already be produced.
	return {
	**os.environ,
	"AIRFLOW__LOGGING__LOGGING_LEVEL": conf.get("logging", "logging_level", fallback="INFO"),
	"AIRFLOW__LOGGING__NAMESPACE_LEVELS": conf.get("logging", "namespace_levels", fallback=""),
	"AIRFLOW__API__BASE_URL": conf.get("api", "base_url", fallback="/"),
	"AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE": str(conf.getboolean("operators", "default_deferrable")),
	"AIRFLOW__TRIGGERER__QUEUES_ENABLED": str(conf.getboolean("triggerer", "queues_enabled")),
	"AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS": conf.get("state_store", "default_retention_days"),
	}

2. Make each SDK emit the SetExecutionTimeout just like the normal Python task does, so that we can respect the value that user explicitly set, worker env and the default value if unset while we won't introduce the inconsistent behavior as mentioned in #75387 (we shouldn't allow this be happened)

Sorry, I didn't realized we will load serialized Dag for every task on execution API server back then.

Go and TS SDKs built from main now need Airflow 3.5, since a 3.4 supervisor rejects schema 2027-02-28. Is that OK?

Sure, it's fine.

A language SDK task mapped with expand(execution_timeout=...) still gets None. I listed it as a known limitation.

Yes, all the SDKs so far haven't supported the mapped task yet.

[jason810496]

As discussed offline, the appropriate direction will be having the API server still return the execution_timeout (but only under the is_stub path. We already load the serialized Dag in the is_stub path at API server, there's no harm to return the execution_timeout, the cost is still the same) since we couldn't resolve the execution_timeout defined "StubOperator" at the SDK runtime.

We could resolve the "execution_timeout" in the backend process (#75184) though, but there's more risk of having the subprocess get stuck before returning the execution_timeout to supervisor.

[uranusjr]

The first sentence in the description seems backwards? It was tru before #75795, but in current main, a Python task's backstop is never armed, and that regression is the reason to merge this PR.

[ChickenBenny]

The first sentence in the description seems backwards? It was tru before #75795, but in current main, a Python task's backstop is never armed, and that regression is the reason to merge this PR.

Thanks for correction. Updated the description: it now starts with the backstop never being armed for Python tasks on main, and keeps the timer mismatch as the reason the runner sends its own value instead of reading the run context again.

[jason810496]

Thanks for the update, LGTM after address the following nits, thanks.

[jason810496]

Thanks.

[shivaam]

LGTM. Tested with ts sdk as well