Re: [PR] Add task loop authoring ADRs for the TypeScript, Java and Go SDKs [airflow]

[kaxil (via GitHub) via commits]

kaxil commented on code in PR #76365:
URL: https://github.com/apache/airflow/pull/76365#discussion_r4481395546


##########
ts-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,166 @@
+<!--
+ 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-0003: Task Loop — TypeScript Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the TypeScript spelling of them.
+
+## Decision
+
+1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and 
fills it as usual.
+2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, 
the shape
+   `dag.if(handler, inputs, spec)` already has 
([ADR-0002](0002-native-dag-interface.md)).
+3. **`getLoop()` supplies the loop context**, beside `getContext()` and 
`getClient()` (ADR-0002
+   decision 6). It is awaited, because reading the previous iteration is a 
round trip.
+4. **`index`, `maxIterations` and `previous` are fields**, and 
`getPrevious<T>()` is the typed view.
+   That is `client.getXCom`'s split between a value and the type the caller 
names.
+5. **A body task returns a `LoopTaskRef<T>`**, accepted where either 
`TaskRef<T>` or
+   `TaskRef<XComSequence<T>>` is wanted, so `tsc` checks whichever reading the 
consumer declares.
+6. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```ts
+import { Dag, getLoop, type LoopDecision, type XComSequence } from 
"apache-airflow-ts-sdk";
+
+type Metrics = { estimate: number; error: number };
+
+const dag = new Dag("ts_refine", { schedule: "@daily" });
+const refine = dag.taskGroup("refine");
+
+const improve = refine.task(async function improve(): Promise<number> {
+  const loop = await getLoop();
+  console.log(`iteration ${loop.index} of ${loop.maxIterations}`);
+  const estimate = loop.getPrevious<Metrics>()?.estimate ?? 1.0;
+  return (estimate + 2.0 / estimate) / 2.0;
+});
+
+const evaluate = refine.task(async function evaluate({ estimate }: { estimate: 
number }): Promise<Metrics> {
+  return { estimate, error: Math.abs(estimate * estimate - 2) };
+});
+
+async function accurateEnough({ result }: { result: Metrics }): 
Promise<LoopDecision> {
+  return { stop: result.error < 1e-6, reason: `error 
${result.error.toExponential(1)}` };
+}
+
+const report = dag.task(async function report({ history }: { history: 
XComSequence<Metrics> }) {
+  console.log(`settled after ${await history.length()} iterations`);
+});
+
+const estimate = improve();
+const evaluated = evaluate({ estimate });
+
+const refinement = refine.loop(accurateEnough, { result: evaluated }, { 
maxIterations: 10 });
+
+report({ history: evaluated }); // every iteration's metrics, through the gate
+refinement.before(cleanup()); // order only: the whole loop, not one iteration
+```
+
+`evaluated` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 5 is what makes both type-check.
+
+A loop that only counts needs no handler:
+
+```ts
+const poll = dag.taskGroup("poll");
+poll.task(checkOnce)();
+poll.loop({ maxIterations: 3 });

Review Comment:
   What task id does the gate get here? The serialized `loop` requires 
`gate_task_id` even with `has_until: false`, and these ADRs derive the gate id 
from the until handler's name, which a fixed-count loop doesn't have. Python 
falls back to `__loop_gate`. Can all three ADRs say the same?



##########
go-sdk/adr/0009-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,211 @@
+<!--
+ 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.
+ -->
+
+# 9. Task loop authoring interface
+
+Date: 2026-11-03
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Go spelling of them.
+
+## Decision
+
+1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and 
fills it as usual
+   ([ADR 8](0008-native-dag-interface.md)).
+2. **`maxIterations` is positional and `Until` is an option.** Every loop has 
a bound; only some add a
+   callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as 
`TaskOption` is.
+3. **The gate is a task handler wired with `airflow.Inputs`.** Registration 
checks its parameter
+   against the terminal task's recorded output type, as it already does for 
`Inputs`.
+4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, 
`TaskInstance` and
+   `DagRun`. It returns a value and an error, because reading the previous 
iteration is a round trip.
+5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as 
`DagRun`'s are, and
+   `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split 
between the value it returns
+   and the `out` it decodes into.
+6. **A body ref is read from outside the loop unchanged.** The consuming 
handler declares `T` for one
+   iteration or `airflow.XComSequence[T]` for every iteration, checked at 
registration.
+7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a 
body ref to a task
+   outside is rejected when it is declared.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```go
+type Metrics struct {
+    Estimate float64 `json:"estimate"`
+    Error    float64 `json:"error"`
+}
+
+func improve(actx airflow.Context) (float64, error) {
+    loop, err := actx.Loop()
+    if err != nil {
+        return 0, err
+    }
+    actx.Logger().Info("refining", "iteration", loop.Index, "of", 
loop.MaxIterations)
+
+    estimate := 1.0
+    if loop.Previous != nil {
+        var previous Metrics
+        if err := loop.UnmarshalPrevious(&previous); err != nil {
+            return 0, err
+        }
+        estimate = previous.Estimate
+    }
+    return (estimate + 2.0/estimate) / 2.0, nil
+}
+
+func evaluate(actx airflow.Context, estimate float64) (Metrics, error) {
+    return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 
2)}, nil
+}
+
+func accurateEnough(actx airflow.Context, result Metrics) 
(airflow.LoopDecision, error) {
+    return airflow.LoopDecision{
+        Stop:   result.Error < 1e-6,
+        Reason: fmt.Sprintf("error %.2g", result.Error),
+    }, nil
+}
+
+func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error 
{
+    n, err := history.Len()
+    if err != nil {
+        return err
+    }
+    actx.Logger().Info("settled", "iterations", n)
+    return nil
+}
+```
+
+```go
+dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"})
+refine := dag.TaskGroup("refine")
+
+improved := refine.Task(improve)
+evaluated := refine.Task(evaluate, airflow.Inputs(improved))
+
+refinement := refine.Loop(10, airflow.Until(accurateEnough, 
airflow.Inputs(evaluated)))
+
+dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, 
through the gate
+refinement.Before(dag.Task(cleanup))        // order only: the whole loop, not 
one iteration
+```
+
+`evaluated` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an 
`airflow.XComSequence[Metrics]`, which is what
+`report` takes. Decision 6 is what makes registration accept both.
+
+A loop that only counts needs no callable:
+
+```go
+poll := dag.TaskGroup("poll")
+poll.Task(checkOnce)
+poll.Loop(3)
+```
+
+## Signature
+
+```go
+package airflow
+
+// Loop turns the group into a task loop. maxIterations must be positive.
+func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef
+
+// Until supplies the gate's callable and the gate's own task options, 
including the Inputs that
+// name the loop's terminal task. fn takes airflow.Context first and the wired 
inputs after, as
+// every handler does, and returns either shape:
+//
+//    func(actx airflow.Context, args ...) (LoopDecision, error)
+//    func(actx airflow.Context, args ...) (bool, error)
+func Until(fn any, opts ...TaskOption) LoopOption
+
+// LoopOption is sealed: its only method is unexported, so a loop takes 
SDK-defined options and
+// nothing else.
+type LoopOption interface{ applyLoop(*loopConfig) error }
+
+// LoopDecision is what an until callable returns. A bool return is a decision 
with no reason.
+type LoopDecision struct {
+    Stop   bool
+    Reason string
+}
+
+// LoopRef is a Node, so Before and After carry edges for the whole loop.
+type LoopRef struct{ /* unexported */ }
+
+// Loop reads the previous iteration's result once and caches it. It panics 
outside a loop body or
+// gate, as the other accessors do on a zero Context, and returns an error 
when the read fails.
+func (c Context) Loop() (Loop, error)
+
+// Loop is handed back by value, as DagRun is: the fields are the runtime's 
answer, not state for
+// a handler to write.
+type Loop struct {
+    Index         int
+    MaxIterations int
+
+    // Previous is the previous iteration's terminal result as it arrived: nil 
at Index 1, and an
+    // XComSequence when the terminal task is mapped.
+    Previous any
+}
+
+// UnmarshalPrevious binds Previous to a DTO, the way XComClient.GetXCom 
decodes into its out
+// parameter. It leaves out untouched when there is no previous iteration.
+func (l Loop) UnmarshalPrevious(out any) error
+
+// XComSequence is the lazy sequence of ADR-0016: every iteration of a loop 
task read from outside
+// the loop, or every instance of a mapped task. It is generic because it is 
reached as a declared
+// parameter type, which the SDK reads by reflection as it does any other 
parameter.
+type XComSequence[T any] interface {

Review Comment:
   How does the runtime build a value of type `XComSequence[Metrics]`? `Loop()` 
has no type parameter, handlers register as `fn any`, and `reflect` can't 
instantiate a generic type. So whatever the SDK constructs (say a `seq[any]`) 
has `At(int) (any, error)`, and the 
`loop.Previous.(airflow.XComSequence[Metrics])` assertion on L199 panics. 
Binding `report`'s `history airflow.XComSequence[Metrics]` parameter hits the 
same wall. A non-generic `XComSequence` with `At(offset int, out any) error`, 
matching `UnmarshalPrevious`, would be constructible. Would that work?



##########
ts-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,166 @@
+<!--
+ 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-0003: Task Loop — TypeScript Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the TypeScript spelling of them.
+
+## Decision
+
+1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and 
fills it as usual.
+2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, 
the shape
+   `dag.if(handler, inputs, spec)` already has 
([ADR-0002](0002-native-dag-interface.md)).
+3. **`getLoop()` supplies the loop context**, beside `getContext()` and 
`getClient()` (ADR-0002
+   decision 6). It is awaited, because reading the previous iteration is a 
round trip.
+4. **`index`, `maxIterations` and `previous` are fields**, and 
`getPrevious<T>()` is the typed view.
+   That is `client.getXCom`'s split between a value and the type the caller 
names.
+5. **A body task returns a `LoopTaskRef<T>`**, accepted where either 
`TaskRef<T>` or
+   `TaskRef<XComSequence<T>>` is wanted, so `tsc` checks whichever reading the 
consumer declares.
+6. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```ts
+import { Dag, getLoop, type LoopDecision, type XComSequence } from 
"apache-airflow-ts-sdk";
+
+type Metrics = { estimate: number; error: number };
+
+const dag = new Dag("ts_refine", { schedule: "@daily" });
+const refine = dag.taskGroup("refine");
+
+const improve = refine.task(async function improve(): Promise<number> {
+  const loop = await getLoop();
+  console.log(`iteration ${loop.index} of ${loop.maxIterations}`);
+  const estimate = loop.getPrevious<Metrics>()?.estimate ?? 1.0;
+  return (estimate + 2.0 / estimate) / 2.0;
+});
+
+const evaluate = refine.task(async function evaluate({ estimate }: { estimate: 
number }): Promise<Metrics> {
+  return { estimate, error: Math.abs(estimate * estimate - 2) };
+});
+
+async function accurateEnough({ result }: { result: Metrics }): 
Promise<LoopDecision> {
+  return { stop: result.error < 1e-6, reason: `error 
${result.error.toExponential(1)}` };
+}
+
+const report = dag.task(async function report({ history }: { history: 
XComSequence<Metrics> }) {
+  console.log(`settled after ${await history.length()} iterations`);
+});
+
+const estimate = improve();
+const evaluated = evaluate({ estimate });
+
+const refinement = refine.loop(accurateEnough, { result: evaluated }, { 
maxIterations: 10 });
+
+report({ history: evaluated }); // every iteration's metrics, through the gate
+refinement.before(cleanup()); // order only: the whole loop, not one iteration
+```
+
+`evaluated` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 5 is what makes both type-check.
+
+A loop that only counts needs no handler:
+
+```ts
+const poll = dag.taskGroup("poll");
+poll.task(checkOnce)();
+poll.loop({ maxIterations: 3 });
+```
+
+## Signature
+
+```ts
+/** Options of a loop, the trailing argument of `loop()`. */
+interface LoopOptions extends TaskSpec {
+  maxIterations: number;
+}
+
+/** What `until` returns. A bare `boolean` is `{ stop }`. */
+interface LoopDecision {
+  stop: boolean;
+  reason?: string;
+}
+
+/**
+ * The gate's handler. `A` is its wired inputs, one key per argument, exactly 
as a task handler
+ * takes its own. Nothing of the SDK's is in `A`; the loop context comes from 
`getLoop()`.
+ */
+type UntilCallable<A> = (args: A) => Promise<LoopDecision | boolean>;
+
+interface TaskGroupRef {
+  loop(options: LoopOptions): LoopRef;
+  loop<A>(until: UntilCallable<A>, inputs: A, options: LoopOptions): LoopRef;

Review Comment:
   With `inputs: A`, `A` has to be both `{ result: Metrics }` (what 
`accurateEnough` takes) and `{ result: evaluated }` (a ref), so the example on 
L81 doesn't type-check. `dag.if` avoids this with `...args: 
DeciderArgs<NoInfer<TArgs>>`, which maps the inputs through `TaskInputs`. 
Should `loop` take the same? `LoopRef` and `LoopTaskRef` from decisions 5 and 6 
are also missing from this block.



##########
ts-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,166 @@
+<!--
+ 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-0003: Task Loop — TypeScript Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the TypeScript spelling of them.
+
+## Decision
+
+1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and 
fills it as usual.
+2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, 
the shape
+   `dag.if(handler, inputs, spec)` already has 
([ADR-0002](0002-native-dag-interface.md)).
+3. **`getLoop()` supplies the loop context**, beside `getContext()` and 
`getClient()` (ADR-0002
+   decision 6). It is awaited, because reading the previous iteration is a 
round trip.
+4. **`index`, `maxIterations` and `previous` are fields**, and 
`getPrevious<T>()` is the typed view.
+   That is `client.getXCom`'s split between a value and the type the caller 
names.
+5. **A body task returns a `LoopTaskRef<T>`**, accepted where either 
`TaskRef<T>` or
+   `TaskRef<XComSequence<T>>` is wanted, so `tsc` checks whichever reading the 
consumer declares.
+6. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```ts
+import { Dag, getLoop, type LoopDecision, type XComSequence } from 
"apache-airflow-ts-sdk";
+
+type Metrics = { estimate: number; error: number };
+
+const dag = new Dag("ts_refine", { schedule: "@daily" });
+const refine = dag.taskGroup("refine");
+
+const improve = refine.task(async function improve(): Promise<number> {
+  const loop = await getLoop();
+  console.log(`iteration ${loop.index} of ${loop.maxIterations}`);
+  const estimate = loop.getPrevious<Metrics>()?.estimate ?? 1.0;
+  return (estimate + 2.0 / estimate) / 2.0;
+});
+
+const evaluate = refine.task(async function evaluate({ estimate }: { estimate: 
number }): Promise<Metrics> {
+  return { estimate, error: Math.abs(estimate * estimate - 2) };
+});
+
+async function accurateEnough({ result }: { result: Metrics }): 
Promise<LoopDecision> {
+  return { stop: result.error < 1e-6, reason: `error 
${result.error.toExponential(1)}` };
+}
+
+const report = dag.task(async function report({ history }: { history: 
XComSequence<Metrics> }) {
+  console.log(`settled after ${await history.length()} iterations`);
+});
+
+const estimate = improve();
+const evaluated = evaluate({ estimate });
+
+const refinement = refine.loop(accurateEnough, { result: evaluated }, { 
maxIterations: 10 });
+
+report({ history: evaluated }); // every iteration's metrics, through the gate
+refinement.before(cleanup()); // order only: the whole loop, not one iteration

Review Comment:
   `cleanup` isn't declared anywhere in this example (same in the Go one, L112).



##########
go-sdk/adr/0009-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,211 @@
+<!--
+ 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.
+ -->
+
+# 9. Task loop authoring interface
+
+Date: 2026-11-03
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Go spelling of them.
+
+## Decision
+
+1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and 
fills it as usual
+   ([ADR 8](0008-native-dag-interface.md)).
+2. **`maxIterations` is positional and `Until` is an option.** Every loop has 
a bound; only some add a
+   callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as 
`TaskOption` is.
+3. **The gate is a task handler wired with `airflow.Inputs`.** Registration 
checks its parameter
+   against the terminal task's recorded output type, as it already does for 
`Inputs`.
+4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, 
`TaskInstance` and
+   `DagRun`. It returns a value and an error, because reading the previous 
iteration is a round trip.
+5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as 
`DagRun`'s are, and
+   `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split 
between the value it returns
+   and the `out` it decodes into.
+6. **A body ref is read from outside the loop unchanged.** The consuming 
handler declares `T` for one
+   iteration or `airflow.XComSequence[T]` for every iteration, checked at 
registration.
+7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a 
body ref to a task
+   outside is rejected when it is declared.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```go
+type Metrics struct {
+    Estimate float64 `json:"estimate"`
+    Error    float64 `json:"error"`
+}
+
+func improve(actx airflow.Context) (float64, error) {
+    loop, err := actx.Loop()
+    if err != nil {
+        return 0, err
+    }
+    actx.Logger().Info("refining", "iteration", loop.Index, "of", 
loop.MaxIterations)
+
+    estimate := 1.0
+    if loop.Previous != nil {
+        var previous Metrics
+        if err := loop.UnmarshalPrevious(&previous); err != nil {
+            return 0, err
+        }
+        estimate = previous.Estimate
+    }
+    return (estimate + 2.0/estimate) / 2.0, nil
+}
+
+func evaluate(actx airflow.Context, estimate float64) (Metrics, error) {
+    return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 
2)}, nil
+}
+
+func accurateEnough(actx airflow.Context, result Metrics) 
(airflow.LoopDecision, error) {
+    return airflow.LoopDecision{
+        Stop:   result.Error < 1e-6,
+        Reason: fmt.Sprintf("error %.2g", result.Error),
+    }, nil
+}
+
+func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error 
{
+    n, err := history.Len()
+    if err != nil {
+        return err
+    }
+    actx.Logger().Info("settled", "iterations", n)
+    return nil
+}
+```
+
+```go
+dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"})
+refine := dag.TaskGroup("refine")
+
+improved := refine.Task(improve)
+evaluated := refine.Task(evaluate, airflow.Inputs(improved))
+
+refinement := refine.Loop(10, airflow.Until(accurateEnough, 
airflow.Inputs(evaluated)))
+
+dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, 
through the gate
+refinement.Before(dag.Task(cleanup))        // order only: the whole loop, not 
one iteration
+```
+
+`evaluated` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an 
`airflow.XComSequence[Metrics]`, which is what
+`report` takes. Decision 6 is what makes registration accept both.
+
+A loop that only counts needs no callable:
+
+```go
+poll := dag.TaskGroup("poll")
+poll.Task(checkOnce)
+poll.Loop(3)
+```
+
+## Signature
+
+```go
+package airflow
+
+// Loop turns the group into a task loop. maxIterations must be positive.
+func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef
+
+// Until supplies the gate's callable and the gate's own task options, 
including the Inputs that

