gh-161083: Fix stale awaited_by nodes on invalid asyncio.gather() arguments

Lib/asyncio/tasks.py

@@ -926,8 +926,6 @@ def _done_callback(fut, cur_task=cur_task):
             if fut.done():
                 done_futs.append(fut)
             else:
-                if cur_task is not None:
-                    futures.future_add_to_awaited_by(fut, cur_task)
                 fut.add_done_callback(_done_callback)
 
         else:
@@ -937,6 +935,11 @@ def _done_callback(fut, cur_task=cur_task):
         children.append(fut)
 
     outer = _GatheringFuture(children, loop=loop)
+    # gh-161083: Invalid arguments must not leave stale awaited-by edges.
+    if cur_task is not None and len(done_futs) < nfuts:
+        for fut in arg_to_fut.values():
+            if not fut.done():
+                futures.future_add_to_awaited_by(fut, 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_tasks.py

@@ -1279,6 +1279,19 @@ async def coro():
 
         self.loop.run_until_complete(self.new_task(self.loop, coro()))
 
+    def test_gather_invalid_arg_does_not_add_awaited_by(self):
+        # gh-161083
+        async def coro():
+            task = self.new_task(self.loop, asyncio.sleep(0))
+            try:
+                with self.assertRaises(TypeError):
+                    asyncio.gather(task, None)
+                self.assertFalse(task._asyncio_awaited_by)
+            finally:
+                await task
+
+        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():

Misc/NEWS.d/next/Library/2026-11-01-19-38-33.gh-issue-161083.PM21lv.rst

@@ -0,0 +1,2 @@
+Fix stale task nodes left by :func:`asyncio.gather` when argument validation
+fails. Patch by Timofei Ivankov