TS SDK: Add triggerDagRun to native Dags

airflow-core/adr/lang-sdk/0008-control-flow-constructs.md

@@ -30,7 +30,8 @@ Proposed.
 2. **A branch selects a task, not a string.** A case is the task reference the SDK already handed back, and the value on the wire is that task's task_id, so the compiler checks the candidate exists and no label has to be kept in step with it.
 3. **No default case.** `BranchPythonOperator` has none to serialize, and a one-sided `If` whose condition is false follows nothing at all.
    A decider returning a ref that is not one of the declared cases is a run-time error the SDK raises, a narrower check than `skip_all_except`, which only rejects a task_id missing from the whole Dag.
-4. **Triggering a Dag run is an ordinary DSL task**, it is pure DSL purpose instead of a new runtime.
+4. **Triggering a Dag run is an ordinary task that the SDK's own runtime runs.** It has no handler.
+   The runtime sends the `TriggerDagRun` request itself and follows `TriggerDagRunOperator` for every option it offers, including `wait_for_completion` and `deferrable`.
 5. **Grouping keeps Python's semantics**: a scope offering the same task and nesting methods as the Dag, prefixing each task_id with the group id (`prefix_group_id`),
    and can be ordered against a task or another group, as `group1 >> group2` does in Python.
 
@@ -89,7 +90,12 @@ That operator defaults to skipping every task in its downstream closure and igno
   a single-ref return cannot express that, and no Lang SDK offers it for now.
   An author needing several paths together puts them behind one task, or gates each with its own condition.
   This limitation is accepted rather than open.
-- **No Lang SDK needs a deferral mechanism for now** to offer `deferrable` or `wait_for_completion`, because the trigger task runs in Python.
+- **Each Lang SDK runtime implements the trigger task**, so a native Dag needs no Python worker.
+  `deferrable` needs no trigger written in that language: the runtime defers to the standard provider's `DagStateTrigger`, which runs in the Python triggerer, and the task keeps its queue, so it resumes in the same runtime with `next_method` set.
+- **The runtime reads the config the trigger task needs from its environment**, since it cannot read `airflow.cfg`.
+  It reads `AIRFLOW__API__BASE_URL`, `AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE` and `AIRFLOW__TRIGGERER__QUEUES_ENABLED`, and an unset one takes Python's fallback.
+  The coordinator exports them from Airflow's config (#73842).
+- **The trigger's arguments are not Jinja-rendered**, because no Python code runs the task. A value is sent as the author wrote it.
 - **Nothing extra reaches the Dag JSON**, which carries no branch-candidate field at all. A ref is a task_id by the time the decision is sent, so each SDK stores its cases and nothing else.
 - **A group edge needs one base type per SDK** that both a task and a group satisfy, since either can sit at the end of an edge.
   Python already has it: `TaskGroup(TaskGroupMixin, DAGNode)` (`task-sdk/src/airflow/sdk/definitions/taskgroup.py:96`) and every operator inherit `DependencyMixin` (`.../definitions/_internal/mixins.py:35`), where `set_upstream` and `set_downstream` live.

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

@@ -443,6 +443,50 @@ Exactly one case is selected. Python's branch callable may return a list of task
 SDK offers that yet: put the paths that run together behind one task, or gate each with its own
 condition.
 
