TS SDK: declare task dependencies by calling tasks

airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst

@@ -121,16 +121,12 @@ That top-level ``await`` makes the module a runnable bundle entry point.
 The ``dagId`` a handler binds must match the ``dag_id`` of the Python Dag, and the ``taskId`` a
 ``@task.stub`` function in that Dag, including any TaskGroup prefix.
 
-``register`` takes any number of task handlers and ``bundle.serve()`` serves exactly what is registered,
-so a task left out is not part of the packed bundle and is marked removed at runtime.
+``register`` takes any number of task handlers and Dags, and ``bundle.serve()`` serves exactly what is
+registered, so a task left out is not part of the packed bundle and is marked removed at runtime.
 A second ``bundle.serve()`` call is rejected.
 Registering holds no sockets and starts nothing, so a unit test can build a bundle and dispatch a handler
 through ``bundle.getTaskHandler(dagId, taskId)`` without a coordinator runtime.
 
-``Dag`` is another interface, for a Dag declared in TypeScript rather than in Python, and is still a work
-in progress. ``new Dag`` and ``dag.task`` take a trailing options object (``spec`` on both, plus
-``inputs`` on a task) that is not used yet; do not set them.
-
 TaskFlow arguments
 ~~~~~~~~~~~~~~~~~~
 
@@ -242,6 +238,47 @@ task instance.
   inside the scheduler process, so they must be present where the scheduler can read them. The API server
   and Dag processor do not need them.
 
+.. _typescript-sdk/native-dag:
+
+Declaring a Dag in TypeScript
+-----------------------------
+
+A ``Dag`` is declared on this side rather than in Python: its tasks and the edges between them are
+written in TypeScript. The surface is still growing, so a Dag declared this way is not served to
+Airflow yet.
+
+``dag.task(taskId, handler)`` returns a *factory*. Calling it places the task in the Dag and supplies the
+handler's arguments, so the call graph is the task graph:
+
+.. code-block:: typescript
+
+    import { Dag } from "apache-airflow-ts-sdk";
+
+    const dag = new Dag("ts_etl");
+
+    const extract = dag.task("extract", async (): Promise<number> => 42);
+    const transform = dag.task("transform", async (rows: number, region: string) => rows * 2);
+    const load = dag.task("load", async (total: number) => {});
+
+    load(transform(extract(), "us"));
+
+Arguments are passed in the order the handler declares them. A handler that declares a single object of
+named arguments can also be called with that object, which names each input instead of ordering it:
+
+.. code-block:: typescript
+
+    const store = dag.task("store", async ({ total }: { total: number }) => {});
+
+    store({ total: extract() });
+
+Each argument takes either an upstream reference or a literal JSON value. A reference has to be the
+argument itself: one buried inside an array or an object is a literal, and draws no edge.
+
+Every task has to be called exactly once. An uncalled task fails when the Dag is read, so none can be
+left out of the graph by accident.
+
+``new Dag`` and ``dag.task`` also take a trailing ``spec`` object that is not used yet; do not set it.
+
 Writing tasks
 -------------
 

ts-sdk/adr/0002-native-dag-interface.md

@@ -28,11 +28,15 @@ Proposed. Revised after the review on #72047.
 1. **`dag.task(handler)` returns a factory, and the task id is optional.** With no id the task takes
    the handler's function name (`dag.task(extract)` → task `"extract"`); `dag.task(taskId, handler)`
    sets it explicitly, which an anonymous handler must do. Calling the factory both places the task in
-   the Dag and supplies its arguments by name — the shape Python TaskFlow uses for
+   the Dag and supplies its arguments, in the order the handler declares them
+   (`load(transform(extract(), "us"))`). A handler that declares a single object of named arguments can
+   also be called with that object — 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.
+   through arguments is unrepresentable rather than rejected by a validator. A reference passed by
+   position is checked against the argument's own type, which is what tells the two call shapes apart
+   when a handler declares a single argument.
 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
@@ -145,21 +149,20 @@ convention.
   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.
+- `TaskOptions` carries the spec and the handler's positional argument names, which the packer fills in
+  from the parameter list so the Dag names each argument as its handler does. With wiring moved to the
+  factory call, `inputs` is no longer an option.
 - `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<typeof handler>` 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.
+- **Named-only wiring**, rejected in the review on #73435: naming every input reads well at twenty
+  tasks but forces an object around a single argument, and positional calls are what TypeScript
+  authors write. Both are offered, and the handler's own parameter list decides which one a task can
+  use.
 - **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.
 
