diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index 7e1b1929c..9069a2e07 100644
--- a/backend/AGENTS.md
+++ b/backend/AGENTS.md
@@ -343,6 +343,12 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti
 **Implementations**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`.
 - `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths.
+- `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.
+  `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()`.
+
+
+**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. AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper.
 
 **Virtual Path System**:
 - Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills`
diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md
index 5c4c875b3..f2f93e868 100644
--- a/backend/docs/CONFIGURATION.md
+++ b/backend/docs/CONFIGURATION.md
@@ -342,6 +342,32 @@ sandbox:
    use: deerflow.community.aio_sandbox:AioSandboxProvider # Docker-based sandbox
 ```
 
+**BoxLite micro-VM Sandbox** (runs sandbox code in daemonless OCI micro-VMs):
+```yaml
+sandbox:
+   use: deerflow.community.boxlite:BoxliteProvider
+   image: python:3.12-slim
+   memory_mib: 1024                 # optional per-box memory cap
+   cpus: 2                          # optional per-box vCPUs
+   replicas: 3                      # max active + warm VMs per gateway process
+   idle_timeout: 600                # warm VM idle seconds before stop; 0 disables idle reaping
+   environment:
+      PYTHONUNBUFFERED: "1"
+```
+
+Install the optional runtime before selecting this provider:
+
+```bash
+pip install "deerflow-harness[boxlite]"
+```
+
+BoxLite boxes are named from the effective `(user_id, thread_id)` scope and are
+released into an in-process warm pool after each turn. The same user/thread can
+reclaim its warm VM on the next acquire; different threads cannot share a VM.
+`replicas` caps active plus warm VMs. When the cap is reached only warm VMs are
+evicted; active VMs continue and the provider may temporarily exceed the cap if
+all boxes are active.
+
 **Docker Execution with Kubernetes** (runs sandbox code in Kubernetes pods via provisioner service):
 
 This mode runs each sandbox in an isolated Kubernetes Pod on your **host machine's cluster**. Requires Docker Desktop K8s, OrbStack, or similar local K8s setup.
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
index b0c7c635b..8874cd47e 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
@@ -27,6 +27,14 @@ except ImportError:  # pragma: no cover - Windows fallback
     fcntl = None  # type: ignore[assignment]
     import msvcrt
 
+from deerflow.community.warm_pool_lifecycle import (
+    DEFAULT_IDLE_TIMEOUT,
+    DEFAULT_REPLICAS,
+    WarmPoolLifecycleMixin,
+)
+from deerflow.community.warm_pool_lifecycle import (
+    IDLE_CHECK_INTERVAL as _SHARED_IDLE_CHECK_INTERVAL,
+)
 from deerflow.config import get_app_config
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
 from deerflow.runtime.user_context import get_effective_user_id
@@ -45,9 +53,7 @@ logger = logging.getLogger(__name__)
 DEFAULT_IMAGE = "enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest"
 DEFAULT_PORT = 8080
 DEFAULT_CONTAINER_PREFIX = "deer-flow-sandbox"
-DEFAULT_IDLE_TIMEOUT = 600  # 10 minutes in seconds
-DEFAULT_REPLICAS = 3  # Maximum concurrent sandbox containers
-IDLE_CHECK_INTERVAL = 60  # Check every 60 seconds
+IDLE_CHECK_INTERVAL = _SHARED_IDLE_CHECK_INTERVAL
 THREAD_LOCK_EXECUTOR_WORKERS = min(32, (os.cpu_count() or 1) + 4)
 _THREAD_LOCK_EXECUTOR = ThreadPoolExecutor(max_workers=THREAD_LOCK_EXECUTOR_WORKERS, thread_name_prefix="sandbox-lock-wait")
 atexit.register(_THREAD_LOCK_EXECUTOR.shutdown, wait=False, cancel_futures=True)
@@ -105,7 +111,7 @@ def _release_cancelled_lock_acquire(lock: threading.Lock, task: asyncio.Future[b
         lock.release()
 
 
-class AioSandboxProvider(SandboxProvider):
+class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
     """Sandbox provider that manages containers running the AIO sandbox.
 
     Architecture:
@@ -352,28 +358,13 @@ class AioSandboxProvider(SandboxProvider):
 
     # ── Idle timeout management ──────────────────────────────────────────
 
-    def _start_idle_checker(self) -> None:
-        """Start the background thread that checks for idle sandboxes."""
-        self._idle_checker_thread = threading.Thread(
-            target=self._idle_checker_loop,
-            name="sandbox-idle-checker",
-            daemon=True,
-        )
-        self._idle_checker_thread.start()
-        logger.info(f"Started idle checker thread (timeout: {self._config.get('idle_timeout', DEFAULT_IDLE_TIMEOUT)}s)")
-
-    def _idle_checker_loop(self) -> None:
-        idle_timeout = self._config.get("idle_timeout", DEFAULT_IDLE_TIMEOUT)
-        while not self._idle_checker_stop.wait(timeout=IDLE_CHECK_INTERVAL):
-            try:
-                self._cleanup_idle_sandboxes(idle_timeout)
-            except Exception as e:
-                logger.error(f"Error in idle checker loop: {e}")
+    def _cleanup_idle_resources(self, idle_timeout: float) -> None:
+        """Clean AIO resources idle longer than ``idle_timeout`` seconds."""
+        self._cleanup_idle_sandboxes(idle_timeout)
 
     def _cleanup_idle_sandboxes(self, idle_timeout: float) -> None:
         current_time = time.time()
         active_to_destroy = []
-        warm_to_destroy: list[tuple[str, SandboxInfo]] = []
 
         with self._lock:
             # Active sandboxes: tracked via _last_activity
@@ -383,14 +374,6 @@ class AioSandboxProvider(SandboxProvider):
                     active_to_destroy.append(sandbox_id)
                     logger.info(f"Sandbox {sandbox_id} idle for {idle_duration:.1f}s, marking for destroy")
 
-            # Warm pool: tracked via release_timestamp stored in _warm_pool
-            for sandbox_id, (info, release_ts) in list(self._warm_pool.items()):
-                warm_duration = current_time - release_ts
-                if warm_duration > idle_timeout:
-                    warm_to_destroy.append((sandbox_id, info))
-                    del self._warm_pool[sandbox_id]
-                    logger.info(f"Warm-pool sandbox {sandbox_id} idle for {warm_duration:.1f}s, marking for destroy")
-
         # Destroy active sandboxes (re-verify still idle before acting)
         for sandbox_id in active_to_destroy:
             try:
@@ -412,13 +395,7 @@ class AioSandboxProvider(SandboxProvider):
             except Exception as e:
                 logger.error(f"Failed to destroy idle sandbox {sandbox_id}: {e}")
 
-        # Destroy warm-pool sandboxes (already removed from _warm_pool under lock above)
-        for sandbox_id, info in warm_to_destroy:
-            try:
-                self._backend.destroy(info)
-                logger.info(f"Destroyed idle warm-pool sandbox {sandbox_id}")
-            except Exception as e:
-                logger.error(f"Failed to destroy idle warm-pool sandbox {sandbox_id}: {e}")
+        self._reap_expired_warm(idle_timeout)
 
     # ── Signal handling ──────────────────────────────────────────────────
 
@@ -650,23 +627,29 @@ class AioSandboxProvider(SandboxProvider):
 
         logger.warning(f"Dropped unhealthy sandbox {sandbox_id}: {reason}")
 
-    def _replica_count(self) -> tuple[int, int]:
-        """Return configured replicas and currently tracked sandbox count."""
-        replicas = self._config.get("replicas", DEFAULT_REPLICAS)
-        with self._lock:
-            total = len(self._sandboxes) + len(self._warm_pool)
-        return replicas, total
+    def _active_count_locked(self) -> int:
+        """Return active AIO sandbox count while ``_lock`` is held."""
+        return len(self._sandboxes)
 
-    def _log_replicas_soft_cap(self, replicas: int, sandbox_id: str, evicted: str | None) -> None:
-        """Log the result of enforcing the warm-pool replica budget."""
-        if evicted:
-            logger.info(f"Evicted warm-pool sandbox {evicted} to stay within replicas={replicas}")
+    def _destroy_warm_entry(self, sandbox_id: str, entry: SandboxInfo, *, reason: str) -> None:
+        """Destroy a warm-pool sandbox using AIO-specific backend logging."""
+        try:
+            self._backend.destroy(entry)
+        except Exception as e:
+            if reason == "idle_timeout":
+                logger.error(f"Failed to destroy idle warm-pool sandbox {sandbox_id}: {e}")
+            elif reason == "replica_enforcement":
+                logger.error(f"Failed to destroy warm-pool sandbox {sandbox_id}: {e}")
+            else:
+                logger.error(f"Failed to destroy warm-pool sandbox {sandbox_id} for {reason}: {e}")
             return
 
-        # All slots are occupied by active sandboxes — proceed anyway and log.
-        # The replicas limit is a soft cap; we never forcibly stop a container
-        # that is actively serving a thread.
-        logger.warning(f"All {replicas} replica slots are in active use; creating sandbox {sandbox_id} beyond the soft limit")
+        if reason == "idle_timeout":
+            logger.info(f"Destroyed idle warm-pool sandbox {sandbox_id}")
+        elif reason == "replica_enforcement":
+            logger.info(f"Destroyed warm-pool sandbox {sandbox_id}")
+        else:
+            logger.info(f"Destroyed warm-pool sandbox {sandbox_id} for {reason}")
 
     # ── Core: acquire / get / release / shutdown ─────────────────────────
 
@@ -822,26 +805,6 @@ class AioSandboxProvider(SandboxProvider):
                 await asyncio.to_thread(_unlock_file, lock_file)
             await asyncio.to_thread(lock_file.close)
 
-    def _evict_oldest_warm(self) -> str | None:
-        """Destroy the oldest container in the warm pool to free capacity.
-
-        Returns:
-            The evicted sandbox_id, or None if warm pool is empty.
-        """
-        with self._lock:
-            if not self._warm_pool:
-                return None
-            oldest_id = min(self._warm_pool, key=lambda sid: self._warm_pool[sid][1])
-            info, _ = self._warm_pool.pop(oldest_id)
-
-        try:
-            self._backend.destroy(info)
-            logger.info(f"Destroyed warm-pool sandbox {oldest_id}")
-        except Exception as e:
-            logger.error(f"Failed to destroy warm-pool sandbox {oldest_id}: {e}")
-            return None
-        return oldest_id
-
     def _create_sandbox(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
         """Create a new sandbox via the backend.
 
@@ -989,11 +952,7 @@ class AioSandboxProvider(SandboxProvider):
             warm_items = list(self._warm_pool.items())
             self._warm_pool.clear()
 
-        # Stop idle checker
-        self._idle_checker_stop.set()
-        if self._idle_checker_thread is not None and self._idle_checker_thread.is_alive():
-            self._idle_checker_thread.join(timeout=5)
-            logger.info("Stopped idle checker thread")
+        self._stop_idle_checker()
 
         logger.info(f"Shutting down {len(sandbox_ids)} active + {len(warm_items)} warm-pool sandbox(es)")
 
diff --git a/backend/packages/harness/deerflow/community/boxlite/README.md b/backend/packages/harness/deerflow/community/boxlite/README.md
index c0c1e745d..bedecba90 100644
--- a/backend/packages/harness/deerflow/community/boxlite/README.md
+++ b/backend/packages/harness/deerflow/community/boxlite/README.md
@@ -16,14 +16,24 @@ sandbox:
   image: python:3.12-slim   # any OCI image, run unchanged (default: python:3.12-slim)
   memory_mib: 1024          # per-box memory cap (optional)
   cpus: 2                   # per-box vCPUs (optional)
+  replicas: 3              # active + warm VM cap per gateway process (default: 3)
+  idle_timeout: 600        # warm VM idle seconds before stop; 0 disables reaping
   environment:              # injected into every command
     PYTHONUNBUFFERED: "1"
 ```
 
