From 3bc1c15c789b8c0cb8d80ef186e673e2d86709b9 Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Fri, 13 Feb 2026 22:40:28 +0000
Subject: [PATCH 1/9] Remove the need for a GatewaySessionManager type.

The only differences between GatewaySessionManager and the underlying SessionManager type were:

1. Calling `get_kernel` inside the `kernel_cull` method to determine if the kernel exists.

   The underlying `SessionManager` class was instead checking if the specified kernel_id was
   *in* the MultiKernelManager.

   However, the MultiKernelManager class implements `get_kernel` by indexing into itself,
   so we could make the base SessionManager class also call `get_kernel` and it would be
   made compatible with `GatewayMappingKernelManager` while still remaining compatible
   with any other subclass of MultiKernelManager.

2. Specifying that the `kernel_manager` was of type `GatewayMappingKernelManager`.

   This was unnecessary because the kernel_manager_class was already being set to
   that inside of `serverapp.py`, and the `test_gateway.py` test has been updated
   to verify that the GatewayMappingKernelManager class is in fact being used.
---
 jupyter_server/gateway/managers.py            | 24 +++++++------------
 jupyter_server/serverapp.py                   |  2 --
 .../services/sessions/sessionmanager.py       |  9 ++++++-
 tests/test_gateway.py                         |  5 +++-
 4 files changed, 20 insertions(+), 20 deletions(-)

diff --git a/jupyter_server/gateway/managers.py b/jupyter_server/gateway/managers.py
index 270001f30..ba6608633 100644
--- a/jupyter_server/gateway/managers.py
+++ b/jupyter_server/gateway/managers.py
@@ -8,6 +8,7 @@
 import datetime
 import json
 import os
+import warnings
 from queue import Empty, Queue
 from threading import Thread
 from time import monotonic
@@ -359,22 +360,13 @@ class GatewaySessionManager(SessionManager):
 
     kernel_manager = Instance("jupyter_server.gateway.managers.GatewayMappingKernelManager")
 
-    async def kernel_culled(self, kernel_id: str) -> bool:  # typing: ignore
-        """Checks if the kernel is still considered alive and returns true if it's not found."""
-        km: Optional[GatewayKernelManager] = None
-        try:
-            # Since we keep the models up-to-date via client polling, use that state to determine
-            # if this kernel no longer exists on the gateway server rather than perform a redundant
-            # fetch operation - especially since this is called at approximately the same interval.
-            # This has the effect of reducing GET /api/kernels requests against the gateway server
-            # by 50%!
-            # Note that should the redundant polling be consolidated, or replaced with an event-based
-            # notification model, this will need to be revisited.
-            km = self.kernel_manager.get_kernel(kernel_id)
-        except Exception:
-            # Let exceptions here reflect culled kernel
-            pass
-        return km is None
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        warnings.warn(
+            "The GatewaySessionManager class is deprecated and will not be supported in Jupyter Server 3.0",
+            DeprecationWarning,
+            stacklevel=2,
+        )
 
 
 class GatewayKernelManager(ServerKernelManager):
