Fail fast on None invariants instead of degraded fallbacks

Address review feedback on the typing pass: replace silently-degraded None fallbacks with fail-fast asserts, and make ThreadSafeSingleton.assert_locked side-effect-free.

- singleton.py: assert_locked now checks lock._is_owned() instead of acquire()/release(), avoiding RLock recursion-count mutation and an unbalanced release() under python -O.

- servers.py: assert pydevdSystemInfo result is non-exception (pid/ppid always assigned before use); assert listener in inject() rather than connecting to an empty address.

- clients.py: assert servers.listener in attach_request and the client listener in notify_of_subprocess instead of substituting ("", 0)/None.

- debuggee.py: assert process in wait_for_exit rather than reporting a bogus clean exit (code 0).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: rchiodo

src/debugpy/adapter/clients.py

@@ -490,11 +490,10 @@ def attach_request(self, request):
         else:
             if not servers.is_serving():
                 servers.serve(localhost)
-            host, port = (
-                sockets.get_address(servers.listener)
-                if servers.listener is not None
-                else ("", 0)
-            )
+            # servers.serve() above guarantees a listener; fail fast if it's missing
+            # rather than handing the debuggee an empty ("", 0) address it can't use.
+            assert servers.listener is not None
+            host, port = sockets.get_address(servers.listener)
 
         # There are four distinct possibilities here.
         #
@@ -772,7 +771,11 @@ def notify_of_subprocess(self, conn):
             localhost = sockets.get_default_localhost()
             body["connect"]["host"] = host or localhost
         if "port" not in body["connect"]:
-            if port is None and listener is not None:
+            if port is None:
+                # A subprocess is only reported once the client is connected, so the
+                # client listener must be serving; fail fast rather than sending the
+                # child a null port it can't connect to.
+                assert listener is not None
                 _, port = sockets.get_address(listener)
             body["connect"]["port"] = port
 

src/debugpy/adapter/servers.py

@@ -80,10 +80,13 @@ def __init__(self, sock: socket.socket):
         try:
             self.authenticate()
             info = self.channel.request("pydevdSystemInfo")
-            if not isinstance(info, Exception):
-                process_info: Callable[..., int] = cast(Callable[..., int], info("process", json.object()))
-                self.pid = process_info("pid", int)
-                self.ppid = process_info("ppid", int, optional=True)
+            # channel.request() either returns a non-exception result or raises, so
+            # assert (rather than silently skipping) to fail fast and to narrow the
+            # type, keeping pid/ppid unconditionally assigned before they're read.
+            assert not isinstance(info, Exception)
+            process_info: Callable[..., int] = cast(Callable[..., int], info("process", json.object()))
+            self.pid = process_info("pid", int)
+            self.ppid = process_info("ppid", int, optional=True)
             if self.ppid == ():
                 self.ppid = None
             self.channel.name = stream.name = str(self)
@@ -484,7 +487,11 @@ def dont_wait_for_first_connection():
 
 
 def inject(pid, debugpy_args, on_output):
-    host, port = sockets.get_address(listener) if listener is not None else ("", 0)
+    # inject() is only reached for attach-by-PID, where the server listener must be
+    # serving so the injected debug server can connect back. Fail fast if it isn't,
+    # rather than spawning an injector that connects to ":0" and never attaches.
+    assert listener is not None
+    host, port = sockets.get_address(listener)
 
     cmdline = [
         sys.executable,

src/debugpy/common/singleton.py

@@ -144,12 +144,15 @@ def __init__(self, *args, **kwargs):
     def assert_locked(self):
         lock = type(self)._lock
         assert lock is not None
-        assert lock.acquire(blocking=False), (
+        # Side-effect-free check: verify the current thread already owns the lock,
+        # rather than acquire()/release() (which mutates the RLock recursion count
+        # and, under `python -O` with the assert stripped, would leak a release()).
+        # _is_owned() is a private but stable RLock helper (used by threading itself).
+        assert lock._is_owned(), (  # pyright: ignore[reportAttributeAccessIssue]
             "ThreadSafeSingleton accessed without locking. Either use with-statement, "
             "or if it is a method or property, mark it as @threadsafe_method or with "
             "@autolocked_method, as appropriate."
         )
-        lock.release()
 
     def __getattribute__(self, name):
         value = object.__getattribute__(self, name)

src/debugpy/launcher/debuggee.py

@@ -198,7 +198,11 @@ def kill():
 
 def wait_for_exit():
     try:
-        code = process.wait() if process is not None else 0
+        # wait_for_exit() only runs after the debuggee was spawned successfully, so
+        # process is always set here; assert to fail fast rather than reporting a
+        # bogus clean exit (code 0) if that invariant is ever broken.
+        assert process is not None
+        code = process.wait()
         if sys.platform != "win32" and code < 0:
             # On POSIX, if the process was terminated by a signal, Popen will use
             # a negative returncode to indicate that - but the actual exit code of