feat: asyncio.run() works in sync tasks; sync-task calls from async must use .aio (#1472)

Two related changes to how sync tasks and sync-task calls behave.

## 1. `asyncio.run()` now works inside sync tasks

Sync task bodies used to run inside a hidden coroutine on a per-call
helper event loop (`run_sync_with_loop`, from #565). That loop was never
usable — `asyncio.run()` raised "cannot be called from a running event
loop" — it just made the thread look async. Sync bodies now run as plain
sync code on a dedicated daemon thread (`run_sync_in_thread`).
`asyncio.run()` and third-party sync wrappers (e.g. pydantic_ai
`run_sync`) work there now.

## 2. Calling a sync task from async code requires `.aio`

```python
@env.task
def child() -> str: ...

@env.task
async def parent() -> str:
    return child()              # raises SyncTaskCallInAsyncContextError
    return await child.aio()    # correct
```

A blocking call from async code parks the event loop that also runs the
controller failure watch. If the actions service goes down, the process
can never observe it and the pod hangs forever. It now raises
immediately with the fix in the message.

## What breaks

- **Blocking sync-task calls from async code** now raise. Migrate to
`await task.aio(...)`. (In-repo callers updated: one test, two
examples.)
- **Sync task bodies no longer see a running event loop** —
`asyncio.get_running_loop()` raises there. Nothing in-repo relied on it;
the agents plugin bridge doesn't depend on it (docstring updated).
- **`flyte._utils.asyncify.run_sync_with_loop` → `run_sync_in_thread`**
(private; papermill and docker_builder callers updated — plugin releases
should pair with this SDK version).

Sync → sync calls, `flyte.map`, sandbox orchestrators, and syncify-based
sync APIs (`download_sync`, `storage.*`, `flyte.run`) are unaffected:
they never used the caller's loop.

## Verified on a real cluster

Remote run on demo with this branch baked into the image:
https://demo.hosted.unionai.cloud/v2/domain/development/project/ketan/runs/ukf2hnfjhtrpvpbc4dqc
— sync body sees no running loop, `asyncio.run()` returns, sync→sync
blocking call works, `await leaf.aio()` from async works, and a blocking
call from an async body raises `SyncTaskCallInAsyncContextError` with
the `.aio` hint.
`tests/flyte/test_sync_tasks.py::test_sync_task_through_runtime_taskrunner`
covers the same through the real runtime entrypoint (local runs don't go
through taskrunner).

## Companion

`flyteplugins-union` is the only external importer of the renamed
helper: unionai/flyteplugins-union#100 adopts `run_sync_in_thread` with
an `ImportError` fallback, so it loads on both SDK versions.

## Tests

`tests/flyte/test_sync_call_guard.py` (raises from async, `.aio` ok,
sync→sync ok, end-to-end local run),
`tests/flyte/utils/test_asyncify.py` updated for the new semantics. Full
sweep: 4577 passed; 3 environment-dependent failures reproduce without
this diff.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Ketan Umare <kumare3@users.noreply.github.com>
Co-authored-by: Ketan Umare <kumare3@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: kumare3

examples/advanced/local_tasks.py

@@ -101,7 +101,7 @@ def noio_task():
 @env.task
 async def parallel_main_no_io(q: str) -> int:
     print("Starting parallel_main_no_io", flush=True)
-    noio_task()
+    await noio_task.aio()
     await input_trace("hello world", "blah", 42)
     a = await output_trace()
     await noio_trace()

examples/basics/dir_download_sync_repro.py

@@ -49,7 +49,9 @@ def download_directory_sync(d: Dir) -> list[str]:
 async def main() -> list[str]:
     remote_dir = await create_remote_directory()
     # Pass the Dir as an input to a sync task, which downloads it via download_sync().
-    return download_directory_sync(d=remote_dir)
+    # Sync tasks must be awaited via .aio() from an async parent — a blocking call would
+    # stall the parent's event loop and now raises SyncTaskCallInAsyncContextError.
+    return await download_directory_sync.aio(d=remote_dir)
 
 
 if __name__ == "__main__":

plugins/agents/core/src/flyteplugins/agents/core/_sync.py

@@ -5,10 +5,12 @@
 and report I/O) is async end to end. A synchronous variant therefore has to run
 that coroutine on some event loop. Two loops are deliberately avoided:
 
-- The caller's thread's loop. Inside a Flyte sync task the body executes via
-  run_sync_with_loop, so the current thread already has a running loop and
-  asyncio.run() would raise. This also breaks agent SDKs' own sync wrappers
-  (for example pydantic_ai run_sync), which is why adapters need this bridge.
+- A per-call throwaway loop (asyncio.run). Inside a Flyte sync task the body
+  executes as plain sync code on a dedicated thread (run_sync_in_thread), so
+  asyncio.run() works — but it tears its loop down on every call, which breaks
+  async resources the agent SDKs cache across calls (HTTP clients, connection
+  pools bound to a dead loop). It also raises if the caller happens to be on a
+  thread that already runs a loop, which is why adapters need this bridge.
 - The SDK-global syncify loop. That loop is reserved for short control-plane
   I/O (see the guidance in flyte._trace); parking a whole agent run on it
   stalls every other syncify user in the process and risks deadlocks when

plugins/agents/core/tests/test_sync.py

@@ -40,9 +40,9 @@ async def _read() -> str:
 
 
 def test_run_coro_sync_inside_running_loop_thread():
-    # A Flyte sync task body executes on a thread that already has a running
-    # event loop (run_sync_with_loop), where asyncio.run() would raise. The
-    # bridge must still work from that shape.
+    # The bridge must also work when the calling thread already runs an event
+    # loop (e.g. sync helper code invoked from async code), where asyncio.run()
+    # would raise.
     async def _outer() -> str:
         return run_coro_sync(_echo("nested"))
 

plugins/mlflow/src/flyteplugins/mlflow/_decorator.py

@@ -481,7 +481,7 @@ def decorator(func: F) -> F:
 
         # Task template — wrap func.func (not func.execute) so that
         # mlflow.start_run() runs in the same thread as the task function.
-        # Flyte runs sync tasks in a separate thread via run_sync_with_loop;
+        # Flyte runs sync tasks in a separate thread via run_sync_in_thread;
         # MLflow uses threading.local for its active run stack, so starting
         # the run in the async execute thread would be invisible to the task.
         if isinstance(func, AsyncFunctionTaskTemplate):

plugins/papermill/src/flyteplugins/papermill/task.py

@@ -655,7 +655,7 @@ async def _build():
     async def execute(self, *args: Any, **kwargs: Any) -> Any:
         """Execute the notebook within a Flyte task context."""
         from flyte._context import internal_ctx
-        from flyte._utils.asyncify import run_sync_with_loop
+        from flyte._utils.asyncify import run_sync_in_thread
 
         kwargs = self.interface.convert_to_kwargs(*args, **kwargs)
 
@@ -674,7 +674,7 @@ def _run():
             literal_map = None
             execution_error: Optional[BaseException] = None
             try:
-                literal_map = await run_sync_with_loop(_run)
+                literal_map = await run_sync_in_thread(_run)
             except Exception as exc:
                 execution_error = exc
             finally:

plugins/pytorch/src/flyteplugins/pytorch/task.py

@@ -348,7 +348,7 @@ async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R:
 
             # elastic_launch must run on the main thread so it can register
             # signal handlers (SIGTERM/SIGINT) for cleaning up worker
-            # subprocesses.  Running it in a thread pool (run_sync_with_loop)
+            # subprocesses.  Running it in a thread pool (run_sync_in_thread)
             # would cause the "Failed to register signal handlers" warning
             # and leave orphaned workers on exit.
             #

src/flyte/_internal/imagebuild/docker_builder.py

@@ -54,7 +54,7 @@
     pixi_script_to_project,
 )
 from flyte._logging import logger
