TS SDK: bind TaskFlow call arguments by folding names on both sides by jason810496 · Pull Request #73189 · apache/airflow · GitHub
Skip to content

TS SDK: bind TaskFlow call arguments by folding names on both sides - #73189

Draft
jason810496 wants to merge 4 commits into
apache:mainfrom
jason810496:feature/ts-sdk/taskflow-arg-folding
Draft

jason810496 wants to merge 4 commits into
apache:mainfrom
jason810496:feature/ts-sdk/taskflow-arg-folding

Conversation

@jason810496

@jason810496 jason810496 commented Sep 15, 2026

Copy link
Copy Markdown
Member

Why

A Python Dag calls a @task.stub task TaskFlow-style, but the handler 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.

How

  • Consume ti_context.arg_bindings at the dispatch site and hand it to the handler as its parameter.
  • Names bind by folding on both sides, lowercased with underscores removed.
    That is the Go SDK's rule (strings.ToLower(strings.ReplaceAll(name, "_", ""))), so one Python signature binds identically in either SDK,
    and neither needs a rename declared for ordinary snake_case.
  • The bound object is a Proxy so folding happens per read. 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. A null prototype means a Python argument named toString binds like any
    other, and one that was not passed misses rather than resolving to a function.
  • in folds like a read. Object.keys and rest destructuring yield Python's names, since there are no TypeScript-side names to enumerate.
  • An unmatched name logs rather than throws: a destructuring default such as { runId = "manual" } is a legitimate miss and nothing can tell
    one from a typo, so the warning names both the requested name and what the call delivered. A failing task reports the same list.
  • Two Python names that fold to one token fail the task before the handler runs, naming both.
    An XCom-backed binding fails the same way for now, with the getXCom call to write instead; the next PR resolves those.
    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.

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.
@jason810496
jason810496 force-pushed the feature/ts-sdk/taskflow-arg-folding branch from 4f3d936 to 8133d20 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