[github-actions]

• Add tests for language SDK tasks that cannot launch

• Report language SDK runtime launch failures in the task log

• Resolve artifact bundles the same way as native Dag bundles

• Fail coordinator bugs before launch without a retry

• Say why execute_task uses an ExitStack

The lock and the scan roots are entered inside the launch-error guard but
released only after wait() returns, which is the whole reason the stack is
there rather than a plain with.

• Fail an unsupported supervisor schema version without a retry

A worker whose Task SDK does not know the schema version an artifact speaks
cannot learn it by trying again, so the task should not burn its retries on
it. The check also moves out of the native Dag branch and now runs once for
every task, so an artifact found by scanning the bundle is rejected before
the runtime starts rather than by the attrs validator inside start().

• Write a failed launch's runtime output to the task log

The output a runtime wrote before it failed to connect went into the error
message, so the task log rendered it as one event whose continuation lines
carry no timestamp or level. It now goes to the log the way a connected
runtime's output does, one event per line under task.stdout and task.stderr,
and the message stays on one line.

The kept tail grows to 16 KiB, since a JVM or Node stack trace does not fit
in 4 KiB, and says so when it drops earlier output. Draining a runtime that
keeps printing while it never connects is capped, so it cannot hold up the
failure.

• Assert why a runtime that cannot be executed fails

The test matched the path with a wildcard, which hid whether the reason the
OS gave survives the wrap. It now asserts the whole message, and covers a
runtime that exists but cannot be executed as well as one that is missing.

_connect_without_output moves above the class that uses it, where every
other helper in the file sits.

• Keep the runtime's sockets tracked until the launch has finished

The pipes and the accepted connections were untracked as the argument to
_register_pipe_readers, so they were disowned before it ran. A failure in it,
or in _on_child_started after it, left four sockets that nothing closed: the
tracker no longer owned them and the half-built subprocess was discarded.

They now stay with the tracker until the task instance has started, which is
the point 'self' takes them over and closes them in wait().

• Forward a failed launch's output through the supervisor's forwarder

_report_startup_output decoded and logged the lines itself, which repeated
forward_to_log with small differences. It now feeds forward_to_log instead,
so a failed launch and a connected runtime split lines and decode them the
same way.

The loggers to forward to come from WatchedSubprocess._target_loggers, split
out of _get_target_loggers so both callers share it: 'self' does not exist
yet when the runtime fails to connect, and the module logger the tuple ends
with belongs to supervisor.py.

• State the launch-error contract in terms of the task instance

TaskLaunchError said a coordinator raises it "before it starts the runtime",
but a runtime that exits or times out before connecting has already been
spawned. The invariant custom coordinators implement against is that the
task instance has not been started, which is what the supervisor does itself
when it handles the error.

The command builders now also say to raise TaskLaunchError with an explicit
retryable when the implementation knows the answer, rather than leaving it
to be guessed from the exception type.

• Drop the comment on execute_task's ExitStack

• Say what a cleared task does to a failed launch's output

"the log keeps output for a try that never ran" reads as if the output shows
up against the next try. It does not: the log file is rendered per try, so
the lines stay in the file of the try whose launch failed. What happens is
narrower, and worth naming: the supervisor stops without reporting a state,
so the file is never uploaded and is left on the worker.

• Keep a Dag file command's launch error and its retryability

• Expect a launch error when a runtime dies before connecting

The process group test landed on main while this branch was open, against
the RuntimeError _accept_connections raised there. This branch raises
TaskLaunchError instead, which is not a RuntimeError.

---

(cherry picked from commit 21a8a98)

Co-authored-by: ChickenBenny zxc123benny14159@gmail.com
Co-authored-by: ZHE YOU LIU zhu424.dev@gmail.com

[github-actions]

Hi maintainer, this PR was merged without a milestone set.
We've automatically set the milestone to Airflow 3.4.1 based on: merged to version branch
If this milestone is not correct, please update it to the appropriate milestone.

This comment was generated by Milestone Tag Assistant.