TS SDK: resolve upstream XComs for bound TaskFlow arguments by jason810496 · Pull Request #73190 · apache/airflow · GitHub
Skip to content

TS SDK: resolve upstream XComs for bound TaskFlow arguments - #73190

Draft
jason810496 wants to merge 5 commits into
apache:mainfrom
jason810496:feature/ts-sdk/taskflow-xcom-args
Draft

jason810496 wants to merge 5 commits into
apache:mainfrom
jason810496:feature/ts-sdk/taskflow-xcom-args

Conversation

@jason810496

@jason810496 jason810496 commented Sep 15, 2026

Copy link
Copy Markdown
Member

Why

summarize(extract()) means "hand the handler what extract returned", but only literals could bind,
so such an argument failed the task and told the author to pull the XCom by hand.

How

  • Airflow names the upstream task in the binding spec, so the runtime pulls those return_value XComs itself, all of them concurrently:
    a task called with four upstream outputs waits for one round-trip, not four.
  • The whole spec is still validated before anything is pulled, so a binding this SDK cannot honour costs no round-trip,
    and leaves no half-resolved call behind.
  • Arguments resolve before the handler runs, the one stretch of a task's life with nothing else listening for termination,
    so the abort signal cuts the pulls short rather than leaving a killed task to sit out the force-exit grace period.
  • An upstream that pushed no output fails the task, naming both the argument and the task it came from.
    Telling that apart from an upstream that pushed null needs more than getXCom, which answers null for both,
    so the coordinator's own client keeps the found flag the supervisor already sends. Handlers stay typed against TaskClient and never see it.
  • A Python int beyond the range a JavaScript number holds exactly is refused rather than bound.
    Airflow stamps format: "int64" on the argument, and the value arrives with its low digits already lost, so nothing downstream could notice.

Note

ADR-0001 decision 6 is unchanged: being upstream is not being passed.
A >> dependency still declares order only, and a value the call did not pass is still read explicitly.


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

`ctx` and `client` arrived as properties of the handler's only parameter, so
they occupied the top level of the namespace that TaskFlow argument binding
needs. Every typed handler would have had to spell its parameter
`TArgs & TaskHandlerArgs` to say so.

Both are properties of the invocation rather than of the author's data, so they
move to getters backed by an `AsyncLocalStorage` store the runtime installs
around the single dispatch site. The store follows the handler across every
`await` and into every promise it creates, so a helper several frames deep
reads it without anything being threaded through. Both throw outside a handler.

The storage is keyed on a global symbol, as the serve latch already is: two
resolved copies of the package would otherwise hold one storage each, and a
handler reaching through the copy that is not running the task would find
nothing in scope.

`TaskHandlerArgs` and the `TaskHandler` type that took it are gone. The handler
function type is now `TaskFunction`, which frees the `TaskHandler` name for the
class that binds a handler to the Python-owned task it implements.
A bundle process was described entirely in terms of Dags: `DagRegistry` held
them, `serveDags(registry)` served them, and the pack error told an author to
pass them to `serveDags(new DagRegistry(...))`. A mixed-language bundle
provides task handlers for a Dag that Python owns and registers no Dag at all,
so all three are about to be wrong.

The collection becomes a `Bundle`, and it serves itself.
`register(...items: Registerable[])` is its one registration verb, over a union
that gains an arm per kind rather than a second verb per kind. Today the union
has one arm, `Dag`, and registering keeps the path it already had.

`bundle.serve()` replaces the free function. The existing intent was that Dag
authors reach the runtime through one call and never name the coordinator; a
method on the object that already holds everything keeps that while dropping
the function, and the coordinator subpath now exports no entry point at all.
The one-shot serve latch stays in the runtime with the sockets it protects.

The guard `serveDags` performed on its argument survives as a guard on the
receiver, which is where it still has something to catch: `const { serve } =
bundle` detaches the method, and a bundle from a second resolved copy is
reported by its cause rather than as a missing private field.

`sdk/bundle.ts` and `coordinator/runtime.ts` now reference each other. Both
directions are hoisted function declarations referenced from inside function
bodies, so neither module touches the other's bindings while it is still
evaluating, and the metadata path in the packed example bundle exercises it.

`tests/cli/fixtures/bundle-v1.mjs` is regenerated. That golden bundle embeds
the byte ranges and SHA-256 digest of `empty-entry.ts`, which this change
rewrites onto the new surface, so its header moves with it.
A mixed-language Dag is declared in Python: the `@task.stub` per task, the
queue that routes it to the Node coordinator, and the order between them all
live there. The SDK said otherwise. The only way to attach a handler was
`new Dag(dagId)` plus `dag.task(taskId, fn)`, so every mixed-language bundle
constructed a Dag object for a Dag it does not own.

