TS SDK: embed the entrypoint source in the Dag bundle

.rat-excludes

@@ -328,6 +328,9 @@ www-hash.txt
 # Vendored-in code
 /src/airflow/providers/google/_vendor/*
 
+# TypeScript bundle golden fixture: its bytes are digest-pinned, so no license header can be added
+/ts-sdk/tests/cli/fixtures/bundle-v1.min.mjs
+
 # Java SDK build outputs
 /java-sdk/bin/*
 /java-sdk/build/*

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

@@ -281,6 +281,10 @@ read or edit in place. The ``/*! */`` license banners of bundled dependencies ar
 a function name, so minified names are safe: a Dag and a task are named by the string ids their registration
 states, and a handler is dispatched by reference.
 
+Because the shipped code is not the code anyone wrote, the packer also embeds the entry module verbatim in a
+``/*# airflowSource ... #*/`` block comment, verified by its own digest, so Airflow has something readable to
+display for the Dag. Only the entry module is embedded, not the modules it imports.
+
 ``esbuild`` is an optional peer dependency: packing is build-time only, so the runtime install of
 ``apache-airflow-ts-sdk`` skips it, and it must be installed separately before running ``airflow-ts-pack``.
 

docs/spelling_wordlist.txt

@@ -1840,6 +1840,7 @@ uncompress
 undeploy
 undeployed
 Undeploys
+unescaped
 unicode
 unicodecsv
 unindent

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

@@ -32,16 +32,16 @@ that suffix.
 Container
 ---------
 
-The bundle remains an ECMAScript module that runs directly with ``node bundle.min.mjs``. It has three regions:
+The bundle remains an ECMAScript module that runs directly with ``node bundle.min.mjs``. It has four regions:
 
 .. code-block:: text
 
     //# airflowBundle=<compact JSON layout>\n
     //# airflowMetadata=<compact JSON>\n
+    /*# airflowSource\n<escaped entrypoint source>\n#*/\n
     <minified, bundled ECMAScript code>
 
-The layout comes first so readers can locate and verify the other regions. The
-current format has no embedded source region.
+The layout comes first so readers can locate and verify the other regions.
 
 Layout Header
 -------------
@@ -60,6 +60,11 @@ The ``airflowBundle`` payload is a compact UTF-8 JSON object:
         "start": "0000000000000300",
         "end": "0000000000000400",
         "sha256": "123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0"
+      },
+      "source": {
+        "start": "0000000000000412",
+        "end": "00000000000003f0",
+        "sha256": "23456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef01"
       }
     }
 
@@ -75,13 +80,41 @@ including this physical framing and the decoded metadata schema. A reader parses
 can locate and verify that version.
 
 The metadata range points to the UTF-8 JSON payload only, excluding the JavaScript comment marker and newline. Its
-digest therefore covers the exact JSON bytes stored in that range. The code range covers every byte after the
-metadata line through the end of the file, and its digest covers those raw JavaScript bytes.
-
-The file begins with the layout line. The metadata marker immediately follows that line, and exactly one newline
-separates the metadata payload from the code range. These prescribed framing bytes are outside the hashed metadata
-and code ranges, and no additional bytes are permitted before, between, or after them. Post-pack formatters,
-compressors, source-map injectors, and other tools that rewrite the bundle invalidate the offsets or digests.
+digest therefore covers the exact JSON bytes stored in that range. The source range likewise points to the escaped
+entrypoint payload only, excluding the ``/*# airflowSource\n`` opener and the ``\n#*/\n`` closer. The code range
+covers every byte after the source comment through the end of the file, and its digest covers those raw JavaScript
+bytes.
+
+The file begins with the layout line. The metadata marker immediately follows that line, exactly one newline
+separates the metadata payload from the source comment's opener, and the source comment's closer is immediately
+followed by the code range. These prescribed framing bytes are outside the hashed ranges, and no additional bytes are
+permitted before, between, or after them. Post-pack formatters, compressors, source-map injectors, and other tools
+that rewrite the bundle invalidate the offsets or digests.
+
+Unlike the metadata range, the source range's length is declared rather than derivable from a newline, so a reader
+pins it by checking that the prescribed opener and closer sit exactly where the declared range implies.
+
+Source
+------
+
+The source region carries the entrypoint as its author wrote it, so the Airflow UI has something readable to show
+for a natively authored TypeScript Dag. The shipped code region is minified and is not the code anyone wrote. Only
+the entrypoint is embedded, not the module graph behind it:
+`ADR-0006 <https://github.com/apache/airflow/blob/main/airflow-core/adr/lang-sdk/0006-no-lang-sdk-source-display.md>`__
+declined multi-file source display for mixed-language Dags.
+
+It is a block comment rather than the line comments the layout and metadata use, because the entrypoint spans the
+lines it was written on and a ``//`` comment would end at the first of them. Line terminators, including ``\r`` and
+U+2028/U+2029, are therefore legal inside the region and need no escaping.
+
+``*/`` is the one sequence that must not appear: it would end the comment where Node reads the file, putting the rest
+of the payload into executable position while the layout still calls those bytes source and both digests still match.
+The packer escapes it by inserting a ``\`` between the two characters, and escapes ``*\`` the same way so the
+transformation is reversible. A reader recovers the entrypoint by dropping the ``\`` that follows a ``*`` and keeping
+the character behind it, and MUST reject a bundle whose source range contains an unescaped ``*/``.
+
+The region is capped at 1 MiB. Larger entrypoints should move code into imported modules, which are bundled into the
+code region as usual.
 
 Metadata
 --------
