Reply with busy -> error -> idle if subshell unknown

[krassowski]

Fix #1557.

When subshell with given ID does not exist reply with busy then error then idle.

Assisted with Fable 5.1

[krassowski]

A small futureproofing at a cost of an extra UnknownSubshellError class in public API would be:

Details

diff --git a/ipykernel/kernelbase.py b/ipykernel/kernelbase.py
index 343c5ad..b46cc66 100644
--- a/ipykernel/kernelbase.py
+++ b/ipykernel/kernelbase.py
@@ -59,6 +59,7 @@ from ipykernel.jsonutil import json_clean
 
 from ._version import kernel_protocol_version
 from .iostream import OutStream
+from .subshell_manager import UnknownSubshellError
 from .utils import LazyDict, _async_in_context
 
 psutil: t.Any | None = None
@@ -603,8 +604,8 @@ class Kernel(SingletonConfigurable):
                 subshell_manager = self.shell_channel_thread.manager
                 try:
                     socket = subshell_manager.get_shell_channel_to_subshell_socket(subshell_id)
-                except KeyError:
-                    self._send_unknown_subshell_reply(idents, msg3, subshell_id)
+                except UnknownSubshellError as err:
+                    self._send_unknown_subshell_reply(idents, msg3, err)
                     return
                 assert socket is not None
                 socket.send_multipart(msg, copy=False)
@@ -1382,7 +1383,7 @@ class Kernel(SingletonConfigurable):
             ident=idents,
         )
 
-    def _send_unknown_subshell_reply(self, idents, msg, subshell_id) -> None:
+    def _send_unknown_subshell_reply(self, idents, msg, err: UnknownSubshellError) -> None:
         """Send an error reply to a request addressed to a subshell that is not there.
 
         Runs in the shell channel thread, so it writes to the shell socket
@@ -1396,17 +1397,12 @@ class Kernel(SingletonConfigurable):
         if not self.session:
             return
         msg_type = msg["header"]["msg_type"]
-        self.log.warning(
-            "Cannot handle %s %s: unknown subshell_id %r",
-            msg_type,
-            msg["header"]["msg_id"],
-            subshell_id,
-        )
+        self.log.warning("Cannot handle %s %s: %s", msg_type, msg["header"]["msg_id"], err)
         self._publish_status("busy", "shell", parent=msg)
         content = {
             "status": "error",
-            "ename": "KeyError",
-            "evalue": f"Unknown subshell_id {subshell_id!r}",
+            "ename": type(err).__name__,
+            "evalue": str(err),
             "traceback": [],
         }
         md = self.init_metadata(msg)
diff --git a/ipykernel/subshell_manager.py b/ipykernel/subshell_manager.py
index 1f23085..ce4accc 100644
--- a/ipykernel/subshell_manager.py
+++ b/ipykernel/subshell_manager.py
@@ -19,6 +19,20 @@ from .thread import SHELL_CHANNEL_THREAD_NAME
 from .utils import _async_in_context
 
 
+class UnknownSubshellError(KeyError):
+    """A subshell_id that does not name an existing subshell.
+
+    .. versionadded:: 7.4
+    """
+
+    def __init__(self, subshell_id: str) -> None:
+        super().__init__(subshell_id)
+        self.subshell_id = subshell_id
+
+    def __str__(self) -> str:
+        return f"Unknown subshell_id {self.subshell_id!r}"
+
+
 class SubshellManager:
     """A manager of subshells.
 
@@ -87,13 +101,24 @@ class SubshellManager:
         self._main_to_shell_channel.close()
         self._shell_channel_to_main.close()
 
+    def _get_subshell(self, subshell_id: str) -> SubshellThread:
+        """Return the thread of the specified subshell.
+
+        The caller must hold ``_lock_cache``.  Raises ``UnknownSubshellError`` if
+        there is no such subshell.
+        """
+        try:
+            return self._cache[subshell_id]
+        except KeyError:
+            raise UnknownSubshellError(subshell_id) from None
+
     def get_shell_channel_to_subshell_pair(self, subshell_id: str | None) -> SocketPair:
         """Return the inproc socket pair used to send messages from the shell channel
         to a particular subshell or main shell."""
         if subshell_id is None:
             return self._shell_channel_to_main
         with self._lock_cache:
