From 7cb757bd70978338d52dce1399745d459c804741 Mon Sep 17 00:00:00 2001
From: Trevor Manz <trevor.j.manz@gmail.com>
Date: Wed, 22 Jul 2026 17:05:37 -0500
Subject: [PATCH] Make control loop test deterministic

Replace the timing-based synthetic test with a deterministic assertion
that the production queue reader offloads blocking work from the
event-loop thread.

Closes MO-6895
---
 tests/_runtime/test_control_loop.py     | 49 -------------------------
 tests/_runtime/test_kernel_lifecycle.py | 26 +++++++++++++
 2 files changed, 26 insertions(+), 49 deletions(-)
 delete mode 100644 tests/_runtime/test_control_loop.py

diff --git a/tests/_runtime/test_control_loop.py b/tests/_runtime/test_control_loop.py
deleted file mode 100644
index 2a886bc96f1..00000000000
--- a/tests/_runtime/test_control_loop.py
+++ /dev/null
@@ -1,49 +0,0 @@
-# Copyright 2026 Marimo. All rights reserved.
-"""Tests for the control loop's event loop behavior.
-
-The kernel control loop polls a synchronous queue for messages.
-These tests verify that the polling strategy does not starve
-background asyncio tasks.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import queue
-import time
-
-import pytest
-
-
-@pytest.mark.asyncio
-async def test_executor_queue_get_does_not_starve_background_tasks():
-    """run_in_executor keeps the event loop free for background tasks.
-
-    This is the pattern used by the fixed control loop: the blocking
-    queue.get() runs in a thread, so the event loop can service other
-    tasks while waiting.
-    """
-    q: queue.Queue[None] = queue.Queue()
-    loop = asyncio.get_running_loop()
-    count = 0
-
-    async def background():
-        nonlocal count
-        end = time.time() + 0.5
-        while time.time() < end:
-            count += 1
-            await asyncio.sleep(0.01)
-
-    task = asyncio.create_task(background())
-
-    # Simulate the NEW control loop: run_in_executor
-    end = time.time() + 0.5
-    while time.time() < end:
-        try:
-            await loop.run_in_executor(None, lambda: q.get(timeout=0.1))
-        except queue.Empty:
-            pass
-
-    await task
-    # With run_in_executor, the task runs freely: ~45 iterations in 0.5s.
-    assert count > 30
diff --git a/tests/_runtime/test_kernel_lifecycle.py b/tests/_runtime/test_kernel_lifecycle.py
index 1c11233a6a8..9b488f4c93b 100644
--- a/tests/_runtime/test_kernel_lifecycle.py
+++ b/tests/_runtime/test_kernel_lifecycle.py
@@ -3,6 +3,7 @@
 
 import asyncio
 import queue as _queue
+import threading
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
@@ -10,6 +11,7 @@
 
 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],