@@ -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
 
@@ -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
@@ -143,6 +140,32 @@ def test_proc_exited_no_invalid_state_error_on_exit_waiters(self):
 
         transport.close()
 
+    def test_wait_returns_on_exit_with_open_pipe(self):
+        # gh-119710: wait() must resolve when the process exits even if a
+        # pipe is still open and never reaches EOF (e.g. inherited by a
+        # grandchild). Otherwise _call_connection_lost() never runs and
+        # _wait() would hang forever despite the returncode being known.
+        transport, protocol = self.create_transport()
+
+        # Pipes are fully connected, but fd 1 stays open (never disconnects).
+        pipe = mock.Mock()
+        pipe.disconnected = False
+        transport._pipes[1] = pipe
+
+        # A waiter registered via _wait() before the process exits.
+        exit_waiter = self.loop.create_future()
+        transport._exit_waiters.append(exit_waiter)
+
+        # _process_exited() must resolve exit_waiter even though the pipe
+        # never disconnects (so _call_connection_lost() never runs). Without
+        # the fix, exit_waiter stays pending forever and this hangs.
+        transport._process_exited(7)
+        self.loop.run_until_complete(exit_waiter)
+
+        self.assertEqual(exit_waiter.result(), 7)
+
+        transport.close()
+
 
 class SubprocessMixin:
 
@@ -436,6 +459,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():
 
@@ -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.