python
diff --git a/‎Lib/test/test_external_inspection.py
Lines changed: 94 additions & 0 deletions b/‎Lib/test/test_external_inspection.py
Lines changed: 94 additions & 0 deletions
@@ -144,6 +144,100 @@ def foo():
             else:
                 self.fail("Main thread stack trace not found in result")
 
+    @skip_if_not_supported
+    @unittest.skipIf(
+        sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
+        "Test only runs on Linux with process_vm_readv support",
+    )
+    def test_async_remote_stack_trace_all_tasks(self):
+        port = find_unused_port()
+        script = textwrap.dedent(
+            f"""\
+            import asyncio
+            import time
+            import sys
+            import socket
+            # Connect to the test process
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.connect(('localhost', {port}))
+
+            def c5():
+                sock.sendall(b"ready"); time.sleep(10_000)  # same line number
+
+            async def c4():
+                await asyncio.sleep(0)
+                c5()
+
+            async def c3():
+                await c4()
+
+            async def c2():
+                await c3()
+
+            async def c1(task):
+                await task
+
+            async def main():
+                async with asyncio.TaskGroup() as tg:
+                    task = tg.create_task(c2(), name="c2_root")
+                    tg.create_task(c1(task), name="sub_main_1")
+                    tg.create_task(c1(task), name="sub_main_2")
+
+            def new_eager_loop():
+                loop = asyncio.new_event_loop()
+                eager_task_factory = asyncio.create_eager_task_factory(
+                    asyncio.Task)
+                loop.set_task_factory(eager_task_factory)
+                return loop
+
+            asyncio.run(main(), loop_factory={{TASK_FACTORY}})
+            """
+        )
+        stack_trace = None
+        for task_factory_variant in "asyncio.new_event_loop", "new_eager_loop":
+            with (
+                self.subTest(task_factory_variant=task_factory_variant),
+                os_helper.temp_dir() as work_dir,
+            ):
+                script_dir = os.path.join(work_dir, "script_pkg")
+                os.mkdir(script_dir)
+                server_socket = socket.socket(
+                    socket.AF_INET, socket.SOCK_STREAM
+                )
+                server_socket.setsockopt(
+                    socket.SOL_SOCKET, socket.SO_REUSEADDR, 1
+                )
+                server_socket.bind(("localhost", port))
+                server_socket.settimeout(SHORT_TIMEOUT)
+                server_socket.listen(1)
+                script_name = _make_test_script(
+                    script_dir,
+                    "script",
+                    script.format(TASK_FACTORY=task_factory_variant),
+                )
+                client_socket = None
+                try:
+                    p = subprocess.Popen([sys.executable, script_name])
+                    client_socket, _ = server_socket.accept()
+                    server_socket.close()
+                    response = client_socket.recv(1024)
+                    self.assertEqual(response, b"ready")
+                    stack_trace = get_all_awaited_by(p.pid)
+                except PermissionError:
+                    self.skipTest(
+                        "Insufficient permissions to read the stack trace"
+                    )
+                finally:
+                    if client_socket is not None:
+                        client_socket.close()
+                    p.kill()
+                    p.terminate()
+                    p.wait(timeout=SHORT_TIMEOUT)
+                
+                self.assertGreater(len(stack_trace), 0)
+
+ 
+
     @skip_if_not_supported
     @unittest.skipIf(
         sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,