This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 762495647a9 Add task loop authoring ADRs for the TypeScript, Java and
Go SDKs (#76365)
762495647a9 is described below
commit 762495647a9c5ba23d49267230c69c79087653a2
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Tue Nov 10 13:24:17 2026 +0800
Add task loop authoring ADRs for the TypeScript, Java and Go SDKs (#76365)
* Add task loop authoring ADRs for the TypeScript, Java and Go SDKs
Each SDK records how an author writes a task loop in its own language, with
the shared semantics left to the core ADR the three of them cite.
* Take the terminal task from the body's leaf, not the gate's wiring
A gate that bound a non-leaf body task was treated as naming the terminal,
and nothing then ordered the real leaf before it. Say the terminal is the
leaf in all three SDKs, and name the gate a fixed-count loop gets.
* Say that a single-value read from outside a loop is rejected
The three ADRs let the consumer's declared type pick the reading, but
which reading is legal depends on where the consumer sits, which no type
can see. Core serves only the all-iterations read from outside the loop,
so say that the SDK rejects the other one when the Dag is read.
* Type the TypeScript loop signature the way dag.if is typed
The gate's inputs were typed as the handler's own argument object, so the
worked example did not type-check: one type had to be both the arguments
and the refs supplying them. Take them through TaskInputs under NoInfer,
as dag.if does, and add the two refs the decisions name.
TaskInputs accepts any TaskRef for any argument, so decision 5 cannot
promise a compile-time check of the reading either. Say what tsc does
check.
* Drop the type parameter from the Go XComSequence
Handlers register as fn any and the SDK binds their parameters by
reflection, which instantiates no generic type, so the runtime could
never build an XComSequence[Metrics] to pass or to assert Previous to.
Name the element type at the read instead, next to UnmarshalPrevious.
* Report NotInLoop from actx.Loop instead of panicking
Logger, Client, TaskInstance and DagRun all hand back a stored field and
never panic; only NewContext panics, and only on a nil argument. Loop
already returns an error, so a handler outside a loop gets one there.
* Give the Java interface surface the reads it needs
getPrevious(Class) had no counterpart on Client, which returns Object and
takes no Class, and a Class cannot carry XComSequence<Metrics>, so a
mapped terminal lost its element type. Add both overloads to Client and
name them. The report example also drew an ordering edge out of the body
and read nothing, so point it at the loop and show the sequence read.
* Name the Java types the loop refs need changed
TaskRef and TaskGroupRef are final Kotlin classes, so neither LoopTaskRef
nor LoopRef compiled as written, and UntilTask did not extend Task the way
ConditionTask and SwitchTask do while taking decide, which is already
ConditionTask's verb. Say both, and give the gate a verb of its own.
* Declare cleanup in the TypeScript and Go examples
Both worked examples drew an ordering edge to a task neither declared.
* Generate the loop overload pair only where both readings exist
A pair per XComSequence parameter is a cross-product: four methods for a
task reading two loop sequences, eight for three. Generate it only where
a loop body task returns the element type, and say what it would cost
otherwise.
* Default UntilTask.execute the way ConditionTask does
A gate class implements check(), so Task.execute has to come from
somewhere; ConditionTask overrides it with a throwing default and the
gate takes the same one. Also line the interface surface's terminal read
up with rule 3, which now covers a surface that wires no data.
* Ask for the outside reading with all()
Decision 6 had the consumer's declared parameter type pick between one
iteration's value and every iteration's. Nothing can check that:
`TaskInputs` takes any `TaskRef` for any argument, and
`serializeArgBindings` records a task id and no type, so a wrong reading
serialized cleanly and failed partway through the run.
`all()` puts the outside reading in the Dag, and a bare body ref outside
the loop is rejected when the Dag is read. Go and Java keep the bare ref,
which their registration and generated overloads can check.
* Make getPrevious the only previous-iteration accessor
Decision 5 cited a split in `client.getXCom` that does not exist: it is
one generic method, `getXCom<T = unknown>()`. ADR-0017 also asks every
runtime for one accessor rather than two, so the `previous` field goes
and `getPrevious<T = unknown>()` stands alone.
* Match the UnmarshalPrevious comment to decision 5
The comment said `XComClient.GetXCom` decodes into its out parameter.
Decision 5 already says that parameter is reserved for the decoding and
ignored today; the comment now says the same.
---
go-sdk/adr/0009-task-loop-authoring-interface.md | 245 +++++++++++++++
java-sdk/adr/0003-task-loop-authoring-interface.md | 327 +++++++++++++++++++++
ts-sdk/adr/0003-task-loop-authoring-interface.md | 214 ++++++++++++++
3 files changed, 786 insertions(+)
diff --git a/go-sdk/adr/0009-task-loop-authoring-interface.md
b/go-sdk/adr/0009-task-loop-authoring-interface.md
new file mode 100644
index 00000000000..6ba0b1e2d78
--- /dev/null
+++ b/go-sdk/adr/0009-task-loop-authoring-interface.md
@@ -0,0 +1,245 @@
+<!--
+ 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`**, which is its
own data, as any task's
+ is. The terminal task is the body's single leaf, which the SDK orders
before the gate. Wiring the
+ leaf in is what gets the decision checked against what the body produces,
since registration
+ already checks a parameter against the recorded output type of whatever
`Inputs` names.
+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,
+ and outside a loop that error is `airflow.NotInLoop`.
+5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as
`DagRun`'s are, and
+ `UnmarshalPrevious(out)` binds the raw one. That is the split `GetXCom`
declares, whose `value`
+ parameter is reserved for exactly this decoding and ignored today.
+6. **A body ref is read from outside the loop unchanged.** The consuming
handler declares `T` for one
+ iteration or `airflow.XComSequence` for every iteration, and registration
checks which it asked
+ for against where it sits.
+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) error {
+ n, err := history.Len()
+ if err != nil {
+ return err
+ }
+ var last Metrics
+ if err := history.UnmarshalAt(n-1, &last); err != nil {
+ return err
+ }
+ actx.Logger().Info("settled", "iterations", n, "error", last.Error)
+ return nil
+}
+
+func cleanup(actx airflow.Context) error {
+ 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`, 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. Outside a
loop body or its gate it
+// reports NotInLoop, and it reports the read's own error when that fails.
+func (c Context) Loop() (Loop, error)
+
+// NotInLoop is what Loop reports for a handler that is not in a loop. Spelled
like the SDK's other
+// sentinels, sdk.XComNotFound among them: errors.Is(err, airflow.NotInLoop).
+var NotInLoop = errors.New("not in a loop")
+
+// 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 decoding
XComClient.GetXCom's value parameter is
+// reserved for and does not do today. 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 not generic. The SDK builds it by reflection from a handler
registered as fn any, and Go
+// reflection instantiates no generic type and synthesizes no method set, so
an XComSequence[T] is
+// one the runtime could never hand to a handler. The element type is named at
the read instead.
+type XComSequence interface {
+ Len() (int, error)
+
+ // At returns one element as it arrived, as GetXCom returns a stored value.
+ At(offset int) (any, error)
+
+ // UnmarshalAt decodes one element into a DTO, as UnmarshalPrevious
decodes Previous.
+ UnmarshalAt(offset int, out any) error
+}
+```
+
+## Consequences
+
+- **A fixed-count loop needs no runtime support.** `Loop(3)` emits `has_until:
false`, which core
+ completes without a worker (ADR-0017 decision 4).
+- **The gate is an ordinary task.** `airflow.Until` takes `TaskOption`, so
`airflow.TaskSpec{Queue: ...}`
+ lands on it, and its task_id defaults to the Go function name under the
group's prefix.
+ `TaskSpec{TaskID: "accurate_enough"}` is how it gets a snake_case id, as for
any task. A loop with
+ no `airflow.Until` has no name to take, so its gate is `__loop_gate`, as
Python's is.
+- **The body's single leaf is checked twice**: at registration, 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.
+- **`actx.Loop()` reports `NotInLoop` outside a loop** rather than panicking.
The other accessors
+ hand back a stored field and never panic; only `NewContext` panics, and only
on a nil argument. A
+ handler moved out of a group still fails where it was moved, through the
error it already checks.
+- **A mapped terminal needs no second accessor.** `Previous` holds an
`airflow.XComSequence` when
+ `terminal_is_mapped` is true and a plain value otherwise, which is what
Python's one
+ `loop.previous` property does. The mapped spelling is a type assertion,
+ `loop.Previous.(airflow.XComSequence)`, since a sequence is not something to
decode into.
+- **The reading has to match where the consumer sits.** Core serves only the
all-iterations read from
+ outside the loop, so a consumer outside it that declares one iteration's
value is rejected when the
+ Dag is read rather than failing partway through the run.
+- **Nothing in Go carries the element type.** `Previous` is a struct field and
`XComSequence` is
+ built by reflection, and neither can be generic, so both name the type at
the read:
+ `UnmarshalPrevious` for one value, `UnmarshalAt` for one element. Go is the
only one of the three
+ SDKs where the signature does not say it.
+
+## Alternatives
+
+- **`*Loop` rather than `Loop`.** Rejected: the fields are the runtime's
answer, and a pointer invites
+ writing to them. `DagRun()` hands back a value for the same reason.
+- **A `Loop` parameter on the handler**, as the Java SDK injects one.
Rejected: a Go handler already
+ takes `airflow.Context` first, and everything else Airflow gives a task
hangs off it.
+- **A generic `XComSequence[T]`**, so the element type sits in the handler's
signature. Rejected
+ because it cannot be built: handlers register as `fn any`, and Go reflection
instantiates no
+ generic type, so the runtime has nothing to pass for such a parameter and
nothing to assert
+ `Previous` to.
+- **Fetching `Previous` before every body handler runs**, which would drop the
error from `Loop()`.
+ Rejected: a handler that never reads it would pay a round trip from
iteration 2 on.
diff --git a/java-sdk/adr/0003-task-loop-authoring-interface.md
b/java-sdk/adr/0003-task-loop-authoring-interface.md
new file mode 100644
index 00000000000..ad788fc34db
--- /dev/null
+++ b/java-sdk/adr/0003-task-loop-authoring-interface.md
@@ -0,0 +1,327 @@
+<!--
+ 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 terminal task is the body's single leaf**, which the SDK orders
before the gate. The gate's
+ data parameters are bound in `depends()` like any other task's, so binding
the leaf is what gets
+ the decision checked against what the body produces. The interface surface
binds nothing and reads
+ the terminal off `Loop` (decision 6).
+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
`DagRun.conf` and
+ `DagRun.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.
+ `Client.getXCom` gains that same overload, so naming the type at the read
is one shape across the
+ surface rather than something the loop invents.
+7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor
generates a wiring
+ parameter of type `XComSequence<E>` as an overloaded pair — one taking
`Arg<XComSequence<E>>`, one
+ taking `LoopTaskRef<E>` — where some loop body task in the Dag returns `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
[email protected](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 Flow {
+ 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
[email protected](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);
+refine.before(report); // the whole loop: this surface wires no data out of
the body
+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 check(Context context, Client client) {
+ Metrics result = context.loop().getResult(Metrics.class);
+ return new LoopDecision(result.error() < 1e-6, "error
%.2g".formatted(result.error()));
+ }
+}
+
+public static class Report implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ XComSequence<Metrics> history = client.getXComSequence(Metrics.class,
"refine.evaluate");
+ System.getLogger("report").log(INFO, "settled after {0} iterations",
history.length());
+ }
+}
+```
+
+`UntilTask` extends `Task` and joins `ConditionTask` and `SwitchTask` as a
task kind with its own
+verb — `decide` is already `ConditionTask`'s — 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);
+}
+
+// Added to Client. getXCom returns Object today, so naming the type at the
read is new; these are
+// the overloads getPrevious and getResult are modelled on. Both take the rest
of getXCom's
+// parameters, which keep their defaults.
+public <T> T getXCom(Class<T> type, String taskId);
+
+// A Class cannot carry XComSequence<Metrics>, so the sequence read names the
element type instead.
+public <E> XComSequence<E> getXComSequence(Class<E> elementType, String
taskId);
+
+/** 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, beside ConditionTask and SwitchTask, which both extend
Task as well. */
+public interface UntilTask extends Task {
+ LoopDecision check(Context context, Client client);
+
+ /** Never called: the SDK runs a gate through check(), as it runs a
condition through decide(). */
+ @Override
+ default void execute(Context context, Client client) {
+ throw new IllegalStateException("...");
+ }
+}
+
+/** A loop is a group with a bound, so its ref is a TaskGroupRef that also
declares the gate. */
+public class LoopRef extends TaskGroupRef {
+ /** Declares the gate. The terminal task is the body's single leaf. */
+ public 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 class 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).
+- **Neither surface names the terminal.** Both take the body's single leaf and
serialize the same
+ `terminal_task_id`. The annotation surface binds it into the gate as well,
which is what buys the
+ compile-time check; the interface surface has no wiring to bind it with.
+- **A fixed-count loop's gate is `__loop_gate`.** `@Airflow.Until` carries the
gate's id, and a loop
+ with no gate method has none to carry, so the id falls back under the
group's prefix as Python's
+ does.
+- **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 only where both readings
exist**, that is for an
+ `XComSequence<E>` parameter whose `E` some loop body task returns. A
sequence fed only by a mapped
+ task keeps the one `Arg<XComSequence<E>>` method. The narrowing is what
keeps the pair from
+ multiplying: a task reading two loop sequences would otherwise need four
methods and one reading
+ three would need eight, where in practice almost every such parameter has a
single reading.
+- **`context.loop()` throws outside a loop**, so a task class moved out of a
group fails where it was
+ moved.
+- **A mapped terminal needs no second accessor.** `loop.previous` holds
ADR-0016's lazy sequence when
+ `terminal_is_mapped` is true and a plain value otherwise, which is what
Python's one
+ `loop.previous` property does. The mapped spelling is the declaration,
`Loop<XComSequence<Metrics>>`.
+ A `Class` cannot carry that type, so the interface surface reads the
sequence with
+ `client.getXComSequence(Metrics.class, taskId)`.
+- **The interface surface reads the terminal off `Loop`**, which is the second
half of ADR-0017
+ rule 3: `Task.execute(Context, Client)` is fixed, so there is no wiring to
bind the terminal with.
+ `loop.getResult(Metrics.class)` names the type at the read as
`Client.getXCom` does, and core pins
+ the read to the gate's iteration either way.
+- **The interface surface wires no data out of the body**, so ADR-0017
decision 3 has nothing to
+ rewire there: a consumer outside the loop orders itself after the loop and
reads the sequence
+ through `Client`.
+- **Two shipped types open up.** `TaskRef<T>` and `TaskGroupRef` are final
Kotlin classes today, and
+ `LoopTaskRef<T>` and `LoopRef` sit under them, so both become `open`. Their
constructors stay
+ internal, so the SDK is still the only thing that subclasses them, and `Arg`
stays sealed because
+ sealing constrains only its direct subclasses. The generated wiring view of
a loop extends `Flow`
+ rather than `LoopRef`, since an interface cannot extend a class.
+- **The reading has to match where the consumer sits.** Core serves only the
all-iterations read from
+ outside the loop, so a consumer outside it that declares one iteration's
value is rejected when the
+ Dag is read rather than failing partway through the run.
+
+## Alternatives
+
+- **A getter on an SDK base class**, `getLoop()`, rather than an injected
argument. Rejected for the
+ reason [ADR-0002](0002-native-dag-interface.md) rejects it for `Client` and
`Context`: a task method
+ declares what it needs, and the SDK binds it.
+- **Fetching `previous` for every body method**, which would let `Loop` be
injected everywhere.
+ Rejected: a method that never declares `Loop` would pay a round trip from
iteration 2 on.
diff --git a/ts-sdk/adr/0003-task-loop-authoring-interface.md
b/ts-sdk/adr/0003-task-loop-authoring-interface.md
new file mode 100644
index 00000000000..bd6e1abf3b9
--- /dev/null
+++ b/ts-sdk/adr/0003-task-loop-authoring-interface.md
@@ -0,0 +1,214 @@
+<!--
+ 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)). `inputs` is
+ the gate's own data, as any task's is.
+3. **The terminal task is the body's single leaf**, which the SDK orders
before the gate. Wiring the
+ leaf into `inputs` is what gets the decision checked against what the body
produces.
+4. **`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.
+5. **`index` and `maxIterations` are fields, and `getPrevious<T>()` is the one
accessor** for the
+ previous iteration's terminal result — one, not two, as ADR-0017 asks of
every runtime. It is a
+ generic method naming its own return type, as `client.getXCom<T>()` is, not
a field with a typed
+ view beside it.
+6. **A body task returns a `LoopTaskRef<T>`**, a `TaskRef<T>` that records the
loop it came from.
+ Inside the loop it reads as one iteration's `T`. Outside it, `all()`
returns a
+ `TaskRef<XComSequence<T>>` and a bare ref is rejected when the Dag is read,
since core serves only
+ the all-iterations read from outside and the SDK records no argument type
it could check instead.
+7. **`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 cleanup = dag.task(async function cleanup() {});
+
+const estimate = improve();
+const evaluated = evaluate({ estimate });
+
+const refinement = refine.loop(accurateEnough, { result: evaluated }, {
maxIterations: 10 });
+
+report({ history: evaluated.all() }); // every iteration's metrics, through
the gate
+refinement.before(cleanup()); // order only: the whole loop, not one iteration
+```
+
+`evaluated` is written twice. Inside the loop it is one iteration's `Metrics`,
which is what the
+gate takes. Outside it only `evaluated.all()` is allowed, an
`XComSequence<Metrics>`, which is what
+`report` takes; a bare `evaluated` out there is rejected when the Dag is read.
+
+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. `TArgs` is its own arguments, one key each, exactly as
a task handler takes
+ * its own. Nothing of the SDK's is in `TArgs`; the loop context comes from
`getLoop()`.
+ */
+type UntilCallable<TArgs extends object> = (
+ args: TArgs,
+) => LoopDecision | boolean | Promise<LoopDecision | boolean>;
+
+interface TaskGroupRef {
+ loop(options: LoopOptions): LoopRef;
+ /**
+ * `NoInfer` keeps `TArgs` coming from the handler alone, so `inputs` is
checked key by key against
+ * it rather than widening it. `dag.if` takes its inputs the same way.
+ */
+ loop<TArgs extends object>(
+ until: UntilCallable<TArgs>,
+ inputs: TaskInputs<NoInfer<TArgs>>,
+ options: LoopOptions,
+ ): LoopRef;
+}
+
+/** What `loop()` returns: the whole loop as one node, which is what an
ordering edge leaves from. */
+interface LoopRef extends Node {
+ readonly groupId: string;
+ before(...downstream: readonly Node[]): LoopRef;
+ after(...upstream: readonly Node[]): LoopRef;
+}
+
+/**
+ * What a loop body task returns. It is a `TaskRef<T>`, so it goes wherever
one goes; the loop it
+ * records is what lets the SDK tell a body ref from an ordinary one when the
Dag is read.
+ */
+interface LoopTaskRef<T> extends TaskRef<T> {
+ readonly loopId: string;
+
+ /** Every iteration's value, the one reading core serves from outside the
loop. */
+ all(): TaskRef<XComSequence<T>>;
+}
+
+/**
+ * Read inside a loop body or its gate. Throws anywhere else.
+ * Awaited because it reads the previous iteration's result, which it caches
per task run.
+ */
+function getLoop(): Promise<LoopContext>;
+
+interface LoopContext {
+ readonly index: number;
+ readonly maxIterations: number;
+
+ /**
+ * The previous iteration's terminal result, as the type the caller names,
or `null` at iteration
+ * 1. An `XComSequence<T>` when the terminal task is mapped.
+ */
+ getPrevious<T = unknown>(): T | null;
+}
+```
+
+## Consequences
+
+- **A fixed-count loop needs no runtime support.** `loop({ maxIterations })`
emits `has_until: false`,
+ which core completes without a worker (ADR-0017 decision 4).
+- **The gate is an ordinary task.** `LoopOptions` extends `TaskSpec`, so
`queue` and `retries` land on
+ it, and its id defaults to the handler's function name under the group's
prefix. A fixed-count loop
+ has no handler to take a name from, so its gate is `__loop_gate`, as
Python's is.
+- **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.
+- **`getLoop()` throws outside a loop**, so a handler moved out of a group
fails where it was moved.
+- **A mapped terminal needs no new call.**
`getPrevious<XComSequence<Metrics>>()` names the sequence
+ type, and `terminal_is_mapped` tells the runtime which to build.
+- **The reading is written, not inferred.** `serializeArgBindings` records a
task id and no argument
+ type, so the SDK cannot tell which reading a consumer declared. `all()` puts
the outside reading in
+ the Dag instead, and `tsc` checks it because the accessor names its own
return type. A bare body
+ ref outside the loop is rejected when the Dag is read rather than failing
partway through the run.
+- **`loop()` runs once per group.** A group already on an edge can still
become one, since `LoopRef`
+ takes its edges over.
+
+## Alternatives
+
+- **The loop context as another key of the gate's argument**,
`accurateEnough({ loop, result })`.
+ Rejected for the reason [ADR-0002](0002-native-dag-interface.md) rejects it
for `ctx` and `client`:
+ a handler's argument is its own data, and merging SDK values into it leaves
the top-level names open
+ to collide with the author's. Python puts `loop` among the context kwargs
because every Python
+ handler already takes them. No TypeScript handler does.
+- **A synchronous `getLoop()`**, with the previous result fetched before every
body handler runs.
+ Rejected: a handler that never reads it would pay a round trip from
iteration 2 on. Python pays
+ nothing for the same reason, since `loop.previous` is a lazy property.
+- **A bare body ref outside the loop, with the consumer's declared parameter
type picking the
+ reading.** Rejected: nothing can check it. `TaskInputs` accepts any
`TaskRef` for any argument, and
+ `serializeArgBindings` records no type, so a consumer that declared one
iteration's value would
+ serialize cleanly and fail partway through the run. The Go SDK keeps the
bare ref because
+ registration records its handlers' parameter types, and the Java SDK because
it generates an
+ overload per reading; TypeScript has neither, so it asks for the reading
instead.
