Parse native Go Dags and show each Dag's own source

airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst

@@ -62,6 +62,8 @@ is the same coordinator mechanism the Java SDK uses. Because the mature Python s
 Airflow-facing concerns, Go tasks inherit remote task logs (S3/GCS), the full range of task states, and
 alternate XCom backends rather than implementing them again in Go.
 
+.. _go-sdk/quick-start:
+
 Quick start
 -----------
 
@@ -221,6 +223,33 @@ There is no separate Go worker to run: the Airflow worker forks the bundle binar
   and the Dag processor resolve ``task_handler_bundle_name`` through it, and wherever the ``[sdk]`` config
   is read it is rejected if the name is missing there.
 
+A Dag processor with this ``[sdk]`` configuration also parses the bundle binaries of every Dag bundle (see
+:ref:`go-sdk/native-dag-parsing`). Keep it from parsing the binaries that only register task handlers with
+that Dag bundle's ``.airflowignore``, whose pattern syntax follows ``[core] dag_ignore_file_syntax``. Tasks do
+not read ``.airflowignore``, so the binaries still run.
+
+With ``task_handler_bundle_name`` set to its own Dag bundle, as above, ``go-task-handlers`` holds only handler
+binaries, so ignore everything in it:
+
+.. code-block:: bash
+
+    # [core] dag_ignore_file_syntax = glob (the default)
+    echo '*' > /opt/airflow/go-task-handlers/.airflowignore
+
+    # [core] dag_ignore_file_syntax = regexp; a bare * is not a valid pattern and is dropped
+    echo '.' > /opt/airflow/go-task-handlers/.airflowignore
+
+With ``task_handler_bundle_name`` unset, the binaries sit in the same Dag bundle as the Python Dag file. Go
+binaries have no extension, so put them in a folder such as ``bin/`` and ignore that folder:
+
+.. code-block:: bash
+
+    # [core] dag_ignore_file_syntax = glob (the default)
+    echo 'bin/*' >> /opt/airflow/dags/.airflowignore
+
+    # [core] dag_ignore_file_syntax = regexp
+    echo '^bin/' >> /opt/airflow/dags/.airflowignore
+
 Writing tasks
 -------------
 
