Skip to content

Add task-handler parse messages and their parse-channel union - #73973

Draft
jason810496 wants to merge 43 commits into
mainfrom
jason/core-taskhandler-refactor/04-task-handler-parse-messages
Draft

jason810496 wants to merge 43 commits into
mainfrom
jason/core-taskhandler-refactor/04-task-handler-parse-messages

Conversation

@jason810496

@jason810496 jason810496 commented Sep 30, 2026 •

Copy link
Copy Markdown
Member

Stack (bottom to top), on native #74037 (stack #74170): #73973, #73974, #73975, #74317, #73976, #74067, #74135, #74140, #74141, #73971, #74030, #74031, #74032, #74136, #74137, #74138, #74139

To check each stub task against the task handler its artifact registers (ADR-0011), the Dag processor has to ask a Lang-SDK runtime which handlers an artifact registers. This PR adds the messages for that (ADR-0012). Nothing sends them yet: #73974 adds the probe.

TaskHandlerParseRequest(  # Dag processor -> runtime
    file="/bundles/go-task-handlers/etl",
    bundle_path=Path("/bundles/go-task-handlers"),
    bundle_name="go-task-handlers",
)

TaskHandlerParsingResult(  # runtime -> Dag processor
    fileloc="/bundles/go-task-handlers/etl",
    task_handlers={  # every handler the artifact registers, keyed by Dag id
        "etl": [
            TaskHandlerDeclaration(
                task_id="extract",
                binding="named",
                params=[TaskHandlerParam(name="day", value_schema={"type": "string"}, exact_name=True)],
            ),
            TaskHandlerDeclaration(
                task_id="load",
                binding="positional",
                params=[TaskHandlerParam(name=None, value_schema={"type": "string"})],
            ),
        ],
    },
)

# A runtime that cannot list a handler's params, such as TypeScript
TaskHandlerDeclaration(task_id="report", binding="named", params=None)
  • The request names no Dags. The runtime answers with every handler the artifact registers, and the answer depends only on the artifact, so one probe serves every Dag whose stub tasks resolve to it.
  • Each SDK binds stub-task arguments differently, so every declaration carries a required binding:
    • positional: by position (Go flat params, Java TaskArgs). A count or type mismatch fails the Dag file's import.
    • named: by name in any order, ignoring case and _ unless exact_name is set (Go structs, Java TaskInput). A type mismatch fails the import, while an argument or param that matches nothing is only a warning.
  • params=None means the runtime cannot list the handler's params (TypeScript, whose types are erased), so only the handler's presence is checked.
  • value_schema reuses ArgValueSchema, which stub tasks' arg_bindings already use, so the check compares one type.
  • ToManager gains TaskHandlerParsingResult with no shared handler: only the probe class in Probe a Lang-SDK artifact for its task handlers #73974 handles it, and a Dag parse answers it as an unhandled request.
  • The supervisor schema version does not move, because a brand-new message needs no VersionChange. The TS SDK types and the Go SDK models are regenerated, and the Java copy is refreshed in Java SDK: answer TaskHandlerParseRequest with registered handlers #74317.
  • ADR-0011, ADR-0012 and ADR-0013 describe the messages as built.

Was generative AI tooling used to co-author this PR?
  • Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

