FIX: Ensure coroutines are converted to tasks prior to asyncio.wait

Author: ghisvail

pydra/engine/workers.py

@@ -58,7 +58,11 @@ async def fetch_finished(self, futures):
         done = set()
         try:
             done, pending = await asyncio.wait(
-                futures, return_when=asyncio.FIRST_COMPLETED
+                [
+                    asyncio.create_task(f) if not isinstance(f, asyncio.Task) else f
+                    for f in futures
+                ],
+                return_when=asyncio.FIRST_COMPLETED,
             )
         except ValueError:
             # nothing pending!
@@ -105,7 +109,11 @@ async def fetch_finished(self, futures):
         try:
             self._jobs += len(futures)
             done, pending = await asyncio.wait(
-                futures, return_when=asyncio.FIRST_COMPLETED
+                [
+                    asyncio.create_task(f) if not isinstance(f, asyncio.Task) else f
+                    for f in futures
+                ],
+                return_when=asyncio.FIRST_COMPLETED,
             )
         except ValueError:
             # nothing pending!