gh-157259: Record awaited-by edge for asyncio subprocess pipe-connect task

Lib/asyncio/base_subprocess.py

@@ -5,6 +5,7 @@
 import signal
 import sys
 
+from . import futures
 from . import protocols
 from . import transports
 from .log import logger
@@ -53,7 +54,12 @@ def __init__(self, loop, protocol, args, shell,
             logger.debug('process %r created: pid %s',
                          program, self._pid)
 
-        self._loop.create_task(self._connect_pipes(waiter))
+        task = self._loop.create_task(self._connect_pipes(waiter))
+        if waiter is not None:
+            futures.future_add_to_awaited_by(task, waiter)
+            task.add_done_callback(
+                lambda t: futures.future_discard_from_awaited_by(t, waiter)
+            )
 
     def __repr__(self):
         info = [self.__class__.__name__]

Lib/test/test_asyncio/test_graph.py

@@ -2,6 +2,7 @@
 import io
 import sys
 import unittest
+from test import support
 from unittest import mock
 
 
@@ -450,6 +451,53 @@ async def main(t):
             ]
         ])
 
+    @support.requires_subprocess()
+    async def test_stack_subprocess_connect_pipes(self):
+        # gh-157259: create_subprocess_exec must record the awaiting waiter / task
+        # on the internal _connect_pipes task
+        loop = asyncio.get_running_loop()
+        in_connect_pipes = asyncio.Event()
+        can_finish = asyncio.Event()
+
+        pipe_task = None
+        orig_connect = loop.connect_write_pipe
+        async def mocked_connect(*args, **kwargs):
+            nonlocal pipe_task
+            pipe_task = asyncio.current_task()
+            in_connect_pipes.set()
+            await can_finish.wait()
+            return await orig_connect(*args, **kwargs)
+
+        loop.connect_write_pipe = mocked_connect
+        spawn = None
+        try:
+            spawn = asyncio.create_task(
+                asyncio.create_subprocess_exec(
+                    sys.executable, '-c', 'pass',
+                    stdin=asyncio.subprocess.PIPE,
+                ),
+                name='spawn',
+            )
+
+            await asyncio.wait_for(in_connect_pipes.wait(), timeout=support.SHORT_TIMEOUT)
+            self.assertIsNotNone(pipe_task)
+
+            stack = capture_test_stack(fut=pipe_task)
+
+            self.assertEqual(stack[0][0], 'T<anon>')
+            awaited_by = stack[0][2]
+            self.assertEqual(len(awaited_by), 1)
+            self.assertEqual(awaited_by[0][0], 'F')
+            self.assertEqual(awaited_by[0][2][0][0], 'T<spawn>')
+        finally:
+            can_finish.set()
+            loop.connect_write_pipe = orig_connect
+            if spawn is not None:
+                proc = await spawn
+                await proc.wait()
+
+        self.assertFalse(pipe_task._asyncio_awaited_by)
+
     async def test_stack_task(self):
 
         stack_for_inner = None

Misc/NEWS.d/next/Library/2026-10-16-21-45-00.gh-issue-157259.wK8pLz.rst

@@ -0,0 +1 @@
+Record the awaiting task in the :mod:`asyncio` call graph for internal pipe-connection tasks created during subprocess spawning.