`new TaskHandler(dagId, taskId, handler)` binds a function to the task it
implements and carries nothing else: no schedule, no task order, no dag_id of
its own. Both ids are written out, so nothing depends on a function name a
bundler is free to rename or inline. A handler is a value with no call
signature, so wiring one the way a natively declared task is wired is a compile
error rather than a runtime throw.

`Dag` becomes exclusively the native case, and a dag_id is one or the other: a
native Dag attaches its own tasks, so registering a handler for one is rejected
rather than silently becoming a second, disagreeing source for its task list.

`Registerable` gains its second arm rather than the bundle gaining a second
verb, so one `register` call still lists everything a bundle provides in any
mixture. What that changes underneath is the key: a bundle now holds one entry
per dag_id and dispatches on the `(dag_id, task_id)` pair, so one bundle can
provide for several Dags and the same task_id under two of them is two
different handlers. The example and the end-to-end test both exercise that,
with `typescript_example` and the new `typescript_taskflow_example` each
declaring a `build_message` served from one bundle.mjs.
A Python Dag declares a task that runs in TypeScript with `@task.stub` and
calls it TaskFlow-style, but the handler could not see those arguments: it was
called with nothing, so the call site was decoration and its values had to be
hardcoded in the handler or re-fetched from XCom. Airflow already delivers that
call site as an ordered, named spec in the task's run context.

The names are the problem worth solving. Python spells a parameter
`region_code` and TypeScript wants to read it as `regionCode`, and making an
author declare that for every ordinary snake_case parameter would be a tax on
the common case. So names bind by folding on both sides, lowercased with
underscores removed, which is exactly the Go SDK's rule. One Python signature
then binds identically in either SDK with nothing declared.

Folding on read rather than up front is why the bound object is a `Proxy`: the
SDK sees Python's names and cannot know which spelling a handler will
destructure, so no guess at a TypeScript name is ever materialized. `in` folds
like a read, and `Object.keys` and rest destructuring yield Python's names. The
object has a null prototype, so a Python argument named `toString` binds like
any other and one that was not passed misses rather than resolving to a
function.

An unmatched name logs rather than throws, since a destructuring default is a
legitimate miss and nothing can tell one from a typo. The warning names both
the requested name and what the call delivered, and a failing task reports the
same list, because a handler that destructured an argument under a name nothing
folds to gets no error of its own.

Two Python names that fold to one token do fail the task, before the handler
runs and naming both: neither could be reached, and picking either silently
would hand the handler the wrong value. An XCom-backed binding fails the same
way for now, with the getXCom call to write instead. Anything else the SDK
cannot honour fails rather than being dropped, because an unbound argument
reaches the handler as `undefined` and corrupts its output instead of stopping
it.
A Python Dag calling a TypeScript task TaskFlow-style, as in
`summarize(extract())`, means "hand the handler what extract returned". But the
runtime could only bind literals: an argument taking an upstream task's output
failed the task and told the author to pull the XCom by hand, so the call site
stopped being the contract exactly where it mattered most.

Airflow names the upstream task in the binding spec, so the runtime pulls those
outputs itself, all of them at once, since a task called with four upstream
outputs should wait for one round-trip rather than four. The whole spec is
still checked before anything is pulled, so a binding this SDK cannot honour
costs no round-trip.

They resolve before the handler is called, which is also the one stretch of a
task's life with nothing else listening for termination, so the task's abort
signal now cuts them short instead of leaving a killed task to sit out the
force-exit grace period.

An upstream that pushed no output fails the task, naming both the argument and
the task it came from: a task that returns nothing pushes no XCom, and an
unbound argument corrupts the handler's output rather than stopping it. Telling
that apart from an upstream that pushed null takes more than the task client's
JS-friendly `getXCom`, which answers null for both, so the coordinator's own
client keeps the found flag the supervisor already sends. Handlers stay typed
against `TaskClient` and never see it.

A Python `int` beyond the range a JavaScript number holds exactly is refused
for the same reason: Airflow stamps `format: "int64"` on the argument, and the
value arrives with its low digits already lost, so binding it would hand the
handler a different number from the one the Dag produced.

What does not change is ADR-0001 decision 6: an upstream's return value is not
a bound argument unless the call passes it.
@jason810496
jason810496 force-pushed the feature/ts-sdk/taskflow-xcom-args branch from f108499 to f443eb5 Compare September 15, 2026 13:11
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant