diff --git a/README.md b/README.md index 41645e757d..85c5ffad66 100644 --- a/README.md +++ b/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. diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py index 746cd71f54..ab35daa06e 100644 --- a/backend/packages/harness/deerflow/community/boxlite/provider.py +++ b/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() diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 2b6ebcd193..98151ae52b 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/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` diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index 138fbf56ef..aebb1e5f89 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/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