+Triggering another Dag
+~~~~~~~~~~~~~~~~~~~~~~
+
+``triggerDagRun`` starts another Dag's run. Pass it to ``dag.task`` in place of a handler:
+
+.. code-block:: typescript
+
+    import { triggerDagRun } from "apache-airflow-ts-sdk";
+
+    const trigger = dag.task(triggerDagRun({ dagId: "downstream_etl", waitForCompletion: true }), {
+      taskId: "trigger_downstream",
+    })();
+
+    trigger.after(loaded);
+
+The task has no handler to take an id from, so ``taskId`` is required, and the factory takes no
+inputs. The trailing spec carries the task's other options, as for any task.
+
+The task runs in the TypeScript runtime, like the Dag's other tasks, and inherits the Dag's queue. It
+behaves as ``TriggerDagRunOperator`` does. ``waitForCompletion`` polls the run every ``pokeInterval``
+seconds until it reaches one of ``allowedStates`` or ``failedStates``. With ``deferrable`` as well, the
+task defers to ``DagStateTrigger`` instead of holding a worker slot, and resumes in this runtime when
+the run finishes. ``DagStateTrigger`` runs in the Python triggerer, so the triggerer needs the standard
+provider installed.
+
+The task pushes the ``trigger_run_id`` XCom, and the task's "Triggered DAG" link opens the run it
+started.
+
+The task follows these config options, with Python's fallback when one is unset:
+
+- ``[api] base_url`` is the base of the "Triggered DAG" link. It falls back to ``/``.
+- ``[operators] default_deferrable`` is the default of ``deferrable``. It falls back to ``false``.
+- ``[triggerer] queues_enabled`` decides whether the deferred trigger gets the task's queue. It falls
+  back to ``false``, so the trigger gets none.
+
+Some of what ``TriggerDagRunOperator`` does is not offered:
+
+- ``logical_date`` and ``run_after``. The triggered run's logical date is the time the task triggers
+  it, as in Python when neither is set.
+- Jinja. Values are sent as written, so ``{{ ds }}`` in ``conf`` reaches the triggered run as that
+  literal string.
+- OpenLineage parent injection (``openlineage_inject_parent_info``). The runtime does not add the
+  parent task's OpenLineage details to ``conf``.
+
 ``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/src/coordinator/client.ts

@@ -31,6 +31,9 @@ import type {
   SetXCom,
   GetConnection,
   SkipDownstreamTasks,
+  TriggerDagRun,
+  GetDagRunState,
+  GetDag,
   ConnectionResult as WireConnectionResult,
 } from "./protocol.js";
 