Review Comment:
   If the gate binds `improved` instead of `evaluated`, this signature still 
accepts it, and then nothing orders the real leaf before the gate. Python takes 
the terminal from `group.get_leaves()` and adds `terminal >> gate` itself, so 
the gate can read whatever else it needs. Can Go (and TS `loop(until, inputs, 
...)`) do the same, instead of naming the terminal through the gate's `Inputs`?



##########
go-sdk/adr/0009-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,211 @@
+<!--
+ 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.
+ -->
+
+# 9. Task loop authoring interface
+
+Date: 2026-11-03
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Go spelling of them.
+
+## Decision
+
+1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and 
fills it as usual
+   ([ADR 8](0008-native-dag-interface.md)).
+2. **`maxIterations` is positional and `Until` is an option.** Every loop has 
a bound; only some add a
+   callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as 
`TaskOption` is.
+3. **The gate is a task handler wired with `airflow.Inputs`.** Registration 
checks its parameter
+   against the terminal task's recorded output type, as it already does for 
`Inputs`.
+4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, 
`TaskInstance` and
+   `DagRun`. It returns a value and an error, because reading the previous 
iteration is a round trip.
+5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as 
`DagRun`'s are, and
+   `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split 
between the value it returns
+   and the `out` it decodes into.
+6. **A body ref is read from outside the loop unchanged.** The consuming 
handler declares `T` for one
+   iteration or `airflow.XComSequence[T]` for every iteration, checked at 
registration.
+7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a 
body ref to a task
+   outside is rejected when it is declared.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```go
+type Metrics struct {
+    Estimate float64 `json:"estimate"`
+    Error    float64 `json:"error"`
+}
+
+func improve(actx airflow.Context) (float64, error) {
+    loop, err := actx.Loop()
+    if err != nil {
+        return 0, err
+    }
+    actx.Logger().Info("refining", "iteration", loop.Index, "of", 
loop.MaxIterations)
+
+    estimate := 1.0
+    if loop.Previous != nil {
+        var previous Metrics
+        if err := loop.UnmarshalPrevious(&previous); err != nil {
+            return 0, err
+        }
+        estimate = previous.Estimate
+    }
+    return (estimate + 2.0/estimate) / 2.0, nil
+}
+
+func evaluate(actx airflow.Context, estimate float64) (Metrics, error) {
+    return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 
2)}, nil
+}
+
+func accurateEnough(actx airflow.Context, result Metrics) 
(airflow.LoopDecision, error) {
+    return airflow.LoopDecision{
+        Stop:   result.Error < 1e-6,
+        Reason: fmt.Sprintf("error %.2g", result.Error),
+    }, nil
+}
+
+func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error 
{
+    n, err := history.Len()
+    if err != nil {
+        return err
+    }
+    actx.Logger().Info("settled", "iterations", n)
+    return nil
+}
+```
+
+```go
+dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"})
+refine := dag.TaskGroup("refine")
+
+improved := refine.Task(improve)
+evaluated := refine.Task(evaluate, airflow.Inputs(improved))
+
+refinement := refine.Loop(10, airflow.Until(accurateEnough, 
airflow.Inputs(evaluated)))
+
+dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, 
through the gate
+refinement.Before(dag.Task(cleanup))        // order only: the whole loop, not 
one iteration
+```
+
+`evaluated` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an 
`airflow.XComSequence[Metrics]`, which is what
+`report` takes. Decision 6 is what makes registration accept both.
+
+A loop that only counts needs no callable:
+
+```go
+poll := dag.TaskGroup("poll")
+poll.Task(checkOnce)
+poll.Loop(3)
+```
+
+## Signature
+
+```go
+package airflow
+
+// Loop turns the group into a task loop. maxIterations must be positive.
+func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef
+
+// Until supplies the gate's callable and the gate's own task options, 
including the Inputs that
+// name the loop's terminal task. fn takes airflow.Context first and the wired 
inputs after, as
+// every handler does, and returns either shape:
+//
+//    func(actx airflow.Context, args ...) (LoopDecision, error)
+//    func(actx airflow.Context, args ...) (bool, error)
+func Until(fn any, opts ...TaskOption) LoopOption
+
+// LoopOption is sealed: its only method is unexported, so a loop takes 
SDK-defined options and
+// nothing else.
+type LoopOption interface{ applyLoop(*loopConfig) error }
+
+// LoopDecision is what an until callable returns. A bool return is a decision 
with no reason.
+type LoopDecision struct {
+    Stop   bool
+    Reason string
+}
+
+// LoopRef is a Node, so Before and After carry edges for the whole loop.
+type LoopRef struct{ /* unexported */ }
+
+// Loop reads the previous iteration's result once and caches it. It panics 
outside a loop body or