@jason810496
jason810496 added this pull request to stack #73978 September 30, 2026 17:59
@jason810496 jason810496 changed the title jason/core taskhandler refactor/04 task handler parse messages Add task-handler parse messages and their parse-channel union Sep 30, 2026
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch 2 times, most recently from 545f056 to 6b23ddc Compare October 1, 2026 06:30
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch from 6b23ddc to 020bdcc Compare October 1, 2026 10:55
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch from 020bdcc to b88bd4f Compare October 1, 2026 14:44
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch from 48f58db to 017a401 Compare October 2, 2026 05:48
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch from 017a401 to 07b5199 Compare October 2, 2026 11:36
The Dag processor and the CLI only run the JAR, so they need a JRE, as
the prerequisites already say. The artifact bundle tests go back to
main's MagicMock(spec=BundleInfo), which they no longer needed to drop.
A NodeCoordinator configured without bundles_root now hands out a Dag
importer, so the Dag bundles it serves have their *.min.mjs bundles parsed
by running them with node. Only files that start with the airflow-ts-pack
layout header are parsed. The Dag source shown for such a Dag is the
bundle's entry module, as the bundle embeds it, or a short notice when
there is none to show.
Describe the coordinator configuration that parses native Dags, what the
Dag processor needs for it, what the Code view shows, the cost of
handler-only bundles, and that cluster policies and some CLI commands do
not apply to a native Dag.
A task used to run the first sorted *.min.mjs bundle that declares its
dag_id, which can differ from the file the Dag was parsed from. When the
task's Dag file is a bundle under the scan roots that still declares the
Dag, run that file. Otherwise fall back to the search.
The Dag processor runs node on every packed bundle a Node coordinator
serves, handler-only ones included, not only on bundles that declare
TypeScript Dags.
The build section said each Dag's own file is embedded so Airflow can
display it, but the Code view shows the entry module for every Dag in
the bundle. Say what is embedded and what the Code view shows today.
A native TypeScript Dag and a Python stub with the same dag_id in one
bundle overwrite each other's Dag on every parse. Tell users to remove
the stub when they move a Dag to new Dag(...), as java.rst does.
The importer only imports the coordinator for type checking, so there
is no cycle. This matches how JavaCoordinator imports JavaDagImporter.
Both callers of get_source_code already catch errors and store a
placeholder, as they do for the Java and Python importers. Raising
reports a broken bundle in one place with one wording.
The tests used a Dag's own file as the entrypoint, so a resolver that
returned the first dag_source_paths entry would still pass. Use a
separate main.ts entrypoint instead.
The importer names the NodeCoordinator class that parses its files and is registered in a Dag bundle whenever a NodeCoordinator is configured. The coordinator no longer hands it out or picks the task's bundle itself, so a task scans the bundle for its Dag as before.
NodeCoordinator builds the command for the bundle its Dag was parsed from with _build_dag_file_command. The runtime reports a Dag the bundle does not declare, so the coordinator does not check the Dag id.
Once a NodeCoordinator is configured, every Dag bundle is parsed. With several, dag_bundle_to_coordinator picks one per Dag bundle. Artifact-only Dag bundles list * in .airflowignore, and a native task runs on the coordinator its queue routes to.
airflow tasks list parses a native TypeScript Dag, so it is not among
the commands that refuse one. Also break the new TypeScript SDK text
only at sentence or clause ends.
The Dag processor will ask a Lang-SDK runtime which task handlers an artifact registers, so a later change can check stub tasks against them (ADR-0011, ADR-0012). This adds the request, the result and the declarations it carries, and ToSDKTaskHandlerProcessor, which shares every response with ToDagProcessor. ToManager gains the result, but nothing handles it yet, so DagFileProcessorProcess and LangSDKDagFileProcessorProcess answer it as an unhandled request.

The schema registry introspects the new union, so the snapshot gains four definitions; new bodies need no VersionChange. params has its own title because go-jsonschema names a nullable list's type by its title, and Params is already the HITL params map.
check-ts-sdk-supervisor-schema regenerates supervisor.ts from the schema snapshot and fails on any difference. CI runs it with --all-files whenever a ts-sdk file changes, so a stale copy would fail the next ts-sdk PR. The Type<N> renumbering is the generator's own output.
go-sdk generates its coordinator-protocol models from its vendored copy of the schema snapshot. This refreshes the copy with sync-go-sdk-schemas and regenerates the models, so the Go SDK can decode the request and build the result, and check-go-sdk-generated-drift passes. go-jsonschema moves some type blocks, so the diff is larger than the new types.
AGENTS.md said a new body needs no VersionChange but not why, and check-supervisor-schemas-versions still fails a local commit that changes the snapshot without touching versions/. The next author of a new body could not tell that failure from a real one.
…ADR-0012

ADR-0012 still sketched a request that names Dags, params compared by name in order, and a required flag, and ADR-0011's query step sent those Dag ids. Both now match the messages: the request names no Dags, the answer has every handler the artifact registers, and each declaration names how its arguments bind.
ADR-0013 called the messages SDKTaskHandlerParseRequest and SDKTaskHandlerParsingResult and sketched their earlier fields. It now uses the names and fields the code defines.
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch from 07b5199 to a6ba6e2 Compare October 5, 2026 18:12
@jason810496
jason810496 removed this pull request from stack #73978 October 6, 2026 02:27
@jason810496
jason810496 changed the base branch from jason/core-taskhandler-refactor/03-parse-process-base to jason/lang-sdk-e2e/05-node-dag-importer October 6, 2026 02:30
@jason810496
jason810496 added this pull request to stack #74318 October 6, 2026 02:35
@jason810496
jason810496 force-pushed the jason/core-taskhandler-refactor/04-task-handler-parse-messages branch from a6ba6e2 to 1b43f7b Compare October 6, 2026 02:44
Base automatically changed from jason/lang-sdk-e2e/05-node-dag-importer to main October 6, 2026 15:13

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant