gh-119710: Let asyncio Process.wait() finish on only process exit.

Letting Process.wait() only wait on actual process return is closer to
how it's documented and consistent with Popen.wait(). This also reduces
complexity for waking waiters which was inconsistend depending on
ordering of wait/exit.

Author: tapetersen

Lib/asyncio/base_subprocess.py

@@ -26,7 +26,6 @@ def __init__(self, loop, protocol, args, shell,
         self._pending_calls = collections.deque()
         self._pipes = {}
         self._finished = False
-        self._pipes_connected = False
 
         if stdin == subprocess.PIPE:
             self._pipes[0] = None
@@ -214,7 +213,6 @@ async def _connect_pipes(self, waiter):
         else:
             if waiter is not None and not waiter.cancelled():
                 waiter.set_result(None)
-            self._pipes_connected = True
 
     def _call(self, cb, *data):
         if self._pending_calls is not None:
@@ -235,6 +233,16 @@ def _process_exited(self, returncode):
         if self._loop.get_debug():
             logger.info('%r exited with return code %r', self, returncode)
         self._returncode = returncode
+
+        # gh-119710: Wake up futures waiting for wait() as soon as the process
+        # exits. The pipe transports now check for the loop being closed before
+        # scheduling a callback preventing gh-114177. This is consistent with
+        # the behavior prior to 3.11 and the documented semantics in _wait().
+        for waiter in self._exit_waiters:
+            if not waiter.done():
+                waiter.set_result(returncode)
+        self._exit_waiters = None
+
         if self._proc.returncode is None:
             # asyncio uses a child watcher: copy the status into the Popen
             # object. On Python 3.6, it is required to avoid a ResourceWarning.
@@ -258,15 +266,7 @@ def _try_finish(self):
         assert not self._finished
         if self._returncode is None:
             return
-        if not self._pipes_connected:
-            # self._pipes_connected can be False if not all pipes were connected
-            # because either the process failed to start or the self._connect_pipes task
-            # got cancelled. In this broken state we consider all pipes disconnected and
-            # to avoid hanging forever in self._wait as otherwise _exit_waiters
-            # would never be woken up, we wake them up here.
-            for waiter in self._exit_waiters:
-                if not waiter.done():
-                    waiter.set_result(self._returncode)
+
         if all(p is not None and p.disconnected
                for p in self._pipes.values()):
             self._finished = True
@@ -276,11 +276,6 @@ def _call_connection_lost(self, exc):
         try:
             self._protocol.connection_lost(exc)
         finally:
-            # wake up futures waiting for wait()
-            for waiter in self._exit_waiters:
-                if not waiter.done():
-                    waiter.set_result(self._returncode)
-            self._exit_waiters = None
             self._loop = None
             self._proc = None
             self._protocol = None

Lib/test/test_asyncio/test_subprocess.py

@@ -128,9 +128,6 @@ def test_proc_exited_no_invalid_state_error_on_exit_waiters(self):
         exit_waiter = self.loop.create_future()
         transport._exit_waiters.append(exit_waiter)
 
-        # _connect_pipes hasn't completed, so _pipes_connected is False.
-        self.assertFalse(transport._pipes_connected)
-
         # Simulate process exit. _try_finish() will set the result on
         # exit_waiter because _pipes_connected is False, and then schedule
         # _call_connection_lost() because _pipes is empty (vacuously all
@@ -436,6 +433,47 @@ async def len_message(message):
         self.assertEqual(output.rstrip(), b'3')
         self.assertEqual(exitcode, 0)
 
+    def test_wait_even_if_pipe_is_open(self):
+        # gh-119710: Process.wait() must return once the process exits even
+        # if its stdout pipe is inherited by a grandchild that keeps it open,
+        # so the pipe never reaches EOF. Otherwise wait() hangs forever
+        # despite the returncode being known.
+
+        async def run():
+            # Just setup a pipe to pass to the grandchild for reading to ensure it dies.
+            # Inheritable is to allow it to be passed on windows
+            r, w = os.pipe()
+            os.set_inheritable(r, True)
+
+            code = textwrap.dedent(f"""\
+                import subprocess, sys
+                subprocess.run([sys.executable, "-c", "import sys;sys.stdin.read()"])
+                """)
+
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable, "-c", code,
+                # This will be inherited by granchild and should not prevent
+                # *this* process from firing .wait().
+                stdout=subprocess.PIPE,
+                stdin=r,
+                pass_fds=(r,) if sys.platform != "win32" else (),
+                close_fds=False if sys.platform == "win32" else True,
+            )
+            os.close(r)
+
+            try:
+                # Ensure we start waiting before the process is killed.
+                wait_proc = asyncio.create_task(proc.wait())
+                await asyncio.sleep(0.1)
+                proc.kill()
+                await asyncio.wait_for(wait_proc, timeout=2.0)
+            finally:
+                os.close(w) # Allows the grandchild to exit
+                if proc.stdout is not None:
+                    await proc.stdout.read()
+
+        self.loop.run_until_complete(run())
+
     def test_empty_input(self):
 
         async def empty_input():

Misc/NEWS.d/next/Library/2026-06-11-10-00-00.gh-issue-119710.Qz7Kp2.rst

@@ -0,0 +1,4 @@
+Fix :mod:`asyncio` subprocess :meth:`~asyncio.subprocess.Process.wait`
+hanging when the process has exited but one of its pipes is kept open by an
+inherited child process (so the pipe never reaches EOF). ``wait()`` now
+returns as soon as the process exits, regardless of the pipes' state.