Review Comment:
   The other accessors don't panic. `Logger()`, `Client()`, `TaskInstance()` 
and `DagRun()` return the stored field, and only `NewContext` panics, on nil 
inputs. Since `Loop()` already returns an error, could it return something like 
`airflow.ErrNotInLoop` instead?



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and
+   `context.runId` already are. `previous` is `null` at iteration 1.
+6. **`Loop<T>` is generic over the terminal task's result**, so 
`Loop<Metrics>` gives a typed
+   `previous` with nothing to cast. The interface surface has no declaration 
to read, so
+   `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names 
the type at the read,
+   as `Client.getXCom` does.
+7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor 
generates every wiring
+   parameter of type `XComSequence<E>` as an overloaded pair, one taking 
`Arg<XComSequence<E>>` and
+   one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the 
consumer declares.
+8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```java
+@Airflow.Dag(id = "java_refine", schedule = "@daily")
+public class RefineEstimate {
+  private static final System.Logger log = 
System.getLogger(RefineEstimate.class.getName());
+
+  record Metrics(double estimate, double error) {}
+
+  // A loop body: a task group with a bound. "refine.improve", 
"refine.evaluate",
+  // "refine.accurate_enough".
+  @Airflow.Loop(id = "refine", maxIterations = 10)
+  static class Refine {
+
+    @Airflow.Task(id = "improve")
+    public double improve(Loop<Metrics> loop) {
+      log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations);
+      double estimate = loop.previous == null ? 1.0 : loop.previous.estimate();
+      return (estimate + 2.0 / estimate) / 2.0;
+    }
+
+    @Airflow.Task(id = "evaluate")
+    public Metrics evaluate(double estimate) {
+      return new Metrics(estimate, Math.abs(estimate * estimate - 2));
+    }
+
+    // The gate. Its parameter is the loop's terminal task, bound in depends().
+    @Airflow.Until(id = "accurate_enough")
+    public LoopDecision accurateEnough(Metrics result) {
+      return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+    }
+  }
+
+  @Airflow.Task(id = "report")
+  public void report(XComSequence<Metrics> history) {
+    log.log(INFO, "settled after {0} iterations", history.length());
+  }
+
+  @Airflow.Task(id = "cleanup")
+  public void cleanup() {}
+
+  @Airflow.Deps
+  static class Wiring implements RefineEstimateDeps {
+    void depends() {
+      var estimate = refine().improve();
+      var metrics = refine().evaluate(estimate);
+      refine().accurateEnough(metrics); // the gate, and the terminal it reads
+      report(metrics);                  // every iteration's metrics, through 
the gate
+      refine().before(cleanup());       // order only: the whole loop, not one 
iteration
+    }
+  }
+}
+```
+
+`metrics` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 7 is what makes both compile:
+
+```java
+// Generated by the Airflow Java SDK annotation processor. Do not edit.
+interface RefineEstimateDeps extends Deps {
+  default Refine refine() { return Flow.loop("refine"); }
+
+  interface Refine extends LoopRef {
+    default LoopTaskRef<Double> improve() { return 
Flow.node("refine.improve"); }
+    default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return 
Flow.call("refine.evaluate", estimate); }
+    default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return 
Flow.call("refine.accurate_enough", result); }
+  }
+
+  default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> cleanup() { return Flow.node("cleanup"); }
+}
+```
+
+A loop that only counts needs no gate method at all:
+
+```java
+@Airflow.Loop(id = "poll", maxIterations = 3)
+static class Poll {
+  @Airflow.Task(id = "check_once")
+  public void checkOnce() {}
+}
+```
+
+### Interface surface
+
+```java
+Dag dag = new Dag("java_refine").config("schedule", "@daily");
+
+LoopRef refine = dag.loop("refine", 10);
+TaskRef improve = refine.task("improve", Improve.class);
+TaskRef evaluate = refine.task("evaluate", Evaluate.class);
+improve.before(evaluate);
+refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's 
single leaf
+
+TaskRef report = dag.task("report", Report.class);
+evaluate.before(report); // data edge out of the body, through the gate
+refine.before(dag.task("cleanup", Cleanup.class));
+```
+
+```java
+public static class Improve implements Task {
+  @Override
+  public void execute(Context context, Client client) {
+    Metrics previous = context.loop().getPrevious(Metrics.class);
+    double estimate = previous == null ? 1.0 : previous.estimate();
+    client.setXCom((estimate + 2.0 / estimate) / 2.0);
+  }
+}
+
+public static class AccurateEnough implements UntilTask {
+  @Override
+  public LoopDecision decide(Context context, Client client) {
+    Metrics result = context.loop().getResult(Metrics.class);

Review Comment:
   ADR-0017 rule 3 says the gate reads the terminal as wired data, not off the 
loop context. Could the interface-surface gate use `client.getXCom(taskId = 
"refine.evaluate")` like any other interface task, since core already pins that 
read to the gate's iteration? That would also drop `Loop.result` and 
`getResult`, which are null everywhere except here. If the deviation is 
intentional, it belongs in Consequences.



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and
+   `context.runId` already are. `previous` is `null` at iteration 1.
+6. **`Loop<T>` is generic over the terminal task's result**, so 
`Loop<Metrics>` gives a typed
+   `previous` with nothing to cast. The interface surface has no declaration 
to read, so
+   `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names 
the type at the read,
+   as `Client.getXCom` does.
+7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor 
generates every wiring
+   parameter of type `XComSequence<E>` as an overloaded pair, one taking 
`Arg<XComSequence<E>>` and
+   one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the 
consumer declares.
+8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```java
+@Airflow.Dag(id = "java_refine", schedule = "@daily")
+public class RefineEstimate {
+  private static final System.Logger log = 
System.getLogger(RefineEstimate.class.getName());
+
+  record Metrics(double estimate, double error) {}
+
+  // A loop body: a task group with a bound. "refine.improve", 
"refine.evaluate",
+  // "refine.accurate_enough".
+  @Airflow.Loop(id = "refine", maxIterations = 10)
+  static class Refine {
+
+    @Airflow.Task(id = "improve")
+    public double improve(Loop<Metrics> loop) {
+      log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations);
+      double estimate = loop.previous == null ? 1.0 : loop.previous.estimate();
+      return (estimate + 2.0 / estimate) / 2.0;
+    }
+
+    @Airflow.Task(id = "evaluate")
+    public Metrics evaluate(double estimate) {
+      return new Metrics(estimate, Math.abs(estimate * estimate - 2));
+    }
+
+    // The gate. Its parameter is the loop's terminal task, bound in depends().
+    @Airflow.Until(id = "accurate_enough")
+    public LoopDecision accurateEnough(Metrics result) {
+      return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+    }
+  }
+
+  @Airflow.Task(id = "report")
+  public void report(XComSequence<Metrics> history) {
+    log.log(INFO, "settled after {0} iterations", history.length());
+  }
+
+  @Airflow.Task(id = "cleanup")
+  public void cleanup() {}
+
+  @Airflow.Deps
+  static class Wiring implements RefineEstimateDeps {
+    void depends() {
+      var estimate = refine().improve();
+      var metrics = refine().evaluate(estimate);
+      refine().accurateEnough(metrics); // the gate, and the terminal it reads
+      report(metrics);                  // every iteration's metrics, through 
the gate
+      refine().before(cleanup());       // order only: the whole loop, not one 
iteration
+    }
+  }
+}
+```
+
+`metrics` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 7 is what makes both compile:
+
+```java
+// Generated by the Airflow Java SDK annotation processor. Do not edit.
+interface RefineEstimateDeps extends Deps {
+  default Refine refine() { return Flow.loop("refine"); }
+
+  interface Refine extends LoopRef {
+    default LoopTaskRef<Double> improve() { return 
Flow.node("refine.improve"); }
+    default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return 
Flow.call("refine.evaluate", estimate); }
+    default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return 
Flow.call("refine.accurate_enough", result); }
+  }
+
+  default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> cleanup() { return Flow.node("cleanup"); }
+}
+```
+
+A loop that only counts needs no gate method at all:
+
+```java
+@Airflow.Loop(id = "poll", maxIterations = 3)
+static class Poll {
+  @Airflow.Task(id = "check_once")
+  public void checkOnce() {}
+}
+```
+
+### Interface surface
+
+```java
+Dag dag = new Dag("java_refine").config("schedule", "@daily");
+
+LoopRef refine = dag.loop("refine", 10);
+TaskRef improve = refine.task("improve", Improve.class);
+TaskRef evaluate = refine.task("evaluate", Evaluate.class);
+improve.before(evaluate);
+refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's 
single leaf
+
+TaskRef report = dag.task("report", Report.class);
+evaluate.before(report); // data edge out of the body, through the gate
+refine.before(dag.task("cleanup", Cleanup.class));
+```
+
+```java
+public static class Improve implements Task {
+  @Override
+  public void execute(Context context, Client client) {
+    Metrics previous = context.loop().getPrevious(Metrics.class);
+    double estimate = previous == null ? 1.0 : previous.estimate();
+    client.setXCom((estimate + 2.0 / estimate) / 2.0);
+  }
+}
+
+public static class AccurateEnough implements UntilTask {
+  @Override
+  public LoopDecision decide(Context context, Client client) {
+    Metrics result = context.loop().getResult(Metrics.class);
+    return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+  }
+}
+```
+
+`UntilTask` joins `ConditionTask` and `SwitchTask` as a task kind with its own 
verb, so the gate is
+recognized by the interface it implements rather than by a naming rule.
+
+## Signature
+
+```java
+// Annotation surface. @Airflow.Until carries @Airflow.Task's own attributes 
as well.
+@interface Airflow.Loop { String id(); int maxIterations(); }
+@interface Airflow.Until { String id() default ""; }
+
+// The gate's method under @Airflow.Until. Its injected arguments come first 
and its wired data
+// after, as any task method's do, and it returns either shape; a boolean is a 
decision with no
+// reason.
+//
+//   LoopDecision accurateEnough(Metrics result)
+//   LoopDecision accurateEnough(Loop<Metrics> loop, Metrics result)
+//   boolean accurateEnough(Metrics result)
+
+// Injected into a loop body task or its gate, beside Client and Context. On 
the interface
+// surface it is context.loop(). It throws outside a loop.
+// T is the terminal task's result type. Declare it (Loop<Metrics>) and 
previous is typed;
+// declare none (Loop<Object>, which context.loop() returns) and name the type 
at the read.
+public final class Loop<T> {
+  public final int index;
+
+  public final int maxIterations;
+
+  /** The previous iteration's terminal result, or null at iteration 1. */
+  public final T previous;
+
+  /**
+   * This iteration's terminal result, for a gate on the interface surface, 
where there is no
+   * ref to bind it from. Null in a body task. The annotation surface takes it 
as a parameter.
+   */
+  public final T result;
+
+  /** previous bound to a DTO, as Client.getXCom's caller names the type it 
wants. */
+  public <R> R getPrevious(Class<R> type);
+
+  public <R> R getResult(Class<R> type);
+}
+
+/** The lazy sequence of ADR-0016, reached as a declared parameter type or as 
Loop's T. */
+public interface XComSequence<E> {
+  int length();
+
+  E get(int offset);
+}
+
+/** What an until method returns. A boolean return is a decision with no 
reason. */
+public record LoopDecision(boolean stop, String reason) {}
+
+/** Interface surface. */
+public interface UntilTask {
+  LoopDecision decide(Context context, Client client);
+}
+
+public interface LoopRef extends TaskGroupRef, Flow {

Review Comment:
   `TaskGroupRef` and `TaskRef` are final Kotlin classes on main (`class 
TaskGroupRef internal constructor`, `class TaskRef<T> internal constructor(...) 
: Arg<T>()`), so neither `LoopRef extends TaskGroupRef` nor `LoopTaskRef<T> 
extends TaskRef<T>` (L246) compiles. Decision 7's overload pair depends on 
`LoopTaskRef` being a subtype of `TaskRef`, so this needs a change to shipped 
public types. Can the ADR say which?



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and
+   `context.runId` already are. `previous` is `null` at iteration 1.
+6. **`Loop<T>` is generic over the terminal task's result**, so 
`Loop<Metrics>` gives a typed
+   `previous` with nothing to cast. The interface surface has no declaration 
to read, so
+   `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names 
the type at the read,
+   as `Client.getXCom` does.
+7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor 
generates every wiring
+   parameter of type `XComSequence<E>` as an overloaded pair, one taking 
`Arg<XComSequence<E>>` and
+   one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the 
consumer declares.
+8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```java
+@Airflow.Dag(id = "java_refine", schedule = "@daily")
+public class RefineEstimate {
+  private static final System.Logger log = 
System.getLogger(RefineEstimate.class.getName());
+
+  record Metrics(double estimate, double error) {}
+
+  // A loop body: a task group with a bound. "refine.improve", 
"refine.evaluate",
+  // "refine.accurate_enough".
+  @Airflow.Loop(id = "refine", maxIterations = 10)
+  static class Refine {
+
+    @Airflow.Task(id = "improve")
+    public double improve(Loop<Metrics> loop) {
+      log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations);
+      double estimate = loop.previous == null ? 1.0 : loop.previous.estimate();
+      return (estimate + 2.0 / estimate) / 2.0;
+    }
+
+    @Airflow.Task(id = "evaluate")
+    public Metrics evaluate(double estimate) {
+      return new Metrics(estimate, Math.abs(estimate * estimate - 2));
+    }
+
+    // The gate. Its parameter is the loop's terminal task, bound in depends().
+    @Airflow.Until(id = "accurate_enough")
+    public LoopDecision accurateEnough(Metrics result) {
+      return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+    }
+  }
+
+  @Airflow.Task(id = "report")
+  public void report(XComSequence<Metrics> history) {
+    log.log(INFO, "settled after {0} iterations", history.length());
+  }
+
+  @Airflow.Task(id = "cleanup")
+  public void cleanup() {}
+
+  @Airflow.Deps
+  static class Wiring implements RefineEstimateDeps {
+    void depends() {
+      var estimate = refine().improve();
+      var metrics = refine().evaluate(estimate);
+      refine().accurateEnough(metrics); // the gate, and the terminal it reads
+      report(metrics);                  // every iteration's metrics, through 
the gate
+      refine().before(cleanup());       // order only: the whole loop, not one 
iteration
+    }
+  }
+}
+```
+
+`metrics` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 7 is what makes both compile:
+
+```java
+// Generated by the Airflow Java SDK annotation processor. Do not edit.
+interface RefineEstimateDeps extends Deps {
+  default Refine refine() { return Flow.loop("refine"); }
+
+  interface Refine extends LoopRef {
+    default LoopTaskRef<Double> improve() { return 
Flow.node("refine.improve"); }
+    default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return 
Flow.call("refine.evaluate", estimate); }
+    default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return 
Flow.call("refine.accurate_enough", result); }
+  }
+
+  default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> cleanup() { return Flow.node("cleanup"); }
+}
+```
+
+A loop that only counts needs no gate method at all:
+
+```java
+@Airflow.Loop(id = "poll", maxIterations = 3)
+static class Poll {
+  @Airflow.Task(id = "check_once")
+  public void checkOnce() {}
+}
+```
+
+### Interface surface
+
+```java
+Dag dag = new Dag("java_refine").config("schedule", "@daily");
+
+LoopRef refine = dag.loop("refine", 10);
+TaskRef improve = refine.task("improve", Improve.class);
+TaskRef evaluate = refine.task("evaluate", Evaluate.class);
+improve.before(evaluate);
+refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's 
single leaf
+
+TaskRef report = dag.task("report", Report.class);
+evaluate.before(report); // data edge out of the body, through the gate
+refine.before(dag.task("cleanup", Cleanup.class));
+```
+
+```java
+public static class Improve implements Task {
+  @Override
+  public void execute(Context context, Client client) {
+    Metrics previous = context.loop().getPrevious(Metrics.class);
+    double estimate = previous == null ? 1.0 : previous.estimate();
+    client.setXCom((estimate + 2.0 / estimate) / 2.0);
+  }
+}
+
+public static class AccurateEnough implements UntilTask {
+  @Override
+  public LoopDecision decide(Context context, Client client) {
+    Metrics result = context.loop().getResult(Metrics.class);
+    return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+  }
+}
+```
+
+`UntilTask` joins `ConditionTask` and `SwitchTask` as a task kind with its own 
verb, so the gate is

Review Comment:
   `ConditionTask` and `SwitchTask` both extend `Task`, but `UntilTask` in the 
Signature block doesn't, and `decide` is already `ConditionTask`'s verb. Should 
this be `interface UntilTask extends Task` with a verb of its own?



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and
+   `context.runId` already are. `previous` is `null` at iteration 1.
+6. **`Loop<T>` is generic over the terminal task's result**, so 
`Loop<Metrics>` gives a typed
+   `previous` with nothing to cast. The interface surface has no declaration 
to read, so
+   `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names 
the type at the read,
+   as `Client.getXCom` does.
+7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor 
generates every wiring
+   parameter of type `XComSequence<E>` as an overloaded pair, one taking 
`Arg<XComSequence<E>>` and
+   one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the 
consumer declares.
+8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```java
+@Airflow.Dag(id = "java_refine", schedule = "@daily")
+public class RefineEstimate {
+  private static final System.Logger log = 
System.getLogger(RefineEstimate.class.getName());
+
+  record Metrics(double estimate, double error) {}
+
+  // A loop body: a task group with a bound. "refine.improve", 
"refine.evaluate",
+  // "refine.accurate_enough".
+  @Airflow.Loop(id = "refine", maxIterations = 10)
+  static class Refine {
+
+    @Airflow.Task(id = "improve")
+    public double improve(Loop<Metrics> loop) {
+      log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations);
+      double estimate = loop.previous == null ? 1.0 : loop.previous.estimate();
+      return (estimate + 2.0 / estimate) / 2.0;
+    }
+
+    @Airflow.Task(id = "evaluate")
+    public Metrics evaluate(double estimate) {
+      return new Metrics(estimate, Math.abs(estimate * estimate - 2));
+    }
+
+    // The gate. Its parameter is the loop's terminal task, bound in depends().
+    @Airflow.Until(id = "accurate_enough")
+    public LoopDecision accurateEnough(Metrics result) {
+      return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+    }
+  }
+
+  @Airflow.Task(id = "report")
+  public void report(XComSequence<Metrics> history) {
+    log.log(INFO, "settled after {0} iterations", history.length());
+  }
+
+  @Airflow.Task(id = "cleanup")
+  public void cleanup() {}
+
+  @Airflow.Deps
+  static class Wiring implements RefineEstimateDeps {
+    void depends() {
+      var estimate = refine().improve();
+      var metrics = refine().evaluate(estimate);
+      refine().accurateEnough(metrics); // the gate, and the terminal it reads
+      report(metrics);                  // every iteration's metrics, through 
the gate
+      refine().before(cleanup());       // order only: the whole loop, not one 
iteration
+    }
+  }
+}
+```
+
+`metrics` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 7 is what makes both compile:
+
+```java
+// Generated by the Airflow Java SDK annotation processor. Do not edit.
+interface RefineEstimateDeps extends Deps {
+  default Refine refine() { return Flow.loop("refine"); }
+
+  interface Refine extends LoopRef {
+    default LoopTaskRef<Double> improve() { return 
Flow.node("refine.improve"); }
+    default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return 
Flow.call("refine.evaluate", estimate); }
+    default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return 
Flow.call("refine.accurate_enough", result); }
+  }
+
+  default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> cleanup() { return Flow.node("cleanup"); }
+}
+```
+
+A loop that only counts needs no gate method at all:
+
+```java
+@Airflow.Loop(id = "poll", maxIterations = 3)
+static class Poll {
+  @Airflow.Task(id = "check_once")
+  public void checkOnce() {}
+}
+```
+
+### Interface surface
+
+```java
+Dag dag = new Dag("java_refine").config("schedule", "@daily");
+
+LoopRef refine = dag.loop("refine", 10);
+TaskRef improve = refine.task("improve", Improve.class);
+TaskRef evaluate = refine.task("evaluate", Evaluate.class);
+improve.before(evaluate);
+refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's 
single leaf
+
+TaskRef report = dag.task("report", Report.class);
+evaluate.before(report); // data edge out of the body, through the gate
+refine.before(dag.task("cleanup", Cleanup.class));
+```
+
+```java
+public static class Improve implements Task {
+  @Override
+  public void execute(Context context, Client client) {
+    Metrics previous = context.loop().getPrevious(Metrics.class);
+    double estimate = previous == null ? 1.0 : previous.estimate();
+    client.setXCom((estimate + 2.0 / estimate) / 2.0);
+  }
+}
+
+public static class AccurateEnough implements UntilTask {
+  @Override
+  public LoopDecision decide(Context context, Client client) {
+    Metrics result = context.loop().getResult(Metrics.class);
+    return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+  }
+}
+```
+
+`UntilTask` joins `ConditionTask` and `SwitchTask` as a task kind with its own 
verb, so the gate is
+recognized by the interface it implements rather than by a naming rule.
+
+## Signature
+
+```java
+// Annotation surface. @Airflow.Until carries @Airflow.Task's own attributes 
as well.
+@interface Airflow.Loop { String id(); int maxIterations(); }
+@interface Airflow.Until { String id() default ""; }
+
+// The gate's method under @Airflow.Until. Its injected arguments come first 
and its wired data
+// after, as any task method's do, and it returns either shape; a boolean is a 
decision with no
+// reason.
+//
+//   LoopDecision accurateEnough(Metrics result)
+//   LoopDecision accurateEnough(Loop<Metrics> loop, Metrics result)
+//   boolean accurateEnough(Metrics result)
+
+// Injected into a loop body task or its gate, beside Client and Context. On 
the interface
+// surface it is context.loop(). It throws outside a loop.
+// T is the terminal task's result type. Declare it (Loop<Metrics>) and 
previous is typed;
+// declare none (Loop<Object>, which context.loop() returns) and name the type 
at the read.
+public final class Loop<T> {
+  public final int index;
+
+  public final int maxIterations;
+
+  /** The previous iteration's terminal result, or null at iteration 1. */
+  public final T previous;
+
+  /**
+   * This iteration's terminal result, for a gate on the interface surface, 
where there is no
+   * ref to bind it from. Null in a body task. The annotation surface takes it 
as a parameter.
+   */
+  public final T result;
+
+  /** previous bound to a DTO, as Client.getXCom's caller names the type it 
wants. */
+  public <R> R getPrevious(Class<R> type);
+
+  public <R> R getResult(Class<R> type);
+}
+
+/** The lazy sequence of ADR-0016, reached as a declared parameter type or as 
Loop's T. */
+public interface XComSequence<E> {
+  int length();
+
+  E get(int offset);
+}
+
+/** What an until method returns. A boolean return is a decision with no 
reason. */
+public record LoopDecision(boolean stop, String reason) {}
+
+/** Interface surface. */
+public interface UntilTask {
+  LoopDecision decide(Context context, Client client);
+}
+
+public interface LoopRef extends TaskGroupRef, Flow {
+  /** Declares the gate. The terminal task is the body's single leaf. */
+  TaskRef<Void> until(String taskId, Class<? extends UntilTask> gate);
+}
+
+public class Dag {
+  public LoopRef loop(String groupId, int maxIterations);
+}
+
+/** Wiring view, generated. A loop body task's ref reads as one iteration or 
as all of them. */
+public interface LoopTaskRef<T> extends TaskRef<T> {}
+```
+
+## Consequences
+
+- **A fixed-count loop needs no runtime support.** A `@Airflow.Loop` with no 
`@Airflow.Until` emits
+  `has_until: false`, which core completes without a worker (ADR-0017 decision 
4).
+- **The two surfaces name the terminal differently.** The annotation surface 
binds it and gets a
+  compile-time check; the interface surface infers the body's single leaf. 
Both serialize the same
+  `terminal_task_id`.
+- **The body's single leaf is checked twice**: by the SDK when the Dag is 
read, for the faster
+  message, and by the Dag processor, which decides (ADR-0017 decision 3). A 
nested loop and a loop
+  inside a mapped task group are rejected the same way.
+- **A wrong `Loop<T>` is caught when the Dag is read.** The processor knows 
the terminal task's return
+  type, so `Loop<String>` over a `Metrics` terminal is named as an error 
rather than failing as a cast
+  at run time. `Loop<Object>` is always accepted.
+- **The overload pair of decision 7 is generated for every `XComSequence<E>` 
parameter**, used or not.

