[ChickenBenny]

• related: Enforce execution_timeout for tasks run by a language SDK #74869
• related: Child processes of language SDK tasks survive when the task is stopped #75533

---

A follow-up to #74869, reworked after review.

With #74869 on main, the Python task runner arms SIGALRM from task.execution_timeout, while the supervisor arms its backstop from TIRunContext.execution_timeout. When the two differ, the supervisor sends SIGTERM before the runner raises AirflowTaskTimeout, so on_kill, the retry policy, callbacks and listeners are skipped. They differ more often than expected: a different [core] default_task_execution_timeout on the Dag processor and the worker, a Dag file changed after serialization, pre_execute, or a mapped expand(execution_timeout=...), which the server cannot resolve and sends as None.

Making the run context the only source does not work either, since the server cannot resolve every value. So each kind of task now has a single source, and both timers use it:

• Python tasks: the runner uses task.execution_timeout, as in 3.3, and sends that value to the supervisor with SetExecutionTimeout(timeout_seconds). This restores the message from Enforce execution_timeout from the task supervisor side #73806, identical to what 3.4.0 ships, in place of StartExecutionTimeout.
• Language SDK tasks: unchanged from Enforce execution_timeout for tasks run by a language SDK #74869. The supervisor uses TIRunContext.execution_timeout and the coordinator arms it once the runtime has the task.

 Python task                                  Language SDK task
 ┌─ Task runner ──────────────────────┐       ┌─ API server ──────────────────────┐
 │ SIGALRM = task.execution_timeout   │       │ TIRunContext.execution_timeout    │
 └────────────────────────────────────┘       └───────────────────────────────────┘
          │ SetExecutionTimeout(timeout_seconds)       │
          ▼                                            ▼
 ┌─ Supervisor ───────────────────────┐       ┌─ Supervisor / coordinator ────────┐
 │ backstop = the same value          │       │ backstop = run context value      │
 └────────────────────────────────────┘       └───────────────────────────────────┘

#74869 was reverted from v3-4-test (#75469), so 3.4.0 does not have TIRunContext.execution_timeout. Its schema changes are therefore moved from 2026-10-30 to 2027-02-28 in this PR, since they ship together with the change above:

• Execution API: AddExecutionTimeoutToTIRunContext moves to v2027_02_28.py.
• Supervisor schema: adds v2027_02_28.py, the first 3.5 supervisor schema version, dated like the Execution API one, with AddExecutionTimeoutToSupervisorTIRunContext. SetExecutionTimeout needs no entry because it matches 3.4.0.

Because the two kinds of task take the timeout from different places, a task that does not set execution_timeout reads [core] default_task_execution_timeout in different places too:

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

This is documented in the Timeouts section of core-concepts/tasks.rst and in the option's description, which recommend setting the same value on the Dag processor and the workers.

---

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

• Yes, Claude Code (Opus 5.5)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[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]

Let's remove this part, since this part should be fixed after the new plan I proposed above.

[jason810496]

We should revert this part from the API server in this PR. The root cause was mentioned in #75791 and #75795 just merged.

airflow/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

Lines 407 to 408 in 840d3a4

	if (execution_timeout := get_execution_timeout(serialized_task)) is not None:
	context.execution_timeout = execution_timeout