-            return self._cache[subshell_id].shell_channel_to_subshell
+            return self._get_subshell(subshell_id).shell_channel_to_subshell
 
     def get_subshell_to_shell_channel_socket(self, subshell_id: str | None) -> zmq.Socket[t.Any]:
         """Return the socket used by a particular subshell or main shell to send
@@ -102,7 +127,7 @@ class SubshellManager:
         if subshell_id is None:
             return self._main_to_shell_channel.from_socket
         with self._lock_cache:
-            return self._cache[subshell_id].subshell_to_shell_channel.from_socket
+            return self._get_subshell(subshell_id).subshell_to_shell_channel.from_socket
 
     def get_shell_channel_to_subshell_socket(self, subshell_id: str | None) -> zmq.Socket[t.Any]:
         """Return the socket used by the shell channel to send messages to a particular
@@ -113,12 +138,12 @@ class SubshellManager:
     def get_subshell_aborting(self, subshell_id: str) -> bool:
         """Get the boolean aborting flag of the specified subshell."""
         with self._lock_cache:
-            return self._cache[subshell_id].aborting
+            return self._get_subshell(subshell_id).aborting
 
     def get_subshell_asyncio_lock(self, subshell_id: str) -> asyncio.Lock:
         """Return the asyncio lock belonging to the specified subshell."""
         with self._lock_cache:
-            return self._cache[subshell_id].asyncio_lock
+            return self._get_subshell(subshell_id).asyncio_lock
 
     def list_subshell(self) -> list[str]:
         """Return list of current subshell ids.
@@ -141,7 +166,7 @@ class SubshellManager:
     def set_subshell_aborting(self, subshell_id: str, aborting: bool) -> None:
         """Set the aborting flag of the specified subshell."""
         with self._lock_cache:
-            self._cache[subshell_id].aborting = aborting
+            self._get_subshell(subshell_id).aborting = aborting
 
     def subshell_id_from_thread_id(self, thread_id: int) -> str | None:
         """Return subshell_id of the specified thread_id.
@@ -185,14 +210,15 @@ class SubshellManager:
     def _delete_subshell(self, subshell_id: str) -> None:
         """Delete subshell identified by subshell_id.
 
-        Raises key error if subshell_id not in cache.
+        Raises ``UnknownSubshellError`` if subshell_id not in cache.
         """
         assert current_thread().name == SHELL_CHANNEL_THREAD_NAME
 
         with self._lock_cache:
-            subshell_threwad = self._cache.pop(subshell_id)
+            subshell_thread = self._get_subshell(subshell_id)
+            del self._cache[subshell_id]
 
-        self._stop_subshell(subshell_threwad)
+        self._stop_subshell(subshell_thread)
 
     def _process_control_request(
         self,
diff --git a/tests/test_subshells.py b/tests/test_subshells.py
index 485361c..a8e1a1a 100644
--- a/tests/test_subshells.py
+++ b/tests/test_subshells.py
@@ -399,6 +399,11 @@ def test_unknown_subshell_id():
     with new_kernel() as kc:
         subshell_id = create_subshell_helper(kc)["subshell_id"]
         delete_subshell_helper(kc, subshell_id)
+
+        # Deleting it again names the missing subshell in the same way.
+        content = delete_subshell_helper(kc, subshell_id)
+        assert content["status"] == "error"
+        assert content["evalue"] == f"Unknown subshell_id {subshell_id!r}"
         flush_channels(kc)
 
         msg = execute_request(kc, "a = 1", subshell_id)
@@ -406,7 +411,8 @@ def test_unknown_subshell_id():
 
         reply = get_reply(kc, msg_id, TIMEOUT)
         assert reply["content"]["status"] == "error"
-        assert subshell_id in reply["content"]["evalue"]
+        assert reply["content"]["ename"] == "UnknownSubshellError"
+        assert reply["content"]["evalue"] == f"Unknown subshell_id {subshell_id!r}"
 
         states = []
         while True:

LMK if you want me to push a commit with it. Also fine by me to keep it as-is, I just like the error subclassing pattern over using generic classes which can be emitted from other code too.