You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Commit 0220ce7
Browse filesBrowse the repository at this point in the historyBrowse files
Lead each Lang-SDK ADR with its diagrams and state the decisions plainly
The diagrams carry the decision faster than the prose around them did, so they
belong where a reader lands rather than behind an appendix. The argument for
each choice is still worth keeping, but it is what someone reads second.
Coordinator lookup gains a named answer: for_bundle, beside the existing
for_queue, rather than an open question about how a registry finds the
coordinators serving a bundle.
The Dag processor needs two different answers from a Lang-SDK runtime, and they
29
-
are not variations of one question:
28
+
The Dag processor asks a Lang-SDK runtime two different questions. "Which Dags does this artifact define?" is answered by the runtime standing in for the parser process, over the
29
+
messages [ADR-0004](0004-dag-parsing.md) already defines. "Which task handlers does this artifact register for a `dag_id` Python already owns?" has no answer in those messages,
30
+
because a `TaskHandler` registration carries no Dag ([ADR-0010](0010-mixed-language-dag-processing.md)).
30
31
31
-
-**Which Dags does this artifact define?** Answered by the runtime standing in
32
-
for the parser process, over the manager ↔ parser messages
33
-
[ADR-0004](0004-dag-parsing.md) already defines.
34
-
-**Which task handlers does this artifact register for a `dag_id` Python
35
-
already owns?** A `TaskHandler` registration carries no Dag
36
-
([ADR-0010](0010-mixed-language-dag-processing.md)), so the answer is not a
37
-
serialized Dag and there is nothing in the existing messages to carry it.
32
+
This ADR defines the channel and payload for the second question, and names both parse-side entry points on the coordinator.
38
33
39
-
This ADR defines the channel and payload for the second question, and names
40
-
both parse-side entry points on the coordinator.
41
-
42
-
Terms follow the Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`, spec
43
-
version `1.0`): `Dag`, `TaskHandler`, `DagRef`, `TaskHandlerRef`, `bundle`,
44
-
`register`, `serve`.
34
+
Terms follow the Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`, spec version `1.0`).
45
35
46
36
## Decision
47
37
48
-
### Handler Declarations Travel on Their Own Channel
49
-
50
-
Handler declarations are carried by a dedicated message pair in their own
51
-
discriminated unions — `ToRuntime` / `ToCoordinator` — not by a field added to
52
-
`DagFileParseRequest`.
53
-
54
-
Union membership is the only registration step in the supervisor schema
55
-
package: it is what puts a body into the generated snapshot, and therefore into
56
-
every SDK's generated models. So the choice of union *is* the wire-contract
57
-
decision. Neither existing pair fits, because the runtime plays two roles. When
58
-
it parses native Dags it **replaces** the parser process and answers the
59
-
manager, which the manager ↔ parser unions already describe. When it answers a
60
-
handler query it is a short-lived child of the Python parser, and the manager is
61
-
not a peer at all. Reusing the parser unions there would make one union mean two
62
-
recipients.
63
-
64
-
The payload is deliberately a stronger contract than the one it parallels.
65
-
`DagFileParsingResult.serialized_dags` is `list[LazyDeserializedDAG]`, which
66
-
reduces in the snapshot to an opaque object — which is why every SDK
67
-
reimplements DagSerialization by hand and validates it against shared fixtures
68
-
([ADR-0004](0004-dag-parsing.md)). A handler declaration carries no Dag, so it
69
-
is fully typed, code-generated, and schema-validated in every SDK. **The
70
-
mixed-language path never asks a Lang SDK to implement DagSerialization at
71
-
all.**
72
-
73
-
Parameter schemas reuse the existing `ArgValueSchema` definition that
74
-
`arg_bindings` already carries
75
-
([ADR-0007](0007-taskflow-across-language-boundary.md)), so validation compares
76
-
like against like rather than translating between two vocabularies. Appendix A
77
-
gives the message definitions and the two properties of that field —
78
-
nullability and ordering — that validation has to respect.
79
-
80
-
### Two Parse Verbs, Because There Are Two Conversations
81
-
82
-
`BaseCoordinator` gains `parse_dag` and `parse_task_handler` alongside the
83
-
shipped `execute_task`. They are **not symmetric siblings**, which is the
84
-
reason not to collapse them into one `parse(request)`:
85
-
86
-
-`parse_dag` is **bridge mode**. The coordinator becomes the parser process
87
-
and raw-forwards the manager's request and the runtime's reply. It never
88
-
reads the payload.
89
-
-`parse_task_handler` is **query mode**. The coordinator is itself the peer:
90
-
it sends one request and decodes one reply on the channel above.
91
-
92
-
One collapsed method would also erase a real deployment distinction — a
93
-
coordinator can serve mixed-language handlers with no interest in native Dag
94
-
parsing, and an absent method says so better than a runtime rejection.
95
-
Appendix B gives the layering and the command hooks.
96
-
97
-
## Consequences
98
-
99
-
-**The handler channel is typed where the Dag channel is opaque.** Every SDK
100
-
gets generated models for the declaration shape, and nothing on this path
101
-
requires a DagSerialization implementation.
102
-
- Each Lang-SDK runtime implements a second request type rather than a new field
103
-
on the existing one. That is the cost of handlers not being Dags: a `DagRef`
104
-
and a `TaskHandlerRef` have different shapes, so one request/result pair could
105
-
not have carried both.
106
-
- Adding `ToRuntime` / `ToCoordinator` extends the set of unions the supervisor
107
-
schema package introspects, and regenerating the snapshot is what propagates
`parse_dag` forwards bytes and never reads the payload. `parse_task_handler` is itself the peer: it sends one request and decodes one reply. They are two methods, not one
55
+
`parse(request)`, because a coordinator can serve handlers without serving native Dag parsing. Appendix A has the longer argument.
Not a field on `DagFileParseRequest`. Union membership is the only registration step in the supervisor schema package, so picking the union is the wire-contract decision. The
65
+
existing pairs describe a runtime answering the manager; here the runtime answers the coordinator.
66
+
67
+
### Message shapes
68
+
137
69
```
138
70
class TaskHandlerParseRequest:
139
-
file: str # the artifact the coordinator resolved
140
-
dag_id: str # scope: handlers bound to this dag_id only
71
+
file: str # the artifact the coordinator resolved
72
+
dag_id: str # scope: handlers bound to this dag_id only
141
73
type: Literal["TaskHandlerParseRequest"]
142
74
143
75
class TaskHandlerParsingResult:
@@ -158,30 +90,13 @@ class TaskHandlerParam:
158
90
required: bool # the handler declares no default
159
91
```
160
92
161
-
Two properties of `value_schema` are load-bearing for validation:
162
-
163
-
-**It is nullable on both sides.** An unannotated `@task.stub` parameter
164
-
produces `value_schema: null` today, so validation compares schemas only
165
-
where neither side is null and falls back to name-and-arity otherwise.
166
-
Without that, every untyped stub argument becomes a parse error.
167
-
-**`params` is ordered.**`LiteralArgBinding` and `XComArgBinding` are each
168
-
documented as "one positional stub-task argument", so position is part of the
169
-
contract rather than incidental.
170
-
171
-
A declaration carries no class, method, or source location.
172
-
[ADR-0006](0006-no-lang-sdk-source-display.md) rules out Lang-SDK source
173
-
display, and putting it on the wire would invite a consumer to render it.
93
+
`value_schema` reuses the `ArgValueSchema` definition `arg_bindings` already carries ([ADR-0007](0007-taskflow-across-language-boundary.md)), so both sides of a comparison are the
94
+
same type. Two properties matter to validation: the field is nullable on both sides, and `params` is ordered. Appendix B says what that forces.
174
95
175
-
Registration mechanics: `registered_models_by_name()` in
176
-
`task-sdk/src/airflow/sdk/execution_time/schema/` introspects a fixed set of
177
-
unions — `ToTask` / `ToSupervisor` and `ToManager` / `ToDagProcessor` — so
178
-
adding `ToRuntime` / `ToCoordinator` extends that set. Two prek hooks guard the
179
-
snapshot, and their interaction on a first-introduction body needs checking
180
-
against the hooks rather than against that package's `AGENTS.md`, which says no
181
-
`VersionChange` is required while the second hook fails when the snapshot moves
182
-
with nothing under `versions/` touched.
96
+
`task_handlers` is the counterpart to `DagFileParsingResult.serialized_dags`, but fully typed. `serialized_dags` is `list[LazyDeserializedDAG]`, which is an opaque object in the
97
+
schema snapshot. A handler declaration carries no Dag, so it code-generates and schema-validates in every SDK, and nothing on this path needs a DagSerialization implementation.
supply the three commands; no socket or protocol code
201
116
```
202
117
203
-
Each hook returns its own `subprocess_schema_version`, so handler parsing
204
-
negotiates the supervisor schema through the mechanism task execution already
205
-
uses, and `None` disables migration exactly as it does today.
206
-
`can_handle_dag_file` still gates native Dag parsing only; handler queries
207
-
resolve through the coordinator registry instead.
208
-
209
-
Artifact roots are resolved per mode by `_init_root_source`, but published
210
-
through `_get_scan_roots()`, which is scoped to an active task and raises
211
-
outside one. Both parse-side commands need the same roots with no
212
-
`TaskInstance` in hand, so the scope that publishes them has to open for a
213
-
parse as well as for a task. [ADR-0009](0009-native-dag-processing.md) covers
214
-
the modes themselves.
118
+
Names follow the shipped `execute_task` / `_build_execute_task_command` pair and supersede ADR-0004's `run_dag_parsing` / `dag_parsing_cmd`. Each hook returns its own
119
+
`subprocess_schema_version`, so handler parsing negotiates the schema the same way task execution does.
120
+
121
+
## Consequences
122
+
123
+
- The handler channel is typed; the Dag channel is opaque. Every SDK gets generated models for the declaration shape.
124
+
- Each runtime implements a second request type instead of a new field on the existing one. A `DagRef` and a `TaskHandlerRef` have different shapes, so one request/result pair
125
+
could not carry both.
126
+
-`ToRuntime` / `ToCoordinator` extends the set of unions the supervisor schema package introspects. Regenerating the snapshot is what propagates the bodies into each SDK's
127
+
generated models.
128
+
-`can_handle_dag_file` still gates native Dag parsing only. Handler queries resolve through the coordinator registry.
129
+
- Terms track Language SDK spec `1.0`. A spec rename of `TaskHandler` lands here too.
-[ADR-0009](0009-native-dag-processing.md) — who calls `parse_dag`
137
+
-[ADR-0010](0010-mixed-language-dag-processing.md) — who calls `parse_task_handler`, and what it compares the reply against
138
+
- Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`)
139
+
-`task-sdk/src/airflow/sdk/execution_time/schema/` — supervisor schema version bundle, union registry, generated snapshot
140
+
-[AIP-108](https://cwiki.apache.org/confluence/x/pY4mGQ) — Language SDKs
141
+
142
+
## Appendix
143
+
144
+
### Appendix A — Why two verbs and a separate channel
145
+
146
+
The runtime plays two roles, and they differ in who it is talking to. Parsing native Dags, it replaces the parser process and answers the manager, which is exactly what the manager
147
+
↔ parser unions describe. Answering a handler query, it is a short-lived child of the Python parser, and the manager is not a peer at all. Reusing the parser unions for both would
148
+
make one union mean two different recipients, and a runtime could not tell from the union which role it was in.
149
+
150
+
The same split is why the verbs stay separate. Collapsing them into one `parse(request)` would hide a real deployment distinction: a coordinator can serve mixed-language handlers
151
+
with no interest in native Dag parsing, and an absent method states that better than a runtime rejection does. The two also differ in mechanism — one is a byte forwarder that never
152
+
decodes the payload, the other decodes a reply it asked for — so the shared code is the subprocess plumbing underneath, not the public method.
153
+
154
+
A boolean on `DagFileParseRequest` was the alternative. It cannot work: a `DagRef` and a `TaskHandlerRef` are different payloads, not two subsets of one, so the flag would select
155
+
between shapes the result type cannot both hold.
156
+
157
+
### Appendix B — What the nullable, ordered parameter list forces
158
+
159
+
`value_schema` is nullable on both sides. An unannotated `@task.stub` parameter produces `value_schema: null` today, so validation compares schemas only where neither side is null,
160
+
and falls back to name-and-arity otherwise. A strict comparison would turn every untyped stub argument into a parse error.
161
+
162
+
`params` is ordered because `LiteralArgBinding` and `XComArgBinding` are each documented as "one positional stub-task argument". Position is part of the contract, not incidental,
163
+
and both sides bind positionally.
164
+
165
+
A declaration carries no class, method, or source location. [ADR-0006](0006-no-lang-sdk-source-display.md) rules out Lang-SDK source display, and putting it on the wire would
166
+
invite a consumer to render it.
167
+
168
+
### Appendix C — Schema registration mechanics
169
+
170
+
`registered_models_by_name()` in `task-sdk/src/airflow/sdk/execution_time/schema/` introspects a fixed set of unions — `ToTask` / `ToSupervisor` and `ToManager` / `ToDagProcessor`
171
+
— so adding `ToRuntime` / `ToCoordinator` extends that set.
172
+
173
+
Two prek hooks guard the generated snapshot. Their interaction on a first-introduction body needs checking against the hooks rather than against that package's `AGENTS.md`: the doc
174
+
says no `VersionChange` is required for a new body, while `check-supervisor-schemas-versions` fails when the snapshot moves and nothing under `versions/` was touched.
175
+
176
+
Artifact roots are resolved per mode by `_init_root_source`, but published through `_get_scan_roots()`, which is scoped to an active task and raises outside one. Both parse-side
177
+
commands need those roots with no `TaskInstance` in hand, so the scope that publishes them has to open for a parse as well as for a task. [ADR-0009](0009-native-dag-processing.md)
0 commit comments