Go SDK: run TriggerDagRun tasks in the Go runtime

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

@@ -1621,7 +1621,7 @@
               "ratio": 0.5,
               "rows": 2
             },
-            "logical_date": "{{ ds }}",
+            "logical_date": "2026-09-30 00:00:00+00:00",
             "skip_when_already_exists": true,
             "task_id": "trigger_all",
             "task_type": "TriggerDagRunOperator",
@@ -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.go

@@ -229,8 +229,9 @@ func (m *dagMap) ListDagSourceFiles() map[string]string {
 	return files
 }
 
-// lookupTask returns the Go function of a task of a registered Dag. A task from TriggerDagRun has
-// none, because a Python worker runs it.
+// lookupTask returns the task of a registered Dag. It is the Go function of a task from DagRef.Task,
+// or a bundle.TriggerTask with the options of a task from TriggerDagRun, which the runtime runs
+// itself.
 func (m *dagMap) lookupTask(dagID, taskID string) (bundle.Task, bool) {
 	m.mu.Lock()
 	dag, ok := m.dags[dagID]
@@ -242,7 +243,13 @@ func (m *dagMap) lookupTask(dagID, taskID string) (bundle.Task, bool) {
 	dag.mu.Lock()
 	defer dag.mu.Unlock()
 	ref, ok := dag.tasksByID[taskID]
-	if !ok || ref.task == nil {
+	if !ok {
+		return nil, false
+	}
+	if ref.triggerDagRun != nil {
+		return &bundle.TriggerTask{Spec: triggerSpec(*ref.triggerDagRun)}, true
+	}
+	if ref.task == nil {
 		return nil, false
 	}
 	return ref.task, true

go-sdk/airflow/bundle_test.go

@@ -25,6 +25,7 @@ import (
 	"reflect"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -471,7 +472,6 @@ func TestServeLooksUpTheTasksOfDagsAndTaskHandlers(t *testing.T) {
 		assert.NotNil(t, task)
 	}
 	for _, id := range [][2]string{
-		{"native_etl", "trigger"},
 		{"native_etl", "load"},
 		{"py_etl", "extract"},
 		{"unknown", "extract"},
@@ -480,3 +480,60 @@ func TestServeLooksUpTheTasksOfDagsAndTaskHandlers(t *testing.T) {
 		assert.False(t, ok, "%s.%s", id[0], id[1])
 	}
 }
+
+func TestServeLooksUpATriggerDagRunTaskWithACopyOfItsSpec(t *testing.T) {
+	poke := 30 * time.Second
+	logicalDate := time.Date(2026, 9, 30, 0, 0, 0, 0, time.UTC)
+	dag := Dag("native_etl")
+	dag.Task(TriggerDagRun(TriggerDagRunSpec{
+		DagID:                 "downstream_etl",
+		RunID:                 "etl_downstream",
+		Note:                  "from etl",
+		Conf:                  map[string]any{"tables": []any{"rows"}},
+		LogicalDate:           logicalDate,
+		ResetDagRun:           true,
+		WaitForCompletion:     true,
+		SkipWhenAlreadyExists: true,
+		FailWhenDagIsPaused:   true,
+		PokeInterval:          &poke,
+		AllowedStates:         []DagRunState{DagRunStateSuccess, DagRunStateQueued},
+		FailedStates:          []DagRunState{},
+		Deferrable:            ptr(true),
+	}), TaskSpec{TaskID: "trigger"})
+	dag.Task(TriggerDagRun(TriggerDagRunSpec{DagID: "bare"}), TaskSpec{TaskID: "bare"})
+	b := Bundle()
+	b.Register(dag)
+	source := coordinatorSource{&b.taskHandlers, &b.dags}
+
+	task, ok := source.LookupTask("native_etl", "trigger")
+	require.True(t, ok)
+	trigger, ok := task.(*bundle.TriggerTask)
+	require.True(t, ok, "got %T", task)
+	assert.Equal(t, bundle.TriggerSpec{
+		DagID:                 "downstream_etl",
+		RunID:                 "etl_downstream",
+		Note:                  "from etl",
+		Conf:                  map[string]any{"tables": []any{"rows"}},
+		LogicalDate:           logicalDate,
+		ResetDagRun:           true,
+		WaitForCompletion:     true,
+		SkipWhenAlreadyExists: true,
+		FailWhenDagIsPaused:   true,
+		PokeInterval:          &poke,
+		AllowedStates:         []string{"success", "queued"},
+		FailedStates:          []string{},
+		Deferrable:            ptr(true),
+	}, trigger.Spec)
+
+	// The runtime gets a copy, so that nothing it does changes the task.
+	trigger.Spec.Conf["tables"].([]any)[0] = "changed"
+	again, _ := source.LookupTask("native_etl", "trigger")
+	assert.Equal(t, []any{"rows"}, again.(*bundle.TriggerTask).Spec.Conf["tables"])
+
+	task, ok = source.LookupTask("native_etl", "bare")
+	require.True(t, ok)
+	spec := task.(*bundle.TriggerTask).Spec
+	assert.Nil(t, spec.AllowedStates)
+	assert.Nil(t, spec.FailedStates, "an unset FailedStates stays nil, which means the default")
+	assert.Nil(t, spec.Conf)
+}

go-sdk/airflow/conformance_test.go

@@ -289,6 +289,13 @@ func conformanceTriggerDagRun(
 				key,
 			)
 		}
+		if moment, ok := target.(*time.Time); ok {
+			// The value is spelled as Python's str(datetime), such as "2026-09-30 00:00:00+00:00".
+			parsed, err := time.Parse("2006-01-02 15:04:05Z07:00", value.Value)
+			require.NoError(t, err, "%s: %s", label, key)
+			*moment = parsed
+			continue
+		}
 		require.NoError(t, value.Decode(target), "%s: %s", label, key)
 	}
 	return spec

go-sdk/airflow/serialize.go

@@ -210,7 +210,7 @@ var dagSpecRules = specRules{
 	fields: dagSpecFields,
 	set:    map[string]bool{"Tags": true},
 	skip: map[string]bool{
-		// serializeTaskLocked writes Queue as the queue of each Go task whose TaskSpec sets no Queue.
+		// serializeTaskLocked writes Queue as the queue of each task whose TaskSpec sets no Queue.
 		"Queue": true,
 		// serializeTimetable writes Schedule as the timetable.
 		"Schedule": true,
@@ -330,6 +330,17 @@ func jsonNumber(value any) (float64, bool) {
 // to the microsecond. A Python datetime holds nothing finer than a microsecond.
 func encodeTime(t time.Time) float64 { return float64(t.UnixMicro()) / 1e6 }
 
+// pythonStr returns t as Python's str(datetime) writes it in UTC, such as
+// "2026-09-30 00:00:00+00:00". It writes the microseconds, six digits, only when they are not zero.
+func pythonStr(t time.Time) string {
+	t = t.UTC()
+	layout := "2006-01-02 15:04:05"
+	if t.Nanosecond()/1000 != 0 {
+		layout += ".000000"
+	}
+	return t.Format(layout) + "+00:00"
+}
+
 // checkTime returns an error for a time that a Python datetime cannot hold, which is a time whose
 // year in UTC is not from 1 to 9999. Airflow would reject a serialized Dag with such a time when it
 // loads the Dag. field names the field that holds t. The zero Time passes, because it means that
@@ -425,9 +436,10 @@ func (d *DagRef) serializeTaskLocked(task *TaskRef) map[string]any {
 			// condition.
 			data["_can_skip_downstream"] = true
 		}
-		if spec.Queue == "" {
-			spec.Queue = d.spec.Queue
-		}
+	}
+	// A task from TriggerDagRun takes the queue of the Dag too, because the Go runtime runs it.
+	if spec.Queue == "" {
+		spec.Queue = d.spec.Queue
 	}
 	writeSpecFields(data, spec, taskSpecRules)
 	if len(task.downstreams) > 0 {
@@ -461,12 +473,12 @@ func serializeArgBindings(inputs []taskInput) []any {
 }
 
 // writeTriggerDagRun writes a task from TriggerDagRun as Python's serializer writes a
-// TriggerDagRunOperator, because a Python worker runs the task.
+// TriggerDagRunOperator, so that the Airflow UI and the Dag dependencies treat it as one.
 //
 // For a Python Dag, Python's serializer writes the template fields of the operator and leaves out
-// its other parameters. A Python worker gets those other parameters by parsing the Python Dag file
-// again. A Go Dag has no Python Dag file, so writeTriggerDagRun also writes each other parameter
-// that spec sets, under the name that TriggerDagRunOperator gives the parameter.
+// its other parameters. A Go Dag has no Python Dag file, so writeTriggerDagRun also writes each
+// other parameter that spec sets, under the name that TriggerDagRunOperator gives the parameter.
+// The Go runtime runs the task from its own copy of spec and reads none of this.
 func writeTriggerDagRun(data map[string]any, spec TriggerDagRunSpec) {
 	data["task_type"] = triggerDagRunTaskType
 	data["_task_module"] = triggerDagRunTaskModule
@@ -481,8 +493,12 @@ func writeTriggerDagRun(data map[string]any, spec TriggerDagRunSpec) {
 	}
 	// Python leaves out a template field that is None, but the default logical_date is NOTSET, a
 	// sentinel that lets TriggerDagRunOperator pick the logical date. Python writes the sentinel as
-	// the string NOTSET, and Airflow loads the string as it is.
-	data["logical_date"] = cmp.Or(spec.LogicalDate, "NOTSET")
+	// the string NOTSET, and Airflow loads the string as it is. Python writes a datetime in a
+	// template field as str(datetime), not as a {"__type": "datetime"} object.
+	data["logical_date"] = "NOTSET"
+	if !spec.LogicalDate.IsZero() {
+		data["logical_date"] = pythonStr(spec.LogicalDate)
+	}
 	if spec.Conf != nil {
 		data["conf"] = copyJSON(spec.Conf)
 	}

go-sdk/airflow/serialize_test.go

@@ -355,7 +355,7 @@ func TestSerializeWritesTimesAndDurationsToTheMicrosecond(t *testing.T) {
 	assert.NotContains(t, got, "end_date")
 }
 
-func TestSerializeGivesEachGoTaskTheQueueOfTheDag(t *testing.T) {
+func TestSerializeGivesEachTaskTheQueueOfTheDag(t *testing.T) {
 	dag := Dag("etl", DagSpec{Queue: "golang"})
 	dag.Task(extract)
 	dag.Task(ping, TaskSpec{Queue: "golang_large"})
@@ -371,9 +371,13 @@ func TestSerializeGivesEachGoTaskTheQueueOfTheDag(t *testing.T) {
 	assert.Equal(t, "golang", serializedTask(t, got, "extract")["queue"])
 	assert.Equal(t, "golang_large", serializedTask(t, got, "ping")["queue"])
 	assert.Equal(t, "golang", serializedTask(t, got, "hasRows")["queue"])
-	// A task from TriggerDagRun runs on a Python worker.
-	assert.NotContains(t, serializedTask(t, got, "trigger"), "queue")
+	// The Go runtime runs a task from TriggerDagRun, so it takes the queue as a Go task does.
+	assert.Equal(t, "golang", serializedTask(t, got, "trigger")["queue"])
 	assert.Equal(t, "python", serializedTask(t, got, "trigger_on_python")["queue"])
+
+	noQueue := Dag("etl")
+	noQueue.Task(TriggerDagRun(TriggerDagRunSpec{DagID: "reports"}), TaskSpec{TaskID: "trigger"})
+	assert.NotContains(t, serializedTask(t, serializedDag(t, noQueue), "trigger"), "queue")
 }
 
 func TestSerializeWritesTheTimetableOfTheSchedule(t *testing.T) {
@@ -901,9 +905,9 @@ 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:           "{{ ds }}",
+		LogicalDate:           time.Date(2026, 1, 31, 8, 30, 0, 500000000, time.UTC),
 		RunAfter:              time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC),
 		ResetDagRun:           true,
 		WaitForCompletion:     true,
@@ -934,8 +938,8 @@ 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 }}",
-		"logical_date": "{{ ds }}",
+		"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,
 		"skip_when_already_exists": true,
@@ -947,10 +951,48 @@ func TestSerializeWritesATriggerDagRunAsATriggerDagRunOperator(t *testing.T) {
 		"fail_when_dag_is_paused": true,
 		"note": "from etl",
 		"deferrable": false,
+		"queue": "golang",
 		"retries": 1
 	}`, got)
 }
 
+func TestSerializeWritesTheLogicalDateOfATriggerDagRunAsPythonStr(t *testing.T) {
+	tests := []struct {
+		name string
+		date time.Time
+		want string
+	}{
+		{
+			"whole seconds",
+			time.Date(2026, 9, 30, 0, 0, 0, 0, time.UTC),
+			"2026-09-30 00:00:00+00:00",
+		},
+		{
+			"microseconds are not trimmed",
+			time.Date(2026, 9, 30, 1, 2, 3, 120000000, time.UTC),
+			"2026-09-30 01:02:03.120000+00:00",
+		},
+		{
+			"another zone is written in UTC",
+			time.Date(2026, 9, 30, 9, 0, 0, 0, time.FixedZone("UTC+9", 9*3600)),
+			"2026-09-30 00:00:00+00:00",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			dag := Dag("etl")
+			dag.Task(
+				TriggerDagRun(TriggerDagRunSpec{DagID: "reports", LogicalDate: tc.date}),
+				TaskSpec{TaskID: "trigger"},
+			)
+
+			got := serializedTask(t, serializedDag(t, dag), "trigger")
+
+			assert.Equal(t, tc.want, got["logical_date"])
+		})
+	}
+}
+
 func TestSerializeWritesOnlyTheTriggerDagRunOptionsThatAreSet(t *testing.T) {
 	dag := Dag("etl")
 	dag.Task(TriggerDagRun(TriggerDagRunSpec{DagID: "reports"}), TaskSpec{TaskID: "trigger"})