Record the Go SDK's Mixed Lang Task and Native Dag task interfaces

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)).

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.

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).

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