Migrate work for finished CurrentThreadExecutor to previous executor

tests/test_sync.py

@@ -1208,3 +1208,72 @@ def test_function(**kwargs: Any) -> None:
 
     # SyncToAsync.__call__.loop.run_in_executor has a param named `task_context`.
     await test_function(task_context=1)
+
+
+def test_nested_task() -> None:
+    async def inner() -> asyncio.Task[None]:
+        return asyncio.create_task(sync_to_async(print)("inner"))
+
+    async def main() -> None:
+        task = await sync_to_async(async_to_sync(inner))()
+        await task
+
+    async_to_sync(main)()
+
+
+def test_nested_task_later() -> None:
+    def later(fut: asyncio.Future[asyncio.Task[None]]) -> None:
+        task = asyncio.create_task(sync_to_async(print)("later"))
+        fut.set_result(task)
+
+    async def inner() -> asyncio.Future[asyncio.Task[None]]:
+        loop = asyncio.get_running_loop()
+        fut = loop.create_future()
+        loop.call_later(0.1, later, fut)
+        return fut
+
+    async def main() -> None:
+        fut = await sync_to_async(async_to_sync(inner))()
+        task = await fut
+        await task
+
+    async_to_sync(main)()
+
+
+def test_double_nested_task() -> None:
+    async def inner() -> asyncio.Task[None]:
+        return asyncio.create_task(sync_to_async(print)("inner"))
+
+    async def outer() -> asyncio.Task[asyncio.Task[None]]:
+        return asyncio.create_task(sync_to_async(async_to_sync(inner))())
+
+    async def main() -> None:
+        outer_task = await sync_to_async(async_to_sync(outer))()
+        inner_task = await outer_task
+        await inner_task
+
+    async_to_sync(main)()
+
+
+def test_two_nested_tasks_with_asyncio_run() -> None:
+    barrier = asyncio.Barrier(3)
+    event = threading.Event()
+
+    async def inner() -> None:
+        task = asyncio.create_task(sync_to_async(event.wait)())
+        await barrier.wait()
+        await task
+
+    async def outer() -> tuple[asyncio.Task[None], asyncio.Task[None]]:
+        task0 = asyncio.create_task(inner())
+        task1 = asyncio.create_task(inner())
+        await barrier.wait()
+        event.set()
+        return task0, task1
+
+    async def main() -> None:
+        task0, task1 = await sync_to_async(async_to_sync(outer))()
+        await task0
+        await task1
+
+    asyncio.run(main())