+Install the optional runtime before selecting this provider:
+
 ```bash
-pip install boxlite   # an optional `[boxlite]` extra + uv.lock update will follow once the approach lands
+pip install "deerflow-harness[boxlite]"
 ```
 
+The `boxlite` package is an optional DeerFlow harness extra, not part of the
+default install. It is also limited to the host platforms and architectures
+where BoxLite publishes wheels and can boot micro-VMs. Unsupported development
+hosts, such as Windows, should use another sandbox provider or run DeerFlow from
+a supported Linux/macOS environment.
+
 **Host requirement:** BoxLite boots micro-VMs, so a Linux host needs KVM — i.e.
 nested virtualization when DeerFlow runs inside a cloud VM. macOS uses
 Hypervisor.framework. This is the main deployment constraint to weigh vs. the
@@ -34,10 +44,9 @@ container-based providers.
 DeerFlow's `Sandbox` contract is synchronous; BoxLite's SDK is async-native and
 its box handles are event-loop-affine. The provider owns **one** private asyncio
 loop on a daemon thread and marshals every coroutine onto it via
-`run_coroutine_threadsafe`. This keeps all operations on the loop the box was
-started on and is safe under DeerFlow's `asyncio.to_thread` worker pool — without
-using BoxLite's greenlet sync facade, which refuses to run inside an async
-context and is thread-affine.
+`run_coroutine_threadsafe`. BoxLite boxes are named deterministically from
+`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.
 
 | File | Role |
 | --- | --- |
@@ -57,8 +66,9 @@ inside the box and reuse `deerflow.sandbox.search`, mirroring `e2b_sandbox`:
 The provider creates `/mnt/user-data/{workspace,uploads,outputs}` and
 `/mnt/skills` on box start so those virtual paths resolve natively.
 
-**Out of scope for this pass** (follow-ups): warm pooling, idle reaping, mount
-syncing, and remote/provisioner modes.
+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.
 
 ## Status
 
diff --git a/backend/packages/harness/deerflow/community/boxlite/__init__.py b/backend/packages/harness/deerflow/community/boxlite/__init__.py
index 125a8eadc..20adefb01 100644
--- a/backend/packages/harness/deerflow/community/boxlite/__init__.py
+++ b/backend/packages/harness/deerflow/community/boxlite/__init__.py
@@ -17,13 +17,14 @@ Configuration example (``config.yaml``)::
       image: python:3.12-slim      # any OCI image; runs unchanged
       memory_mib: 1024             # per-box memory cap (optional)
       cpus: 2                      # per-box vCPUs (optional)
+      replicas: 3                  # active + warm VM cap per gateway process
+      idle_timeout: 600            # warm VM idle seconds before stop; 0 disables
       environment:                 # injected into every command
         PYTHONUNBUFFERED: "1"
 
-Install the runtime (an optional ``[boxlite]`` extra + lockfile update will
-follow once the approach lands)::
+Install the optional runtime before selecting this provider::
 
-    pip install boxlite
+    pip install "deerflow-harness[boxlite]"
 
 Host requirement: BoxLite boots micro-VMs, so a Linux host needs KVM (nested
 virtualization when DeerFlow itself runs inside a cloud VM); macOS uses
diff --git a/backend/packages/harness/deerflow/community/boxlite/box.py b/backend/packages/harness/deerflow/community/boxlite/box.py
index d670e8e1f..b923c29b2 100644
--- a/backend/packages/harness/deerflow/community/boxlite/box.py
+++ b/backend/packages/harness/deerflow/community/boxlite/box.py
@@ -29,7 +29,7 @@ from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env
 from deerflow.sandbox.search import GrepMatch, path_matches, should_ignore_path, truncate_line
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable, Callable
+    from collections.abc import Callable
 
     from boxlite import SimpleBox
 
@@ -61,7 +61,7 @@ class BoxliteBox(Sandbox):
         self,
         id: str,
         box: SimpleBox,
-        run: Callable[[Awaitable[T]], T],
+        run: Callable[..., T],
         *,
         default_env: dict[str, str] | None = None,
     ) -> None:
@@ -124,6 +124,11 @@ class BoxliteBox(Sandbox):
         DeerFlow passes a bash command *string*; BoxLite's ``exec`` takes argv, so
         it runs through ``sh -lc``. Per-call ``env`` is layered over the static
         config environment and scoped to this command only.
+
+        *timeout* bounds both layers: BoxLite's SDK ``exec(timeout=...)`` handles
+        command timeout inside the VM, and the event-loop bridge receives the
+        same value so ``run_coroutine_threadsafe(...).result(timeout)`` cannot
+        block the caller forever if the SDK future itself never resolves.
         """
         _validate_extra_env(env)  # POSIX env-var key rule; raises ValueError on a bad key
         merged_env = {**self._default_env, **(env or {})} or None
@@ -132,7 +137,7 @@ class BoxliteBox(Sandbox):
                 return "Error: sandbox has been closed"
             box = self._box
         try:
-            result = self._run(box.exec("sh", "-lc", command, env=merged_env, timeout=timeout))
+            result = self._run(box.exec("sh", "-lc", command, env=merged_env, timeout=timeout), timeout=timeout)
         except Exception as e:
             logger.error("Failed to execute command in BoxLite box %s: %s", self.id, e)
             return f"Error: {e}"
diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py
index e1e61f941..33c62a259 100644
--- a/backend/packages/harness/deerflow/community/boxlite/provider.py
+++ b/backend/packages/harness/deerflow/community/boxlite/provider.py
@@ -8,15 +8,18 @@ Config is read off :class:`SandboxConfig` (``extra="allow"``), so BoxLite keys
 may appear under ``sandbox:`` in ``config.yaml`` even though they are not declared
 on the model — see this package's ``__init__`` docstring for the full set. The
 provider creates one micro-VM per ``(user, thread)`` and reuses it within the
-process; warm pooling, idle reaping and remote modes are out of scope for now.
+process.
 """
 
 from __future__ import annotations
 
 import asyncio
 import atexit
+import hashlib
 import logging
 import threading
+import time
+import uuid
 from collections.abc import Awaitable
 from typing import TYPE_CHECKING, Any, TypeVar
 
@@ -26,6 +29,7 @@ from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
 from deerflow.sandbox.sandbox import Sandbox
 from deerflow.sandbox.sandbox_provider import SandboxProvider
 
+from ..warm_pool_lifecycle import WarmPoolLifecycleMixin
 from .box import BoxliteBox
 
 if TYPE_CHECKING:
@@ -36,6 +40,7 @@ logger = logging.getLogger(__name__)
 T = TypeVar("T")
 
 DEFAULT_IMAGE = "python:3.12-slim"
+_BOX_NAME_PREFIX = "deer-flow-boxlite-"
 # DeerFlow's virtual prefixes, materialised on the box rootfs at start so the
 # Sandbox file APIs (which address /mnt/user-data/...) resolve natively.
 _VIRTUAL_DIRS = (
@@ -55,10 +60,19 @@ def _import_simplebox() -> type[SimpleBox]:
     try:
         from boxlite import SimpleBox
     except ImportError as e:  # pragma: no cover - depends on the optional dependency
-        raise ImportError("BoxliteProvider requires the 'boxlite' package. Install it with: pip install boxlite.") from e
+        raise ImportError("BoxliteProvider requires the optional 'boxlite' dependency. Install it with: pip install 'deerflow-harness[boxlite]' or pip install boxlite.") from e
     return SimpleBox
 
 
+def _import_sync_boxlite_runtime():
+    """Import BoxLite's sync runtime lazily for startup reconciliation."""
+    try:
+        from boxlite import SyncBoxlite
+    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
+
+
 class _EventLoopThread:
     """A private asyncio event loop running on a dedicated daemon thread.
 
@@ -71,34 +85,106 @@ class _EventLoopThread:
     """
 
     def __init__(self) -> None:
-        self._loop = asyncio.new_event_loop()
-        self._thread = threading.Thread(target=self._loop.run_forever, name="boxlite-loop", daemon=True)
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._ready = threading.Event()
+        self._thread = threading.Thread(target=self._run_forever, name="boxlite-loop", daemon=True)
         self._thread.start()
+        self._ready.wait(timeout=5)
+
+    def _run_forever(self) -> None:
+        self._loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._loop)
+        self._loop.call_soon(self._ready.set)
+        self._loop.run_forever()
 
     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)
 
     def close(self) -> None:
+        if self._loop is None:
+            return
         self._loop.call_soon_threadsafe(self._loop.stop)
+        wake = getattr(self._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()
 
 
-class BoxliteProvider(SandboxProvider):
+class _SyncBoxAdapter:
+    """Adapt a sync BoxLite ``Box`` handle to the async ``SimpleBox`` methods we use."""
+
+    def __init__(self, runtime: Any, box: Any) -> None:
+        self._runtime = runtime
+        self._box = box
+
+    async def exec(
+        self,
+        cmd: str,
+        *args: str,
+        env: dict[str, str] | None = None,
+        user: str | None = None,
+        timeout: float | None = None,
+        cwd: str | None = None,
+    ) -> Any:
+        return self._box.exec(
+            cmd,
+            *args,
+            env=env,
+            user=user,
+            timeout=timeout,
+            cwd=cwd,
+        )
+
+    async def stop(self) -> None:
+        try:
+            self._box.stop()
+        finally:
+            self._runtime.stop()
+
+
+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))
+
+
+class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
     """Run each DeerFlow sandbox as a BoxLite micro-VM."""
 
     uses_thread_data_mounts = False
     needs_upload_permission_adjustment = True
+    _idle_checker_thread_name = "boxlite-idle-reaper"
+
+    @staticmethod
+    def _sandbox_id(thread_id: str, user_id: str) -> str:
+        """Deterministic sandbox ID from user/thread scope.
+
+        Includes user_id so a box created for one user's bucket cannot be
+        reclaimed by another user's thread with the same thread_id.
+        """
+        return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:8]
+
+    # ── Provider ────────────────────────────────────────────────────────
 
     def __init__(self) -> None:
         self._lock = threading.Lock()
         self._boxes: dict[str, BoxliteBox] = {}
         self._thread_boxes: dict[tuple[str, str], str] = {}
+        self._warm_pool: dict[str, tuple[BoxliteBox, float]] = {}
+        self._acquire_locks: dict[str, threading.Lock] = {}
+        self._idle_checker_stop = threading.Event()
+        self._idle_checker_thread: threading.Thread | None = None
         self._shutdown_called = False
         self._config = self._load_config()
         self._loop = _EventLoopThread()
         atexit.register(self.shutdown)
+        self._reconcile_orphans()
+        self._start_idle_checker()
 
     def _load_config(self) -> dict[str, Any]:
         sandbox_config = get_app_config().sandbox
@@ -108,39 +194,172 @@ class BoxliteProvider(SandboxProvider):
 
         # $VARS in config.yaml are already resolved by AppConfig.resolve_env_variables
         # (which raises on a missing var), so the environment dict is used as-is.
+        replicas = _opt("replicas")
+        idle_timeout = _opt("idle_timeout")
         return {
             "image": _opt("image") or DEFAULT_IMAGE,
             "memory_mib": _opt("memory_mib"),
             "cpus": _opt("cpus"),
             "environment": dict(_opt("environment") or {}),
+            "replicas": replicas if replicas is not None else self.DEFAULT_REPLICAS,
+            "idle_timeout": idle_timeout if idle_timeout is not None else self.DEFAULT_IDLE_TIMEOUT,
         }
 
     @staticmethod
     def _thread_key(thread_id: str, user_id: str | None) -> tuple[str, str]:
         return (user_id or "", thread_id)
 
+    @staticmethod
+    def _box_name(sandbox_id: str) -> str:
+        return f"{_BOX_NAME_PREFIX}{sandbox_id}"
+
+    @staticmethod
+    def _sandbox_id_from_box_name(name: str | None) -> str | None:
+        if not name or not name.startswith(_BOX_NAME_PREFIX):
+            return None
+        sandbox_id = name[len(_BOX_NAME_PREFIX) :]
+        return sandbox_id or None
+
+    def _lock_for_sandbox(self, sandbox_id: str) -> threading.Lock:
+        """Return the per-sandbox acquire lock for a deterministic sandbox id."""
+        with self._lock:
+            lock = self._acquire_locks.get(sandbox_id)
+            if lock is None:
+                lock = threading.Lock()
+                self._acquire_locks[sandbox_id] = lock
+            return lock
+
+    def _start_idle_checker(self) -> None:
+        """Start idle cleanup when enabled; idle_timeout=0 keeps it disabled."""
+        if self._config["idle_timeout"] <= 0:
+            return
+        super()._start_idle_checker()
+
+    def _active_count_locked(self) -> int:
+        """Return active BoxLite box count while ``_lock`` is held."""
+        return len(self._boxes)
+
+    def _destroy_warm_entry(self, sandbox_id: str, entry: BoxliteBox, *, reason: str) -> None:
+        """Close a removed warm-pool entry and log with context."""
+        try:
+            entry.close()
+            if reason == "idle_timeout":
+                logger.info("Idle reaper destroyed expired warm-pool box %s", sandbox_id)
+            elif reason == "replica_enforcement":
+                logger.info("Replica enforcement evicted oldest warm-pool box %s", sandbox_id)
+            else:
+                logger.info("Destroyed warm-pool box %s (reason=%s)", sandbox_id, reason)
+        except Exception as e:
+            if reason == "idle_timeout":
+                logger.warning("Error closing expired BoxLite box %s: %s", sandbox_id, e)
+            elif reason == "replica_enforcement":
+                logger.warning("Error closing evicted BoxLite box %s: %s", sandbox_id, e)
+            else:
+                logger.warning("Error closing BoxLite box %s (reason=%s): %s", sandbox_id, reason, e)
+
+    def _reconcile_orphans(self) -> None:
+        """Adopt DeerFlow-owned BoxLite boxes left by a previous provider/process.
+
+        BoxLite boxes are discovered by a DeerFlow-specific name prefix. Adopted
+        boxes enter the warm pool so the normal idle reaper can reclaim them.
+        """
+        try:
+            adopted = self._adopt_existing_boxes()
+        except ImportError:
+            logger.debug("BoxLite is not installed; skipping startup reconciliation")
+            return
+        except Exception as e:
+            logger.warning("Failed to reconcile existing BoxLite boxes: %s", e)
+            return
+
+        if adopted:
+            logger.info("Startup reconciliation adopted %s BoxLite box(es)", adopted)
+
+    def _adopt_existing_boxes(self) -> int:
+        runtime_cls = _import_sync_boxlite_runtime()
+        now = time.time()
+        adopted = 0
+
+        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:
+                    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"])
+            with self._lock:
+                if sandbox_id in self._boxes or sandbox_id in self._warm_pool:
+                    box_runtime.stop()
+                    continue
+                self._warm_pool[sandbox_id] = (wrapped, now)
+                adopted += 1
+            logger.info("Adopted existing BoxLite box %s (%s) into warm pool", sandbox_id, name)
+
+        return adopted
+
+    # ── Acquire / release ────────────────────────────────────────────────
+
     def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
-        if thread_id is not None:
-            key = self._thread_key(thread_id, user_id)
+        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
+            return box.id
+
+        key = self._thread_key(thread_id, user_id)
+        sandbox_id = self._sandbox_id(thread_id, user_id)
+        acquire_lock = self._lock_for_sandbox(sandbox_id)
+        with acquire_lock:
             with self._lock:
                 existing = self._thread_boxes.get(key)
                 if existing is not None and existing in self._boxes:
                     return existing
 
-        box = self._create_box()
+            reclaimed = self._reclaim_warm_pool(sandbox_id)
+            if reclaimed is not None:
+                with self._lock:
+                    self._thread_boxes[key] = reclaimed
+                return reclaimed
 
-        with self._lock:
-            self._boxes[box.id] = box
-            if thread_id is not None:
-                self._thread_boxes[self._thread_key(thread_id, user_id)] = box.id
-        return box.id
+            box = self._create_box(sandbox_id)
+            with self._lock:
+                self._boxes[box.id] = box
+                self._thread_boxes[key] = box.id
+            return box.id
 
-    def _create_box(self) -> BoxliteBox:
+    def _create_box(self, sandbox_id: str) -> BoxliteBox:
+        # Enforce replica limit: evict oldest warm-pool box if active + warm boxes are at capacity.
+        replicas, total = self._replica_count()
+        if total >= replicas:
+            evicted = self._evict_oldest_warm()
+            self._log_replicas_soft_cap(replicas, sandbox_id, evicted)
         simplebox_cls = _import_simplebox()
         mkdir_cmd = "mkdir -p " + " ".join(_VIRTUAL_DIRS)
 
         async def _make() -> SimpleBox:
             box = simplebox_cls(
+                name=self._box_name(sandbox_id),
                 image=self._config["image"],
                 memory_mib=self._config["memory_mib"],
                 cpus=self._config["cpus"],
@@ -151,36 +370,109 @@ class BoxliteProvider(SandboxProvider):
             return box
 
         box = self._loop.run(_make())
-        logger.info("Created BoxLite box %s (image=%s)", box.id, self._config["image"])
-        return BoxliteBox(box.id, box, self._loop.run, default_env=self._config["environment"])
+        logger.info("Created BoxLite box %s (name=%s, image=%s)", sandbox_id, self._box_name(sandbox_id), self._config["image"])
+        return BoxliteBox(sandbox_id, box, self._loop.run, default_env=self._config["environment"])
 
     def get(self, sandbox_id: str) -> Sandbox | None:
         with self._lock:
             return self._boxes.get(sandbox_id)
 
     def release(self, sandbox_id: str) -> None:
+        """Release a sandbox into the warm pool — VM stays running.
+
+        The box is moved from _boxes to _warm_pool; _thread_boxes entries are
+        cleared so the thread no longer holds an active reference. The VM is
+        NOT stopped unless shutdown has already begun.
+        """
+        close_box: BoxliteBox | None = None
         with self._lock:
             box = self._boxes.pop(sandbox_id, None)
             for key in [k for k, sid in self._thread_boxes.items() if sid == sandbox_id]:
                 self._thread_boxes.pop(key, None)
-        if box is not None:
+            if box is None:
+                return
+            if self._shutdown_called:
+                close_box = box
+            else:
+                self._warm_pool[sandbox_id] = (box, time.time())
+
+        if close_box is not None:
+            close_box.close()
+            logger.info("Closed released sandbox %s because shutdown is in progress", sandbox_id)
+        else:
+            logger.info("Released sandbox %s to warm pool (VM still running)", sandbox_id)
+
+    def _reclaim_warm_pool(self, sandbox_id: str) -> str | None:
+        """Try to reclaim a warm-pool box by sandbox_id.
+
+        Returns sandbox_id on success, None if not found or dead.
+        """
+        with self._lock:
+            if sandbox_id not in self._warm_pool:
+                return None
+            box, _ = self._warm_pool[sandbox_id]
+
+        # Health check: run a simple command to verify the VM is alive
+        try:
+            result = box.execute_command("echo ok", timeout=5)
+            if "ok" not in result:
+                logger.warning("Warm pool box %s health check failed: %s", sandbox_id, result)
+                with self._lock:
+                    self._warm_pool.pop(sandbox_id, None)
+                box.close()
+                return None
+        except Exception as e:
+            logger.warning("Warm pool box %s health check error: %s", sandbox_id, e)
+            with self._lock:
+                self._warm_pool.pop(sandbox_id, None)
             box.close()
+            return None
+
+        # Promote from warm pool to active
+        with self._lock:
+            warm_entry = self._warm_pool.pop(sandbox_id, None)
+            if warm_entry is None:
+                return None  # Raced with another thread
+            box, _ = warm_entry
+            self._boxes[sandbox_id] = box
+
+        logger.info("Reclaimed warm-pool box %s", sandbox_id)
+        return sandbox_id
 
     def reset(self) -> None:
+        """Release tracked BoxLite VMs to this instance's warm-pool cleanup.
+
+        ``reset_sandbox_provider()`` drops the provider singleton and calls this
+        lightweight hook so config changes take effect on the next provider
+        construction. Teardown belongs to ``shutdown()``; reset intentionally
+        leaves running VMs alive, but keeps them visible to this instance's idle
+        reaper and atexit shutdown instead of orphaning them.
+        """
         with self._lock:
+            now = time.time()
+            for sandbox_id, box in self._boxes.items():
+                self._warm_pool.setdefault(sandbox_id, (box, now))
             self._boxes.clear()
             self._thread_boxes.clear()
+            self._acquire_locks.clear()
 
     def shutdown(self) -> None:
         with self._lock:
             if self._shutdown_called:
                 return
             self._shutdown_called = True
-            active = list(self._boxes.values())
-            self._boxes.clear()
-            self._thread_boxes.clear()
 
-        for box in active:
+        self._stop_idle_checker()
+
+        with self._lock:
+            active = list(self._boxes.values())
+            warm = [box for box, _ in self._warm_pool.values()]
+            self._boxes.clear()
+            self._warm_pool.clear()
+            self._thread_boxes.clear()
+            self._acquire_locks.clear()
+
+        for box in active + warm:
             try:
                 box.close()
             except Exception as e:  # pragma: no cover - defensive
diff --git a/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py b/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py
new file mode 100644
index 000000000..ffb5c066e
--- /dev/null
+++ b/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py
@@ -0,0 +1,127 @@
+"""Shared warm-pool lifecycle helpers for community sandbox providers."""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_IDLE_TIMEOUT = 600
+DEFAULT_REPLICAS = 3
+IDLE_CHECK_INTERVAL = 60
+
+
+class WarmPoolLifecycleMixin[WarmEntryT]:
+    """Mixin for provider warm-pool expiry and replica lifecycle mechanics."""
+
+    DEFAULT_IDLE_TIMEOUT = DEFAULT_IDLE_TIMEOUT
+    DEFAULT_REPLICAS = DEFAULT_REPLICAS
+    IDLE_CHECK_INTERVAL = IDLE_CHECK_INTERVAL
+    _idle_checker_thread_name = "warm-pool-idle-checker"
+
+    _lock: threading.Lock
+    _warm_pool: dict[str, tuple[WarmEntryT, float]]
+    _config: dict[str, Any]
+    _idle_checker_stop: threading.Event
+    _idle_checker_thread: threading.Thread | None
+
+    def _active_count_locked(self) -> int:
+        """Return active entry count while ``_lock`` is held."""
+        raise NotImplementedError
+
+    def _destroy_warm_entry(self, sandbox_id: str, entry: WarmEntryT, *, reason: str) -> None:
+        """Destroy a warm-pool entry after it has been removed from the pool."""
+        raise NotImplementedError
+
+    def _replica_count(self) -> tuple[int, int]:
+        """Return configured replicas and current active + warm entry count."""
+        replicas = int(self._config.get("replicas", DEFAULT_REPLICAS))
+        with self._lock:
+            total = self._active_count_locked() + len(self._warm_pool)
+        return replicas, total
+
+    def _log_replicas_soft_cap(self, replicas: int, sandbox_id: str, evicted: str | None) -> None:
+        """Log the result of enforcing the warm-pool replica soft cap."""
+        if evicted is not None:
+            logger.info("Evicted warm-pool sandbox %s to stay within replicas=%s", evicted, replicas)
+            return
+
+        logger.warning(
+            "All %s replica slots are in active use; creating sandbox %s beyond the soft limit",
+            replicas,
+            sandbox_id,
+        )
+
+    def _evict_oldest_warm(self) -> str | None:
+        """Remove and destroy the oldest warm entry by timestamp."""
+        with self._lock:
+            if not self._warm_pool:
+                return None
+            sandbox_id, (entry, _) = min(self._warm_pool.items(), key=lambda item: item[1][1])
+            self._warm_pool.pop(sandbox_id)
+
+        self._destroy_warm_entry(sandbox_id, entry, reason="replica_enforcement")
+        return sandbox_id
+
+    def _reap_expired_warm(self, idle_timeout: float | None = None) -> None:
+        """Remove and destroy warm entries older than ``idle_timeout`` seconds."""
+        timeout = float(self._config.get("idle_timeout", DEFAULT_IDLE_TIMEOUT) if idle_timeout is None else idle_timeout)
+        if timeout <= 0:
+            return
+
+        now = time.time()
+        expired: list[tuple[str, WarmEntryT]] = []
+        with self._lock:
+            for sandbox_id, (entry, timestamp) in self._warm_pool.items():
+                if now - timestamp > timeout:
+                    expired.append((sandbox_id, entry))
+            for sandbox_id, _ in expired:
+                self._warm_pool.pop(sandbox_id, None)
+
+        for sandbox_id, entry in expired:
+            self._destroy_warm_entry(sandbox_id, entry, reason="idle_timeout")
+
+    def _start_idle_checker(self) -> None:
+        """Start the daemon thread that periodically cleans idle warm entries."""
+        if self._idle_checker_thread is not None and self._idle_checker_thread.is_alive():
+            return
+
+        self._idle_checker_stop.clear()
+        self._idle_checker_thread = threading.Thread(
+            target=self._idle_checker_loop,
+            name=self._idle_checker_thread_name,
+            daemon=True,
+        )
+        self._idle_checker_thread.start()
+        logger.info("Started warm-pool idle checker thread (timeout: %ss)", self._config.get("idle_timeout", DEFAULT_IDLE_TIMEOUT))
+
+    def _stop_idle_checker(self) -> None:
+        """Stop the idle checker thread and wait for it to exit when running."""
+        self._idle_checker_stop.set()
+        thread = self._idle_checker_thread
+        if thread is not None and thread.is_alive() and thread is not threading.current_thread():
+            thread.join(timeout=5)
+
+    def _idle_checker_loop(self) -> None:
+        """Run periodic idle cleanup until the stop event is set."""
+        idle_timeout = float(self._config.get("idle_timeout", DEFAULT_IDLE_TIMEOUT))
+        while not self._idle_checker_stop.wait(self.IDLE_CHECK_INTERVAL):
+            try:
+                self._cleanup_idle_resources(idle_timeout)
+            except Exception:
+                logger.exception("Error in warm-pool idle checker loop")
+
+    def _cleanup_idle_resources(self, idle_timeout: float) -> None:
+        """Clean resources idle longer than ``idle_timeout`` seconds."""
+        self._reap_expired_warm(idle_timeout)
+
+
+__all__ = [
+    "DEFAULT_IDLE_TIMEOUT",
+    "DEFAULT_REPLICAS",
+    "IDLE_CHECK_INTERVAL",
+    "WarmPoolLifecycleMixin",
+]
diff --git a/backend/packages/harness/deerflow/config/sandbox_config.py b/backend/packages/harness/deerflow/config/sandbox_config.py
index d8434ccf8..d2f0eb382 100644
--- a/backend/packages/harness/deerflow/config/sandbox_config.py
+++ b/backend/packages/harness/deerflow/config/sandbox_config.py
@@ -30,14 +30,16 @@ class SandboxConfig(BaseModel):
         allow_host_bash: Enable host-side bash execution for LocalSandboxProvider.
             Dangerous and intended only for fully trusted local workflows.
 
+    AioSandboxProvider and BoxliteProvider shared options:
+        image: Sandbox image to use (Docker/AIO image or BoxLite OCI image)
+        replicas: Maximum active + warm sandboxes/VMs per gateway process (default: 3). When the limit is reached, warm/least-recently-used sandboxes are evicted to make room; active sandboxes are not forcibly stopped.
+        idle_timeout: Idle timeout in seconds before released warm sandboxes/VMs are stopped (default: 600 = 10 minutes). Set to 0 to disable.
+        environment: Environment variables to inject into the sandbox (values starting with $ are resolved from host env)
+
     AioSandboxProvider specific options:
-        image: Docker image to use (default: enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest)
         port: Base port for sandbox containers (default: 8080)
-        replicas: Maximum number of concurrent sandbox containers (default: 3). When the limit is reached the least-recently-used sandbox is evicted to make room.
         container_prefix: Prefix for container names (default: deer-flow-sandbox)
-        idle_timeout: Idle timeout in seconds before sandbox is released (default: 600 = 10 minutes). Set to 0 to disable.
         mounts: List of volume mounts to share directories with the container
-        environment: Environment variables to inject into the container (values starting with $ are resolved from host env)
     """
 
     use: str = Field(
@@ -50,7 +52,7 @@ class SandboxConfig(BaseModel):
     )
     image: str | None = Field(
         default=None,
-        description="Docker image to use for the sandbox container",
+        description="Sandbox image to use (Docker/AIO image or BoxLite OCI image)",
     )
     port: int | None = Field(
         default=None,
@@ -58,7 +60,7 @@ class SandboxConfig(BaseModel):
     )
     replicas: int | None = Field(
         default=None,
-        description="Maximum number of concurrent sandbox containers (default: 3). When the limit is reached the least-recently-used sandbox is evicted to make room.",
+        description="Maximum active + warm sandboxes/VMs per gateway process (default: 3). Warm/least-recently-used entries are evicted to make room; active sandboxes are not forcibly stopped.",
     )
     container_prefix: str | None = Field(
         default=None,
@@ -66,7 +68,7 @@ class SandboxConfig(BaseModel):
     )
     idle_timeout: int | None = Field(
         default=None,
-        description="Idle timeout in seconds before sandbox is released (default: 600 = 10 minutes). Set to 0 to disable.",
+        description="Idle timeout in seconds before released warm sandboxes/VMs are stopped (default: 600 = 10 minutes). Set to 0 to disable.",
     )
     mounts: list[VolumeMountConfig] = Field(
         default_factory=list,
diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml
index 8b6d6e6e6..746d5badf 100644
--- a/backend/packages/harness/pyproject.toml
+++ b/backend/packages/harness/pyproject.toml
@@ -64,6 +64,7 @@ postgres = [
 # always installs this extra because Docker defaults to the redis bridge.
 redis = ["redis>=5.0.0"]
 pymupdf = ["pymupdf4llm>=0.0.17"]
+boxlite = ["boxlite>=0.9.7"]
 
 [build-system]
 requires = ["hatchling"]
diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py
index 3c0aea357..e0349f0a0 100644
--- a/backend/tests/test_aio_sandbox_provider.py
+++ b/backend/tests/test_aio_sandbox_provider.py
@@ -626,3 +626,66 @@ def test_destroy_swallows_close_errors_and_still_destroys_backend(tmp_path, capl
 
     assert "Error closing sandbox sandbox-dest-err during destroy" in caplog.text
     provider._backend.destroy.assert_called_once()
+
+
+def test_cleanup_idle_sandboxes_keeps_active_cleanup_and_delegates_warm_expiry(tmp_path):
+    """AIO active-idle cleanup must remain local while warm expiry uses the shared lifecycle."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._lock = aio_mod.threading.Lock()
+    provider._sandboxes = {"active-old": MagicMock()}
+    provider._sandbox_infos = {
+        "active-old": aio_mod.SandboxInfo(sandbox_id="active-old", sandbox_url="http://active-old"),
+    }
+    provider._thread_sandboxes = {("default", "thread-old"): "active-old"}
+    provider._last_activity = {"active-old": 0.0}
+    provider._warm_pool = {
+        "warm-old": (
+            aio_mod.SandboxInfo(sandbox_id="warm-old", sandbox_url="http://warm-old"),
+            0.0,
+        )
+    }
+
+    calls = []
+    provider.destroy = MagicMock(side_effect=lambda _sandbox_id: calls.append("active"))
+    provider._reap_expired_warm = MagicMock(side_effect=lambda _idle_timeout: calls.append("warm"))
+
+    provider._cleanup_idle_sandboxes(1.0)
+
+    provider.destroy.assert_called_once_with("active-old")
+    provider._reap_expired_warm.assert_called_once_with(1.0)
+    assert calls == ["active", "warm"]
+
+
+def test_create_sandbox_evicts_oldest_warm_replica_via_shared_lifecycle(tmp_path, monkeypatch):
+    """Replica enforcement must destroy the oldest warm SandboxInfo before creating another."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._lock = aio_mod.threading.Lock()
+    provider._config = {"replicas": 2}
+    provider._sandboxes = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._last_activity = {}
+
+    oldest_info = aio_mod.SandboxInfo(sandbox_id="warm-oldest", sandbox_url="http://warm-oldest")
+    newest_info = aio_mod.SandboxInfo(sandbox_id="warm-newest", sandbox_url="http://warm-newest")
+    created_info = aio_mod.SandboxInfo(sandbox_id="created", sandbox_url="http://created")
+    provider._warm_pool = {
+        "warm-newest": (newest_info, 20.0),
+        "warm-oldest": (oldest_info, 10.0),
+    }
+    provider._backend = SimpleNamespace(
+        create=MagicMock(return_value=created_info),
+        destroy=MagicMock(),
+    )
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_extra_mounts", lambda _self, _thread_id, *, user_id=None: [])
+    monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda _url, *, timeout=60: True)
+
+    sandbox_id = provider._create_sandbox(None, "created", user_id="default")
+
+    assert sandbox_id == "created"
+    provider._backend.destroy.assert_called_once_with(oldest_info)
+    assert "warm-oldest" not in provider._warm_pool
+    assert provider._warm_pool == {"warm-newest": (newest_info, 20.0)}
+    assert provider._sandbox_infos["created"] is created_info
diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py
index 4c92ff1c8..98c1abe08 100644
--- a/backend/tests/test_boxlite_provider.py
+++ b/backend/tests/test_boxlite_provider.py
@@ -1,12 +1,15 @@
 """Unit tests for the BoxLite community provider.
-
 These run in CI without BoxLite installed: they cover the lazy-import error path,
-provider lifecycle, and the path-safety guards — none of which need a live box.
+provider lifecycle, the path-safety guards, and the warm pool lifecycle — none of
+which need a live box.
 """
 
 from __future__ import annotations
 
+import asyncio
 import sys
+import threading
+import time
 import types
 
 import pytest
@@ -14,15 +17,80 @@ import pytest
 from deerflow.community.boxlite.box import BoxliteBox
 from deerflow.community.boxlite.provider import BoxliteProvider, _import_simplebox
 
+# ── Fake BoxLite SDK ──────────────────────────────────────────────────
+
+
+class _FakeBox:
+    """A fake SimpleBox that records lifecycle calls without starting real VMs."""
+
+    def __init__(self, *, image=None, name=None, memory_mib=None, cpus=None, **kwargs):
+        self.id = name or "auto-gen-id"
+        self.name = name
+        self._image = image
+        self._started = False
+        self._stopped = False
+        self._exec_history: list[tuple] = []
+
+    async def start(self):
+        self._started = True
+
+    async def exec(self, *argv, env=None, timeout=None):
+        self._exec_history.append((argv, env, timeout))
+        _FakeResult = type("_FakeResult", (), {"stdout": "", "stderr": "", "exit_code": 0})
+        # Health check: box.execute_command("echo ok") → exec("sh", "-lc", "echo ok")
+        if len(argv) >= 3 and argv[0] == "sh" and argv[1] == "-lc" and argv[2] == "echo ok":
+            return type("_FakeResult", (), {"stdout": "ok\n", "stderr": "", "exit_code": 0})()
+        return _FakeResult()
+
+    async def stop(self):
+        self._stopped = True
+
+
+def _fake_run(coro, *, timeout=None):
+    """Sync runner that executes coroutines on a temporary event loop (no daemon thread)."""
+    return asyncio.run(coro)
+
+
+# ── Config stub ───────────────────────────────────────────────────────
+
+
+def _stub_config(sandbox_attrs=None):
+    """Stub get_app_config to return a config with given sandbox attrs."""
+    attrs = sandbox_attrs or {}
+    stub = types.SimpleNamespace(sandbox=types.SimpleNamespace(**attrs))
+    return stub
+
 
 def _no_boxlite(monkeypatch: pytest.MonkeyPatch) -> None:
     """Make ``import boxlite`` raise, regardless of whether it is installed."""
     monkeypatch.setitem(sys.modules, "boxlite", None)
 
 
+@pytest.fixture(autouse=True)
+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):
+            return []
+
+    class _EmptyBoxlite:
+        @staticmethod
+        def default():
+            return _EmptyRuntime()
+
+    monkeypatch.setattr("deerflow.community.boxlite.provider._import_sync_boxlite_runtime", lambda: _EmptyBoxlite)
+
+
 def test_import_simplebox_missing_raises_actionable(monkeypatch: pytest.MonkeyPatch) -> None:
     _no_boxlite(monkeypatch)
-    with pytest.raises(ImportError, match=r"pip install boxlite"):
+    with pytest.raises(ImportError, match=r"deerflow-harness\[boxlite\]"):
         _import_simplebox()
 
 
@@ -34,7 +102,7 @@ def test_acquire_without_boxlite_raises_and_shuts_down_cleanly(monkeypatch: pyte
 
     provider = BoxliteProvider()
     try:
-        with pytest.raises(ImportError, match=r"pip install boxlite"):
+        with pytest.raises(ImportError, match=r"deerflow-harness\[boxlite\]"):
             provider.acquire("thread-1", user_id="u")
     finally:
         provider.shutdown()  # must not raise even though no box was ever created
@@ -70,3 +138,573 @@ def test_execute_command_rejects_invalid_env_key() -> None:
     box = BoxliteBox("box-id", box=object(), run=_fail_run)
     with pytest.raises(ValueError, match=r"POSIX"):
         box.execute_command("echo hi", env={"BAD KEY": "x"})
+
+
+def test_execute_command_forwards_timeout_to_sdk_and_loop_runner() -> None:
+    """Command timeout must bound both BoxLite exec and the loop bridge future."""
+    run_timeouts: list[float | None] = []
+
+    def _recording_run(coro, *, timeout=None):
+        run_timeouts.append(timeout)
+        return asyncio.run(coro)
+
+    fake = _FakeBox(name="box-id")
+    box = BoxliteBox("box-id", box=fake, run=_recording_run)
+
+    output = box.execute_command("echo ok", timeout=5)
+
+    assert "ok" in output
+    assert fake._exec_history[-1] == (("sh", "-lc", "echo ok"), None, 5)
+    assert run_timeouts == [5]
+
+
+def test_sandbox_id_deterministic(monkeypatch):
+    """_sandbox_id produces the same id for the same inputs."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    provider = BoxliteProvider()
+    id1 = provider._sandbox_id("thread-1", "user-a")
+    id2 = provider._sandbox_id("thread-1", "user-a")
+    assert id1 == id2
+    assert len(id1) == 8
+
+
+def test_sandbox_id_different_users(monkeypatch):
+    """Different users produce different ids for the same thread."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    provider = BoxliteProvider()
+    id_a = provider._sandbox_id("thread-1", "user-a")
+    id_b = provider._sandbox_id("thread-1", "user-b")
+    assert id_a != id_b
+
+
+def test_sandbox_id_different_threads(monkeypatch):
+    """Different threads produce different ids for the same user."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    provider = BoxliteProvider()
+    id_a = provider._sandbox_id("thread-1", "user-a")
+    id_b = provider._sandbox_id("thread-2", "user-a")
+    assert id_a != id_b
+
+
+def test_idle_timeout_zero_is_preserved_and_disables_reaper(monkeypatch):
+    """idle_timeout=0 is a valid config value and disables the reaper thread."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"idle_timeout": 0}),
+    )
+
+    provider = BoxliteProvider()
+
+    assert provider._config["idle_timeout"] == 0
+    assert provider._idle_checker_thread is None
+    provider.shutdown()
+
+
+def test_create_box_passes_prefixed_sandbox_id_as_name(monkeypatch):
+    """_create_box gives BoxLite a DeerFlow-owned name prefix."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    # Inject fake SimpleBox and fake loop runner
+    created_boxes = []
+
+    class _RecordingBox(_FakeBox):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            created_boxes.append(kwargs)
+
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _RecordingBox,
+    )
+
+    provider = BoxliteProvider()
+    # Replace _loop.run with our sync runner
+    provider._loop.run = _fake_run
+
+    box = provider._create_box("test-sandbox-id")
+    assert len(created_boxes) == 1
+    assert created_boxes[0]["name"] == "deer-flow-boxlite-test-sandbox-id"
+    assert box.id == "test-sandbox-id"
+
+
+def test_startup_reconciliation_adopts_prefixed_existing_boxes(monkeypatch):
+    """Existing DeerFlow-named BoxLite boxes are adopted into the warm pool."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    stopped: list[str] = []
+
+    class _NativeBox:
+        def stop(self):
+            stopped.append("adopted")
+
+    class _Runtime:
+        def start(self):
+            return self
+
+        def stop(self):
+            pass
+
+        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):
+            if name == "deer-flow-boxlite-adopted":
+                return _NativeBox()
+            raise AssertionError(f"unexpected box lookup: {name}")
+
+    class _Boxlite:
+        @staticmethod
+        def default():
+            return _Runtime()
+
+    monkeypatch.setattr("deerflow.community.boxlite.provider._import_sync_boxlite_runtime", lambda: _Boxlite)
+
+    provider = BoxliteProvider()
+
+    assert list(provider._warm_pool) == ["adopted"]
+    adopted_box = provider._warm_pool["adopted"][0]
+    assert adopted_box.id == "adopted"
+
+    provider.shutdown()
+    assert stopped == ["adopted"]
+
+
+def test_release_parks_in_warm_pool(monkeypatch):
+    """After release, box is in warm pool, not destroyed."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    # Acquire a box
+    sid = provider.acquire("thread-1", user_id="u1")
+
+    # Verify box is active
+    assert sid in provider._boxes
+    assert sid not in provider._warm_pool
+
+    # Release
+    provider.release(sid)
+
+    # Verify box is in warm pool, not active
+    assert sid not in provider._boxes
+    assert sid in provider._warm_pool
+    box, ts = provider._warm_pool[sid]
+    assert isinstance(box, BoxliteBox)
+    assert not box._box._stopped  # VM not destroyed
+
+
+def test_acquire_reclaims_from_warm_pool(monkeypatch):
+    """acquire reclaims a warm pool box for the same thread."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    # First acquire → create
+    sid1 = provider.acquire("thread-1", user_id="u1")
+    provider.release(sid1)
+
+    # Second acquire → should reclaim from warm pool
+    sid2 = provider.acquire("thread-1", user_id="u1")
+    assert sid1 == sid2  # Same deterministic ID
+    assert sid2 in provider._boxes
+    assert sid2 not in provider._warm_pool
+
+
+def test_acquire_different_threads_dont_reclaim_each_other(monkeypatch):
+    """Thread A's box can't be reclaimed by thread B."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    sid_a = provider.acquire("thread-a", user_id="u1")
+    provider.release(sid_a)
+
+    # Thread B acquires — should NOT get thread A's box
+    sid_b = provider.acquire("thread-b", user_id="u1")
+    assert sid_b != sid_a  # Different deterministic ID
+    assert sid_a in provider._warm_pool  # A's box still in warm pool
+    assert sid_b in provider._boxes  # B's box is new
+
+
+def test_warm_pool_reclaim_failed_health_check_creates_new(monkeypatch):
+    """Dead warm pool box is evicted and a new one created."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    sid1 = provider.acquire("thread-1", user_id="u1")
+    provider.release(sid1)
+    assert sid1 in provider._warm_pool
+
+    # Corrupt the warm pool box: close it so health check fails
+    box, _ = provider._warm_pool[sid1]
+    box.close()  # Stop VM, marks _closed=True
+
+    # Re-acquire — health check should fail on the dead box
+    # A new box is created with the same deterministic ID
+    sid2 = provider.acquire("thread-1", user_id="u1")
+    assert sid2 == sid1  # Same deterministic ID
+    assert sid2 in provider._boxes
+
+
+def test_concurrent_same_thread_acquire_creates_one_box(monkeypatch):
+    """Concurrent acquires for one thread serialize before creating a named box."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+    original_create_box = provider._create_box
+    create_started = threading.Event()
+    created: list[str] = []
+
+    def slow_create_box(sandbox_id: str) -> BoxliteBox:
+        create_started.set()
+        time.sleep(0.1)
+        created.append(sandbox_id)
+        return original_create_box(sandbox_id)
+
+    provider._create_box = slow_create_box  # type: ignore[method-assign]
+    results: list[str] = []
+
+    def acquire() -> None:
+        results.append(provider.acquire("thread-1", user_id="u1"))
+
+    first = threading.Thread(target=acquire)
+    second = threading.Thread(target=acquire)
+    first.start()
+    assert create_started.wait(timeout=2)
+    second.start()
+    first.join(timeout=2)
+    second.join(timeout=2)
+
+    assert len(results) == 2
+    assert results[0] == results[1]
+    assert len(created) == 1
+    assert results[0] in provider._boxes
+    provider.shutdown()
+
+
+def test_release_during_shutdown_closes_instead_of_reparking(monkeypatch):
+    """release() must not park a VM after shutdown has begun."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    sid = provider.acquire("thread-1", user_id="u1")
+    box = provider._boxes[sid]
+    with provider._lock:
+        provider._shutdown_called = True
+
+    provider.release(sid)
+
+    assert sid not in provider._boxes
+    assert sid not in provider._warm_pool
+    assert box._closed
+    provider._loop.close()
+
+
+def test_reset_parks_running_resources_for_later_cleanup(monkeypatch):
+    """reset() stops thread reuse but leaves VMs tracked for cleanup."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    sid_active = provider.acquire("thread-active", user_id="u1")
+    sid_warm = provider.acquire("thread-warm", user_id="u1")
+    provider.release(sid_warm)
+    active_box = provider._boxes[sid_active]
+    warm_box = provider._warm_pool[sid_warm][0]
+    checker_thread = provider._idle_checker_thread
+    loop_thread = provider._loop._thread
+
+    provider.reset()
+
+    assert provider._boxes == {}
+    assert provider._warm_pool[sid_active][0] is active_box
+    assert provider._warm_pool[sid_warm][0] is warm_box
+    assert provider._thread_boxes == {}
+    assert provider._acquire_locks == {}
+    assert not active_box._closed
+    assert not warm_box._closed
+    assert not provider._shutdown_called
+    assert not provider._idle_checker_stop.is_set()
+    assert checker_thread is not None
+    assert checker_thread.is_alive()
+    assert loop_thread.is_alive()
+
+    provider.shutdown()
+    assert active_box._closed
+    assert warm_box._closed
+    assert provider._idle_checker_stop.is_set()
+    assert not checker_thread.is_alive()
+
+
+def test_reset_parked_resources_are_reaped_after_idle_timeout(monkeypatch):
+    """VMs parked by reset remain visible to warm-pool idle cleanup."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    sid_active = provider.acquire("thread-active", user_id="u1")
+    sid_warm = provider.acquire("thread-warm", user_id="u1")
+    provider.release(sid_warm)
+    active_box = provider._boxes[sid_active]
+    warm_box = provider._warm_pool[sid_warm][0]
+
+    provider.reset()
+
+    provider._warm_pool[sid_active] = (active_box, time.time() - 9999)
+    provider._warm_pool[sid_warm] = (warm_box, time.time() - 9999)
+    provider._reap_expired_warm(idle_timeout=1)
+
+    assert provider._warm_pool == {}
+    assert active_box._closed
+    assert warm_box._closed
+    provider.shutdown()
+
+
+# ── Task 6: Idle reaper ───────────────────────────────────────────────
+
+
+def test_idle_reaper_destroys_expired_warm_boxes(monkeypatch):
+    """Idle reaper daemon destroys warm pool boxes that exceed the idle timeout."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    # Use a very short check interval so the reaper runs quickly
+    monkeypatch.setattr(BoxliteProvider, "IDLE_CHECK_INTERVAL", 0.1)
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    # Acquire and release a box into the warm pool
+    sid = provider.acquire("thread-1", user_id="u1")
+    provider.release(sid)
+
+    assert sid in provider._warm_pool
+
+    # Backdate the warm-pool timestamp so it appears long-expired
+    warm_box = provider._warm_pool[sid][0]
+    provider._warm_pool[sid] = (warm_box, time.time() - 9999)
+
+    # Wait long enough for the reaper to detect and destroy it
+    time.sleep(0.3)
+
+    # Box should be gone from warm pool and closed
+    assert sid not in provider._warm_pool
+    assert warm_box._closed
+
+    provider.shutdown()
+
+
+# ── Task 7: Replica enforcement ───────────────────────────────────────
+
+
+def test_replica_enforcement_evicts_oldest_warm(monkeypatch):
+    """When warm pool exceeds replica limit, the oldest box is evicted."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"replicas": 2}),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    # Fill warm pool with 2 boxes from different threads
+    sid_a = provider.acquire("thread-a", user_id="u1")
+    provider.release(sid_a)
+
+    sid_b = provider.acquire("thread-b", user_id="u1")
+    provider.release(sid_b)
+
+    assert len(provider._warm_pool) == 2
+    assert sid_a in provider._warm_pool
+    assert sid_b in provider._warm_pool
+
+    # Make sid_a definitely older by backdating its timestamp
+    box_a = provider._warm_pool[sid_a][0]
+    provider._warm_pool[sid_a] = (box_a, time.time() - 100)
+    # Refresh sid_b's timestamp so it's newer
+    box_b = provider._warm_pool[sid_b][0]
+    provider._warm_pool[sid_b] = (box_b, time.time())
+
+    # Acquiring a third thread triggers replica enforcement:
+    # warm pool count (2) >= replicas (2) → evict oldest (sid_a)
+    sid_c = provider.acquire("thread-c", user_id="u1")
+
+    # Oldest (sid_a) should be evicted (gone from warm pool, closed)
+    assert sid_a not in provider._warm_pool
+    assert box_a._closed
+    # Newer (sid_b) should remain in warm pool
+    assert sid_b in provider._warm_pool
+    # New box (sid_c) should be active
+    assert sid_c in provider._boxes
+    assert sid_c not in provider._warm_pool
+
+    provider.shutdown()
+
+
+def test_replica_enforcement_counts_active_and_warm(monkeypatch):
+    """replicas caps active + warm boxes, not warm boxes alone."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"replicas": 2}),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    sid_active = provider.acquire("thread-active", user_id="u1")
+    sid_warm = provider.acquire("thread-warm", user_id="u1")
+    provider.release(sid_warm)
+    warm_box = provider._warm_pool[sid_warm][0]
+
+    sid_new = provider.acquire("thread-new", user_id="u1")
+
+    assert sid_active in provider._boxes
+    assert sid_new in provider._boxes
+    assert sid_warm not in provider._warm_pool
+    assert warm_box._closed
+    provider.shutdown()
+
+
+# ── Task 8: Shutdown and reset including warm pool ────────────────────
+
+
+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(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config(),
+    )
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider._import_simplebox",
+        lambda: _FakeBox,
+    )
+
+    provider = BoxliteProvider()
+    provider._loop.run = _fake_run
+
+    # Create one active box (thread-1) and one warm pool box (thread-2 released)
+    sid_active = provider.acquire("thread-1", user_id="u1")
+    sid_warm = provider.acquire("thread-2", user_id="u1")
+    provider.release(sid_warm)
+
+    assert sid_active in provider._boxes
+    assert sid_warm in provider._warm_pool
+
+    # Get box references before shutdown
+    box_active = provider._boxes[sid_active]
+    box_warm = provider._warm_pool[sid_warm][0]
+
+    # Remember the idle checker thread
+    checker_thread = provider._idle_checker_thread
+
+    provider.shutdown()
+
+    # Idle checker should be stopped
+    assert provider._idle_checker_stop.is_set()
+    assert checker_thread is not None
+    assert not checker_thread.is_alive()
+
+    # All boxes (active + warm) should be closed
+    assert box_active._closed
+    assert box_warm._closed
+
+    # All collections should be empty
+    assert len(provider._boxes) == 0
+    assert len(provider._warm_pool) == 0
+    assert len(provider._thread_boxes) == 0
diff --git a/backend/tests/test_harness_packaging.py b/backend/tests/test_harness_packaging.py
new file mode 100644
index 000000000..1e80ae5ae
--- /dev/null
+++ b/backend/tests/test_harness_packaging.py
@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+
+
+def test_boxlite_is_optional_harness_dependency() -> None:
+    """BoxLite should not make core harness installs platform-dependent."""
+    pyproject_path = Path(__file__).resolve().parents[1] / "packages" / "harness" / "pyproject.toml"
+    pyproject = tomllib.loads(pyproject_path.read_text(encoding="utf-8"))
+
+    core_dependencies = pyproject["project"]["dependencies"]
+    optional_dependencies = pyproject["project"]["optional-dependencies"]
+
+    assert not any(dep.startswith("boxlite") for dep in core_dependencies)
+    assert any(dep.startswith("boxlite>=0.9.7") for dep in optional_dependencies["boxlite"])
diff --git a/backend/tests/test_warm_pool_lifecycle.py b/backend/tests/test_warm_pool_lifecycle.py
new file mode 100644
index 000000000..ae917b3c1
--- /dev/null
+++ b/backend/tests/test_warm_pool_lifecycle.py
@@ -0,0 +1,95 @@
+"""Unit tests for shared warm-pool lifecycle mechanics."""
+
+from __future__ import annotations
+
+import threading
+import time
+from typing import Any
+
+from deerflow.community.warm_pool_lifecycle import DEFAULT_IDLE_TIMEOUT, DEFAULT_REPLICAS, WarmPoolLifecycleMixin
+
+
+class _Provider(WarmPoolLifecycleMixin[str]):
+    _idle_checker_thread_name = "test-warm-pool-reaper"
+
+    def __init__(self, *, replicas: int = DEFAULT_REPLICAS, idle_timeout: float = DEFAULT_IDLE_TIMEOUT, active_count: int = 0) -> None:
+        self._lock = threading.Lock()
+        self._warm_pool: dict[str, tuple[str, float]] = {}
+        self._config: dict[str, Any] = {"replicas": replicas, "idle_timeout": idle_timeout}
+        self._idle_checker_stop = threading.Event()
+        self._idle_checker_thread: threading.Thread | None = None
+        self.active_count = active_count
+        self.destroyed: list[tuple[str, str, str]] = []
+
+    def _active_count_locked(self) -> int:
+        return self.active_count
+
+    def _destroy_warm_entry(self, sandbox_id: str, entry: str, *, reason: str) -> None:
+        self.destroyed.append((sandbox_id, entry, reason))
+
+
+def test_replica_count_includes_active_and_warm_entries() -> None:
+    provider = _Provider(replicas=2, active_count=1)
+    provider._warm_pool["warm-1"] = ("entry-1", time.time())
+
+    assert provider._replica_count() == (2, 2)
+
+
+def test_evict_oldest_warm_removes_and_destroys_oldest_entry() -> None:
+    provider = _Provider()
+    provider._warm_pool["new"] = ("entry-new", 200.0)
+    provider._warm_pool["old"] = ("entry-old", 100.0)
+
+    evicted = provider._evict_oldest_warm()
+
+    assert evicted == "old"
+    assert "old" not in provider._warm_pool
+    assert "new" in provider._warm_pool
+    assert provider.destroyed == [("old", "entry-old", "replica_enforcement")]
+
+
+def test_evict_oldest_warm_returns_none_when_pool_empty() -> None:
+    provider = _Provider()
+
+    assert provider._evict_oldest_warm() is None
+    assert provider.destroyed == []
+
+
+def test_reap_expired_warm_destroys_only_expired_entries() -> None:
+    provider = _Provider()
+    now = time.time()
+    provider._warm_pool["expired"] = ("entry-expired", now - 100)
+    provider._warm_pool["fresh"] = ("entry-fresh", now)
+
+    provider._reap_expired_warm(idle_timeout=10)
+
+    assert "expired" not in provider._warm_pool
+    assert "fresh" in provider._warm_pool
+    assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")]
+
+
+def test_reap_expired_warm_noops_when_timeout_disabled() -> None:
+    provider = _Provider(idle_timeout=0)
+    provider._warm_pool["expired"] = ("entry-expired", time.time() - 100)
+
+    provider._reap_expired_warm(idle_timeout=0)
+
+    assert "expired" in provider._warm_pool
+    assert provider.destroyed == []
+
+
+def test_start_idle_checker_uses_monkeypatchable_interval(monkeypatch) -> None:
+    provider = _Provider(idle_timeout=0.01)
+    monkeypatch.setattr(_Provider, "IDLE_CHECK_INTERVAL", 0.01)
+    provider._warm_pool["expired"] = ("entry-expired", time.time() - 10)
+
+    provider._start_idle_checker()
+    deadline = time.time() + 1
+    while "expired" in provider._warm_pool and time.time() < deadline:
+        time.sleep(0.01)
+    provider._stop_idle_checker()
+
+    assert "expired" not in provider._warm_pool
+    assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")]
+    assert provider._idle_checker_thread is not None
+    assert not provider._idle_checker_thread.is_alive()
diff --git a/backend/uv.lock b/backend/uv.lock
index 0d9df677d..6c6bdf153 100644
--- a/backend/uv.lock
+++ b/backend/uv.lock
@@ -485,6 +485,17 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/95/c1/84fc6811122f54b20de2e5afb312ee07a3a47a328755587d1e505475239b/blockbuster-1.5.26-py3-none-any.whl", hash = "sha256:f8e53fb2dd4b6c6ec2f04907ddbd063ca7cd1ef587d24448ef4e50e81e3a79bb", size = 13226, upload-time = "2025-12-05T10:43:48.778Z" },
 ]
 
+[[package]]
+name = "boxlite"
+version = "0.9.7"
+source = { registry = "https://pypi.org/simple" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/58/77/34fe448e7a1411b95c33df673305d64bacdfb031a5faae9faec82a32eaf8/boxlite-0.9.7-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:715ba342a80432279115ad41edf42b92993134402a7f4c73f23eb1327aa7ec29", size = 35518202, upload-time = "2026-07-01T12:55:41.757Z" },
+    { url = "https://files.pythonhosted.org/packages/02/2f/8ebf72c81afd07b3779ce206e85b32cf7b7baed98b811e966508ab9509cc/boxlite-0.9.7-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:01a2dd8b871213beb037a0c332bd94d9568aea7db0c1d39385ae789b73610055", size = 37616271, upload-time = "2026-07-01T12:55:45.01Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/32/6d9d8e9dca8f6e387dfe874a6872b49529b7795b75e814b2a75015a66036/boxlite-0.9.7-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:2281d4613af8f56e05856883aac2cbaf0028d50895a67299cd4932585661f13a", size = 33781989, upload-time = "2026-07-01T12:55:47.944Z" },
+    { url = "https://files.pythonhosted.org/packages/e0/79/068b8a7de1fc9724d89ca7086754f0e6c6c80f8c74f9d95267006c49fe33/boxlite-0.9.7-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:b8e68f2efe493112445f3150adac74de592974ba9c05025440e0ddf54870bc53", size = 35519183, upload-time = "2026-07-01T12:55:50.896Z" },
+]
+
 [[package]]
 name = "bracex"
 version = "2.7"
@@ -885,6 +896,9 @@ dependencies = [
 ]
 
 [package.optional-dependencies]
+boxlite = [
+    { name = "boxlite" },
+]
 ollama = [
     { name = "langchain-ollama" },
 ]
@@ -911,6 +925,7 @@ requires-dist = [
     { name = "aiosqlite", specifier = ">=0.19" },
     { name = "alembic", specifier = ">=1.13" },
     { name = "asyncpg", marker = "extra == 'postgres'", specifier = ">=0.29" },
+    { name = "boxlite", marker = "extra == 'boxlite'", specifier = ">=0.9.7" },
     { name = "croniter", specifier = ">=6.0.0" },
     { name = "cryptography", specifier = ">=48.0.1" },
     { name = "ddgs", specifier = ">=9.10.0" },
@@ -950,7 +965,7 @@ requires-dist = [
     { name = "textual", marker = "extra == 'tui'", specifier = ">=0.80" },
     { name = "tiktoken", specifier = ">=0.8.0" },
 ]
-provides-extras = ["redis", "tui", "groundroute", "ollama", "postgres", "pymupdf"]
+provides-extras = ["tui", "groundroute", "ollama", "postgres", "redis", "pymupdf", "boxlite"]
 
 [[package]]
 name = "defusedxml"
diff --git a/config.example.yaml b/config.example.yaml
index 733359be5..b7d3c04ec 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -1086,7 +1086,32 @@ sandbox:
 #   #   API_KEY: $MY_API_KEY        # Reads from host's MY_API_KEY env var
 #   #   DATABASE_URL: $DATABASE_URL  # Reads from host's DATABASE_URL env var
 
-# Option 3: Provisioner-managed AIO Sandbox (docker-compose-dev)
+# Option 3: BoxLite micro-VM Sandbox
+# Runs each sandbox as a BoxLite micro-VM. Released boxes stay in an in-process
+# warm pool and can be reclaimed by the same user/thread without a cold start.
+# Requires the boxlite runtime and host virtualization support (KVM on Linux,
+# Hypervisor.framework on macOS).
+# sandbox:
+#   use: deerflow.community.boxlite:BoxliteProvider
+#   image: python:3.12-slim
+#
+#   # Optional: Per-box memory and CPU limits.
+#   # memory_mib: 1024
+#   # cpus: 2
+#
+#   # Optional: Maximum active + warm BoxLite VMs per gateway process (default: 3).
+#   # Active boxes are never evicted; only warm-pool boxes are stopped to make room.
+#   # replicas: 3
+#
+#   # Optional: Seconds before an idle warm-pool VM is stopped (default: 600).
+#   # Set to 0 to keep warm VMs until shutdown or replica eviction.
+#   # idle_timeout: 600
+#
+#   # Optional: Environment variables to inject into every command.
+#   # environment:
+#   #   PYTHONUNBUFFERED: "1"
+#
+# Option 4: Provisioner-managed AIO Sandbox (docker-compose-dev)
 # Each sandbox_id gets a dedicated Pod in k3s, managed by the provisioner.
 # Recommended for production or advanced users who want better isolation and scalability.:
 # sandbox: