Make control loop test deterministic

tests/_runtime/test_kernel_lifecycle.py

@@ -3,13 +3,15 @@
 
 import asyncio
 import queue as _queue
+import threading
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from marimo._runtime.commands import (
     CodeCompletionCommand,
+    CommandMessage,
     ExecuteCellsCommand,
     ModelCommand,
     SetBreakpointsCommand,
@@ -22,6 +24,7 @@
     drain_stale,
     listen_messages,
     make_control_enqueuer,
+    threaded_queue_reader,
 )
 from marimo._types.ids import CellId_t, UIElementId, WidgetModelId
 
@@ -55,6 +58,29 @@ def _ui_update(
     )
 
 
+async def test_threaded_queue_reader_offloads_blocking_get(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    q: _queue.Queue[CommandMessage] = _queue.Queue()
+    command = StopKernelCommand()
+    q.put(command)
+
+    event_loop_thread = threading.current_thread()
+    reader_thread: threading.Thread | None = None
+    original_get = q.get
+
+    def tracked_get() -> CommandMessage:
+        nonlocal reader_thread
+        reader_thread = threading.current_thread()
+        return original_get()
+
+    monkeypatch.setattr(q, "get", tracked_get)
+
+    assert await threaded_queue_reader(q) is command
+    assert reader_thread is not None
+    assert reader_thread is not event_loop_thread
+
+
 async def test_listen_messages_exits_on_stop_command(
     kernel: Any,
     control: asyncio.Queue[Any],