Go SDK: reject a templated DagID or RunID of TriggerDagRun

Mark the skipped and deferred states supported, since a TriggerDagRun
task now emits both.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: jason810496

airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json

@@ -1637,7 +1637,7 @@
               "conf": "py"
             },
             "trigger_dag_id": "conformance_minimal",
-            "trigger_run_id": "triggered_{{ run_id }}",
+            "trigger_run_id": "triggered_by_conformance",
             "ui_color": "#ffefeb",
             "wait_for_completion": true
           }

go-sdk/README.md

@@ -337,8 +337,8 @@ prek hook regenerate it.
 | state: `success` | MUST | ✓ | 3.3 |  |
 | state: `failed` | MUST | ✓ | 3.3 |  |
 | state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask |
-| state: `skipped` | SHOULD | ✗ | – | runtime does not emit TaskState skipped yet |
-| state: `deferred` | MAY | ✗ | – | runtime does not emit DeferTask yet |
+| state: `skipped` | SHOULD | ✓ | 3.4 | airflow.TriggerDagRun with SkipWhenAlreadyExists only |
+| state: `deferred` | MAY | ✓ | 3.4 | airflow.TriggerDagRun with Deferrable only; resumes in the Go runtime |
 | state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet |
 | state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet |
 | state: `removed` | MAY | ✓ | 3.3 |  |

go-sdk/airflow/bundle_test.go

