This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 24e067aeef0 Record the Go SDK's Mixed Lang Task and Native Dag task
interfaces (#72043)
24e067aeef0 is described below
commit 24e067aeef058cb66681f5a01551b3d7ba145d09
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Sep 16 22:31:22 2026 +0800
Record the Go SDK's Mixed Lang Task and Native Dag task interfaces (#72043)
---
.../adr/lang-sdk/0008-control-flow-constructs.md | 96 +++++++++++
.../0009-provider-operators-as-generated-dsl.md | 82 +++++++++
airflow-core/adr/lang-sdk/README.md | 2 +
go-sdk/README.md | 9 +
.../adr/0006-mixed-lang-task-handler-interface.md | 171 ++++++++++++++++++
go-sdk/adr/0007-native-dag-interface.md | 191 +++++++++++++++++++++
6 files changed, 551 insertions(+)
diff --git a/airflow-core/adr/lang-sdk/0008-control-flow-constructs.md
b/airflow-core/adr/lang-sdk/0008-control-flow-constructs.md
new file mode 100644
index 00000000000..c8599c4f551
--- /dev/null
+++ b/airflow-core/adr/lang-sdk/0008-control-flow-constructs.md
@@ -0,0 +1,96 @@
+<!--
+ 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.
+ -->
+
+# ADR-0008: Control-Flow Constructs in Lang SDKs
+
+## Status
+
+Proposed.
+
+## Decision
+
+1. **Name each construct after the host language's control flow**, Conditional
skipping and branching are `if`/`else` and `switch`/`case` in most languages;
the `Operator` suffix and the `Python` infix carry nothing an author of another
language needs.
+ `If`, with or without `Else`, and `Switch` all serialize as a branch
operator.
+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.
+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.
+
+## Context
+
+A native Dag interface starts with "register a task, declare an edge".
+Four constructs follow immediately, because Python Dags use them everywhere:
+
+- grouping (`TaskGroup`)
+- conditional skipping (`ShortCircuitOperator`, `@task.short_circuit`)
+- branching (`BranchPythonOperator`, `@task.branch`)
+- triggering another Dag's run (`TriggerDagRunOperator`)
+
+Every Lang SDK has to decide how to spell them, and copying Python's class
names is the tempting default but the existing convention of Python SDK might
not be straightforward for other Lang SDKs.
+For example `dag.ShortCircuitOperator(...)` asks an author who has never seen
Airflow to learn Python's operator taxonomy in order to write an `if`.
+Feeling native to its own language matters more than the SDKs looking alike.
+
+## Example
+
+The Go SDK spelling of all four. Every callable takes `airflow.Context` first,
and a decider returns the case it picked, which the SDK sends as that task's
task_id:
+
+```go
+func hasRows(actx airflow.Context, rows Rows) (bool, error)
+func pickPath(actx airflow.Context) (*airflow.TaskRef, error)
+```
+
+```go
+group := dag.TaskGroup("transform")
+cleaned := group.Task(cleanRows)
+validated := group.Task(validateRows, airflow.Inputs(cleaned))
+
+loadIfReadyRef := dag.Task(loadIfReady)
+loadFallbackRef := dag.Task(loadFallback)
+handleLongRef := dag.Task(handleLong)
+handleShortRef := dag.Task(handleShort)
+
+gate := dag.If(hasRows, airflow.Inputs(validated)) // BranchOperator: skips
the side not taken
+gate.Then(loadIfReadyRef)
+gate.Else(loadFallbackRef)
+
+pick := dag.Switch(pickPath) // task_id pickPath, from the function name
+pick.Case(handleLongRef).
+ Case(handleShortRef)
+
+dag.Task(airflow.TriggerDagRun(airflow.TriggerDagRunSpec{DagId:
"downstream_etl"}), airflow.TaskSpec{TaskId: "trigger_downstream"}).After(gate)
+```
+
+A decider has to see the refs it returns, so either they are package-level or
it is a closure where the Dag is built.
+
+A one-sided `If` is a branch with one candidate rather than a
`ShortCircuitOperator`.
+That operator defaults to skipping every task in its downstream closure and
ignoring their trigger rules, where a branch skips only the immediate
downstream it did not take and keeps a task that several branches converge on
running.
+
+## Consequences
+
+- **A branch selects exactly one task.** Python's branch callable may return a
list of task_ids, and `skip_all_except` handles it;
+ 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.
+- **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.
+ The Go shape is `airflow.Node`
([`go-sdk/adr/0007`](../../../go-sdk/adr/0007-native-dag-interface.md)).
diff --git
a/airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md
b/airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md
new file mode 100644
index 00000000000..035e4f1f78b
--- /dev/null
+++ b/airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md
@@ -0,0 +1,82 @@
+<!--
+ 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.
+ -->
+
+# ADR-0009: Provider Operators as Generated Lang-SDK DSL
+
+## Status
+
+Proposed.
+
+## Decision
+
+1. **Provider operators reach every Lang SDK as generated, serialization-only
bindings.**
+ Only a task wrapping a host-language function executes in that language; a
generated operator carries no body.
+ Both go through the SDK's ordinary task registration.
+2. **A provider DSL task runs on a Python worker**, so it must not inherit the
SDK's queue, and the deployment must have that provider installed.
+3. **Generate from the Python constructors, commit the output, and guard it
with a prek hook.**
+4. **Generate an operator only when every required constructor parameter is
JSON-serializable**
+ (primitive, list, dict, or a nested spec of those); omit optional
parameters that are not;
+ skip entirely any operator requiring a callable or a live object.
+5. **The namespace mirrors `providers/`, adapted to each language's naming
rules.**
+6. **Templated fields pass through untouched.** The SDK writes the Jinja
string; rendering stays server-side, where it already happens.
+7. **Version skew warns at Dag parsing time and never blocks execution.**
Warning on SDK provider DSL version and the server-side Python provider runtime
version mismatch, shouldn't be a fatal error.
+8. **Bindings ship as one package per provider from day one**, on that
provider's release cadence, with an aggregate pin published alongside.
+
+## Context
+
+The design review on #72043 asked whether authoring a Dag in Go means giving
up Python provider operators.
+
+**It must not**, for any language: Airflow's ~100 provider distributions are
what a native Lang-SDK Dag cannot afford to lose, and what no workflow engine
outside Airflow's ecosystem can offer.
+A native Dag serializes into the same Dag JSON a Python Dag produces, so any
operator whose
+constructor arguments are JSON-representable can be expressed from another
language as a DSL that
+emits serialization and nothing else.
+
+## Example
+
+The Go SDK spelling, mixing a native task with two generated operators:
+
+```go
+import (
+ "github.com/apache/airflow/go-sdk/airflowprovider/amazon"
+ "github.com/apache/airflow/go-sdk/airflowprovider/cncf/kubernetes"
+)
+
+extracted := dag.Task(extract) // native Go: runs on a Go worker
+
+staged := dag.Task(amazon.S3ToRedshiftOperator{
+ SchemaName: "public", TableName: "events", S3Bucket: "raw", S3Key:
"events/{{ ds }}",
+}, airflow.TaskSpec{TaskId: "stage"}).After(extracted) // DSL only: runs on a
Python worker
+
+dag.Task(kubernetes.KubernetesPodOperator{
+ Namespace: "airflow", Image: "report:latest", Name: "report",
+}, airflow.TaskSpec{TaskId: "report"}).After(staged)
+```
+
+`airflow.TriggerDagRun` ([ADR-0008](0008-control-flow-constructs.md)) follows
the same concept but it is the hand-written.
+
+## Consequences
+
+- **Authors should pin the DSL to the provider version their deployment has
installed.** The bindings
+ describe one provider version's constructors, the operator that actually
runs is whatever the
+ Python worker imports, and skew between them only warns. Matching the two is
what keeps that
+ warning from becoming a surprise at run time.
+- **Most skew is harmless, so it must not be fatal.** Python already fails
loudly and precisely when
+ a class or argument genuinely is not there, and a parse-time hard failure
would take a whole Dag
+ out over a version difference its tasks may not even touch. The hook also
emits a coverage report,
+ so which operators each SDK can reach is reviewable.
diff --git a/airflow-core/adr/lang-sdk/README.md
b/airflow-core/adr/lang-sdk/README.md
index b6768d61396..359275ac269 100644
--- a/airflow-core/adr/lang-sdk/README.md
+++ b/airflow-core/adr/lang-sdk/README.md
@@ -33,6 +33,8 @@ bind core interfaces and apply to every language SDK, not
just the Java SDK.
- [ADR-0005](0005-coordinator-packaging.md): coordinator packaging, module
layout, and registration.
- [ADR-0006](0006-no-lang-sdk-source-display.md): no Lang-SDK source display
for mixed-language (`@task.stub`) Dags.
- [ADR-0007](0007-taskflow-across-language-boundary.md): TaskFlow across the
language boundary — argument binding for Lang-SDK tasks.
+- [ADR-0008](0008-control-flow-constructs.md): control-flow constructs —
grouping, conditional skipping, branching, and triggering a Dag run.
+- [ADR-0009](0009-provider-operators-as-generated-dsl.md): provider operators
as generated, serialization-only DSL in every Lang SDK.
Decisions specific to a single SDK stay next to that SDK — for example, the Go
SDK's bundle-format
decisions live in [`go-sdk/adr/`](../../../go-sdk/adr). Java-SDK-only
interface-design decisions —
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 8fb6564c9e7..14e44f3c531 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -323,9 +323,18 @@ The [`adr/`](./adr) directory records the design decisions
behind the SDK:
the executable *is* the bundle.
- [ADR 0005](./adr/0005-retire-go-edge-worker.md): retire the standalone Go
Edge Worker and make the
coordinator the only execution path.
+- [ADR 0006](./adr/0006-mixed-lang-task-handler-interface.md): bundle
registration and the Mixed Lang
+ task handler interface — `airflow.Bundle`/`Register`/`Serve`; Cross language
TaskFlow: flat positional binding, `arg:` tagged structs, and the untagged
folded-name fallback.
+- [ADR 0007](./adr/0007-native-dag-interface.md): the proposed Native Dag
interface (`airflow.Dag`/
+ `dag.Task`/`airflow.Inputs`/`Before`-`After`)
Cross-cutting Lang-SDK decisions — the coordinator architecture and how
non-Python tasks integrate with
Airflow core surfaces — are recorded in
[`airflow-core/adr/lang-sdk/`](../airflow-core/adr/lang-sdk).
+Two of them shape the interfaces above:
+[ADR-0008](../airflow-core/adr/lang-sdk/0008-control-flow-constructs.md) for
grouping, conditions,
+branching, and triggering a Dag run, and
+[ADR-0009](../airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md)
for reaching
+Python provider operators from a native Dag.
The normative, language-agnostic on-disk bundle format (the footer layout,
manifest fields, and what the
`ExecutableCoordinator` reads) is specified in
diff --git a/go-sdk/adr/0006-mixed-lang-task-handler-interface.md
b/go-sdk/adr/0006-mixed-lang-task-handler-interface.md
new file mode 100644
index 00000000000..6150ca4c2fc
--- /dev/null
+++ b/go-sdk/adr/0006-mixed-lang-task-handler-interface.md
@@ -0,0 +1,171 @@
+<!--
+ 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.
+ -->
+
+# 6. Bundle registration and Mixed Lang task handlers
+
+Date: 2026-09-09
+
+## Status
+
+Proposed.
+
+## Decision
+
+1. **A "bundle" is a value the author builds.** `airflow.Bundle()` returns a
`*airflow.BundleRef`;
+ `main` reads build, register, serve, with `bundle.Serve()` as its last
statement.
+ It replaces `BundleProvider` and `Registry`, the callback and the write
half of the same bundle.
+2. **`bundle.Register(items ...airflow.Registraterable)`** is the single
registration verb, taking native Dags and task handlers.
+3. **A Go bundle registers task handlers, not Dags**:
`airflow.TaskHandler(dagId, taskId, fn)`, the Go body for a task Python
declares with `@task.stub`.
+4. **Both dag_id and task_id are written out on TaskHandler definition**,
because Python owns them; nothing is derived from the Go function name.
+5. **Every handler must take an `airflow.Context` first**: a struct embedding
`context.Context`, exposing `Logger()`, `Client()`, `TaskInstance()`, and
`DagRun()`.
+ What Airflow supplies a task arrives as a method on that value rather than
as a parameter of its own.
+6. **Every remaining parameter is data**, bound positionally, or by field when
it is a single struct: `arg:"..."` when tagged, else the folded Go field name.
+
+## Context
+
+Python owns everything but the body of a Mixed Lang task: `@task.stub`
declares the task, its arguments, and its place in the graph.
+The Go side has no Dag to define, so Dag vocabulary misleads.
+
+Renaming the Go function must not change which task body Airflow matches, so
`TaskHandler` names the dag_id and the task_id explicitly instead of inferring
them from the Go function name.
+Additionally, the TaskHandler shouldn't accept any spec as it should only
define the implementation of stub operator, so the `airflow.TaskHandler(dag_id,
task_id, fn)` is a much cleaner interface.
+
+Registration is inverted today. An author declares a struct with no state,
asserts it implements
+`v1.BundleProvider`, fills in `RegisterDags(dagbag v1.Registry) error`, and
hands the struct to
+`bundlev1server.Serve` — three concepts and an empty type before a single task
is declared.
+
+The term naming should be refined to reduce the new terminologies across user
interface.
+The `Registry` should be `Bundle` and the `AddDag` is mis-used for registering
the TaskHandler.
+
+The shipped signature (#70209) injects `sdk.TIRunContext`, `*slog.Logger`, and
`sdk.Client` by type.
+Calling them still needs the `context.Context` passed in by hand, which is
awkward from a Go author's perspective.
+Exposing the logger and the client on the context itself removes that, and
`airflow.Context` in the Signature section below is that shape.
+Both keep taking a context, so a call reads `actx.Logger().InfoContext(actx,
...)` or `actx.Client().GetVariable(actx, ...)` and neither holds one in a
field.
+
+## Example
+
+```go
+func main() {
+ bundle := airflow.Bundle()
+
+ bundle.Register(
+ airflow.TaskHandler("py_etl", "transform", transform),
+ airflow.TaskHandler("py_etl", "via_struct_arg_tag", ViaStructArgTag),
+ airflow.TaskHandler("py_etl", "via_struct", ViaStruct),
+ )
+
+ if err := bundle.Serve(); err != nil {
+ log.Fatal(err)
+ }
+}
+```
+
+Registration can be spread across packages, either by passing the bundle along
or by returning `[]airflow.Registraterable` for the caller:
`bundle.Register(taskflowbinding.Handlers()...)`.
+
+Three ways a Go function receives a stub task's data, all live in
`go-sdk/example/bundle/`.
+
+**Flat positional**, for `def transform(country: str, extracted: dict)`:
+
+```go
+func transform(actx airflow.Context, country string, extracted map[string]any)
error {
+ actx.Logger().InfoContext(actx, "transforming", "country", country, "try",
actx.TaskInstance().TryNumber)
+
+ threshold, err := actx.Client().GetVariable(actx, "etl_threshold")
+ if err != nil {
+ return err
+ }
+ return writeRows(actx, extracted, threshold)
+}
+```
+
+**A single `arg:`-tagged struct**, for `def via_struct_arg_tag(region_code:
str, threshold: float)`:
+
+```go
+type ViaStructArgTagInput struct {
+ Region string `arg:"region_code"`
+ Threshold float64 `arg:"threshold"`
+}
+
+func ViaStructArgTag(actx airflow.Context, input ViaStructArgTagInput) (any,
error)
+```
+
+**A single untagged struct**, where the field name folds to the Python
argument, for
+`def via_struct(region_code: str, threshold: float)`:
+
+```go
+type ViaStructInput struct {
+ RegionCode string // folds to region_code
+ Threshold float64 // folds to threshold
+}
+
+func ViaStruct(actx airflow.Context, input ViaStructInput) (any, error)
+```
+
+Folding lowercases a name and strips its underscores, on both sides:
`RegionCode` and `region_code`
+both fold to `regioncode`.
+
+## Signature
+
+```go
+package airflow
+
+func Bundle() *BundleRef
+
+func (b *BundleRef) Register(items ...Registraterable)
+func (b *BundleRef) Serve() error
+
+// Registraterable is sealed: its only method is unexported, so the set of
things a bundle
+// accepts stays closed to the SDK's own types — task handlers today, a Dag
authored in Go
+// once there is one.
+type Registraterable interface{ registraterable() }
+
+func TaskHandler(dagId, taskId string, fn any) Registraterable
+
+// Context is what every handler takes first. It is a context, so it passes
straight to the
+// logger and the client rather than being stored inside either of them.
+type Context struct {
+ context.Context
+ // unexported fields
+}
+
+func (c Context) Logger() *slog.Logger
+func (c Context) Client() sdk.Client
+func (c Context) TaskInstance() TaskInstance
+func (c Context) DagRun() DagRun
+
+// FromContext recovers the SDK surface inside a helper typed as a plain
context.Context.
+func FromContext(ctx context.Context) (Context, bool)
+```
+
+## Consequences
+
+- **Registration closes when `Serve` is called.** Registering afterwards is a
programming error and panics.
+- Graceful termination needs no unwrapping — `actx.Done()` fires on supervisor
shutdown, and
+ `http.NewRequestWithContext(actx, ...)` accepts it — while cleanup that must
outlive cancellation
+ uses `context.WithoutCancel(actx)`.
+
+## Alternatives
+
+- **`airflow.TaskHandler(dagId, fn, airflow.WithTaskId(...))`**, defaulting
the task_id to the Go
+ function name. Rejected: see the ids in Context above.
+- **Package-level accessors over a plain `context.Context`**
(`airflow.Logger(ctx)`, `airflow.Client(ctx)`), leaving the handler's first
parameter as `context.Context`.
+ Rejected: `airflow.Logger(ctx)` reads oddly next to `actx.Logger()`, asking
the package for something the context already holds.
+ It also keeps the SDK surface in package functions instead of on the value,
and a context from anywhere else still compiles, so a missing logger or client
only shows up when the task runs.
+- **A logger and a client that hold the context themselves**, so a call reads
`actx.Logger().Info(...)` with nothing passed. Rejected: that is what puts a
request-scoped context in a struct field, which the context package warns
against.
+ `airflow.Context` carrying one is a different case, since it is a
purpose-built context rather than a domain type, which is why the context
itself stays a struct.
+- **Two registration verbs**, one per registerable kind. Rejected: Having
`bundle.registerTaskHandler(airflow.TaskHandler(...))` spell the exact term
twice, having a sealed type is a much cleaner interface.
diff --git a/go-sdk/adr/0007-native-dag-interface.md
b/go-sdk/adr/0007-native-dag-interface.md
new file mode 100644
index 00000000000..de1ff773a6f
--- /dev/null
+++ b/go-sdk/adr/0007-native-dag-interface.md
@@ -0,0 +1,191 @@
+<!--
+ 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.
+ -->
+
+# 7. Native Dag interface
+
+Date: 2026-09-09
+
+## Status
+
+Proposed.
+
+## Decision
+
+1. **One Dag type, constructed then registered.** `airflow.Dag(dagId, spec)`
returns a `*airflow.DagRef` that is complete before `bundle.Register(dag)`
takes it — the same verb that registers Mixed Lang task handlers ([ADR
6](0006-mixed-lang-task-handler-interface.md)).
+ Naming rule: `airflow.X(...)` constructs, `*airflow.XRef` is the entity;
what every Dag must have is a positional parameter for dag_id, and the rest
travels in a spec struct.
+2. **Tasks register through `dag.Task(fn any, opts ...airflow.TaskOption)`**,
returning a `*airflow.TaskRef`. `airflow.Inputs(...)` and a bare
`airflow.TaskSpec{}` both implement `TaskOption`.
+3. **At most one `airflow.TaskSpec` per task.** A second one is a registration
error rather than something to merge, so a task's attributes are only ever
written in one place.
+4. **task_id is the Go function name by default**, spelled exactly as the
function is, and `airflow.TaskSpec{TaskId: ...}` sets it to anything else.
+5. **`airflow.Inputs(refs...)` declares the data and the edge in one call**
for defining graph with TaskFlow syntax.
+6. **`Before` and `After` are order-only edges on `airflow.Node`**, which both
`*airflow.TaskRef` and `*airflow.TaskGroupRef` satisfy. They are the Go pair
for `>>` and `<<`, and both return their argument set as one `Node`, so
`a.Before(b, c).Before(d)` is Python's `a >> [b, c] >> d`.
+7. **An edge label wraps the endpoint**: `loaded.Before(airflow.Label(notify,
"when empty"))` is Python's `loaded >> Label("when empty") >> notify`.
Labelling the endpoint rather than the call lets one fan-out give each edge its
own label.
+8. **Trigger rules belong to the task**, as `airflow.TaskSpec{TriggerRule:
...}`, never to an edge.
+9. **A user-facing enum carries its type in the constant name** — e.g.
`airflow.TriggerRuleAllDone`.
+10. **Everything an author writes comes from one `airflow` package.**
+11. **No Go-native deferral**, and none is needed: the constructs that defer
are DSL tasks Python executes.
+12. **`DagSpec` and `TaskSpec` are generated from Airflow core's serialization
schema** (`airflow-core/src/airflow/serialization/schema.json`) into the
`airflow` package itself and committed, the way `models.gen.go` already is for
the supervisor schema.
+ `TaskSpec` implements `airflow.TaskOption`, so a generated struct travels
in the same variadic as `airflow.Inputs`.
+
+## Context
+
+A native Dag is authored entirely in Go — schedule, tasks, and dependencies —
and serializes into the Dag JSON a Python Dag would produce.
+Dependencies between Go functions have to be typed rather than looked up by
task ID, and a Dag should read like Go rather than transliterated Python.
+
+The interfaces sketched in #67155 and #70158 spread their surface across `v1`,
`sdk`, and `slog`, published a half-built Dag to the registry and mutated it
afterwards, and could declare an edge in only one direction.
+
+## Example
+
+Both forms build the same graph; which one an author writes depends on whether
the edge carries a value.
+
+**Data dependencies — the TaskFlow equivalent.** `airflow.Inputs` passes an
upstream's return value in and declares the edge in one call, as calling one
TaskFlow function with another's output does in Python (`extracted = extract();
transformed = transform(extracted); load(transformed)`).
+
+```go
+dag := airflow.Dag("etl", airflow.DagSpec{Schedule: "@daily"})
+
+extracted := dag.Task(extract)
+transformed := dag.Task(transform, airflow.Inputs(extracted))
+dag.Task(load, airflow.Inputs(transformed), airflow.TaskSpec{Retries: 2})
+
+bundle.Register(dag)
+```
+
+The task functions, where `Result` is any type the SDK can serialize to XCom —
the Go equivalent of what a TaskFlow function returns:
+
+```go
+type Result struct {
+ Message string `json:"message"`
+}
+
+func extract(actx airflow.Context) (Result, error) {
+ return Result{Message: "native Dag data"}, nil
+}
+
+func transform(actx airflow.Context, extracted Result) (Result, error) {
+ return Result{Message: "transformed " + extracted.Message}, nil
+}
+
+func load(actx airflow.Context, transformed Result) error { return nil }
+```
+
+The task_ids are the function names by default: `extract`, `transform`, and
`load`.
+
+**Order-only dependencies — the `>>` and `<<` equivalent.** For tasks that
must be ordered but exchange no data; the functions take no parameter for such
an edge.
+
+```go
+loaded := dag.Task(load, airflow.Inputs(transformed))
+cleaned := dag.Task(cleanup, airflow.TaskSpec{TriggerRule:
airflow.TriggerRuleAllDone})
+notified := dag.Task(notify)
+emptyNotice := dag.Task(notifyEmpty, airflow.TaskSpec{TaskId: "notify_empty"})
+staging := dag.TaskGroup("staging") // a group carries edges like a task
+staging.Task(stageRows) // tasks join a group through the group
+
+staging.Before(loaded) // staging >> load
+loaded.Before(notified, cleaned) // loaded >> [notify, cleanup]
+cleaned.After(extracted) // cleanup << extracted
+
+loaded.Before(airflow.Label(emptyNotice, "when empty")) // loaded >>
Label("when empty") >> notify_empty
+```
+
+## Signature
+
+```go
+package airflow
+
+func Dag(dagId string, spec DagSpec) *DagRef
+
+func (d *DagRef) Task(fn any, opts ...TaskOption) *TaskRef
+func (d *DagRef) TaskGroup(groupId string, opts ...TaskGroupOption)
*TaskGroupRef
+
+func (g *TaskGroupRef) Task(fn any, opts ...TaskOption) *TaskRef
+func (g *TaskGroupRef) TaskGroup(groupId string, opts ...TaskGroupOption)
*TaskGroupRef
+
+// DagSpec and TaskSpec are generated into this package from
+// airflow-core/src/airflow/serialization/schema.json and committed.
+type DagSpec struct {
+ Schedule string
+ StartDate time.Time
+ Catchup bool
+ Tags []string
+ // ...
+}
+
+// TaskSpec implements TaskOption, so it travels in the same variadic as
Inputs.
+type TaskSpec struct {
+ TaskId string // defaults to the Go function name
+ Retries int
+ TriggerRule TriggerRule
+ // ...
+}
+
+// TaskOption is sealed: its only method is unexported, so a task takes
SDK-defined options and
+// nothing else. TaskSpec and the value Inputs returns both implement it.
+type TaskOption interface{ applyTask(*taskConfig) }
+
+func Inputs(refs ...*TaskRef) TaskOption
+
+// Node is what an edge connects. *TaskRef and *TaskGroupRef implement it,
sealed the same way.
+// It is the Go counterpart of Python's DAGNode / DependencyMixin.
+// Before and After return their argument set as one Node, which is what makes
a chain work.
+type Node interface {
+ Before(nodes ...Node) Node
+ After(nodes ...Node) Node
+ node()
+}
+
+// Label carries an edge label into Before or After. The Node it returns
stands for node itself.
+func Label(node Node, text string) Node
+
+// A user-facing enum is a named string type with its type in each constant
name.
+type TriggerRule string
+
+const (
+ TriggerRuleAllSuccess TriggerRule = "all_success"
+ TriggerRuleAllDone TriggerRule = "all_done"
+ TriggerRuleOneFailed TriggerRule = "one_failed"
+ // ... one constant per Python trigger rule
+)
+```
+
+## Consequences
+
+- **A cycle check becomes necessary.** `b := dag.Task(B, airflow.Inputs(a));
b.Before(a)` is a genuine cycle in accepted
+ syntax. Either the build-time or the Dag-processing time should reject this.
+- **A count or type mismatch panics at registration**, not at run time,
because each `*TaskRef` carries its recorded output type.
+- **An edge verb returns what it pointed at, not its receiver.** That is what
makes `a.Before(b, c).Before(d)` mean `a >> [b, c] >> d`.
+ Returning the receiver would read like a chain and mean a second fan-out
from `a`.
+- **The specs generate into the `airflow` package, not a `gen` package beside
it.** An unexported method belongs to the package that declares it, so a
generated type living elsewhere could not implement the sealed `TaskOption`,
and a type alias cannot gain methods either.
+ Generating in place is what keeps both `airflow.TaskSpec` and the seal.
+- **The generated names need a mapping.** The core schema carries no `title`
fields, unlike the supervisor schema `models.gen.go` reads, so its `dag` and
`operator` definitions would generate as `Dag`, a name the constructor already
takes, and `Operator`, which is not the SDK's vocabulary.
+ Either the schema gains titles or the generate step keeps the map.
+- **The schema is the serialized shape, not the authoring shape.** It requires
`fileloc` and `tasks` on a Dag, and `task_type`, `_task_module`, `ui_color`,
`ui_fgcolor`, and `template_fields` on an operator, all of which the SDK fills
in, and it carries a serialized `timetable` object where an author writes a
schedule.
+ Generation needs an exclusion list and a hand-written field or two, the same
kind of rule
[ADR-0009](../../airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md)
states for provider operators.
+- **A data edge is labelled by redeclaring it.** Declaring an edge that
already exists is idempotent, so `extracted.Before(airflow.Label(transformed,
"rows"))` labels the edge `Inputs` created.
+- **Renaming a Go function renames the task.** The id is derived, and history,
clears, and the UI all key on task_id, so renaming a function whose task
carries no `TaskSpec` id is a Dag change.
+ `airflow.TaskSpec{TaskId: ...}` pins an id that has to outlive the
function's name, and it is also how a Dag gets snake_case ids, since nothing
transforms a Go name.
+
+## Alternatives
+
+- **Fetching upstream values at run time**, where a task reads an upstream
result inside its own body
+ (`result.Get(&out)`) and the graph falls out of the order the Go code
executes. Rejected: Airflow
+ materializes the whole graph at Dag-processing time and then invokes a
single task instance's
+ callable per run, so an edge existing only in execution order cannot be
parsed without running the
+ program to completion. `Inputs` keeps the typed outputs that style is
reached for.
+- **Labelling the call**, `loaded.Before(notify).Label("when empty")`.
Rejected: one call fans out, so a single label on the call cannot give
`a.Before(b, c)` a different label per edge.
+- **A positional task_id**, `dag.Task(taskId, fn, opts...)`. Rejected: It's
more straightforward and native for Go user to define a Task without defining
the task_id explicitly. They could still set the task_id other than the
function name in the TaskSpec.
+- **Separate `Dag` and `MixedLangDag` types.** Rejected: Python has one Dag
class, and the Mixed Lang case is not a Dag at all ([ADR
6](0006-mixed-lang-task-handler-interface.md)).
