gh-157213: Fix stale asyncio.gather() edges in the await graph

Lib/asyncio/tasks.py

@@ -770,6 +770,11 @@ def cancel(self, msg=None):
         return ret
 
 
+def _discard_awaited_by(children, waiter, outer):
+    for fut in children:
+        futures.future_discard_from_awaited_by(fut, waiter)
+
+
 def gather(*coros_or_futures, return_exceptions=False):
     """Return a future aggregating results from the given coroutines/futures.
 
@@ -903,6 +908,10 @@ def _done_callback(fut, cur_task=cur_task):
         children.append(fut)
 
     outer = _GatheringFuture(children, loop=loop)
+    if cur_task is not None:
+        # gh-157213: a child outliving gather() must lose the awaited-by edge
+        outer.add_done_callback(
+            functools.partial(_discard_awaited_by, children, cur_task))
     # Run done callbacks after GatheringFuture created so any post-processing
     # can be performed at this point
     # optimization: in the special case that *all* futures finished eagerly,

Lib/test/test_asyncio/test_graph.py

@@ -202,6 +202,29 @@ async def main():
             ]
         ])
 
+    async def test_stack_gather_survivor(self):
+        # gh-157213: a child that outlives gather() must not be shown as awaited
+
+        async def fail():
+            raise ValueError
+
+        async def survivor():
+            await asyncio.Future()
+
+        t = asyncio.create_task(survivor(), name='survivor')
+        with self.assertRaises(ValueError):
+            await asyncio.gather(t, fail())
+
+        self.assertEqual(capture_test_stack(fut=t)[0], [
+            'T<survivor>',
+            ['a survivor'],
+            []
+        ])
+
+        t.cancel()
+        with self.assertRaises(asyncio.CancelledError):
+            await t
+
     async def test_stack_shield(self):
 
         stack_for_shield = None

Lib/test/test_asyncio/test_tasks.py

@@ -1231,6 +1231,25 @@ async def coro():
 
         self.loop.run_until_complete(self.new_task(self.loop, coro()))
 
+    def test_gather_discards_awaited_by_for_pending(self):
+        # gh-157213: a child outliving gather() must lose the awaited-by edge
+        async def fail():
+            raise ValueError
+
+        async def survivor():
+            await asyncio.Future()
+
+        async def coro():
+            t = self.new_task(self.loop, survivor())
+            with self.assertRaises(ValueError):
+                await asyncio.gather(t, fail())
+            self.assertFalse(t._asyncio_awaited_by)
+            t.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await t
+
+        self.loop.run_until_complete(self.new_task(self.loop, coro()))
+
     def test_wait_really_done(self):
         # there is possibility that some tasks in the pending list
         # became done but their callbacks haven't all been called yet

Misc/NEWS.d/next/Library/2026-09-09-13-46-38.gh-issue-157213.rYg7pO.rst

@@ -0,0 +1,2 @@
+Fix :func:`asyncio.gather` leaving stale await-graph edges on children that
+outlive it.