diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py index 734928fac..47c398c3e 100644 --- a/backend/packages/harness/deerflow/community/boxlite/provider.py +++ b/backend/packages/harness/deerflow/community/boxlite/provider.py @@ -115,7 +115,18 @@ class _EventLoopThread: def run(self, coro: Awaitable[T], *, timeout: float | None = None) -> T: if self._loop is None: raise RuntimeError("BoxLite event loop is not ready") - return asyncio.run_coroutine_threadsafe(coro, self._loop).result(timeout) + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + try: + return future.result(timeout) + except TimeoutError: + # If the bridge wait itself timed out, stop the loop-affine + # operation instead of letting it keep mutating the sandbox after + # the synchronous caller has already observed a timeout. A + # coroutine that completed by raising its own TimeoutError is + # already done and must not be reclassified here. + if not future.done(): + future.cancel() + raise def close(self) -> None: if self._loop is None: diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 3d51f9899..71748f9cf 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -90,8 +90,8 @@ to respond. - **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/`) - BoxLite micro-VM isolation. The `boxlite` runtime is optional (`deerflow-harness[boxlite]`) and lazy-imported only when this provider is selected. The provider owns one private asyncio event loop on a daemon thread because BoxLite handles are loop-affine; sync `Sandbox` calls marshal onto that loop with `run_coroutine_threadsafe`. - Boxes are named deterministically from `user_id:thread_id`, released into an in-process warm pool after each agent turn, and reclaimed only by the same user/thread. Warm-pool health checks use a short explicit timeout and forward that timeout through both BoxLite `exec(timeout=...)` and the private-loop `.result(timeout)` bridge so a hung VM cannot pin the per-thread acquire lock indefinitely. +- `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()`. - `TenkiSandboxProvider` (`packages/harness/deerflow/community/tenki/`) - Tenki cloud microVM isolation. The `tenki` SDK is optional (`deerflow-harness[tenki]`, importable as `tenki_sandbox`) and lazy-imported (`_import_client`) only when this provider is selected. Unlike Boxlite, the SDK is synchronous, so the adapter calls it directly with no event-loop bridge. File transport uses Tenki's native `sandbox.fs` API (`read_text`/`read_stream`/`write_stream`/`mkdir`/`stat`) — binary-safe and streaming, no base64/shell hop; only directory/content *search* (`list_dir`/`glob`/`grep`) shells out to busybox-portable `find`/`grep`, parsed with the shared `deerflow.sandbox.search` helpers like `community/e2b_sandbox`. Sandboxes run as the unprivileged `tenki` user, so DeerFlow's `/mnt/user-data` prefix is remapped under a writable HOME (`_resolve_path`) and best-effort `sudo`-symlinked at bootstrap. Boxes are named deterministically from `sha256(user_id:thread_id)[:16]` (64-bit, matching E2B; the warm pool is keyed by this id alone with no full-seed fallback), released into an in-process warm pool, and reclaimed only by the same user/thread after a liveness check. A terminal session error (named SDK errors plus builtin `ConnectionError`/`BrokenPipeError`/`EOFError`) routes through `_invalidate_sandbox` to evict the dead microVM. Cross-process orphan reconciliation is a follow-up (single-process warm pool today). diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index a887db7af..02803dd28 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/backend/tests/test_boxlite_provider.py @@ -1506,3 +1506,40 @@ def test_grep_single_file_path_with_matching_glob(tmp_path, monkeypatch) -> None assert [m.path for m in matches] == [str(target)] assert truncated is False assert box.grep(str(target), "needle", glob="*.md") == ([], False) + + +def test_event_loop_thread_timeout_cancels_submitted_coroutine() -> None: + from concurrent.futures import TimeoutError as FutureTimeoutError + + from deerflow.community.boxlite.provider import _EventLoopThread + + loop_thread = _EventLoopThread() + started = threading.Event() + cancelled = threading.Event() + finished = threading.Event() + release_holder: dict[str, asyncio.Event] = {} + + async def blocking_operation() -> None: + release = asyncio.Event() + release_holder["event"] = release + started.set() + try: + await release.wait() + except asyncio.CancelledError: + cancelled.set() + raise + finally: + finished.set() + + try: + with pytest.raises(FutureTimeoutError): + loop_thread.run(blocking_operation(), timeout=0.05) + + assert started.wait(1.0) + assert cancelled.wait(1.0), "timed-out BoxLite coroutine kept running on the private loop" + finally: + release = release_holder.get("event") + if release is not None and loop_thread._loop is not None: + loop_thread._loop.call_soon_threadsafe(release.set) + finished.wait(1.0) + loop_thread.close()