gh-124958: Revert "gh-125472: Revert "gh-124958: fix asyncio.TaskGroup and _PyFuture refcycles ... (#125486)

* Revert "gh-125472: Revert "gh-124958: fix asyncio.TaskGroup and _PyFuture refcycles (#12… (#125476)"

This reverts commit e99650b80a.

* fix incompatability with gh-124392

Lib/test/test_asyncio/test_taskgroups.py

@@ -1,7 +1,8 @@
 # Adapted with permission from the EdgeDB project;
 # license: PSFL.
 
-
+import sys
+import gc
 import asyncio
 import contextvars
 import contextlib
@@ -11,7 +12,6 @@ import warnings
 
 from test.test_asyncio.utils import await_without_task
 
-
 # To prevent a warning "test altered the execution environment"
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
@@ -29,6 +29,15 @@ def get_error_types(eg):
     return {type(exc) for exc in eg.exceptions}
 
 
+def no_other_refs():
+    # due to gh-124392 coroutines now refer to their locals
+    coro = asyncio.current_task().get_coro()
+    frame = sys._getframe(1)
+    while coro.cr_frame != frame:
+        coro = coro.cr_await
+    return [coro]
+
+
 class TestTaskGroup(unittest.IsolatedAsyncioTestCase):
 
     async def test_taskgroup_01(self):
@@ -899,6 +908,95 @@ class TestTaskGroup(unittest.IsolatedAsyncioTestCase):
 
         await outer()
 
+    async def test_exception_refcycles_direct(self):
+        """Test that TaskGroup doesn't keep a reference to the raised ExceptionGroup"""
+        tg = asyncio.TaskGroup()
+        exc = None
+
+        class _Done(Exception):
+            pass
+
+        try:
+            async with tg:
+                raise _Done
+        except ExceptionGroup as e:
+            exc = e
+
+        self.assertIsNotNone(exc)
+        self.assertListEqual(gc.get_referrers(exc), no_other_refs())
+
+
+    async def test_exception_refcycles_errors(self):
+        """Test that TaskGroup deletes self._errors, and __aexit__ args"""
+        tg = asyncio.TaskGroup()
+        exc = None
+
+        class _Done(Exception):
+            pass
+
+        try:
+            async with tg:
+                raise _Done
+        except* _Done as excs:
+            exc = excs.exceptions[0]
+
+        self.assertIsInstance(exc, _Done)
+        self.assertListEqual(gc.get_referrers(exc), no_other_refs())
+
+
+    async def test_exception_refcycles_parent_task(self):
+        """Test that TaskGroup deletes self._parent_task"""
+        tg = asyncio.TaskGroup()
+        exc = None
+
+        class _Done(Exception):
+            pass
+
+        async def coro_fn():
+            async with tg:
+                raise _Done
+
+        try:
+            async with asyncio.TaskGroup() as tg2:
+                tg2.create_task(coro_fn())
+        except* _Done as excs:
+            exc = excs.exceptions[0].exceptions[0]
+
+        self.assertIsInstance(exc, _Done)
+        self.assertListEqual(gc.get_referrers(exc), no_other_refs())
+
+    async def test_exception_refcycles_propagate_cancellation_error(self):
+        """Test that TaskGroup deletes propagate_cancellation_error"""
+        tg = asyncio.TaskGroup()
+        exc = None
+
+        try:
+            async with asyncio.timeout(-1):
+                async with tg:
+                    await asyncio.sleep(0)
+        except TimeoutError as e:
+            exc = e.__cause__
+
+        self.assertIsInstance(exc, asyncio.CancelledError)
+        self.assertListEqual(gc.get_referrers(exc), no_other_refs())
+
+    async def test_exception_refcycles_base_error(self):
+        """Test that TaskGroup deletes self._base_error"""
+        class MyKeyboardInterrupt(KeyboardInterrupt):
+            pass
+
+        tg = asyncio.TaskGroup()
+        exc = None
+
+        try:
+            async with tg:
+                raise MyKeyboardInterrupt
+        except MyKeyboardInterrupt as e:
+            exc = e
+
+        self.assertIsNotNone(exc)
+        self.assertListEqual(gc.get_referrers(exc), no_other_refs())
+
 
 if __name__ == "__main__":
     unittest.main()