@@ -175,9 +178,12 @@ 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.
+- **The spec argument already has its slot.** `dag.task(taskId, handler, options)` reads `{ spec = {} }`
+  and runs `validateEmptySpec` on it (`ts-sdk/src/sdk/dag.ts`), so task fields land on a path that
+  exists rather than a new one.
+- **A positional argument binds by order, and its name is a label.** The serialized Dag names each
+  argument, so the packer reads the names from the handler's parameter list; `arg0`, `arg1` and so on
+  stand in for a name it cannot see, without changing which value reaches which argument.
 - **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a Dag file executes a task
   body.
 - **A defaulted task id is resolved at pack time, not read at runtime.** esbuild renames function

ts-sdk/src/coordinator/manifest.ts

@@ -18,7 +18,7 @@
  */
 
 import { SUPERVISOR_API_VERSION } from "./protocol.js";
-import { listBundleDags, type Bundle } from "../sdk/bundle.js";
+import { bundleDagTaskIds, finalizeBundleDags, type Bundle } from "../sdk/bundle.js";
 
 export const AIRFLOW_METADATA_FLAG = "--airflow-metadata";
 
@@ -36,7 +36,10 @@ export interface BundleManifest {
 
 export function buildBundleManifest(bundle: Bundle): BundleManifest {
   const taskHandlers: BundleManifest["task_handlers"] = {};
-  for (const { dagId, tasks } of listBundleDags(bundle)) {
+  // The manifest is the bundle reporting what it provides, so this is where its
+  // Dags are finalized: a Dag missing an edge is reported here rather than packed.
+  finalizeBundleDags(bundle);
+  for (const [dagId, tasks] of bundleDagTaskIds(bundle)) {
     if (typeof dagId !== "string") {
       throw new Error("Dag ID must be a string");
     }

ts-sdk/src/coordinator/runtime.ts

@@ -55,7 +55,7 @@ import {
   type StartupDetails,
 } from "./protocol.js";
 import { getArgNames } from "../sdk/arg-names.js";
-import { listBundleTasks, type Bundle } from "../sdk/bundle.js";
+import { bundleDagTaskIds, type Bundle } from "../sdk/bundle.js";
 import { runInTaskScope, type TaskContext } from "../sdk/task.js";
 import type { JsonValue } from "../sdk/client-types.js";
 
@@ -148,10 +148,10 @@ export async function startCoordinator(
     const runtimeLogs = logs.child("runtime");
     runtimeLogs.debug("Connecting log socket", { logs_addr: parsed.logsAddr });
     await logs.connect(parsed.logsAddr);
-    const tasks = listBundleTasks(bundle);
+    const byDag = bundleDagTaskIds(bundle);
     runtimeLogs.info("Coordinator runtime started", {
-      registered_tasks: tasks,
-      count: tasks.length,
+      registered_tasks: Object.fromEntries(byDag),
+      count: [...byDag.values()].reduce((total, tasks) => total + tasks.length, 0),
       // Cadwyn schema version this SDK was generated against. Logged
       // for operator visibility; not sent on the wire.
       supervisor_api_version: SUPERVISOR_API_VERSION,
@@ -270,7 +270,7 @@ function handleParse(
   // TypeScript-native Dag parsing is not yet supported.
   // Respond with an empty result so the Python-stub-Dag workflow works.
   logs.info("Parse-mode response (TS Dag parsing not yet supported)", {
-    registered_tasks: listBundleTasks(bundle),
+    registered_tasks: Object.fromEntries(bundleDagTaskIds(bundle)),
   });
   const response: RuntimeDagFileParsingResult = {
     type: "DagFileParsingResult",
@@ -295,7 +295,7 @@ async function handleTask(
     logs.warning("No handler registered for task", {
       dag_id: ti.dag_id,
       task_id: ti.task_id,
-      available: listBundleTasks(bundle),
+      available: Object.fromEntries(bundleDagTaskIds(bundle)),
     });
     // A missing handler means this bundle cannot run the task, so retrying the
     // same bundle/configuration mismatch would not help.

ts-sdk/src/index.ts

@@ -26,7 +26,16 @@ export { ConnectionNotFoundError, VariableNotFoundError } from "./sdk/client.js"
 export { SUPERVISOR_API_VERSION } from "./coordinator/index.js";
 export type { ArgNameMap } from "./sdk/arg-names.js";
 export type { Registerable } from "./sdk/bundle.js";
-export type { DagSpec, TaskInputs, TaskOptions, TaskRef, TaskSpec } from "./sdk/dag.js";
+export type {
+  DagSpec,
+  PositionalInputs,
+  TaskFactory,
+  TaskInput,
+  TaskInputs,
+  TaskOptions,
+  TaskRef,
+  TaskSpec,
+} from "./sdk/dag.js";
 export type { TaskClient } from "./sdk/client.js";
 export type { ConnectionResult, GetXComOpts, JsonValue, SetXComOpts } from "./sdk/client-types.js";
 export type { TaskContext, TaskFunction } from "./sdk/task.js";