diff --git a/airflow-core/newsfragments/68548.feature.rst b/airflow-core/newsfragments/68548.feature.rst
new file mode 100644
index 0000000000000..b27224b817d00
--- /dev/null
+++ b/airflow-core/newsfragments/68548.feature.rst
@@ -0,0 +1 @@
+Add a Node.js task coordinator for running TypeScript task bundles through coordinator mode.
diff --git a/task-sdk/src/airflow/sdk/coordinators/node/__init__.py b/task-sdk/src/airflow/sdk/coordinators/node/__init__.py
new file mode 100644
index 0000000000000..9030ca17ebbfa
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/coordinators/node/__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.
+"""
+Node.js runtime coordinator for the Apache Airflow Task SDK.
+
+See :class:`NodeCoordinator` for details.
+"""
+
+from __future__ import annotations
+
+from airflow.sdk.coordinators.node.coordinator import NodeCoordinator
+
+__all__ = ["NodeCoordinator"]
diff --git a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
new file mode 100644
index 0000000000000..c1b23615b89a1
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
@@ -0,0 +1,174 @@
+#
+# 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.
+"""Node.js 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.node")
+
+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 _NodeBundle:
+    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:
+        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:
+        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]) -> _NodeBundle:
+    """
+    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.
+    """
+    rejected: list[tuple[pathlib.Path, str]] = []
+    for root in bundles_root:
+        candidate = root / BUNDLE_FILENAME
+        if not candidate.is_file():
+            continue
+        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,
+                exc_info=True,
+            )
+            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}")
+
+
+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 NodeCoordinator(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.node.NodeCoordinator",
+                "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/node/__init__.py b/task-sdk/tests/task_sdk/coordinators/node/__init__.py
new file mode 100644
index 0000000000000..217e5db960782
--- /dev/null
+++ b/task-sdk/tests/task_sdk/coordinators/node/__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/node/test_coordinator.py b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
new file mode 100644
index 0000000000000..3538c389a19b2
--- /dev/null
+++ b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
@@ -0,0 +1,205 @@
+#
+# 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.node.coordinator import (
+    NodeCoordinator,
+    _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 TestNodeCoordinatorAttributes:
+    def test_default_kwargs(self):
+        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 = NodeCoordinator(
+            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"):
+            NodeCoordinator(bundles_root=None)
+
+
+class TestNodeCoordinatorBundleSelection:
+    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(FileNotFoundError, 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(FileNotFoundError, 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_open(self, *args, **kwargs)
+
+        original_open = pathlib.Path.open
+        monkeypatch.setattr(pathlib.Path, "open", raise_os_error)
+
+        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(FileNotFoundError, match="Version 'banana' not found"):
+            _find_bundle([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")
+        bundle = write_bundle(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"
+        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 TestNodeCoordinatorExecuteTaskCommand:
+    def test_build_execute_task_command_returns_node_bundle_and_schema_version(self, tmp_path):
+        bundle = write_bundle(tmp_path)
+        coordinator = NodeCoordinator(
+            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
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