Review Comment:
   This is one extra method only when a task has a single sequence parameter. A 
task reading two loop outputs needs all four combinations, and three needs 
eight. Worth saying so, or picking a shape that doesn't multiply?



##########
go-sdk/adr/0009-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,211 @@
+<!--
+ 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.
+ -->
+
+# 9. Task loop authoring interface
+
+Date: 2026-11-03
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Go spelling of them.
+
+## Decision
+
+1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and 
fills it as usual
+   ([ADR 8](0008-native-dag-interface.md)).
+2. **`maxIterations` is positional and `Until` is an option.** Every loop has 
a bound; only some add a
+   callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as 
`TaskOption` is.
+3. **The gate is a task handler wired with `airflow.Inputs`.** Registration 
checks its parameter
+   against the terminal task's recorded output type, as it already does for 
`Inputs`.
+4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, 
`TaskInstance` and
+   `DagRun`. It returns a value and an error, because reading the previous 
iteration is a round trip.
+5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as 
`DagRun`'s are, and
+   `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split 
between the value it returns
+   and the `out` it decodes into.
+6. **A body ref is read from outside the loop unchanged.** The consuming 
handler declares `T` for one
+   iteration or `airflow.XComSequence[T]` for every iteration, checked at 
registration.
+7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a 
body ref to a task
+   outside is rejected when it is declared.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```go
+type Metrics struct {
+    Estimate float64 `json:"estimate"`
+    Error    float64 `json:"error"`
+}
+
+func improve(actx airflow.Context) (float64, error) {
+    loop, err := actx.Loop()
+    if err != nil {
+        return 0, err
+    }
+    actx.Logger().Info("refining", "iteration", loop.Index, "of", 
loop.MaxIterations)
+
+    estimate := 1.0
+    if loop.Previous != nil {
+        var previous Metrics
+        if err := loop.UnmarshalPrevious(&previous); err != nil {
+            return 0, err
+        }
+        estimate = previous.Estimate
+    }
+    return (estimate + 2.0/estimate) / 2.0, nil
+}
+
+func evaluate(actx airflow.Context, estimate float64) (Metrics, error) {
+    return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 
2)}, nil
+}
+
+func accurateEnough(actx airflow.Context, result Metrics) 
(airflow.LoopDecision, error) {
+    return airflow.LoopDecision{
+        Stop:   result.Error < 1e-6,
+        Reason: fmt.Sprintf("error %.2g", result.Error),
+    }, nil
+}
+
+func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error 
{
+    n, err := history.Len()
+    if err != nil {
+        return err
+    }
+    actx.Logger().Info("settled", "iterations", n)
+    return nil
+}
+```
+
+```go
+dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"})
+refine := dag.TaskGroup("refine")
+
+improved := refine.Task(improve)
+evaluated := refine.Task(evaluate, airflow.Inputs(improved))
+
+refinement := refine.Loop(10, airflow.Until(accurateEnough, 
airflow.Inputs(evaluated)))
+
+dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, 
through the gate
+refinement.Before(dag.Task(cleanup))        // order only: the whole loop, not 
one iteration
+```
+
+`evaluated` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an 
`airflow.XComSequence[Metrics]`, which is what
+`report` takes. Decision 6 is what makes registration accept both.
+
+A loop that only counts needs no callable:
+
+```go
+poll := dag.TaskGroup("poll")
+poll.Task(checkOnce)
+poll.Loop(3)
+```
+
+## Signature
+
+```go
+package airflow
+
+// Loop turns the group into a task loop. maxIterations must be positive.
+func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef
+
+// Until supplies the gate's callable and the gate's own task options, 
including the Inputs that
+// name the loop's terminal task. fn takes airflow.Context first and the wired 
inputs after, as
+// every handler does, and returns either shape:
+//
+//    func(actx airflow.Context, args ...) (LoopDecision, error)
+//    func(actx airflow.Context, args ...) (bool, error)
+func Until(fn any, opts ...TaskOption) LoopOption
+
+// LoopOption is sealed: its only method is unexported, so a loop takes 
SDK-defined options and
+// nothing else.
+type LoopOption interface{ applyLoop(*loopConfig) error }
+
+// LoopDecision is what an until callable returns. A bool return is a decision 
with no reason.
+type LoopDecision struct {
+    Stop   bool
+    Reason string
+}
+
+// LoopRef is a Node, so Before and After carry edges for the whole loop.
+type LoopRef struct{ /* unexported */ }
+
+// Loop reads the previous iteration's result once and caches it. It panics 
outside a loop body or
+// gate, as the other accessors do on a zero Context, and returns an error 
when the read fails.
+func (c Context) Loop() (Loop, error)
+
+// Loop is handed back by value, as DagRun is: the fields are the runtime's 
answer, not state for
+// a handler to write.
+type Loop struct {
+    Index         int
+    MaxIterations int
+
+    // Previous is the previous iteration's terminal result as it arrived: nil 
at Index 1, and an
+    // XComSequence when the terminal task is mapped.
+    Previous any
+}
+
+// UnmarshalPrevious binds Previous to a DTO, the way XComClient.GetXCom 
decodes into its out

