Support TaskFlow Dag definitions natively in the Go-SDK

go-sdk/Justfile

@@ -50,3 +50,7 @@ protoc:
 # Regenerate the coordinator-protocol data models from the task-sdk supervisor schema
 generate-models:
   go generate ./pkg/execution/genmodels/...
+
+# Regenerate the bundle spec structs from the Airflow dag-serialization schema
+generate-specs:
+  go generate ./bundle/bundlev1/...

go-sdk/README.md

@@ -87,9 +87,9 @@ func (m *myBundle) GetBundleVersion() v1.BundleInfo {
 }
 
 func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
-    simpleDag := dagbag.AddDag("simple_dag")
-    simpleDag.AddTask(extract)
-    simpleDag.AddTask(transform)
+    simpleDag := dagbag.AddDag(v1.DagSpec{DagId: "simple_dag"})
+    extracted := simpleDag.Task(extract)
+    simpleDag.Task(transform, v1.Inputs(extracted))
     return nil
 }
 
@@ -100,24 +100,32 @@ func main() {
 }
 ```
 
-A task is an ordinary Go function. The runtime inspects its signature and injects arguments by type:
-`sdk.TIRunContext`, `*slog.Logger`, and an `sdk.Client` (or a narrower interface such as
-`sdk.VariableClient`). An optional `(any, error)` return becomes the task's XCom; an `error` return marks
-the task failed.
+A task is an ordinary Go function. The runtime inspects its signature and fills parameters by kind:
+runtime values (`sdk.TIRunContext`, `*slog.Logger`, and an `sdk.Client` or a narrower interface such as
+`sdk.VariableClient`) are injected by type, and every other parameter is a **data parameter**, bound
+positionally to the `v1.Inputs(...)` refs — at run time it receives the upstream task's return value,
+pulled from that task's return-value XCom and decoded into the parameter's type. A `(result, error)`
+return becomes the task's XCom; an `error` return marks the task failed.
 
 ```go
-func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (any, error) {
+type ExtractResult struct {
+    GoVersion string `json:"go_version"`
+}
+
+func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (ExtractResult, error) {
     conn, err := client.GetConnection(ctx, "test_http")
     // ... do work, honour ctx cancellation ...
-    return map[string]any{"go_version": runtime.Version()}, nil
+    return ExtractResult{GoVersion: runtime.Version()}, nil
 }
 
-func transform(ctx sdk.TIRunContext, client sdk.VariableClient, log *slog.Logger) error {
+// `in` is a data parameter: it receives extract's return value because
+// RegisterDags wired v1.Inputs(extracted).
+func transform(ctx sdk.TIRunContext, client sdk.VariableClient, log *slog.Logger, in ExtractResult) error {
     val, err := client.GetVariable(ctx, "my_variable")
     if err != nil {
         return err
     }
-    log.Info("Obtained variable", "my_variable", val)
+    log.Info("Obtained variable", "my_variable", val, "upstream_go_version", in.GoVersion)
     return nil
 }
 ```

go-sdk/bundle/bundlev1/gen.go

@@ -0,0 +1,30 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// spec.gen.go is generated from the Airflow dag-serialization schema by the
+// local gen tool: the TaskSpec field set, its Go types, and its SchemaFields
+// omit-if-default rules all derive from the "operator" definition, so they
+// cannot drift from what the scheduler deserializes. Don't edit spec.gen.go
+// by hand; change the schema (or the exclusion rules in ./gen) and re-run
+// `just generate-specs` (go generate). DagSpec and the registration Info
+// structs stay hand-written
+// in spec.go: Schedule is an SDK-level concept the schema has no scalar for,
+// and several dag keys are always-emitted with [core]-config fallbacks the
+// schema cannot express.
+package bundlev1
+
+//go:generate go run ./gen -schema ../../../airflow-core/src/airflow/serialization/schema.json -out spec.gen.go