@@ -487,7 +487,7 @@ func TestServeLooksUpATriggerDagRunTaskWithACopyOfItsSpec(t *testing.T) {
 	dag := Dag("native_etl")
 	dag.Task(TriggerDagRun(TriggerDagRunSpec{
 		DagID:                 "downstream_etl",
-		RunID:                 "{{ run_id }}",
+		RunID:                 "etl_downstream",
 		Note:                  "from etl",
 		Conf:                  map[string]any{"tables": []any{"rows"}},
 		LogicalDate:           logicalDate,
@@ -511,7 +511,7 @@ func TestServeLooksUpATriggerDagRunTaskWithACopyOfItsSpec(t *testing.T) {
 	require.True(t, ok, "got %T", task)
 	assert.Equal(t, bundle.TriggerSpec{
 		DagID:                 "downstream_etl",
-		RunID:                 "{{ run_id }}",
+		RunID:                 "etl_downstream",
 		Note:                  "from etl",
 		Conf:                  map[string]any{"tables": []any{"rows"}},
 		LogicalDate:           logicalDate,

go-sdk/airflow/serialize_test.go

@@ -905,7 +905,7 @@ func TestSerializeWritesATriggerDagRunAsATriggerDagRunOperator(t *testing.T) {
 	dag := Dag("etl", DagSpec{Queue: "golang"})
 	dag.Task(TriggerDagRun(TriggerDagRunSpec{
 		DagID:                 "downstream_etl",
-		RunID:                 "{{ run_id }}",
+		RunID:                 "etl_downstream",
 		Conf:                  map[string]any{"rows": 2, "nested": map[string]any{"ok": true}},
 		LogicalDate:           time.Date(2026, 1, 31, 8, 30, 0, 500000000, time.UTC),
 		RunAfter:              time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC),
@@ -938,7 +938,7 @@ func TestSerializeWritesATriggerDagRunAsATriggerDagRunOperator(t *testing.T) {
 		"template_fields_renderers": {"conf": "py"},
 		"_operator_extra_links": {"Triggered DAG": "_link_TriggerDagRunLink"},
 		"trigger_dag_id": "downstream_etl",
-		"trigger_run_id": "{{ run_id }}",
+		"trigger_run_id": "etl_downstream",
 		"logical_date": "2026-01-31 08:30:00.500000+00:00",
 		"conf": {"rows": 2, "nested": {"ok": true}},
 		"wait_for_completion": true,

go-sdk/airflow/trigger_dag_run.go

@@ -38,8 +38,9 @@ import (
 // leaves its parameter at the Python default. PokeInterval and Deferrable are pointers so that a
 // pointer to 0 or false can set that value instead of leaving the default.
 //
-// The task does not render templates. DagID, RunID, Note and the values in Conf are sent as they
-// are, so a string such as "{{ ds }}" reaches the new Dag run unchanged. The task does not offer
+// The task does not render templates. Note and the values in Conf are sent as they are, so a
+// string such as "{{ ds }}" reaches the new Dag run unchanged. DagID and RunID must not contain
+// "{{", since Airflow rejects such a dag_id or run_id. The task does not offer
 // openlineage_inject_parent_info, so it adds nothing to Conf.
 type TriggerDagRunSpec struct {
 	// DagID is the dag_id of the Dag to trigger. It is required.
@@ -125,6 +126,16 @@ func copyTriggerDagRunSpec(spec TriggerDagRunSpec) (TriggerDagRunSpec, error) {
 	if spec.DagID == "" {
 		return TriggerDagRunSpec{}, errors.New("airflow.TriggerDagRunSpec has no DagID")
 	}
+	for _, field := range []struct{ name, value string }{
+		{"DagID", spec.DagID}, {"RunID", spec.RunID},
+	} {
+		if strings.Contains(field.value, "{{") {
+			return TriggerDagRunSpec{}, fmt.Errorf(
+				"airflow.TriggerDagRunSpec.%s is %q; the task does not render templates",
+				field.name, field.value,
+			)
+		}
+	}
 	if poke := spec.PokeInterval; poke != nil && (*poke < 0 || *poke%time.Second != 0) {
 		return TriggerDagRunSpec{}, fmt.Errorf(
 			"airflow.TriggerDagRunSpec.PokeInterval is %v; "+

go-sdk/airflow/trigger_dag_run_test.go

@@ -33,7 +33,7 @@ func TestTriggerDagRunIsATask(t *testing.T) {
 	gate := dag.Task(extract)
 	spec := TriggerDagRunSpec{
 		DagID:                 "downstream_etl",
-		RunID:                 "{{ run_id }}_downstream",
+		RunID:                 "etl_downstream",
 		Conf:                  map[string]any{"source": "etl"},
 		LogicalDate:           time.Date(2026, 9, 29, 0, 0, 0, 0, time.UTC),
 		RunAfter:              time.Date(2026, 9, 30, 0, 0, 0, 0, time.UTC),
@@ -111,6 +111,20 @@ func TestTriggerDagRunRejectsAnInvalidSpec(t *testing.T) {
 			trigger: TriggerDagRun(TriggerDagRunSpec{WaitForCompletion: true}),
 			want:    "airflow.TriggerDagRunSpec has no DagID",
 		},
+		{
+			name:    "templated DagID",
+			trigger: TriggerDagRun(TriggerDagRunSpec{DagID: "{{ params.target }}"}),
+			want: `airflow.TriggerDagRunSpec.DagID is "{{ params.target }}"; ` +
+				"the task does not render templates",
+		},
+		{
+			name: "templated RunID",
+			trigger: TriggerDagRun(
+				TriggerDagRunSpec{DagID: "downstream_etl", RunID: "triggered_{{ run_id }}"},
+			),
+			want: `airflow.TriggerDagRunSpec.RunID is "triggered_{{ run_id }}"; ` +
+				"the task does not render templates",
+		},
 		{
 			name: "negative PokeInterval",
 			trigger: TriggerDagRun(

go-sdk/capabilities.yaml

@@ -23,8 +23,9 @@ min_airflow_version: "3.4"
 # bundle metadata. The render hook fails if the two disagree.
 supervisor_schema_version: "2026-10-30"
 
-# The task runner terminates a task with SucceedTask, RetryTask, or TaskState (failed/removed); it
-# does not yet emit skipped, DeferTask, RescheduleTask, or AwaitInputTask.
+# The task runner terminates a task with SucceedTask, RetryTask, or TaskState (failed/removed); only
+# an airflow.TriggerDagRun task emits TaskState skipped or DeferTask. It does not emit RescheduleTask
+# or AwaitInputTask.
 states:
   success:
     supported: true
@@ -37,11 +38,13 @@ states:
     since: "3.3"
     note: "RetryTask"
   skipped:
-    supported: false
-    note: "runtime does not emit TaskState skipped yet"
+    supported: true
+    since: "3.4"
+    note: "airflow.TriggerDagRun with SkipWhenAlreadyExists only"
   deferred:
-    supported: false
-    note: "runtime does not emit DeferTask yet"
+    supported: true
+    since: "3.4"
+    note: "airflow.TriggerDagRun with Deferrable only; resumes in the Go runtime"
   up_for_reschedule:
     supported: false
     note: "runtime does not emit RescheduleTask yet"

go-sdk/pkg/execution/trigger_runner_test.go

@@ -204,7 +204,7 @@ func TestTriggerDagRunSendsTheOptionsAsWritten(t *testing.T) {
 	supervisor := &triggerSupervisor{}
 	spec := bundle.TriggerSpec{
 		DagID:       "downstream",
-		RunID:       "{{ run_id }}_x",
+		RunID:       "downstream_x",
 		Note:        "{{ ds }}",
 		Conf:        map[string]any{"day": "{{ ds }}", "rows": int64(2)},
 		ResetDagRun: true,
@@ -214,11 +214,11 @@ func TestTriggerDagRunSendsTheOptionsAsWritten(t *testing.T) {
 
 	assertSucceedTask(t, result)
 	sent := supervisor.request(t, "TriggerDagRun")
-	assert.Equal(t, "{{ run_id }}_x", sent["run_id"], "an explicit run_id is not generated")
+	assert.Equal(t, "downstream_x", sent["run_id"], "an explicit run_id is not generated")
 	assert.Equal(t, "{{ ds }}", sent["note"])
 	assert.Equal(t, map[string]any{"day": "{{ ds }}", "rows": int8(2)}, sent["conf"])
 	assert.Equal(t, true, sent["reset_dag_run"])
-	assert.Equal(t, "{{ run_id }}_x", supervisor.sent()[2]["value"])
+	assert.Equal(t, "downstream_x", supervisor.sent()[2]["value"])
 }
 
 func TestTriggerDagRunDatesAndRunID(t *testing.T) {

scripts/ci/lang_sdk_serialization/test_dags.yaml

@@ -303,7 +303,7 @@ dags:
       - task_id: trigger_all
         trigger_dag_run:
           trigger_dag_id: conformance_minimal
-          trigger_run_id: "triggered_{{ run_id }}"
+          trigger_run_id: "triggered_by_conformance"
           conf: {rows: 2, ratio: 0.5, nested: {tags: [a, b]}}
           logical_date: !datetime 2026-09-30T00:00:00+00:00
           wait_for_completion: true