Review Comment:
   `GetXCom` doesn't decode into `out` today. `sdk.go` says `value` "is 
reserved for future typed decoding and is currently ignored", and the stored 
value comes back as the first result. Decision 5 (L45-47) cites the same 
precedent.



##########
ts-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,166 @@
+<!--
+ 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-0003: Task Loop — TypeScript Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the TypeScript spelling of them.
+
+## Decision
+
+1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and 
fills it as usual.
+2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, 
the shape
+   `dag.if(handler, inputs, spec)` already has 
([ADR-0002](0002-native-dag-interface.md)).
+3. **`getLoop()` supplies the loop context**, beside `getContext()` and 
`getClient()` (ADR-0002
+   decision 6). It is awaited, because reading the previous iteration is a 
round trip.
+4. **`index`, `maxIterations` and `previous` are fields**, and 
`getPrevious<T>()` is the typed view.
+   That is `client.getXCom`'s split between a value and the type the caller 
names.
+5. **A body task returns a `LoopTaskRef<T>`**, accepted where either 
`TaskRef<T>` or

Review Comment:
   Outside the loop, core only serves the all-iterations read. 
`task_coordinates.py` raises "A loop producer requires an explicit scope or a 
consumer inside the loop" for a single-value read from outside. So 
`dag.task(f)({ m: evaluated })` with `m: Metrics` outside `refine` passes `tsc` 
here and then fails at run time. Should the SDK reject a non-sequence consumer 
outside the loop when the Dag is read? Go decision 6 and Java decision 7 make 
the same promise, and it traces back to rule 6 in ADR-0017.



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and

