gh-157044: Fix asyncio call graph truncation at aiter(callable, stop)

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, 'aw_wrapped'):
+            # gh-157044: aiter(callable, stop) awaitable wrapping the callable
+            coro = coro.aw_wrapped
         else:
             break
 

Lib/test/test_asyncgen.py

@@ -906,6 +906,26 @@ async def spam():
         self.assertEqual(self.loop.run_until_complete(awaitable), 1)
         self.assertEqual(calls, [1])
 
+    def test_aiter_callable_aw_wrapped(self):
+        # gh-157044: two-arg aiter's awaitable exposes the wrapped coroutine
+        class Pend:
+            def __await__(self):
+                yield
+
+        async def produce():
+            await Pend()
+            return 1
+
+        aw = anext(aiter(produce, None))
+        self.assertFalse(hasattr(aw, 'aw_wrapped'))
+        try:
+            aw.send(None)
+            self.assertIsNotNone(aw.aw_wrapped)
+            with self.assertRaises(AttributeError):
+                aw.aw_wrapped = None
+        finally:
+            aw.close()
+
     def test_aiter_callable_awaitable(self):
         it = aiter(self.make_counter(), 10)
         awaitable = it.__anext__()

Lib/test/test_asyncio/test_graph.py

@@ -173,6 +173,42 @@ class FakeCoro:
 
         self.assertEqual(len(result.call_stack), 2)
 
+    async def test_stack_aiter_callable(self):
+        # gh-157044: aiter(callable, stop) must not truncate the stack
+        fut = asyncio.Future()
+        stack_for_worker = None
+
+        async def deep():
+            await fut
+
+        async def produce():
+            await deep()
+            return 'stop'
+
+        async def worker():
+            async for _ in aiter(produce, 'stop'):
+                pass
+
+        async def main():
+            nonlocal stack_for_worker
+
+            async with asyncio.TaskGroup() as g:
+                t = g.create_task(worker(), name='worker')
+                for _ in range(5):
+                    await asyncio.sleep(0)
+
+                stack_for_worker = capture_test_stack(fut=t)
+                aw = t.get_coro().cr_await
+                self.assertIsNotNone(getattr(aw, 'aw_wrapped', None))
+                fut.set_result(None)
+
+        await main()
+
+        self.assertEqual(stack_for_worker[0][:2], [
+            'T<worker>',
+            ['a deep', 'a produce', 'a worker'],
+        ])
+
     async def test_stack_gather(self):
 
         stack_for_deep = None

Misc/NEWS.d/next/Library/2026-09-08-03-05-02.gh-issue-157044.jL4L3j.rst

@@ -0,0 +1,2 @@
+:func:`asyncio.print_call_graph` no longer truncates the call stack of a
+task suspended inside ``aiter(callable, stop_value)``.

Objects/iterobject.c

@@ -1,6 +1,7 @@
 /* Iterator objects */
 
 #include "Python.h"
+#include "structmember.h"         // PyMemberDef
 #include "pycore_abstract.h"      // _PyObject_HasLen()
 #include "pycore_call.h"          // _PyObject_CallNoArgs()
 #include "pycore_ceval.h"         // _PyEval_GetBuiltin()
@@ -939,6 +940,12 @@ static PyMethodDef acallawaitable_methods[] = {
     {NULL, NULL}        /* Sentinel */
 };
 
+static PyMemberDef acallawaitable_members[] = {
+    {"aw_wrapped", Py_T_OBJECT_EX, offsetof(acallawaitableobject, aw_wrapped),
+     Py_READONLY, "the awaitable returned by the callable"},
+    {NULL}      /* Sentinel */
+};
+
 static PyAsyncMethods acallawaitable_as_async = {
     PyObject_SelfIter,                          /* am_await */
     0,                                          /* am_aiter */
@@ -958,4 +965,5 @@ PyTypeObject _PyACallIterAwaitable_Type = {
     .tp_iter = PyObject_SelfIter,
     .tp_iternext = acallawaitable_iternext,
     .tp_methods = acallawaitable_methods,
+    .tp_members = acallawaitable_members,
 };