@@ -86,6 +89,16 @@ export interface CoordinatorClient extends TaskClient {
 
   /** Mark direct downstream tasks of the running task as skipped; none is a no-op. */
   skipDownstreamTasks(taskIds: readonly string[]): Promise<void>;
+
+  /**
+   * Trigger a Dag run. `"already_exists"` is the one refusal a trigger task
+   * handles itself; any other error throws.
+   */
+  triggerDagRun(msg: Omit<TriggerDagRun, "type">): Promise<"triggered" | "already_exists">;
+  /** The state of a Dag run, as `GetDagRunState` reports it. */
+  getDagRunState(dagId: string, runId: string): Promise<string>;
+  /** Whether a Dag is paused. */
+  isDagPaused(dagId: string): Promise<boolean>;
 }
 
 export function createCoordinatorClient(
@@ -99,11 +112,16 @@ export function createCoordinatorClient(
     request: unknown,
     extract: (body: Record<string, unknown> | null) => T,
     absent: AbsentRowPolicy,
+    allowedError?: string,
   ): Promise<T | null> {
     logs?.debug(`${op} request`);
     const frame = await comm.request(request);
     const err = parseFrameError(frame);
     if (err) {
+      if (err.code === allowedError) {
+        logs?.debug(`${op} answered ${err.code}`);
+        return null;
+      }
       if (absent === "null" && isNotFound(err)) {
         logs?.debug(`${op} not found`, { error: err.code });
         return null;
@@ -214,6 +232,45 @@ export function createCoordinatorClient(
       await rpc("SkipDownstreamTasks", null, msg, () => undefined, "throw");
     },
 
+    // ---- Dag runs ----
+
+    async triggerDagRun(msg: Omit<TriggerDagRun, "type">) {
+      const request: TriggerDagRun = { type: "TriggerDagRun", ...msg };
+      const triggered = await rpc(
+        "TriggerDagRun",
+        null,
+        request,
+        () => "triggered" as const,
+        "throw",
+        "DAGRUN_ALREADY_EXISTS",
+      );
+      return triggered ?? "already_exists";
+    },
+
+    async getDagRunState(dagId: string, runId: string): Promise<string> {
+      const msg: GetDagRunState = { type: "GetDagRunState", dag_id: dagId, run_id: runId };
+      const state = await rpc(
+        "GetDagRunState",
+        "DagRunStateResult",
+        msg,
+        (body) => body!.state as string,
+        "throw",
+      );
+      return state!;
+    },
+
+    async isDagPaused(dagId: string): Promise<boolean> {
+      const msg: GetDag = { type: "GetDag", dag_id: dagId };
+      const paused = await rpc(
+        "GetDag",
+        "DagResult",
+        msg,
+        (body) => body!.is_paused === true,
+        "throw",
+      );
+      return paused!;
+    },
+
     // ---- Connections ----
 
     async getConnection(connId: string): Promise<ConnectionResult | null> {

ts-sdk/src/coordinator/protocol.ts

@@ -28,6 +28,7 @@ import type {
   DagFileParseRequest,
   ErrorResponse,
   DagFileParsingResult,
+  DeferTask,
   RetryTask,
   SucceedTask,
   TaskState,
@@ -47,6 +48,7 @@ export type {
 
 export type {
   DagFileParsingResult,
+  DeferTask,
   RetryTask,
   SucceedTask,
   TaskState,
@@ -65,6 +67,11 @@ export type {
   SetXCom,
   GetConnection,
   SkipDownstreamTasks,
+  TriggerDagRun,
+  GetDagRunState,
+  DagRunStateResult,
+  GetDag,
+  DagResult,
 } from "../generated/supervisor.js";
 
 // -------- Frames from supervisor --------
@@ -82,6 +89,7 @@ export type MsgFromSupervisor =
 
 export type RuntimeTaskState = WithRequiredType<TaskState>;
 export type RuntimeRetryTask = WithRequiredType<RetryTask>;
+export type RuntimeDeferTask = WithRequiredType<DeferTask>;
 export type RuntimeSucceedTask = WithRequiredType<SucceedTask> & {
   task_outlets: NonNullable<SucceedTask["task_outlets"]>;
   outlet_events: NonNullable<SucceedTask["outlet_events"]>;

ts-sdk/src/coordinator/runtime.ts

@@ -49,16 +49,18 @@ import {
   asMsgFromSupervisor,
   SUPERVISOR_API_VERSION,
   type RuntimeDagFileParsingResult,
+  type RuntimeDeferTask,
   type RuntimeRetryTask,
   type RuntimeSucceedTask,
   type RuntimeTaskState,
   type StartupDetails,
 } from "./protocol.js";
 import { getArgNames } from "../sdk/arg-names.js";
-import { bundleDags, bundleDagTaskIds, type Bundle } from "../sdk/bundle.js";
+import { bundleDags, bundleDagTaskIds, getBundleTrigger, type Bundle } from "../sdk/bundle.js";
 import { finalizeDag } from "../sdk/dag.js";
 import { SERIALIZATION_VERSION } from "../generated/dag-schema-fields.js";
 import { computeRelativeFileloc, serializeDag } from "./serde.js";
+import { runTriggerDagRun } from "./trigger-runner.js";
 import { runInTaskScope, type TaskContext } from "../sdk/task.js";
 import type { JsonValue } from "../sdk/client-types.js";
 
@@ -195,6 +197,8 @@ export async function startCoordinator(
       await sendSupervisorResponse(firstFrame.id, response, comm, runtimeLogs);
       if (response.type === "SucceedTask") {
         runtimeLogs.info("Task succeeded", { task_id: body.ti.task_id });
+      } else if (response.type === "DeferTask") {
+        runtimeLogs.info("Task deferred", { task_id: body.ti.task_id });
       }
     } else {
       const errMsg = `First frame must be DagFileParseRequest or StartupDetails, got ${body.type}`;
@@ -338,8 +342,22 @@ async function handleTask(
   logs: LogChannel,
   clientLogs: LogChannel,
   signal: AbortSignal,
-): Promise<RuntimeSucceedTask | RuntimeRetryTask | RuntimeTaskState> {
+): Promise<RuntimeSucceedTask | RuntimeRetryTask | RuntimeTaskState | RuntimeDeferTask> {
   const ti = details.ti;
+  const trigger = getBundleTrigger(bundle, ti.dag_id, ti.task_id);
+  if (trigger) {
+    const ctx = buildContext(details, signal);
+    const client = createCoordinatorClient(comm, ctx, clientLogs);
+    const fail = (message: string) => {
+      logs.error("Task failed", { task_id: ctx.taskId, error: message });
+      return buildFailureResponse(details, message);
+    };
+    try {
+      return await runTriggerDagRun(details, trigger, client, logs, ctx.signal, fail);
+    } catch (err) {
+      return fail((err as Error).message ?? String(err));
+    }
+  }
   const handler = bundle.getTaskHandler(ti.dag_id, ti.task_id);
 
   if (!handler) {

ts-sdk/src/coordinator/serde.ts

@@ -50,6 +50,7 @@ import {
   type Dag,
   type RecordedInputs,
   type TaskGroupRecord,
+  type TaskRecord,
 } from "../sdk/dag.js";
 
 /** A serialized Dag: JSON, by the time it reaches the supervisor as msgpack. */
@@ -83,6 +84,29 @@ const TASK_MODULE = "airflow.sdk.coordinators.node";
  */
 const TASK_LANGUAGE = "typescript";
 
+/** The Dags this one triggers, for the UI dependency graph. */
+function serializeDagDependencies(dag: Dag): SerializedValue {
+  const dependencies: SerializedValue[] = [];
+  for (const [taskId, record] of getDagTaskRecords(dag)) {
+    if (!record.trigger) continue;
+    dependencies.push({
+      source: dag.dagId,
+      target: record.trigger.dagId,
+      label: taskId,
+      dependency_type: "trigger",
+      dependency_id: taskId,
+    });
+  }
+  return dependencies;
+}
+
+/** Makes the UI draw a trigger task as `TriggerDagRunOperator` with its link. */
+const TRIGGER_DAG_RUN_FIELDS: Readonly<Record<string, SerializedValue>> = {
+  _operator_name: "TriggerDagRunOperator",
+  ui_color: "#ffefeb",
+  _operator_extra_links: { "Triggered DAG": "_link_TriggerDagRunLink" },
+};
+
 /** How one set of authoring fields is written into a serialized object. */
 interface FieldRules {
   readonly fields: Readonly<Record<string, SchemaField>>;
@@ -144,13 +168,13 @@ export function serializeDag(
       serializeTask(
         dag.dagId,
         taskId,
-        withDagQueue(record.spec, dag.spec.queue),
+        record,
         graph.downstreamTaskIds.get(taskId),
         inputs.get(taskId),
-        record.canSkipDownstream === true,
+        dag.spec.queue,
       ),
     ),
-    dag_dependencies: [],
+    dag_dependencies: serializeDagDependencies(dag),
     task_group: serializeTaskGroups(dag, graph),
     edge_info: {},
     params: [],
@@ -182,10 +206,10 @@ function withDagQueue(spec: object, dagQueue: string | undefined): object {
 function serializeTask(
   dagId: string,
   taskId: string,
-  spec: object,
+  record: TaskRecord,
   downstream: ReadonlySet<string> | undefined,
   inputs: RecordedInputs | undefined,
-  canSkipDownstream: boolean,
+  dagQueue: string | undefined,
 ): SerializedValue {
   const data: Record<string, SerializedValue> = {
     task_id: taskId,
@@ -202,11 +226,17 @@ function serializeTask(
     is_stub: true,
   };
   const label = `task "${taskId}" of Dag "${dagId}"`;
+  if (record.trigger) {
+    if (record.trigger.conf !== undefined) {
+      toPlainJson(record.trigger.conf, `conf of ${label}`);
+    }
+    Object.assign(data, structuredClone(TRIGGER_DAG_RUN_FIELDS));
+  }
   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 (record.canSkipDownstream) data["_can_skip_downstream"] = true;
+  applySchemaFields(data, withDagQueue(record.spec, dagQueue), TASK_FIELD_RULES, label);
   if (downstream?.size) {
     data["downstream_task_ids"] = [...downstream].sort();
   }