From 18288e9108145f9e6330f7afe0d103a6187980e0 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 28 Sep 2026 06:50:57 +0000 Subject: [PATCH 1/3] Parse native TypeScript Dags with the Node coordinator A NodeCoordinator configured without bundles_root now hands out a Dag importer, so the Dag bundles it serves have their *.min.mjs bundles parsed by running them with node. Only files that start with the airflow-ts-pack layout header are parsed. The Dag source shown for such a Dag is the bundle's entry module, as the bundle embeds it, or a short notice when there is none to show. --- .../sdk/coordinators/node/_bundle_reader.py | 29 +++- .../sdk/coordinators/node/_dag_importer.py | 71 +++++++++ .../sdk/coordinators/node/coordinator.py | 21 ++- .../coordinators/node/test_bundle_reader.py | 7 + .../coordinators/node/test_coordinator.py | 56 +++++++ .../coordinators/node/test_dag_importer.py | 143 ++++++++++++++++++ 6 files changed, 314 insertions(+), 13 deletions(-) create mode 100644 task-sdk/src/airflow/sdk/coordinators/node/_dag_importer.py create mode 100644 task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py diff --git a/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py b/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py index 934bb8f6837b9..8bb39953ac124 100644 --- a/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py +++ b/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py @@ -40,6 +40,8 @@ if TYPE_CHECKING: from typing import BinaryIO +BUNDLE_SUFFIX = ".min.mjs" + # Format prefixes and whole-line limits must agree with the TypeScript encoder. _LAYOUT_COMMENT_PREFIX = b"//# airflowBundle=" _MAX_LAYOUT_LINE_BYTES = 4096 @@ -143,19 +145,32 @@ def read_bundle_source(bundle_path: pathlib.Path, dag_id: str | None = None) -> """ if dag_id is None: return None - payloads = _read_verified_payloads(bundle_path) + return _read_source(_read_verified_payloads(bundle_path), dag_id) + + +def read_bundle_entrypoint_source(bundle_path: pathlib.Path) -> str | None: + """Return the entrypoint source ``airflow-ts-pack`` embedded, or ``None`` when it embeds none.""" + return _read_source(_read_verified_payloads(bundle_path), None) + + +def _read_source(payloads: _VerifiedPayloads, dag_id: str | None) -> str | None: source_path = _resolve_source_path(payloads.metadata, dag_id) if source_path is None: return None try: payload = payloads.sources[source_path] except KeyError: - raise ValueError( - f"bundle declares no source region at path {source_path!r} for dag_id {dag_id!r}" - ) from None + owner = "the entrypoint" if dag_id is None else f"dag_id {dag_id!r}" + raise ValueError(f"bundle declares no source region at path {source_path!r} for {owner}") from None return _decode_source(payload) +def has_bundle_layout_prefix(bundle_path: pathlib.Path) -> bool: + """Return whether *bundle_path* starts with the layout comment ``airflow-ts-pack`` writes.""" + with bundle_path.open("rb") as bundle_file: + return bundle_file.read(len(_LAYOUT_COMMENT_PREFIX)) == _LAYOUT_COMMENT_PREFIX + + @attrs.define(frozen=True) class _VerifiedPayloads: """The metadata and per-path source payloads of a bundle whose digests have been checked.""" @@ -451,9 +466,9 @@ def _parse_bundle_metadata(payload: bytes) -> BundleMetadata: ) -def _resolve_source_path(payload: bytes, dag_id: str) -> str | None: +def _resolve_source_path(payload: bytes, dag_id: str | None) -> str | None: """ - Return the embedded source path for *dag_id*, or ``None``. + Return the embedded source path for *dag_id*, or for the entrypoint when it is ``None``. A native TypeScript Dag maps to its own file through ``dag_source_paths``. A Dag that is not mapped (one constructed dynamically, so the packer could not attribute a file) falls back to @@ -466,7 +481,7 @@ def _resolve_source_path(payload: bytes, dag_id: str) -> str | None: if not isinstance(metadata, dict): raise ValueError("embedded airflow metadata must contain a mapping") dag_source_paths = metadata.get("dag_source_paths") - if isinstance(dag_source_paths, dict): + 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 diff --git a/task-sdk/src/airflow/sdk/coordinators/node/_dag_importer.py b/task-sdk/src/airflow/sdk/coordinators/node/_dag_importer.py new file mode 100644 index 0000000000000..ef7d079458e39 --- /dev/null +++ b/task-sdk/src/airflow/sdk/coordinators/node/_dag_importer.py @@ -0,0 +1,71 @@ +# +# 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. +"""The Dag importer that parses native TypeScript Dags from packed bundles.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Final + +import structlog + +from airflow.sdk.coordinators._dag_importer import CoordinatorDagImporter +from airflow.sdk.coordinators.node._bundle_reader import ( + BUNDLE_SUFFIX, + has_bundle_layout_prefix, + read_bundle_entrypoint_source, +) +from airflow.sdk.importers.base import DagSourceCode + +if TYPE_CHECKING: + from structlog.typing import FilteringBoundLogger + + from airflow.sdk.coordinators.node.coordinator import NodeCoordinator + from airflow.sdk.importers.base import DagDefinition + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.node") + +_NO_SOURCE: Final = "// Source code is not available: the bundle embeds no entrypoint source.\n" + + +class NodeDagImporter(CoordinatorDagImporter): + """Parse the native Dags of packed ``*.min.mjs`` TypeScript bundles with a :class:`NodeCoordinator`.""" + + artifact_suffix = BUNDLE_SUFFIX + supported_extensions = [".mjs"] + coordinator: NodeCoordinator + + def might_contain_dag(self, definition: DagDefinition, safe_mode: bool) -> bool: + # The header identifies a packed bundle rather than guessing at content, so safe_mode keeps it. + with definition.as_file() as path: + try: + return has_bundle_layout_prefix(path) + except OSError: + # Keep the file, so parsing it records why it cannot be read. + return True + + def get_source_code(self, definition: DagDefinition) -> DagSourceCode: + """Return the embedded entrypoint source of the bundle, or a notice when there is none.""" + with definition.as_file() as path: + try: + source = read_bundle_entrypoint_source(path) + except (OSError, ValueError) as e: + log.warning( + "Cannot read the Dag source embedded in a TypeScript bundle", path=path, reason=str(e) + ) + source = f"// Source code is not available: {e}\n" + return DagSourceCode(source_code=source or _NO_SOURCE, language="typescript") diff --git a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py index 3dafbb9c33cbc..292893d619a24 100644 --- a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py +++ b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py @@ -15,7 +15,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -"""Node.js runtime coordinator that launches a Node.js subprocess for task execution.""" +"""Node.js runtime coordinator that launches a Node.js subprocess for task execution and Dag parsing.""" from __future__ import annotations @@ -33,7 +33,7 @@ walk_files, ) from airflow.sdk.coordinators._subprocess import SubprocessCoordinator -from airflow.sdk.coordinators.node._bundle_reader import read_bundle +from airflow.sdk.coordinators.node._bundle_reader import BUNDLE_SUFFIX, read_bundle if TYPE_CHECKING: from collections.abc import Sequence @@ -42,11 +42,10 @@ from typing_extensions import Self from airflow.sdk.api.datamodels._generated import TaskInstance + from airflow.sdk.coordinators.node._dag_importer import NodeDagImporter log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.node") -BUNDLE_SUFFIX = ".min.mjs" - def _is_bundle(path: pathlib.Path) -> bool: return path.name.endswith(BUNDLE_SUFFIX) @@ -98,7 +97,7 @@ def find(cls, bundles_root: Sequence[pathlib.Path], dag_id: str) -> Self: @attrs.define(kw_only=True) class NodeCoordinator(SubprocessCoordinator): """ - Coordinator that launches a Node.js subprocess for task execution. + Coordinator that launches a Node.js subprocess for task execution and Dag parsing. Configuration is taken from the ``[sdk] coordinators`` entry that constructs this instance:: @@ -117,7 +116,8 @@ class NodeCoordinator(SubprocessCoordinator): ``"node"``, which relies on ``$PATH``). :param bundles_root: Directories searched recursively, in order, for the first verified ``*.min.mjs`` bundle declaring the task instance's Dag. See :class:`SubprocessCoordinator` - for its interaction with ``dag_bundle_name``. + for its interaction with ``dag_bundle_name``. Without it, the coordinator also parses the + native Dags of the ``*.min.mjs`` bundles in the Dag bundles it serves. :param task_startup_timeout: Maximum time the coordinator waits for a task process to start, in seconds. The default is 10 seconds. """ @@ -134,3 +134,12 @@ def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], roots = self._get_scan_roots() bundle = _Bundle.find(roots, what.dag_id) return [self.node_executable, os.fspath(bundle.path)], bundle.schema_version + + @classmethod + def get_dag_importer_class(cls) -> type[NodeDagImporter]: + from airflow.sdk.coordinators.node._dag_importer import NodeDagImporter + + return NodeDagImporter + + def _build_parse_dag_command(self, *, path: pathlib.Path) -> tuple[list[str], str | None]: + return [self.node_executable, os.fspath(path)], read_bundle(path).supervisor_schema_version diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py b/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py index 93440a2118223..77156d16e44b0 100644 --- a/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py +++ b/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py @@ -46,6 +46,7 @@ _digest_cache, _hash_region, read_bundle, + read_bundle_entrypoint_source, read_bundle_source, ) @@ -364,6 +365,12 @@ def test_reads_one_source_region_per_dag(self, tmp_path): assert read_bundle(bundle).dag_ids == frozenset({"sales", "inventory"}) assert read_bundle_source(bundle, "sales") == "export const sales = 1;\n" assert read_bundle_source(bundle, "inventory") == "export const inv = 2;\n" + assert read_bundle_entrypoint_source(bundle) == "export const sales = 1;\n" + + def test_reads_no_entrypoint_source_when_none_is_declared(self, tmp_path): + bundle = write_bundle(tmp_path, "sales", entrypoint_path=None) + + assert read_bundle_entrypoint_source(bundle) is None def test_returns_none_without_a_dag_id(self, tmp_path): # A Dag owned by another language (Python) shows its own source; the caller omits dag_id diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py index 146cb91c1d94c..b439c7ddc47f6 100644 --- a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py +++ b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py @@ -35,7 +35,11 @@ from airflow.sdk.api.datamodels._generated import TaskInstance from airflow.sdk.coordinators.node import _bundle_reader as _reader from airflow.sdk.coordinators.node._bundle_reader import _digest_cache +from airflow.sdk.coordinators.node._dag_importer import NodeDagImporter from airflow.sdk.coordinators.node.coordinator import NodeCoordinator, _Bundle +from airflow.sdk.importers import DagImporterRegistry, reset_importer_registry + +from tests_common.test_utils.config import conf_vars SCHEMA_VERSION = "2026-06-16" @@ -141,6 +145,58 @@ def test_build_execute_task_command_returns_node_bundle_and_schema_version(self, assert schema_version == SCHEMA_VERSION +class TestNodeCoordinatorParseDagCommand: + def test_returns_node_and_bundle_schema_version(self, tmp_path): + bundle = write_bundle(tmp_path / "typescript", "native_dag") + coordinator = NodeCoordinator(node_executable="/opt/node/bin/node") + + command, schema_version = coordinator._build_parse_dag_command(path=bundle) + + assert command == ["/opt/node/bin/node", str(bundle)] + assert schema_version == SCHEMA_VERSION + + def test_tampered_bundle_raises(self, tmp_path): + bundle = write_bundle(tmp_path, "native_dag") + mutate_byte(bundle, int(read_layout(bundle)["code"]["start"], 16)) # type: ignore[index, call-overload] + + with pytest.raises(ValueError, match="code SHA-256 mismatch"): + NodeCoordinator()._build_parse_dag_command(path=bundle) + + +class TestNodeCoordinatorDagImporter: + @pytest.fixture(autouse=True) + def _reset_registry(self): + reset_importer_registry() + yield + reset_importer_registry() + + def test_hands_out_an_importer_bound_to_itself(self): + coordinator = NodeCoordinator() + + importer = coordinator.get_dag_importer() + + assert NodeCoordinator.get_dag_importer_class() is NodeDagImporter + assert isinstance(importer, NodeDagImporter) + assert importer.coordinator is coordinator + assert importer.supported_extensions == [".mjs"] + assert importer.can_handle("dags/bundle.min.mjs") is True + assert importer.can_handle("dags/helper.mjs") is False + + @pytest.mark.parametrize( + ("kwargs", "parses"), + [({}, True), ({"bundles_root": ["/opt/airflow/ts-bundles"]}, False)], + ids=["task-bundle", "explicit-root"], + ) + def test_registry_routes_bundles_only_to_a_serving_coordinator(self, kwargs, parses): + coordinators = { + "ts": {"classpath": "airflow.sdk.coordinators.node.NodeCoordinator", "kwargs": kwargs} + } + with conf_vars({("sdk", "coordinators"): json.dumps(coordinators)}): + importer = DagImporterRegistry.from_config("dags-folder").get_importer("dags/bundle.min.mjs") + + assert isinstance(importer, NodeDagImporter) is parses + + class TestBundleFind: @pytest.mark.parametrize( "name", diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py b/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py new file mode 100644 index 0000000000000..98bbe0f0cda4f --- /dev/null +++ b/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py @@ -0,0 +1,143 @@ +# +# 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. +from __future__ import annotations + +import pathlib +from types import SimpleNamespace + +import pytest +from task_sdk.coordinators.node._bundle_test_utils import ( + BUNDLE_NAME, + LAYOUT_PREFIX, + SOURCE_OPEN, + mutate_byte, + read_layout, + write_bundle, +) + +from airflow.sdk.coordinators.node._bundle_reader import _digest_cache +from airflow.sdk.coordinators.node._dag_importer import NodeDagImporter +from airflow.sdk.coordinators.node.coordinator import NodeCoordinator +from airflow.sdk.importers import DagSourceCode, FilesystemDagDefinition + + +@pytest.fixture(autouse=True) +def clear_digest_cache(): + _digest_cache.clear() + + +@pytest.fixture +def importer() -> NodeDagImporter: + return NodeDagImporter(coordinator=NodeCoordinator()) + + +def _mutate_section(bundle: pathlib.Path, section: str) -> None: + mutate_byte(bundle, int(read_layout(bundle)[section]["start"], 16)) # type: ignore[index, call-overload] + + +def test_lists_only_packed_bundles(importer, tmp_path): + nested = write_bundle(tmp_path / "team", "sales") + tampered = write_bundle(tmp_path, "inventory", name="tampered.min.mjs") + _mutate_section(tampered, "code") + write_bundle(tmp_path, "orders", name="plain.mjs") + (tmp_path / "vendor.min.mjs").write_bytes(b"export {};\n") + + definitions = importer.list_dag_definitions(SimpleNamespace(name="dags-folder", path=tmp_path)) + + # A tampered bundle keeps its header, so parsing it reports the integrity failure. + assert sorted(d.path for d in definitions) == sorted([nested, tampered]) + + +class TestMightContainDag: + @pytest.mark.parametrize("safe_mode", [True, False]) + @pytest.mark.parametrize( + ("content", "expected"), + [ + (None, True), + (b"export {};\n", False), + (b"", False), + (LAYOUT_PREFIX[:-1], False), + ], + ids=["bundle", "plain-module", "empty", "truncated-prefix"], + ) + def test_checks_the_layout_header(self, importer, tmp_path, content, expected, safe_mode): + path = write_bundle(tmp_path, "sales") + if content is not None: + path.write_bytes(content) + + assert importer.might_contain_dag(FilesystemDagDefinition(path), safe_mode) is expected + + def test_keeps_an_unreadable_file(self, importer, tmp_path, monkeypatch): + path = write_bundle(tmp_path, "sales") + original_open = pathlib.Path.open + + def raise_permission_error(self, *args, **kwargs): + if self.name == BUNDLE_NAME: + raise PermissionError("denied") + return original_open(self, *args, **kwargs) + + monkeypatch.setattr(pathlib.Path, "open", raise_permission_error) + + assert importer.might_contain_dag(FilesystemDagDefinition(path), True) is True + + +class TestGetSourceCode: + def test_returns_the_entrypoint_source_for_every_dag(self, importer, tmp_path): + sales = '/* the */ export const sales = new Dag({ dagId: "sales" });\n' + inventory = 'export const inventory = new Dag({ dagId: "inventory" });\n' + path = write_bundle( + tmp_path, + "sales", + "inventory", + sources=[("sales.ts", sales.encode()), ("inventory.ts", inventory.encode())], + dag_source_paths={"sales": "sales.ts", "inventory": "inventory.ts"}, + entrypoint_path="sales.ts", + ) + + assert importer.get_source_code(FilesystemDagDefinition(path)) == DagSourceCode(sales, "typescript") + + def test_returns_a_notice_without_an_entrypoint_source(self, importer, tmp_path): + path = write_bundle(tmp_path, "sales", entrypoint_path=None) + + source_code = importer.get_source_code(FilesystemDagDefinition(path)) + + assert source_code == DagSourceCode( + "// Source code is not available: the bundle embeds no entrypoint source.\n", "typescript" + ) + + @pytest.mark.parametrize( + ("break_bundle", "reason"), + [ + ( + lambda path: mutate_byte(path, path.read_bytes().index(SOURCE_OPEN) + len(SOURCE_OPEN)), + "source main.ts SHA-256 mismatch", + ), + (lambda path: write_bundle(path.parent, "sales", source=b"x" * (1024 * 1024 + 1)), "exceeds"), + (lambda path: path.unlink(), "cannot read bundle.min.mjs"), + ], + ids=["tampered", "oversize", "missing"], + ) + def test_returns_a_notice_when_the_source_cannot_be_read(self, importer, tmp_path, break_bundle, reason): + path = write_bundle(tmp_path, "sales") + break_bundle(path) + + source_code = importer.get_source_code(FilesystemDagDefinition(path)) + + assert source_code.language == "typescript" + assert source_code.source_code.startswith("// Source code is not available: ") + assert reason in source_code.source_code From 69b4b262dae4a70d4061a816454edbcde09d449f Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 28 Sep 2026 06:51:16 +0000 Subject: [PATCH 2/3] Document parsing native TypeScript Dags Describe the coordinator configuration that parses native Dags, what the Dag processor needs for it, what the Code view shows, the cost of handler-only bundles, and that cluster policies and some CLI commands do not apply to a native Dag. --- .../language-sdks/typescript.rst | 54 +++++++++++++++++-- 1 file changed, 50 insertions(+), 4 deletions(-) diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst index 47070c581da15..50998f100b475 100644 --- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst +++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst @@ -48,7 +48,8 @@ The SDK is the ``apache-airflow-ts-sdk`` package (ESM-only). It is currently in Prerequisites ------------- -* Node.js 22 or later must be available on the Airflow worker nodes. +* Node.js 22 or later must be available on the Airflow worker nodes, and on the Dag processor to parse + Dags declared in TypeScript. * The packed bundle (a single ``bundle.min.mjs`` file, see :ref:`typescript-sdk/build`) must be accessible from the worker, under a directory the coordinator scans. * The ``apache-airflow-task-sdk`` package (installed with Airflow) provides the coordinator; no additional @@ -236,7 +237,8 @@ task instance. bundles in ``bundles_root``) only need to be present wherever tasks actually execute. With ``CeleryExecutor``, setting them on the Celery workers is sufficient. With ``LocalExecutor``, tasks run inside the scheduler process, so they must be present where the scheduler can read them. The API server - and Dag processor do not need them. + does not need them, and the Dag processor needs them only to parse Dags declared in TypeScript, + see :ref:`typescript-sdk/native-parsing`. .. _typescript-sdk/native-dag: @@ -244,8 +246,8 @@ Declaring a Dag in TypeScript ----------------------------- A ``Dag`` is declared on this side rather than in Python: its schedule, its tasks, their options and -the edges between them are all written in TypeScript. The surface is still growing, so a Dag declared -this way is not served to Airflow yet. +the edges between them are all written in TypeScript. Airflow parses such a Dag from its packed bundle, +see :ref:`typescript-sdk/native-parsing`. ``dag.task(taskId, handler)`` returns a *factory*. A handler takes one object of named arguments, and calling the factory names each input, so the call graph is the task graph: @@ -514,6 +516,42 @@ instance's Dag. The artifact's name does not matter beyond that suffix, so one r and a Dag is routed to whichever declares it. If multiple bundles declare the same Dag, the first configured root wins, and within a root the first in sorted path order. +.. _typescript-sdk/native-parsing: + +Parsing native Dags +------------------- + +A Dag declared in TypeScript ships in its packed bundle, so the bundle goes into a Dag bundle, +next to any Python Dag files. A coordinator configured without ``bundles_root`` parses the ``*.min.mjs`` +bundles of the Dag bundles it serves: every Dag bundle, or only the one ``dag_bundle_name`` names. +The same entry runs the tasks: + +.. code-block:: ini + + [sdk] + coordinators = { + "ts": { + "classpath": "airflow.sdk.coordinators.node.NodeCoordinator", + "kwargs": {"node_executable": "/usr/local/bin/node"} + } + } + queue_to_coordinator = {"typescript": "ts"} + +At most one such coordinator can serve a Dag bundle, and a coordinator with ``bundles_root`` parses nothing. + +The Dag processor runs each bundle with ``node`` to parse it, so it needs Node.js and this ``[sdk]`` +configuration. It parses only files that end in ``.min.mjs`` and start with the header ``airflow-ts-pack`` +writes. A bundle that fails its integrity check, or whose Dags cannot be serialized or fail validation, +such as a cycle drawn with ``before`` and ``after``, is reported as an import error. + +The Code view shows the bundle's entry module for each of its Dags, as ``airflow-ts-pack`` embeds it +(see :ref:`typescript-sdk/build`). If the source cannot be read from the bundle, the view shows a short +notice instead. + +A bundle that only registers ``TaskHandler`` objects is parsed too when it sits in a served Dag bundle, +since its metadata does not say whether it declares Dags. Each parse then launches ``node`` once and +finds no Dags, so keep such bundles in a ``bundles_root`` directory to avoid that cost. + .. _typescript-sdk/coordinator-config: :class:`~airflow.sdk.coordinators.node.NodeCoordinator` configuration @@ -559,12 +597,20 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the * Leave both unset (the default) to load the bundle from the **task's own** Dag bundle, pinned to the version the run was created with. + Without ``bundles_root``, the coordinator also parses the native Dags of the Dag bundles it serves, + see :ref:`typescript-sdk/native-parsing`. + Limitations ----------- * **A Python stub Dag is still required.** The Execution API does not yet carry Dag structure for non-Python languages, so task names and dependencies are declared in Python with :func:`@task.stub `. +* **Cluster policies do not apply to a native Dag.** ``dag_policy`` and ``task_policy`` are not run on + it, so they cannot change or reject it. +* **Some CLI commands do not take a native Dag.** ``airflow dags test``, ``tasks test``, ``tasks render`` + and ``tasks list`` refuse it, and ``airflow dags reserialize`` does not store the Dags of + ``*.min.mjs`` bundles, which only the Dag processor stores. * **Beta status.** The SDK API may change in incompatible ways between releases. * **One Node.js subprocess per task instance.** Tasks that need to share in-process state between instances should use XCom or an external store instead. From fe683157090787c921136d54d8d3ce5dea57e906 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 28 Sep 2026 07:12:22 +0000 Subject: [PATCH 3/3] Share the bundle section mutation helper between Node tests --- .../task_sdk/coordinators/node/_bundle_test_utils.py | 5 +++++ .../tests/task_sdk/coordinators/node/test_coordinator.py | 3 ++- .../tests/task_sdk/coordinators/node/test_dag_importer.py | 8 ++------ 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py b/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py index 8039cbc032cdd..1d222d469c25d 100644 --- a/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py +++ b/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py @@ -173,3 +173,8 @@ def mutate_byte(bundle: pathlib.Path, offset: int) -> None: contents = bytearray(bundle.read_bytes()) contents[offset] = ord("A") if contents[offset] != ord("A") else ord("B") bundle.write_bytes(contents) + + +def mutate_section(bundle: pathlib.Path, section: str) -> None: + """Change the first byte of a layout section, so its digest no longer matches.""" + mutate_byte(bundle, int(read_layout(bundle)[section]["start"], 16)) # type: ignore[index, call-overload] diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py index b439c7ddc47f6..b81ab84170528 100644 --- a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py +++ b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py @@ -27,6 +27,7 @@ from task_sdk.coordinators.node._bundle_test_utils import ( BUNDLE_NAME, mutate_byte, + mutate_section, read_layout, write_bundle, ) @@ -157,7 +158,7 @@ def test_returns_node_and_bundle_schema_version(self, tmp_path): def test_tampered_bundle_raises(self, tmp_path): bundle = write_bundle(tmp_path, "native_dag") - mutate_byte(bundle, int(read_layout(bundle)["code"]["start"], 16)) # type: ignore[index, call-overload] + mutate_section(bundle, "code") with pytest.raises(ValueError, match="code SHA-256 mismatch"): NodeCoordinator()._build_parse_dag_command(path=bundle) diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py b/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py index 98bbe0f0cda4f..c3e22869513c1 100644 --- a/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py +++ b/task-sdk/tests/task_sdk/coordinators/node/test_dag_importer.py @@ -26,7 +26,7 @@ LAYOUT_PREFIX, SOURCE_OPEN, mutate_byte, - read_layout, + mutate_section, write_bundle, ) @@ -46,14 +46,10 @@ def importer() -> NodeDagImporter: return NodeDagImporter(coordinator=NodeCoordinator()) -def _mutate_section(bundle: pathlib.Path, section: str) -> None: - mutate_byte(bundle, int(read_layout(bundle)[section]["start"], 16)) # type: ignore[index, call-overload] - - def test_lists_only_packed_bundles(importer, tmp_path): nested = write_bundle(tmp_path / "team", "sales") tampered = write_bundle(tmp_path, "inventory", name="tampered.min.mjs") - _mutate_section(tampered, "code") + mutate_section(tampered, "code") write_bundle(tmp_path, "orders", name="plain.mjs") (tmp_path / "vendor.min.mjs").write_bytes(b"export {};\n")