TS SDK: Add conditional branching with dag.if

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

@@ -359,6 +359,34 @@ there. Set it once on the Dag and each task inherits it:
 ``queue`` on a task wins over the Dag's. See :ref:`typescript-sdk/coordinator-config` for the
 ``queue_to_coordinator`` entry that sends that queue to the coordinator.
 
+Conditional branching
+~~~~~~~~~~~~~~~~~~~~~
+
+``dag.if`` takes a handler that returns a boolean, and names the task each outcome runs:
+
+.. code-block:: typescript
+
+    async function hasRows({ rows }: { rows: number }): Promise<boolean> {
+      return rows > 0;
+    }
+
+    const gate = dag.if(hasRows, { rows: extracted });
+    gate.then(loaded).else(reportedEmpty);
+
+``dag.if`` declares the condition as a task: its id is the function's name, and a trailing spec sets
+its id or other options, as in ``dag.if(hasRows, { rows }, { taskId: "has_rows" })``. The compiler
+checks that the handler returns a boolean. ``else`` is optional: a one-sided condition skips its own
+branch when the condition fails and follows nothing. The condition is also a node, so
+``notified.after(gate)`` orders a task after it.
+
+A guarded task takes no argument for the control edge, because a condition's boolean decides whether
+the task runs rather than what it runs on. Read a value from the condition with
+``getClient().getXCom``.
+
+The side not taken is skipped when the run reaches it, and stays skipped if you clear it later. Only
+the branches named here are skipped, so a task that several branches converge on still runs — unlike
+Python's ``@task.branch``, which skips every immediate downstream it did not follow.
+
 ``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options:
 ``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2, retryDelay: 30 }`` for a task.
 

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

@@ -135,6 +135,49 @@ data is the wiring object itself — `summarize({ north: extractNorth(), south:
 does. This matches `Before`/`After` in the Go SDK's native Dag interface, spelled to TypeScript
 convention.
 
+### Conditional branching: `if` and `else`
+
+`dag.if(handler)` is TypeScript's spelling of the construct
+[`airflow-core/adr/lang-sdk/0008`](../../airflow-core/adr/lang-sdk/0008-control-flow-constructs.md)
+names after the host language's control flow:
+
+```ts
+async function hasRows({ rows }: { rows: number }): Promise<boolean> {
+  return rows > 0;
+}
+
+const gate = dag.if(hasRows, { rows: validated });
+gate.then(loadIfReady).else(loadFallback);
+```
+
+**The condition is a handler, declared and wired in one call.** That ADR writes
+`dag.If(hasRows, airflow.Inputs(validated), airflow.TaskSpec{...})` in Go, and
+`dag.if(hasRows, { rows: validated }, { ... })` reads the same way: inputs, then an optional
+`TaskSpec`. The deciding task takes its id from the function's name unless the spec's `taskId` sets
+it, and the spec carries its other options, such as `queue` and `retries`. The compiler checks that
+the handler returns a `boolean`, and that the inputs match its argument.
+
+**A condition is a node.** What `dag.if` returns carries `before` and `after`, and stands at the
+other end of an edge as the deciding task, so `notified.after(gate)` reads as Go's `.After(gate)`. It
+is also a valid branch of another condition.
+
+**A `then` chain is a thenable, and is guarded rather than avoided.** An object with a callable
+`then` is a *thenable*: were the condition `dag.if` returns to reach an `await`, the runtime would hand
+its `then` a resolve function where a task reference belongs. Two things contain that. `.then(...)`
+returns an object carrying only `.else`, so nothing past the first step is awaitable at all; and
+`.then` rejects a function argument by naming the cause, so an author who does await it reads "this
+builds a branch, drop the await" rather than a type error about references.
+
+**A branch is a real branch to Airflow.** The control edges serialize as ordinary order-only edges
+and carry no branch-candidate field, as that ADR's consequences require. The condition task is
+serialized with `_can_skip_downstream`, and it writes the `skipmixin_key` XCom alongside the skip, so
+clearing a skipped branch re-skips it the way a Python `@task.branch` does rather than running the
+side the condition rejected.
+
+A one-sided `if` is a branch with one candidate — it skips `then` and follows nothing — rather than a
+`ShortCircuitOperator`, which would also skip the whole downstream closure and ignore trigger rules.
+A guarded task takes no argument for the control edge: a condition's boolean is a signal, not data.
+
 ## Consequences
 
 - One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments,

ts-sdk/src/coordinator/client.ts

@@ -30,6 +30,7 @@ import type {
   GetXCom,
   SetXCom,
   GetConnection,
+  SkipDownstreamTasks,
   ConnectionResult as WireConnectionResult,
 } from "./protocol.js";
 
@@ -82,6 +83,9 @@ export interface CoordinatorClient extends TaskClient {
    * output fails the task, while one that pushed null binds null.
    */
   getXComEntry(opts: GetXComOpts): Promise<XComEntry>;
+
+  /** Mark direct downstream tasks of the running task as skipped; none is a no-op. */
+  skipDownstreamTasks(taskIds: readonly string[]): Promise<void>;
 }
 
 export function createCoordinatorClient(
@@ -202,6 +206,14 @@ export function createCoordinatorClient(
       await rpc("SetXCom", null, msg, () => undefined, "throw");
     },
 
+    // ---- Control flow ----
+
+    async skipDownstreamTasks(taskIds: readonly string[]): Promise<void> {
+      if (taskIds.length === 0) return;
+      const msg: SkipDownstreamTasks = { type: "SkipDownstreamTasks", tasks: [...taskIds] };
+      await rpc("SkipDownstreamTasks", null, msg, () => undefined, "throw");
+    },
+
     // ---- Connections ----
 
     async getConnection(connId: string): Promise<ConnectionResult | null> {

ts-sdk/src/coordinator/protocol.ts

@@ -64,6 +64,7 @@ export type {
   GetXCom,
   SetXCom,
   GetConnection,
+  SkipDownstreamTasks,
 } from "../generated/supervisor.js";
 
 // -------- Frames from supervisor --------

ts-sdk/src/coordinator/serde.ts

@@ -143,6 +143,7 @@ export function serializeDag(
         withDagQueue(record.spec, dag.spec.queue),
         graph.downstreamTaskIds.get(taskId),
         inputs.get(taskId),
+        record.canSkipDownstream === true,
       ),
     ),
     dag_dependencies: [],
@@ -180,6 +181,7 @@ function serializeTask(
   spec: object,
   downstream: ReadonlySet<string> | undefined,
   inputs: RecordedInputs | undefined,
+  canSkipDownstream: boolean,
 ): SerializedValue {
   const data: Record<string, SerializedValue> = {
     task_id: taskId,
@@ -198,6 +200,8 @@ function serializeTask(
   const label = `task "${taskId}" of Dag "${dagId}"`;
   const bindings = serializeArgBindings(inputs, label);
   if (bindings) data["_arg_bindings"] = bindings;
+  // Lets `NotPreviouslySkippedDep` re-skip a cleared downstream, as Python's SkipMixin does.
+  if (canSkipDownstream) data["_can_skip_downstream"] = true;
   applySchemaFields(data, spec, TASK_FIELD_RULES, label);
   if (downstream?.size) {
     data["downstream_task_ids"] = [...downstream].sort();

ts-sdk/src/index.ts

@@ -27,6 +27,9 @@ 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 {
+  Condition,
+  ConditionElse,
+  DeciderArgs,
   DagSpec,
   Node,
   TaskFactory,