Review Comment:
   `Context` on main has only `dagRun` and `ti`. `conf` and `runId` live on 
`context.dagRun`.



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and
+   `context.runId` already are. `previous` is `null` at iteration 1.
+6. **`Loop<T>` is generic over the terminal task's result**, so 
`Loop<Metrics>` gives a typed
+   `previous` with nothing to cast. The interface surface has no declaration 
to read, so
+   `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names 
the type at the read,
+   as `Client.getXCom` does.
+7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor 
generates every wiring
+   parameter of type `XComSequence<E>` as an overloaded pair, one taking 
`Arg<XComSequence<E>>` and
+   one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the 
consumer declares.
+8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body 
ref to a task outside is
+   rejected when the Dag is read.
+
+## Example
+
+Newton's method for the square root of two, stopping on the error or at ten 
iterations.
+
+```java
+@Airflow.Dag(id = "java_refine", schedule = "@daily")
+public class RefineEstimate {
+  private static final System.Logger log = 
System.getLogger(RefineEstimate.class.getName());
+
+  record Metrics(double estimate, double error) {}
+
+  // A loop body: a task group with a bound. "refine.improve", 
"refine.evaluate",
+  // "refine.accurate_enough".
+  @Airflow.Loop(id = "refine", maxIterations = 10)
+  static class Refine {
+
+    @Airflow.Task(id = "improve")
+    public double improve(Loop<Metrics> loop) {
+      log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations);
+      double estimate = loop.previous == null ? 1.0 : loop.previous.estimate();
+      return (estimate + 2.0 / estimate) / 2.0;
+    }
+
+    @Airflow.Task(id = "evaluate")
+    public Metrics evaluate(double estimate) {
+      return new Metrics(estimate, Math.abs(estimate * estimate - 2));
+    }
+
+    // The gate. Its parameter is the loop's terminal task, bound in depends().
+    @Airflow.Until(id = "accurate_enough")
+    public LoopDecision accurateEnough(Metrics result) {
+      return new LoopDecision(result.error() < 1e-6, "error 
%.2g".formatted(result.error()));
+    }
+  }
+
+  @Airflow.Task(id = "report")
+  public void report(XComSequence<Metrics> history) {
+    log.log(INFO, "settled after {0} iterations", history.length());
+  }
+
+  @Airflow.Task(id = "cleanup")
+  public void cleanup() {}
+
+  @Airflow.Deps
+  static class Wiring implements RefineEstimateDeps {
+    void depends() {
+      var estimate = refine().improve();
+      var metrics = refine().evaluate(estimate);
+      refine().accurateEnough(metrics); // the gate, and the terminal it reads
+      report(metrics);                  // every iteration's metrics, through 
the gate
+      refine().before(cleanup());       // order only: the whole loop, not one 
iteration
+    }
+  }
+}
+```
+
+`metrics` is written twice and means two things. Inside the loop it is one 
iteration's `Metrics`,
+which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which 
is what `report` takes.
+Decision 7 is what makes both compile:
+
+```java
+// Generated by the Airflow Java SDK annotation processor. Do not edit.
+interface RefineEstimateDeps extends Deps {
+  default Refine refine() { return Flow.loop("refine"); }
+
+  interface Refine extends LoopRef {
+    default LoopTaskRef<Double> improve() { return 
Flow.node("refine.improve"); }
+    default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return 
Flow.call("refine.evaluate", estimate); }
+    default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return 
Flow.call("refine.accurate_enough", result); }
+  }
+
+  default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return 
Flow.call("report", history); }
+  default TaskRef<Void> cleanup() { return Flow.node("cleanup"); }
+}
+```
+
+A loop that only counts needs no gate method at all:
+
+```java
+@Airflow.Loop(id = "poll", maxIterations = 3)
+static class Poll {
+  @Airflow.Task(id = "check_once")
+  public void checkOnce() {}
+}
+```
+
+### Interface surface
+
+```java
+Dag dag = new Dag("java_refine").config("schedule", "@daily");
+
+LoopRef refine = dag.loop("refine", 10);
+TaskRef improve = refine.task("improve", Improve.class);
+TaskRef evaluate = refine.task("evaluate", Evaluate.class);
+improve.before(evaluate);
+refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's 
single leaf
+
+TaskRef report = dag.task("report", Report.class);
+evaluate.before(report); // data edge out of the body, through the gate