@@ -98,7 +131,7 @@ The ``airflowMetadata`` payload is compact UTF-8 JSON with this logical shape:
         "supervisor_schema_version": "2026-06-16"
       },
       "source": "main.ts",
-      "dags": {
+      "task_handlers": {
         "example": {
           "tasks": ["extract", "load"]
         }
@@ -109,8 +142,13 @@ The packer serializes this object without insignificant whitespace and escapes t
 separators (U+2028 and U+2029), keeping it in one newline-terminated JavaScript comment without a second encoding
 layer. The SHA-256 digest detects changes to the exact serialized bytes.
 
-The coordinator uses the ``dags`` keys to choose a bundle for a task instance. The ``source`` value is a logical
-authoring name only, not embedded source content, and it is not used to execute the bundle.
+``task_handlers`` is keyed by Dag ID and lists the task IDs the bundle handles for each. It is named for what a
+TypeScript bundle actually provides: handlers for Dags declared elsewhere, not Dag definitions of its own. The
+coordinator uses its keys to choose a bundle for a task instance.
+
+The metadata ``source`` value is the logical authoring name displayed for the Dag, a filename rather than content.
+The source region named in the layout header is what carries the content. The two are separate fields in separate
+documents, and neither is used to execute the bundle.
 
 Reader and Selection Algorithm
 ------------------------------
@@ -122,14 +160,15 @@ For each candidate in ``bundles_root``, the coordinator:
    a filesystem returns entries in. Directories are deduplicated by ``(st_dev, st_ino)``, so a symlink loop
    terminates the walk instead of exhausting the interpreter stack.
 2. Reads a bounded first line and decodes the named metadata and code ranges.
-3. Reads the bounded metadata line and checks that the declared metadata and code ranges exactly match their
-   physical locations and the file size.
-4. Computes SHA-256 for both ranges before parsing or using metadata.
-5. Confirms with ``fstat`` that the open file did not change during verification.
-6. Parses metadata and requires a supported bundle contract major version from
+3. Reads the bounded metadata line and checks that the declared metadata range matches its physical location.
+4. Reads the bounded source region, checks that its prescribed opener and closer frame the declared range, rejects an
+   unescaped ``*/`` inside it, and checks that the declared code range matches its physical location and file size.
+5. Computes SHA-256 for all three ranges before parsing or using metadata.
+6. Confirms with ``fstat`` that the open file did not change during verification.
+7. Parses metadata and requires a supported bundle contract major version from
    ``airflow_bundle_metadata_version``.
-7. Skips the verified bundle if its ``dags`` mapping does not contain the requested ``dag_id``.
-8. Resolves the supervisor schema version and selects the first usable match.
+8. Skips the verified bundle if its ``task_handlers`` mapping does not contain the requested ``dag_id``.
+9. Resolves the supervisor schema version and selects the first usable match.
 
 A missing, unrelated, unreadable, malformed, corrupt, or incompatible earlier candidate does not prevent selection
 of a later usable match. When more than one usable bundle declares the same Dag, the first configured match wins. If

task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py

@@ -18,8 +18,8 @@
 """
 Read and verify TypeScript Dag bundles.
 
-File order: layout comment, metadata comment, then executable JavaScript.
-Read the headers, verify the section digests, and return coordinator metadata.
+File order: layout comment, metadata comment, source block comment, then executable JavaScript.
+Read the headers, verify the section digests, and return coordinator metadata or the source.
 """
 
 from __future__ import annotations
@@ -44,7 +44,13 @@
 _MAX_LAYOUT_LINE_BYTES = 4096
 _METADATA_COMMENT_PREFIX = b"//# airflowMetadata="
 _MAX_METADATA_LINE_BYTES = 1024 * 1024
+# A block comment, because the entrypoint spans more than one line.
+_SOURCE_COMMENT_OPEN = b"/*# airflowSource\n"
+_SOURCE_COMMENT_CLOSE = b"\n#*/\n"
+_MAX_SOURCE_BYTES = 1024 * 1024
 _SUPPORTED_BUNDLE_MAJOR_VERSION = 1
+# Reverses the packer's escaping of ``*/``, the only sequence that could close the comment early.
+_SOURCE_ESCAPE = re.compile(rb"\*\\([\\/])")
 
 # Bound hashing memory and process-local cache growth independently of the format.
 _HASH_CHUNK_BYTES = 1024 * 1024
@@ -63,9 +69,10 @@ class _DeclaredSection:
 
 @attrs.define(frozen=True)
 class _BundleLayout:
-    """The metadata and executable sections described by the layout comment."""
+    """The metadata, source, and executable sections described by the layout comment."""
 
     metadata: _DeclaredSection
+    source: _DeclaredSection
     code: _DeclaredSection
 
 
@@ -75,6 +82,7 @@ class _DigestCacheKey:
 
     path: str
     metadata: _DeclaredSection
+    source: _DeclaredSection
     code: _DeclaredSection
     device: int
     inode: int
@@ -88,6 +96,7 @@ class _ComputedDigests:
     """SHA-256 digests calculated from the actual section bytes."""
 
     metadata: bytes
+    source: bytes
     code: bytes
 
 
@@ -106,6 +115,25 @@ class BundleMetadata:
 
 def read_bundle(bundle_path: pathlib.Path) -> BundleMetadata:
     """Read and verify one exact TypeScript bundle file."""
+    # Interpret metadata only after checking its serialized bytes against the declared digest.
+    return _parse_bundle_metadata(_read_verified_payloads(bundle_path).metadata)
+
+
+def read_bundle_source(bundle_path: pathlib.Path) -> str:
+    """Return the entrypoint source ``airflow-ts-pack`` embedded in *bundle_path*."""
+    return _decode_source(_read_verified_payloads(bundle_path).source)
+
+
+@attrs.define(frozen=True)
+class _VerifiedPayloads:
+    """The metadata and source payloads of a bundle whose digests have been checked."""
+
+    metadata: bytes
+    source: bytes
+
+
+def _read_verified_payloads(bundle_path: pathlib.Path) -> _VerifiedPayloads:
+    """Open *bundle_path* once, check its framing and digests, and return its payloads."""
     try:
         bundle_file = bundle_path.open("rb")
     except OSError as exc:
@@ -118,13 +146,19 @@ def read_bundle(bundle_path: pathlib.Path) -> BundleMetadata:
         except OSError as exc:
             raise OSError(f"cannot read {bundle_path.name}: {exc}") from exc
 
-        layout, metadata_payload = _read_bundle_headers(
+        layout, payloads = _read_bundle_headers(
             bundle_file, path=bundle_path, file_size=initial_file_info.st_size
         )
         _verify_integrity(bundle_file, path=bundle_path, layout=layout, initial_file_info=initial_file_info)
 
-    # Interpret metadata only after checking its serialized bytes against the declared digest.
-    return _parse_bundle_metadata(metadata_payload)
+    return payloads
+
+
+def _decode_source(payload: bytes) -> str:
+    try:
+        return _SOURCE_ESCAPE.sub(rb"*\1", payload).decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ValueError(f"embedded airflow source is not valid UTF-8: {exc}") from exc
 
 
 class _BundleDigestCache:
@@ -191,6 +225,7 @@ def _parse_layout(payload: bytes) -> _BundleLayout:
         raise ValueError("embedded airflow bundle layout must contain a mapping")
     return _BundleLayout(
         metadata=_parse_section(layout, "metadata"),
+        source=_parse_section(layout, "source"),
         code=_parse_section(layout, "code"),
     )
 
@@ -266,6 +301,9 @@ def _compute_stable_digests(
         metadata=_hash_region(
             bundle_file, start=layout.metadata.start, end=layout.metadata.end, path=path, section="metadata"
         ),
+        source=_hash_region(
+            bundle_file, start=layout.source.start, end=layout.source.end, path=path, section="source"
+        ),
         code=_hash_region(
             bundle_file, start=layout.code.start, end=layout.code.end, path=path, section="code"
         ),
@@ -290,6 +328,7 @@ def _verify_integrity(
     cache_key = _DigestCacheKey(
         path=os.fspath(path),
         metadata=layout.metadata,
+        source=layout.source,
         code=layout.code,
         device=initial_file_info.st_dev,
         inode=initial_file_info.st_ino,
@@ -307,6 +346,7 @@ def _verify_integrity(
 
     for section, computed_digest, declared_digest in (
         ("metadata", computed_digests.metadata, layout.metadata.sha256),
+        ("source", computed_digests.source, layout.source.sha256),
         ("code", computed_digests.code, layout.code.sha256),
     ):
         if computed_digest != declared_digest:
@@ -335,19 +375,20 @@ def _parse_bundle_metadata(payload: bytes) -> BundleMetadata:
             f"unsupported airflow bundle metadata version {value!r}; "
             f"this runtime supports major version {_SUPPORTED_BUNDLE_MAJOR_VERSION}"
         )
-    dags = metadata.get("dags")
-    if not isinstance(dags, dict):
-        raise ValueError("embedded airflow metadata must contain a dags mapping")
+    # Keyed by Dag, but named for what the bundle provides: handlers for Dags declared elsewhere.
+    task_handlers = metadata.get("task_handlers")
+    if not isinstance(task_handlers, dict):
+        raise ValueError("embedded airflow metadata must contain a task_handlers mapping")
     return BundleMetadata(
-        dag_ids=frozenset(dags),
+        dag_ids=frozenset(task_handlers),
         supervisor_schema_version=extract_supervisor_schema_version(metadata),
     )
 
 
 def _read_bundle_headers(
     bundle_file: BinaryIO, *, path: pathlib.Path, file_size: int
-) -> tuple[_BundleLayout, bytes]:
-    """Read both comment payloads and check their declared ranges against the file."""
+) -> tuple[_BundleLayout, _VerifiedPayloads]:
+    """Read every comment payload and check their declared ranges against the file."""
     layout_payload = _read_prefixed_line(
         bundle_file,
         path=path,
@@ -368,9 +409,39 @@ def _read_bundle_headers(
     layout_line_size = len(_LAYOUT_COMMENT_PREFIX) + len(layout_payload) + 1
     metadata_start = layout_line_size + len(_METADATA_COMMENT_PREFIX)
     metadata_end = metadata_start + len(metadata_payload)
-    code_start = metadata_end + 1
     if (layout.metadata.start, layout.metadata.end) != (metadata_start, metadata_end):
         raise ValueError("bundle layout metadata offsets do not match the metadata section")
+    source_payload = _read_source_region(
+        bundle_file, path=path, start=metadata_end + 1 + len(_SOURCE_COMMENT_OPEN), layout=layout
+    )
+    code_start = layout.source.end + len(_SOURCE_COMMENT_CLOSE)
     if (layout.code.start, layout.code.end) != (code_start, file_size):
         raise ValueError("bundle layout code offsets do not match the executable section")
-    return layout, metadata_payload
+    return layout, _VerifiedPayloads(metadata=metadata_payload, source=source_payload)
+
+
+def _read_source_region(
+    bundle_file: BinaryIO, *, path: pathlib.Path, start: int, layout: _BundleLayout
+) -> bytes:
+    """Read the source block comment and check that it frames the declared range exactly."""
+    # The length is declared, not derivable, so the comment markers pin the range to real bytes.
+    if layout.source.start != start:
+        raise ValueError("bundle layout source offsets do not match the source section")
+    length = layout.source.end - start
+    if length > _MAX_SOURCE_BYTES:
+        raise ValueError(f"embedded airflow source exceeds {_MAX_SOURCE_BYTES} bytes")
+    try:
+        opener = bundle_file.read(len(_SOURCE_COMMENT_OPEN))
+        payload = bundle_file.read(length)
+        closer = bundle_file.read(len(_SOURCE_COMMENT_CLOSE))
+    except OSError as exc:
+        raise OSError(f"cannot read {path.name}: {exc}") from exc
+    if opener != _SOURCE_COMMENT_OPEN:
+        raise ValueError(f"{path.name} has no embedded airflow source after its metadata")
+    if len(payload) != length or closer != _SOURCE_COMMENT_CLOSE:
+        raise ValueError("embedded airflow source is not closed by its block comment")
+    # An unescaped ``*/`` would end the comment where Node reads the file,
+    # so bytes the layout calls source would execute while both digests still match.
+    if b"*/" in payload:
+        raise ValueError("embedded airflow source contains an unescaped block comment terminator")
+    return payload