Parse native Lang-SDK Dags in the Dag processor with their runtime

airflow-core/adr/lang-sdk/0010-native-dag-processing.md

@@ -152,6 +152,15 @@ DagModelOperation → PERSIST
 
 `DagImportResult.dags` is `list[DAG]`, so the importer wraps each serialized entry as a `LazyDeserializedDAG` and transforms it into an `airflow.sdk.DAG`.
 
+### Interim: the Dag processor manager routes claimed files
+
+Until importers own their parse process (#73457), the Dag processor manager routes a coordinator-claimed file itself, as of #74035. When a file's importer in the bundle's
+registry is a coordinator's, the manager starts `LangSDKDagFileProcessorProcess` for it instead of `DagFileProcessorProcess`. That process execs the runtime, which connects
+back to it. The importer's `import_definition` only reports that the Dag processor parses the file, which is what a Dag bag outside the Dag processor sees.
+
+This is a bridge, not the end state. Running the runtime inside the Python parse child would need that child to relay every request and reply between the runtime and the
+manager over its fd 0. Starting the runtime from the manager lets it reach the manager's request handlers directly, with no relay.
+
 ## Consequences
 
 - The Java and TypeScript importers are never configured by hand. The runtime is declared once, in `[sdk] coordinators`, and the importer follows from it.

airflow-core/src/airflow/dag_processing/dagbag.py

@@ -33,6 +33,7 @@
 from airflow._shared.timezones import timezone
 from airflow.configuration import conf
 from airflow.dag_processing.bundles.local import LocalDagBundle
+from airflow.dag_processing.importer_routing import get_claiming_importer
 from airflow.exceptions import (
     AirflowClusterPolicyError,
     AirflowClusterPolicySkipDag,
@@ -57,6 +58,7 @@
     from airflow import DAG
     from airflow.models.dagwarning import DagWarning
     from airflow.sdk.importers import AbstractDagImporter, DagDefinition, DagImportWarning, DagSourceCode
+    from airflow.serialization.definitions.dag import SerializedDAG
 
 
 class FileLoadStat(NamedTuple):
@@ -103,27 +105,24 @@ def _executor_exists(executor_name: str, team_name: str | None) -> bool:
     return False
 
 
-def _validate_executor_fields(dag: DAG, bundle_name: str | None = None) -> None:
+def _get_bundle_team_name(bundle_name: str | None) -> str | None:
+    """Return the team that owns *bundle_name* when multi-team is on, else ``None``."""
+    if not conf.getboolean("core", "multi_team") or not bundle_name:
+        return None
+    from airflow.dag_processing.bundles.manager import DagBundlesManager
+
+    return DagBundlesManager()._bundle_config[bundle_name].team_name
+
+
+def _validate_executor_fields(dag: DAG | SerializedDAG, bundle_name: str | None = None) -> None:
     """Validate that executors specified in tasks are available and owned by the same team as the dag bundle."""
     import logging
 
     log = logging.getLogger(__name__)
-    dag_team_name = None
-
-    # Check if multi team is available by reading the multi_team configuration (which is boolean)
-    if conf.getboolean("core", "multi_team"):
-        # Get team name from bundle configuration if available
-        if bundle_name:
-            from airflow.dag_processing.bundles.manager import DagBundlesManager
 
-            bundle_manager = DagBundlesManager()
-            bundle_config = bundle_manager._bundle_config[bundle_name]
-
-            dag_team_name = bundle_config.team_name
-            if dag_team_name:
-                log.debug(
-                    "Found team '%s' for DAG '%s' via bundle '%s'", dag_team_name, dag.dag_id, bundle_name
-                )
+    dag_team_name = _get_bundle_team_name(bundle_name)
+    if dag_team_name:
+        log.debug("Found team '%s' for DAG '%s' via bundle '%s'", dag_team_name, dag.dag_id, bundle_name)
 
     for task in dag.tasks:
         if not task.executor:
@@ -150,18 +149,7 @@ def _assign_default_team_pools(
     bundle_name: str | None = None,
 ) -> None:
     """Assign the default team pool to tasks that do not explicitly specify a pool."""
-    dag_team_name = None
-
-    if conf.getboolean("core", "multi_team"):
-        if bundle_name:
-            from airflow.dag_processing.bundles.manager import DagBundlesManager
-
-            bundle_manager = DagBundlesManager()
-            bundle_config = bundle_manager._bundle_config[bundle_name]
-
-            dag_team_name = bundle_config.team_name
-
-    if not dag_team_name:
+    if not (dag_team_name := _get_bundle_team_name(bundle_name)):
         return
 
     for task in dag.tasks:
@@ -583,16 +571,30 @@ def sync_bag_to_db(
     version_data: dict[str, Any] | None = None,
     session: Session = NEW_SESSION,
 ) -> None:
-    """Save attributes about list of DAG to the DB."""
+    """
+    Save attributes about list of DAG to the DB.
+
+    Files that a Lang-SDK runtime parses are left out, with their import errors: the Dag processor
+    stores those.
+    """
     from airflow.dag_processing.collection import update_dag_parsing_results_in_db
 
-    import_errors = {(bundle_name, rel_path): error for rel_path, error in dagbag.import_errors.items()}
+    def is_parsed_by_runtime(rel_path: str) -> bool:
+        return get_claiming_importer(Path(dagbag.bundle_path or "", rel_path), bundle_name) is not None
+
+    import_errors = {
+        (bundle_name, rel_path): error
+        for rel_path, error in dagbag.import_errors.items()
+        if not is_parsed_by_runtime(rel_path)
+    }
 
     # Build the set of all files that were parsed and include files with import errors
     # in case they are not in parsed_definitions
     files_parsed = set(import_errors)
     if dagbag.bundle_path:
         for rel_path in dagbag.parsed_definitions:
+            if is_parsed_by_runtime(rel_path):
+                continue
             files_parsed.add((bundle_name, rel_path))
             # A definition nested in an archive also clears the archive's own discovery errors.
             if enclosing_file := find_enclosing_file(Path(dagbag.bundle_path, rel_path)):

airflow-core/src/airflow/dag_processing/importer_routing.py

@@ -0,0 +1,47 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Route Dag files to the process that parses them, by the bundle's Dag importer registry."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import TYPE_CHECKING
+
+from airflow.sdk.coordinators._dag_importer import find_claiming_importer
+
+if TYPE_CHECKING:
+    from airflow.sdk.coordinators._dag_importer import CoordinatorDagImporter  # noqa: SDK001
+
+log = logging.getLogger(__name__)
+
+
+def get_claiming_importer(
+    path: str | os.PathLike[str], bundle_name: str | None
+) -> CoordinatorDagImporter | None:
+    """
+    Return the coordinator Dag importer that claims ``path``, or ``None`` when a Python child parses it.
+
+    A runtime parses the file when its importer is a coordinator's Dag importer. When the bundle's
+    importers cannot be loaded, the error is logged and a Python child parses the file.
+    """
+    try:
+        return find_claiming_importer(path, bundle_name)
+    except Exception:
+        log.exception("Cannot load the Dag importer for %s in bundle %s", path, bundle_name)
+        return None