diff --git a/Doc/library/functions.rst b/Doc/library/functions.rst
index 013150535cb089c..5389d9b12332904 100644
--- a/Doc/library/functions.rst
+++ b/Doc/library/functions.rst
@@ -139,8 +139,14 @@ are always available.  They are listed here in alphabetical order.
    iterator. If *default* is given, it is returned if the iterator is exhausted,
    otherwise :exc:`StopAsyncIteration` is raised.
 
+   The awaitable returned by :func:`anext` exposes the generator it drives as a
+   read-only ``ag_gen`` attribute if *async_iterator* is an asynchronous generator.
+
    .. versionadded:: 3.10
 
+   .. versionadded:: 3.16
+      The ``ag_gen`` attribute.
+
 .. function:: any(iterable, /)
 
    Return ``True`` if any element of the *iterable* is true.  If the iterable
diff --git a/Doc/library/stdtypes.rst b/Doc/library/stdtypes.rst
index ec37d29ef283ad0..aa4db5d274f6f9f 100644
--- a/Doc/library/stdtypes.rst
+++ b/Doc/library/stdtypes.rst
@@ -1238,6 +1238,14 @@ the underlying generator function:
    that does nothing.
 
 
+The awaitables returned by :meth:`~agen.asend`, :meth:`~agen.athrow` and
+:meth:`~agen.aclose` expose the generator they drive as a read-only
+``ag_gen`` attribute.
+
+.. versionadded:: 3.16
+   The ``ag_gen`` attribute.
+
+
 .. _typesseq:
 
 Sequence Types --- :class:`list`, :class:`tuple`, :class:`range`
diff --git a/Lib/asyncio/graph.py b/Lib/asyncio/graph.py
index d5db59f5d6f5f36..a3b7ca00f18dedb 100644
--- a/Lib/asyncio/graph.py
+++ b/Lib/asyncio/graph.py
@@ -65,6 +65,9 @@ def _build_graph_for_future(
             # A native async generator or duck-type compatible iterator
             st.append(FrameCallGraphEntry(coro.ag_frame))
             coro = coro.ag_await
+        elif hasattr(coro, 'ag_gen'):
+            # gh-156980, gh-157032: An asend()/athrow()/anext() awaitable: step over it to its generator
+            coro = coro.ag_gen
         else:
             break
 
diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index cdae58b3e89ae36..219f22dc912f1bf 100644
--- a/Lib/test/test_asyncgen.py
+++ b/Lib/test/test_asyncgen.py
@@ -397,6 +397,50 @@ async def gen():
                 r"cannot reuse already awaited __anext__\(\)/asend\(\)"):
             an.send(None)
 
+    def test_async_gen_asend_athrow_ag_gen(self):
+        # gh-156980: 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.ag_gen, g)
+            self.assertIs(athrow.ag_gen, g)
+            with self.assertRaises(AttributeError):
+                asend.ag_gen = None
+        finally:
+            asend.close()
+            athrow.close()
+
+    def test_async_gen_anext_default_ag_gen(self):
+        # gh-157032: anext(agen, default) awaitables expose the generator they drive
+        async def gen():
+            yield 1
+
+        g = gen()
+        an = anext(g, None)
+        try:
+            self.assertIs(an.ag_gen, g)
+            with self.assertRaises(AttributeError):
+                an.ag_gen = None
+        finally:
+            an.close()
+
+        # Non-generator async iterator should not have ag_gen
+        class CustomAsyncIter:
+            async def __anext__(self):
+                return 1
+
+        c = CustomAsyncIter()
+        an_custom = anext(c, None)
+        try:
+            with self.assertRaises(AttributeError):
+                _ = an_custom.ag_gen
+        finally:
+            an_custom.close()
+
     def test_async_gen_asend_throw_concurrent_with_send(self):
         import types
 
