diff --git a/airflow-core/adr/lang-sdk/README.md b/airflow-core/adr/lang-sdk/README.md index ba7800a97407b..b6768d6139610 100644 --- a/airflow-core/adr/lang-sdk/README.md +++ b/airflow-core/adr/lang-sdk/README.md @@ -35,4 +35,6 @@ bind core interfaces and apply to every language SDK, not just the Java SDK. - [ADR-0007](0007-taskflow-across-language-boundary.md): TaskFlow across the language boundary — argument binding for Lang-SDK tasks. Decisions specific to a single SDK stay next to that SDK — for example, the Go SDK's bundle-format -decisions live in [`go-sdk/adr/`](../../../go-sdk/adr). +decisions live in [`go-sdk/adr/`](../../../go-sdk/adr). Java-SDK-only interface-design decisions — +the Java call-site syntax itself, rather than anything binding a core surface — live in +[`java-sdk/adr/`](../../../java-sdk/adr), distinct from the cross-cutting ADRs kept here. diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md new file mode 100644 index 0000000000000..305f9c7df6ac5 --- /dev/null +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -0,0 +1,200 @@ + + +# ADR-0001: Mixed-Lang Dag — Java Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. Java tasks should consume those +bindings directly instead of repeating upstream ids with `@Builder.XCom`. + +This ADR is scoped to the **Java call-site interface only** — the shape of the argument-binding +spec itself, how it is materialized, and how it travels over the wire is the protocol-level +decision recorded in +[`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). +This ADR only answers: given that spec, what does the Java code a user writes look like? + +## Decision + +The annotation is `@Builder.TaskHandler(dag, task)`, not `@Builder.Task(id)`. Python declares the +task with `@task.stub` and Java supplies only its body, so the two are different things and should +not share a name: `@Builder.Dag`/`@Builder.Task` stay with the native surface, where Java owns the +Dag ([ADR-0002](0002-native-dag-interface.md)). Python owns both ids too, so a handler names the pair +it binds to and the processor generates its registration from them. + +We support 3 authoring syntaxes for the TaskFlow binding. All three resolve to the same internal +`TaskArgs`, which is not part of the public surface: a Dag author never names it, and only generated +code reads it. + +### 1. Annotation based with positional injection + +```java +@Builder.TaskHandler(dag = "etl", task = "score") +public long score(Client client, long rows, double threshold, List regions) { ... } +``` + +Data parameters bind by position, and `Client`/`Context` are injected without consuming a position. +The processor compiles this into a `Task`: + +```java +// Generated by the Airflow Java SDK annotation processor. Do not edit. +public static final class Score implements Task { + @Override + public void execute(Context context, Client client) throws Exception { + TaskArgs args = TaskArgs.of(context); + long rows = args.require(0, Long.class); + double threshold = args.require(1, Double.class); + List regions = args.require(2, new TypeReference>() {}); + client.setXCom(new ExampleTasks().score(client, rows, threshold, regions)); + } +} +``` + +Each local carries its parameter's declared type, and `TaskArgs` converts by position — including a +generic type, which a cast cannot express. + +### 2. Annotation based with explicit struct + +```java +public static class ReportInput implements TaskInput { + @ArgName("run_label") + public String runLabel; + + public long transformed; +} + +@Builder.TaskHandler(dag = "etl", task = "report") +public void report(ReportInput input) { + log.log(INFO, "Report {0} for transformed value {1}", input.runLabel, input.transformed); + if (!"nightly".equals(input.runLabel)) { + throw new RuntimeException("expected run label 'nightly' but got " + input.runLabel); + } +} +``` + +### 3. Interface based with an explicit `TaskInput` + +```java +public static class SummarizeInput implements TaskInput { + @ArgName("region_code") + public String region; + + public long transformed; + } + + // summarize(region_code=..., transformed=...) is called with keyword + // arguments, which bind to the bundle's fields by name. + public static class Summarize implements InputTask { + public void execute(@NotNull Context context, Client client, SummarizeInput input) { + log.log( + INFO, "Summarize region {0} for transformed value {1}", input.region, input.transformed); + if (!"emea".equals(input.region)) { + throw new RuntimeException("expected region 'emea' but got " + input.region); + } + } + } +``` + +Defining a `TaskInput` is the only way an interface task receives bound values. `InputTask` +is not offered: `TaskArgs` is positional and untyped, which is right for generated code and wrong for +code a person writes and later reads. + +### Registering and serving + +A `Bundle` is what the server serves, and where handlers and Dags are registered — create it, +register, serve: + +```java +public static void main(String[] args) { + Bundle bundle = new Bundle() + .register(ScoreTasks.class) // annotation based: ids from the annotations + .register("etl", "summarize", Summarize.class) // interface based: ids at the call + .register(EtlPipeline.class); // native Dags share the bundle + + Server.create(args).serve(bundle); +} +``` + +One verb, overloaded, and everything registers as a class: nothing is constructed here. + +## Consequences + +- One binding contract serves both the annotation and interface authoring surfaces, because all + three syntaxes resolve to the same internal `TaskArgs`. +- Keeping `TaskArgs` internal means positional access stays available where it is cheap and safe — + in code the processor writes and type-checks — without asking a Dag author to track argument + order by hand. +- `@Builder.XCom` is retired; the Python call site is the only source of data-flow wiring. +- Renaming the shipped `@Builder.Task` to `@Builder.TaskHandler(dag, task)` for this surface + breaks existing mixed-language Dags, which change one annotation and name the Dag they belong to. +- `Client`/`Context` are injected as method arguments. This follows the same rule the native surface settles + ([ADR-0002](0002-native-dag-interface.md)), so one injection model spans both surfaces. + +## Appendix: Implementation Notes + +- Annotated task data parameters bind by position; injected `Client` and `Context` do not consume + positions. Positional binding also needs no parameter names in the class file, so it does not + depend on compiling with `-parameters`. +- Keyword arguments bind through one `TaskInput` bundle, with `@ArgName` only when Python and Java + names differ. +- Interface tasks use `InputTask` for named fields. There is no public positional form. +- `TaskArgs` is internal. It is what generated code reads, and what every syntax above resolves to; + nothing in the public API returns or accepts it. It exposes `require(index, type)`, which fails + when the position resolves to nothing, and `get(index, type)`, which yields `null` — each with a + `Class` and a `TypeReference` form. +- Annotation processing generates code, it never rewrites it, so the method a user writes stays + exactly as written and the generated `Task` delegates to it. That also sets the two constraints + the processor enforces at compile time: the annotated method has to be reachable from the + generated class in the same package, and its declaring class needs a no-arg constructor, since + the generated body calls `new Parent().method(...)`. +- A parameter whose type has type arguments binds through the `TypeReference` overload, so the + element type survives and the generated code never contains an unchecked cast. +- Missing reference or boxed values become `null`; primitive inputs fail clearly. +- `@Builder.XCom` is removed, keeping the Python call site as the single source of data-flow + wiring. +- **The generated `Task` is one more branch of an existing classifier.** `BuilderProcessor` already + compiles annotated methods into a `Task` implementation, classifying each parameter by type and + emitting `new Parent().method(...)`; positional binding adds a case to that classifier rather than + a mechanism. +- **Locals are emitted with the parameter's declared type, never `var`,** so the generated file reads + as the method signature it serves, and a binding that does not fit is a compile error in generated + code rather than a surprise at run time. +- **`TaskArgs` converts rather than casts,** through the `ObjectMapper` the SDK already uses. That is + what makes a generic parameter work: a cast cannot express `List`, and erasure would let + `(List) value` succeed over a list of numbers and fail much later at the first element + read. A `Class` overload covers simple types and a `TypeReference` overload carries the full + generic type, the processor choosing by whether the declared type has type arguments. Conversion + also subsumes numeric coercion, which the XCom path handles today by widening through `Number` by + hand, since the wire decodes integers to `Long` and floats to `Double`. `require` names the + parameter it failed on, so a missing value for a primitive fails clearly instead of surfacing as an + unboxing `NullPointerException`. +- **`register` is overloaded, not suffixed**: `register(Class)` for anything annotated, + `register(String, String, Class)` for an `InputTask` with no annotation to read, + and `register(Dag)` for a Dag built through the interface surface + ([ADR-0002](0002-native-dag-interface.md)). `register(ScoreTasks.class)` registers every handler + the class holds. It is this ADR's proposed addition: today's `Bundle` takes an `Iterable` + through `BundleBuilder.getDags()`, with no place for a handler that belongs to a Python-owned Dag. +- **Registration holds classes, never instances.** `Dag.addTask(id, Class)` already + states that Airflow "instantiates the class via its no-argument constructor, then calls `execute` + once per task-instance run", so an instance exists only for one task-instance run — never across + runs, and never at registration. diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md new file mode 100644 index 0000000000000..80c2d7101cdbd --- /dev/null +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -0,0 +1,348 @@ + + +# ADR-0002: Native Java Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so +Java itself must express the graph, the Dag/task configuration, and the task bodies. This ADR is +scoped to what that Java call site looks like for a user. It also settles the injectable +`client`/`context` question shared with [ADR-0001](0001-mixed-lang-dag-interface.md): they are +**injected as method arguments**, not exposed through getters. It shares the protocol substrate +(the argument-binding spec) with +[`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +### Annotation based + +`Client` and `Context` are injected as **method arguments**, exactly as on the mixed-language +surface ([ADR-0001](0001-mixed-lang-dag-interface.md)) — there are no getters and no SDK base class +to extend. A task method's signature is its injected arguments, if any, followed by its data: + +```java +@Builder.Dag(id = "java_etl", schedule = "@daily") +public class EtlPipeline { // extends nothing of ours; your own base class stays free + + @Builder.Task(id = "extract", retries = 2) + public long extract(Client client) { + return ((Number) client.getVariable("row_count")).longValue(); + } + + @Builder.Task(id = "transform") + public long transform(Client client, Context context, long extracted, double threshold) { + client.setXCom("scaled_in", context.runId); + return (long) (extracted * threshold); + } + + @Builder.Task(id = "load") + public void load(Context context, long transformed) { + log.log(INFO, "Loaded {0} for run {1}", transformed, context.runId); + } + + @Builder.Task(id = "audit") + public void audit(Client client) { /* side effect only, no data in or out */ } + + @Builder.Task(id = "notify") + public void notify(Client client) { /* side effect only, no data in or out */ } + + @Builder.Deps + static class Wiring implements EtlPipelineDeps { + void depends() { + // TaskFlow (data) edges: implied by passing a TaskRef; a constant is wrapped with lit(...) + var rows = extract(); + var scaled = transform(rows, lit(0.9)); + var loaded = load(scaled); + + // non-TaskFlow (ordering-only) edges: sequence with no data flowing + rows.then(audit()); // extract >> audit + Flow.of(loaded, audit()).then(notify()); // [load, audit] >> notify + } + } + + public static void main(String[] args) { + Bundle bundle = new Bundle().register(EtlPipeline.class); + Server.create(args).serve(bundle); + } +} +``` + +Because the injected arguments are real parameters, the wiring expression cannot call the task +methods directly — `extract(client)` would demand a `Client` the wiring has no business supplying. +So the processor generates one **wiring view** per Dag: the interface `Deps` (here +`EtlPipelineDeps`), one default method per task with the injected arguments stripped, each data +argument lifted to `Arg`, and the return lifted to `TaskRef`: + +```java +// Generated by the Airflow Java SDK annotation processor. Do not edit. +interface EtlPipelineDeps extends Deps { // Deps: the shared base that nests Flow + default TaskRef extract() { return Flow.node("extract"); } + default TaskRef transform(Arg extracted, Arg threshold) { return Flow.call("transform", extracted, threshold); } + default TaskRef load(Arg transformed) { return Flow.call("load", transformed); } + default TaskRef audit() { return Flow.node("audit"); } + default TaskRef notify() { return Flow.node("notify"); } +} +``` + +The `@Builder.Deps` class implements this interface, so `depends()` composes the wiring methods and +`javac` type-checks the graph — `extract()` yields a `TaskRef`, which is an `Arg` for +`transform`, and so on. The class only *implements an interface*, so its `extends` stays free, and +being `static` it cannot reach the outer class's real task methods — wiring can only touch the +generated views. + +`@Builder.Deps` is read by executing it once, in **recording mode**, at registration: each wiring +call records a memoized node (so `extract()` is the same `TaskRef` everywhere), passing a `TaskRef` +into another call records a data edge, and the ordering verbs below record ordering edges. No task +body runs, and — because a `TaskRef` carries the node's identity as an object reference — the +processor never reads the method's syntax tree, so this works under any compiler, not only javac. + +The result is the Dag below, equivalent to the interface-based form written by hand: + +```java +// Generated by the Airflow Java SDK annotation processor. Do not edit. +public final class EtlPipeline_Dag { + + public static Dag dag() { + Dag dag = new Dag("java_etl").config("schedule", "@daily"); + TaskRef extract = dag.task("extract", Extract.class).config("retries", 2); + TaskRef transform = dag.task("transform", Transform.class); + TaskRef load = dag.task("load", Load.class); + TaskRef audit = dag.task("audit", Audit.class); + TaskRef notify = dag.task("notify", Notify.class); + extract.then(transform).then(load); // data path + extract.then(audit); // ordering-only edge + Flow.of(load, audit).then(notify); // ordering-only edge + return dag; + } + + public static final class Transform implements Task { + @Override + public void execute(Context context, Client client) throws Exception { + TaskArgs args = TaskArgs.of(context); + long extracted = args.require(0, Long.class); + double threshold = 0.9; // baked from lit(0.9) at Dag-build time + client.setXCom(new EtlPipeline().transform(client, context, extracted, threshold)); + } + } + + // Extract, Load, Audit, and Notify follow the same shape. +} +``` + +The injected arguments are passed straight into the user method, and the data arguments bind by +position through the same internal `TaskArgs` the mixed-language surface uses +([ADR-0001](0001-mixed-lang-dag-interface.md)) — no getters, no `bind()`, nothing ambient. + +**Literals.** A data argument is an `Arg`, which a `TaskRef` satisfies; a constant is wrapped +with `lit(...)` — `transform(extract(), lit(0.9))` — and recorded as a baked value with no edge. +Wrapping is required because a bare `Integer` cannot implement `Arg`; boxed types only, no +primitives. + +### Non-data dependencies + +A data edge comes for free from passing a `TaskRef` into another wiring method. A dependency +where no data flows, similar to Python's `a >> b`, is expressed with `then`, a variadic verb +that every `TaskRef` carries. It lives on a small `Flow` interface, and a `TaskRef` is a `Flow` +of one (`interface TaskRef extends Arg, Flow`): + +```java +a.then(b); // a >> b +a.then(b, c); // a >> [b, c] +``` + +`then` returns the **new frontier** (the set it just pointed at), the way `>>` evaluates to its +right operand, so a chain walks through a fan: + +```java +a.then(b, c).then(d); // a >> [b, c] >> d (a->b, a->c, then b->d, c->d) +``` + +The one thing `then` cannot do is start from a *set*: Java can't overload `>>` the way Python +does, and there is no list literal to call `.then` on, so `Flow.of` opens a chain from one: + +```java +Flow.of(a, b).then(c); // [a, b] >> c +Flow.of(a, b).then(c, d); // [a, b] >> [c, d] +``` + +Both edge kinds compile to the same `then(...)` in the generated Dag above. The only difference is +that a data edge also binds an argument while an ordering-only edge binds nothing. + +### Interface based + +```java +Dag dag = new Dag("java_etl").config("schedule", "@daily"); + +TaskRef extract = dag.task("extract", Extract.class).config("retries", 2); +TaskRef transform = dag.task("transform", Transform.class); +TaskRef load = dag.task("load", Load.class); +TaskRef notify = dag.task("notify", Notify.class); + +extract.then(transform).then(load).then(notify); +``` + +`dag.task(id, class)` registers the task as it creates it and hands back a `TaskRef`, so there is no +second `addTask(...)` call to forget. `then` is the same variadic edge verb as the +annotation surface — Java's spelling of Python's `>>` — and `.config(key, value)` carries +Dag and task configuration. This surface wires edges through object references and reads no syntax +tree, so it works under any toolchain and is exactly what the annotation surface's recording +produces. + +### Registering and serving + +Both surfaces produce a `Dag`, and a `Bundle` is what the server serves — create it, register, serve: + +```java +public static void main(String[] args) { + Bundle bundle = new Bundle() + .register(EtlPipeline.class) // annotation based: the user's own annotated class + .register(dag); // interface based: the Dag built above + + Server.create(args).serve(bundle); +} +``` + +An annotated Dag registers as the class the user wrote — the same class the annotations are on — so +there is no second name to learn or keep in sync. + +Nothing is constructed at registration. Task classes are instantiated per task-instance +run ([ADR-0001](0001-mixed-lang-dag-interface.md)), and the Dag definition itself is read once, at +registration, without running a task body. + +One bundle carries native Dags and mixed-language task handlers alike +([ADR-0001](0001-mixed-lang-dag-interface.md)), so one process serves both. + +## Alternatives + +- **Getters for `Client`/`Context`** (`getClient()`/`getContext()` on an SDK base class the Dag + extends), so a task method takes only its data and `depends()` composes the real signatures + directly, with no wiring view. Rejected: the team does not want ambient getters — they invite the + "reach for the context anywhere" pattern we already avoid in Python — and a base class spends the + Dag class's single `extends` on the SDK, so the user can no longer extend their own. Injected + arguments keep the getter out of the language and the Dag class's inheritance free; the wiring view + is the price, and it is generated rather than written. + +- **Reading the `@Wiring` body's syntax tree** instead of executing it. With getters the task + methods are data-only, so `depends()` composes real signatures and a processor could walk the tree + and emit one edge per nested call. Rejected on two counts: it depends on + `com.sun.source.util.Trees`, javac's tree model, so it does not run under a compiler with its own + (ecj); and it must recover identity syntactically — a result held in a local + (`long e = extract(); fanin(e, ...)`) is an identifier, not a call, so the walker has to resolve it + through the initializer and bail on anything it cannot attribute. Recording sidesteps both: nothing + parses source, and a reused local just works. + +- **Capturing the graph by overriding task methods in a generated subclass**, so an inherited + `depends()` calls the overrides. Rejected because a raw return (`long`) carries no identity, so a + result stored in a local and used twice cannot be attributed to its producer. Recording avoids this + precisely by having the wiring view return `TaskRef` rather than the task's real return type — + identity travels with the handle, so this is not the same mechanism as the subclass capture. + +- **Naming the type `Flow`, over `Order` and `Chain`.** The type every `TaskRef` carries, and that + `Flow.of` returns, needed a name. `Chain` was rejected because Airflow's Python `chain()` is a + *sequential* helper (`chain(a, b)` means `a >> b`), so `Chain.of(a, b)` would read as the edge + `a -> b` rather than the parallel set `{a, b}`. `Order` reads well for pure sequencing, but misnames + a group that can carry data: a chain like `Flow.of(a(), b(c())).then(d(e()))` also records the data + edges `c -> b` and `e -> d`, so the group is tasks *in the flow*, not an ordering. It shares a + name with `java.util.concurrent.Flow`, but nesting defuses that: `Flow` is a member type of the + shared `Deps` base every `Deps` extends, so wiring code inherits it by simple name and never + imports it (see Implementation Notes). + +- **Only `then`, not Python's fuller set of relation helpers.** Python offers `>>` and `<<`, + `chain()`, etc. This surface has only one forward verb, plus `Flow.of`. A single left-to-right verb reads + consistently, and is less to learn. Nothing is lost; the reverse direction and the shorthands can still be + properly expressed. + +## Consequences + +- Two authoring surfaces (annotation, interface) flow into the same `Dag`/`TaskRef` model, and the + annotation surface's recording produces exactly the interface-based calls a user could have + written. +- `Client`/`Context` are injected as arguments on both the native and mixed-language surfaces, so a + single injection rule spans the SDK and no getter or SDK base class exists. +- The wiring view (`Deps`) is generated, so the graph is real, type-checked Java — but the + interface name resolves in an IDE only after the first build that runs the processor. This is the + standard annotation-processing tradeoff (Dagger, AutoValue, MapStruct) and needs the processor + wired into the build so generated sources are indexed. +- Recording reads no syntax tree, so the annotation surface works under any compiler, not only javac. +- A task method keeps its real signature — injected arguments then data — and the only generated + user-facing type is the wiring-view interface it implements: no twin class, no `In`, no getters, + no `bind()`. +- Native tasks keep `@Builder.Task`, while the mixed-language surface uses + `@Builder.TaskHandler` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names which + of the two a method is. + +## Appendix: Implementation Notes + +- The processor generates, per Dag, the `Deps` wiring-view interface and a registrar. The + registrar instantiates the `@Builder.Deps` class, installs a recorder, calls `depends()`, collects + the graph, and validates it (no cycles, no self-edges), reporting a bad edge on its own source line + through `Messager`. It reads no syntax tree. +- The wiring view strips a parameter only when its type is `Client` or `Context`; every other + parameter becomes an `Arg` in declared order, and the return becomes `TaskRef` + (`TaskRef` for `void`). +- **The nested `@Builder.Deps` class is required, not stylistic.** The Dag class holds + `long extract(Client)` and the wiring view holds `TaskRef extract()` — two methods that would + be an overload clash in one class. A separate type carries the view, and it `implements` (never + `extends`) so the Dag class keeps its single inheritance slot. +- Wiring methods are memoized by task id, so `extract()` returns the same `TaskRef` wherever it is + called; that identity is what lets a data edge and a later ordering edge refer to one node. +- **`then` is variadic and returns the new frontier**, so `a.then(b, c).then(d)` walks through a + fan, and linear runs, fan-outs, and fan-ins all fall out of the one verb. `Flow.of(...)` is the + only entry point beyond it — it opens a chain from a set, which `then` cannot. A data edge never + needs any of this — it is implied by passing a `TaskRef`. +- **`Flow` is a nested member of a shared base, not a top-level type.** Every generated `Deps` + extends one library interface, `Deps`, that declares the nested interface `Flow`. Member types are + inherited, so a `@Builder.Deps` class implementing `Deps` names `Flow` (as in `Flow.of(...)`) + by simple name with no import — so there is nothing for `java.util.concurrent.Flow` to clash with. + A single-type-import does not override this: a visible member type outranks a single-type-import in + simple-name resolution (JLS §6.5.5.1; and §6.4.1, whose single-type-import shadowing list covers + only top-level types in other units and on-demand imports, not member types), so even an explicit + `import java.util.concurrent.Flow` in the same file still resolves `Flow` to the inherited member. + The imperative interface-based surface does not extend `Deps`, so it qualifies the factory as + `Deps.Flow.of(...)` — or, since it always has named refs, skips it entirely, writing a fan-in as + `x.then(z); y.then(z)`. +- **`Arg` is the data-argument type in a wiring method**; `TaskRef extends Arg`, and + `lit(value)` wraps a boxed constant. A `TaskRef` argument records an edge; a `lit(...)` records a + baked value with none. Because both flow through `Arg` and the recorder distinguishes them by type, + there is no ambiguity even at an `Object`-typed position. +- **At run time the getters are gone.** The generated `Task` passes the injected `Client`/`Context` + straight into the user method and binds data arguments by position through `TaskArgs` + ([ADR-0001](0001-mixed-lang-dag-interface.md)) — the same conversion-not-cast path, including the + `TypeReference` overload for generics. Nothing is ambient, so the SDK's Java 11 pin (no + `ScopedValue`) does not matter here. +- **`dag.task(...)`, `.config(...)`, and `.then(...)` are proposed additions.** + Today's shipped `Dag` has only `addTask(id, definition)`, which returns the Dag rather than a ref; + `.config(...)` is keyed to the Dag serialization schema. `dag.task(...)` is a factory method, not a + constructor — Java spells qualified inner-class creation `dag.new TaskRef(...)`, which reads as + compiler trivia rather than as a Dag definition. +- **`register` is one overloaded verb**: `register(Class)` for an annotated Dag class, + `register(Dag)` for a built one, and `register(dag, task, Class)` for a task + handler ([ADR-0001](0001-mixed-lang-dag-interface.md)). Today a bundle is built from an + `Iterable` through `BundleBuilder.getDags()`. +- **Registration holds classes, never instances.** `Dag.addTask(id, Class)` already + states that Airflow "instantiates the class via its no-argument constructor, then calls `execute` + once per task-instance run", so an instance exists only for one task-instance run — never across + runs, and never at registration. +- Testing a task is testing an ordinary method with a real return type whose injected arguments are + passed in directly — no base class to subclass, no getters to override, no `In` to unwrap.