Fix microsoft#865: Debugging through poetry drops subprocess

Handle "exited" { "pydevdReason": "processReplaced" } appropriately.

Add test for os.exec() in-place.

pyproject.toml

@@ -12,3 +12,9 @@ exclude = '''
 | ^/src/debugpy/_version.py
 )
 '''
+
+[tool.pyright]
+pythonVersion = "3.7"
+include = ["src"]
+extraPaths = ["src/debugpy/_vendored/pydevd"]
+ignore = ["src/debugpy/_vendored/pydevd", "src/debugpy/_version.py"]

src/debugpy/adapter/clients.py

@@ -2,6 +2,8 @@
 # Licensed under the MIT License. See LICENSE in the project root
 # for license information.
 
+from __future__ import annotations
+
 import atexit
 import os
 import sys
@@ -17,6 +19,10 @@ class Client(components.Component):
 
     message_handler = components.Component.message_handler
 
+    known_subprocesses: set[servers.Connection]
+    """Server connections to subprocesses that this client has been made aware of.
+    """
+
     class Capabilities(components.Capabilities):
         PROPERTIES = {
             "supportsVariableType": False,
@@ -70,10 +76,7 @@ def __init__(self, sock):
             only if and when the "launch" or "attach" response is sent.
             """
 
-            self._known_subprocesses = set()
-            """servers.Connection instances for subprocesses that this client has been
-            made aware of.
-            """
+            self.known_subprocesses = set()
 
             session.client = self
             session.register()
@@ -630,8 +633,9 @@ def disconnect_request(self, request):
         return {}
 
     def notify_of_subprocess(self, conn):
+        log.info("{1} is a subprocess of {0}.", self, conn)
         with self.session:
-            if self.start_request is None or conn in self._known_subprocesses:
+            if self.start_request is None or conn in self.known_subprocesses:
                 return
             if "processId" in self.start_request.arguments:
                 log.warning(
@@ -643,7 +647,8 @@ def notify_of_subprocess(self, conn):
 
             log.info("Notifying {0} about {1}.", self, conn)
             body = dict(self.start_request.arguments)
-            self._known_subprocesses.add(conn)
+            self.known_subprocesses.add(conn)
+            self.session.notify_changed()
 
         for key in "processId", "listen", "preLaunchTask", "postDebugTask":
             body.pop(key, None)

src/debugpy/adapter/servers.py

@@ -2,6 +2,8 @@
 # Licensed under the MIT License. See LICENSE in the project root
 # for license information.
 
+from __future__ import annotations
+
 import os
 import subprocess
 import sys
@@ -37,14 +39,31 @@ class Connection(object):
     once the session ends.
     """
 
+    disconnected: bool
+
+    process_replaced: bool
+    """Whether this is a connection to a process that is being replaced in situ
+    by another process, e.g. via exec().
+    """
+
+    server: Server | None
+    """The Server component, if this debug server belongs to Session.
+    """
+
+    pid: int | None
+
+    ppid: int | None
+
+    channel: messaging.JsonMessageChannel
+
     def __init__(self, sock):
         from debugpy.adapter import sessions
 
         self.disconnected = False
 
+        self.process_replaced = False
+
         self.server = None
-        """The Server component, if this debug server belongs to Session.
-        """
 
         self.pid = None
 
@@ -109,7 +128,13 @@ def __init__(self, sock):
                 if self.disconnected:
                     return
 
-                if any(conn.pid == self.pid for conn in _connections):
+                # An existing connection with the same PID and process_replaced == True
+                # corresponds to the process that replaced itself with this one, so it's
+                # not an error.
+                if any(
+                    conn.pid == self.pid and not conn.process_replaced
+                    for conn in _connections
+                ):
                     raise KeyError(f"{self} is already connected to this adapter")
 
                 is_first_server = len(_connections) == 0
@@ -130,9 +155,17 @@ def __init__(self, sock):
             return
 
         parent_session = sessions.get(self.ppid)
+        if parent_session is None:
+            parent_session = sessions.get(self.pid)
         if parent_session is None:
             log.info("No active debug session for parent process of {0}.", self)
         else:
+            if self.pid == parent_session.pid:
+                parent_server = parent_session.server
+                if not (parent_server and parent_server.connection.process_replaced):
+                    log.error("{0} is not expecting replacement.", parent_session)
+                    self.channel.close()
+                    return
             try:
                 parent_session.client.notify_of_subprocess(self)
                 return
@@ -218,6 +251,8 @@ class Server(components.Component):
 
     message_handler = components.Component.message_handler
 
+    connection: Connection
+
     class Capabilities(components.Capabilities):
         PROPERTIES = {
             "supportsCompletionsRequest": False,
@@ -340,11 +375,18 @@ def continued_event(self, event):
             self.client.propagate_after_start(event)
 
     @message_handler
-    def exited_event(self, event):
+    def exited_event(self, event: messaging.Event):
         # If there is a launcher, it's handling the exit code.
         if not self.launcher:
             self.client.propagate_after_start(event)
 
+        if event("pydevdReason", str, optional=True) == "processReplaced":
+            # The parent process used some API like exec() that replaced it with another
+            # process in situ. The connection will shut down immediately afterwards, but
+            # we need to keep the corresponding session alive long enough to report the
+            # subprocess to it.
+            self.connection.process_replaced = True
+
     @message_handler
     def terminated_event(self, event):
         # Do not propagate this, since we'll report our own.
@@ -358,6 +400,27 @@ def detach_from_session(self):
             self.connection.server = None
 
     def disconnect(self):
+        if self.connection.process_replaced:
+            # Wait for the replacement server to connect to the adapter, and to report
+            # itself to the client for this session if there is one.
+            log.info("{0} is waiting for replacement subprocess.", self)
+            session = self.session
+            if not session.client or not session.client.is_connected:
+                wait_for_connection(
+                    session, lambda conn: conn.pid == self.pid, timeout=30
+                )
+            else:
+                self.wait_for(
+                    lambda: (
+                        not session.client
+                        or not session.client.is_connected
+                        or any(
+                            conn.pid == self.pid
+                            for conn in session.client.known_subprocesses
+                        )
+                    ),
+                    timeout=30,
+                )
         with _lock:
             _connections.remove(self.connection)
             _connections_changed.set()
@@ -383,8 +446,8 @@ def connections():
 
 
 def wait_for_connection(session, predicate, timeout=None):
-    """Waits until there is a server with the specified PID connected to this adapter,
-    and returns the corresponding Connection.
+    """Waits until there is a server matching the specified predicate connected to
+    this adapter, and returns the corresponding Connection.
 
     If there is more than one server connection already available, returns the oldest
     one.

src/debugpy/adapter/sessions.py

@@ -192,8 +192,9 @@ def _finalize(self, why, terminate_debuggee):
 
         if self.launcher and self.launcher.is_connected:
             # If there was a server, we just disconnected from it above, which should
-            # cause the debuggee process to exit - so let's wait for that first.
-            if self.server:
+            # cause the debuggee process to exit, unless it is being replaced in situ -
+            # so let's wait for that first.
+            if self.server and not self.server.connection.process_replaced:
                 log.info('{0} waiting for "exited" event...', self)
                 if not self.wait_for(
                     lambda: self.launcher.exit_code is not None,
@@ -203,12 +204,16 @@ def _finalize(self, why, terminate_debuggee):
 
             # Terminate the debuggee process if it's still alive for any reason -
             # whether it's because there was no server to handle graceful shutdown,
-            # or because the server couldn't handle it for some reason.
-            self.launcher.terminate_debuggee()
+            # or because the server couldn't handle it for some reason - unless the
+            # process is being replaced in situ.
+            if not (self.server and self.server.connection.process_replaced):
+                self.launcher.terminate_debuggee()
 
             # Wait until the launcher message queue fully drains. There is no timeout
             # here, because the final "terminated" event will only come after reading
-            # user input in wait-on-exit scenarios.
+            # user input in wait-on-exit scenarios. In addition, if the process was
+            # replaced in situ, the launcher might still have more output to capture
+            # from its replacement.
             log.info("{0} waiting for {1} to disconnect...", self, self.launcher)
             self.wait_for(lambda: not self.launcher.is_connected)
 
@@ -229,6 +234,7 @@ def _finalize(self, why, terminate_debuggee):
             if (
                 self.client.start_request is not None
                 and self.client.start_request.command == "launch"
+                and not (self.server and self.server.connection.process_replaced)
             ):
                 servers.stop_serving()
                 log.info(

src/debugpy/common/messaging.py

@@ -9,6 +9,8 @@
 https://microsoft.github.io/debug-adapter-protocol/overview#base-protocol
 """
 
+from __future__ import annotations
+
 import collections
 import contextlib
 import functools
@@ -460,7 +462,7 @@ def describe(self):
         raise NotImplementedError
 
     @property
-    def payload(self):
+    def payload(self) -> MessageDict:
         """Payload of the message - self.body or self.arguments, depending on the
         message type.
         """

src/debugpy/public_api.py

@@ -3,6 +3,7 @@
 # for license information.
 
 from __future__ import annotations
+
 import functools
 import typing
 
@@ -17,10 +18,7 @@
 # than 72 characters per line! - and must be readable when retrieved via help().
 
 
-# Type aliases and protocols must be guarded to avoid runtime errors due to unsupported
-# syntax in Python <3.9; since they aren't annotations, they're eagerly evaluated!
-if typing.TYPE_CHECKING:
-    Endpoint = tuple[str, int]
+Endpoint = typing.Tuple[str, int]
 
 
 def _api(cancelable=False):
@@ -57,7 +55,7 @@ def log_to(__path: str) -> None:
 
 
 @_api()
-def configure(__properties: dict[str] = None, **kwargs) -> None:
+def configure(__properties: dict[str, typing.Any] | None = None, **kwargs) -> None:
     """Sets debug configuration properties that cannot be set in the
     "attach" request, because they must be applied as early as possible
     in the process being debugged.
@@ -113,7 +111,7 @@ def listen(__endpoint: Endpoint | int) -> Endpoint:
 
 
 @_api()
-def connect(__endpoint: Endpoint | int, *, access_token: str = None) -> Endpoint:
+def connect(__endpoint: Endpoint | int, *, access_token: str | None = None) -> Endpoint:
     """Tells an existing debug adapter instance that is listening on the
     specified address to debug this process.
 

tests/debugpy/test_multiproc.py

@@ -537,3 +537,46 @@ def parent():
 
     log.info("Waiting for child process...")
     child_process.wait()
+
+
+def test_subprocess_replace(pyfile, target, run):
+    @pyfile
+    def child():
+        import os
+        import sys
+
+        assert "debugpy" in sys.modules
+
+        from debuggee import backchannel
+
+        backchannel.send(os.getpid())
+
+    @pyfile
+    def parent():
+        import debuggee
+        import os
+        import sys
+
+        debuggee.setup()
+        print(f"execl({sys.executable!r}, {sys.argv[1]!r})")
+        os.execl(sys.executable, sys.executable, sys.argv[1])
+
+    with debug.Session() as parent_session:
+        backchannel = parent_session.open_backchannel()
+        with run(parent_session, target(parent, args=[child])):
+            pass
+
+        expected_child_config = expected_subprocess_config(parent_session)
+        child_config = parent_session.wait_for_next_event("debugpyAttach")
+        child_config.pop("isOutputRedirected", None)
+        assert child_config == expected_child_config
+        parent_session.proceed()
+
+        with debug.Session(child_config) as child_session:
+            with child_session.start():
+                pass
+
+            child_pid = backchannel.receive()
+            #assert child_pid == parent_session.debuggee.pid
+            assert child_pid == child_config["subProcessId"]
+            assert str(child_pid) in child_config["name"]

tests/timeline.py

@@ -295,7 +295,7 @@ def _explain_how_realized(self, expectation, reasons):
             # Otherwise, break it down expectation by expectation.
             message += ":"
             for exp, reason in reasons.items():
-                message += "\n\n   where {exp!r}\n      == {reason!r}"
+                message += f"\n\n   where {exp!r}\n      == {reason!r}"
         else:
             message += "."