fix(pyodide): address code review findings in the runtime shims

- Schedule threading.Timer on the host event loop instead of running it
  inline, which deadlocked RPMController (it re-arms under a held Lock).
- Route exceptions from inline Thread targets to threading.excepthook, as
  a real thread would, instead of raising out of start().
- Map JsException while streaming a response body to httpx.ReadError so
  SDK retry handling sees a transport error.
- Read sync request bodies before handing off to the async transport.
- On emscripten, event bus shutdown() now shuts down the executor and
  clears handlers while still leaving the host loop running.
- Turn the entropy stand-ins off once the import finishes, so references
  captured during import return real entropy afterwards.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01A3tq2y6MGDFxCWvmt1BUsj

lib/crewai-pyodide/src/crewai_pyodide/_entropy.py

@@ -68,9 +68,12 @@ def _without_entropy(module_name: str) -> Iterator[None]:
     stay visible to :func:`audit_entropy`.
     """
     real_uuid4, real_urandom = uuid.uuid4, os.urandom
+    # Names bound with ``from os import urandom`` keep these stand-ins after
+    # the import, so they must hand out real entropy from then on.
+    active = True
 
     def from_module() -> bool:
-        return sys._getframe(2).f_globals.get("__name__") == module_name
+        return active and sys._getframe(2).f_globals.get("__name__") == module_name
 
     def uuid4() -> uuid.UUID:
         return _deterministic_uuid4() if from_module() else real_uuid4()
@@ -82,6 +85,7 @@ def _without_entropy(module_name: str) -> Iterator[None]:
     try:
         yield
     finally:
+        active = False
         uuid.uuid4, os.urandom = real_uuid4, real_urandom
 
 

lib/crewai-pyodide/src/crewai_pyodide/_threads.py

@@ -8,9 +8,11 @@ without parallelism.
 
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Callable
 import concurrent.futures
 import concurrent.futures.thread
+import sys
 import threading
 from typing import Any, TypeVar
 
@@ -42,8 +44,31 @@ class InlineExecutor(concurrent.futures.Executor):
 
 
 def _start_inline(self: threading.Thread) -> None:
-    """``Thread.start`` replacement: run the target to completion now."""
-    self.run()
+    """``Thread.start`` replacement: run the target to completion now.
+
+    As in a real thread, an exception from the target goes to
+    ``threading.excepthook`` instead of propagating to the caller.
+    """
+    try:
+        self.run()
+    except SystemExit:
+        pass
+    except BaseException:
+        threading.excepthook(threading.ExceptHookArgs((*sys.exc_info(), self)))
+
+
+def _start_timer(self: threading.Timer) -> None:
+    """``Timer.start`` replacement: schedule the call on the event loop.
+
+    Running a timer inline would block the caller for its whole interval.
+    """
+
+    def fire() -> None:
+        if not self.finished.is_set():
+            self.function(*self.args, **self.kwargs)
+        self.finished.set()
+
+    asyncio.get_event_loop().call_later(self.interval, fire)
 
 
 def _join_inline(self: threading.Thread, timeout: float | None = None) -> None:
@@ -56,3 +81,4 @@ def install_inline_threads() -> None:
     concurrent.futures.thread.ThreadPoolExecutor = InlineExecutor  # type: ignore[misc,assignment]
     threading.Thread.start = _start_inline  # type: ignore[method-assign]
     threading.Thread.join = _join_inline  # type: ignore[method-assign]
+    threading.Timer.start = _start_timer  # type: ignore[method-assign,assignment]

lib/crewai-pyodide/src/crewai_pyodide/http.py

@@ -31,10 +31,16 @@ class _FetchByteStream(httpx.AsyncByteStream):
         self._reader = body.getReader() if body is not None else None
 
     async def __aiter__(self) -> AsyncIterator[bytes]:
+        from pyodide.ffi import JsException  # type: ignore[import-not-found]
+
         if self._reader is None:
             return
         while True:
-            chunk = await self._reader.read()
+            try:
+                chunk = await self._reader.read()
+            except JsException as e:
+                # The connection dropped mid-body; let httpx and SDKs retry.
+                raise httpx.ReadError(str(e)) from e
             if chunk.done:
                 return
             yield chunk.value.to_bytes()
@@ -56,7 +62,7 @@ class FetchAsyncTransport(httpx.AsyncBaseTransport):
         pass
 
     async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
-        from pyodide.ffi import JsException, to_js  # type: ignore[import-not-found]
+        from pyodide.ffi import JsException, to_js
 
         js: Any = importlib.import_module("js")  # Pyodide's bridge to JavaScript
 
@@ -108,6 +114,9 @@ class FetchTransport(httpx.BaseTransport):
 
             raise RuntimeError(JSPI_REQUIRED_MESSAGE)
 
+        # Read a streamed sync body here: the async transport can only await it.
+        request.read()
+
         async def send() -> httpx.Response:
             response = await self._async.handle_async_request(request)
             content = await response.aread()

lib/crewai-pyodide/tests/test_http.py

@@ -144,3 +144,23 @@ def test_sync_requests_without_jspi_explain_the_requirement(
     with httpx.Client(transport=FetchTransport()) as client:
         with pytest.raises(RuntimeError, match="JSPI"):
             client.get("https://api.example.com/")
+
+
+@pytest.mark.asyncio
+async def test_body_failures_raise_read_error(fake_js: SimpleNamespace) -> None:
+    async def fail() -> SimpleNamespace:
+        raise _JsException("TypeError: network error")
+
+    async with httpx.AsyncClient(transport=FetchAsyncTransport()) as client:
+        async with client.stream("GET", "https://api.example.com/sse") as response:
+            fake_js.reader.read = fail
+            with pytest.raises(httpx.ReadError, match="network error"):
+                await response.aread()
+
+
+def test_sync_requests_send_streamed_bodies(fake_js: SimpleNamespace) -> None:
+    with httpx.Client(transport=FetchTransport()) as client:
+        client.post("https://api.example.com/", content=iter([b"a", b"b"]))
+
+    ((_, init),) = fake_js.requests
+    assert init["body"] == b"ab"

lib/crewai-pyodide/tests/test_threads.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import pytest
 
-from crewai_pyodide._threads import InlineExecutor
+from crewai_pyodide._threads import InlineExecutor, install_inline_threads
 
 
 def test_submit_runs_immediately_and_returns_a_done_future() -> None:
@@ -30,3 +30,57 @@ def test_works_as_a_context_manager_and_refuses_work_after_shutdown() -> None:
 
     with pytest.raises(RuntimeError):
         executor.submit(sum, [1])
+
+
+@pytest.fixture
+def inline_threads(monkeypatch: pytest.MonkeyPatch) -> None:
+    import threading
+
+    monkeypatch.setattr(threading.Thread, "start", threading.Thread.start)
+    monkeypatch.setattr(threading.Thread, "join", threading.Thread.join)
+    monkeypatch.setattr(threading.Timer, "start", threading.Timer.start, raising=False)
+    import concurrent.futures
+
+    monkeypatch.setattr(
+        concurrent.futures, "ThreadPoolExecutor", concurrent.futures.ThreadPoolExecutor
+    )
+    monkeypatch.setattr(
+        concurrent.futures.thread,
+        "ThreadPoolExecutor",
+        concurrent.futures.thread.ThreadPoolExecutor,
+    )
+    install_inline_threads()
+
+
+def test_thread_errors_go_to_excepthook_not_the_caller(
+    inline_threads: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    import threading
+
+    hooked: list[type[BaseException]] = []
+    monkeypatch.setattr(threading, "excepthook", lambda args: hooked.append(args.exc_type))
+
+    def boom() -> None:
+        raise ValueError("nope")
+
+    threading.Thread(target=boom).start()
+
+    assert hooked == [ValueError]
+
+
+@pytest.mark.asyncio
+async def test_timers_fire_later_without_blocking(inline_threads: None) -> None:
+    import asyncio
+    import threading
+
+    fired = asyncio.Event()
+    threading.Timer(0.01, fired.set).start()
+    assert not fired.is_set()
+    await asyncio.wait_for(fired.wait(), timeout=5)
+
+    cancelled: list[int] = []
+    timer = threading.Timer(0.01, cancelled.append, args=(1,))
+    timer.start()
+    timer.cancel()
+    await asyncio.sleep(0.05)
+    assert cancelled == []

lib/crewai/src/crewai/events/event_bus.py

@@ -917,10 +917,14 @@ class CrewAIEventsBus:
                 return
             loop = getattr(self, "_loop", None)
 
-        if loop is None or loop.is_closed() or self._loop_thread is None:
+        if loop is None or loop.is_closed():
             return
 
-        if wait:
+        if self._loop_thread is None:
+            # The loop belongs to the host runtime (e.g. Pyodide): leave it
+            # running, but still release the executor and handlers below.
+            pass
+        elif wait:
 
             async def _wait_for_all_tasks() -> None:
                 tasks = {
@@ -945,9 +949,10 @@ class CrewAIEventsBus:
 
             loop.call_soon_threadsafe(_cancel_tasks)
 
-        loop.call_soon_threadsafe(loop.stop)
-        self._loop_thread.join()
-        loop.close()
+        if self._loop_thread is not None:
+            loop.call_soon_threadsafe(loop.stop)
+            self._loop_thread.join()
+            loop.close()
         self._sync_executor.shutdown(wait=wait)
 
         with self._rwlock.w_locked():

lib/crewai/tests/events/test_event_bus_wasm.py

@@ -43,9 +43,14 @@ async def test_async_handlers_run_on_the_host_event_loop(
 async def test_shutdown_leaves_the_host_event_loop_running(
     wasm_bus: CrewAIEventsBus,
 ) -> None:
+    @wasm_bus.on(_PingEvent)
+    def on_ping(source: object, event: _PingEvent) -> None:
+        pass
+
     wasm_bus.emit(object(), _PingEvent())
 
     wasm_bus.shutdown(wait=False)
 
     assert asyncio.get_running_loop().is_running()
     assert not asyncio.get_running_loop().is_closed()
+    assert not wasm_bus._sync_handlers