[jason810496]

Stack (bottom to top): #74042, #74035, #74043, #74036, #74037, #73845, #73846, #73847

Why

Nothing can parse a Dag written entirely in a Lang SDK yet. This is the Dag processor side of the parse_dag half of #71929.

What changes

• Routing. The manager parses a file whose importer is a CoordinatorDagImporter with the new LangSDKDagFileProcessorProcess (dag_processing/lang_sdk_processor.py), and every other file as before. A callback for such a file is dropped with a warning before the file is queued.
• Parse child. The forked child resolves get_dagbag_import_timeout for the file, asks the file's importer for its coordinator and execs the runtime, which connects back to the Dag processor over TCP and answers the parse request. Fork, the macOS exec path and log forwarding work as for a Python file (Move the Dag file processor's shared plumbing into a base class #74040).
• Result. Each returned Dag gets fill_config_defaults and must pass validate_serialized_dag (Validate a serialized Dag and fill its unset settings from the Airflow config #74041). A failed start, a missing result, an invalid frame or message, an invalid Dag, or a parse past its timeout gives an import error. A runtime still running 5s after its result is killed, with anything left in its process group. A native Dag gets the same multi-team executor and pool checks as a Python Dag. The file's Dag source is read with its Dag importer.
• Dag bag. A Dag bag cannot parse such a file yet: the importer reports an import error, and sync_bag_to_db leaves the file's import errors to the Dag processor. Parse native Lang-SDK Dags in a Dag bag #74043 adds the Dag bag path.
• Worker. A Python worker handed a task of a native Dag fails it without retries, before it builds a Dag bag. The error names the coordinator class to route the Dag's queue to.

class NodeDagImporter(CoordinatorDagImporter):
    coordinator_classpath = "airflow.sdk.coordinators.node.NodeCoordinator"
    artifact_suffix = ".min.mjs"
    supported_extensions = [".mjs"]  # a registry routes by the last suffix

    def get_source_code(self, definition): ...


class NodeCoordinator(SubprocessCoordinator):
    def _build_parse_dag_command(self, *, path):
        # parse_dag appends --comm and --logs
        return [self.node_executable, os.fspath(path)], read_bundle(path).supervisor_schema_version

Limitations

• Cluster policies do not run on a native Dag.
• Only the checks above run on a native Dag. A rule the Python DAG class enforces, such as that a scheduled Dag's required params have defaults, is not checked.

Deviations from #71929

• The Python parse child execs the runtime and hands it TCP addresses, so no fd-0 bridge stays in between. parse_dag is on SubprocessCoordinator only.

Follow-ups

• Java Serde.kt (Java SDK: Serialize native Dags to DagSerialization v3 #71190) still writes the stock values of the five config settings, so a native Java Dag does not get the config values yet. The Go SDK serializer (Support TaskFlow Dag definitions natively in the Go-SDK #70158) writes them too.
• The Go SDK does not parse native Dags yet: it has no CoordinatorDagImporter, and a packed Go bundle has no file extension to claim.
• Decision 2 of airflow-core/adr/lang-sdk/0009-provider-operators-as-generated-dsl.md says a provider DSL task runs on a Python worker, which a task of a native Dag cannot do. It needs its own amendment.

---

Was generative AI tooling used to co-author this PR?

• Yes, with help of Claude Code Opus 5.5 following the guidelines

[pierrejeambrun]

LGTM just one thing to address I believe.

[jason810496]

Thanks for the review!