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 <willem.jiang@gmail.com>

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
 

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

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.

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.

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 ────────────────────────────────────────────────
 

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",

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"]

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()