fix(runtime): run user code on a Proactor loop on Windows in run and script mode

marimo/_runtime/app/script_runner.py

@@ -1,7 +1,6 @@
 # Copyright 2026 Marimo. All rights reserved.
 from __future__ import annotations
 
-import asyncio
 from typing import TYPE_CHECKING, Any
 
 from marimo._ast.names import SETUP_CELL_NAME
@@ -32,6 +31,7 @@
 from marimo._runtime.runner.result import RunResult
 from marimo._runtime.runner.scheduler import SequentialScheduler
 from marimo._types.ids import CellId_t
+from marimo._utils.asyncio_utils import run_on_subprocess_capable_loop
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -225,7 +225,7 @@ def run(self) -> RunOutput:
                 post_execute_hooks.append(close_figures)
 
             if is_async:
-                outputs, defs = asyncio.run(
+                outputs, defs = run_on_subprocess_capable_loop(
                     self._run_asynchronous(
                         post_execute_hooks=post_execute_hooks,
                     )

marimo/_runtime/runtime.py

@@ -166,6 +166,7 @@
 from marimo._tracer import attach_trace_context, kernel_tracer
 from marimo._types.ids import CellId_t, UIElementId, VariableName
 from marimo._types.lifespan import Lifespan
+from marimo._utils.asyncio_utils import run_on_subprocess_capable_loop
 from marimo._utils.lifespans import Lifespans
 from marimo._utils.paths import normalize_path
 from marimo._utils.platform import is_pyodide
@@ -2735,7 +2736,9 @@ def launch_kernel(
             )
             if loop_factory is not None:
                 asyncio.run(coro, loop_factory=loop_factory)
-            else:
+            elif is_subprocess:
                 asyncio.run(coro)
+            else:
+                run_on_subprocess_capable_loop(coro)
 
         streams.close(use_fd_redirect)

marimo/_utils/asyncio_utils.py

@@ -7,6 +7,8 @@
   where the loop's default handler would otherwise swallow them).
 - `cancel_and_wait`: the `task.cancel(); await task` /
   `except CancelledError` dance, in one place.
+- `run_on_subprocess_capable_loop`: `asyncio.run` on a loop that can spawn
+  subprocesses on Windows.
 """
 
 from __future__ import annotations
@@ -128,6 +130,60 @@ def fire_and_forget(
     return supervised_task(coro, name=name, registry=_BACKGROUND_TASKS)
 
 
+def run_on_subprocess_capable_loop(coro: Coroutine[Any, Any, T]) -> T:
+    """`asyncio.run(coro)`, but on a `ProactorEventLoop` on Windows.
+
+    The loop is created locally instead of changing the global policy,
+    because other threads create event loops at the same time.
+    """
+    if sys.platform != "win32":
+        return asyncio.run(coro)
+    if asyncio._get_running_loop() is not None:
+        raise RuntimeError(
+            "asyncio.run() cannot be called from a running event loop"
+        )
+    if sys.version_info >= (3, 12):
+        return asyncio.run(coro, loop_factory=asyncio.ProactorEventLoop)
+    if sys.version_info >= (3, 11):
+        with asyncio.Runner(loop_factory=asyncio.ProactorEventLoop) as runner:
+            return runner.run(coro)
+    return _run_with_loop_factory_py310(coro, asyncio.ProactorEventLoop)
+
+
+def _run_with_loop_factory_py310(
+    coro: Coroutine[Any, Any, T],
+    loop_factory: Callable[[], asyncio.AbstractEventLoop],
+) -> T:
+    # Mirrors CPython 3.10's `asyncio.run`, which has no `loop_factory`.
+    loop = loop_factory()
+    try:
+        asyncio.set_event_loop(loop)
+        return loop.run_until_complete(coro)
+    finally:
+        try:
+            to_cancel = asyncio.all_tasks(loop)
+            for task in to_cancel:
+                task.cancel()
+            loop.run_until_complete(
+                asyncio.gather(*to_cancel, return_exceptions=True)
+            )
+            for task in to_cancel:
+                if not task.cancelled() and task.exception() is not None:
+                    loop.call_exception_handler(
+                        {
+                            "message": "unhandled exception during "
+                            "asyncio.run() shutdown",
+                            "exception": task.exception(),
+                            "task": task,
+                        }
+                    )
+            loop.run_until_complete(loop.shutdown_asyncgens())
+            loop.run_until_complete(loop.shutdown_default_executor())
+        finally:
+            asyncio.set_event_loop(None)
+            loop.close()
+
+
 async def cancel_and_wait(task: asyncio.Task[Any]) -> None:
     """Cancel `task` and await its completion, suppressing `CancelledError`.
 
@@ -151,5 +207,6 @@ async def cancel_and_wait(task: asyncio.Task[Any]) -> None:
     "cancel_and_wait",
     "fire_and_forget",
     "initialize_asyncio",
+    "run_on_subprocess_capable_loop",
     "supervised_task",
 ]

tests/_ast/test_app.py

@@ -631,6 +631,32 @@ def __(x: int) -> tuple[int]:
         assert defs["x"] == 0
         assert defs["y"] == 1
 
+    @staticmethod
+    @pytest.mark.skipif(
+        sys.platform != "win32", reason="Windows event loop regression"
+    )
+    def test_run_async_subprocess() -> None:
+        app = App()
+
+        @app.cell
+        async def __() -> tuple[str]:
+            import asyncio
+            import sys
+
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable,
+                "-c",
+                "print('ok')",
+                stdout=asyncio.subprocess.PIPE,
+            )
+            stdout, _ = await proc.communicate()
+            output = stdout.decode().strip()
+            return (output,)
+
+        _, defs = app.run()
+
+        assert defs["output"] == "ok"
+
     @staticmethod
     def test_run_mo_stop() -> None:
         app = App()

