From d03f1234f58b38ed7c9d76d04d1b683779fa1ab7 Mon Sep 17 00:00:00 2001 From: qkycir <108872958+qkycir@users.noreply.github.com> Date: Tue, 10 Nov 2026 15:05:33 +0800 Subject: [PATCH] fix(sandbox): recover BoxLite VMs on the provider event loop (#8072) * fix: recover BoxLite sandboxes on the async runtime loop * test: register the BoxLite scoped guidance * test(sandbox): pin BoxLite recovery lifecycle signatures * docs(changelog): link BoxLite recovery PR --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 5 + README.md | 3 + .../deerflow/community/boxlite/AGENTS.md | 19 ++ .../deerflow/community/boxlite/README.md | 12 ++ .../deerflow/community/boxlite/provider.py | 126 +++++++------ backend/tests/test_agent_guidance_check.py | 1 + backend/tests/test_boxlite_provider.py | 44 +++-- backend/tests/test_boxlite_recovery_sdk.py | 171 ++++++++++++++++++ 8 files changed, 303 insertions(+), 78 deletions(-) create mode 100644 backend/packages/harness/deerflow/community/boxlite/AGENTS.md create mode 100644 backend/tests/test_boxlite_recovery_sdk.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 407a1cd76e..3afabcf5c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,6 +58,10 @@ This section accumulates work toward the **2.3.0** milestone they carry per-call environment variables, including Lark CLI commands. Commands within one subagent remain ordered and client cleanup drains them. +- **sandbox:** BoxLite VMs recovered at startup now run health checks, commands, + and cleanup on the provider's event loop, avoiding sync/async SDK mismatches + that made healthy recovered VMs fail their first command. ([#8072]) + - **uploads:** Re-uploading a file whose name collides with a compound archive suffix no longer inserts the dedupe counter inside the suffix. A second `report.tar.gz` is now named `report_1.tar.gz` (matching `report_1.tar` and @@ -16238,4 +16242,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#8043]: https://github.com/bytedance/deer-flow/pull/8043 [#8057]: https://github.com/bytedance/deer-flow/pull/8057 [#8064]: https://github.com/bytedance/deer-flow/pull/8064 +[#8072]: https://github.com/bytedance/deer-flow/pull/8072 diff --git a/README.md b/README.md index 87a0b67c9d..d76616f45e 100644 --- a/README.md +++ b/README.md @@ -1170,6 +1170,9 @@ failed run. Live runs require a disposable OpenSandbox service and runtime home. 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. +Startup-recovered BoxLite VMs use the same private async loop as newly created +VMs for health checks, commands, and shutdown. Recovery skips boxes removed +after discovery without creating replacement VMs. Successful `read_file` calls treat a JSON document's `error` field as file data, not a failure of the read tool. This keeps receipts accurate when inspecting diff --git a/backend/packages/harness/deerflow/community/boxlite/AGENTS.md b/backend/packages/harness/deerflow/community/boxlite/AGENTS.md new file mode 100644 index 0000000000..c8f99ed055 --- /dev/null +++ b/backend/packages/harness/deerflow/community/boxlite/AGENTS.md @@ -0,0 +1,19 @@ +# BoxLite provider + +- `provider.py` owns one private asyncio loop. Fresh and startup-recovered boxes + must execute, collect output, wait, and stop on that loop, including calls from + arbitrary synchronous tool workers. Do not use the greenlet sync SDK or + `asyncio.run()` for recovered handles. +- Startup recovery is get-only: list DeerFlow-prefixed names and look them up + through the async runtime. Skip missing entries. `SimpleBox(reuse_existing=True)` + uses `get_or_create` and can create a VM, so it is not a get-only attach API. +- The recovered-box adapter translates argv and environment to native lists, + forwards `user`, `cwd`, and `timeout_secs`, drains stdout/stderr concurrently, + and waits for completion before returning the public SDK `ExecResult`. + Preserve `SimpleBox.exec` semantics for bytes, partial stream failures, and + failed waits (`exit_code=-1`). +- Keep SDK imports lazy: ordinary harness installs do not include BoxLite. + Provider unit tests run without it. The installed-SDK contract tests in + `backend/tests/test_boxlite_recovery_sdk.py` replace only the native runtime + boundary and require the optional extra on a supported SDK platform; they + must not depend on external audit files or a live hypervisor. diff --git a/backend/packages/harness/deerflow/community/boxlite/README.md b/backend/packages/harness/deerflow/community/boxlite/README.md index 5852022874..943de1a1ff 100644 --- a/backend/packages/harness/deerflow/community/boxlite/README.md +++ b/backend/packages/harness/deerflow/community/boxlite/README.md @@ -51,6 +51,12 @@ loop on a daemon thread and marshals every coroutine onto it via `user_id:thread_id`, released into an in-process warm pool after each agent turn, and reclaimed by the same thread on the next acquire. +Startup recovery uses the async runtime's `list_info()` and `get()` on that same +private loop. Recovered native box handles collect stdout and stderr concurrently +and return the SDK's `ExecResult`, matching fresh `SimpleBox` commands. Recovery +does not use the thread-affine greenlet sync API or create replacements for boxes +that disappear between listing and lookup; normal acquisition owns creation. + | File | Role | | --- | --- | | `provider.py` | `SandboxProvider` lifecycle + the private-loop bridge | @@ -76,6 +82,12 @@ from the SDK's `boxlite.TimeoutError`. The provider creates `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/skills` on box start so those virtual paths resolve natively. +`tests/test_boxlite_provider.py` covers recovery, health checks, commands, and +shutdown without the optional SDK. With `deerflow-harness[boxlite]` installed, +`tests/test_boxlite_recovery_sdk.py` additionally checks the installed SDK's native +command signature and `SimpleBox` result semantics against an offline native +runtime double. These contract tests do not start a VM or require KVM/HVF. + Warm-pool capacity is governed by `sandbox.replicas` across active + warm VMs. `sandbox.idle_timeout` controls how long released warm VMs stay running; `0` disables idle reaping. Active boxes are never evicted to satisfy the cap. diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py index 57e647c0e7..66aa5b3570 100644 --- a/backend/packages/harness/deerflow/community/boxlite/provider.py +++ b/backend/packages/harness/deerflow/community/boxlite/provider.py @@ -35,7 +35,7 @@ from ..warm_pool_lifecycle import WarmPoolLifecycleMixin from .box import BoxliteBox if TYPE_CHECKING: - from boxlite import SimpleBox + from boxlite import ExecResult, SimpleBox logger = logging.getLogger(__name__) @@ -83,13 +83,13 @@ def _import_simplebox() -> type[SimpleBox]: return SimpleBox -def _import_sync_boxlite_runtime(): - """Import BoxLite's sync runtime lazily for startup reconciliation.""" +def _import_boxlite_runtime(): + """Import BoxLite's async runtime lazily for startup reconciliation.""" try: - from boxlite import SyncBoxlite + from boxlite import Boxlite except ImportError as e: # pragma: no cover - depends on the optional dependency raise ImportError("BoxliteProvider requires the optional 'boxlite' dependency. Install it with: pip install 'deerflow-harness[boxlite]' or pip install boxlite.") from e - return SyncBoxlite + return Boxlite class _EventLoopThread: @@ -147,11 +147,10 @@ class _EventLoopThread: loop.close() -class _SyncBoxAdapter: - """Adapt a sync BoxLite ``Box`` handle to the async ``SimpleBox`` methods we use.""" +class _AsyncBoxAdapter: + """Expose SimpleBox command results for an existing async Box without creating a VM.""" - def __init__(self, runtime: Any, box: Any) -> None: - self._runtime = runtime + def __init__(self, box: Any) -> None: self._box = box async def exec( @@ -162,28 +161,43 @@ class _SyncBoxAdapter: user: str | None = None, timeout: float | None = None, cwd: str | None = None, - ) -> Any: - return self._box.exec( + ) -> ExecResult: + from boxlite import ExecResult + + execution = await self._box.exec( cmd, - *args, - env=env, + list(args) or None, + list(env.items()) if env else None, user=user, - timeout=timeout, + timeout_secs=timeout, cwd=cwd, ) - async def stop(self) -> None: + async def collect(stream_name: str) -> str: + chunks: list[str] = [] + try: + stream = getattr(execution, stream_name)() + if stream is not None: + async for chunk in stream: + chunks.append(chunk.decode("utf-8", errors="replace") if isinstance(chunk, bytes) else chunk) + except Exception: + logger.exception("Failed to collect recovered BoxLite %s", stream_name) + return "".join(chunks) + + # Like SimpleBox.exec, drain both streams before waiting, retaining partial output on stream errors. + stdout, stderr = await asyncio.gather(collect("stdout"), collect("stderr")) + error_message = None try: - self._box.stop() - finally: - self._runtime.stop() + result = await execution.wait() + exit_code = result.exit_code + error_message = result.error_message + except Exception: + logger.exception("Failed to wait for recovered BoxLite execution") + exit_code = -1 + return ExecResult(exit_code=exit_code, stdout=stdout, stderr=stderr, error_message=error_message) - -def _run_sync_adapter[T](coro: Awaitable[T], *, timeout: float | None = None) -> T: - """Run sync-adapter coroutines without using the BoxLite async loop.""" - if timeout is None: - return asyncio.run(coro) - return asyncio.run(asyncio.wait_for(coro, timeout=timeout)) + async def stop(self) -> None: + await self._box.stop() class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): @@ -337,46 +351,38 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): logger.info("Startup reconciliation adopted %s BoxLite box(es)", adopted) def _adopt_existing_boxes(self) -> int: - runtime_cls = _import_sync_boxlite_runtime() - adopted = 0 + runtime_cls = _import_boxlite_runtime() - list_runtime = runtime_cls.default().start() - try: - infos = list_runtime.list_info() - finally: - list_runtime.stop() - - for info in infos: - name = getattr(info, "name", None) - sandbox_id = self._sandbox_id_from_box_name(name) - if sandbox_id is None: - continue - with self._lock: - if sandbox_id in self._boxes or sandbox_id in self._warm_pool: + async def adopt() -> int: + runtime = runtime_cls.default() + infos = await runtime.list_info() + adopted = 0 + for info in infos: + name = getattr(info, "name", None) + sandbox_id = self._sandbox_id_from_box_name(name) + if sandbox_id is None: continue - - box_runtime = runtime_cls.default().start() - try: - box = box_runtime.get(name) - except Exception as e: - box_runtime.stop() - logger.warning("Failed to retrieve existing BoxLite box %s: %s", name, e) - continue - if box is None: - box_runtime.stop() - continue - - wrapped = BoxliteBox(sandbox_id, _SyncBoxAdapter(box_runtime, box), _run_sync_adapter, default_env=self._config["environment"], on_terminal_failure=self._invalidate_box) - with self._lock: - if sandbox_id in self._boxes or sandbox_id in self._warm_pool: - box_runtime.stop() + with self._lock: + if sandbox_id in self._boxes or sandbox_id in self._warm_pool: + continue + try: + box = await runtime.get(name) + except Exception as e: + logger.warning("Failed to retrieve existing BoxLite box %s: %s", name, e) continue - self._warm_pool[sandbox_id] = (wrapped, time.monotonic()) - self._warm_pool_identity[sandbox_id] = None - adopted += 1 - logger.info("Adopted existing BoxLite box %s (%s) into warm pool", sandbox_id, name) + if box is None: + continue + wrapped = BoxliteBox(sandbox_id, _AsyncBoxAdapter(box), self._loop.run, default_env=self._config["environment"], on_terminal_failure=self._invalidate_box) + with self._lock: + if sandbox_id in self._boxes or sandbox_id in self._warm_pool: + continue + self._warm_pool[sandbox_id] = (wrapped, time.monotonic()) + self._warm_pool_identity[sandbox_id] = None + adopted += 1 + logger.info("Adopted existing BoxLite box %s (%s) into warm pool", sandbox_id, name) + return adopted - return adopted + return self._loop.run(adopt()) # ── Acquire / release ──────────────────────────────────────────────── diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py index b85373dde8..57a9ade458 100644 --- a/backend/tests/test_agent_guidance_check.py +++ b/backend/tests/test_agent_guidance_check.py @@ -34,6 +34,7 @@ EXPECTED_GUIDANCE_PATHS = { "backend/packages/harness/deerflow/community/ddg_search/AGENTS.md", "backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md", "backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md", + "backend/packages/harness/deerflow/community/boxlite/AGENTS.md", "backend/packages/harness/deerflow/community/browser_automation/AGENTS.md", "backend/packages/harness/deerflow/community/opensandbox/AGENTS.md", "backend/packages/harness/deerflow/config/AGENTS.md", diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index 9ac1bbc588..df0d32c4d3 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/backend/tests/test_boxlite_provider.py @@ -87,13 +87,7 @@ def _no_existing_boxlite_boxes(monkeypatch: pytest.MonkeyPatch) -> None: """Keep provider startup reconciliation isolated from the real SDK state.""" class _EmptyRuntime: - def start(self): - return self - - def stop(self): - pass - - def list_info(self): + async def list_info(self): return [] class _EmptyBoxlite: @@ -101,7 +95,7 @@ def _no_existing_boxlite_boxes(monkeypatch: pytest.MonkeyPatch) -> None: def default(): return _EmptyRuntime() - monkeypatch.setattr("deerflow.community.boxlite.provider._import_sync_boxlite_runtime", lambda: _EmptyBoxlite) + monkeypatch.setattr("deerflow.community.boxlite.provider._import_boxlite_runtime", lambda: _EmptyBoxlite) def test_import_simplebox_missing_raises_actionable(monkeypatch: pytest.MonkeyPatch) -> None: @@ -443,26 +437,34 @@ def test_startup_reconciliation_adopts_prefixed_existing_boxes(monkeypatch): lambda: _stub_config(), ) stopped: list[str] = [] + commands = [] + monkeypatch.setitem(sys.modules, "boxlite", types.SimpleNamespace(ExecResult=types.SimpleNamespace)) class _NativeBox: - def stop(self): + async def exec(self, command, args=None, env=None, tty=False, user=None, timeout_secs=None, cwd=None): + commands.append((command, args, env, timeout_secs, threading.get_ident())) + + async def stream(*values): + for value in values: + yield value + + async def wait(): + return types.SimpleNamespace(exit_code=0, error_message=None) + + return types.SimpleNamespace(stdout=lambda: stream("ok\n" if args[-1] == "echo ok" else "restored\n"), stderr=lambda: stream(), wait=wait) + + async def stop(self): stopped.append("adopted") class _Runtime: - def start(self): - return self - - def stop(self): - pass - - def list_info(self): + async def list_info(self): return [ types.SimpleNamespace(name="deer-flow-boxlite-adopted"), types.SimpleNamespace(name="unrelated-box"), types.SimpleNamespace(name=None), ] - def get(self, name): + async def get(self, name): if name == "deer-flow-boxlite-adopted": return _NativeBox() raise AssertionError(f"unexpected box lookup: {name}") @@ -472,13 +474,19 @@ def test_startup_reconciliation_adopts_prefixed_existing_boxes(monkeypatch): def default(): return _Runtime() - monkeypatch.setattr("deerflow.community.boxlite.provider._import_sync_boxlite_runtime", lambda: _Boxlite) + monkeypatch.setattr("deerflow.community.boxlite.provider._import_boxlite_runtime", lambda: _Boxlite) provider = BoxliteProvider() assert list(provider._warm_pool) == ["adopted"] adopted_box = provider._warm_pool["adopted"][0] assert adopted_box.id == "adopted" + assert provider._reclaim_warm_pool("adopted", ("user", "thread")) == "adopted" + assert adopted_box.execute_command("echo restored", env={"KEY": "value"}, timeout=2) == "restored\n" + assert commands == [ + ("sh", ["-lc", "echo ok"], None, 5, provider._loop._thread.ident), + ("sh", ["-lc", "echo restored"], [("KEY", "value")], 2, provider._loop._thread.ident), + ] provider.shutdown() assert stopped == ["adopted"] diff --git a/backend/tests/test_boxlite_recovery_sdk.py b/backend/tests/test_boxlite_recovery_sdk.py new file mode 100644 index 0000000000..3388bc9ab2 --- /dev/null +++ b/backend/tests/test_boxlite_recovery_sdk.py @@ -0,0 +1,171 @@ +"""Recovered boxes exercise the installed optional SDK without starting a VM.""" + +from __future__ import annotations + +import asyncio +import importlib +import inspect +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest + +from deerflow.community.boxlite.provider import BoxliteProvider + + +@pytest.fixture +def sdk_runtime(monkeypatch): + sdk = pytest.importorskip("boxlite", reason="Install deerflow-harness[boxlite] on a supported SDK platform") + exec_signature = inspect.signature(sdk.Box.exec) + list_signature = inspect.signature(sdk.Boxlite.list_info) + get_signature = inspect.signature(sdk.Boxlite.get) + stop_signature = inspect.signature(sdk.Box.stop) + events = [] + commands = [] + + class NativeExecution: + def __init__(self, health_check, fail_wait): + self.health_check = health_check + self.fail_wait = fail_wait + self.stderr_started = asyncio.Event() + self.stdout_done = self.stderr_done = False + + def stdout(self): + async def lines(): + yield b"ok\n" if self.health_check else b"output\n" + # Reading stdout alone would block: both SDK output streams must be drained concurrently. + await self.stderr_started.wait() + self.stdout_done = True + + return lines() + + def stderr(self): + async def lines(): + self.stderr_started.set() + if not self.health_check: + yield "error\n" + self.stderr_done = True + + return lines() + + async def wait(self): + assert self.stdout_done and self.stderr_done + if self.fail_wait: + raise RuntimeError("execution wait failed") + return SimpleNamespace(exit_code=0 if self.health_check else 7, error_message=None) + + class NativeBox: + id = "recovered-native-box" + + async def __aenter__(self): + return self + + async def exec(self, *args, **kwargs): + # Bind against the installed native API rather than a permissive hand-written SDK signature. + bound = exec_signature.bind(self, *args, **kwargs) + bound.apply_defaults() + commands.append(bound.arguments) + events.append(("exec", asyncio.get_running_loop(), threading.get_ident())) + health_check = bound.arguments["command"] == "sh" and bound.arguments["args"] == ["-lc", "echo ok"] + return NativeExecution(health_check, bound.arguments["command"] == "wait-error") + + async def stop(self): + stop_signature.bind(self) + events.append(("stop", asyncio.get_running_loop(), threading.get_ident())) + + class NativeRuntime: + existing = True + + async def list_info(self): + list_signature.bind(self) + events.append(("list", asyncio.get_running_loop(), threading.get_ident())) + return [SimpleNamespace(name="deer-flow-boxlite-existing"), SimpleNamespace(name="foreign-box")] + + async def get(self, name): + get_signature.bind(self, name) + assert name == "deer-flow-boxlite-existing" + events.append(("get", asyncio.get_running_loop(), threading.get_ident())) + return native_box if self.existing else None + + async def create(self, *args, **kwargs): + raise AssertionError("Recovery must never create a VM") + + async def get_or_create(self, *args, **kwargs): + raise AssertionError("Recovery must use get, not get_or_create") + + native_box = NativeBox() + runtime = NativeRuntime() + + class RuntimeFactory: + @staticmethod + def default(): + return runtime + + # Only the native runtime boundary is replaced. SimpleBox and the old SyncBox wrappers remain real SDK code. + monkeypatch.setattr(sdk, "Boxlite", RuntimeFactory) + monkeypatch.setattr(importlib.import_module("boxlite.boxlite"), "Boxlite", RuntimeFactory) + monkeypatch.setattr( + "deerflow.community.boxlite.provider.get_app_config", + lambda: SimpleNamespace(sandbox=SimpleNamespace(idle_timeout=0, environment={"DEFAULT": "value"})), + ) + return SimpleNamespace(sdk=sdk, runtime=runtime, box=native_box, events=events, commands=commands) + + +def test_recovered_box_reclaims_executes_and_stops_on_the_private_loop(sdk_runtime): + provider = BoxliteProvider() + sync_runtime = None + try: + adopted = provider._warm_pool["existing"][0] + sync_runtime = getattr(adopted._box, "_runtime", None) + assert provider._reclaim_warm_pool("existing", ("user", "thread")) == "existing" + with ThreadPoolExecutor(max_workers=1) as worker: + output = worker.submit(adopted.execute_command, "run-tool", {"CALL": "value"}, 3).result(timeout=5) + assert "output\n" in output and "error\n" in output and "Exit Code: 7" in output + + options = {"env": {"KEY": "value"}, "user": "1000", "timeout": 2, "cwd": "/work"} + result = provider._loop.run(adopted._box.exec("tool", "argument", **options), timeout=5) + assert isinstance(result, sdk_runtime.sdk.ExecResult) + assert (result.stdout, result.stderr, result.exit_code) == ("output\n", "error\n", 7) + assert sdk_runtime.commands[-1] == { + "self": sdk_runtime.box, + "command": "tool", + "args": ["argument"], + "env": [("KEY", "value")], + "tty": False, + "user": "1000", + "timeout_secs": 2, + "cwd": "/work", + } + + async def sdk_control(): + async def create(options, name=None): + return sdk_runtime.box + + control = sdk_runtime.sdk.SimpleBox(image="python:3.12-slim", runtime=SimpleNamespace(create=create)) + await control.start() + return await control.exec("tool", "argument", **options) + + assert result == provider._loop.run(sdk_control(), timeout=5) + failed_wait = provider._loop.run(adopted._box.exec("wait-error"), timeout=5) + assert (failed_wait.stdout, failed_wait.stderr, failed_wait.exit_code) == ("output\n", "error\n", -1) + private_loop = provider._loop._loop + private_thread = provider._loop._thread.ident + provider.shutdown() + assert sdk_runtime.events[-1][0] == "stop" + assert all(loop is private_loop and thread == private_thread for _, loop, thread in sdk_runtime.events) + finally: + provider.shutdown() + # Reap the original implementation's creator-thread-owned greenlet after a red regression. + if sync_runtime is not None and getattr(sync_runtime, "_started", False): + sync_runtime.stop() + + +def test_recovery_skips_a_box_removed_after_listing_without_creating_a_replacement(sdk_runtime): + sdk_runtime.runtime.existing = False + provider = BoxliteProvider() + try: + assert provider._warm_pool == {} + assert [event[0] for event in sdk_runtime.events] == ["list", "get"] + finally: + provider.shutdown()