diff --git a/jupyter_server/serverapp.py b/jupyter_server/serverapp.py
index ed2bd1361..4bcda1396 100644
--- a/jupyter_server/serverapp.py
+++ b/jupyter_server/serverapp.py
@@ -1632,8 +1632,6 @@ def _default_kernel_manager_class(self) -> t.Union[str, type[AsyncMappingKernelM
 
     @default("session_manager_class")
     def _default_session_manager_class(self) -> t.Union[str, type[SessionManager]]:
-        if self.gateway_config.gateway_enabled:
-            return "jupyter_server.gateway.managers.GatewaySessionManager"
         return SessionManager
 
     kernel_websocket_connection_class = Type(
diff --git a/jupyter_server/services/sessions/sessionmanager.py b/jupyter_server/services/sessions/sessionmanager.py
index f02e04bc4..b13a05ce3 100644
--- a/jupyter_server/services/sessions/sessionmanager.py
+++ b/jupyter_server/services/sessions/sessionmanager.py
@@ -18,6 +18,7 @@
 
 from dataclasses import dataclass, fields
 
+from jupyter_client.manager import KernelManager
 from jupyter_core.utils import ensure_async
 from tornado import web
 from traitlets import Instance, TraitError, Unicode, validate
@@ -468,7 +469,13 @@ async def update_session(self, session_id, **kwargs):
 
     async def kernel_culled(self, kernel_id: str) -> bool:
         """Checks if the kernel is still considered alive and returns true if its not found."""
-        return kernel_id not in self.kernel_manager
+        km: Optional[KernelManager] = None
+        try:
+            km = self.kernel_manager.get_kernel(kernel_id)
+        except Exception:
+            # Let exceptions here reflect culled kernel
+            pass
+        return km is None
 
     async def row_to_model(self, row, tolerate_culled=False):
         """Takes sqlite database session row and turns it into a dictionary"""
diff --git a/tests/test_gateway.py b/tests/test_gateway.py
index 6f4fcebc1..d50c2b23c 100644
--- a/tests/test_gateway.py
+++ b/tests/test_gateway.py
@@ -439,7 +439,10 @@ def test_gateway_request_with_expiring_cookies(
 async def test_gateway_class_mappings(init_gateway, jp_serverapp):
     # Ensure appropriate class mappings are in place.
     assert jp_serverapp.kernel_manager_class.__name__ == "GatewayMappingKernelManager"
-    assert jp_serverapp.session_manager_class.__name__ == "GatewaySessionManager"
+    assert (
+        jp_serverapp.session_manager.kernel_manager.__class__.__name__
+        == "GatewayMappingKernelManager"
+    )
     assert jp_serverapp.kernel_spec_manager_class.__name__ == "GatewayKernelSpecManager"
 
 

From b3b9cb66f06b9273530d5b4318df0e50ec30798d Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Fri, 13 Feb 2026 23:40:54 +0000
Subject: [PATCH 2/9] Allow each kernel manager to configure its own websocket
 connection class rather than requiring a single, global config

---
 jupyter_server/gateway/managers.py               |  2 ++
 jupyter_server/serverapp.py                      |  4 ----
 jupyter_server/services/kernels/kernelmanager.py | 11 +++++++++++
 jupyter_server/services/kernels/websocket.py     |  4 ++++
 4 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/jupyter_server/gateway/managers.py b/jupyter_server/gateway/managers.py
index ba6608633..74c1814e0 100644
--- a/jupyter_server/gateway/managers.py
+++ b/jupyter_server/gateway/managers.py
@@ -32,6 +32,7 @@
 )
 from ..services.sessions.sessionmanager import SessionManager
 from ..utils import url_path_join
+from .connections import GatewayWebSocketConnection
 from .gateway_client import GatewayClient, gateway_request
 
 if TYPE_CHECKING:
@@ -398,6 +399,7 @@ def has_kernel(self):
 
     client_class = DottedObjectName("jupyter_server.gateway.managers.GatewayKernelClient")
     client_factory = Type(klass="jupyter_server.gateway.managers.GatewayKernelClient")
+    websocket_connection_class = GatewayWebSocketConnection
 
     # --------------------------------------------------------------------------
     # create a Client connected to our Kernel
diff --git a/jupyter_server/serverapp.py b/jupyter_server/serverapp.py
index 4bcda1396..32903f558 100644
--- a/jupyter_server/serverapp.py
+++ b/jupyter_server/serverapp.py
@@ -103,7 +103,6 @@
 from jupyter_server.extension.config import ExtensionConfigManager
 from jupyter_server.extension.manager import ExtensionManager
 from jupyter_server.extension.serverextension import ServerExtensionApp
-from jupyter_server.gateway.connections import GatewayWebSocketConnection
 from jupyter_server.gateway.gateway_client import GatewayClient
 from jupyter_server.gateway.managers import (
     GatewayKernelSpecManager,
@@ -896,7 +895,6 @@ class ServerApp(JupyterApp):
         GatewayMappingKernelManager,
         GatewayKernelSpecManager,
         GatewaySessionManager,
-        GatewayWebSocketConnection,
         GatewayClient,
         Authorizer,
         EventLogger,
@@ -1644,8 +1642,6 @@ def _default_session_manager_class(self) -> t.Union[str, type[SessionManager]]:
     def _default_kernel_websocket_connection_class(
         self,
     ) -> t.Union[str, type[ZMQChannelsWebsocketConnection]]:
-        if self.gateway_config.gateway_enabled:
-            return "jupyter_server.gateway.connections.GatewayWebSocketConnection"
         return ZMQChannelsWebsocketConnection
 
     websocket_ping_interval = Integer(
diff --git a/jupyter_server/services/kernels/kernelmanager.py b/jupyter_server/services/kernels/kernelmanager.py
index 8eb0b947a..e18720023 100644
--- a/jupyter_server/services/kernels/kernelmanager.py
+++ b/jupyter_server/services/kernels/kernelmanager.py
@@ -42,6 +42,7 @@
     Integer,
     List,
     TraitError,
+    Type,
     Unicode,
     default,
     validate,
@@ -50,6 +51,8 @@
 from jupyter_server import DEFAULT_EVENTS_SCHEMA_PATH
 from jupyter_server._tz import isoformat, utcnow
 from jupyter_server.prometheus.metrics import KERNEL_CURRENTLY_RUNNING_TOTAL
+from jupyter_server.services.kernels.connection.base import BaseKernelWebsocketConnection
+from jupyter_server.services.kernels.connection.channels import ZMQChannelsWebsocketConnection
 from jupyter_server.utils import ApiPath, import_item, to_os_path
 
 
@@ -895,6 +898,14 @@ def _default_event_logger(self):
                 pass
         return logger
 
+    websocket_connection_class = Type(
+        default_value=ZMQChannelsWebsocketConnection,
+        klass=BaseKernelWebsocketConnection,
+        help="""
+        The websocket connection class to use for this manager's kernels.
+        """,
+    ).tag(config=True)
+
     def emit(self, schema_id, data):
         """Emit an event from the kernel manager."""
         self.event_logger.emit(schema_id=schema_id, data=data)
diff --git a/jupyter_server/services/kernels/websocket.py b/jupyter_server/services/kernels/websocket.py
index a24b0539f..2d0ab4974 100644
--- a/jupyter_server/services/kernels/websocket.py
+++ b/jupyter_server/services/kernels/websocket.py
@@ -21,6 +21,10 @@ class KernelWebsocketHandler(WebSocketMixin, WebSocketHandler, JupyterHandler):
     @property
     def kernel_websocket_connection_class(self):
         """The kernel websocket connection class."""
+        if self.kernel_manager and self.kernel_id:
+            kernel = self.kernel_manager.get_kernel(self.kernel_id)
+            if hasattr(kernel, "websocket_connection_class"):
+                return kernel.websocket_connection_class
         return self.settings.get("kernel_websocket_connection_class")
 
     def set_default_headers(self):

From fcc9f5aab668d16f905298f384adc4399af8493a Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Fri, 13 Feb 2026 23:59:31 +0000
Subject: [PATCH 3/9] Allow kernel managers to print their own info messages at
 server startup

---
 jupyter_server/gateway/managers.py | 8 ++++++++
 jupyter_server/serverapp.py        | 7 ++-----
 2 files changed, 10 insertions(+), 5 deletions(-)

diff --git a/jupyter_server/gateway/managers.py b/jupyter_server/gateway/managers.py
index 74c1814e0..a5ab4012b 100644
--- a/jupyter_server/gateway/managers.py
+++ b/jupyter_server/gateway/managers.py
@@ -31,6 +31,7 @@
     emit_kernel_action_event,
 )
 from ..services.sessions.sessionmanager import SessionManager
+from ..transutils import _i18n
 from ..utils import url_path_join
 from .connections import GatewayWebSocketConnection
 from .gateway_client import GatewayClient, gateway_request
@@ -213,6 +214,13 @@ async def cull_kernels(self):
         await self.list_kernels()
         await super().cull_kernels()
 
+    @property
+    def info(self):
+        return (
+            _i18n("\nKernels will be managed by the Gateway server running at:\n%s")
+            % self.kernels_url
+        )
+
 
 class GatewayKernelSpecManager(KernelSpecManager):
     """A gateway kernel spec manager."""
diff --git a/jupyter_server/serverapp.py b/jupyter_server/serverapp.py
index 32903f558..617f06e62 100644
--- a/jupyter_server/serverapp.py
+++ b/jupyter_server/serverapp.py
@@ -2871,11 +2871,8 @@ def running_server_info(self, kernel_count: bool = True) -> str:
         info += _i18n("Jupyter Server {version} is running at:\n{url}").format(
             version=ServerApp.version, url=self.display_url
         )
-        if self.gateway_config.gateway_enabled:
-            info += (
-                _i18n("\nKernels will be managed by the Gateway server running at:\n%s")
-                % self.gateway_config.url
-            )
+        if hasattr(self.kernel_manager, "info"):
+            info += self.kernel_manager.info
         return info
 
     def server_info(self) -> dict[str, t.Any]:

From 28046d197482fff9a79aba6ec8b959dc6b3c0475 Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Sat, 14 Feb 2026 05:27:38 +0000
Subject: [PATCH 4/9] Support configuring multiple MappingKernelManagers to be
 used side-by-side

---
 jupyter_server/serverapp.py                |  22 +-
 jupyter_server/services/kernels/routing.py | 356 +++++++++++++++++++++
 2 files changed, 364 insertions(+), 14 deletions(-)
 create mode 100644 jupyter_server/services/kernels/routing.py

diff --git a/jupyter_server/serverapp.py b/jupyter_server/serverapp.py
index 617f06e62..a8b50dc44 100644
--- a/jupyter_server/serverapp.py
+++ b/jupyter_server/serverapp.py
@@ -104,11 +104,6 @@
 from jupyter_server.extension.manager import ExtensionManager
 from jupyter_server.extension.serverextension import ServerExtensionApp
 from jupyter_server.gateway.gateway_client import GatewayClient
-from jupyter_server.gateway.managers import (
-    GatewayKernelSpecManager,
-    GatewayMappingKernelManager,
-    GatewaySessionManager,
-)
 from jupyter_server.log import log_request
 from jupyter_server.prometheus.metrics import (
     ACTIVE_DURATION,
@@ -130,6 +125,10 @@
     AsyncMappingKernelManager,
     MappingKernelManager,
 )
+from jupyter_server.services.kernels.routing import (
+    RoutingKernelSpecManager,
+    RoutingMappingKernelManager,
+)
 from jupyter_server.services.sessions.sessionmanager import SessionManager
 from jupyter_server.utils import (
     JupyterServerAuthWarning,
@@ -892,13 +891,12 @@ class ServerApp(JupyterApp):
         AsyncContentsManager,
         AsyncFileContentsManager,
         NotebookNotary,
-        GatewayMappingKernelManager,
-        GatewayKernelSpecManager,
-        GatewaySessionManager,
         GatewayClient,
         Authorizer,
         EventLogger,
         ZMQChannelsWebsocketConnection,
+        RoutingKernelSpecManager,
+        RoutingMappingKernelManager,
     ]
 
     subcommands: dict[str, t.Any] = {
@@ -1619,9 +1617,7 @@ def template_file_path(self) -> list[str]:
 
     @default("kernel_manager_class")
     def _default_kernel_manager_class(self) -> t.Union[str, type[AsyncMappingKernelManager]]:
-        if self.gateway_config.gateway_enabled:
-            return "jupyter_server.gateway.managers.GatewayMappingKernelManager"
-        return AsyncMappingKernelManager
+        return RoutingMappingKernelManager
 
     session_manager_class = Type(
         config=True,
@@ -1691,9 +1687,7 @@ def _default_kernel_websocket_connection_class(
 
     @default("kernel_spec_manager_class")
     def _default_kernel_spec_manager_class(self) -> t.Union[str, type[KernelSpecManager]]:
-        if self.gateway_config.gateway_enabled:
-            return "jupyter_server.gateway.managers.GatewayKernelSpecManager"
-        return KernelSpecManager
+        return RoutingKernelSpecManager
 
     login_handler_class = Type(
         default_value=LoginHandler,
diff --git a/jupyter_server/services/kernels/routing.py b/jupyter_server/services/kernels/routing.py
new file mode 100644
index 000000000..46e884e75
--- /dev/null
+++ b/jupyter_server/services/kernels/routing.py
@@ -0,0 +1,356 @@
+import copy
+import typing as t
+
+from jupyter_client.kernelspec import KernelSpecManager
+from jupyter_client.manager import in_pending_state
+from jupyter_client.managerabc import KernelManagerABC
+from jupyter_core.utils import ensure_async, run_sync
+from traitlets import (
+    Dict,
+    Instance,
+    List,
+    Type,
+    Unicode,
+    default,
+    observe,
+)
+from traitlets.config import LoggingConfigurable
+
+from jupyter_server.gateway.gateway_client import GatewayClient
+from jupyter_server.gateway.managers import GatewayKernelSpecManager, GatewayMappingKernelManager
+from jupyter_server.services.kernels.connection.base import BaseKernelWebsocketConnection
+from jupyter_server.services.kernels.connection.channels import ZMQChannelsWebsocketConnection
+from jupyter_server.services.kernels.kernelmanager import (
+    AsyncMappingKernelManager,
+    ServerKernelManager,
+)
+from jupyter_server.transutils import _i18n
+
+
+class RoutingProvider(LoggingConfigurable):
+    connection_dir = Unicode("")
+
+    primary_manager = Instance(AsyncMappingKernelManager)
+
+    additional_managers = List(trait=Instance(AsyncMappingKernelManager))
+
+    @default("primary_manager")
+    def _default_primary_manager(self):
+        ksm = KernelSpecManager(parent=self.parent)
+        return AsyncMappingKernelManager(
+            parent=self.parent,
+            log=self.log,
+            connection_dir=self.connection_dir,
+            kernel_spec_manager=ksm,
+        )
+
+    info = Unicode("")
+
+
+class RemoteOnlyRoutingProvider(RoutingProvider):
+    @default("primary_manager")
+    def _default_primary_manager(self):
+        ksm = GatewayKernelSpecManager(parent=self.parent)
+        return GatewayMappingKernelManager(
+            parent=self.parent,
+            log=self.log,
+            connection_dir=self.connection_dir,
+            kernel_spec_manager=ksm,
+        )
+
+    @default("info")
+    def _default_info(self):
+        return (
+            _i18n("\nKernels will be managed by the Gateway server running at:\n%s")
+            % self.gateway_config.url
+        )
+
+
+class SideBySideRoutingProvider(RoutingProvider):
+    @default("additional_managers")
+    def _default_additional_managers(self):
+        ksm = GatewayKernelSpecManager(parent=self.parent)
+        return [
+            GatewayMappingKernelManager(
+                parent=self.parent,
+                log=self.log,
+                connection_dir=self.connection_dir,
+                kernel_spec_manager=ksm,
+            )
+        ]
+
+
+class AsyncRoutingKernelSpecManager(KernelSpecManager):
+    """KernelSpecManager that routes to multiple nested kernel spec managers.
+
+    This async version of the wrapper exists because the base KernelSpecManager
+    class only has synchronous methods, but some child classes (in particular,
+    GatewayKernelManager) change those methods to be async.
+
+    In order to support both versions, we first implement the routing in this async
+    class, but then make it synchronous in the child, RoutingKernelSpecManager class.
+    """
+
+    @property
+    def primary_manager(self):
+        if hasattr(self.parent.kernel_manager, "routing_provider"):
+            return self.parent.kernel_manager.routing_provider.primary_manager
+
+        # This happens if the server administrator configured just the kernel_manager_class
+        # while leaving the kernel_spec_manager_class as the default.
+        # In that case we want this class to just pass through to the kernel manager's
+        # nested kernelspec manager.
+        km = self.parent.kernel_manager
+
+        # On the odd chance that an administrator explicitly configured a non
+        # RoutingKernelManager with an instance of AsyncRoutingKernelSpecManager as
+        # its nested `kernel_spec_manager`, all attempts to list or get kernelspecs
+        # will result in an infinite loop.
+        #
+        # Accordingly, we use an assert to catch this early.
+        assert not isinstance(km, AsyncRoutingKernelSpecManager)
+
+        return km
+
+    @property
+    def additional_managers(self):
+        if hasattr(self.parent.kernel_manager, "routing_provider"):
+            return self.parent.kernel_manager.routing_provider.additional_managers
+        return []
+
+    spec_to_manager_map = Dict(key_trait=Unicode(), value_trait=Instance(AsyncMappingKernelManager))
+
+    async def get_all_specs(self):
+        ks = await ensure_async(self.primary_manager.kernel_spec_manager.get_all_specs())
+        for spec_name, _spec in ks.items():
+            self.spec_to_manager_map[spec_name] = self.primary_manager
+        for additional_manager in self.additional_managers:
+            additional_ks = await ensure_async(
+                additional_manager.kernel_spec_manager.get_all_specs()
+            )
+            for spec_name, spec in additional_ks.items():
+                if spec_name not in ks:
+                    ks[spec_name] = spec
+                    self.spec_to_manager_map[spec_name] = additional_manager
+        return ks
+
+    def get_mapping_kernel_manager(self, kernel_name: str) -> AsyncMappingKernelManager:
+        km = self.spec_to_manager_map.get(kernel_name, None)
+        if km is None:
+            return self.primary_manager
+        return km
+
+    async def get_kernel_spec(self, kernel_name, **kwargs):
+        wrapped_manager = self.get_mapping_kernel_manager(kernel_name).kernel_spec_manager
+        return ensure_async(wrapped_manager.get_kernel_spec(kernel_name, **kwargs))
+
+    async def get_kernel_spec_resource(self, kernel_name, path):
+        wrapped_manager = self.get_mapping_kernel_manager(kernel_name).kernel_spec_manager
+        if hasattr(wrapped_manager, "get_kernel_spec_resource"):
+            return await ensure_async(wrapped_manager.get_kernel_spec_resource(kernel_name, path))
+        return None
+
+    def is_remote(self, kernel_name):
+        wrapped_manager = self.get_mapping_kernel_manager(kernel_name).kernel_spec_manager
+        return isinstance(wrapped_manager, GatewayKernelSpecManager)
+
+
+class RoutingKernelSpecManager(AsyncRoutingKernelSpecManager):
+    """KernelSpecManager that routes to multiple nested kernel spec managers."""
+
+    def get_all_specs(self):
+        return run_sync(super().get_all_specs)()
+
+    def get_kernel_spec(self, kernel_name, *args, **kwargs):
+        return run_sync(super().get_kernel_spec)(kernel_name, *args, **kwargs)
+
+
+class RoutingKernelManagerWebsocketConnection(BaseKernelWebsocketConnection):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        km = self.kernel_manager.wrapped_kernel_manager
+        wrapped_class = ZMQChannelsWebsocketConnection
+        if hasattr(km, "websocket_connection_class"):
+            wrapped_class = km.websocket_connection_class
+        self.wrapped = wrapped_class(
+            parent=km, websocket_handler=self.websocket_handler, config=self.config
+        )
+
+    async def connect(self):
+        """Connect the kernel websocket to the kernel ZMQ connections"""
+        return await self.wrapped.connect()
+
+    # N.B. The disconnect method in the BaseKernelWebsocketConnection is defined
+    # to be async, but in all of the implementing subclasses it is sync, and
+    # the Jupyter server does not await the value returned from this method.
+    def disconnect(self):
+        """Disconnect the kernel websocket from the kernel ZMQ connections"""
+        return self.wrapped.disconnect()
+
+    def handle_incoming_message(self, incoming_msg: str) -> None:
+        """Broker the incoming websocket message to the appropriate ZMQ channel."""
+        self.wrapped.handle_incoming_message(incoming_msg)
+
+    def handle_outgoing_message(self, stream: str, outgoing_msg: list) -> None:
+        """Broker outgoing ZMQ messages to the kernel websocket."""
+        self.wrapped.handle_outgoing_message(stream, outgoing_msg)
+
+    async def prepare(self):
+        if hasattr(self.wrapped, "prepare"):
+            return await self.wrapped.prepare()
+
+
+class RoutingKernelManager(ServerKernelManager):
+    kernel_id_map: dict[str, str] = {}
+
+    @property
+    def is_remote(self):
+        if not self.kernel_name or not self.kernel_id:
+            return False
+        return self.parent.kernel_spec_manager.is_remote(self.kernel_name)
+
+    @property
+    def wrapped_multi_kernel_manager(self):
+        return self.parent.kernel_spec_manager.get_mapping_kernel_manager(self.kernel_name)
+
+    @property
+    def wrapped_kernel_manager(self):
+        if not self.kernel_id:
+            return None
+        wrapped_kernel_id = RoutingKernelManager.kernel_id_map.get(self.kernel_id, self.kernel_id)
+        return self.wrapped_multi_kernel_manager.get_kernel(wrapped_kernel_id)
+
+    @property
+    def websocket_connection_class(self):
+        return RoutingKernelManagerWebsocketConnection
+
+    @property
+    def has_kernel(self):
+        if not self.kernel_id:
+            return False
+        return self.wrapped_kernel_manager.has_kernel
+
+    def client(self, *args, **kwargs):
+        if not self.kernel_id:
+            return None
+        return self.wrapped_kernel_manager.client(*args, **kwargs)
+
+    @in_pending_state
+    async def start_kernel(self, *args, **kwargs):
+        kernel_id: t.Optional[str] = kwargs.pop("kernel_id", self.kernel_id)
+        if kernel_id:
+            self.kernel_id = kernel_id
+
+        km = self.wrapped_multi_kernel_manager
+        wrapped_kernel_id: str = await ensure_async(
+            km.start_kernel(kernel_name=self.kernel_name, **kwargs)
+        )
+        self.kernel_id = self.kernel_id or wrapped_kernel_id
+        RoutingKernelManager.kernel_id_map[self.kernel_id] = wrapped_kernel_id
+        self.log.debug(
+            f"Created kernel {self.kernel_id} corresponding to {wrapped_kernel_id} in {km}"
+        )
+        self.log.debug(RoutingKernelManager.kernel_id_map)
+
+    async def shutdown_kernel(self, now=False, restart=False):
+        wrapped_kernel_id = RoutingKernelManager.kernel_id_map.get(self.kernel_id, self.kernel_id)
+        km = self.wrapped_multi_kernel_manager
+        await ensure_async(km.shutdown_kernel(wrapped_kernel_id, now=now, restart=restart))
+        RoutingKernelManager.kernel_id_map.pop(self.kernel_id, None)
+
+    async def restart_kernel(self, now=False):
+        wrapped_kernel_id = RoutingKernelManager.kernel_id_map.get(self.kernel_id, self.kernel_id)
+        km = self.wrapped_multi_kernel_manager
+        return await ensure_async(km.restart_kernel(wrapped_kernel_id, now=now))
+
+    async def interrupt_kernel(self):
+        km = self.wrapped_kernel_manager
+        return await ensure_async(km.interrupt_kernel())
+
+    async def model(self):
+        wrapped_kernel_id = RoutingKernelManager.kernel_id_map.get(self.kernel_id, self.kernel_id)
+        wrapped_model = await ensure_async(
+            self.wrapped_multi_kernel_manager.kernel_model(wrapped_kernel_id)
+        )
+        model = copy.deepcopy(wrapped_model)
+        model["id"] = self.kernel_id
+        return model
+
+
+class RoutingMappingKernelManager(AsyncMappingKernelManager):
+    @default("kernel_manager_class")
+    def _default_kernel_manager_class(self):
+        return "jupyter_server.services.kernels.routing.RoutingKernelManager"
+
+    kernel_spec_manager = Instance(
+        "jupyter_server.services.kernels.routing.RoutingKernelSpecManager"
+    )
+
+    _routing_provider = None
+    routing_provider_class = Type(
+        klass=RoutingProvider,
+        config=True,
+        help=_i18n(
+            "The class defining how kernelspec and kernel requests are routed "
+            + "to the various supported managers."
+        ),
+    )
+
+    @default("routing_provider_class")
+    def _default_routing_provider_class(self):
+        gateway_config = GatewayClient.instance(parent=self.parent)
+        if gateway_config.gateway_enabled:
+            return RemoteOnlyRoutingProvider
+        return RoutingProvider
+
+    @property
+    def routing_provider(self):
+        if not self._routing_provider:
+            self._routing_provider = self.routing_provider_class(
+                parent=self.parent, log=self.log, connection_dir=self.connection_dir
+            )
+        return self._routing_provider
+
+    def has_remote_kernels(self):
+        for kid in self._kernels:
+            if self._kernels[kid].is_remote:
+                return True
+        return False
+
+    async def list_kernels(self):
+        if self.has_remote_kernels():
+            # We have remote kernels, so we must call `list_kernels` on the
+            # wrapped Gateway kernel managers to update our kernel models.
+            try:
+                await ensure_async(self.routing_provider.primary_manager.list_kernels())
+                for wrapped in self.routing_provider.additional_managers:
+                    await ensure_async(wrapped.list_kernels())
+            except Exception as ex:
+                self.log.exception("Failure listing kernels: %s", ex)
+                # Ignore the exception listing remote kernels, so that local kernels are still usable.
+        return super().list_kernels()
+
+    def kernel_model(self, kernel_id):
+        self._check_kernel_id(kernel_id)
+        kernel = self._kernels[kernel_id]
+        # Normally, calls to `run_sync` pose a danger of locking up Tornado's
+        # single-threaded event loop.
+        #
+        # However, the call below should be fine because it cannot block for an
+        # arbitrary amount of time.
+        #
+        # This call blocks on the `model` method defined below, which in turn
+        # blocks on the `GatewayMappingKernelManager`'s `kernel_model` method
+        # (https://github.com/jupyter-server/jupyter_server/blob/547f7a244d89f79dd09fa7d382322d1c40890a3f/jupyter_server/gateway/managers.py#L94).
+        #
+        # That will only take a small, deterministic amount of time to complete
+        # because that `kernel_model` only operates on existing, in-memory data
+        # and does not block on any outgoing network requests.
+        return run_sync(kernel.model)()
+
+    @property
+    def info(self):
+        return self.routing_provider.info
+
+
+KernelManagerABC.register(RoutingKernelManager)

From b9988d5a488424788641489c0305ace6d4ace778 Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Mon, 23 Feb 2026 22:24:12 +0000
Subject: [PATCH 5/9] Add checks to make sure that routing managers are
 configured correctly and pass calls to `is_alive` down to nested kernel
 managers

---
 jupyter_server/serverapp.py                | 12 +++++-
 jupyter_server/services/kernels/routing.py | 46 ++++++++++++++--------
 2 files changed, 40 insertions(+), 18 deletions(-)

diff --git a/jupyter_server/serverapp.py b/jupyter_server/serverapp.py
index a8b50dc44..8d91e0c34 100644
--- a/jupyter_server/serverapp.py
+++ b/jupyter_server/serverapp.py
@@ -1638,6 +1638,11 @@ def _default_session_manager_class(self) -> t.Union[str, type[SessionManager]]:
     def _default_kernel_websocket_connection_class(
         self,
     ) -> t.Union[str, type[ZMQChannelsWebsocketConnection]]:
+        if issubclass(
+            self.kernel_manager_class,
+            RoutingMappingKernelManager,
+        ):
+            return "jupyter_server.services.kernels.routing.RoutingKernelManagerWebsocketConnection"
         return ZMQChannelsWebsocketConnection
 
     websocket_ping_interval = Integer(
@@ -1687,7 +1692,12 @@ def _default_kernel_websocket_connection_class(
 
     @default("kernel_spec_manager_class")
     def _default_kernel_spec_manager_class(self) -> t.Union[str, type[KernelSpecManager]]:
-        return RoutingKernelSpecManager
+        if issubclass(
+            self.kernel_manager_class,
+            RoutingMappingKernelManager,
+        ):
+            return RoutingKernelSpecManager
+        return KernelSpecManager
 
     login_handler_class = Type(
         default_value=LoginHandler,
diff --git a/jupyter_server/services/kernels/routing.py b/jupyter_server/services/kernels/routing.py
index 46e884e75..dbd311608 100644
--- a/jupyter_server/services/kernels/routing.py
+++ b/jupyter_server/services/kernels/routing.py
@@ -93,30 +93,37 @@ class only has synchronous methods, but some child classes (in particular,
 
     @property
     def primary_manager(self):
-        if hasattr(self.parent.kernel_manager, "routing_provider"):
-            return self.parent.kernel_manager.routing_provider.primary_manager
-
-        # This happens if the server administrator configured just the kernel_manager_class
-        # while leaving the kernel_spec_manager_class as the default.
-        # In that case we want this class to just pass through to the kernel manager's
-        # nested kernelspec manager.
-        km = self.parent.kernel_manager
-
-        # On the odd chance that an administrator explicitly configured a non
-        # RoutingKernelManager with an instance of AsyncRoutingKernelSpecManager as
-        # its nested `kernel_spec_manager`, all attempts to list or get kernelspecs
-        # will result in an infinite loop.
+        # This kernelspec manager can only be used when the corresponding kernel
+        # manager can tell us how to route requests to the nested managers.
+        assert hasattr(self.parent.kernel_manager, "routing_provider")
+
+        km = self.parent.kernel_manager.routing_provider.primary_manager
+
+        # On the odd chance that an administrator explicitly configured a routing
+        # provider with a nested routing kernelspec manager, all attempts to list
+        # or get kernelspecs will result in an infinite loop.
         #
         # Accordingly, we use an assert to catch this early.
-        assert not isinstance(km, AsyncRoutingKernelSpecManager)
+        assert not isinstance(km.kernel_spec_manager, AsyncRoutingKernelSpecManager)
 
         return km
 
     @property
     def additional_managers(self):
-        if hasattr(self.parent.kernel_manager, "routing_provider"):
-            return self.parent.kernel_manager.routing_provider.additional_managers
-        return []
+        # This kernelspec manager can only be used when the corresponding kernel
+        # manager can tell us how to route requests to the nested managers.
+        assert hasattr(self.parent.kernel_manager, "routing_provider")
+
+        kms = self.parent.kernel_manager.routing_provider.additional_managers
+
+        # Similarly to the `primary_manager` property, we want to ensure that
+        # none of the nested kernelspec managers are instances of this same class,
+        # in order to prevent infinite loops and to catch the configuration
+        # issues that could cause such loops early.
+        for km in kms:
+            assert not isinstance(km.kernel_spec_manager, AsyncRoutingKernelSpecManager)
+
+        return kms
 
     spec_to_manager_map = Dict(key_trait=Unicode(), value_trait=Instance(AsyncMappingKernelManager))
 
@@ -230,6 +237,11 @@ def has_kernel(self):
             return False
         return self.wrapped_kernel_manager.has_kernel
 
+    async def is_alive(self):
+        if not self.has_kernel:
+            return False
+        return await self.wrapped_kernel_manager.is_alive()
+
     def client(self, *args, **kwargs):
         if not self.kernel_id:
             return None

From b7241b634ea2ede66fd4b185807c336465828bfe Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Mon, 23 Feb 2026 22:42:05 +0000
Subject: [PATCH 6/9] Make the test_jupyter_handler_contextvar test agnostic of
 which kernel manager class is used

---
 tests/base/test_call_context.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/base/test_call_context.py b/tests/base/test_call_context.py
index f3e48522f..ed683f869 100644
--- a/tests/base/test_call_context.py
+++ b/tests/base/test_call_context.py
@@ -7,7 +7,7 @@
 from jupyter_server.services.kernels.kernelmanager import AsyncMappingKernelManager
 
 
-async def test_jupyter_handler_contextvar(jp_fetch, monkeypatch):
+async def test_jupyter_handler_contextvar(jp_serverapp, jp_fetch, monkeypatch):
     # Create some mock kernel Ids
     kernel1 = "x-x-x-x-x"
     kernel2 = "y-y-y-y-y"
@@ -45,7 +45,7 @@ async def kernel_model(self, kernel_id):
         context_tracker[kernel_id]["ended"] = current.current_user
         return {"id": kernel_id, "name": "blah"}
 
-    monkeypatch.setattr(AsyncMappingKernelManager, "kernel_model", kernel_model)
+    monkeypatch.setattr(jp_serverapp.kernel_manager.__class__, "kernel_model", kernel_model)
 
     # Make two requests in parallel.
     await asyncio.gather(

From 7b02ed8fdb2e68f58478b322e471c8539d056963 Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Mon, 23 Feb 2026 22:47:01 +0000
Subject: [PATCH 7/9] Update API docs

---
 docs/source/api/jupyter_server.services.kernels.rst | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/docs/source/api/jupyter_server.services.kernels.rst b/docs/source/api/jupyter_server.services.kernels.rst
index 683e340a7..9fa4d1c4e 100644
--- a/docs/source/api/jupyter_server.services.kernels.rst
+++ b/docs/source/api/jupyter_server.services.kernels.rst
@@ -25,6 +25,12 @@ Submodules
    :undoc-members:
 
 
+.. automodule:: jupyter_server.services.kernels.routing
+   :members:
+   :show-inheritance:
+   :undoc-members:
+
+
 .. automodule:: jupyter_server.services.kernels.websocket
    :members:
    :show-inheritance:

From 43bd1b156af7b38d6fbd4d62ea7301d644c8e087 Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Mon, 23 Feb 2026 23:01:55 +0000
Subject: [PATCH 8/9] Fix lint warnings on the routing package

---
 jupyter_server/services/kernels/routing.py | 23 +++++++++++-----------
 1 file changed, 12 insertions(+), 11 deletions(-)

diff --git a/jupyter_server/services/kernels/routing.py b/jupyter_server/services/kernels/routing.py
index dbd311608..8fff4798f 100644
--- a/jupyter_server/services/kernels/routing.py
+++ b/jupyter_server/services/kernels/routing.py
@@ -46,6 +46,12 @@ def _default_primary_manager(self):
 
     info = Unicode("")
 
+    @default("info")
+    def _default_info(self):
+        if hasattr(self.primary_manager, "info"):
+            return self.primary_manager.info
+        return ""
+
 
 class RemoteOnlyRoutingProvider(RoutingProvider):
     @default("primary_manager")
@@ -58,13 +64,6 @@ def _default_primary_manager(self):
             kernel_spec_manager=ksm,
         )
 
-    @default("info")
-    def _default_info(self):
-        return (
-            _i18n("\nKernels will be managed by the Gateway server running at:\n%s")
-            % self.gateway_config.url
-        )
-
 
 class SideBySideRoutingProvider(RoutingProvider):
     @default("additional_managers")
@@ -92,9 +91,10 @@ class only has synchronous methods, but some child classes (in particular,
     """
 
     @property
-    def primary_manager(self):
+    def primary_manager(self) -> AsyncMappingKernelManager:
         # This kernelspec manager can only be used when the corresponding kernel
         # manager can tell us how to route requests to the nested managers.
+        assert self.parent is not None
         assert hasattr(self.parent.kernel_manager, "routing_provider")
 
         km = self.parent.kernel_manager.routing_provider.primary_manager
@@ -112,6 +112,7 @@ def primary_manager(self):
     def additional_managers(self):
         # This kernelspec manager can only be used when the corresponding kernel
         # manager can tell us how to route requests to the nested managers.
+        assert self.parent is not None
         assert hasattr(self.parent.kernel_manager, "routing_provider")
 
         kms = self.parent.kernel_manager.routing_provider.additional_managers
@@ -198,7 +199,7 @@ def handle_incoming_message(self, incoming_msg: str) -> None:
         """Broker the incoming websocket message to the appropriate ZMQ channel."""
         self.wrapped.handle_incoming_message(incoming_msg)
 
-    def handle_outgoing_message(self, stream: str, outgoing_msg: list) -> None:
+    def handle_outgoing_message(self, stream: str, outgoing_msg: list[t.Any]) -> None:
         """Broker outgoing ZMQ messages to the kernel websocket."""
         self.wrapped.handle_outgoing_message(stream, outgoing_msg)
 
@@ -227,8 +228,8 @@ def wrapped_kernel_manager(self):
         wrapped_kernel_id = RoutingKernelManager.kernel_id_map.get(self.kernel_id, self.kernel_id)
         return self.wrapped_multi_kernel_manager.get_kernel(wrapped_kernel_id)
 
-    @property
-    def websocket_connection_class(self):
+    @default("websocket_connection_class")
+    def _default_websocket_connection_class(self):
         return RoutingKernelManagerWebsocketConnection
 
     @property

From a58ea6572a8e8b8b72d1503acfaa14664d4d4051 Mon Sep 17 00:00:00 2001
From: Omar Jarjur <ojarjur@google.com>
Date: Mon, 23 Feb 2026 23:46:41 +0000
Subject: [PATCH 9/9] Document internal assumptions about what values the
 various routing managers are instantiated with

---
 jupyter_server/services/kernels/routing.py | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/jupyter_server/services/kernels/routing.py b/jupyter_server/services/kernels/routing.py
index 8fff4798f..1ed229e8e 100644
--- a/jupyter_server/services/kernels/routing.py
+++ b/jupyter_server/services/kernels/routing.py
@@ -96,6 +96,7 @@ def primary_manager(self) -> AsyncMappingKernelManager:
         # manager can tell us how to route requests to the nested managers.
         assert self.parent is not None
         assert hasattr(self.parent.kernel_manager, "routing_provider")
+        assert isinstance(self.parent.kernel_manager.routing_provider, RoutingProvider)
 
         km = self.parent.kernel_manager.routing_provider.primary_manager
 
@@ -129,7 +130,9 @@ def additional_managers(self):
     spec_to_manager_map = Dict(key_trait=Unicode(), value_trait=Instance(AsyncMappingKernelManager))
 
     async def get_all_specs(self):
-        ks = await ensure_async(self.primary_manager.kernel_spec_manager.get_all_specs())
+        ksm = self.primary_manager.kernel_spec_manager
+        assert ksm is not None
+        ks = await ensure_async(ksm.get_all_specs())
         for spec_name, _spec in ks.items():
             self.spec_to_manager_map[spec_name] = self.primary_manager
         for additional_manager in self.additional_managers:
@@ -143,23 +146,27 @@ async def get_all_specs(self):
         return ks
 
     def get_mapping_kernel_manager(self, kernel_name: str) -> AsyncMappingKernelManager:
-        km = self.spec_to_manager_map.get(kernel_name, None)
-        if km is None:
-            return self.primary_manager
+        km = self.spec_to_manager_map.get(kernel_name, None) or self.primary_manager
         return km
 
     async def get_kernel_spec(self, kernel_name, **kwargs):
         wrapped_manager = self.get_mapping_kernel_manager(kernel_name).kernel_spec_manager
+        assert wrapped_manager is not None
+
         return ensure_async(wrapped_manager.get_kernel_spec(kernel_name, **kwargs))
 
     async def get_kernel_spec_resource(self, kernel_name, path):
         wrapped_manager = self.get_mapping_kernel_manager(kernel_name).kernel_spec_manager
+        assert wrapped_manager is not None
+
         if hasattr(wrapped_manager, "get_kernel_spec_resource"):
             return await ensure_async(wrapped_manager.get_kernel_spec_resource(kernel_name, path))
         return None
 
     def is_remote(self, kernel_name):
         wrapped_manager = self.get_mapping_kernel_manager(kernel_name).kernel_spec_manager
+        assert wrapped_manager is not None
+
         return isinstance(wrapped_manager, GatewayKernelSpecManager)
 
 
@@ -215,10 +222,12 @@ class RoutingKernelManager(ServerKernelManager):
     def is_remote(self):
         if not self.kernel_name or not self.kernel_id:
             return False
+        assert self.parent is not None
         return self.parent.kernel_spec_manager.is_remote(self.kernel_name)
 
     @property
     def wrapped_multi_kernel_manager(self):
+        assert self.parent is not None
         return self.parent.kernel_spec_manager.get_mapping_kernel_manager(self.kernel_name)
 
     @property