[PR] Add task-handler parse messages and their parse-channel union [airflow]

["jason810496 (via GitHub)" <gi...@apache.org>]

jason810496 opened a new pull request, #73973:
URL: https://github.com/apache/airflow/pull/73973

   Stack (bottom to top): #73970, #73971, #73972, **#73973**, #73974, #73975, #73976, #73977
   
   ## Why
   
   The Dag-file parse will ask a Lang-SDK runtime which task handlers an artifact registers for Dags that Python already owns ([ADR-0011](https://github.com/apache/airflow/blob/main/airflow-core/adr/lang-sdk/0011-mixed-language-dag-processing.md)). This adds the messages for that question, shaped as [ADR-0012](https://github.com/apache/airflow/blob/main/airflow-core/adr/lang-sdk/0012-lang-sdk-parse-protocol.md) "Message shapes" defines them. Each SDK binds stub-task arguments to its handler differently, so each declaration says how its arguments bind and the later stub-to-handler check follows that. Nothing sends or receives these messages yet; the next layer adds the transport and `parse_task_handler`.
   
   ## What changes
   
   ```python
   TaskHandlerParseRequest(  # parent -> runtime, on ToSDKTaskHandlerProcessor
       file="/bundles/go-task-handlers/etl",
       dag_ids=["etl"],
       bundle_path=Path("/bundles/go-task-handlers"),
       bundle_name="go-task-handlers",
   )
   
   TaskHandlerParsingResult(  # runtime -> parent, on ToManager
       fileloc="/bundles/go-task-handlers/etl",
       task_handlers={
           "etl": [
               TaskHandlerDeclaration(  # Go struct with arg: tags
                   task_id="extract",
                   binding="named",  # "positional" | "named" | "named_or_whole" | "named_open"
                   params=[TaskHandlerParam(name="day", value_schema={"type": "string"}, required=False, exact_name=True)],
               ),
               TaskHandlerDeclaration(  # Go flat params
                   task_id="load",
                   binding="positional",
                   params=[TaskHandlerParam(name=None, value_schema={"type": "string"}, required=True)],
               ),
           ],
       },
   )
   ```
   
   `binding` is required and says how stub-task arguments match `params`:
   
   - `positional`: by position. `name` is informative only, and `None` when the runtime has none (Go flat params). Java's `TaskArgs` binds this way although it has names.
   - `named`: by name in any order, case-insensitively with underscores ignored unless `exact_name` is set (Go `arg:` tags, explicit Java names). A tagged Go struct and Java's `TaskInput` bind this way.
   - `named_or_whole`: as `named`, except that when there is exactly one argument and it matches no parameter, it is decoded as the whole value (an untagged Go struct).
   - `named_open`: as `named`, but `params` need not list every parameter. Only the listed names are checked, and an argument none of them names is not a mismatch. TypeScript binds this way: its types are erased, so a handler knows only its explicit `withArgNames` renames.
   
   Other changes:
   
   - `ToSDKTaskHandlerProcessor = TaskHandlerParseRequest | _ParseSideResponses` is new. `ToDagProcessor` keeps its members, now written as `DagFileParseRequest | _ParseSideResponses`. `ToManager` gains `TaskHandlerParsingResult`, and `DagFileProcessorProcess` rejects it as an unhandled request.
   - A requested Dag id the artifact registers nothing for is omitted from `task_handlers`. `params` keeps declaration order, which matters only for `positional`. `value_schema` reuses `ArgValueSchema`, the type `arg_bindings` carry, so the stub-to-handler check compares one type.
   - The supervisor schema registry introspects the new union, and the snapshot gains four definitions.
   - No `VersionChange` is added: as `versions/v2026_10_30.py` documents since #69403, a brand-new message body needs none. The schema package's AGENTS.md already had this rule. It now gives the reason, and notes that `check-supervisor-schemas-versions` fails a local commit that changes the snapshot without touching `versions/`, while CI's `--all-files` run passes.
   - `ts-sdk/src/generated/supervisor.ts` is regenerated. Selective checks skip `check-ts-sdk-supervisor-schema` on PRs with no ts-sdk file and call that "the ts-sdk follow-up PR's job", but CI runs it with `--all-files` whenever a ts-sdk file changes, so a stale copy fails the next ts-sdk PR (as after #73027, fixed by #73663).
   - The Go genmodels and the vendored Java schema are refreshed in the SDK layers above. The Java copy needs another sync path there, because its sync hook only fires when `api_version` changes, and this PR does not change it.
   - ADR-0012's message shapes and Appendix B describe the binding modes, and ADR-0011 and ADR-0013 say the parse check follows each declaration's `binding`.
   
   ## How to test
   
   ```bash
   uv run --project airflow-core --with-editable shared/secrets_masker pytest \
     airflow-core/tests/unit/dag_processing/test_processor.py -q
   uv run --project task-sdk pytest task-sdk/tests/task_sdk/execution_time/schema \
     task-sdk/tests/task_sdk/execution_time/test_supervisor_schemas_name_type_sync.py -q
   ```
   
   Ran: `test_processor.py` 132 passed; the task-sdk selection 60 passed. `prek` on the changed files passed (`generate-supervisor-schemas-snapshot`, `mypy-airflow-core`, `mypy-task-sdk`, `mypy-scripts`, the ts-sdk lint and supervisor-schema checks), with `check-supervisor-schemas-versions` skipped at commit as the AGENTS.md note above explains. `compile-ts-sdk` (manual stage) passed.
   
   ---
   
   ##### Was generative AI tooling used to co-author this PR?
   
   - [X] Yes (please specify the tool below)
   
   Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
   


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

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

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