diff --git a/Lib/test/test_asyncio/test_graph.py b/Lib/test/test_asyncio/test_graph.py
index 36841672e1f0f65..a1d839ec7a9cfe6 100644
--- a/Lib/test/test_asyncio/test_graph.py
+++ b/Lib/test/test_asyncio/test_graph.py
@@ -173,6 +173,110 @@ class FakeCoro:
 
         self.assertEqual(len(result.call_stack), 2)
 
+    async def test_stack_async_gen_asend(self):
+        # gh-156980: 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-156980: 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_async_gen_anext_default(self):
+        # gh-157032: an anext(agen, default) awaitable 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():
+            await anext(gen(), None)
+
+        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
diff --git a/Misc/NEWS.d/next/Library/2026-09-05-17-29-50.gh-issue-156980.W6kHva.rst b/Misc/NEWS.d/next/Library/2026-09-05-17-29-50.gh-issue-156980.W6kHva.rst
new file mode 100644
index 000000000000000..701e42548d5d269
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-09-05-17-29-50.gh-issue-156980.W6kHva.rst
@@ -0,0 +1,2 @@
+:func:`asyncio.print_call_graph` no longer truncates the call stack of a
+task suspended inside an asynchronous generator.
diff --git a/Misc/NEWS.d/next/Library/2026-09-06-12-00-00.gh-issue-157032.XyZ123.rst b/Misc/NEWS.d/next/Library/2026-09-06-12-00-00.gh-issue-157032.XyZ123.rst
new file mode 100644
index 000000000000000..1e859ee561e7915
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-09-06-12-00-00.gh-issue-157032.XyZ123.rst
@@ -0,0 +1,3 @@
+:func:`asyncio.print_call_graph` no longer loses the call stack of a task
+waiting on :func:`anext` with a default value. The awaitable returned by
+``anext(agen, default)`` now exposes the underlying generator as ``ag_gen``.
\ No newline at end of file
diff --git a/Objects/genobject.c b/Objects/genobject.c
index c313002c723e317..ca3002e8438a6f3 100644
--- a/Objects/genobject.c
+++ b/Objects/genobject.c
@@ -2171,6 +2171,11 @@ async_gen_asend_finalize(PyObject *self)
     }
 }
 
+static PyMemberDef async_gen_asend_memberlist[] = {
+    {"ag_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},
@@ -2215,7 +2220,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 */
@@ -2634,6 +2639,11 @@ async_gen_athrow_finalize(PyObject *op)
     }
 }
 
+static PyMemberDef async_gen_athrow_memberlist[] = {
+    {"ag_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),
@@ -2681,7 +2691,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 */
diff --git a/Objects/iterobject.c b/Objects/iterobject.c
index b5783c92c8eb689..bde25c4dcdd08d0 100644
--- a/Objects/iterobject.c
+++ b/Objects/iterobject.c
@@ -582,6 +582,19 @@ static PyMethodDef anextawaitable_methods[] = {
 };
 
 
+static PyObject *
+anextawaitable_getag_gen(PyObject *op, void *Py_UNUSED(ignored))
+{
+    anextawaitableobject *obj = anextawaitableobject_CAST(op);
+    return PyObject_GetAttrString(obj->wrapped, "ag_gen");
+}
+
+static PyGetSetDef anextawaitable_getsetlist[] = {
+    {"ag_gen", anextawaitable_getag_gen, NULL, NULL},
+    {NULL}      /* Sentinel */
+};
+
+
 static PyAsyncMethods anextawaitable_as_async = {
     PyObject_SelfIter,                          /* am_await */
     0,                                          /* am_aiter */
@@ -619,6 +632,8 @@ PyTypeObject _PyAnextAwaitable_Type = {
     PyObject_SelfIter,                          /* tp_iter */
     anextawaitable_iternext,                    /* tp_iternext */
     anextawaitable_methods,                     /* tp_methods */
+    0,                                          /* tp_members */
+    anextawaitable_getsetlist,                  /* tp_getset */
 };
 
 PyObject *