Fix ready promise and session send (#852)

.github/workflows/main.yml

@@ -130,7 +130,7 @@ jobs:
           pip check
       - name: Run the tests
         run: |
-          pytest -vv jupyter_client || pytest -vv jupyter_client --lf
+          pytest -vv -W default jupyter_client || pytest -vv -W default jupyter_client --lf
 
   make_sdist:
     name: Make SDist

jupyter_client/manager.py

@@ -65,11 +65,12 @@ def in_pending_state(method: F) -> F:
     @functools.wraps(method)
     async def wrapper(self, *args, **kwargs):
         # Create a future for the decorated method
-        try:
-            self._ready = Future()
-        except RuntimeError:
-            # No event loop running, use concurrent future
-            self._ready = CFuture()
+        if self._attempted_start:
+            try:
+                self._ready = Future()
+            except RuntimeError:
+                # No event loop running, use concurrent future
+                self._ready = CFuture()
         try:
             # call wrapped method, await, and set the result or exception.
             out = await method(self, *args, **kwargs)
@@ -96,6 +97,7 @@ class KernelManager(ConnectionFileMixin):
     def __init__(self, *args, **kwargs):
         super().__init__(**kwargs)
         self._shutdown_status = _ShutdownStatus.Unset
+        self._attempted_start = False
         # Create a place holder future.
         try:
             asyncio.get_running_loop()
@@ -382,6 +384,7 @@ async def _async_start_kernel(self, **kw: t.Any) -> None:
              keyword arguments that are passed down to build the kernel_cmd
              and launching the kernel (e.g. Popen kwargs).
         """
+        self._attempted_start = True
         kernel_cmd, kw = await ensure_async(self.pre_start_kernel(**kw))
 
         # launch the kernel subprocess

jupyter_client/session.py

@@ -29,7 +29,7 @@
 from typing import Optional
 from typing import Union
 
-import zmq
+import zmq.asyncio
 from traitlets import Any
 from traitlets import Bool
 from traitlets import CBytes
@@ -807,6 +807,10 @@ def send(
             # ZMQStreams and dummy sockets do not support tracking.
             track = False
 
+        if isinstance(stream, zmq.asyncio.Socket):
+            assert stream is not None
+            stream = zmq.Socket.shadow(stream.underlying)
+
         if isinstance(msg_or_type, (Message, dict)):
             # We got a Message or message dict, not a msg_type so don't
             # build a new Message.