@@ -596,13 +625,39 @@ Deploying
 Copy or mount the packed bundle into the Dag bundle named by the coordinator's ``task_handler_bundle_name``.
 The :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` scans that Dag bundle recursively,
 matches the incoming ``dag_id`` against each bundle's manifest, verifies the bundle's integrity hash, and
-launches the matching bundle. Bundles are identified by the trailer magic, not by filename (no extension on
-Linux/macOS, ``.exe`` on Windows), so the file name on the worker is irrelevant.
+launches the matching bundle. This scan identifies bundles by the trailer magic, not by file name. The Dag
+processor also requires a file name without an extension, see :ref:`go-sdk/native-dag-parsing`.
 
 The matching bundle is marked executable before it is launched, so any Dag bundle works, including an
 object-store one such as ``S3DagBundle`` that has no concept of file permissions and so cannot preserve
 the execute bit the build produced.
 
+.. _go-sdk/native-dag-parsing:
+
+Parsing native Dags
+~~~~~~~~~~~~~~~~~~~
+
+Once an :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` is configured, the Dag processor
+parses every bundle binary in each Dag bundle and runs it to collect the Dags it defines. The Dag processor
+recognizes a bundle binary only when its file name has no extension, such as ``bin/orders``, not
+``bin/orders.bin``. A binary that only registers task handlers is parsed too: each parse runs it and finds
+no Dags, and a binary built with a Go SDK that cannot answer the Dag-parse request records an import error.
+Keep those out of the Dag processor with ``.airflowignore``, as described in :ref:`Quick start <go-sdk/quick-start>`.
+
+With one :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`, it parses the binaries of every Dag
+bundle. With several, for example a second one for another ``task_handler_bundle_name``, map each Dag bundle that
+holds native Go Dags to one of them in ``[sdk] dag_bundle_to_coordinator``. A binary in a Dag bundle that has no
+entry, or whose entry names no :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`, fails to parse
+with an import error:
+
+.. code-block:: ini
+
+    [sdk]
+    dag_bundle_to_coordinator = {"dags-folder": "go"}
+
+* The Code view shows each native Dag's own source file, taken from the sources the bundle embeds.
+* A task of a native Go Dag runs the bundle binary its Dag was parsed from.
+
 .. _go-sdk/coordinator-config:
 
 :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` configuration

task-sdk/docs/airflow-metadata.schema.json

@@ -4,7 +4,7 @@
   "title": "Airflow Executable SDK Bundle Metadata",
   "description": "Build-time manifest declaring DAG and task identifiers exposed by an Airflow native-executable SDK bundle. See the Executable Bundle Spec documentation in the Airflow Task SDK.",
   "type": "object",
-  "required": ["airflow_bundle_metadata_version", "sdk", "source", "dags"],
+  "required": ["airflow_bundle_metadata_version", "sdk", "dags"],
   "additionalProperties": true,
   "properties": {
     "airflow_bundle_metadata_version": {
@@ -35,11 +35,26 @@
         }
       }
     },
-    "source": {
+    "entrypoint_path": {
       "type": "string",
-      "description": "Original filename of the primary Dag source file (e.g. 'example.go'). Bundle formats with an embedded source region use this as its display name; other formats treat it as the logical authoring name.",
+      "description": "Path of the entrypoint source file (e.g. 'example/bundle/main.go'). It must be one of the sources paths. Shown for a Dag that dag_source_paths does not map.",
       "minLength": 1
     },
+    "dag_source_paths": {
+      "type": "object",
+      "description": "Mapping of dag_id to the path of the file that defines the Dag. Every value must be one of the sources paths.",
+      "additionalProperties": {
+        "type": "string",
+        "minLength": 1
+      }
+    },
+    "sources": {
+      "type": "array",
+      "description": "The embedded source files, in the order they appear in the source region. A bundle without it embeds no source. Paths must be unique.",
+      "items": {
+        "$ref": "#/$defs/sourceEntry"
+      }
+    },
     "dags": {
       "type": "object",
       "description": "Mapping of dag_id to DAG entry. Every dag_id the bundle exposes must appear here.",
@@ -50,6 +65,34 @@
     }
   },
   "$defs": {
+    "sourceEntry": {
+      "type": "object",
+      "description": "One embedded source file inside the bundle's source region.",
+      "required": ["path", "offset", "length", "sha256"],
+      "additionalProperties": true,
+      "properties": {
+        "path": {
+          "type": "string",
+          "description": "The file's path as the author knows it (e.g. 'example/bundle/main.go').",
+          "minLength": 1
+        },
+        "offset": {
+          "type": "integer",
+          "description": "The file's start, in bytes from the start of the source region.",
+          "minimum": 0
+        },
+        "length": {
+          "type": "integer",
+          "description": "The file's length in bytes.",
+          "minimum": 0
+        },
+        "sha256": {
+          "type": "string",
+          "description": "Lower-case hexadecimal SHA-256 of the file's bytes.",
+          "pattern": "^[0-9a-f]{64}$"
+        }
+      }
+    },
     "dagEntry": {
       "type": "object",
       "description": "Static description of a single DAG declared in the bundle.",

task-sdk/docs/executable-bundle-spec.rst

@@ -42,29 +42,37 @@ A bundle file therefore has three regions, in order from offset 0:
 
 1. The native executable (ELF / Mach-O / PE), including any code-signing
    structures the platform appends.
-2. The primary DAG source file, embedded verbatim (UTF-8). MAY have length 0.
+2. The embedded source files, each verbatim (UTF-8), back to back. MAY have length 0.
+   The metadata indexes them. See :ref:`the source region <bundle-source-region>`.
 3. The build-time manifest (``airflow-metadata.yaml`` content, UTF-8).
 
 The file ends with a fixed 64-byte trailer that locates regions (2) and (3),
 carries an integrity hash of the binary region, and identifies the file as a
 bundle. See :ref:`bundle-trailer-layout`.
 
-Filenames follow OS conventions for executables: no extension on Linux/macOS,
-``.exe`` on Windows. The scanner identifies bundles by the trailer's magic,
-not by the filename.
+A bundle file has no file extension, and the Dag processor does not treat a
+file with an extension as a bundle. The scanner identifies bundles by the
+trailer's magic.
 
 The complete bundle file regions are:
 
 .. code-block:: text
 
     [0,            source_start)    native binary (must be non-empty)
-    [source_start, metadata_start)  embedded source (may be zero length)
+    [source_start, metadata_start)  embedded source files (may be zero length)
     [metadata_start, file_size-64)  build-time manifest
     [file_size-64, file_size)       64-byte trailer
 
 where ``metadata_start = file_size - 64 - metadata_len`` and
 ``source_start = metadata_start - source_len``.
 
+.. _bundle-source-region:
+
+The source region holds the source files of the Dags the bundle defines, one entry per file. Files
+are concatenated with no separator, and the ``sources`` list in the manifest gives each file's
+``path``, ``offset`` and ``length`` within the region and its ``sha256``. A file's ``offset`` is
+relative to ``source_start``, so the first file has offset ``0``.
+
 Reference Implementation
 ------------------------
 
@@ -84,15 +92,15 @@ for SDK users. Go SDK's ``airflow-go-pack`` is a good example.
 
     BINARY = pathlib.Path(...)  # Path to the compiled executable.
     OUTPUT = pathlib.Path(...)  # Where to put the processed executable.
-    SOURCE = b"..."  # Source code to embed.
+    SOURCES = [b"..."]  # Source files to embed, in the order the manifest lists them.
     METADATA = b"..."  # UTF-8-encoded YAML metadata.
 
     # SHA-256 covers the binary region only: bytes [0, source_start).
     binary_sha256 = hashlib.sha256(BINARY.read_bytes()).digest()
 
     trailer = struct.pack(
         "<III 32s 12s 8s",
-        len(SOURCE),  # source_len
+        sum(map(len, SOURCES)),  # source_len
         len(METADATA),  # metadata_len
         1,  # footer_ver
         binary_sha256,
@@ -103,7 +111,7 @@ for SDK users. Go SDK's ``airflow-go-pack`` is a good example.
 
     shutil.copy(BINARY, OUTPUT)
     with OUTPUT.open("ab") as fh:
-        fh.write(SOURCE)  # Embedded source region.
+        fh.writelines(SOURCES)  # Embedded source region.
         fh.write(METADATA)  # Metadata region.
         fh.write(trailer)
     OUTPUT.chmod(0o755)
@@ -156,9 +164,17 @@ Reader algorithm:
    re-hash on every exec; a cache miss (file replaced, mtime bumped)
    triggers re-verification.
 7. Read ``metadata_len`` bytes from ``metadata_start`` for the manifest.
-8. Read ``source_len`` bytes from ``source_start`` for the source view.
-   If ``source_len == 0``, no source is embedded; the UI displays
-   "(source not available)".
+8. Read the source files through the manifest's ``sources`` list. For each entry, check that
+   ``offset`` and ``length`` are non-negative integers and that ``offset + length <= source_len``.
+   To read a file, read ``length`` bytes from ``source_start + offset`` and compare their SHA-256 to
+   ``sha256``. A duplicate ``path`` or a digest mismatch is an error. Without a ``sources`` key, no
+   source is embedded, and the UI shows a notice in place of the source.
+
+   A Dag's source file is the ``dag_source_paths`` entry for its ``dag_id``. A Dag with no entry,
+   such as one built dynamically, shows ``entrypoint_path``. A Dag owned by another language, such as
+   the Python Dag that a bundle's task handlers run for, shows its own source and not an embedded
+   file. ``entrypoint_path`` and every ``dag_source_paths`` value MUST be one of the ``sources``
+   paths.
 
 Source comes *before* metadata so a future ``footer_ver`` MAY introduce
 additional trailing blobs (e.g. signed checksums, compressed deps) by
@@ -182,7 +198,19 @@ and editors.
       language: go
       version: "0.1.0"
       supervisor_schema_version: "2026-06-16"
-    source: example.go
+    entrypoint_path: example/bundle/main.go
+    dag_source_paths:
+      example_dag: example/bundle/main.go
+      another_dag: example/bundle/dags/another.go
+    sources:
+      - path: example/bundle/main.go
+        offset: 0
+        length: 1532
+        sha256: 0f3a...e91c
+      - path: example/bundle/dags/another.go
+        offset: 1532
+        length: 811
+        sha256: 7b21...04d8
     dags:
       example_dag:
         tasks:
@@ -214,12 +242,24 @@ Top-level keys:
       task-execution time, and an unknown version causes that bundle to be
       skipped.
 
-``source`` (string, required)
-    Original filename of the primary DAG source file (e.g. ``example.go``).
-    The file's bytes live in the source region of the bundle, not at this
-    path; this field is a display name the Airflow UI uses to label the
-    source-view panel and pick a syntax-highlighting mode from the
-    extension.
+``entrypoint_path`` (string, optional)
+    Path of the entrypoint source file, such as the Go ``main`` package file. It MUST be one of
+    the ``sources`` paths. The Airflow UI shows it for a Dag that ``dag_source_paths`` does not map.
+
+``dag_source_paths`` (mapping, optional)
+    Mapping of ``dag_id`` to the path of the file that defines the Dag. Every value MUST be one of
+    the ``sources`` paths.
+
+``sources`` (list, optional)
+    The embedded source files, in the order they appear in the source region. A bundle without it
+    embeds no source. Each entry has:
+
+    - ``path`` (string, required): the file's path as the author knows it, such as
+      ``example/bundle/main.go``. Paths MUST be unique. The Airflow UI picks a syntax-highlighting
+      mode from the extension.
+    - ``offset`` (integer, required): the file's start, in bytes from ``source_start``.
+    - ``length`` (integer, required): the file's length in bytes.
+    - ``sha256`` (string, required): the lower-case hexadecimal SHA-256 of the file's bytes.
 
 ``dags`` (mapping, required)
     Mapping of ``dag_id`` to a *DAG entry*. Every ``dag_id`` the bundle
@@ -243,16 +283,16 @@ Go bundle::
 
     example
     ├── ELF/Mach-O/PE executable
-    ├── source region:   contents of example.go
-    ├── metadata region: airflow-metadata.yaml (source: example.go)
+    ├── source region:   example/bundle/main.go, example/bundle/dags/another.go
+    ├── metadata region: airflow-metadata.yaml (entrypoint_path, dag_source_paths, sources)
     └── trailer (64 B):  lengths + binary_sha256 + AFBNDL01 magic
 
 Rust bundle::
 
     pipeline
     ├── ELF/Mach-O/PE executable
-    ├── source region:   contents of main.rs
-    ├── metadata region: airflow-metadata.yaml (source: main.rs)
+    ├── source region:   src/main.rs
+    ├── metadata region: airflow-metadata.yaml (entrypoint_path: src/main.rs)
     └── trailer (64 B):  lengths + binary_sha256 + AFBNDL01 magic
 
 The bundle is one file. ``./example`` runs the binary; the appended data
@@ -272,7 +312,7 @@ that perform additional post-build steps MUST observe the following order:
   binary region; nothing has been written past its OS-defined end yet, so
   the digest matches what the reader will recompute over
   ``[0, source_start)`` after the append.
-- **Append** ``<source><metadata><trailer>`` in a single write so a
+- **Append** ``<sources><metadata><trailer>`` in a single write so a
   partially written file fails the magic or hash check rather than
   appearing as a half-valid bundle.
 

task-sdk/src/airflow/sdk/coordinators/_bundle_metadata.py

@@ -124,3 +124,19 @@ def extract_supervisor_schema_version(metadata: dict[str, Any]) -> str:
     if not isinstance(value, str) or not value:
         raise ValueError("missing or invalid sdk.supervisor_schema_version")
     return value
+
+
+def resolve_source_path(metadata: dict[str, Any], dag_id: str | None) -> str | None:
+    """
+    Return the embedded source path to show for *dag_id*, or ``None`` when there is none.
+
+    A Dag mapped in ``dag_source_paths`` resolves to its own file. Any other Dag, such as one built
+    dynamically, and a *dag_id* of ``None`` resolve to ``entrypoint_path``.
+    """
+    dag_source_paths = metadata.get("dag_source_paths")
+    if dag_id is not None and isinstance(dag_source_paths, dict):
+        mapped = dag_source_paths.get(dag_id)
+        if isinstance(mapped, str):
+            return mapped
+    entrypoint_path = metadata.get("entrypoint_path")
+    return entrypoint_path if isinstance(entrypoint_path, str) else None

task-sdk/src/airflow/sdk/coordinators/_dag_importer.py

@@ -44,6 +44,7 @@
 COORDINATOR_DAG_IMPORTERS: Final[tuple[str, ...]] = (
     "airflow.sdk.coordinators.java._dag_importer.JavaDagImporter",
     "airflow.sdk.coordinators.node._dag_importer.NodeDagImporter",
+    "airflow.sdk.coordinators.executable._dag_importer.ExecutableDagImporter",
 )
 """
 The classpaths of the :class:`CoordinatorDagImporter` subclasses a Dag bundle's registry may hold.