Catch missing DELIM in feed_identities on shell and control channels

ipykernel/kernelbase.py

@@ -375,8 +375,8 @@ async def process_control(self, msg):
         """dispatch control requests"""
         if not self.session:
             return
-        idents, msg = self.session.feed_identities(msg, copy=False)
         try:
+            idents, msg = self.session.feed_identities(msg, copy=False)
             msg = self.session.deserialize(msg, content=True, copy=False)
         except Exception:
             self.log.error("Invalid Control Message", exc_info=True)  # noqa: G201
@@ -443,8 +443,8 @@ async def dispatch_shell(
                 self.shell_channel_thread,
             )
 
-        idents, msg = self.session.feed_identities(msg, copy=False)
         try:
+            idents, msg = self.session.feed_identities(msg, copy=False)
             msg = self.session.deserialize(msg, content=True, copy=False)
         except Exception:
             self.log.error("Invalid Message", exc_info=True)  # noqa: G201
@@ -604,8 +604,8 @@ async def shell_channel_thread_main(self, msg):
 
             # deserialize only the header to get subshell_id
             # Keep original message to send to subshell_id unmodified.
-            idents, msg2 = self.session.feed_identities(msg, copy=False)
             try:
+                idents, msg2 = self.session.feed_identities(msg, copy=False)
                 msg3 = self.session.deserialize(msg2, content=False, copy=False)
                 subshell_id = msg3["header"].get("subshell_id")
 

tests/test_kernel_direct.py

@@ -116,13 +116,18 @@ async def test_deprecated_features(kernel):
         kernel.shell_streams = [kernel.shell_stream, kernel.shell_stream]
 
 
-async def test_process_control(kernel):
+async def test_process_control(kernel, caplog):
     from jupyter_client.session import DELIM
 
     class FakeMsg:
         def __init__(self, bytes):
             self.bytes = bytes
 
+    caplog.clear()
+    await kernel.process_control([FakeMsg(b"bad")])
+    assert "Invalid Control Message" in caplog.text
+    assert "DELIM not in msg_list" in caplog.text
+
     await kernel.process_control([FakeMsg(DELIM), 1])
     msg = kernel._prep_msg("does_not_exist")
     await kernel.process_control(msg)
@@ -133,18 +138,48 @@ def test_should_handle(kernel):
     assert kernel.should_handle(kernel.control_stream, msg, []) is True
 
 
-async def test_dispatch_shell(kernel):
+async def test_dispatch_shell(kernel, caplog):
     from jupyter_client.session import DELIM
 
     class FakeMsg:
         def __init__(self, bytes):
             self.bytes = bytes
+            self.buffer = bytes
+
+    caplog.clear()
+    await kernel.dispatch_shell([FakeMsg(b"bad")])
+    assert "Invalid Message" in caplog.text
+    assert "DELIM not in msg_list" in caplog.text
 
     await kernel.dispatch_shell([FakeMsg(DELIM), 1])
     msg = kernel._prep_msg("does_not_exist")
     await kernel.dispatch_shell(msg)
 
 
+async def test_shell_channel_thread_main(kernel, monkeypatch, caplog):
+    import asyncio
+    import threading
+
+    from jupyter_client.session import DELIM
+
+    class FakeMsg:
+        def __init__(self, bytes):
+            self.bytes = bytes
+
+    current = threading.current_thread()
+    monkeypatch.setattr(current, "asyncio_lock", asyncio.Lock(), raising=False)
+    monkeypatch.setattr(kernel, "shell_channel_thread", current)
+
+    caplog.clear()
+    await kernel.shell_channel_thread_main([FakeMsg(b"bad")])
+    assert "Invalid message" in caplog.text
+    assert "DELIM not in msg_list" in caplog.text
+
+    caplog.clear()
+    await kernel.shell_channel_thread_main([FakeMsg(DELIM), 1])
+    assert "Invalid message" in caplog.text
+
+
 async def test_publish_debug_event(kernel):
     kernel._publish_debug_event({})