Review Comment:
   `evaluate.before(report)` is an ordering-only edge (`TaskRef.before` in 
`Arg.kt`), which is the body-to-outside edge decision 8 rejects, and it binds 
no data either. How does `Report` read every iteration on the interface 
surface? `client.getXCom` is a single-value read, and core refuses that from 
outside the loop.



##########
java-sdk/adr/0003-task-loop-authoring-interface.md:
##########
@@ -0,0 +1,276 @@
+<!--
+ 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-0003: Task Loop — Java Authoring Interface
+
+## Status
+
+Proposed.
+
+## Context
+
+A task loop runs a task group again and again until a gate says stop. What a 
loop means, and the
+eight authoring rules every SDK follows, are in
+[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md).
+This ADR records the Java spelling of them, on both surfaces
+[ADR-0002](0002-native-dag-interface.md) defines.
+
+## Decision
+
+1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface 
surface,
+   `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` 
does.
+2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence 
is a fixed-count loop.
+3. **The gate's data parameter is the terminal task**, bound in `depends()` 
like any other argument.
+   The interface surface has no typed refs, so `refine.until(id, class)` takes 
the body's single leaf.
+4. **`Loop` is injected as a method argument**, beside `Client` and `Context` 
(ADR-0002), into a body
+   task or the gate. On the interface surface, where `Task.execute(Context, 
Client)` is fixed, it is
+   `context.loop()`. The processor injects it only where it is declared, so 
nothing else pays for the
+   read.
+5. **`index`, `maxIterations` and `previous` are public final fields**, as 
`Context.conf` and
+   `context.runId` already are. `previous` is `null` at iteration 1.
+6. **`Loop<T>` is generic over the terminal task's result**, so 
`Loop<Metrics>` gives a typed
+   `previous` with nothing to cast. The interface surface has no declaration 
to read, so
+   `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names 
the type at the read,

Review Comment:
   `Client.getXCom` takes no `Class` and returns `Any?`, so this is a new 
pattern rather than the existing one. A `Class` also can't carry 
`XComSequence<Metrics>`, so for a mapped terminal 
`getPrevious(XComSequence.class)` loses the element type. How does the 
interface surface get a typed sequence in that case?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org