From d887d7ff9651e2f48d0b8ada50c57a663fd1152c Mon Sep 17 00:00:00 2001 From: Shivam Rastogi <6463385+shivaam@users.noreply.github.com> Date: Sun, 14 Jun 2026 17:25:06 -0700 Subject: [PATCH 1/5] Add TypeScript task coordinator --- .../sdk/coordinators/typescript/__init__.py | 28 +++ .../coordinators/typescript/coordinator.py | 158 ++++++++++++++ .../coordinators/typescript/__init__.py | 17 ++ .../typescript/test_coordinator.py | 203 ++++++++++++++++++ 4 files changed, 406 insertions(+) create mode 100644 task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py create mode 100644 task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py create mode 100644 task-sdk/tests/task_sdk/coordinators/typescript/__init__.py create mode 100644 task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py diff --git a/task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py b/task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py new file mode 100644 index 0000000000000..75d8bdecb4ea1 --- /dev/null +++ b/task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py @@ -0,0 +1,28 @@ +# +# 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. +""" +TypeScript runtime coordinator for the Apache Airflow Task SDK. + +See :class:`TypescriptCoordinator` for details. +""" + +from __future__ import annotations + +from airflow.sdk.coordinators.typescript.coordinator import TypescriptCoordinator + +__all__ = ["TypescriptCoordinator"] diff --git a/task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py b/task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py new file mode 100644 index 0000000000000..15c8fe3c10e1c --- /dev/null +++ b/task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py @@ -0,0 +1,158 @@ +# +# 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. +"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution.""" + +from __future__ import annotations + +import os +import pathlib +from typing import TYPE_CHECKING, Any + +import attrs +import structlog +import yaml + +from airflow.sdk.coordinators._subprocess import SubprocessCoordinator +from airflow.sdk.execution_time.schema import get_schema_version_migrator + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + + from airflow.sdk.api.datamodels._generated import TaskInstance + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript") + +BUNDLE_FILENAME = "bundle.mjs" +METADATA_FILENAME = "airflow-metadata.yaml" + + +def _validate_schema_version(instance, _, value) -> str: + return get_schema_version_migrator().resolve_version(str(value)) + + +@attrs.define +class _TypescriptBundle: + path: pathlib.Path + schema_version: str = attrs.field(validator=_validate_schema_version) + + +def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]: + if not metadata_path.is_file(): + raise ValueError(f"missing {METADATA_FILENAME}") + + try: + data = yaml.safe_load(metadata_path.read_text(encoding="utf-8")) + except OSError as exc: + raise ValueError(f"cannot read {METADATA_FILENAME}: {exc}") from exc + except yaml.YAMLError as exc: + raise ValueError(f"cannot parse {METADATA_FILENAME}: {exc}") from exc + + if not isinstance(data, dict): + raise ValueError(f"{METADATA_FILENAME} must contain a mapping") + return data + + +def _supervisor_schema_version(metadata: dict[str, Any]) -> str: + sdk = metadata.get("sdk") + if not isinstance(sdk, dict): + raise ValueError("missing sdk metadata mapping") + + value = sdk.get("supervisor_schema_version") + if not isinstance(value, str) or not value: + raise ValueError("missing or invalid sdk.supervisor_schema_version") + return value + + +def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _TypescriptBundle: + """ + Locate the ``.mjs`` entry point in *bundles_root*. + + Scans each configured directory for ``bundle.mjs`` and reads the sibling + ``airflow-metadata.yaml`` for the bundle's supervisor schema version. + + This is an ordered fallback search, not Dag/task-aware multi-bundle + routing. The first bundle found wins. A future version can use the + metadata's ``dags`` section together with ``TaskInstance.dag_id`` and + ``TaskInstance.task_id`` to select the bundle that owns a specific task. + """ + for root in bundles_root: + candidate = root / BUNDLE_FILENAME + if not candidate.is_file(): + continue + metadata = _read_bundle_metadata(root / METADATA_FILENAME) + log.debug("Selected TypeScript bundle", path=candidate, root=root) + return _TypescriptBundle( + path=candidate, + schema_version=_supervisor_schema_version(metadata), + ) + + searched = os.pathsep.join(os.fspath(p.resolve()) for p in bundles_root) + raise FileNotFoundError(f"Cannot find {BUNDLE_FILENAME} in {searched}") + + +def _convert_bundles_root( + value: None | os.PathLike[str] | pathlib.Path | list[os.PathLike[str] | pathlib.Path], +) -> list[pathlib.Path]: + if value is None: + return [] + if isinstance(value, (str, os.PathLike, pathlib.Path)): + return [pathlib.Path(value).expanduser()] + return [pathlib.Path(v).expanduser() for v in value] + + +@attrs.define(kw_only=True) +class TypescriptCoordinator(SubprocessCoordinator): + """ + Coordinator that launches a Node.js subprocess for task execution. + + Configuration is taken from the ``[sdk] coordinators`` entry that constructs + this instance:: + + { + "ts": { + "classpath": "airflow.sdk.coordinators.typescript.TypescriptCoordinator", + "kwargs": { + "node_executable": "node", + "bundles_root": ["/opt/airflow/ts-bundles"], + }, + } + } + + :param node_executable: Path to the ``node`` binary (defaults to + ``"node"``, which relies on ``$PATH``). + :param bundles_root: Ordered list of directories scanned for a usable + TypeScript bundle. Each bundle directory must contain ``bundle.mjs`` + and ``airflow-metadata.yaml``. This is a fallback search path; it does + not yet route different Dag/task pairs to different bundles. + :param task_startup_timeout: Maximum time the coordinator waits for a task + process to start, in seconds. The default is 10 seconds. + """ + + node_executable: str = "node" + bundles_root: list[pathlib.Path] = attrs.field( + converter=_convert_bundles_root, + validator=attrs.validators.min_len(1), + ) + + def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]: + # Multi-bundle routing should be added here by passing `what.dag_id` and + # `what.task_id` into bundle selection and matching against metadata["dags"]. + bundle = _find_bundle(self.bundles_root) + return [self.node_executable, os.fspath(bundle.path)], bundle.schema_version diff --git a/task-sdk/tests/task_sdk/coordinators/typescript/__init__.py b/task-sdk/tests/task_sdk/coordinators/typescript/__init__.py new file mode 100644 index 0000000000000..217e5db960782 --- /dev/null +++ b/task-sdk/tests/task_sdk/coordinators/typescript/__init__.py @@ -0,0 +1,17 @@ +# +# 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. diff --git a/task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py b/task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py new file mode 100644 index 0000000000000..ed51e4aef05b8 --- /dev/null +++ b/task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py @@ -0,0 +1,203 @@ +# +# 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 + +import pytest +from uuid6 import uuid7 + +from airflow.sdk.api.datamodels._generated import TaskInstance +from airflow.sdk.coordinators.typescript.coordinator import ( + TypescriptCoordinator, + _find_bundle, +) + +SCHEMA_VERSION = "2026-06-16" + + +def _make_ti(dag_id: str = "test_dag", queue: str = "ts") -> TaskInstance: + return TaskInstance( + id=uuid7(), + dag_version_id=uuid7(), + task_id="test_task", + dag_id=dag_id, + run_id="run_1", + try_number=1, + map_index=-1, + queue=queue, + ) + + +def write_bundle(root: pathlib.Path, schema_version: str = SCHEMA_VERSION) -> pathlib.Path: + bundle = root / "bundle.mjs" + bundle.write_text("export {};\n", encoding="utf-8") + (root / "airflow-metadata.yaml").write_text( + "\n".join( + [ + 'airflow_bundle_metadata_version: "1.0"', + "sdk:", + " language: typescript", + ' version: "0.1.0"', + f' supervisor_schema_version: "{schema_version}"', + "source: src/airflow.ts", + "dags:", + " test_dag:", + " tasks:", + " - test_task", + "", + ] + ), + encoding="utf-8", + ) + return bundle + + +class TestTypescriptCoordinatorAttributes: + def test_default_kwargs(self): + coordinator = TypescriptCoordinator(bundles_root="/airflow/ts-bundles") + + assert coordinator.node_executable == "node" + assert coordinator.bundles_root == [pathlib.Path("/airflow/ts-bundles")] + assert coordinator.task_startup_timeout == 10.0 + + def test_custom_kwargs(self): + coordinator = TypescriptCoordinator( + node_executable="/opt/node/bin/node", + bundles_root=["/airflow/ts-bundles", "~/extra-bundles"], + task_startup_timeout=30.0, + ) + + assert coordinator.node_executable == "/opt/node/bin/node" + assert coordinator.bundles_root == [ + pathlib.Path("/airflow/ts-bundles"), + pathlib.Path("~/extra-bundles").expanduser(), + ] + assert coordinator.task_startup_timeout == 30.0 + + def test_bundles_root_is_required(self): + with pytest.raises(ValueError, match="Length of 'bundles_root' must be >= 1"): + TypescriptCoordinator(bundles_root=None) + + +class TestTypescriptCoordinatorBundleSelection: + def test_find_bundle_returns_bundle_mjs(self, tmp_path): + bundle = write_bundle(tmp_path) + + found = _find_bundle([tmp_path]) + + assert found.path == bundle + assert found.schema_version == SCHEMA_VERSION + + def test_find_bundle_checks_multiple_roots(self, tmp_path): + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + bundle = write_bundle(second) + + found = _find_bundle([first, second]) + + assert found.path == bundle + assert found.schema_version == SCHEMA_VERSION + + def test_find_bundle_ignores_other_mjs_names(self, tmp_path): + (tmp_path / "tasks.mjs").write_text("export {};\n") + + with pytest.raises(FileNotFoundError, match="Cannot find bundle.mjs"): + _find_bundle([tmp_path]) + + def test_find_bundle_rejects_bundle_without_metadata(self, tmp_path): + (tmp_path / "bundle.mjs").write_text("export {};\n", encoding="utf-8") + + with pytest.raises(ValueError, match="missing airflow-metadata.yaml"): + _find_bundle([tmp_path]) + + @pytest.mark.parametrize( + ("metadata", "message"), + [ + ("[not-a-mapping]\n", "airflow-metadata.yaml must contain a mapping"), + ("sdk: [not-a-mapping]\n", "missing sdk metadata mapping"), + ("sdk:\n language: typescript\n", "missing or invalid sdk.supervisor_schema_version"), + ], + ) + def test_find_bundle_rejects_invalid_metadata_shape(self, tmp_path, metadata, message): + (tmp_path / "bundle.mjs").write_text("export {};\n", encoding="utf-8") + (tmp_path / "airflow-metadata.yaml").write_text(metadata, encoding="utf-8") + + with pytest.raises(ValueError, match=message): + _find_bundle([tmp_path]) + + def test_find_bundle_reports_unreadable_metadata(self, tmp_path, monkeypatch): + write_bundle(tmp_path) + + def raise_os_error(self, *args, **kwargs): + if self.name == "airflow-metadata.yaml": + raise PermissionError("denied") + return original_read_text(self, *args, **kwargs) + + original_read_text = pathlib.Path.read_text + monkeypatch.setattr(pathlib.Path, "read_text", raise_os_error) + + with pytest.raises(ValueError, match="cannot read airflow-metadata.yaml"): + _find_bundle([tmp_path]) + + def test_find_bundle_rejects_invalid_schema_version(self, tmp_path): + write_bundle(tmp_path, schema_version="banana") + + with pytest.raises(ValueError, match="Version 'banana' not found"): + _find_bundle([tmp_path]) + + def test_find_bundle_fails_on_invalid_first_bundle(self, tmp_path): + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + (first / "bundle.mjs").write_text("export {};\n", encoding="utf-8") + write_bundle(second) + + with pytest.raises(ValueError, match="missing airflow-metadata.yaml"): + _find_bundle([first, second]) + + def test_find_bundle_raises_with_searched_roots(self, tmp_path): + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + + with pytest.raises(FileNotFoundError) as exc_info: + _find_bundle([first, second]) + + msg = str(exc_info.value) + assert str(first.resolve()) in msg + assert str(second.resolve()) in msg + + +class TestTypescriptCoordinatorExecuteTaskCommand: + def test_build_execute_task_command_returns_node_bundle_and_schema_version(self, tmp_path): + bundle = write_bundle(tmp_path) + coordinator = TypescriptCoordinator( + node_executable="/opt/node/bin/node", + bundles_root=tmp_path, + ) + + command, schema_version = coordinator._build_execute_task_command(what=_make_ti()) + + assert command == ["/opt/node/bin/node", str(bundle)] + assert schema_version == SCHEMA_VERSION From 530aa997c965630c8f2d543a60ae80f0108f64ed Mon Sep 17 00:00:00 2001 From: Shivam Rastogi <6463385+shivaam@users.noreply.github.com> Date: Sun, 14 Jun 2026 17:42:08 -0700 Subject: [PATCH 2/5] Add newsfragment for TypeScript coordinator --- airflow-core/newsfragments/68548.feature.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 airflow-core/newsfragments/68548.feature.rst diff --git a/airflow-core/newsfragments/68548.feature.rst b/airflow-core/newsfragments/68548.feature.rst new file mode 100644 index 0000000000000..107eff3cf7bcd --- /dev/null +++ b/airflow-core/newsfragments/68548.feature.rst @@ -0,0 +1 @@ +Add a TypeScript task coordinator for running TypeScript task bundles through coordinator mode. From c34c2acd05d20015961caf9a14899398c5f78c05 Mon Sep 17 00:00:00 2001 From: Shivam Rastogi <6463385+shivaam@users.noreply.github.com> Date: Sun, 28 Jun 2026 12:16:25 -0700 Subject: [PATCH 3/5] Rename TypeScript coordinator to Node coordinator --- .../{typescript => node}/__init__.py | 8 ++-- .../{typescript => node}/coordinator.py | 39 +++++++++++------ .../{typescript => node}/__init__.py | 0 .../{typescript => node}/test_coordinator.py | 42 ++++++++++--------- 4 files changed, 52 insertions(+), 37 deletions(-) rename task-sdk/src/airflow/sdk/coordinators/{typescript => node}/__init__.py (78%) rename task-sdk/src/airflow/sdk/coordinators/{typescript => node}/coordinator.py (80%) rename task-sdk/tests/task_sdk/coordinators/{typescript => node}/__init__.py (100%) rename task-sdk/tests/task_sdk/coordinators/{typescript => node}/test_coordinator.py (84%) diff --git a/task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py b/task-sdk/src/airflow/sdk/coordinators/node/__init__.py similarity index 78% rename from task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py rename to task-sdk/src/airflow/sdk/coordinators/node/__init__.py index 75d8bdecb4ea1..9030ca17ebbfa 100644 --- a/task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py +++ b/task-sdk/src/airflow/sdk/coordinators/node/__init__.py @@ -16,13 +16,13 @@ # specific language governing permissions and limitations # under the License. """ -TypeScript runtime coordinator for the Apache Airflow Task SDK. +Node.js runtime coordinator for the Apache Airflow Task SDK. -See :class:`TypescriptCoordinator` for details. +See :class:`NodeCoordinator` for details. """ from __future__ import annotations -from airflow.sdk.coordinators.typescript.coordinator import TypescriptCoordinator +from airflow.sdk.coordinators.node.coordinator import NodeCoordinator -__all__ = ["TypescriptCoordinator"] +__all__ = ["NodeCoordinator"] diff --git a/task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py similarity index 80% rename from task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py rename to task-sdk/src/airflow/sdk/coordinators/node/coordinator.py index 15c8fe3c10e1c..458f19f039767 100644 --- a/task-sdk/src/airflow/sdk/coordinators/typescript/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. -"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution.""" +"""Node.js runtime coordinator that launches a Node.js subprocess for task execution.""" from __future__ import annotations @@ -37,7 +37,7 @@ from airflow.sdk.api.datamodels._generated import TaskInstance -log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript") +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.node") BUNDLE_FILENAME = "bundle.mjs" METADATA_FILENAME = "airflow-metadata.yaml" @@ -48,7 +48,7 @@ def _validate_schema_version(instance, _, value) -> str: @attrs.define -class _TypescriptBundle: +class _NodeBundle: path: pathlib.Path schema_version: str = attrs.field(validator=_validate_schema_version) @@ -58,7 +58,8 @@ def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]: raise ValueError(f"missing {METADATA_FILENAME}") try: - data = yaml.safe_load(metadata_path.read_text(encoding="utf-8")) + with metadata_path.open(encoding="utf-8") as metadata_file: + data = yaml.safe_load(metadata_file) except OSError as exc: raise ValueError(f"cannot read {METADATA_FILENAME}: {exc}") from exc except yaml.YAMLError as exc: @@ -80,7 +81,7 @@ def _supervisor_schema_version(metadata: dict[str, Any]) -> str: return value -def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _TypescriptBundle: +def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _NodeBundle: """ Locate the ``.mjs`` entry point in *bundles_root*. @@ -92,18 +93,30 @@ def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _TypescriptBundle: metadata's ``dags`` section together with ``TaskInstance.dag_id`` and ``TaskInstance.task_id`` to select the bundle that owns a specific task. """ + rejected: list[tuple[pathlib.Path, str]] = [] for root in bundles_root: candidate = root / BUNDLE_FILENAME if not candidate.is_file(): continue - metadata = _read_bundle_metadata(root / METADATA_FILENAME) - log.debug("Selected TypeScript bundle", path=candidate, root=root) - return _TypescriptBundle( - path=candidate, - schema_version=_supervisor_schema_version(metadata), - ) + try: + metadata = _read_bundle_metadata(root / METADATA_FILENAME) + log.debug("Selected TypeScript bundle", path=candidate, root=root) + return _NodeBundle( + path=candidate, + schema_version=_supervisor_schema_version(metadata), + ) + except (TypeError, ValueError) as exc: + log.debug( + "TypeScript bundle metadata rejected; skipping", path=candidate, root=root, error=str(exc) + ) + rejected.append((candidate.resolve(), str(exc))) searched = os.pathsep.join(os.fspath(p.resolve()) for p in bundles_root) + if rejected: + details = "; ".join(f"{path}: {reason}" for path, reason in rejected) + raise FileNotFoundError( + f"Cannot find usable TypeScript bundle in {searched}: matching bundles were rejected ({details})" + ) raise FileNotFoundError(f"Cannot find {BUNDLE_FILENAME} in {searched}") @@ -118,7 +131,7 @@ def _convert_bundles_root( @attrs.define(kw_only=True) -class TypescriptCoordinator(SubprocessCoordinator): +class NodeCoordinator(SubprocessCoordinator): """ Coordinator that launches a Node.js subprocess for task execution. @@ -127,7 +140,7 @@ class TypescriptCoordinator(SubprocessCoordinator): { "ts": { - "classpath": "airflow.sdk.coordinators.typescript.TypescriptCoordinator", + "classpath": "airflow.sdk.coordinators.node.NodeCoordinator", "kwargs": { "node_executable": "node", "bundles_root": ["/opt/airflow/ts-bundles"], diff --git a/task-sdk/tests/task_sdk/coordinators/typescript/__init__.py b/task-sdk/tests/task_sdk/coordinators/node/__init__.py similarity index 100% rename from task-sdk/tests/task_sdk/coordinators/typescript/__init__.py rename to task-sdk/tests/task_sdk/coordinators/node/__init__.py diff --git a/task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py similarity index 84% rename from task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py rename to task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py index ed51e4aef05b8..3538c389a19b2 100644 --- a/task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py +++ b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py @@ -24,8 +24,8 @@ from uuid6 import uuid7 from airflow.sdk.api.datamodels._generated import TaskInstance -from airflow.sdk.coordinators.typescript.coordinator import ( - TypescriptCoordinator, +from airflow.sdk.coordinators.node.coordinator import ( + NodeCoordinator, _find_bundle, ) @@ -69,16 +69,16 @@ def write_bundle(root: pathlib.Path, schema_version: str = SCHEMA_VERSION) -> pa return bundle -class TestTypescriptCoordinatorAttributes: +class TestNodeCoordinatorAttributes: def test_default_kwargs(self): - coordinator = TypescriptCoordinator(bundles_root="/airflow/ts-bundles") + coordinator = NodeCoordinator(bundles_root="/airflow/ts-bundles") assert coordinator.node_executable == "node" assert coordinator.bundles_root == [pathlib.Path("/airflow/ts-bundles")] assert coordinator.task_startup_timeout == 10.0 def test_custom_kwargs(self): - coordinator = TypescriptCoordinator( + coordinator = NodeCoordinator( node_executable="/opt/node/bin/node", bundles_root=["/airflow/ts-bundles", "~/extra-bundles"], task_startup_timeout=30.0, @@ -93,10 +93,10 @@ def test_custom_kwargs(self): def test_bundles_root_is_required(self): with pytest.raises(ValueError, match="Length of 'bundles_root' must be >= 1"): - TypescriptCoordinator(bundles_root=None) + NodeCoordinator(bundles_root=None) -class TestTypescriptCoordinatorBundleSelection: +class TestNodeCoordinatorBundleSelection: def test_find_bundle_returns_bundle_mjs(self, tmp_path): bundle = write_bundle(tmp_path) @@ -126,7 +126,7 @@ def test_find_bundle_ignores_other_mjs_names(self, tmp_path): def test_find_bundle_rejects_bundle_without_metadata(self, tmp_path): (tmp_path / "bundle.mjs").write_text("export {};\n", encoding="utf-8") - with pytest.raises(ValueError, match="missing airflow-metadata.yaml"): + with pytest.raises(FileNotFoundError, match="missing airflow-metadata.yaml"): _find_bundle([tmp_path]) @pytest.mark.parametrize( @@ -141,7 +141,7 @@ def test_find_bundle_rejects_invalid_metadata_shape(self, tmp_path, metadata, me (tmp_path / "bundle.mjs").write_text("export {};\n", encoding="utf-8") (tmp_path / "airflow-metadata.yaml").write_text(metadata, encoding="utf-8") - with pytest.raises(ValueError, match=message): + with pytest.raises(FileNotFoundError, match=message): _find_bundle([tmp_path]) def test_find_bundle_reports_unreadable_metadata(self, tmp_path, monkeypatch): @@ -150,30 +150,32 @@ def test_find_bundle_reports_unreadable_metadata(self, tmp_path, monkeypatch): def raise_os_error(self, *args, **kwargs): if self.name == "airflow-metadata.yaml": raise PermissionError("denied") - return original_read_text(self, *args, **kwargs) + return original_open(self, *args, **kwargs) - original_read_text = pathlib.Path.read_text - monkeypatch.setattr(pathlib.Path, "read_text", raise_os_error) + original_open = pathlib.Path.open + monkeypatch.setattr(pathlib.Path, "open", raise_os_error) - with pytest.raises(ValueError, match="cannot read airflow-metadata.yaml"): + with pytest.raises(FileNotFoundError, match="cannot read airflow-metadata.yaml"): _find_bundle([tmp_path]) def test_find_bundle_rejects_invalid_schema_version(self, tmp_path): write_bundle(tmp_path, schema_version="banana") - with pytest.raises(ValueError, match="Version 'banana' not found"): + with pytest.raises(FileNotFoundError, match="Version 'banana' not found"): _find_bundle([tmp_path]) - def test_find_bundle_fails_on_invalid_first_bundle(self, tmp_path): + def test_find_bundle_skips_rejected_bundle_metadata(self, tmp_path): first = tmp_path / "first" second = tmp_path / "second" first.mkdir() second.mkdir() (first / "bundle.mjs").write_text("export {};\n", encoding="utf-8") - write_bundle(second) + bundle = write_bundle(second) - with pytest.raises(ValueError, match="missing airflow-metadata.yaml"): - _find_bundle([first, second]) + found = _find_bundle([first, second]) + + assert found.path == bundle + assert found.schema_version == SCHEMA_VERSION def test_find_bundle_raises_with_searched_roots(self, tmp_path): first = tmp_path / "first" @@ -189,10 +191,10 @@ def test_find_bundle_raises_with_searched_roots(self, tmp_path): assert str(second.resolve()) in msg -class TestTypescriptCoordinatorExecuteTaskCommand: +class TestNodeCoordinatorExecuteTaskCommand: def test_build_execute_task_command_returns_node_bundle_and_schema_version(self, tmp_path): bundle = write_bundle(tmp_path) - coordinator = TypescriptCoordinator( + coordinator = NodeCoordinator( node_executable="/opt/node/bin/node", bundles_root=tmp_path, ) From b1062892ec13d780d60137c06d4675bf075d0dd8 Mon Sep 17 00:00:00 2001 From: Shivam Rastogi <6463385+shivaam@users.noreply.github.com> Date: Sun, 28 Jun 2026 12:26:12 -0700 Subject: [PATCH 4/5] Document Node coordinator usage for TypeScript SDK --- airflow-core/newsfragments/68548.feature.rst | 2 +- ts-sdk/README.md | 27 ++++++++++++++++---- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/airflow-core/newsfragments/68548.feature.rst b/airflow-core/newsfragments/68548.feature.rst index 107eff3cf7bcd..b27224b817d00 100644 --- a/airflow-core/newsfragments/68548.feature.rst +++ b/airflow-core/newsfragments/68548.feature.rst @@ -1 +1 @@ -Add a TypeScript task coordinator for running TypeScript task bundles through coordinator mode. +Add a Node.js task coordinator for running TypeScript task bundles through coordinator mode. diff --git a/ts-sdk/README.md b/ts-sdk/README.md index f69f9fe758889..f76ad3ba9597a 100644 --- a/ts-sdk/README.md +++ b/ts-sdk/README.md @@ -52,11 +52,12 @@ key by the active runtime, matching Python `@task` behavior. ## Intended Coordinator Usage -This PR only adds the TypeScript-side public interface. The coordinator runtime -will be added separately. Declaring Airflow Dags in TypeScript is not supported -yet; the Dag is still declared in Python. The intended authoring shape matches -the other non-Python SDKs: a Python Dag declares the scheduling shape with stub -tasks, and the TypeScript module registers handlers with matching task IDs. +Airflow runs TypeScript task bundles through the Python-side +`airflow.sdk.coordinators.node.NodeCoordinator`. Declaring Airflow Dags in +TypeScript is not supported yet; the Dag is still declared in Python. The +intended authoring shape matches the other non-Python SDKs: a Python Dag +declares the scheduling shape with stub tasks, and the TypeScript module +registers handlers with matching task IDs. Python Dag: @@ -78,6 +79,22 @@ def sales_pipeline(): sales_pipeline() ``` +Airflow coordinator config: + +```ini +[sdk] +coordinators = { + "ts": { + "classpath": "airflow.sdk.coordinators.node.NodeCoordinator", + "kwargs": {"bundles_root": ["/opt/airflow/ts-bundles"]} + } +} +queue_to_coordinator = {"typescript": "ts"} +``` + +Each configured bundle directory must contain `bundle.mjs` and +`airflow-metadata.yaml`. + TypeScript handlers: ```ts From e0343fd94e0c82c534e8824b62195b595f98a805 Mon Sep 17 00:00:00 2001 From: Shivam Rastogi <6463385+shivaam@users.noreply.github.com> Date: Sun, 28 Jun 2026 14:33:04 -0700 Subject: [PATCH 5/5] Improve Node coordinator metadata rejection logging --- task-sdk/src/airflow/sdk/coordinators/node/coordinator.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py index 458f19f039767..c1b23615b89a1 100644 --- a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py +++ b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py @@ -107,7 +107,10 @@ def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _NodeBundle: ) except (TypeError, ValueError) as exc: log.debug( - "TypeScript bundle metadata rejected; skipping", path=candidate, root=root, error=str(exc) + "TypeScript bundle metadata rejected; skipping", + path=candidate, + root=root, + exc_info=True, ) rejected.append((candidate.resolve(), str(exc)))