@@ -475,6 +475,17 @@ def _extract_coroutine_stacks_lineno_only(self, stack_trace):
 
 @requires_remote_subprocess_debugging()
 class TestSelfStackTrace(RemoteInspectionTestBase):
+    def _extract_coroutine_funcnames(self, trace):
+        return {
+            task.task_name: [
+                frame.funcname.rpartition(".")[2]
+                for coro in task.coroutine_stack
+                for frame in coro.call_stack
+            ]
+            for info in trace
+            for task in info.awaited_by
+        }
+
     @skip_if_not_supported
     def test_long_task_name_is_truncated(self):
         # gh-157788
@@ -508,6 +519,101 @@ async def rec(n):
 
         self.assertEqual(asyncio.run(rec(3)), ["rec"] * 4)
 
+    @skip_if_not_supported
+    def test_async_generator_stack_is_not_truncated(self):
+        # gh-159620
+        async def agen():
+            await asyncio.sleep(10_000)
+            yield
+
+        async def agen_with_cleanup():
+            try:
+                yield
+            finally:
+                await asyncio.sleep(10_000)
+
+        async def outer_agen():
+            async for _ in agen():
+                yield
+
+        async def consumer(gen):
+            async for _ in gen:
+                pass
+
+        async def main():
+            g = agen_with_cleanup()
+            await anext(g)
+            async_for_task = asyncio.create_task(consumer(agen()), name="async for")
+            nested_task = asyncio.create_task(
+                consumer(outer_agen()), name="nested async for")
+            anext_task = asyncio.create_task(anext(agen()), name="anext")
+            aclose_task = asyncio.create_task(g.aclose(), name="aclose")
+            await asyncio.sleep(0)
+            return RemoteUnwinder(os.getpid()).get_all_awaited_by()
+
+        stacks = self._extract_coroutine_funcnames(asyncio.run(main()))
+        self.assertEqual(stacks["async for"], ["sleep", "agen", "consumer"])
+        self.assertEqual(stacks["nested async for"],
+                         ["sleep", "agen", "outer_agen", "consumer"])
+        self.assertEqual(stacks["anext"], ["sleep", "agen"])
+        self.assertEqual(stacks["aclose"], ["sleep", "agen_with_cleanup"])
+
+    @skip_if_not_supported
+    def test_async_generator_waiter_in_async_stack_trace(self):
+        # gh-159620
+        async def agen(task):
+            await task
+            yield
+
+        async def main():
+            waiter_task = asyncio.create_task(
+                anext(agen(asyncio.current_task())), name="anext")
+            await asyncio.sleep(0)
+            return RemoteUnwinder(os.getpid()).get_async_stack_trace()
+
+        stacks = self._extract_coroutine_funcnames(asyncio.run(main()))
+        self.assertEqual(stacks["anext"], ["agen"])
+
+    @skip_if_not_supported
+    def test_async_generator_running_task_stack(self):
+        # gh-159620
+        async def agen():
+            yield RemoteUnwinder(os.getpid()).get_async_stack_trace()
+
+        async def main():
+            return await asyncio.create_task(anext(agen()), name="anext")
+
+        stacks = self._extract_coroutine_funcnames(asyncio.run(main()))
+        self.assertEqual(stacks["anext"], ["agen"])
+
+    @skip_if_not_supported
+    def test_class_named_like_async_generator_wrapper_is_not_walked(self):
+        # gh-159620
+        async def decoy():
+            await asyncio.sleep(10_000)
+
+        class async_generator_asend:
+            __slots__ = ("gen",)
+            def __init__(self, gen):
+                self.gen = gen
+            def __await__(self):
+                return self
+            def __next__(self):
+                return None
+
+        async def consumer(aw):
+            await aw
+
+        async def main():
+            with contextlib.closing(decoy()) as gen:
+                consumer_task = asyncio.create_task(
+                    consumer(async_generator_asend(gen)), name="consumer")
+                await asyncio.sleep(0)
+                return RemoteUnwinder(os.getpid()).get_all_awaited_by()
+
+        stacks = self._extract_coroutine_funcnames(asyncio.run(main()))
+        self.assertEqual(stacks["consumer"], ["consumer"])
+
     @skip_if_not_supported
     @unittest.skipIf(
         sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
 
@@ -0,0 +1,3 @@
+Fix ``python -m asyncio ps``, ``pstree`` and async-aware
+:mod:`profiling.sampling` stacks stopping at asynchronous generators.
+Patch by Timofei Ivankov.
@@ -240,11 +240,63 @@ parse_task_name(
  * COROUTINE CHAIN PARSING
  * ============================================================================ */
 
+// gh-159620
+/* Return in *coro_addr the coroutine or async generator to walk for the
+   awaitable at addr, or 0 if it is neither (for example, a future). */
+static int
+get_coro_like_address(
+    RemoteUnwinderObject *unwinder,
+    uintptr_t addr,
+    uintptr_t *coro_addr
+) {
+    *coro_addr = 0;
+
+    // Skip user classes that share a name with the generator types below
+    uintptr_t type_addr;
+    unsigned long tp_flags;
+    if (read_ptr(unwinder, addr + (uintptr_t)unwinder->debug_offsets.pyobject.ob_type,
+                 &type_addr) < 0
+        || _Py_RemoteDebug_PagedReadRemoteMemory(&unwinder->handle,
+                type_addr + (uintptr_t)unwinder->debug_offsets.type_object.tp_flags,
+                sizeof(tp_flags), &tp_flags) < 0)
+    {
+        set_exception_cause(unwinder, PyExc_RuntimeError, "Failed to read awaitable type");
+        return -1;
+    }
+
+    if (tp_flags & Py_TPFLAGS_HEAPTYPE) {
+        return 0;
+    }
+
+    uintptr_t name_addr;
+    char name[32] = {0};
+    if (read_ptr(unwinder, type_addr + (uintptr_t)unwinder->debug_offsets.type_object.tp_name,
+                 &name_addr) < 0
+        || _Py_RemoteDebug_PagedReadRemoteMemory(&unwinder->handle, name_addr,
+                                                 sizeof(name) - 1, name) < 0)
+    {
+        set_exception_cause(unwinder, PyExc_RuntimeError, "Failed to read awaitable type name");
+        return -1;
+    }
+
+    if (strcmp(name, "coroutine") == 0) {
+        *coro_addr = addr;
+    }
+    else if (strcmp(name, "async_generator_asend") == 0
+             || strcmp(name, "async_generator_athrow") == 0)
+    {
+        // asend and athrow are separate wrapper objects: read the pointer
+        // to their parent async generator, stored right after PyObject_HEAD
+        return read_ptr(unwinder, addr + (uintptr_t)unwinder->debug_offsets.pyobject.size,
+                        coro_addr);
+    }
+    return 0;
+}
+
 static int
 get_awaited_coro_address(
     RemoteUnwinderObject *unwinder,
     uintptr_t gi_iframe_addr,
-    uintptr_t gen_type_addr,
     uintptr_t *next_coro
 ) {
     *next_coro = 0;
@@ -272,41 +324,14 @@ get_awaited_coro_address(
     uintptr_t stackpointer_addr = GET_MEMBER_NO_TAG(uintptr_t, iframe, unwinder->debug_offsets.interpreter_frame.stackpointer);
 
     if ((void*)stackpointer_addr != NULL) {
-        uintptr_t gi_await_addr;
         err = read_py_ptr(
             unwinder,
             stackpointer_addr - sizeof(void*) * 2,
-            &gi_await_addr);
+            next_coro);
         if (err) {
             set_exception_cause(unwinder, PyExc_RuntimeError, "Failed to read gi_await address");
             return -1;
         }
-
-        if ((void*)gi_await_addr != NULL) {
-            uintptr_t gi_await_addr_type_addr;
-            err = read_ptr(
-                unwinder,
-                gi_await_addr + (uintptr_t)unwinder->debug_offsets.pyobject.ob_type,
-                &gi_await_addr_type_addr);
-            if (err) {
-                set_exception_cause(unwinder, PyExc_RuntimeError, "Failed to read gi_await type address");
-                return -1;
-            }
-
-            if (gen_type_addr == gi_await_addr_type_addr) {
-                /* This needs an explanation. We always start with parsing
-                   native coroutine / generator frames. Ultimately they
-                   are awaiting on something. That something can be
-                   a native coroutine frame or... an iterator.
-                   If it's the latter -- we can't continue building
-                   our chain. So the condition to bail out of this is
-                   to do that when the type of the current coroutine
-                   doesn't match the type of whatever it points to
-                   in its cr_await.
-                */
-                *next_coro = gi_await_addr;
-            }
-        }
     }
 
     return 0;
@@ -329,6 +354,13 @@ parse_coro_chain(
             return -1;
         }
 
+        if (get_coro_like_address(unwinder, coro_address, &coro_address) < 0) {
+            return -1;
+        }
+        if ((void*)coro_address == NULL) {
+            return 0;
+        }
+
         // Read the entire generator object at once
         char gen_object[SIZEOF_GEN_OBJ];
         int err = _Py_RemoteDebug_PagedReadRemoteMemory(
@@ -346,8 +378,6 @@ parse_coro_chain(
             return 0;
         }
 
-        uintptr_t gen_type_addr = GET_MEMBER(uintptr_t, gen_object, unwinder->debug_offsets.pyobject.ob_type);
-
         PyObject* name = NULL;
 
         // Parse the previous frame using the gi_iframe from local copy
@@ -374,8 +404,7 @@ parse_coro_chain(
             return 0;
         }
 
-        if (get_awaited_coro_address(unwinder, gi_iframe_addr, gen_type_addr,
-                                     &coro_address) < 0) {
+        if (get_awaited_coro_address(unwinder, gi_iframe_addr, &coro_address) < 0) {
             return -1;
         }
     }
@@ -733,6 +762,10 @@ get_task_frame(RemoteUnwinderObject *unwinder, uintptr_t task_addr, uintptr_t *f
         return -1;
     }
 
+    if (get_coro_like_address(unwinder, running_coro_addr, &running_coro_addr) < 0) {
+        return -1;
+    }
+
     *frame_addr = running_coro_addr + (uintptr_t)unwinder->debug_offsets.gen_object.gi_iframe;
     return 0;
 }
Original file line number	Diff line number	Diff line change
@@ -0,0 +1,3 @@
	1	+ Fix ``python -m asyncio ps``, ``pstree`` and async-aware
	2	+ :mod:`profiling.sampling` stacks stopping at asynchronous generators.
	3	+ Patch by Timofei Ivankov.