@@ -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.
 
@@ -436,6 +436,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
+~~~~~~~~~~~~~~~~~~~~~~
+
+``dag.triggerDagRun`` starts another Dag's run, as a task of this one:
+
+.. code-block:: typescript
+
+    const trigger = dag.triggerDagRun({
+      taskId: "trigger_downstream",
+      dagId: "downstream_etl",
+      waitForCompletion: true,
+    });
+
+    trigger.after(loaded);
+
+There is nothing to call: the task takes no TypeScript arguments, so it hands back the reference
+directly. A second options object carries the task's own spec, as ``dag.task`` takes one.
+
+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.
 
 
@@ -31,6 +31,9 @@ import type {
   SetXCom,
   GetConnection,
   SkipDownstreamTasks,
+  TriggerDagRun,
+  GetDagRunState,
+  GetDag,
   ConnectionResult as WireConnectionResult,
 } from "./protocol.js";
 
@@ -83,6 +86,15 @@ export interface CoordinatorClient extends TaskClient {
    * output fails the task, while one that pushed null binds null.
    */
   getXComEntry(opts: GetXComOpts): Promise<XComEntry>;
+  /**
+   * 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(
@@ -213,6 +225,47 @@ export function createCoordinatorClient(
       await rpc("SkipDownstreamTasks", null, msg, () => undefined, "throw");
     },
 
+    // ---- Dag runs ----
+
+    async triggerDagRun(msg: Omit<TriggerDagRun, "type">) {
+      const request: TriggerDagRun = { type: "TriggerDagRun", ...msg };
+      logs?.debug("TriggerDagRun request");
+      const frame = await comm.request(request);
+      const err = parseFrameError(frame);
+      // Sent as an ErrorResponse body, not a frame error; see
+      // `ActivitySubprocess._handle_trigger_dag_run`.
+      if (err?.code === "DAGRUN_ALREADY_EXISTS") return "already_exists";
+      if (err) {
+        logs?.warning("TriggerDagRun failed", { error: err.code });
+        throw new Error(`TriggerDagRun failed: ${err.code}`);
+      }
+      return "triggered";
+    },
+
+    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> {
 
@@ -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"]>;
 
@@ -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) {
 
@@ -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,39 @@ const TASK_MODULE = "airflow.sdk.coordinators.node";
  */
 const TASK_LANGUAGE = "typescript";
 
+/**
+ * The Dags this one reaches, as the UI's dependency graph and
+ * `airflow dags show-dependencies` read them.
+ *
+ * Python derives these from the live operator with `detect_dependencies`. Only
+ * a trigger task creates one here.
+ */
+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;
+}
+
+/**
+ * What a trigger task adds to a TypeScript task, so the UI draws it as
+ * `TriggerDagRunOperator` and shows its "Triggered DAG" link. The runtime
+ * pushes the link's XCom under the key named here.
+ */
+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>>;
@@ -140,13 +174,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: [],
@@ -178,10 +212,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,
@@ -197,20 +231,63 @@ function serializeTask(
     // `get_arg_bindings` reads nothing without it.
     is_stub: true,
   };
+  if (record.trigger) {
+    // `conf` reaches Airflow at run time rather than in the Dag JSON, so a
+    // value it cannot store is caught here, as this Dag's import error.
+    if (record.trigger.conf !== undefined) {
+      plainJsonArg(record.trigger.conf, `conf of task "${taskId}" of Dag "${dagId}"`);
+    }
+    Object.assign(data, structuredClone(TRIGGER_DAG_RUN_FIELDS));
+  }
   const label = `task "${taskId}" of Dag "${dagId}"`;
   const bindings = serializeArgBindings(inputs, label);
   if (bindings) data["_arg_bindings"] = bindings;
   // What Python writes for a SkipMixin operator, and what makes
   // `NotPreviouslySkippedDep` consult this task's `skipmixin_key` XCom when one
   // of its skipped downstream tasks is cleared.
-  if (canSkipDownstream) data["_can_skip_downstream"] = true;
+  if (record.canSkipDownstream === true) data["_can_skip_downstream"] = true;
+  const spec = withDagQueue(record.spec, dagQueue);
   applySchemaFields(data, spec, TASK_FIELD_RULES, label);
   if (downstream?.size) {
     data["downstream_task_ids"] = [...downstream].sort();
   }
   return { __type: "operator", __var: data };
 }
 
+/**
+ * A trigger task's `conf`, checked to be plain JSON.
+ *
+ * A value JSON cannot carry, such as a `Date`, a `Map` or a function, is
+ * rejected, because msgpack would otherwise hand Airflow something it cannot
+ * store as a Dag run's conf.
+ */
+function plainJsonArg(value: unknown, label: string): SerializedValue {
+  if (value === null || value === undefined) return null;
+  if (typeof value === "string" || typeof value === "boolean") return value;
+  if (typeof value === "number") {
+    if (!Number.isFinite(value)) throw new Error(`${label} is not a finite number`);
+    return value;
+  }
+  if (Array.isArray(value))
+    return value.map((item, index) => plainJsonArg(item, `${label}[${index}]`));
+  if (isPlainObject(value)) {
+    const copy: Record<string, SerializedValue> = {};
+    for (const [key, item] of Object.entries(value))
+      copy[key] = plainJsonArg(item, `${label}.${key}`);
+    return copy;
+  }
+  throw new Error(
+    `${label} is ${describeType(value)}, which a Dag run's conf cannot carry; ` +
+      "pass a string, a number, a boolean, null, an array or a plain object",
+  );
+}
+
+function isPlainObject(value: unknown): value is Record<string, unknown> {
+  if (typeof value !== "object" || value === null) return false;
+  const prototype = Object.getPrototypeOf(value);
+  return prototype === Object.prototype || prototype === null;
+}
+
 /**
  * The task's arguments as the binding spec the API server hands back at run
  * time, one entry per argument in the order the call named them.