-from flyte._utils.asyncify import run_sync_with_loop
+from flyte._utils.asyncify import run_sync_in_thread
 
 if TYPE_CHECKING:
     from flyte._build import ImageBuild
@@ -770,11 +770,11 @@ async def _build_from_dockerfile(self, image: Image, push: bool, wait: bool = Tr
 
         try:
             if wait:
-                await run_sync_with_loop(
+                await run_sync_in_thread(
                     subprocess.run, command, cwd=str(cast(Path, image.dockerfile).cwd()), check=True
                 )
             else:
-                await run_sync_with_loop(subprocess.Popen, command, cwd=str(cast(Path, image.dockerfile).cwd()))
+                await run_sync_in_thread(subprocess.Popen, command, cwd=str(cast(Path, image.dockerfile).cwd()))
         except subprocess.CalledProcessError as e:
             from flyte.errors import ImageBuildError
 
@@ -790,7 +790,7 @@ async def _ensure_buildx_builder():
 
         # Check if buildx is available
         try:
-            await run_sync_with_loop(
+            await run_sync_in_thread(
                 subprocess.run, ["docker", "buildx", "version"], check=True, stdout=subprocess.DEVNULL
             )
         except FileNotFoundError:
@@ -803,7 +803,7 @@ async def _ensure_buildx_builder():
             raise ImageBuildError("Docker buildx is not available. Make sure BuildKit is installed and enabled.")
 
         try:
-            result = await run_sync_with_loop(
+            result = await run_sync_in_thread(
                 subprocess.run, ["docker", "buildx", "ls"], capture_output=True, text=True, check=True
             )
         except subprocess.CalledProcessError as e:
@@ -817,7 +817,7 @@ async def _ensure_buildx_builder():
         # Check if there's any usable builder with the correct driver options
         if DockerImageBuilder._builder_name in builders:
             # Builder exists — verify it has network=host driver option
-            inspect_result = await run_sync_with_loop(
+            inspect_result = await run_sync_in_thread(
                 subprocess.run,
                 ["docker", "buildx", "inspect", DockerImageBuilder._builder_name],
                 capture_output=True,
@@ -829,7 +829,7 @@ async def _ensure_buildx_builder():
 
             # Builder exists but missing network=host, remove and recreate
             logger.info("Buildx builder exists but missing network=host driver option, recreating...")
-            await run_sync_with_loop(
+            await run_sync_in_thread(
                 subprocess.run,
                 ["docker", "buildx", "rm", DockerImageBuilder._builder_name],
                 check=False,
@@ -838,7 +838,7 @@ async def _ensure_buildx_builder():
             logger.info("No buildx builder found, creating one...")
 
         try:
-            await run_sync_with_loop(
+            await run_sync_in_thread(
                 subprocess.run,
                 [
                     "docker",
@@ -955,9 +955,9 @@ async def _build_image(self, image: Image, *, push: bool = True, dry_run: bool =
 
             try:
                 if wait:
-                    await run_sync_with_loop(subprocess.run, command, check=True)
+                    await run_sync_in_thread(subprocess.run, command, check=True)
                 else:
-                    await run_sync_with_loop(subprocess.Popen, command)
+                    await run_sync_in_thread(subprocess.Popen, command)
             except subprocess.CalledProcessError as e:
                 from flyte.errors import ImageBuildError
 

src/flyte/_task.py

@@ -24,7 +24,12 @@
 )
 
 from flyte._pod import PodTemplate
-from flyte.errors import RuntimeSystemError, RuntimeUserError, TraceDoesNotAllowNestedTasksError
+from flyte.errors import (
+    RuntimeSystemError,
+    RuntimeUserError,
+    SyncTaskCallInAsyncContextError,
+    TraceDoesNotAllowNestedTasksError,
+)
 
 from ._cache import Cache, CacheRequest
 from ._context import internal_ctx
@@ -363,6 +368,21 @@ def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Coroutine[Any, Any, R]
                     raise RuntimeSystemError("BadContext", "Controller is not initialized.")
 
                 if self._call_as_synchronous:
+                    # A blocking call is only safe from a plain thread. Sync task bodies run in one
+                    # (run_sync_in_thread), so this trips only inside async code, where blocking the
+                    # event loop can hang the run (it also services the controller failure watch).
+                    in_event_loop = True
+                    try:
+                        asyncio.get_running_loop()
+                    except RuntimeError:
+                        in_event_loop = False
+                    if in_event_loop:
+                        call_name = getattr(getattr(self, "func", None), "__name__", self.name)
+                        raise SyncTaskCallInAsyncContextError(
+                            f"Sync task '{self.name}' was called in a blocking way from async code. "
+                            f"This blocks the event loop and can hang the run. "
+                            f"Use `await {call_name}.aio(...)` instead."
+                        )
                     fut = controller.submit_sync(self, *args, **kwargs)
                     x = fut.result(None)
                     return x
@@ -573,7 +593,7 @@ async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R:
         This is the execute method that will be called when the task is invoked. It will call the actual function.
         # TODO We may need to keep this as the bare func execute, and need a pre and post execute some other func.
         """
-        from flyte._utils.asyncify import run_sync_with_loop
+        from flyte._utils.asyncify import run_sync_in_thread
 
         ctx = internal_ctx()
         assert ctx.data.task_context is not None, "Function should have already returned if not in a task context"
@@ -583,7 +603,7 @@ async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R:
             if iscoroutinefunction(self.func):
                 v = await self.func(*args, **kwargs)
             else:
-                v = await run_sync_with_loop(self.func, *args, **kwargs)
+                v = await run_sync_in_thread(self.func, *args, **kwargs)
 
             await self.post(v)
         return v

src/flyte/_utils/asyncify.py

@@ -5,6 +5,7 @@
 import inspect
 import random
 import threading
+from concurrent.futures import Future
 from typing import Callable, TypeVar
 
 from typing_extensions import ParamSpec
@@ -15,19 +16,20 @@
 P = ParamSpec("P")
 
 
-async def run_sync_with_loop(
+async def run_sync_in_thread(
     func: Callable[P, T],
     *args: P.args,
     **kwargs: P.kwargs,
 ) -> T:
     """
-    Run a synchronous function from an async context with its own event loop.
+    Run a synchronous function from an async context in a dedicated daemon thread.
 
     This function:
-    - Copies the current context variables and preserves them in the sync function
-    - Creates a new event loop in a separate thread for the sync function
-    - Allows the sync function to potentially use asyncio operations
-    - Returns the result without blocking the calling async event loop
+    - Copies the current context variables into the thread, so the Flyte task context propagates
+    - Runs the function as plain sync code: no event loop exists in the thread, so the function
+      may freely use `asyncio.run()` or third-party sync wrappers (e.g. an SDK's `run_sync`)
+    - Uses a daemon thread, so a stuck function cannot block interpreter exit
+    - Returns the result without blocking the calling event loop
 
     Args:
         func: The synchronous function to run (must not be an async function)
@@ -42,62 +44,40 @@ async def run_sync_with_loop(
 
     Example:
         async def my_async_function():
-            result = await run_sync_with_loop(some_sync_function, arg1, arg2)
+            result = await run_sync_in_thread(some_sync_function, arg1, arg2)
             return result
     """
     # Check if func is an async function
     if inspect.iscoroutinefunction(func):
         raise TypeError(
-            f"Cannot call run_sync_with_loop with async function '{getattr(func, '__name__')}'. "
+            f"Cannot call run_sync_in_thread with async function '{getattr(func, '__name__')}'. "
             "This utility is for running sync functions from async contexts."
         )
 
     copied_ctx = contextvars.copy_context()
-    execute_loop = None
-    execute_loop_created = threading.Event()
 
     # Build thread name with random suffix for uniqueness
     func_name = getattr(func, "__name__", "unknown")
     current_thread = threading.current_thread().name
     random_suffix = f"{random.getrandbits(32):08x}"
     full_thread_name = f"sync-executor-{random_suffix}_from_{current_thread}"
 
-    def _sync_thread_loop_runner() -> None:
-        """This method runs the event loop and should be invoked in a separate thread."""
-        nonlocal execute_loop
+    fut: Future[T] = Future()
+
+    def _runner() -> None:
+        if not fut.set_running_or_notify_cancel():
+            return
         try:
-            execute_loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(execute_loop)
-            logger.debug(f"Created event loop for function '{func_name}' in thread '{full_thread_name}'")
-            execute_loop_created.set()
-            execute_loop.run_forever()
-        except Exception as e:
-            logger.error(f"Exception in thread '{full_thread_name}' running '{func_name}': {e}", exc_info=True)
-            raise
-        finally:
-            if execute_loop:
-                logger.debug(f"Stopping event loop for function '{func_name}' in thread '{full_thread_name}'")
-                execute_loop.stop()
-                execute_loop.close()
-                logger.debug(f"Cleaned up event loop for function '{func_name}' in thread '{full_thread_name}'")
+            fut.set_result(copied_ctx.run(func, *args, **kwargs))
+        except BaseException as e:
+            fut.set_exception(e)
 
     executor_thread = threading.Thread(
         name=full_thread_name,
         daemon=True,
-        target=_sync_thread_loop_runner,
+        target=_runner,
     )
     logger.debug(f"Starting executor thread '{full_thread_name}' for function '{func_name}'")
     executor_thread.start()
 
-    async def async_wrapper():
-        res = copied_ctx.run(func, *args, **kwargs)
-        return res
-
-    # Wait for the loop to be created in a thread to avoid blocking the current thread
-    await asyncio.get_event_loop().run_in_executor(None, execute_loop_created.wait)
-    assert execute_loop is not None
-    fut = asyncio.run_coroutine_threadsafe(async_wrapper(), loop=execute_loop)
-    async_fut = asyncio.wrap_future(fut)
-    result = await async_fut
-
-    return result
+    return await asyncio.wrap_future(fut)