From 556d2d7ce440c06bfd4850b3875a981c1b25fec8 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Tue, 25 Aug 2026 05:37:32 +0000 Subject: [PATCH 1/6] Add TypeScript SDK ADRs for TaskFlow argument binding on mixed-language and native Dags The cross-SDK decisions on argument binding and the TaskFlow Dag DSL were settled against Go and Java, both of which can recover a task function's parameter list, Go with reflect at run time, Java with an annotation processor at compile time. TypeScript can do neither: its idiomatic handler takes a single destructured object whose parameter positions do not survive to run time. That constraint changes the answer to several questions the earlier records treat as settled, so the reasoning needs to live somewhere reviewers and future SDK authors can find it. The mixed-language argument-binding surface and native-Dag authoring are different enough problems to warrant separate ADRs. ADR-0001 covers wiring a TypeScript task body under a Python-declared Dag (MixedLangDag); ADR-0002 covers a Dag whose graph and task bodies are both TypeScript (Dag). Each leads with the call-site code a user actually writes, and each records the ctx/client-as-getters question as open rather than settled, since resolving it would also reshape the binding mechanism and open the door to positional wiring. --- ts-sdk/adr/0001-mixed-lang-dag-interface.md | 89 ++++++++++++ ts-sdk/adr/0002-native-dag-interface.md | 143 ++++++++++++++++++++ 2 files changed, 232 insertions(+) create mode 100644 ts-sdk/adr/0001-mixed-lang-dag-interface.md create mode 100644 ts-sdk/adr/0002-native-dag-interface.md diff --git a/ts-sdk/adr/0001-mixed-lang-dag-interface.md b/ts-sdk/adr/0001-mixed-lang-dag-interface.md new file mode 100644 index 0000000000000..a2490f288c5ea --- /dev/null +++ b/ts-sdk/adr/0001-mixed-lang-dag-interface.md @@ -0,0 +1,89 @@ + + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. + +A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). + +## Decision + +TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. + +```ts +const dag = new MixedLangDag("etl"); + +interface TransformArgs { + region_code: string; + threshold: number; +} + +async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { + const rows = await client.getXCom({ key: "return_value", taskId: "extract" }); + if (rows === null) { + throw new Error(`task ${ctx.taskId} has no upstream row count to transform`); + } + const passed = rows >= threshold; + await client.setXCom({ key: "region", value: region_code }); + return { region_code, passed }; +} + +dag.task("transform", transform); +``` + +Renaming a Python name that isn't idiomatic TypeScript is ordinary destructuring, not a separate mechanism: + +```ts +async function report({ run_label: runLabel }: { run_label: string } & TaskHandlerArgs) { + if (runLabel !== "nightly") { + throw new Error(`expected run label "nightly" but got "${runLabel}"`); + } +} + +dag.task("report", report); +``` + +### How + +- `MixedLangDag.task()` returns a plain `TaskRef`, not a callable factory. There is nothing to wire, since the Python file already owns task order. Calling it the way a native `Dag`'s factory is called (`transform()`) is a compile error, since `TaskRef` has no call signature, not a runtime throw. +- Wire names match the Python parameter names character for character, with no case- or separator-insensitive fallback. Renaming happens once, at the destructuring site. +- `ctx` and `client` are reserved, permanently: bound arguments are merged flat into the same object alongside them, and a bound name that collides with either fails the task at dispatch. +- An upstream's return value is not delivered as a bound argument. Read it explicitly via `client.getXCom({ key: "return_value", taskId: "..." })`. +- `tsc` cannot check a handler's destructuring pattern against the Python call site. A typo binds `undefined` silently; the runtime logs the bound names at dispatch and includes them in a handler-failure message, so the mismatch is diagnosable from the task log. + +## Open Questions + +- Should the decoded bindings also be exposed as a public, positional/raw accessor (name/value pairs, no interface required), or should that stay an internal runtime detail? +- Should `ctx` and `client` become explicit getter functions (`getClient()`, `getContext()`) instead of arguments merged into the handler's object? (See [ADR-0002](0002-native-dag-interface.md) for where this same question resurfaces on the native-Dag side.) + +## Consequences + +- One binding mechanism serves every mixed-language handler; there is no second syntax to keep in sync. +- The Python call site stays the single source of data-flow wiring for mixed-language Dags. +- `MixedLangDag` replaces an earlier `isMixedLanguageDag` spec flag on a single `Dag` class. The authoring class itself states the mode, so a mixed-language task that's wired incorrectly fails to compile under `tsc` instead of throwing at runtime. +- The public surface is not final until the two open questions above are resolved. diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md new file mode 100644 index 0000000000000..184f6c1f63e49 --- /dev/null +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -0,0 +1,143 @@ + + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: + +```ts +const dag = new Dag("ts_etl"); + +const extract = dag.task("extract", async ({ client }): Promise => { + const rows = 42; + await client.setXCom({ key: "row_count", value: rows }); + return rows; +}); + +interface TransformArgs { + extracted: number; +} + +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); + +interface LoadArgs { + transformed: number; +} + +const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { + if (transformed <= 0) { + throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); + } + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load({ transformed: transform({ extracted: extract() }) }); +``` + +The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task can't be silently left out of the graph. + +## If `ctx`/`client` were real getter methods instead of injected arguments + +The intersection type above, `TransformArgs & TaskHandlerArgs`, exists for one reason: today's handler signature carries `ctx`/`client` as arguments, so a handler that wants type safety on its own data has to say so explicitly. If `ctx`/`client` came from getter functions instead, a handler's parameter type would be exactly its own data: + +```ts +// Before: ctx/client share the handler's one argument object, so +// TransformArgs must be intersected with TaskHandlerArgs. +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); +``` + +```ts +// After: ctx/client come from getters, so the handler's parameter type is +// exactly its own data, with no TaskHandlerArgs intersection needed. +import { getClient } from "@apache-airflow/ts-sdk"; + +const transform = dag.task("transform", async ({ extracted }: TransformArgs) => { + const client = getClient(); + await client.setXCom({ key: "doubled", value: extracted * 2 }); + return extracted * 2; +}); +``` + +Open question this ADR does not resolve: + +- What backs `getClient()`/`getContext()` at run time. A Node `AsyncLocalStorage` scoped to the handler's execution is the likely mechanism, but confirming it survives every `await` inside a handler, and any user-spawned concurrency, is a runtime-dispatch question, not a call-site question. This ADR only proposes the shape. + +If that question is answered, the implications are: + +- The top-level argument namespace closes automatically: `ctx`/`client` can never collide with a Dag author's own parameter name, because they no longer share an object with one. +- A handler becomes directly unit-testable with a plain data argument. No `TaskHandlerArgs` fixture is needed, and there's no risk of forgetting to intersect it in. +- The wiring surface (`TaskInputs`, the mapped type checked at the call site) is unaffected either way, since it already derives from the handler's own declared parameter type, whichever shape that type takes. + +### This also reopens positional wiring + +The "after" handler above still takes one object (`{ extracted }: TransformArgs`), out of habit. But the *reason* wiring is named rather than positional today is that a handler takes one destructured object, so `Parameters` is always a one-element tuple with no per-field position to check against. That reason goes away once data no longer has to share an object with `ctx`/`client`: a handler can drop the object entirely and take its arguments positionally, the way an ordinary TypeScript function does. + +```ts +// Before (named wiring, current): the wiring object's keys match each +// handler's destructured object. +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); +const load = dag.task("load", async ({ transformed }: LoadArgs & TaskHandlerArgs) => { + /* ... */ +}); +load({ transformed: transform({ extracted: extract() }) }); +``` + +```ts +// After (positional wiring, hypothetical): each handler takes its data +// positionally, so `Parameters` is a real tuple (`[number]` +// for both transform and load), and a positional TaskFactory can type-check +// a call against it, the same way today's `Wiring` type-checks a +// named one. +const extract = dag.task("extract", async () => 42); +const transform = dag.task("transform", async (extracted: number) => extracted * 2); +const load = dag.task("load", async (transformed: number) => { + const client = getClient(); + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load(transform(extract())); +``` + +This is a second, independent decision stacked on top of the getters question above. A handler could still take one object for its data even without `ctx`/`client` in it, purely for self-documenting call sites once there are several arguments. Positional wiring only follows if handlers *also* drop the object, and it trades that self-documentation away at every multiple-argument call site. Neither this ADR nor the getters question settles it. It's recorded because positional wiring was previously ruled out entirely, for a reason that no longer holds once `ctx`/`client` are getters. + +## Consequences + +- One authoring surface (`dag.task()` + factory) covers both the graph and each task's arguments. +- `DagSpec`/`TaskSpec` are unaffected by this ADR. They are the language-neutral Dag configuration surface, not the argument-binding one. +- Whether `ctx`/`client` stay injected arguments or become getters is still open; adopting getters is a real ergonomics win but depends on the runtime-dispatch question above. +- If getters land, positional data parameters become possible for the first time, and with them, positional wiring (`load(transform(extract()))`). Adopting that is a further, separate, unresolved trade-off against today's named, self-documenting convention. From de939e2dc3c1b7483cd2720aa96091bcfa7040e1 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Tue, 8 Sep 2026 05:47:32 +0000 Subject: [PATCH 2/6] Record the reviewed TypeScript SDK task and Dag authoring interfaces The review on #72047 found three things these ADRs described but should not have. The SDK's context and client shared a handler's single argument object, so every typed handler had to intersect its own data type with the SDK's. A mixed-language Dag was authored through a second Dag class, when TypeScript declares no Dag at all in that mode. And a graph whose every edge carried data had no way to express an edge that carries none, which left a fan-in of order-only dependencies unanswerable and pushed the examples into nesting that stops reading well before a Dag grows large. Recording what the review arrived at keeps these documents a reference for the interface the SDK will ship, before either interface reaches users. --- ts-sdk/adr/0001-mixed-lang-dag-interface.md | 79 +++++++----- ts-sdk/adr/0002-native-dag-interface.md | 128 ++++++-------------- 2 files changed, 88 insertions(+), 119 deletions(-) diff --git a/ts-sdk/adr/0001-mixed-lang-dag-interface.md b/ts-sdk/adr/0001-mixed-lang-dag-interface.md index a2490f288c5ea..18496784ab567 100644 --- a/ts-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/ts-sdk/adr/0001-mixed-lang-dag-interface.md @@ -17,11 +17,11 @@ under the License. --> -# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface +# ADR-0001: Mixed-Lang Dag — TypeScript Task Handler Interface ## Status -Proposed +Proposed. Revised after the review on #72047. ## Context @@ -29,61 +29,82 @@ The Python `@task.stub` call site already defines task data flow. TypeScript tas This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. -A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). +A mixed-language Dag declares its structure in Python, and TypeScript supplies task bodies and nothing else. So TypeScript registers **task handlers**, not a Dag: it owns no dag_id, no schedule, no task order. `Dag` is exclusively the native case, covered in [ADR-0002](0002-native-dag-interface.md). ## Decision -TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. +A handler is a plain function of its own data. The SDK's context and client come from getters, and each handler is registered against the dag_id/task_id pair Python already owns. ```ts -const dag = new MixedLangDag("etl"); +import { DagRegistry, getClient, getContext, serveDags } from "apache-airflow-ts-sdk"; interface TransformArgs { - region_code: string; + regionCode: string; threshold: number; } -async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { +async function transform({ regionCode, threshold }: TransformArgs) { + const client = getClient(); const rows = await client.getXCom({ key: "return_value", taskId: "extract" }); if (rows === null) { - throw new Error(`task ${ctx.taskId} has no upstream row count to transform`); + throw new Error(`task ${getContext().taskId} has no upstream row count to transform`); } - const passed = rows >= threshold; - await client.setXCom({ key: "region", value: region_code }); - return { region_code, passed }; + return { regionCode, passed: rows >= threshold }; } -dag.task("transform", transform); +const registry = new DagRegistry(); +registry.registerTaskHandler("etl", "transform", transform); +await serveDags(registry); ``` -Renaming a Python name that isn't idiomatic TypeScript is ordinary destructuring, not a separate mechanism: +### Wire names + +By default, arguments are transformed automatically: the SDK normalizes the names on **both** sides — lowercased, separators removed — and matches those. Python's `region_code` binds to a handler's `regionCode`, `Name` binds to `name`, and `s3_uri` binds to `s3Uri`, with nothing declared on either side. The Go SDK normalizes the same way (`strings.ToLower(strings.ReplaceAll(name, "_", ""))`), so one Python signature binds identically in either SDK. ```ts -async function report({ run_label: runLabel }: { run_label: string } & TaskHandlerArgs) { - if (runLabel !== "nightly") { - throw new Error(`expected run label "nightly" but got "${runLabel}"`); - } +// Python: def transform(region_code: str, s3_uri: str, threshold: float) +async function transform({ regionCode, s3Uri, threshold }: TransformArgs) { + // ... } +``` -dag.task("report", report); +A user can also specify the binding explicitly, with `withArgNames`: its first argument is the argument mapping, and its second is the handler itself. Normalization only absorbs spelling differences, so this is what to reach for when a handler wants a name the Python side never used — a clearer word, or a TypeScript reserved word like `enum`: + +```ts +const report = withArgNames( + { label: "run_label" }, + async ({ label, transformed }: ReportArgs) => { + if (label !== "nightly") { + throw new Error(`expected run label "nightly" but got "${label}"`); + } + }, +); + +registry.registerTaskHandler("etl", "report", report); ``` +The map's keys are checked against the handler's own parameter type, so `{ labl: "run_label" }` is a compile error naming the right key. Its values are Python names, which `tsc` cannot see and does not check. + ### How -- `MixedLangDag.task()` returns a plain `TaskRef`, not a callable factory. There is nothing to wire, since the Python file already owns task order. Calling it the way a native `Dag`'s factory is called (`transform()`) is a compile error, since `TaskRef` has no call signature, not a runtime throw. -- Wire names match the Python parameter names character for character, with no case- or separator-insensitive fallback. Renaming happens once, at the destructuring site. -- `ctx` and `client` are reserved, permanently: bound arguments are merged flat into the same object alongside them, and a bound name that collides with either fails the task at dispatch. +- `registry.registerTaskHandler(dagId, taskId, handler)` is the second registration verb, beside `registry.registerDag(...dags)` for the native case ([ADR-0002](0002-native-dag-interface.md)). It takes one handler at a time because each needs its own dag_id/task_id pair, where a Dag carries its own id and several can be registered in one call. task_id must match the `@task.stub` id exactly and is always written out: deriving it from the handler's function name would silently rebind the handler when the function is renamed. +- A handler registered this way has no factory to call, so wiring a mixed-language task the way a native one is wired (`transform()`) is a compile error rather than a runtime throw. That is the guarantee an earlier draft's separate `MixedLangDag` class existed to provide. +- `getClient()` and `getContext()` read from an `AsyncLocalStorage` store the runtime wraps around the handler call, and throw outside a handler. TypeScript keeps type and value namespaces separate, so `getContext()` coexists with the `TaskContext` type without either being renamed. +- The bound argument object is a `Proxy`. Destructuring a name triggers a lookup that folds that name on demand and matches it against the folded wire names, so binding needs nothing declared anywhere, and an entry in `withArgNames` takes precedence over folding. +- An unmatched name is logged with both the requested name and the names actually delivered. It cannot throw: a destructuring default (`{ runId = "manual" }`) is a legitimate miss, and the runtime cannot tell one from a typo. +- `in` folds like a read. `Object.keys` and rest destructuring (`{ ...rest }`) yield Python's names, since the SDK has no TypeScript-side names to enumerate. +- Two Python names that fold to the same token fail the task at dispatch, naming both. - An upstream's return value is not delivered as a bound argument. Read it explicitly via `client.getXCom({ key: "return_value", taskId: "..." })`. -- `tsc` cannot check a handler's destructuring pattern against the Python call site. A typo binds `undefined` silently; the runtime logs the bound names at dispatch and includes them in a handler-failure message, so the mismatch is diagnosable from the task log. -## Open Questions +## Alternatives -- Should the decoded bindings also be exposed as a public, positional/raw accessor (name/value pairs, no interface required), or should that stay an internal runtime detail? -- Should `ctx` and `client` become explicit getter functions (`getClient()`, `getContext()`) instead of arguments merged into the handler's object? (See [ADR-0002](0002-native-dag-interface.md) for where this same question resurfaces on the native-Dag side.) +- **Annotating the Python name on the field with a phantom type** (`type Arg = T & { readonly __arg?: N }`), the way Java's `@ArgName` annotates a `TaskInput` field. Rejected: an `interface` is erased, so the annotation cannot reach dispatch and needs a runtime companion regardless. It also has two silent failure modes — `Arg` collapses to a required `string`, because `undefined & object` is `never`, and the phantom key appears in `keyof` for an object-valued argument. +- **Reading the expected names from `handler.toString()`** and parsing the destructuring pattern. Property names do survive bundling, but a handler whose parameter is not destructured (`async (a: ReportArgs) => a.runLabel`) exposes no names at all, so the check would disappear silently for ordinary code. ## Consequences -- One binding mechanism serves every mixed-language handler; there is no second syntax to keep in sync. -- The Python call site stays the single source of data-flow wiring for mixed-language Dags. -- `MixedLangDag` replaces an earlier `isMixedLanguageDag` spec flag on a single `Dag` class. The authoring class itself states the mode, so a mixed-language task that's wired incorrectly fails to compile under `tsc` instead of throwing at runtime. -- The public surface is not final until the two open questions above are resolved. +- One binding mechanism serves every mixed-language handler, and the Python call site stays the single source of data-flow wiring. +- Folding matches the Go SDK's rule, so the same Python signature binds the same way in either SDK, and neither one needs a rename declared for ordinary snake_case parameters. +- `TaskHandlerArgs` is removed, `Dag` no longer serves the mixed-language case, and the registry's `register(...dags)` becomes `registerDag(...dags)` so the two verbs name what they take. All are shipped API (`src/index.ts`, and the `new Dag(...)` + `dag.task(...)` pattern in `README.md`, `docs/index.md`, and `example/src/main.ts`), so this breaks 0.1.0-beta1 authors and those call sites change with the implementation. The package's own status line already reads "API may change". +- A handler is directly unit-testable with a plain data argument: no `TaskHandlerArgs` fixture, and no intersection to remember. +- `withArgNames` is needed only for a genuine rename, never for a spelling difference, so most handlers never mention it. diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md index 184f6c1f63e49..ca1c7dea418ea 100644 --- a/ts-sdk/adr/0002-native-dag-interface.md +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -21,123 +21,71 @@ ## Status -Proposed +Proposed. Revised after the review on #72047. ## Context -A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case registers task handlers instead of a Dag, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). Both share the protocol substrate recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). ## Decision -`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow uses for `load(transformed=transform(...))`. A handler is a function of its own data; `getClient()` and `getContext()` supply the rest. ```ts -const dag = new Dag("ts_etl"); - -const extract = dag.task("extract", async ({ client }): Promise => { - const rows = 42; - await client.setXCom({ key: "row_count", value: rows }); - return rows; -}); +import { Dag, DagRegistry, getClient, serveDags } from "apache-airflow-ts-sdk"; -interface TransformArgs { - extracted: number; -} +const dag = new Dag("ts_etl"); -const transform = dag.task( - "transform", - async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, -); +const extract = dag.task("extract", async (): Promise => 42); -interface LoadArgs { - transformed: number; -} +const transform = dag.task("transform", async ({ extracted }: { extracted: number }) => extracted * 2); -const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { - if (transformed <= 0) { - throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); - } - await client.setXCom({ key: "loaded", value: transformed }); +const load = dag.task("load", async ({ transformed }: { transformed: number }) => { + await getClient().setXCom({ key: "loaded", value: transformed }); }); -load({ transformed: transform({ extracted: extract() }) }); -``` - -The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task can't be silently left out of the graph. - -## If `ctx`/`client` were real getter methods instead of injected arguments - -The intersection type above, `TransformArgs & TaskHandlerArgs`, exists for one reason: today's handler signature carries `ctx`/`client` as arguments, so a handler that wants type safety on its own data has to say so explicitly. If `ctx`/`client` came from getter functions instead, a handler's parameter type would be exactly its own data: +const extracted = extract(); +const transformed = transform({ extracted }); +const loaded = load({ transformed }); -```ts -// Before: ctx/client share the handler's one argument object, so -// TransformArgs must be intersected with TaskHandlerArgs. -const transform = dag.task( - "transform", - async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, -); +const registry = new DagRegistry(); +registry.registerDag(dag); +await serveDags(registry); ``` -```ts -// After: ctx/client come from getters, so the handler's parameter type is -// exactly its own data, with no TaskHandlerArgs intersection needed. -import { getClient } from "@apache-airflow/ts-sdk"; - -const transform = dag.task("transform", async ({ extracted }: TransformArgs) => { - const client = getClient(); - await client.setXCom({ key: "doubled", value: extracted * 2 }); - return extracted * 2; -}); -``` +One statement per task, with each ref named, is the form to write. Nesting the calls (`load({ transformed: transform({ extracted: extract() }) })`) is legal and equivalent, but it is shorthand for a two-task chain, not the general shape: a Dag of twenty tasks reads as twenty flat statements, never as a twenty-deep expression. -Open question this ADR does not resolve: +The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle through arguments is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task cannot be silently left out of the graph. -- What backs `getClient()`/`getContext()` at run time. A Node `AsyncLocalStorage` scoped to the handler's execution is the likely mechanism, but confirming it survives every `await` inside a handler, and any user-spawned concurrency, is a runtime-dispatch question, not a call-site question. This ADR only proposes the shape. +### Order-only edges: `>>` and `<<` -If that question is answered, the implications are: +An edge that carries no data has no key to put in the wiring object, so it is drawn directly between refs. `before` and `after` are the TypeScript pair for Python's `>>` and `<<`: -- The top-level argument namespace closes automatically: `ctx`/`client` can never collide with a Dag author's own parameter name, because they no longer share an object with one. -- A handler becomes directly unit-testable with a plain data argument. No `TaskHandlerArgs` fixture is needed, and there's no risk of forgetting to intersect it in. -- The wiring surface (`TaskInputs`, the mapped type checked at the call site) is unaffected either way, since it already derives from the handler's own declared parameter type, whichever shape that type takes. +```ts +const cleaned = cleanup(); -### This also reopens positional wiring +cleaned.after(loaded, transformed); // [loaded, transformed] >> cleaned +loaded.before(cleaned); // loaded >> cleaned +``` -The "after" handler above still takes one object (`{ extracted }: TransformArgs`), out of habit. But the *reason* wiring is named rather than positional today is that a handler takes one destructured object, so `Parameters` is always a one-element tuple with no per-field position to check against. That reason goes away once data no longer has to share an object with `ctx`/`client`: a handler can drop the object entirely and take its arguments positionally, the way an ordinary TypeScript function does. +Both are variadic, so one call fans out to several tasks, and both return their own receiver, since a fan-out has no single "next" ref to hand back. Fan-*in* with data is the wiring object itself — `summarize({ north: extractNorth(), south: extractSouth() })` — so `[a, b] >> c` has an answer in each direction: named keys when values flow, `after` when only order does. This matches `Before`/`After` in the Go SDK's native Dag interface, spelled to TypeScript convention. -```ts -// Before (named wiring, current): the wiring object's keys match each -// handler's destructured object. -const transform = dag.task( - "transform", - async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, -); -const load = dag.task("load", async ({ transformed }: LoadArgs & TaskHandlerArgs) => { - /* ... */ -}); -load({ transformed: transform({ extracted: extract() }) }); -``` +### How -```ts -// After (positional wiring, hypothetical): each handler takes its data -// positionally, so `Parameters` is a real tuple (`[number]` -// for both transform and load), and a positional TaskFactory can type-check -// a call against it, the same way today's `Wiring` type-checks a -// named one. -const extract = dag.task("extract", async () => 42); -const transform = dag.task("transform", async (extracted: number) => extracted * 2); -const load = dag.task("load", async (transformed: number) => { - const client = getClient(); - await client.setXCom({ key: "loaded", value: transformed }); -}); +- `getClient()` and `getContext()` read from an `AsyncLocalStorage` (`node:async_hooks`) store that the runtime wraps around the handler call, at the single dispatch site in `ts-sdk/src/coordinator/runtime.ts`. The store propagates across every `await` and every promise created inside that scope by construction. What it does not cover is work that outlives the handler: a floating promise still calling `getClient()` after the handler resolved runs after the task's success has been reported, which is equally true of a closed-over client today. +- A handler's parameter type is exactly its own data. An earlier draft merged `ctx`/`client` into the same object, which forced every typed handler to declare `TArgs & TaskHandlerArgs` and left the top-level argument namespace open to collisions with an author's own parameter names. Getters close both. +- `withArgNames` and the name folding behind it ([ADR-0001](0001-mixed-lang-dag-interface.md)) exist for the mixed-language case and are never needed here: both ends of every name are TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. +- A `TaskRef` is inert. It is a handle for wiring, not a promise; nothing in a Dag file executes a task body. +- `registry.registerDag(...dags)` takes any number, since a Dag carries its own dag_id. Its mixed-language counterpart, `registry.registerTaskHandler(dagId, taskId, handler)` ([ADR-0001](0001-mixed-lang-dag-interface.md)), takes one at a time because each handler needs a dag_id/task_id pair supplied with it. -load(transform(extract())); -``` +## Alternatives -This is a second, independent decision stacked on top of the getters question above. A handler could still take one object for its data even without `ctx`/`client` in it, purely for self-documenting call sites once there are several arguments. Positional wiring only follows if handlers *also* drop the object, and it trades that self-documentation away at every multiple-argument call site. Neither this ADR nor the getters question settles it. It's recorded because positional wiring was previously ruled out entirely, for a reason that no longer holds once `ctx`/`client` are getters. +- **Positional wiring** (`load(transform(extract()))`), which becomes expressible once data no longer shares an object with `ctx`/`client`, since a handler can then take its arguments positionally and `Parameters` is a real tuple. Rejected: it removes the key names from every call site, and those names are what keeps flat, one-statement-per-task wiring readable at twenty tasks. A handler may still take several positional arguments; only the *wiring* stays named. +- **Injected `ctx`/`client` arguments**, mimicking the Python signature. Rejected, per the above and because feeling native to TypeScript matters more than matching Python's parameter list. ## Consequences -- One authoring surface (`dag.task()` + factory) covers both the graph and each task's arguments. -- `DagSpec`/`TaskSpec` are unaffected by this ADR. They are the language-neutral Dag configuration surface, not the argument-binding one. -- Whether `ctx`/`client` stay injected arguments or become getters is still open; adopting getters is a real ergonomics win but depends on the runtime-dispatch question above. -- If getters land, positional data parameters become possible for the first time, and with them, positional wiring (`load(transform(extract()))`). Adopting that is a further, separate, unresolved trade-off against today's named, self-documenting convention. +- One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments, and `before`/`after` cover edges that carry nothing. +- `DagSpec`/`TaskSpec` are unaffected. They are the language-neutral Dag configuration surface, not the argument-binding one. +- Handlers are unit-testable as plain functions of their data, with no SDK fixture to construct. +- `TaskHandlerArgs` is removed from the public API, which breaks 0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites that change. From 5b16ef49461d9cb324316c4879d0c7bbd5082601 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 06:13:22 +0000 Subject: [PATCH 3/6] Collapse the TypeScript SDK's registration surface onto the bundle A bundle usually provides both native Dags and bodies for tasks Python declares, and the earlier draft made it describe itself twice: one verb per kind, with a collector named after only one of them. Naming that collector Bundle, letting a single register call take either kind, and hanging serve off the same object leaves an author with one thing to hold and one place to look. StubHandler says what a mixed-language registration is - a body for a Python @task.stub task - instead of borrowing the word Dag for something TypeScript does not own. Both records also open with their decision now, with the AsyncLocalStorage dispatch scope, the name-folding proxy, and the rest of the mechanics moved to an appendix, so a reviewer can see what was decided without reading how it works first. --- ts-sdk/adr/0001-mixed-lang-dag-interface.md | 140 +++++++++++++++----- ts-sdk/adr/0002-native-dag-interface.md | 104 +++++++++++---- 2 files changed, 182 insertions(+), 62 deletions(-) diff --git a/ts-sdk/adr/0001-mixed-lang-dag-interface.md b/ts-sdk/adr/0001-mixed-lang-dag-interface.md index 18496784ab567..ba626292ddd62 100644 --- a/ts-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/ts-sdk/adr/0001-mixed-lang-dag-interface.md @@ -17,26 +17,45 @@ under the License. --> -# ADR-0001: Mixed-Lang Dag — TypeScript Task Handler Interface +# ADR-0001: Mixed-Lang Dag — TypeScript Stub Handler Interface ## Status Proposed. Revised after the review on #72047. -## Context +## Decision -The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. +1. TypeScript registers **stub handlers, not Dags**. `new StubHandler(dagId, taskId, handler)` binds a + handler to the Python-owned task it implements; `Dag` is exclusively the native case + ([ADR-0002](0002-native-dag-interface.md)). +2. **A bundle has one registration verb and serves itself.** `bundle.register(...)` takes Dags and + stub handlers alike, in any mixture, and `await bundle.serve()` starts the runtime over them. + `Bundle` replaces `DagRegistry`, and the free `serveDags(registry)` function goes with it. +3. **task_id is always written out**; nothing is derived from the handler's function name. +4. **A handler is a plain function of its own data**, destructured by name. `getContext()` and + `getClient()` supply the rest, so nothing the SDK injects shares a namespace with an author's + arguments. +5. **Names bind by folding on both sides** — lowercased, separators removed — so Python's + `region_code` reaches a handler's `regionCode` with nothing declared. `withArgNames` is for a + genuine rename, never for a spelling difference. +6. **An upstream's return value is not a bound argument.** Read it with + `client.getXCom({ key: "return_value", taskId })`. -This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. +## Context -A mixed-language Dag declares its structure in Python, and TypeScript supplies task bodies and nothing else. So TypeScript registers **task handlers**, not a Dag: it owns no dag_id, no schedule, no task order. `Dag` is exclusively the native case, covered in [ADR-0002](0002-native-dag-interface.md). +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume +those bindings directly as named arguments instead of re-fetching each value with +`client.getXCom(...)`. A mixed-language Dag declares its structure in Python, and TypeScript supplies +task bodies and nothing else: it owns no dag_id, no schedule, no task order. -## Decision +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself — its +shape, how it is materialized, how it travels over the wire — is a protocol-level decision recorded +in [`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). -A handler is a plain function of its own data. The SDK's context and client come from getters, and each handler is registered against the dag_id/task_id pair Python already owns. +## Example ```ts -import { DagRegistry, getClient, getContext, serveDags } from "apache-airflow-ts-sdk"; +import { Bundle, StubHandler, getClient, getContext } from "apache-airflow-ts-sdk"; interface TransformArgs { regionCode: string; @@ -52,14 +71,26 @@ async function transform({ regionCode, threshold }: TransformArgs) { return { regionCode, passed: rows >= threshold }; } -const registry = new DagRegistry(); -registry.registerTaskHandler("etl", "transform", transform); -await serveDags(registry); +const bundle = new Bundle(); +bundle.register(new StubHandler("etl", "transform", transform)); +await bundle.serve(); +``` + +A bundle usually provides both kinds, and one call lists everything it exposes: + +```ts +bundle.register( + nativeEtl, // a Dag, from ADR-0002 + new StubHandler("py_etl", "transform", transform), +); ``` ### Wire names -By default, arguments are transformed automatically: the SDK normalizes the names on **both** sides — lowercased, separators removed — and matches those. Python's `region_code` binds to a handler's `regionCode`, `Name` binds to `name`, and `s3_uri` binds to `s3Uri`, with nothing declared on either side. The Go SDK normalizes the same way (`strings.ToLower(strings.ReplaceAll(name, "_", ""))`), so one Python signature binds identically in either SDK. +Arguments bind by folding both sides, so `region_code` reaches `regionCode`, `Name` reaches `name`, +and `s3_uri` reaches `s3Uri` with nothing declared on either side. The Go SDK folds the same way +(`strings.ToLower(strings.ReplaceAll(name, "_", ""))`), so one Python signature binds identically in +either SDK. ```ts // Python: def transform(region_code: str, s3_uri: str, threshold: float) @@ -68,7 +99,9 @@ async function transform({ regionCode, s3Uri, threshold }: TransformArgs) { } ``` -A user can also specify the binding explicitly, with `withArgNames`: its first argument is the argument mapping, and its second is the handler itself. Normalization only absorbs spelling differences, so this is what to reach for when a handler wants a name the Python side never used — a clearer word, or a TypeScript reserved word like `enum`: +`withArgNames` states a binding explicitly — its first argument is the mapping, its second the +handler. Folding absorbs spelling differences, so this is for a name the Python side never used: a +clearer word, or a TypeScript reserved word like `enum`. ```ts const report = withArgNames( @@ -80,31 +113,70 @@ const report = withArgNames( }, ); -registry.registerTaskHandler("etl", "report", report); +bundle.register(new StubHandler("etl", "report", report)); ``` -The map's keys are checked against the handler's own parameter type, so `{ labl: "run_label" }` is a compile error naming the right key. Its values are Python names, which `tsc` cannot see and does not check. +The map's keys are checked against the handler's own parameter type, so `{ labl: "run_label" }` is a +compile error naming the right key. Its values are Python names, which `tsc` cannot see and does not +check. -### How +## Consequences -- `registry.registerTaskHandler(dagId, taskId, handler)` is the second registration verb, beside `registry.registerDag(...dags)` for the native case ([ADR-0002](0002-native-dag-interface.md)). It takes one handler at a time because each needs its own dag_id/task_id pair, where a Dag carries its own id and several can be registered in one call. task_id must match the `@task.stub` id exactly and is always written out: deriving it from the handler's function name would silently rebind the handler when the function is renamed. -- A handler registered this way has no factory to call, so wiring a mixed-language task the way a native one is wired (`transform()`) is a compile error rather than a runtime throw. That is the guarantee an earlier draft's separate `MixedLangDag` class existed to provide. -- `getClient()` and `getContext()` read from an `AsyncLocalStorage` store the runtime wraps around the handler call, and throw outside a handler. TypeScript keeps type and value namespaces separate, so `getContext()` coexists with the `TaskContext` type without either being renamed. -- The bound argument object is a `Proxy`. Destructuring a name triggers a lookup that folds that name on demand and matches it against the folded wire names, so binding needs nothing declared anywhere, and an entry in `withArgNames` takes precedence over folding. -- An unmatched name is logged with both the requested name and the names actually delivered. It cannot throw: a destructuring default (`{ runId = "manual" }`) is a legitimate miss, and the runtime cannot tell one from a typo. -- `in` folds like a read. `Object.keys` and rest destructuring (`{ ...rest }`) yield Python's names, since the SDK has no TypeScript-side names to enumerate. -- Two Python names that fold to the same token fail the task at dispatch, naming both. -- An upstream's return value is not delivered as a bound argument. Read it explicitly via `client.getXCom({ key: "return_value", taskId: "..." })`. +- One binding mechanism serves every mixed-language handler, and the Python call site stays the + single source of data-flow wiring. +- Folding matches the Go SDK's rule, so the same Python signature binds the same way in either SDK, + and neither needs a rename declared for ordinary snake_case parameters. +- A handler is directly unit-testable as a plain function of its data: no fixture to construct and no + intersection type to remember. +- This breaks 0.1.0-beta1 authors. `DagRegistry` becomes `Bundle`, and `TaskHandlerArgs` and the + `TaskHandler` type that takes it (`src/sdk/task.ts`) no longer describe a handler and are removed. + The shipped call sites change with the implementation — `src/index.ts`, and the `new Dag(...)` + + `dag.task(...)` pattern in `README.md`, `docs/index.md`, and `example/src/main.ts`. The package's + status line already reads "API may change". +- `serveDags(registry)` is removed in favour of `bundle.serve()`, since a bundle no longer holds + only Dags. Its name is also quoted in a user-facing error string + (`ts-sdk/src/cli/pack.ts:237`), which changes with it. ## Alternatives -- **Annotating the Python name on the field with a phantom type** (`type Arg = T & { readonly __arg?: N }`), the way Java's `@ArgName` annotates a `TaskInput` field. Rejected: an `interface` is erased, so the annotation cannot reach dispatch and needs a runtime companion regardless. It also has two silent failure modes — `Arg` collapses to a required `string`, because `undefined & object` is `never`, and the phantom key appears in `keyof` for an object-valued argument. -- **Reading the expected names from `handler.toString()`** and parsing the destructuring pattern. Property names do survive bundling, but a handler whose parameter is not destructured (`async (a: ReportArgs) => a.runLabel`) exposes no names at all, so the check would disappear silently for ordinary code. - -## Consequences - -- One binding mechanism serves every mixed-language handler, and the Python call site stays the single source of data-flow wiring. -- Folding matches the Go SDK's rule, so the same Python signature binds the same way in either SDK, and neither one needs a rename declared for ordinary snake_case parameters. -- `TaskHandlerArgs` is removed, `Dag` no longer serves the mixed-language case, and the registry's `register(...dags)` becomes `registerDag(...dags)` so the two verbs name what they take. All are shipped API (`src/index.ts`, and the `new Dag(...)` + `dag.task(...)` pattern in `README.md`, `docs/index.md`, and `example/src/main.ts`), so this breaks 0.1.0-beta1 authors and those call sites change with the implementation. The package's own status line already reads "API may change". -- A handler is directly unit-testable with a plain data argument: no `TaskHandlerArgs` fixture, and no intersection to remember. -- `withArgNames` is needed only for a genuine rename, never for a spelling difference, so most handlers never mention it. +- **Annotating the Python name on the field with a phantom type** + (`type Arg = T & { readonly __arg?: N }`), the way Java's `@ArgName` annotates + a `TaskInput` field. Rejected: an `interface` is erased, so the annotation cannot reach dispatch and + needs a runtime companion regardless. It also has two silent failure modes — + `Arg` collapses to a required `string`, because `undefined & object` is + `never`, and the phantom key appears in `keyof` for an object-valued argument. +- **Reading the expected names from `handler.toString()`** and parsing the destructuring pattern. + Property names do survive bundling, but a handler whose parameter is not destructured + (`async (a: ReportArgs) => a.runLabel`) exposes no names at all, so the check would disappear + silently for ordinary code. +- **Two registration verbs**, `registerDag(...dags)` beside `registerTaskHandler(dagId, taskId, fn)`. + Rejected once a stub handler became a value carrying its own ids: the asymmetry that justified the + split — a Dag knows its id, a bare handler does not — disappears, and a bundle that provides both + kinds had to say so in two calls. + +## Appendix: Implementation Notes + +- **`register` widens rather than splits.** The shipped `DagRegistry.register(...dags: Dag[])` + (`ts-sdk/src/sdk/registry.ts`) already narrows each argument with `instanceof Dag` and rejects a + foreign copy by brand. `Bundle.register(...items: Registerable[])`, over + `type Registerable = Dag | StubHandler`, follows the same path with one more arm — a discriminated + union being TypeScript's equivalent of the sealed interface the Go SDK uses for the same purpose. +- **`serve` is a method so the coordinator stays unnamed.** `startCoordinator` is deliberately not + exported — "Dag authors reach the runtime through `serveDags()`, and never name the coordinator + itself" (`ts-sdk/src/coordinator/index.ts`) — and a method on the object that already holds the + Dags and handlers keeps that intent while dropping the free function. +- **A stub handler has no factory to call**, so wiring one the way a native task is wired + (`transform()`) is a compile error rather than a runtime throw. That is the guarantee an earlier + draft's separate `MixedLangDag` class existed to provide. +- **`getClient()` and `getContext()` read from an `AsyncLocalStorage` store** the runtime wraps around + the handler call, and throw outside a handler. TypeScript keeps type and value namespaces separate, + so `getContext()` coexists with the `TaskContext` type without either being renamed. +- **The bound argument object is a `Proxy`.** Destructuring a name triggers a lookup that folds that + name on demand and matches it against the folded wire names, so binding needs nothing declared + anywhere, and an entry in `withArgNames` takes precedence over folding. +- **An unmatched name is logged**, with both the requested name and the names actually delivered. It + cannot throw: a destructuring default (`{ runId = "manual" }`) is a legitimate miss, and the runtime + cannot tell one from a typo. +- `in` folds like a read. `Object.keys` and rest destructuring (`{ ...rest }`) yield Python's names, + since the SDK has no TypeScript-side names to enumerate. Two Python names that fold to the same + token fail the task at dispatch, naming both. diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md index ca1c7dea418ea..9fa1cc4056c9c 100644 --- a/ts-sdk/adr/0002-native-dag-interface.md +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -23,16 +23,35 @@ Proposed. Revised after the review on #72047. -## Context +## Decision -A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case registers task handlers instead of a Dag, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). Both share the protocol substrate recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). +1. **`dag.task(taskId, handler)` returns a factory.** Calling the factory both places the task in the + Dag and supplies its arguments by name — the shape Python TaskFlow uses for + `load(transformed=transform(...))`. +2. **The call graph is the task graph.** `tsc` checks every wired key against the handler's own + parameter type, and a `TaskRef` exists only once its producing call has returned, so a cycle + through arguments is unrepresentable rather than rejected by a validator. +3. **Every task is called exactly once.** An uncalled task fails when the Dag is read, so none can be + silently left out of the graph. +4. **`before` and `after` draw order-only edges** — the TypeScript pair for `>>` and `<<`, both + variadic so one call fans out. +5. **A handler is a plain function of its own data**; `getContext()` and `getClient()` supply the rest. +6. **One registration verb**: `bundle.register(dag)`, the same call that takes stub handlers, with + `await bundle.serve()` starting the runtime ([ADR-0001](0001-mixed-lang-dag-interface.md)). -## Decision +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so +TypeScript itself must express both the graph and the task bodies. This ADR covers only what that +call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case +registers stub handlers instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the +protocol substrate recorded in +[`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). -`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow uses for `load(transformed=transform(...))`. A handler is a function of its own data; `getClient()` and `getContext()` supply the rest. +## Example ```ts -import { Dag, DagRegistry, getClient, serveDags } from "apache-airflow-ts-sdk"; +import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk"; const dag = new Dag("ts_etl"); @@ -48,18 +67,20 @@ const extracted = extract(); const transformed = transform({ extracted }); const loaded = load({ transformed }); -const registry = new DagRegistry(); -registry.registerDag(dag); -await serveDags(registry); +const bundle = new Bundle(); +bundle.register(dag); +await bundle.serve(); ``` -One statement per task, with each ref named, is the form to write. Nesting the calls (`load({ transformed: transform({ extracted: extract() }) })`) is legal and equivalent, but it is shorthand for a two-task chain, not the general shape: a Dag of twenty tasks reads as twenty flat statements, never as a twenty-deep expression. - -The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle through arguments is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task cannot be silently left out of the graph. +One statement per task, with each ref named, is the form to write. Nesting the calls +(`load({ transformed: transform({ extracted: extract() }) })`) is legal and equivalent, but it is +shorthand for a two-task chain, not the general shape: a Dag of twenty tasks reads as twenty flat +statements, never as a twenty-deep expression. ### Order-only edges: `>>` and `<<` -An edge that carries no data has no key to put in the wiring object, so it is drawn directly between refs. `before` and `after` are the TypeScript pair for Python's `>>` and `<<`: +An edge that carries no data has no key to put in the wiring object, so it is drawn directly between +refs: ```ts const cleaned = cleanup(); @@ -68,24 +89,51 @@ cleaned.after(loaded, transformed); // [loaded, transformed] >> cleaned loaded.before(cleaned); // loaded >> cleaned ``` -Both are variadic, so one call fans out to several tasks, and both return their own receiver, since a fan-out has no single "next" ref to hand back. Fan-*in* with data is the wiring object itself — `summarize({ north: extractNorth(), south: extractSouth() })` — so `[a, b] >> c` has an answer in each direction: named keys when values flow, `after` when only order does. This matches `Before`/`After` in the Go SDK's native Dag interface, spelled to TypeScript convention. +Both return their own receiver, since a fan-out has no single "next" ref to hand back. Fan-*in* with +data is the wiring object itself — `summarize({ north: extractNorth(), south: extractSouth() })` — so +`[a, b] >> c` has an answer in each direction: named keys when values flow, `after` when only order +does. This matches `Before`/`After` in the Go SDK's native Dag interface, spelled to TypeScript +convention. -### How +## Consequences -- `getClient()` and `getContext()` read from an `AsyncLocalStorage` (`node:async_hooks`) store that the runtime wraps around the handler call, at the single dispatch site in `ts-sdk/src/coordinator/runtime.ts`. The store propagates across every `await` and every promise created inside that scope by construction. What it does not cover is work that outlives the handler: a floating promise still calling `getClient()` after the handler resolved runs after the task's success has been reported, which is equally true of a closed-over client today. -- A handler's parameter type is exactly its own data. An earlier draft merged `ctx`/`client` into the same object, which forced every typed handler to declare `TArgs & TaskHandlerArgs` and left the top-level argument namespace open to collisions with an author's own parameter names. Getters close both. -- `withArgNames` and the name folding behind it ([ADR-0001](0001-mixed-lang-dag-interface.md)) exist for the mixed-language case and are never needed here: both ends of every name are TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. -- A `TaskRef` is inert. It is a handle for wiring, not a promise; nothing in a Dag file executes a task body. -- `registry.registerDag(...dags)` takes any number, since a Dag carries its own dag_id. Its mixed-language counterpart, `registry.registerTaskHandler(dagId, taskId, handler)` ([ADR-0001](0001-mixed-lang-dag-interface.md)), takes one at a time because each handler needs a dag_id/task_id pair supplied with it. +- One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments, + and `before`/`after` cover edges that carry nothing. +- Handlers are unit-testable as plain functions of their data, with no SDK fixture to construct. +- `DagSpec`/`TaskSpec` are unaffected. They are the language-neutral Dag configuration surface, not + the argument-binding one. +- `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes `Bundle`, and + `serveDags(registry)` becomes `bundle.serve()`, which breaks + 0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites + that change. ## Alternatives -- **Positional wiring** (`load(transform(extract()))`), which becomes expressible once data no longer shares an object with `ctx`/`client`, since a handler can then take its arguments positionally and `Parameters` is a real tuple. Rejected: it removes the key names from every call site, and those names are what keeps flat, one-statement-per-task wiring readable at twenty tasks. A handler may still take several positional arguments; only the *wiring* stays named. -- **Injected `ctx`/`client` arguments**, mimicking the Python signature. Rejected, per the above and because feeling native to TypeScript matters more than matching Python's parameter list. - -## Consequences - -- One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments, and `before`/`after` cover edges that carry nothing. -- `DagSpec`/`TaskSpec` are unaffected. They are the language-neutral Dag configuration surface, not the argument-binding one. -- Handlers are unit-testable as plain functions of their data, with no SDK fixture to construct. -- `TaskHandlerArgs` is removed from the public API, which breaks 0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites that change. +- **Positional wiring** (`load(transform(extract()))`), which becomes expressible once data no longer + shares an object with `ctx`/`client`, since a handler can then take its arguments positionally and + `Parameters` is a real tuple. Rejected: it removes the key names from every call + site, and those names are what keeps flat, one-statement-per-task wiring readable at twenty tasks. + A handler may still take several positional arguments; only the *wiring* stays named. +- **Injected `ctx`/`client` arguments**, mimicking the Python signature. Rejected, per the above and + because feeling native to TypeScript matters more than matching Python's parameter list. + +## Appendix: Implementation Notes + +- **`getClient()` and `getContext()` read from an `AsyncLocalStorage`** (`node:async_hooks`) store + that the runtime wraps around the handler call, at the single dispatch site in + `ts-sdk/src/coordinator/runtime.ts`. The store propagates across every `await` and every promise + created inside that scope by construction. What it does not cover is work that outlives the + handler: a floating promise still calling `getClient()` after the handler resolved runs after the + task's success has been reported, which is equally true of a closed-over client today. +- **A handler's parameter type is exactly its own data.** An earlier draft merged `ctx`/`client` into + the same object, which forced every typed handler to declare `TArgs & TaskHandlerArgs` and left the + top-level argument namespace open to collisions with an author's own parameter names. Getters close + both. +- **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a Dag file executes a task + body. +- **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md)) + exist for the mixed-language case and are never needed here: both ends of every name are + TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. +- **`bundle.register(...)` takes any number of Dags and stub handlers**, since each carries its own + ids; the earlier `registerDag`/`registerTaskHandler` split is recorded as a rejected alternative in + [ADR-0001](0001-mixed-lang-dag-interface.md). From 21fd2c4070ef7892367fb8f6348b5c6774499acf Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 06:23:17 +0000 Subject: [PATCH 4/6] Show what a native TypeScript Dag owns that a stub handler cannot The example declared a bare Dag and bare tasks, which reads as a more verbose way to write the mixed-language case. Carrying a schedule on the Dag and retries on a task shows the reason the native mode exists: the Dag file owns the configuration a Python stub Dag would otherwise own, and a handler for a Python-declared task can carry neither. The fields are recorded as forward-looking. DagSpec and TaskSpec accept only the empty object today, so the record says where an author writes this configuration without claiming to choose the fields themselves. --- ts-sdk/adr/0002-native-dag-interface.md | 40 +++++++++++++++++++------ 1 file changed, 31 insertions(+), 9 deletions(-) diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md index 9fa1cc4056c9c..217000199cc80 100644 --- a/ts-sdk/adr/0002-native-dag-interface.md +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -35,14 +35,19 @@ Proposed. Revised after the review on #72047. silently left out of the graph. 4. **`before` and `after` draw order-only edges** — the TypeScript pair for `>>` and `<<`, both variadic so one call fans out. -5. **A handler is a plain function of its own data**; `getContext()` and `getClient()` supply the rest. -6. **One registration verb**: `bundle.register(dag)`, the same call that takes stub handlers, with +5. **The Dag file owns Dag-level and task-level configuration.** `new Dag(dagId, spec)` carries the + schedule and the rest of `DagSpec`; `dag.task(taskId, handler, spec)` carries per-task options + such as retries. Python owns both in the mixed-language case + ([ADR-0001](0001-mixed-lang-dag-interface.md)), which is the difference between the two modes. +6. **A handler is a plain function of its own data**; `getContext()` and `getClient()` supply the rest. +7. **One registration verb**: `bundle.register(dag)`, the same call that takes stub handlers, with `await bundle.serve()` starting the runtime ([ADR-0001](0001-mixed-lang-dag-interface.md)). ## Context A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so -TypeScript itself must express both the graph and the task bodies. This ADR covers only what that +TypeScript itself must express everything Python would otherwise own: the schedule and the rest of +the Dag-level configuration, each task's own options, the graph, and the task bodies. This ADR covers only what that call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case registers stub handlers instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the protocol substrate recorded in @@ -53,15 +58,19 @@ protocol substrate recorded in ```ts import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk"; -const dag = new Dag("ts_etl"); +const dag = new Dag("ts_etl", { schedule: "@daily", catchup: false, tags: ["etl"] }); const extract = dag.task("extract", async (): Promise => 42); const transform = dag.task("transform", async ({ extracted }: { extracted: number }) => extracted * 2); -const load = dag.task("load", async ({ transformed }: { transformed: number }) => { - await getClient().setXCom({ key: "loaded", value: transformed }); -}); +const load = dag.task( + "load", + async ({ transformed }: { transformed: number }) => { + await getClient().setXCom({ key: "loaded", value: transformed }); + }, + { retries: 2 }, +); const extracted = extract(); const transformed = transform({ extracted }); @@ -72,6 +81,10 @@ bundle.register(dag); await bundle.serve(); ``` +The schedule on the `Dag` and the retries on `load` are the point of a native Dag: nothing outside +this file declares them. A mixed-language handler cannot carry either, because the Python Dag it +belongs to already does. + One statement per task, with each ref named, is the form to write. Nesting the calls (`load({ transformed: transform({ extracted: extract() }) })`) is legal and equivalent, but it is shorthand for a two-task chain, not the general shape: a Dag of twenty tasks reads as twenty flat @@ -100,8 +113,14 @@ convention. - One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments, and `before`/`after` cover edges that carry nothing. - Handlers are unit-testable as plain functions of their data, with no SDK fixture to construct. -- `DagSpec`/`TaskSpec` are unaffected. They are the language-neutral Dag configuration surface, not - the argument-binding one. +- `DagSpec` and `TaskSpec` are empty placeholders today — `Record` + (`ts-sdk/src/sdk/dag.ts`), so `new Dag("d", { schedule: "@daily" })` is currently a compile error + by design. Native declaration is what fills them, generated from the serialized-Dag JSON schema the + way `src/generated/supervisor.ts` is. This ADR does not choose those fields; it fixes where an + author writes them. +- `TaskOptions` collapses into `TaskSpec`. The shipped third argument to `dag.task` is + `{ inputs, spec }`; with wiring moved to the factory call, `inputs` is no longer an option and the + third argument is the spec itself. - `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes `Bundle`, and `serveDags(registry)` becomes `bundle.serve()`, which breaks 0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites @@ -129,6 +148,9 @@ convention. the same object, which forced every typed handler to declare `TArgs & TaskHandlerArgs` and left the top-level argument namespace open to collisions with an author's own parameter names. Getters close both. +- **The spec argument already has its slot.** `dag.task(taskId, handler, options)` reads + `{ inputs = {}, spec = {} }` and runs `validateEmptySpec` on the spec today + (`ts-sdk/src/sdk/dag.ts`), so task fields land on a path that exists rather than a new one. - **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a Dag file executes a task body. - **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md)) From 2bca33cc332acb87b15d6ded49d1f0135ed7c955 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 08:38:51 +0000 Subject: [PATCH 5/6] Rename StubHandler to StubTask in the TypeScript SDK ADRs The name should say what the SDK registers, and what it registers is a task: Python declares it with @task.stub, gives it a task_id, and the SDK supplies only its body. "Handler" described the callable rather than the thing it stands for, leaving these docs using a word no other part of Airflow uses for a task. --- ts-sdk/adr/0001-mixed-lang-dag-interface.md | 20 ++++++++++---------- ts-sdk/adr/0002-native-dag-interface.md | 6 +++--- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/ts-sdk/adr/0001-mixed-lang-dag-interface.md b/ts-sdk/adr/0001-mixed-lang-dag-interface.md index ba626292ddd62..13c896eae0f97 100644 --- a/ts-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/ts-sdk/adr/0001-mixed-lang-dag-interface.md @@ -17,7 +17,7 @@ under the License. --> -# ADR-0001: Mixed-Lang Dag — TypeScript Stub Handler Interface +# ADR-0001: Mixed-Lang Dag — TypeScript Stub Task Interface ## Status @@ -25,11 +25,11 @@ Proposed. Revised after the review on #72047. ## Decision -1. TypeScript registers **stub handlers, not Dags**. `new StubHandler(dagId, taskId, handler)` binds a +1. TypeScript registers **stub tasks, not Dags**. `new StubTask(dagId, taskId, handler)` binds a handler to the Python-owned task it implements; `Dag` is exclusively the native case ([ADR-0002](0002-native-dag-interface.md)). 2. **A bundle has one registration verb and serves itself.** `bundle.register(...)` takes Dags and - stub handlers alike, in any mixture, and `await bundle.serve()` starts the runtime over them. + stub tasks alike, in any mixture, and `await bundle.serve()` starts the runtime over them. `Bundle` replaces `DagRegistry`, and the free `serveDags(registry)` function goes with it. 3. **task_id is always written out**; nothing is derived from the handler's function name. 4. **A handler is a plain function of its own data**, destructured by name. `getContext()` and @@ -55,7 +55,7 @@ in [`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskf ## Example ```ts -import { Bundle, StubHandler, getClient, getContext } from "apache-airflow-ts-sdk"; +import { Bundle, StubTask, getClient, getContext } from "apache-airflow-ts-sdk"; interface TransformArgs { regionCode: string; @@ -72,7 +72,7 @@ async function transform({ regionCode, threshold }: TransformArgs) { } const bundle = new Bundle(); -bundle.register(new StubHandler("etl", "transform", transform)); +bundle.register(new StubTask("etl", "transform", transform)); await bundle.serve(); ``` @@ -81,7 +81,7 @@ A bundle usually provides both kinds, and one call lists everything it exposes: ```ts bundle.register( nativeEtl, // a Dag, from ADR-0002 - new StubHandler("py_etl", "transform", transform), + new StubTask("py_etl", "transform", transform), ); ``` @@ -113,7 +113,7 @@ const report = withArgNames( }, ); -bundle.register(new StubHandler("etl", "report", report)); +bundle.register(new StubTask("etl", "report", report)); ``` The map's keys are checked against the handler's own parameter type, so `{ labl: "run_label" }` is a @@ -150,7 +150,7 @@ check. (`async (a: ReportArgs) => a.runLabel`) exposes no names at all, so the check would disappear silently for ordinary code. - **Two registration verbs**, `registerDag(...dags)` beside `registerTaskHandler(dagId, taskId, fn)`. - Rejected once a stub handler became a value carrying its own ids: the asymmetry that justified the + Rejected once a stub task became a value carrying its own ids: the asymmetry that justified the split — a Dag knows its id, a bare handler does not — disappears, and a bundle that provides both kinds had to say so in two calls. @@ -159,13 +159,13 @@ check. - **`register` widens rather than splits.** The shipped `DagRegistry.register(...dags: Dag[])` (`ts-sdk/src/sdk/registry.ts`) already narrows each argument with `instanceof Dag` and rejects a foreign copy by brand. `Bundle.register(...items: Registerable[])`, over - `type Registerable = Dag | StubHandler`, follows the same path with one more arm — a discriminated + `type Registerable = Dag | StubTask`, follows the same path with one more arm — a discriminated union being TypeScript's equivalent of the sealed interface the Go SDK uses for the same purpose. - **`serve` is a method so the coordinator stays unnamed.** `startCoordinator` is deliberately not exported — "Dag authors reach the runtime through `serveDags()`, and never name the coordinator itself" (`ts-sdk/src/coordinator/index.ts`) — and a method on the object that already holds the Dags and handlers keeps that intent while dropping the free function. -- **A stub handler has no factory to call**, so wiring one the way a native task is wired +- **A stub task has no factory to call**, so wiring one the way a native task is wired (`transform()`) is a compile error rather than a runtime throw. That is the guarantee an earlier draft's separate `MixedLangDag` class existed to provide. - **`getClient()` and `getContext()` read from an `AsyncLocalStorage` store** the runtime wraps around diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md index 217000199cc80..221485dad9f57 100644 --- a/ts-sdk/adr/0002-native-dag-interface.md +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -40,7 +40,7 @@ Proposed. Revised after the review on #72047. such as retries. Python owns both in the mixed-language case ([ADR-0001](0001-mixed-lang-dag-interface.md)), which is the difference between the two modes. 6. **A handler is a plain function of its own data**; `getContext()` and `getClient()` supply the rest. -7. **One registration verb**: `bundle.register(dag)`, the same call that takes stub handlers, with +7. **One registration verb**: `bundle.register(dag)`, the same call that takes stub tasks, with `await bundle.serve()` starting the runtime ([ADR-0001](0001-mixed-lang-dag-interface.md)). ## Context @@ -49,7 +49,7 @@ A Dag authored with no Python stub file has no `@task.stub` call site to declare TypeScript itself must express everything Python would otherwise own: the schedule and the rest of the Dag-level configuration, each task's own options, the graph, and the task bodies. This ADR covers only what that call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case -registers stub handlers instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the +registers stub tasks instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the protocol substrate recorded in [`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). @@ -156,6 +156,6 @@ convention. - **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md)) exist for the mixed-language case and are never needed here: both ends of every name are TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. -- **`bundle.register(...)` takes any number of Dags and stub handlers**, since each carries its own +- **`bundle.register(...)` takes any number of Dags and stub tasks**, since each carries its own ids; the earlier `registerDag`/`registerTaskHandler` split is recorded as a rejected alternative in [ADR-0001](0001-mixed-lang-dag-interface.md). From ff0b411409632cbbe2aa6d13c3009df36719d61f Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 10:53:32 +0000 Subject: [PATCH 6/6] Rename StubTask to TaskHandler in the TypeScript SDK ADRs Python declares the task; an SDK supplies only its body. Naming the SDK object after the task overloads a word the native surface already uses for something the user creates with dag.Task, so it names the role the SDK actually plays instead. --- ts-sdk/adr/0001-mixed-lang-dag-interface.md | 20 ++++++++++---------- ts-sdk/adr/0002-native-dag-interface.md | 6 +++--- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/ts-sdk/adr/0001-mixed-lang-dag-interface.md b/ts-sdk/adr/0001-mixed-lang-dag-interface.md index 13c896eae0f97..f12226c9f2781 100644 --- a/ts-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/ts-sdk/adr/0001-mixed-lang-dag-interface.md @@ -17,7 +17,7 @@ under the License. --> -# ADR-0001: Mixed-Lang Dag — TypeScript Stub Task Interface +# ADR-0001: Mixed-Lang Dag — TypeScript Task Handler Interface ## Status @@ -25,11 +25,11 @@ Proposed. Revised after the review on #72047. ## Decision -1. TypeScript registers **stub tasks, not Dags**. `new StubTask(dagId, taskId, handler)` binds a +1. TypeScript registers **task handlers, not Dags**. `new TaskHandler(dagId, taskId, handler)` binds a handler to the Python-owned task it implements; `Dag` is exclusively the native case ([ADR-0002](0002-native-dag-interface.md)). 2. **A bundle has one registration verb and serves itself.** `bundle.register(...)` takes Dags and - stub tasks alike, in any mixture, and `await bundle.serve()` starts the runtime over them. + task handlers alike, in any mixture, and `await bundle.serve()` starts the runtime over them. `Bundle` replaces `DagRegistry`, and the free `serveDags(registry)` function goes with it. 3. **task_id is always written out**; nothing is derived from the handler's function name. 4. **A handler is a plain function of its own data**, destructured by name. `getContext()` and @@ -55,7 +55,7 @@ in [`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskf ## Example ```ts -import { Bundle, StubTask, getClient, getContext } from "apache-airflow-ts-sdk"; +import { Bundle, TaskHandler, getClient, getContext } from "apache-airflow-ts-sdk"; interface TransformArgs { regionCode: string; @@ -72,7 +72,7 @@ async function transform({ regionCode, threshold }: TransformArgs) { } const bundle = new Bundle(); -bundle.register(new StubTask("etl", "transform", transform)); +bundle.register(new TaskHandler("etl", "transform", transform)); await bundle.serve(); ``` @@ -81,7 +81,7 @@ A bundle usually provides both kinds, and one call lists everything it exposes: ```ts bundle.register( nativeEtl, // a Dag, from ADR-0002 - new StubTask("py_etl", "transform", transform), + new TaskHandler("py_etl", "transform", transform), ); ``` @@ -113,7 +113,7 @@ const report = withArgNames( }, ); -bundle.register(new StubTask("etl", "report", report)); +bundle.register(new TaskHandler("etl", "report", report)); ``` The map's keys are checked against the handler's own parameter type, so `{ labl: "run_label" }` is a @@ -150,7 +150,7 @@ check. (`async (a: ReportArgs) => a.runLabel`) exposes no names at all, so the check would disappear silently for ordinary code. - **Two registration verbs**, `registerDag(...dags)` beside `registerTaskHandler(dagId, taskId, fn)`. - Rejected once a stub task became a value carrying its own ids: the asymmetry that justified the + Rejected once a task handler became a value carrying its own ids: the asymmetry that justified the split — a Dag knows its id, a bare handler does not — disappears, and a bundle that provides both kinds had to say so in two calls. @@ -159,13 +159,13 @@ check. - **`register` widens rather than splits.** The shipped `DagRegistry.register(...dags: Dag[])` (`ts-sdk/src/sdk/registry.ts`) already narrows each argument with `instanceof Dag` and rejects a foreign copy by brand. `Bundle.register(...items: Registerable[])`, over - `type Registerable = Dag | StubTask`, follows the same path with one more arm — a discriminated + `type Registerable = Dag | TaskHandler`, follows the same path with one more arm — a discriminated union being TypeScript's equivalent of the sealed interface the Go SDK uses for the same purpose. - **`serve` is a method so the coordinator stays unnamed.** `startCoordinator` is deliberately not exported — "Dag authors reach the runtime through `serveDags()`, and never name the coordinator itself" (`ts-sdk/src/coordinator/index.ts`) — and a method on the object that already holds the Dags and handlers keeps that intent while dropping the free function. -- **A stub task has no factory to call**, so wiring one the way a native task is wired +- **A task handler has no factory to call**, so wiring one the way a native task is wired (`transform()`) is a compile error rather than a runtime throw. That is the guarantee an earlier draft's separate `MixedLangDag` class existed to provide. - **`getClient()` and `getContext()` read from an `AsyncLocalStorage` store** the runtime wraps around diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md index 221485dad9f57..03c9a9e119745 100644 --- a/ts-sdk/adr/0002-native-dag-interface.md +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -40,7 +40,7 @@ Proposed. Revised after the review on #72047. such as retries. Python owns both in the mixed-language case ([ADR-0001](0001-mixed-lang-dag-interface.md)), which is the difference between the two modes. 6. **A handler is a plain function of its own data**; `getContext()` and `getClient()` supply the rest. -7. **One registration verb**: `bundle.register(dag)`, the same call that takes stub tasks, with +7. **One registration verb**: `bundle.register(dag)`, the same call that takes task handlers, with `await bundle.serve()` starting the runtime ([ADR-0001](0001-mixed-lang-dag-interface.md)). ## Context @@ -49,7 +49,7 @@ A Dag authored with no Python stub file has no `@task.stub` call site to declare TypeScript itself must express everything Python would otherwise own: the schedule and the rest of the Dag-level configuration, each task's own options, the graph, and the task bodies. This ADR covers only what that call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case -registers stub tasks instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the +registers task handlers instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the protocol substrate recorded in [`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). @@ -156,6 +156,6 @@ convention. - **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md)) exist for the mixed-language case and are never needed here: both ends of every name are TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. -- **`bundle.register(...)` takes any number of Dags and stub tasks**, since each carries its own +- **`bundle.register(...)` takes any number of Dags and task handlers**, since each carries its own ids; the earlier `registerDag`/`registerTaskHandler` split is recorded as a rejected alternative in [ADR-0001](0001-mixed-lang-dag-interface.md).