@@ -284,6 +284,12 @@ attributes (see :ref:`import-mod-attrs` for module attributes):
 |                 |                   | ``AGEN_SUSPENDED``, or    |
 |                 |                   | ``AGEN_CLOSED``           |
 +-----------------+-------------------+---------------------------+
+| async generator | aga_gen           | async generator driven by |
+| awaitable       |                   | the awaitable returned    |
+|                 |                   | from ``asend()``,         |
+|                 |                   | ``athrow()``, or          |
+|                 |                   | ``aclose()``              |
++-----------------+-------------------+---------------------------+
 | coroutine       | __name__          | name                      |
 +-----------------+-------------------+---------------------------+
 |                 | __qualname__      | qualified name            |
@@ -360,6 +366,11 @@ attributes (see :ref:`import-mod-attrs` for module attributes):
    Add ``gi_state`` attribute to generators, ``cr_state`` attribute to
    coroutines, and ``ag_state`` attribute to async generators.
 
+.. versionchanged:: 3.16
+
+   Add ``aga_gen`` attribute to the awaitables returned by the ``asend()``,
+   ``athrow()`` and ``aclose()`` methods of async generators.
+
 .. function:: getmembers(object[, predicate])
 
    Return all the members of an object in a list of ``(name, value)``
 
@@ -70,6 +70,9 @@ def _build_graph_for_future(
             if coro.gi_frame is not None:
                 st.append(FrameCallGraphEntry(coro.gi_frame))
             coro = coro.gi_yieldfrom
+        elif hasattr(coro, 'aga_gen'):
+            # An asend()/athrow() awaitable: step over it to its generator
+            coro = coro.aga_gen
         else:
             break
 
 
@@ -399,6 +399,23 @@ async def gen():
                 r"cannot reuse already awaited __anext__\(\)/asend\(\)"):
             an.send(None)
 
+    def test_async_gen_asend_athrow_aga_gen(self):
+        # gh-160837: asend()/athrow() awaitables expose the generator they drive
+        async def gen():
+            yield 1
+
+        g = gen()
+        asend = g.asend(None)
+        athrow = g.athrow(ValueError)
+        try:
+            self.assertIs(asend.aga_gen, g)
+            self.assertIs(athrow.aga_gen, g)
+            with self.assertRaises(AttributeError):
+                asend.aga_gen = None
+        finally:
+            asend.close()
+            athrow.close()
+
     def test_async_gen_asend_throw_concurrent_with_send(self):
         import types
 
 
@@ -430,6 +430,77 @@ async def probe(t):
             ['T<probe>', ['a _cancel_and_wait', 'a wait_for', 'a probe'], []],
         ])
 
+    async def test_stack_async_gen_asend(self):
+        # gh-160837: an async_generator_asend must not truncate the stack
+        fut = asyncio.Future()
+        stack_for_consumer = None
+
+        async def deep():
+            await fut
+
+        async def gen():
+            await deep()
+            yield 1
+
+        async def consumer():
+            async for _ in gen():
+                pass
+
+        async def main():
+            nonlocal stack_for_consumer
+
+            async with asyncio.TaskGroup() as g:
+                t = g.create_task(consumer(), name='consumer')
+                for _ in range(5):
+                    await asyncio.sleep(0)
+
+                stack_for_consumer = capture_test_stack(fut=t)
+                fut.set_result(None)
+
+        await main()
+
+        self.assertEqual(stack_for_consumer[0][:2], [
+            'T<consumer>',
+            ['a deep', 'ag gen', 'a consumer'],
+        ])
+
+    async def test_stack_async_gen_aclose(self):
+        # gh-160837: an async_generator_athrow must not truncate the stack
+        fut = asyncio.Future()
+        stack_for_consumer = None
+
+        async def deep():
+            await fut
+
+        async def gen():
+            try:
+                yield 1
+            finally:
+                await deep()
+
+        async def consumer():
+            agen = gen()
+            await anext(agen)
+            await agen.aclose()
+
+        async def main():
+            nonlocal stack_for_consumer
+
+            async with asyncio.TaskGroup() as g:
+                t = g.create_task(consumer(), name='consumer')
+                for _ in range(5):
+                    await asyncio.sleep(0)
+
+                stack_for_consumer = capture_test_stack(fut=t)
+                fut.set_result(None)
+
+        await main()
+
+        self.assertEqual(stack_for_consumer[0][:2], [
+            'T<consumer>',
+            ['a deep', 'ag gen', 'a consumer'],
+        ])
+
     async def test_stack_gather(self):
 
         stack_for_deep = None
 
@@ -0,0 +1,2 @@
+:func:`asyncio.print_call_graph` no longer truncates the call stack of a
+task suspended inside an asynchronous generator.
@@ -2194,6 +2194,11 @@ async_gen_asend_finalize(PyObject *self)
     }
 }
 
+static PyMemberDef async_gen_asend_memberlist[] = {
+    {"aga_gen", Py_T_OBJECT_EX, offsetof(PyAsyncGenASend, ags_gen), Py_READONLY},
+    {NULL}      /* Sentinel */
+};
+
 static PyMethodDef async_gen_asend_methods[] = {
     {"send", async_gen_asend_send, METH_O, send_doc},
     {"throw", _PyCFunction_CAST(async_gen_asend_throw), METH_FASTCALL, throw_doc},
@@ -2238,7 +2243,7 @@ PyTypeObject _PyAsyncGenASend_Type = {
     PyObject_SelfIter,                          /* tp_iter */
     async_gen_asend_iternext,                   /* tp_iternext */
     async_gen_asend_methods,                    /* tp_methods */
-    0,                                          /* tp_members */
+    async_gen_asend_memberlist,                 /* tp_members */
     0,                                          /* tp_getset */
     0,                                          /* tp_base */
     0,                                          /* tp_dict */
@@ -2657,6 +2662,11 @@ async_gen_athrow_finalize(PyObject *op)
     }
 }
 
+static PyMemberDef async_gen_athrow_memberlist[] = {
+    {"aga_gen", Py_T_OBJECT_EX, offsetof(PyAsyncGenAThrow, agt_gen), Py_READONLY},
+    {NULL}      /* Sentinel */
+};
+
 static PyMethodDef async_gen_athrow_methods[] = {
     {"send", async_gen_athrow_send, METH_O, send_doc},
     {"throw", _PyCFunction_CAST(async_gen_athrow_throw),
@@ -2704,7 +2714,7 @@ PyTypeObject _PyAsyncGenAThrow_Type = {
     PyObject_SelfIter,                          /* tp_iter */
     async_gen_athrow_iternext,                  /* tp_iternext */
     async_gen_athrow_methods,                   /* tp_methods */
-    0,                                          /* tp_members */
+    async_gen_athrow_memberlist,                /* tp_members */
     0,                                          /* tp_getset */
     0,                                          /* tp_base */
     0,                                          /* tp_dict */
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	+:func:`asyncio.print_call_graph` no longer truncates the call stack of a
	`2`	`+task suspended inside an asynchronous generator.`