@@ -1,6 +1,7 @@
 import unittest
 from contextlib import contextmanager
 import asyncio
+import contextlib
 import os
 import textwrap
 import importlib
@@ -68,6 +69,112 @@ def get_all_awaited_by(pid):
 class TestGetStackTrace(unittest.TestCase):
     maxDiff = None
 
+    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_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
     def test_long_task_name_is_truncated(self):
         # gh-157788
@@ -935,22 +1042,10 @@ async def main():
                 for _ in busy_retry(SHORT_TIMEOUT):
                     try:
                         all_awaited_by = get_all_awaited_by(p.pid)
-                    except RuntimeError as re:
-                        # This call reads a linked list in another process with
-                        # no synchronization. That occasionally leads to invalid
-                        # reads. Here we avoid making the test flaky.
-                        msg = str(re)
-                        if msg.startswith("Task list appears corrupted"):
-                            continue
-                        elif msg.startswith(
-                            "Invalid linked list structure reading remote memory"
-                        ):
-                            continue
-                        elif msg.startswith("Unknown error reading memory"):
-                            continue
-                        elif msg.startswith("Unhandled frame owner"):
-                            continue
-                        raise  # Unrecognized exception, safest not to ignore it
+                    except RuntimeError:
+                        # Tasks and awaited objects can change while we read
+                        # the running process. Retry with a fresh sample.
+                        continue
                     else:
                         break
                 # expected: a list of two elements: 1 thread, 1 interp
 
@@ -0,0 +1,2 @@
+Fix ``python -m asyncio ps`` and ``pstree`` stacks stopping at
+asynchronous generators. Patch by Timofei Ivankov.
@@ -1426,11 +1426,63 @@ parse_task_name(
 }
 
 
+// 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;
@@ -1458,41 +1510,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*),
-            &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;
@@ -1515,6 +1540,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(
@@ -1532,8 +1564,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
@@ -1560,8 +1590,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;
         }
     }
@@ -2034,6 +2063,10 @@ get_task_code_object(RemoteUnwinderObject *unwinder, uintptr_t task_addr, uintpt
         return -1;
     }
 
+    if (get_coro_like_address(unwinder, running_coro_addr, &running_coro_addr) < 0) {
+        return -1;
+    }
+
     // note: genobject's gi_iframe is an embedded struct so the address to
     // the offset leads directly to its first field: f_executable
     if (read_py_ptr(
Original file line number	Diff line number	Diff line change
@@ -0,0 +1,2 @@
	1	+ Fix ``python -m asyncio ps`` and ``pstree`` stacks stopping at
	2	+ asynchronous generators. Patch by Timofei Ivankov.