tests/_runtime/test_runtime.py

@@ -4675,11 +4675,11 @@ def _filter_to_error_ops(
 class TestLaunchKernelEventLoop:
     """Event-loop policy / factory selection in launch_kernel.
 
-    The kernel subprocess must run on the Windows ProactorEventLoop so
-    user code can use asyncio.create_subprocess_exec() and other APIs
-    the SelectorEventLoop does not implement. The server keeps the
-    SelectorEventLoop because ConnectionDistributor relies on
-    loop.add_reader().
+    The server keeps the SelectorEventLoop because ConnectionDistributor
+    relies on loop.add_reader(). Kernels that run user code use a
+    ProactorEventLoop on Windows so asyncio.create_subprocess_exec()
+    works: child processes via the policy / loop_factory, in-process
+    run-mode threads via run_on_subprocess_capable_loop.
 
     Each test exercises a single (platform, python-version) branch and
     skips when the current runner doesn't match it. CI runs across all
@@ -4799,16 +4799,31 @@ def test_windows_314_plus_uses_proactor_loop_factory(self, harness):
 
     @pytest.mark.skipif(
         sys.platform != "win32",
-        reason="run-mode guard is only meaningful on Windows",
+        reason="exercises the Windows run-mode branch",
     )
-    def test_run_mode_on_windows_does_not_touch_event_loop_policy(
-        self, harness
-    ):
-        # Run mode (not edit, not IPC) runs in-process on the server's
-        # loop and must NOT mutate the event loop policy — the server
-        # uses the Selector loop for ConnectionDistributor.add_reader().
+    @pytest.mark.usefixtures("harness")
+    def test_run_mode_on_windows_uses_subprocess_capable_loop(self) -> None:
+        # On 3.10/3.11 the helper bypasses the mocked asyncio.run.
+        with (
+            patch(
+                "marimo._runtime.runtime.run_on_subprocess_capable_loop",
+                side_effect=self._fake_asyncio_run,
+            ) as helper,
+            patch.object(asyncio, "set_event_loop_policy") as set_policy,
+        ):
+            self._call_launch_kernel(is_edit_mode=False)
+
+        helper.assert_called_once()
+        set_policy.assert_not_called()
+
+    @pytest.mark.skipif(
+        sys.platform == "win32",
+        reason="exercises the non-Windows run-mode branch",
+    )
+    def test_run_mode_on_non_windows_uses_plain_asyncio_run(self, harness):
         with patch.object(asyncio, "set_event_loop_policy") as set_policy:
             self._call_launch_kernel(is_edit_mode=False)
 
         set_policy.assert_not_called()
+        assert harness.call_count == 1
         assert "loop_factory" not in harness.call_args.kwargs

tests/_utils/test_asyncio_utils.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 
 import pytest
 
 from marimo._utils.asyncio_utils import (
     cancel_and_wait,
     fire_and_forget,
+    run_on_subprocess_capable_loop,
     supervised_task,
 )
 
@@ -129,3 +131,76 @@ async def fast() -> int:
     task = asyncio.create_task(fast())
     await task
     await cancel_and_wait(task)
+
+
+def test_run_on_subprocess_capable_loop_returns_and_cleans_up() -> None:
+    loop: asyncio.AbstractEventLoop | None = None
+    pending: asyncio.Task[None] | None = None
+
+    async def main() -> int:
+        nonlocal loop, pending
+        loop = asyncio.get_running_loop()
+        pending = asyncio.create_task(asyncio.sleep(3600))
+        await asyncio.sleep(0)
+        return 42
+
+    assert run_on_subprocess_capable_loop(main()) == 42
+    assert pending is not None
+    assert pending.cancelled()
+    assert loop is not None
+    assert loop.is_closed()
+
+
+def test_run_on_subprocess_capable_loop_propagates_exception() -> None:
+    async def main() -> None:
+        raise ValueError("boom")
+
+    with pytest.raises(ValueError, match="boom"):
+        run_on_subprocess_capable_loop(main())
+
+
+def test_run_on_subprocess_capable_loop_rejects_running_loop() -> None:
+    async def inner() -> None:
+        pass
+
+    async def outer() -> None:
+        coro = inner()
+        try:
+            with pytest.raises(RuntimeError):
+                run_on_subprocess_capable_loop(coro)
+        finally:
+            coro.close()
+
+    asyncio.run(outer())
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only loop")
+def test_run_on_subprocess_capable_loop_spawns_under_selector_policy() -> None:
+    original = asyncio.get_event_loop_policy()
+    selector_policy = asyncio.WindowsSelectorEventLoopPolicy()
+    asyncio.set_event_loop_policy(selector_policy)
+
+    async def main() -> tuple[bool, int | None, bytes]:
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable,
+            "-c",
+            "print('ok')",
+            stdout=asyncio.subprocess.PIPE,
+        )
+        stdout, _ = await proc.communicate()
+        is_proactor = isinstance(
+            asyncio.get_running_loop(), asyncio.ProactorEventLoop
+        )
+        return is_proactor, proc.returncode, stdout
+
+    try:
+        is_proactor, returncode, stdout = run_on_subprocess_capable_loop(
+            main()
+        )
+        assert asyncio.get_event_loop_policy() is selector_policy
+    finally:
+        asyncio.set_event_loop_policy(original)
+
+    assert is_proactor
+    assert returncode == 0
+    assert stdout.decode().strip() == "ok"