fix(boxlite): keep private-loop shutdown retryable (#6283)

* fix(boxlite): keep private-loop shutdown retryable

* test(boxlite): cover private-loop shutdown retry

* docs(sandbox): pin BoxLite loop shutdown ownership

* fix(ci): format BoxLite shutdown changes and trim guidance

* fix(boxlite): drain acquisitions before shutdown teardown

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -675,6 +675,10 @@ report `Exit Code: 124`; unsuccessful health checks cannot reclaim a warm sandbo
 Health probes tolerate login-shell output around the `ok` line, and failures log
 the sandbox ID and probe output before replacing the sandbox.
 
+BoxLite shutdown rejects late VM registration and keeps its SDK loop open while
+in-flight acquisitions finish. If they cannot drain within five seconds,
+shutdown fails with resources still owned and can be retried.
+
 #### MCP Server
 
 In the chat UI, enable **Token Usage → Debug** to inspect generic/MCP tool calls.

backend/packages/harness/deerflow/community/boxlite/provider.py

@@ -54,6 +54,10 @@ _VIRTUAL_DIRS = (
 )
 
 
+class _BoxliteLoopShutdownTimeout(RuntimeError):
+    """The private BoxLite event-loop thread outlived its shutdown join."""
+
+
 class SandboxIdentityCollisionError(RuntimeError):
     """A deterministic ID is already tracked for a different user/thread."""
 
@@ -129,15 +133,18 @@ class _EventLoopThread:
             raise
 
     def close(self) -> None:
-        if self._loop is None:
+        loop = self._loop
+        if loop is None or loop.is_closed():
             return
-        self._loop.call_soon_threadsafe(self._loop.stop)
-        wake = getattr(self._loop, "_write_to_self", None)
+        loop.call_soon_threadsafe(loop.stop)
+        wake = getattr(loop, "_write_to_self", None)
         if wake is not None:
             wake()
         self._thread.join(timeout=5)
-        if not self._loop.is_running():
-            self._loop.close()
+        if self._thread.is_alive():
+            raise _BoxliteLoopShutdownTimeout("BoxLite event-loop thread is still running after stop timeout")
+        if not loop.is_running():
+            loop.close()
 
 
 class _SyncBoxAdapter:
@@ -204,6 +211,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
 
     def __init__(self) -> None:
         self._lock = threading.Lock()
+        self._acquire_condition = threading.Condition(self._lock)
+        self._pending_acquires = 0
         self._boxes: dict[str, BoxliteBox] = {}
         self._thread_boxes: dict[tuple[str, str], str] = {}
         self._warm_pool: dict[str, tuple[BoxliteBox, float]] = {}
@@ -214,6 +223,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
         self._idle_checker_stop = threading.Event()
         self._idle_checker_thread: threading.Thread | None = None
         self._shutdown_called = False
+        self._loop_cleanup_pending = False
         self._config = self._load_config()
         self._loop = _EventLoopThread()
         atexit.register(self.shutdown)
@@ -372,18 +382,33 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
     # ── Acquire / release ────────────────────────────────────────────────
 
     def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
-        if thread_id is None:
-            sandbox_id = str(uuid.uuid4())[:8]
-            box = self._create_box(sandbox_id)
-            with self._lock:
-                self._boxes[box.id] = box
-                self._active_box_identity[box.id] = None
-            return box.id
+        with self._acquire_condition:
+            if self._shutdown_called:
+                raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
+            self._pending_acquires += 1
 
-        key = self._thread_key(thread_id, user_id)
-        sandbox_id = self._sandbox_id(thread_id, user_id)
-        with self._acquire_serializer.hold(sandbox_id):
-            return self._acquire_scope_locked(key, sandbox_id)
+        try:
+            if thread_id is None:
+                sandbox_id = str(uuid.uuid4())[:8]
+                box = self._create_box(sandbox_id)
+                with self._lock:
+                    shutting_down = self._shutdown_called
+                    if not shutting_down:
+                        self._boxes[box.id] = box
+                        self._active_box_identity[box.id] = None
+                if shutting_down:
+                    box.close()
+                    raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
+                return box.id
+
+            key = self._thread_key(thread_id, user_id)
+            sandbox_id = self._sandbox_id(thread_id, user_id)
+            with self._acquire_serializer.hold(sandbox_id):
+                return self._acquire_scope_locked(key, sandbox_id)
+        finally:
+            with self._acquire_condition:
+                self._pending_acquires -= 1
+                self._acquire_condition.notify_all()
 
     async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
         """Acquire without blocking the event loop.
@@ -399,6 +424,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
 
     def _acquire_scope_locked(self, key: tuple[str, str], sandbox_id: str) -> str:
         with self._lock:
+            if self._shutdown_called:
+                raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
             existing = self._thread_boxes.get(key)
             if existing is not None and existing in self._boxes:
                 return existing
@@ -416,20 +443,27 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
         reclaimed = self._reclaim_warm_pool(sandbox_id, key)
         if reclaimed is not None:
             with self._lock:
+                if self._shutdown_called:
+                    raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
                 self._thread_boxes[key] = reclaimed
             return reclaimed
 
         box = self._create_box(sandbox_id)
         conflict: tuple[str, str] | None | object = _NO_ACTIVE_IDENTITY
         with self._lock:
-            if box.id in self._boxes:
-                conflict = self._active_box_identity.get(box.id)
-                if conflict == key:
+            shutting_down = self._shutdown_called
+            if not shutting_down:
+                if box.id in self._boxes:
+                    conflict = self._active_box_identity.get(box.id)
+                    if conflict == key:
+                        self._thread_boxes[key] = box.id
+                else:
+                    self._boxes[box.id] = box
+                    self._active_box_identity[box.id] = key
                     self._thread_boxes[key] = box.id
-            else:
-                self._boxes[box.id] = box
-                self._active_box_identity[box.id] = key
-                self._thread_boxes[key] = box.id
+        if shutting_down:
+            box.close()
+            raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
         if conflict is not _NO_ACTIVE_IDENTITY:
             box.close()
             if conflict != key:
@@ -530,6 +564,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             # health-check round trip, but never return an adapter that this
             # process already knows is closed.
             with self._lock:
+                if self._shutdown_called:
+                    raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
                 current = self._warm_pool.get(sandbox_id)
                 stored_key = self._warm_pool_identity.get(sandbox_id)
                 if stored_key is not None and stored_key != expected_key:
@@ -605,6 +641,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
 
         # Promote from warm pool to active
         with self._lock:
+            if self._shutdown_called:
+                raise RuntimeError("BoxLite provider is shutting down; cannot acquire sandbox")
             current = self._warm_pool.get(sandbox_id)
             stored_key = self._warm_pool_identity.get(sandbox_id)
             if stored_key is not None and stored_key != expected_key:
@@ -647,14 +685,37 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             self._thread_boxes.clear()
             self._acquire_serializer.close()
 
+    def _close_loop_for_shutdown(self) -> None:
+        try:
+            self._loop.close()
+        except _BoxliteLoopShutdownTimeout:
+            with self._lock:
+                self._loop_cleanup_pending = True
+            raise
+        with self._lock:
+            self._loop_cleanup_pending = False
+
     def shutdown(self) -> None:
         with self._lock:
             if self._shutdown_called:
-                return
-            self._shutdown_called = True
+                retry_loop_cleanup = self._loop_cleanup_pending
+                if not retry_loop_cleanup:
+                    return
+            else:
+                self._shutdown_called = True
+                retry_loop_cleanup = False
+
+        if retry_loop_cleanup:
+            self._close_loop_for_shutdown()
+            return
 
         try:
             self._stop_idle_checker()
+            # In-flight acquires own late-created boxes until they can reject
+            # registration and stop them on the still-running private loop.
+            with self._acquire_condition:
+                if not self._acquire_condition.wait_for(lambda: self._pending_acquires == 0, timeout=5):
+                    raise RuntimeError("BoxLite in-flight acquisitions did not finish before shutdown timeout")
         except Exception:
             with self._lock:
                 self._shutdown_called = False
@@ -676,4 +737,4 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
                 box.close()
             except Exception as e:  # pragma: no cover - defensive
                 logger.warning("Error closing BoxLite box %s during shutdown: %s", box.id, e)
-        self._loop.close()
+        self._close_loop_for_shutdown()

backend/packages/harness/deerflow/sandbox/AGENTS.md

@@ -89,14 +89,14 @@ cancellation before propagating it.
     - **An absent lease means the same thing on both paths, and reconciliation must say so too.** The `LAPSED` rule above only covers an owner renewing its *own* lease; on its own it does not make state loss safe, because reconciliation reads the same absent key as "orphan, adopt". After a Redis flush (restart without persistence, or eviction under `maxmemory`) every owner is alive and merely pre-renewal-tick, so whichever instance reconciles first would adopt every live container, each real owner's next renewal would report `LOST`, and it would drop a sandbox mid-turn for the adopter to idle-destroy — #4206 through the back door. `_adoptable_after_grace` closes it: an untracked container must be seen unowned (`owner()`, a read-only peek — the atomic `claim()` is still what actually gates adoption) across a full lease TTL before it can be adopted, tracked per container in `_unowned_since`. That rebuilds the delay the flush erased — a live owner republishes within one renewal interval, shorter than the TTL by construction (`ttl_multiplier >= 2`) — while a genuinely crashed owner never republishes, so its containers are still adopted one grace later rather than leaking. A republished lease **resets** the grace; a pausing-only timer would still expire over a live owner's lease. The grace is skipped when `supports_cross_process` is `False`: no peer can hold a lease such a store would show us, so single-instance deployments keep instant orphan cleanup, and a grace could not help a multi-worker gateway on `memory` anyway (peers are invisible to each other's leases with or without it).
     - **The `memory` store is single-instance only** and says so via `supports_cross_process = False`; the provider logs a warning at startup when the configured store cannot see peers. A multi-worker gateway on `memory` has no cross-process coordination at all — same contract as `stream_bridge`'s memory backend. This is why the redis inference matters: it reads `app_config.stream_bridge` **and** the env var, in the same order the bridge's own resolver does, so any deployment already pointing the bridge at Redis (i.e. every multi-instance one) gets a redis ownership store without extra config.
     - `get()` stays a pure in-memory lookup and must never call the store (that is blocking filesystem/network IO on the event loop); anchored by `tests/blocking_io/test_aio_sandbox_get.py`, which injects a deliberately-blocking probe store so the anchor keeps its teeth regardless of the configured backend. Tests: `tests/test_sandbox_ownership_store.py` (store contract, defined once for **both** backends — the redis tier is `@pytest.mark.integration`, uses `DEER_FLOW_TEST_REDIS_URL` when set, and otherwise self-skips without a reachable Redis. Backend CI provisions Redis, so the merge gate executes the real Lua tier; there is no fake-redis tier because a fake would not execute the Lua exclusions) and `tests/test_sandbox_orphan_reconciliation.py` (provider behaviour, two providers sharing one store).
-- `BoxliteProvider` (`packages/harness/deerflow/community/boxlite/`) - optional BoxLite micro-VM isolation. Loop-affine handles live on one private daemon-thread asyncio loop; sync `Sandbox` calls marshal through `run_coroutine_threadsafe`.
-  Boxes use deterministic `user_id:thread_id` IDs, park in a same-owner warm pool, and are reclaimed only by that owner. Health-check timeouts reach both BoxLite `exec(timeout=...)` and private-loop `.result(timeout)`; a bridge timeout must cancel its coroutine so work cannot continue or retain acquire ownership in the background.
-  `sandbox.replicas` caps active + warm VMs per gateway process; if capacity is exhausted, only warm-pool VMs are evicted. `sandbox.idle_timeout` stops idle warm VMs after the configured seconds. `reset()` is intentionally a lightweight registry clear for `reset_sandbox_provider()` and does not close boxes, stop the idle reaper, or close the private loop; full teardown remains `shutdown()`.
+- `BoxliteProvider` (`community/boxlite/`): optional micro-VMs. One private daemon asyncio loop owns SDK handles; sync calls use `run_coroutine_threadsafe`. Deterministic `user_id:thread_id` boxes permit owner-only warm reuse. Health checks bound SDK `exec(timeout=...)` and bridge `.result(timeout)`; bridge timeouts cancel coroutines to release work/acquire ownership.
+  `sandbox.replicas` caps active + warm VMs per process; evict only warm VMs. `sandbox.idle_timeout` expires warm VMs. `reset()` retains VMs, reaper, and loop; `shutdown()` performs full teardown.
+  Fence registration under `_lock`; stop late VMs before loop close. Shutdown drains acquires for 5s; timeout retains resources for retry. `boxlite-loop` join must confirm exit before loop close; a join timeout keeps admission fenced and retries only loop cleanup.
 - `TenkiSandboxProvider` (`community/tenki/`): cloud microVMs. Optional `tenki` SDK (`deerflow-harness[tenki]`, module `tenki_sandbox`) loads via `_import_client` only on provider selection; calls are synchronous, without an event-loop bridge. Native `sandbox.fs` (`read_text`/`read_stream`/`write_stream`/`mkdir`/`stat`) streams binary-safe files without shell/base64. Only search (`list_dir`/`glob`/`grep`) uses busybox-portable `find`/`grep` and shared `deerflow.sandbox.search` parsing, like E2B. The unprivileged `tenki` user needs `/mnt/user-data` remapped to writable HOME (`_resolve_path`), with a best-effort bootstrap `sudo` symlink. Deterministic names and warm-pool keys use `sha256(user_id:thread_id)[:16]` (64-bit, matching E2B), with no full-seed fallback. Release parks VMs in-process; only the same user/thread reclaims them after a health check. Named terminal SDK errors and builtin `ConnectionError`/`BrokenPipeError`/`EOFError` evict via `_invalidate_sandbox`. Cross-process orphan reconciliation remains a follow-up.
   Health probes require an exact `ok` line and reject `Error:`/`Exit Code:` diagnostic lines; other login-shell stdout/stderr is tolerated. Failures log sandbox id and output before eviction.
 
 
-**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. Shutdown must not tear down provider resources while the shared idle-checker thread is still alive: `_stop_idle_checker()` uses a bounded join and raises if that bound expires; providers reset their shutdown guard on that failure so a later shutdown can retry instead of silently becoming a no-op.
+**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. `_stop_idle_checker()` must confirm thread exit before provider teardown. A bounded-join timeout raises; reset the provider shutdown guard so shutdown can retry.
 
 **Virtual Path System**:
 - Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills`

backend/tests/test_boxlite_provider.py

@@ -1132,6 +1132,185 @@ def test_replica_enforcement_counts_active_and_warm(monkeypatch):
 # ── Task 8: Shutdown and reset including warm pool ────────────────────
 
 
+@pytest.mark.parametrize("thread_id", [None, "thread-late"])
+def test_shutdown_rejects_and_closes_box_created_by_inflight_acquire(monkeypatch, thread_id):
+    monkeypatch.setattr("deerflow.community.boxlite.provider.get_app_config", lambda: _stub_config({"idle_timeout": 0}))
+    monkeypatch.setattr("deerflow.community.boxlite.provider._import_simplebox", lambda: _FakeBox)
+    provider = BoxliteProvider()
+    created = threading.Event()
+    resume_create = threading.Event()
+    shutdown_started = threading.Event()
+    boxes = []
+    acquired = []
+    acquire_errors = []
+    shutdown_errors = []
+    closed_before_loop = []
+    original_create = provider._create_box
+    original_stop = provider._stop_idle_checker
+    original_close_loop = provider._loop.close
+
+    def paused_create(sandbox_id):
+        box = original_create(sandbox_id)
+        boxes.append(box)
+        created.set()
+        assert resume_create.wait(10)
+        return box
+
+    def signal_shutdown():
+        original_stop()
+        shutdown_started.set()
+
+    def close_loop():
+        closed_before_loop.append(boxes[0]._box._stopped)
+        original_close_loop()
+
+    def acquire():
+        try:
+            acquired.append(provider.acquire(thread_id, user_id="u1"))
+        except Exception as error:
+            acquire_errors.append(error)
+
+    def shutdown():
+        try:
+            provider.shutdown()
+        except Exception as error:
+            shutdown_errors.append(error)
+
+    monkeypatch.setattr(provider, "_create_box", paused_create)
+    monkeypatch.setattr(provider, "_stop_idle_checker", signal_shutdown)
+    monkeypatch.setattr(provider._loop, "close", close_loop)
+    acquirer = threading.Thread(target=acquire)
+    closer = threading.Thread(target=shutdown)
+    try:
+        acquirer.start()
+        assert created.wait(5)
+        closer.start()
+        assert shutdown_started.wait(5)
+        resume_create.set()
+        acquirer.join(5)
+        closer.join(5)
+        assert not acquirer.is_alive()
+        assert not closer.is_alive()
+        assert acquired == []
+        assert len(acquire_errors) == 1
+        assert isinstance(acquire_errors[0], RuntimeError)
+        assert "shutting down" in str(acquire_errors[0])
+        assert shutdown_errors == []
+        assert boxes[0].is_closed
+        assert boxes[0]._box._stopped
+        assert closed_before_loop == [True]
+        assert provider._boxes == {}
+        assert provider._thread_boxes == {}
+        assert provider._active_box_identity == {}
+        assert provider._warm_pool == {}
+        assert provider._loop._loop.is_closed()
+        provider.shutdown()
+        assert closed_before_loop == [True]
+    finally:
+        resume_create.set()
+        acquirer.join(5)
+        if closer.ident is not None:
+            closer.join(5)
+        provider.shutdown()
+
+
+def test_shutdown_acquire_drain_timeout_keeps_resources_owned_for_retry(monkeypatch):
+    monkeypatch.setattr("deerflow.community.boxlite.provider.get_app_config", lambda: _stub_config({"idle_timeout": 0}))
+    monkeypatch.setattr("deerflow.community.boxlite.provider._import_simplebox", lambda: _FakeBox)
+    provider = BoxliteProvider()
+    existing_id = provider.acquire("thread-existing", user_id="u1")
+    existing_box = provider.get(existing_id)
+    created = threading.Event()
+    resume_create = threading.Event()
+    acquired = []
+    errors = []
+    original_create = provider._create_box
+
+    def paused_create(sandbox_id):
+        box = original_create(sandbox_id)
+        created.set()
+        assert resume_create.wait(10)
+        return box
+
+    def acquire():
+        try:
+            acquired.append(provider.acquire("thread-late", user_id="u1"))
+        except Exception as error:
+            errors.append(error)
+
+    monkeypatch.setattr(provider, "_create_box", paused_create)
+    acquirer = threading.Thread(target=acquire)
+    try:
+        acquirer.start()
+        assert created.wait(5)
+        with monkeypatch.context() as timeout:
+            timeout.setattr(threading.Condition, "wait_for", lambda self, predicate, timeout=None: False)
+            with pytest.raises(RuntimeError, match="in-flight acquisitions"):
+                provider.shutdown()
+        assert not provider._shutdown_called
+        assert provider.get(existing_id) is existing_box
+        assert not existing_box.is_closed
+        assert provider._loop._loop.is_running()
+        resume_create.set()
+        acquirer.join(5)
+        assert not acquirer.is_alive()
+        assert errors == []
+        assert len(acquired) == 1
+        late_box = provider.get(acquired[0])
+        assert late_box is not None
+        provider.shutdown()
+        assert existing_box._box._stopped
+        assert late_box._box._stopped
+        assert provider._boxes == {}
+        assert provider._loop._loop.is_closed()
+    finally:
+        resume_create.set()
+        acquirer.join(5)
+        provider.shutdown()
+
+
+def test_shutdown_retries_only_private_loop_cleanup_after_join_timeout(monkeypatch):
+    from deerflow.community.boxlite.provider import _BoxliteLoopShutdownTimeout
+
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"idle_timeout": 0}),
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.close()
+
+    class RetryLoop:
+        def __init__(self) -> None:
+            self.close_calls = 0
+
+        def close(self) -> None:
+            self.close_calls += 1
+            if self.close_calls == 1:
+                raise _BoxliteLoopShutdownTimeout("BoxLite event-loop thread is still running after stop timeout")
+
+    retry_loop = RetryLoop()
+    provider._loop = retry_loop
+
+    with pytest.raises(_BoxliteLoopShutdownTimeout, match="still running"):
+        provider.shutdown()
+
+    assert provider._shutdown_called is True
+    assert provider._loop_cleanup_pending is True
+    assert retry_loop.close_calls == 1
+
+    with pytest.raises(RuntimeError, match="shutting down"):
+        provider.acquire()
+
+    provider.shutdown()
+
+    assert provider._loop_cleanup_pending is False
+    assert retry_loop.close_calls == 2
+
+    provider.shutdown()
+    assert retry_loop.close_calls == 2
+
+
 def test_shutdown_stops_idle_reaper_and_destroys_all_boxes(monkeypatch):
     """shutdown stops the idle reaper thread and destroys all active + warm boxes."""
     monkeypatch.setattr(
@@ -1534,6 +1713,67 @@ def test_grep_single_file_path_with_matching_glob(tmp_path, monkeypatch) -> None
     assert box.grep(str(target), "needle", glob="*.md") == ([], False)
 
 
+def test_event_loop_thread_close_rejects_live_thread_after_timeout() -> None:
+    from deerflow.community.boxlite.provider import (
+        _BoxliteLoopShutdownTimeout,
+        _EventLoopThread,
+    )
+
+    class FakeLoop:
+        def __init__(self) -> None:
+            self.running = True
+            self.closed = False
+            self.stop_calls = 0
+            self.close_calls = 0
+
+        def is_closed(self) -> bool:
+            return self.closed
+
+        def stop(self) -> None:
+            self.stop_calls += 1
+
+        def call_soon_threadsafe(self, callback) -> None:
+            callback()
+
+        def _write_to_self(self) -> None:
+            return None
+
+        def is_running(self) -> bool:
+            return self.running
+
+        def close(self) -> None:
+            self.close_calls += 1
+            self.closed = True
+
+    class JoinControlledThread:
+        def __init__(self) -> None:
+            self.alive = True
+            self.join_timeout: float | None = None
+
+        def join(self, timeout: float | None = None) -> None:
+            self.join_timeout = timeout
+
+        def is_alive(self) -> bool:
+            return self.alive
+
+    loop_thread = _EventLoopThread.__new__(_EventLoopThread)
+    loop_thread._loop = FakeLoop()
+    loop_thread._thread = JoinControlledThread()
+
+    with pytest.raises(_BoxliteLoopShutdownTimeout, match="still running"):
+        loop_thread.close()
+
+    assert loop_thread._thread.join_timeout == 5
+    assert loop_thread._loop.close_calls == 0
+
+    loop_thread._thread.alive = False
+    loop_thread._loop.running = False
+    loop_thread.close()
+
+    assert loop_thread._loop.closed is True
+    assert loop_thread._loop.close_calls == 1
+
+
 def test_event_loop_thread_timeout_cancels_submitted_coroutine() -> None:
     from concurrent.futures import TimeoutError as FutureTimeoutError