diff --git a/.gitignore b/.gitignore
index 0076848e0..6dba53b46 100644
--- a/.gitignore
+++ b/.gitignore
@@ -20,6 +20,12 @@ __pycache__/
 .venv
 venv/
 
+# Benchmark outputs
+bench_results.jsonl
+bench_optimized.jsonl
+results.jsonl
+backend/scripts/benchmark/*.jsonl
+
 # Environment variables
 .env
 
diff --git a/backend/packages/harness/deerflow/community/boxlite/README.md b/backend/packages/harness/deerflow/community/boxlite/README.md
index bedecba90..5ddadb239 100644
--- a/backend/packages/harness/deerflow/community/boxlite/README.md
+++ b/backend/packages/harness/deerflow/community/boxlite/README.md
@@ -13,12 +13,15 @@ the default AIO Docker sandbox in
 ```yaml
 sandbox:
   use: deerflow.community.boxlite:BoxliteProvider
-  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
+  image: python:3.12-slim         # any OCI image (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
+  health_check_skip_seconds: 0.0  # optional low-latency mode: skip reclaim
+                                  #   health checks for recent releases; 0 keeps
+                                  #   reliability-first validation (default: 0.0)
+  environment:                    # injected into every command
     PYTHONUNBUFFERED: "1"
 ```
 
diff --git a/backend/packages/harness/deerflow/community/boxlite/box.py b/backend/packages/harness/deerflow/community/boxlite/box.py
index b923c29b2..2f60dc1ee 100644
--- a/backend/packages/harness/deerflow/community/boxlite/box.py
+++ b/backend/packages/harness/deerflow/community/boxlite/box.py
@@ -57,6 +57,23 @@ class BoxliteBox(Sandbox):
             per-call ``env`` (request-scoped secrets).
     """
 
+    TERMINAL_ERROR_MARKERS = (
+        "vsock",
+        "disconnected",
+        "broken pipe",
+        "connection reset",
+        "connection refused",
+        "no such box",
+        "box has been stopped",
+        "engine reported an error",
+    )
+    RETRYABLE_ERROR_MARKERS = (
+        "transport not ready",
+        "retry later",
+        "temporarily unavailable",
+        "resource busy",
+    )
+
     def __init__(
         self,
         id: str,
@@ -64,25 +81,56 @@ class BoxliteBox(Sandbox):
         run: Callable[..., T],
         *,
         default_env: dict[str, str] | None = None,
+        on_terminal_failure: Callable[[str, str], None] | None = None,
     ) -> None:
         super().__init__(id)
         self._box = box
         self._run = run
         self._default_env = dict(default_env or {})
+        self._on_terminal_failure = on_terminal_failure
         self._lock = threading.Lock()
         self._closed = False
 
+    @classmethod
+    def _is_terminal_box_failure(cls, error: Exception) -> bool:
+        if isinstance(error, (BrokenPipeError, ConnectionError, EOFError)):
+            return True
+        if not isinstance(error, RuntimeError | OSError):
+            return False
+        msg = str(error).lower()
+        if any(marker in msg for marker in cls.RETRYABLE_ERROR_MARKERS):
+            return False
+        return any(marker in msg for marker in cls.TERMINAL_ERROR_MARKERS)
+
     # ── bridge helpers ──────────────────────────────────────────────────
 
-    def _exec(self, *argv: str, env: dict[str, str] | None = None):
-        with self._lock:
-            if self._closed:
-                raise RuntimeError("sandbox has been closed")
-            box = self._box
-        return self._run(box.exec(*argv, env=env))
+    def _exec(
+        self,
+        *argv: str,
+        env: dict[str, str] | None = None,
+        timeout: float | None = None,
+    ):
+        try:
+            with self._lock:
+                if self._closed:
+                    raise RuntimeError("sandbox has been closed")
+                box = self._box
+            return self._run(box.exec(*argv, env=env, timeout=timeout), timeout=timeout)
+        except Exception as e:
+            if self._on_terminal_failure is not None and self._is_terminal_box_failure(e):
+                try:
+                    self._on_terminal_failure(self.id, str(e))
+                except Exception:
+                    logger.exception("Terminal BoxLite failure callback errored for %s", self.id)
+            raise
 
-    def _sh(self, script: str, env: dict[str, str] | None = None):
-        return self._exec("sh", "-lc", script, env=env)
+    def _sh(
+        self,
+        script: str,
+        env: dict[str, str] | None = None,
+        timeout: float | None = None,
+    ):
+        return self._exec("sh", "-lc", script, env=env, timeout=timeout)
 
     def close(self) -> None:
         with self._lock:
@@ -94,6 +142,11 @@ class BoxliteBox(Sandbox):
         except Exception as e:
             logger.warning("Error stopping BoxLite box %s: %s", self.id, e)
 
+    @property
+    def is_closed(self) -> bool:
+        with self._lock:
+            return self._closed
+
     # ── path safety (mirrors community/e2b_sandbox) ─────────────────────
 
     @staticmethod
@@ -131,13 +184,11 @@ class BoxliteBox(Sandbox):
         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
+        if self.is_closed:
+            return "Error: sandbox has been closed"
         merged_env = {**self._default_env, **(env or {})} or None
-        with self._lock:
-            if self._closed:
-                return "Error: sandbox has been closed"
-            box = self._box
         try:
-            result = self._run(box.exec("sh", "-lc", command, env=merged_env, timeout=timeout), timeout=timeout)
+            result = self._exec("sh", "-lc", command, env=merged_env, 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 33c62a259..fa0981be4 100644
--- a/backend/packages/harness/deerflow/community/boxlite/provider.py
+++ b/backend/packages/harness/deerflow/community/boxlite/provider.py
@@ -176,6 +176,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
         self._boxes: dict[str, BoxliteBox] = {}
         self._thread_boxes: dict[tuple[str, str], str] = {}
         self._warm_pool: dict[str, tuple[BoxliteBox, float]] = {}
+        self._skip_health_check_warm_ids: set[str] = set()
         self._acquire_locks: dict[str, threading.Lock] = {}
         self._idle_checker_stop = threading.Event()
         self._idle_checker_thread: threading.Thread | None = None
@@ -196,6 +197,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
         # (which raises on a missing var), so the environment dict is used as-is.
         replicas = _opt("replicas")
         idle_timeout = _opt("idle_timeout")
+        health_check_skip_seconds = _opt("health_check_skip_seconds")
         return {
             "image": _opt("image") or DEFAULT_IMAGE,
             "memory_mib": _opt("memory_mib"),
@@ -203,6 +205,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             "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,
+            "health_check_skip_seconds": float(health_check_skip_seconds if health_check_skip_seconds is not None else 0.0),
         }
 
     @staticmethod
@@ -241,6 +244,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
 
     def _destroy_warm_entry(self, sandbox_id: str, entry: BoxliteBox, *, reason: str) -> None:
         """Close a removed warm-pool entry and log with context."""
+        with self._lock:
+            self._skip_health_check_warm_ids.discard(sandbox_id)
         try:
             entry.close()
             if reason == "idle_timeout":
@@ -257,6 +262,24 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             else:
                 logger.warning("Error closing BoxLite box %s (reason=%s): %s", sandbox_id, reason, e)
 
+    def _invalidate_box(self, sandbox_id: str, reason: str) -> None:
+        """Destroy and deregister a box after a terminal command-path failure."""
+        box_to_close: BoxliteBox | None = None
+        with self._lock:
+            active_box = self._boxes.pop(sandbox_id, None)
+            warm_entry = self._warm_pool.pop(sandbox_id, None)
+            self._skip_health_check_warm_ids.discard(sandbox_id)
+            for key in [k for k, sid in self._thread_boxes.items() if sid == sandbox_id]:
+                self._thread_boxes.pop(key, None)
+            box_to_close = active_box or (warm_entry[0] if warm_entry is not None else None)
+
+        if box_to_close is None:
+            logger.warning("BoxLite box %s failed terminally but was not tracked: %s", sandbox_id, reason)
+            return
+
+        logger.warning("Invalidating BoxLite box %s after terminal failure: %s", sandbox_id, reason)
+        box_to_close.close()
+
     def _reconcile_orphans(self) -> None:
         """Adopt DeerFlow-owned BoxLite boxes left by a previous provider/process.
 
@@ -306,7 +329,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
                 box_runtime.stop()
                 continue
 
-            wrapped = BoxliteBox(sandbox_id, _SyncBoxAdapter(box_runtime, box), _run_sync_adapter, default_env=self._config["environment"])
+            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()
@@ -371,7 +394,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
 
         box = self._loop.run(_make())
         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"])
+        return BoxliteBox(sandbox_id, box, self._loop.run, default_env=self._config["environment"], on_terminal_failure=self._invalidate_box)
 
     def get(self, sandbox_id: str) -> Sandbox | None:
         with self._lock:
@@ -393,8 +416,10 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
                 return
             if self._shutdown_called:
                 close_box = box
+                self._skip_health_check_warm_ids.discard(sandbox_id)
             else:
                 self._warm_pool[sandbox_id] = (box, time.time())
+                self._skip_health_check_warm_ids.add(sandbox_id)
 
         if close_box is not None:
             close_box.close()
@@ -406,11 +431,44 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
         """Try to reclaim a warm-pool box by sandbox_id.
 
         Returns sandbox_id on success, None if not found or dead.
+
+        Only boxes that *this provider instance* placed in the warm pool via
+        ``release()`` may skip the health check when reclaimed shortly after
+        release; startup-adopted/orphaned boxes always validate before reuse.
         """
+
         with self._lock:
             if sandbox_id not in self._warm_pool:
                 return None
-            box, _ = self._warm_pool[sandbox_id]
+            box, released_at = self._warm_pool[sandbox_id]
+            skip_eligible = sandbox_id in self._skip_health_check_warm_ids
+
+        skip_seconds = self._config.get("health_check_skip_seconds", 0.0)
+        if skip_eligible and skip_seconds > 0 and (time.time() - released_at) < skip_seconds:
+            # Recently released by this provider — promote directly without a
+            # health-check round trip, but never return an adapter that this
+            # process already knows is closed.
+            with self._lock:
+                warm_entry = self._warm_pool.pop(sandbox_id, None)
+                if warm_entry is None:
+                    return None  # Raced with another thread
+                self._skip_health_check_warm_ids.discard(sandbox_id)
+                box, _ = warm_entry
+                if box.is_closed:
+                    logger.warning("Warm-pool box %s was closed before skipped health check reclaim", sandbox_id)
+                    close_box = box
+                else:
+                    close_box = None
+                    self._boxes[sandbox_id] = box
+            if close_box is not None:
+                close_box.close()
+                return None
+            logger.debug(
+                "Reclaimed warm-pool box %s (skipped health check, age=%.1fs)",
+                sandbox_id,
+                time.time() - released_at,
+            )
+            return sandbox_id
 
         # Health check: run a simple command to verify the VM is alive
         try:
@@ -418,14 +476,16 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             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()
+                    warm_entry = self._warm_pool.pop(sandbox_id, None)
+                if warm_entry is not None:
+                    self._destroy_warm_entry(sandbox_id, warm_entry[0], reason="health_check_failed")
                 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()
+                warm_entry = self._warm_pool.pop(sandbox_id, None)
+            if warm_entry is not None:
+                self._destroy_warm_entry(sandbox_id, warm_entry[0], reason="health_check_failed")
             return None
 
         # Promote from warm pool to active
@@ -433,6 +493,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             warm_entry = self._warm_pool.pop(sandbox_id, None)
             if warm_entry is None:
                 return None  # Raced with another thread
+            self._skip_health_check_warm_ids.discard(sandbox_id)
             box, _ = warm_entry
             self._boxes[sandbox_id] = box
 
@@ -452,6 +513,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             now = time.time()
             for sandbox_id, box in self._boxes.items():
                 self._warm_pool.setdefault(sandbox_id, (box, now))
+                self._skip_health_check_warm_ids.discard(sandbox_id)
             self._boxes.clear()
             self._thread_boxes.clear()
             self._acquire_locks.clear()
@@ -471,6 +533,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
             self._warm_pool.clear()
             self._thread_boxes.clear()
             self._acquire_locks.clear()
+            self._skip_health_check_warm_ids.clear()
 
         for box in active + warm:
             try:
diff --git a/backend/packages/harness/deerflow/config/sandbox_config.py b/backend/packages/harness/deerflow/config/sandbox_config.py
index d2f0eb382..428335329 100644
--- a/backend/packages/harness/deerflow/config/sandbox_config.py
+++ b/backend/packages/harness/deerflow/config/sandbox_config.py
@@ -36,6 +36,9 @@ class SandboxConfig(BaseModel):
         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)
 
+    BoxliteProvider specific options:
+        health_check_skip_seconds: Optional reclaim-time skip window in seconds for recently released warm VMs. Default behavior is 0.0 = always validate before reuse.
+
     AioSandboxProvider specific options:
         port: Base port for sandbox containers (default: 8080)
         container_prefix: Prefix for container names (default: deer-flow-sandbox)
@@ -70,6 +73,11 @@ class SandboxConfig(BaseModel):
         default=None,
         description="Idle timeout in seconds before released warm sandboxes/VMs are stopped (default: 600 = 10 minutes). Set to 0 to disable.",
     )
+    health_check_skip_seconds: float | None = Field(
+        default=None,
+        ge=0,
+        description="BoxLite-only reclaim skip window in seconds for boxes recently released by this provider instance. Set to 0 to always validate before warm reuse.",
+    )
     mounts: list[VolumeMountConfig] = Field(
         default_factory=list,
         description="List of volume mounts to share directories between host and container",
diff --git a/backend/scripts/benchmark/bench_sandbox_provider.py b/backend/scripts/benchmark/bench_sandbox_provider.py
new file mode 100755
index 000000000..a38ac8c32
--- /dev/null
+++ b/backend/scripts/benchmark/bench_sandbox_provider.py
@@ -0,0 +1,897 @@
+#!/usr/bin/env python3
+"""Provider-agnostic sandbox benchmark.
+
+Measures acquire / run / release latency across providers, scenarios,
+workloads, and concurrency levels.  Outputs JSONL for aggregation.
+
+Usage::
+
+    python scripts/benchmark/bench_sandbox_provider.py \\
+        --provider boxlite \\
+        --scenario warm_same_thread \\
+        --workload noop \\
+        --iterations 50 \\
+        --concurrency 4 \\
+        --output results.jsonl
+
+    python scripts/benchmark/bench_sandbox_provider.py \\
+        --provider boxlite \\
+        --scenario cold_unique_thread \\
+        --no-warmpool \\
+        --iterations 30 \\
+        --output results.jsonl
+
+Providers
+---------
+``boxlite``       BoxLite micro-VM sandbox (requires ``pip install boxlite``).
+``aio-docker``    AIO Docker sandbox (requires Docker daemon + ``deerflow-harness`` extras).
+
+Scenarios
+---------
+``warm_same_thread``       Reuse one ``(user_id, thread_id)`` — warm pool hit after first turn.
+``cold_unique_thread``     Fresh ``thread_id`` per turn — never hits warm pool.
+``warm_miss_many_threads`` Rotate through N distinct threads — verifies isolation.
+``idle_timeout``           Release, sleep > timeout, re-acquire — verify reaper works.
+``replica_pressure``       Push past ``replicas`` — verify eviction only targets warm entries.
+
+Workloads
+---------
+``noop``          ``true`` — exposes acquire/release overhead.
+``python_small``  ``python -c "print(sum(range(100000)))"`` — typical agent code.
+``fs_1mb``        Write + read 1 MB file inside sandbox.
+``sleep_2s``      ``sleep 2`` — verifies timeout handling + active-box protection.
+``state_reuse``   Write state in turn N, verify it persists in turn N+1 (warm only).
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib
+import importlib.metadata
+import json
+import os
+import sys
+import threading
+import time
+import types
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+# ── Output schema ───────────────────────────────────────────────────────
+
+
+@dataclass
+class BenchResult:
+    provider: str
+    scenario: str
+    workload: str
+    iteration: int
+    concurrency: int
+    thread_id: str
+    user_id: str
+    acquire_ms: float
+    run_ms: float
+    release_ms: float
+    total_ms: float
+    warm_hit: bool | None = None
+    success: bool = True
+    error: str | None = None
+    # Provider config snapshot (written once per batch)
+    replicas: int | None = None
+    idle_timeout: float | None = None
+    health_check_skip_seconds: float | None = None
+    image: str | None = None
+    no_warmpool: bool = False
+
+
+# ── Workloads ───────────────────────────────────────────────────────────
+
+WORKLOADS: dict[str, str] = {
+    "noop": "true",
+    "python_small": 'python -c "print(sum(range(100000)))"',
+    "fs_1mb": """python - <<'PY'
+from pathlib import Path
+p = Path("/tmp/bench_file.txt")
+p.write_text("x" * 1024 * 1024)
+print(len(p.read_text()))
+PY""",
+    "sleep_2s": """python - <<'PY'
+import time
+time.sleep(2)
+print("done")
+PY""",
+}
+
+# state_reuse is a two-step workload; handled separately
+_STATE_WRITE = """python - <<'PY'
+from pathlib import Path
+Path("/tmp/warm_state.txt").write_text("benchmark-state-42")
+print("written")
+PY"""
+
+_STATE_READ = """python - <<'PY'
+from pathlib import Path
+print(Path("/tmp/warm_state.txt").read_text())
+PY"""
+
+
+# ── Provider factories ──────────────────────────────────────────────────
+
+
+def _stub_config(sandbox_attrs: dict[str, Any] | None = None) -> types.SimpleNamespace:
+    """Build a stub config namespace mimicking ``get_app_config()``."""
+    attrs = sandbox_attrs or {}
+    return types.SimpleNamespace(sandbox=types.SimpleNamespace(**attrs))
+
+
+@contextmanager
+def _patched_module_attr(module_name: str, attr_name: str, value: Any):
+    module = importlib.import_module(module_name)
+    original = getattr(module, attr_name)
+    setattr(module, attr_name, value)
+    try:
+        yield module
+    finally:
+        setattr(module, attr_name, original)
+
+
+def _boxlite_version() -> str | None:
+    try:
+        return importlib.metadata.version("boxlite")
+    except importlib.metadata.PackageNotFoundError:
+        return None
+
+
+def _chmod_boxlite_shims(boxes_dir: str) -> int:
+    fixed = 0
+    for shim in Path(boxes_dir).glob("*/bin/boxlite-shim"):
+        st = shim.stat()
+        if st.st_mode & 0o111:
+            continue
+        shim.chmod(st.st_mode | 0o111)
+        fixed += 1
+    return fixed
+
+
+def _create_box_with_097_shim_workaround(
+    create_box: Callable[[str], Any],
+    sandbox_id: str,
+    *,
+    boxes_dir: str,
+) -> Any:
+    try:
+        return create_box(sandbox_id)
+    except RuntimeError as exc:
+        version = _boxlite_version()
+        if version != "0.9.7":
+            raise RuntimeError(f"BoxLite benchmark shim workaround only supports boxlite 0.9.7; got {version!r}") from exc
+        fixed = _chmod_boxlite_shims(boxes_dir)
+        if fixed == 0:
+            raise
+        return create_box(sandbox_id)
+
+
+def _make_boxlite_provider(config: dict[str, Any]) -> tuple[Any, dict[str, Any]]:
+    """Create a BoxliteProvider with stub config; returns (provider, config_used).
+
+    On BoxLite 0.9.7 only, retries a failed create after fixing missing execute
+    bits on extracted ``boxlite-shim`` binaries under ``~/.boxlite/boxes``.
+    """
+    from deerflow.community.boxlite.provider import BoxliteProvider
+
+    sandbox_attrs = {
+        "image": config.get("image") or "python:3.12-slim",
+        "replicas": config.get("replicas", 3),
+        "idle_timeout": config.get("idle_timeout", 600),
+        "health_check_skip_seconds": config.get("health_check_skip_seconds", 0.0),
+    }
+    if "memory_mib" in config:
+        sandbox_attrs["memory_mib"] = config["memory_mib"]
+    if "cpus" in config:
+        sandbox_attrs["cpus"] = config["cpus"]
+    if "environment" in config:
+        sandbox_attrs["environment"] = config["environment"]
+
+    with _patched_module_attr(
+        "deerflow.community.boxlite.provider",
+        "get_app_config",
+        lambda: _stub_config(sandbox_attrs),
+    ):
+        provider = BoxliteProvider()
+
+    original_create_box = provider._create_box
+    boxes_dir = os.path.expanduser("~/.boxlite/boxes")
+
+    def _patched_create_box(self: Any, sandbox_id: str) -> Any:
+        return _create_box_with_097_shim_workaround(
+            original_create_box,
+            sandbox_id,
+            boxes_dir=boxes_dir,
+        )
+
+    provider._create_box = types.MethodType(_patched_create_box, provider)
+    return provider, sandbox_attrs
+
+
+def _make_aio_provider(config: dict[str, Any]) -> tuple[Any, dict[str, Any]]:
+    """Create an AioSandboxProvider with stub config."""
+    from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider
+
+    sandbox_attrs = {
+        "image": config.get("image"),
+        "port": config.get("port"),
+        "container_prefix": config.get("container_prefix"),
+        "replicas": config.get("replicas", 3),
+        "idle_timeout": config.get("idle_timeout", 600),
+        "mounts": config.get("mounts", []),
+        "environment": config.get("environment", {}),
+        "provisioner_url": config.get("provisioner_url", ""),
+    }
+
+    with _patched_module_attr(
+        "deerflow.community.aio_sandbox.aio_sandbox_provider",
+        "get_app_config",
+        lambda: _stub_config(sandbox_attrs),
+    ):
+        provider = AioSandboxProvider()
+    return provider, sandbox_attrs
+
+
+PROVIDER_FACTORIES: dict[str, Callable] = {
+    "boxlite": _make_boxlite_provider,
+    "aio-docker": _make_aio_provider,
+}
+
+
+# ── Warm-hit tracking ───────────────────────────────────────────────────
+_WARM_HIT_STATE = threading.local()
+
+
+def _install_warm_hit_tracking(provider: Any) -> None:
+    """Record warm-pool reclaims from inside the provider acquire path."""
+    if getattr(provider, "_bench_warm_hit_tracking_installed", False):
+        return
+
+    installed = False
+    for method_name in ("_reclaim_warm_pool", "_reclaim_warm_pool_sandbox"):
+        original = getattr(provider, method_name, None)
+        if original is None:
+            continue
+
+        def _wrapped(*args: Any, _original: Callable = original, **kwargs: Any):
+            result = _original(*args, **kwargs)
+            if result is not None:
+                _WARM_HIT_STATE.value = True
+            return result
+
+        setattr(provider, method_name, _wrapped)
+        installed = True
+
+    setattr(provider, "_bench_warm_hit_tracking_installed", installed)
+
+
+def _reset_warm_hit_tracking() -> None:
+    _WARM_HIT_STATE.value = False
+
+
+def _warm_hit_from_acquire() -> bool:
+    return bool(getattr(_WARM_HIT_STATE, "value", False))
+
+
+def _compute_sandbox_id(provider: Any, thread_id: str, user_id: str) -> str:
+    """Compute the deterministic sandbox_id the provider would use."""
+    if hasattr(provider, "_sandbox_id"):
+        return provider._sandbox_id(thread_id, user_id)
+    # Fallback: use the provider's own method or hash
+    import hashlib
+
+    return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:8]
+
+
+def _was_warm_hit(provider: Any, sandbox_id: str) -> bool:
+    """Check if the sandbox_id is currently in the provider's warm pool."""
+    with provider._lock:
+        return sandbox_id in provider._warm_pool
+
+
+def _evict_from_warm(provider: Any, sandbox_id: str) -> None:
+    """Forcibly remove and destroy a warm-pool entry (no-warmpool simulation)."""
+    with provider._lock:
+        entry = provider._warm_pool.pop(sandbox_id, None)
+    if entry is not None:
+        box, _ = entry
+        try:
+            box.close()
+        except Exception:
+            pass
+
+
+# ── Core benchmark runner ───────────────────────────────────────────────
+
+
+def _run_one_turn(
+    provider: Any,
+    provider_name: str,
+    scenario: str,
+    workload_name: str,
+    command: str,
+    iteration: int,
+    concurrency: int,
+    user_id: str,
+    thread_id: str,
+    no_warmpool: bool,
+    state_write_turn: bool = False,
+    expected_state: str | None = None,
+) -> BenchResult:
+    """Execute one acquire→run→release cycle and return a BenchResult."""
+    t0 = time.perf_counter()
+    sandbox_id = _compute_sandbox_id(provider, thread_id, user_id)
+
+    sid: str | None = None
+    warm_hit: bool | None = None
+    acquire_ms = 0.0
+    run_ms = 0.0
+    release_ms = 0.0
+    release_needed = False
+
+    try:
+        tracked_warm_hit = getattr(provider, "_bench_warm_hit_tracking_installed", False)
+        _reset_warm_hit_tracking()
+        if not tracked_warm_hit:
+            warm_hit = _was_warm_hit(provider, sandbox_id)
+
+        t_a = time.perf_counter()
+        sid = provider.acquire(thread_id, user_id=user_id)
+        t_b = time.perf_counter()
+        if tracked_warm_hit:
+            warm_hit = _warm_hit_from_acquire()
+        acquire_ms = (t_b - t_a) * 1000
+        release_needed = True
+
+        sandbox = provider.get(sid)
+        if sandbox is None:
+            raise RuntimeError(f"acquire returned {sid!r} but get() returned None")
+
+        cmd = command
+        if state_write_turn:
+            cmd = _STATE_WRITE
+        elif expected_state is not None:
+            cmd = _STATE_READ
+
+        t_c = time.perf_counter()
+        output = sandbox.execute_command(cmd, timeout=30)
+        t_d = time.perf_counter()
+        run_ms = (t_d - t_c) * 1000
+        if output.startswith("Error:"):
+            raise RuntimeError(output)
+
+        if expected_state is not None:
+            if expected_state.strip() not in output.strip():
+                raise RuntimeError(f"State reuse failed: expected {expected_state!r} in output, got {output.strip()!r}")
+
+        t_e = time.perf_counter()
+        provider.release(sid)
+        release_needed = False
+        t_f = time.perf_counter()
+        release_ms = (t_f - t_e) * 1000
+
+        if no_warmpool:
+            _evict_from_warm(provider, sid)
+
+        return BenchResult(
+            provider=provider_name,
+            scenario=scenario,
+            workload=workload_name,
+            iteration=iteration,
+            concurrency=concurrency,
+            thread_id=thread_id,
+            user_id=user_id,
+            acquire_ms=acquire_ms,
+            run_ms=run_ms,
+            release_ms=release_ms,
+            total_ms=(t_f - t0) * 1000,
+            warm_hit=warm_hit,
+            success=True,
+            no_warmpool=no_warmpool,
+        )
+
+    except Exception as exc:
+        release_error: str | None = None
+        if sid is not None and release_needed:
+            t_release = time.perf_counter()
+            try:
+                provider.release(sid)
+                if no_warmpool:
+                    _evict_from_warm(provider, sid)
+            except Exception as release_exc:
+                release_error = repr(release_exc)
+            finally:
+                release_ms = (time.perf_counter() - t_release) * 1000
+        error = str(exc) if str(exc).startswith("Error:") else repr(exc)
+        if release_error is not None:
+            error = f"{error}; release_error={release_error}"
+        return BenchResult(
+            provider=provider_name,
+            scenario=scenario,
+            workload=workload_name,
+            iteration=iteration,
+            concurrency=concurrency,
+            thread_id=thread_id,
+            user_id=user_id,
+            acquire_ms=acquire_ms,
+            run_ms=run_ms,
+            release_ms=release_ms,
+            total_ms=(time.perf_counter() - t0) * 1000,
+            warm_hit=warm_hit,
+            success=False,
+            error=error,
+            no_warmpool=no_warmpool,
+        )
+
+
+def _run_scenario(
+    provider: Any,
+    provider_name: str,
+    scenario: str,
+    workload_name: str,
+    iterations: int,
+    concurrency: int,
+    output_path: Path,
+    no_warmpool: bool,
+    config_used: dict[str, Any],
+    fault_inject_after: int | None = None,
+) -> list[BenchResult]:
+
+    command = WORKLOADS.get(workload_name, WORKLOADS["noop"])
+
+    # For state_reuse workload, run paired turns: write → read
+    is_state_reuse = workload_name == "state_reuse"
+
+    results: list[BenchResult] = []
+
+    def _run_one(i: int) -> BenchResult:
+        if scenario == "cold_unique_thread":
+            tid = f"cold-{i}"
+        elif scenario == "warm_same_thread":
+            tid = "warm-hit"
+        elif scenario == "warm_miss_many_threads":
+            tid = f"thread-{i % max(concurrency, 4)}"
+        elif scenario in ("idle_timeout", "replica_pressure"):
+            tid = f"warm-hit-{i % concurrency}"
+        else:
+            tid = f"default-{i}"
+
+        state_write = is_state_reuse and (i % 2 == 0)
+        expect_state = "benchmark-state-42" if is_state_reuse and (i % 2 == 1) else None
+
+        return _run_one_turn(
+            provider=provider,
+            provider_name=provider_name,
+            scenario=scenario,
+            workload_name=workload_name,
+            command=command,
+            iteration=i,
+            concurrency=concurrency,
+            user_id="bench-user",
+            thread_id=tid,
+            no_warmpool=no_warmpool,
+            state_write_turn=state_write,
+            expected_state=expect_state,
+        )
+
+    def _tid(i: int) -> str:
+        if scenario == "cold_unique_thread":
+            return f"cold-{i}"
+        elif scenario == "warm_same_thread":
+            return "warm-hit"
+        elif scenario == "warm_miss_many_threads":
+            return f"thread-{i % max(concurrency, 4)}"
+        elif scenario in ("idle_timeout", "replica_pressure"):
+            return f"warm-hit-{i % concurrency}"
+        else:
+            return f"default-{i}"
+
+    def _inject_fault(i: int) -> None:
+        if fault_inject_after is None or i != fault_inject_after:
+            return
+        tid = _tid(i)
+        sandbox_id = _compute_sandbox_id(provider, tid, "bench-user")
+        with provider._lock:
+            warm_entry = provider._warm_pool.get(sandbox_id)
+        if warm_entry is not None:
+            box, _ = warm_entry
+            try:
+                box.close()
+            except Exception:
+                pass
+            print(
+                f"  [fault] killed warm-pool box {sandbox_id} after iteration {i}",
+                file=sys.stderr,
+            )
+        else:
+            print(
+                f"  [fault] no warm-pool box {sandbox_id} to kill after iteration {i}",
+                file=sys.stderr,
+            )
+
+    if concurrency == 1:
+        for i in range(iterations):
+            r = _run_one(i)
+            results.append(r)
+            _inject_fault(i)
+    else:
+        sem = threading.BoundedSemaphore(concurrency)
+
+        def _guarded(i: int) -> BenchResult:
+            with sem:
+                return _run_one(i)
+
+        with ThreadPoolExecutor(max_workers=concurrency) as pool:
+            futures = {pool.submit(_guarded, i): i for i in range(iterations)}
+            for future in as_completed(futures):
+                i = futures[future]
+                results.append(future.result())
+                _inject_fault(i)
+
+    # Annotate with config
+    for r in results:
+        r.replicas = config_used.get("replicas")
+        r.idle_timeout = config_used.get("idle_timeout")
+        r.health_check_skip_seconds = config_used.get("health_check_skip_seconds")
+        r.image = config_used.get("image")
+
+    # Write JSONL
+    with output_path.open("a", encoding="utf-8") as f:
+        for r in results:
+            f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")
+
+    return results
+
+
+# ── CLI ─────────────────────────────────────────────────────────────────
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    p = argparse.ArgumentParser(
+        description="Provider-agnostic sandbox benchmark",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    p.add_argument(
+        "--provider",
+        default="boxlite",
+        choices=list(PROVIDER_FACTORIES),
+        help="Sandbox provider to benchmark",
+    )
+    p.add_argument(
+        "--scenario",
+        default="warm_same_thread",
+        choices=[
+            "warm_same_thread",
+            "cold_unique_thread",
+            "warm_miss_many_threads",
+            "idle_timeout",
+            "replica_pressure",
+        ],
+        help="Benchmark scenario",
+    )
+    p.add_argument(
+        "--workload",
+        default="noop",
+        choices=list(WORKLOADS) + ["state_reuse"],
+        help="Command to run inside the sandbox",
+    )
+    p.add_argument(
+        "--iterations",
+        type=int,
+        default=50,
+        help="Number of acquire→run→release turns (default: 50)",
+    )
+    p.add_argument(
+        "--concurrency",
+        type=int,
+        default=1,
+        help="Max concurrent turns (default: 1)",
+    )
+    p.add_argument(
+        "--output",
+        default="bench_results.jsonl",
+        help="JSONL output file (appended, default: bench_results.jsonl)",
+    )
+    p.add_argument(
+        "--no-warmpool",
+        action="store_true",
+        help="Evict from warm pool immediately after release (baseline)",
+    )
+    p.add_argument(
+        "--replicas",
+        type=int,
+        default=3,
+        help="sandbox.replicas config value (default: 3)",
+    )
+    p.add_argument(
+        "--idle-timeout",
+        type=float,
+        default=600,
+        help="sandbox.idle_timeout in seconds (default: 600)",
+    )
+    p.add_argument(
+        "--health-check-skip-seconds",
+        type=float,
+        default=0.0,
+        help="sandbox.health_check_skip_seconds in seconds (default: 0.0)",
+    )
+    p.add_argument(
+        "--image",
+        default=None,
+        help="OCI image override (default: provider-specific)",
+    )
+    p.add_argument(
+        "--warmup-iterations",
+        type=int,
+        default=1,
+        help="Warm-up turns before timed iterations (default: 1)",
+    )
+    p.add_argument(
+        "--fault-inject",
+        type=int,
+        default=None,
+        metavar="N",
+        help="After iteration N, close the warm-pool box to simulate a VM crash",
+    )
+    return p.parse_args(argv)
+
+
+# ── Main ────────────────────────────────────────────────────────────────
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(argv)
+    if args.workload == "state_reuse" and (args.scenario != "warm_same_thread" or args.concurrency != 1):
+        raise SystemExit("state_reuse requires --scenario warm_same_thread --concurrency 1")
+
+    output_path = Path(args.output)
+
+    config: dict[str, Any] = {
+        "replicas": args.replicas,
+        "idle_timeout": args.idle_timeout,
+        "health_check_skip_seconds": args.health_check_skip_seconds,
+        "image": args.image,
+    }
+
+    factory = PROVIDER_FACTORIES[args.provider]
+    provider, config_used = factory(config)
+    _install_warm_hit_tracking(provider)
+
+    if output_path.exists():
+        header = f"# provider={args.provider} scenario={args.scenario} workload={args.workload} concurrency={args.concurrency} iterations={args.iterations} no_warmpool={args.no_warmpool}\n"
+        with output_path.open("a", encoding="utf-8") as f:
+            f.write(header)
+
+    try:
+        # --- Warm-up (not measured) ---
+        if args.warmup_iterations > 0:
+            print(
+                f"Warming up ({args.warmup_iterations} turn(s))...",
+                file=sys.stderr,
+            )
+            for i in range(args.warmup_iterations):
+                _run_one_turn(
+                    provider=provider,
+                    provider_name=args.provider,
+                    scenario="warmup",
+                    workload_name="noop",
+                    command="true",
+                    iteration=-(args.warmup_iterations - i),
+                    concurrency=1,
+                    user_id="bench-user",
+                    thread_id="warmup",
+                    no_warmpool=args.no_warmpool,
+                )
+
+        # --- Idle-timeout scenario: special handling ---
+        if args.scenario == "idle_timeout":
+            return _run_idle_timeout_scenario(provider, args, output_path, config_used)
+
+        # --- Replica pressure scenario: special handling ---
+        if args.scenario == "replica_pressure":
+            return _run_replica_pressure_scenario(provider, args, output_path, config_used)
+
+        # --- Standard scenarios ---
+        print(
+            f"Running: provider={args.provider} scenario={args.scenario} workload={args.workload} concurrency={args.concurrency} iterations={args.iterations}",
+            file=sys.stderr,
+        )
+
+        results = _run_scenario(
+            provider=provider,
+            provider_name=args.provider,
+            scenario=args.scenario,
+            workload_name=args.workload,
+            iterations=args.iterations,
+            concurrency=args.concurrency,
+            output_path=output_path,
+            no_warmpool=args.no_warmpool,
+            config_used=config_used,
+            fault_inject_after=args.fault_inject,
+        )
+
+        _print_summary(results, args)
+
+    finally:
+        provider.shutdown()
+
+    return 0
+
+
+def _run_idle_timeout_scenario(
+    provider: Any,
+    args: argparse.Namespace,
+    output_path: Path,
+    config_used: dict[str, Any],
+) -> int:
+    """Acquire, release, force-reap warm entries, verify re-acquire is cold.
+
+    The idle reaper thread runs every 60 s by default — too slow for a
+    benchmark.  We call ``_reap_expired_warm`` directly after the sleep to
+    simulate the reaper firing.
+    """
+    idle = min(args.idle_timeout, 10)
+    print(
+        f"Idle timeout scenario: acquire, release, force-reap after {idle + 1}s sleep (timeout={idle}s)",
+        file=sys.stderr,
+    )
+
+    results: list[BenchResult] = []
+    for i in range(min(args.iterations, 20)):
+        tid = f"idle-{i}"
+        r1 = _run_one_turn(
+            provider,
+            args.provider,
+            "idle_timeout",
+            args.workload,
+            WORKLOADS.get(args.workload, "true"),
+            i * 2,
+            args.concurrency,
+            "bench-user",
+            tid,
+            args.no_warmpool,
+        )
+        results.append(r1)
+
+        # Sleep past the idle timeout, then force-reap
+        print(f"  Sleeping {idle + 1}s then force-reaping...", file=sys.stderr)
+        time.sleep(idle + 1)
+        provider._reap_expired_warm(idle_timeout=idle)
+
+        r2 = _run_one_turn(
+            provider,
+            args.provider,
+            "idle_timeout",
+            args.workload,
+            WORKLOADS.get(args.workload, "true"),
+            i * 2 + 1,
+            args.concurrency,
+            "bench-user",
+            tid,
+            args.no_warmpool,
+        )
+        if r2.warm_hit:
+            print(
+                f"  WARNING: turn {i * 2 + 1} was a warm hit — reaping may not have removed the entry",
+                file=sys.stderr,
+            )
+        results.append(r2)
+
+    for r in results:
+        r.replicas = config_used.get("replicas")
+        r.idle_timeout = config_used.get("idle_timeout")
+        r.image = config_used.get("image")
+
+    with output_path.open("a", encoding="utf-8") as f:
+        for r in results:
+            f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")
+
+    _print_summary(results, args)
+    return 0
+
+
+def _run_replica_pressure_scenario(
+    provider: Any,
+    args: argparse.Namespace,
+    output_path: Path,
+    config_used: dict[str, Any],
+) -> int:
+    """Push past replicas limit to verify eviction behaviour."""
+    replicas = args.replicas
+    overcommit = replicas * 2
+
+    print(
+        f"Replica pressure: replicas={replicas}, overcommitting to {overcommit} unique threads, {args.iterations} rounds",
+        file=sys.stderr,
+    )
+
+    results: list[BenchResult] = []
+    for i in range(args.iterations):
+        tid = f"pressure-{i % overcommit}"
+        r = _run_one_turn(
+            provider,
+            args.provider,
+            "replica_pressure",
+            args.workload,
+            WORKLOADS.get(args.workload, "true"),
+            i,
+            args.concurrency,
+            "bench-user",
+            tid,
+            args.no_warmpool,
+        )
+
+        # Track warm pool evictions
+        with provider._lock:
+            warm_size = len(provider._warm_pool)
+            active_size = len(provider._boxes)
+        print(
+            f"  iter {i}: warm_pool={warm_size} active={active_size} warm_hit={r.warm_hit}",
+            file=sys.stderr,
+        )
+
+        results.append(r)
+
+    for r in results:
+        r.replicas = config_used.get("replicas")
+        r.idle_timeout = config_used.get("idle_timeout")
+        r.image = config_used.get("image")
+
+    with output_path.open("a", encoding="utf-8") as f:
+        for r in results:
+            f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")
+
+    _print_summary(results, args)
+    return 0
+
+
+def _print_summary(results: list[BenchResult], args: argparse.Namespace) -> None:
+    """Print a quick summary to stderr."""
+    ok = [r for r in results if r.success]
+    fail = [r for r in results if not r.success]
+    warm = [r for r in ok if r.warm_hit]
+    cold = [r for r in ok if r.warm_hit is False]
+
+    if not ok:
+        print("All iterations failed.", file=sys.stderr)
+        for r in fail:
+            print(f"  {r.error}", file=sys.stderr)
+        return
+
+    def _p(arr: list[float], pct: float) -> float:
+        if not arr:
+            return 0.0
+        idx = max(0, min(len(arr) - 1, int(len(arr) * pct / 100)))
+        return sorted(arr)[idx]
+
+    a = [r.acquire_ms for r in ok]
+    t = [r.total_ms for r in ok]
+
+    print(file=sys.stderr)
+    print(
+        f"Results: {len(ok)} ok, {len(fail)} fail, {len(warm)} warm hits, {len(cold)} cold",
+        file=sys.stderr,
+    )
+    print(
+        f"  acquire: p50={_p(a, 50):.1f}ms p95={_p(a, 95):.1f}ms p99={_p(a, 99):.1f}ms",
+        file=sys.stderr,
+    )
+    print(
+        f"  total:   p50={_p(t, 50):.1f}ms p95={_p(t, 95):.1f}ms p99={_p(t, 99):.1f}ms",
+        file=sys.stderr,
+    )
+    print(f"  output → {args.output}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/backend/scripts/benchmark/summarize_bench.py b/backend/scripts/benchmark/summarize_bench.py
new file mode 100755
index 000000000..0dfe7b879
--- /dev/null
+++ b/backend/scripts/benchmark/summarize_bench.py
@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+"""Aggregate JSONL benchmark results into summary tables.
+
+Usage::
+
+    python scripts/benchmark/summarize_bench.py results.jsonl
+    python scripts/benchmark/summarize_bench.py results/*.jsonl --group provider,scenario,workload
+    python scripts/benchmark/summarize_bench.py results.jsonl --csv > summary.csv
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+
+def _p(arr: list[float], pct: float) -> float:
+    if not arr:
+        return 0.0
+    s = sorted(arr)
+    if len(s) == 1:
+        return s[0]
+    rank = (len(s) - 1) * (pct / 100)
+    lower = int(rank)
+    upper = min(lower + 1, len(s) - 1)
+    weight = rank - lower
+    return s[lower] * (1 - weight) + s[upper] * weight
+
+
+def _load_jsonl(paths: list[Path]) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    errors: list[str] = []
+    for p in paths:
+        with p.open("r", encoding="utf-8") as f:
+            for line_no, line in enumerate(f, start=1):
+                line = line.strip()
+                if not line or line.startswith("#"):
+                    continue
+                try:
+                    rows.append(json.loads(line))
+                except json.JSONDecodeError as exc:
+                    errors.append(f"{p}:{line_no}: {exc.msg}")
+    if errors:
+        raise ValueError("Malformed JSONL row(s):\n" + "\n".join(errors))
+    return rows
+
+
+def _group_key(row: dict[str, Any], group_by: list[str]) -> tuple:
+    return tuple(row.get(k, "?") for k in group_by)
+
+
+def _summarize(rows: list[dict[str, Any]], group_by: list[str]) -> list[dict[str, Any]]:
+    groups: dict[tuple, list[dict[str, Any]]] = defaultdict(list)
+    for r in rows:
+        groups[_group_key(r, group_by)].append(r)
+
+    summary: list[dict[str, Any]] = []
+    for key, group in sorted(groups.items()):
+        ok = [r for r in group if r.get("success")]
+        a = [r["acquire_ms"] for r in ok]
+        ru = [r.get("run_ms", 0) for r in ok]
+        rel = [r.get("release_ms", 0) for r in ok]
+        t = [r["total_ms"] for r in ok]
+        warm_hits = sum(1 for r in ok if r.get("warm_hit"))
+        errors = len(group) - len(ok)
+
+        entry: dict[str, Any] = {}
+        for i, k in enumerate(group_by):
+            entry[k] = key[i]
+        entry["count"] = len(group)
+        entry["ok"] = len(ok)
+        entry["errors"] = errors
+        entry["warm_hit_rate"] = round(warm_hits / len(ok), 3) if ok else 0
+        entry["acquire_p50"] = round(_p(a, 50), 1)
+        entry["acquire_p95"] = round(_p(a, 95), 1)
+        entry["acquire_p99"] = round(_p(a, 99), 1)
+        entry["acquire_mean"] = round(sum(a) / len(a), 1) if a else 0.0
+        entry["run_p50"] = round(_p(ru, 50), 1)
+        entry["run_p95"] = round(_p(ru, 95), 1)
+        entry["release_p50"] = round(_p(rel, 50), 1)
+        entry["total_p50"] = round(_p(t, 50), 1)
+        entry["total_p95"] = round(_p(t, 95), 1)
+        entry["total_p99"] = round(_p(t, 99), 1)
+        entry["total_mean"] = round(sum(t) / len(t), 1) if t else 0.0
+
+        summary.append(entry)
+
+    return summary
+
+
+_COLUMNS = [
+    "provider",
+    "scenario",
+    "workload",
+    "concurrency",
+    "count",
+    "ok",
+    "errors",
+    "warm_hit_rate",
+    "acquire_p50",
+    "acquire_p95",
+    "acquire_p99",
+    "acquire_mean",
+    "run_p50",
+    "run_p95",
+    "release_p50",
+    "total_p50",
+    "total_p95",
+    "total_p99",
+    "total_mean",
+]
+
+
+def _print_table(rows: list[dict[str, Any]], fmt: str = "plain") -> None:
+    if fmt == "csv":
+        import csv as _csv
+
+        w = _csv.DictWriter(sys.stdout, fieldnames=_COLUMNS, extrasaction="ignore")
+        w.writeheader()
+        w.writerows(rows)
+        return
+
+    # Plain text table
+    if not rows:
+        print("(no data)")
+        return
+
+    headers = [c for c in _COLUMNS if any(r.get(c) is not None for r in rows)]
+    col_widths = {h: len(h) for h in headers}
+    for r in rows:
+        for h in headers:
+            v = str(r.get(h, ""))
+            col_widths[h] = max(col_widths[h], len(v))
+
+    def _fmt_row(vals: list[str]) -> str:
+        parts = [v.rjust(col_widths[h]) for h, v in zip(headers, vals)]
+        return "  ".join(parts)
+
+    print(_fmt_row(headers))
+    for r in rows:
+        vals = [str(r.get(h, "")) for h in headers]
+        print(_fmt_row(vals))
+
+
+def main(argv: list[str] | None = None) -> int:
+    p = argparse.ArgumentParser(description="Aggregate JSONL benchmark results")
+    p.add_argument("inputs", nargs="+", help="JSONL file(s) from bench_sandbox_provider.py")
+    p.add_argument(
+        "--group",
+        default="provider,scenario,workload,concurrency",
+        help="Comma-separated grouping dimensions (default: provider,scenario,workload,concurrency)",
+    )
+    p.add_argument(
+        "--csv",
+        action="store_true",
+        help="Output CSV instead of aligned text",
+    )
+    p.add_argument(
+        "--json",
+        action="store_true",
+        help="Output JSON instead of aligned text",
+    )
+    args = p.parse_args(argv)
+
+    paths = [Path(i) for i in args.inputs]
+    try:
+        rows = _load_jsonl(paths)
+    except ValueError as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+    if not rows:
+        print("No valid JSONL rows found.", file=sys.stderr)
+        return 1
+
+    group_by = [g.strip() for g in args.group.split(",") if g.strip()]
+    summary = _summarize(rows, group_by)
+
+    if args.json:
+        json.dump(summary, sys.stdout, indent=2)
+    elif args.csv:
+        _print_table(summary, fmt="csv")
+    else:
+        _print_table(summary, fmt="plain")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/backend/tests/test_bench_sandbox_provider.py b/backend/tests/test_bench_sandbox_provider.py
new file mode 100644
index 000000000..1217c3be2
--- /dev/null
+++ b/backend/tests/test_bench_sandbox_provider.py
@@ -0,0 +1,320 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from typing import Any
+
+
+def _load_module(name: str, relative: str):
+    path = Path(__file__).resolve().parents[1] / relative
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec is not None
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+bench = _load_module("bench_sandbox_provider", "scripts/benchmark/bench_sandbox_provider.py")
+summarize = _load_module("summarize_bench", "scripts/benchmark/summarize_bench.py")
+
+
+class _FakeProvider:
+    def __init__(self, sandbox: Any | None = None) -> None:
+        self._lock = bench.threading.Lock()
+        self._warm_pool: dict[str, tuple[Any, float]] = {}
+        self._sandbox = sandbox or _FakeSandbox("ok")
+        self.released: list[str] = []
+        self.shutdown_called = False
+
+    def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        return "sandbox-id"
+
+    def get(self, sandbox_id: str):
+        return self._sandbox
+
+    def release(self, sandbox_id: str) -> None:
+        self.released.append(sandbox_id)
+
+    def shutdown(self) -> None:
+        self.shutdown_called = True
+
+
+class _FakeWarmReclaimProvider(_FakeProvider):
+    def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        reclaimed = self._reclaim_warm_pool("sandbox-id")
+        assert reclaimed is not None
+        return reclaimed
+
+    def _reclaim_warm_pool(self, sandbox_id: str) -> str | None:
+        return sandbox_id
+
+
+class _FakeSandbox:
+    def __init__(self, output: str | Exception) -> None:
+        self.output = output
+
+    def execute_command(self, command: str, timeout: float | None = None) -> str:
+        if isinstance(self.output, Exception):
+            raise self.output
+        return self.output
+
+
+def test_aio_provider_default_leaves_image_unset(monkeypatch, tmp_path):
+    captured_config: dict[str, Any] = {}
+
+    def _factory(config: dict[str, Any]):
+        captured_config.update(config)
+        return _FakeProvider(), {"replicas": config["replicas"], "idle_timeout": config["idle_timeout"], "image": config.get("image")}
+
+    monkeypatch.setitem(bench.PROVIDER_FACTORIES, "aio-docker", _factory)
+
+    rc = bench.main(
+        [
+            "--provider",
+            "aio-docker",
+            "--iterations",
+            "0",
+            "--warmup-iterations",
+            "0",
+            "--output",
+            str(tmp_path / "out.jsonl"),
+        ]
+    )
+
+    assert rc == 0
+    assert captured_config["image"] is None
+
+
+def test_explicit_aio_provider_image_is_forwarded(monkeypatch, tmp_path):
+    captured_config: dict[str, Any] = {}
+
+    def _factory(config: dict[str, Any]):
+        captured_config.update(config)
+        return _FakeProvider(), {"replicas": config["replicas"], "idle_timeout": config["idle_timeout"], "image": config.get("image")}
+
+    monkeypatch.setitem(bench.PROVIDER_FACTORIES, "aio-docker", _factory)
+
+    rc = bench.main(
+        [
+            "--provider",
+            "aio-docker",
+            "--image",
+            "custom/aio:latest",
+            "--iterations",
+            "0",
+            "--warmup-iterations",
+            "0",
+            "--output",
+            str(tmp_path / "out.jsonl"),
+        ]
+    )
+
+    assert rc == 0
+    assert captured_config["image"] == "custom/aio:latest"
+
+
+def test_failed_turn_releases_acquired_sandbox() -> None:
+    provider = _FakeProvider(_FakeSandbox(RuntimeError("boom")))
+
+    result = bench._run_one_turn(
+        provider=provider,
+        provider_name="fake",
+        scenario="warm_same_thread",
+        workload_name="noop",
+        command="true",
+        iteration=0,
+        concurrency=1,
+        user_id="user",
+        thread_id="thread",
+        no_warmpool=False,
+    )
+
+    assert result.success is False
+    assert provider.released == ["sandbox-id"]
+
+
+def test_error_string_output_records_failed_turn() -> None:
+    provider = _FakeProvider(_FakeSandbox("Error: vsock disconnected"))
+
+    result = bench._run_one_turn(
+        provider=provider,
+        provider_name="fake",
+        scenario="warm_same_thread",
+        workload_name="noop",
+        command="true",
+        iteration=0,
+        concurrency=1,
+        user_id="user",
+        thread_id="thread",
+        no_warmpool=False,
+    )
+
+    assert result.success is False
+    assert result.error == "Error: vsock disconnected"
+    assert provider.released == ["sandbox-id"]
+
+
+def test_warm_hit_uses_reclaim_instrumentation_not_pre_acquire_sample() -> None:
+    provider = _FakeWarmReclaimProvider(_FakeSandbox("ok"))
+    bench._install_warm_hit_tracking(provider)
+
+    result = bench._run_one_turn(
+        provider=provider,
+        provider_name="fake",
+        scenario="warm_same_thread",
+        workload_name="noop",
+        command="true",
+        iteration=0,
+        concurrency=1,
+        user_id="user",
+        thread_id="thread",
+        no_warmpool=False,
+    )
+
+    assert result.success is True
+    assert result.warm_hit is True
+
+
+def test_summary_preserves_all_failure_group() -> None:
+    rows = [
+        {
+            "provider": "boxlite",
+            "scenario": "warm_same_thread",
+            "workload": "noop",
+            "concurrency": 1,
+            "success": False,
+            "error": "RuntimeError('boom')",
+            "acquire_ms": 0,
+            "total_ms": 12.3,
+        }
+    ]
+
+    summary = summarize._summarize(rows, ["provider", "scenario", "workload", "concurrency"])
+
+    assert summary == [
+        {
+            "provider": "boxlite",
+            "scenario": "warm_same_thread",
+            "workload": "noop",
+            "concurrency": 1,
+            "count": 1,
+            "ok": 0,
+            "errors": 1,
+            "warm_hit_rate": 0,
+            "acquire_p50": 0.0,
+            "acquire_p95": 0.0,
+            "acquire_p99": 0.0,
+            "acquire_mean": 0.0,
+            "run_p50": 0.0,
+            "run_p95": 0.0,
+            "release_p50": 0.0,
+            "total_p50": 0.0,
+            "total_p95": 0.0,
+            "total_p99": 0.0,
+            "total_mean": 0.0,
+        }
+    ]
+
+
+def test_health_check_skip_seconds_is_forwarded_and_serialized(monkeypatch, tmp_path):
+    captured_config: dict[str, Any] = {}
+
+    def _factory(config: dict[str, Any]):
+        captured_config.update(config)
+        return _FakeProvider(), {
+            "replicas": config["replicas"],
+            "idle_timeout": config["idle_timeout"],
+            "image": config.get("image"),
+            "health_check_skip_seconds": config.get("health_check_skip_seconds"),
+        }
+
+    monkeypatch.setitem(bench.PROVIDER_FACTORIES, "boxlite", _factory)
+    output_path = tmp_path / "out.jsonl"
+
+    rc = bench.main(
+        [
+            "--provider",
+            "boxlite",
+            "--iterations",
+            "1",
+            "--warmup-iterations",
+            "0",
+            "--health-check-skip-seconds",
+            "7.5",
+            "--output",
+            str(output_path),
+        ]
+    )
+
+    assert rc == 0
+    assert captured_config["health_check_skip_seconds"] == 7.5
+    row = __import__("json").loads(output_path.read_text(encoding="utf-8").splitlines()[0])
+    assert row["health_check_skip_seconds"] == 7.5
+
+
+def test_boxlite_factory_restores_module_state(monkeypatch):
+    import deerflow.community.boxlite.provider as provider_mod
+
+    original_get_app_config = provider_mod.get_app_config
+
+    class _FactoryProvider:
+        _create_box = object()
+
+        def __init__(self) -> None:
+            self.created = True
+
+    original_create_box = _FactoryProvider._create_box
+    monkeypatch.setattr(provider_mod, "BoxliteProvider", _FactoryProvider)
+
+    provider, _ = bench._make_boxlite_provider({})
+
+    assert isinstance(provider, _FactoryProvider)
+    assert provider_mod.get_app_config is original_get_app_config
+    assert _FactoryProvider._create_box is original_create_box
+
+
+def test_boxlite_shim_workaround_retries_after_fixing_permissions(monkeypatch, tmp_path):
+    boxes_dir = tmp_path / "boxes"
+    shim = boxes_dir / "deadbeef" / "bin" / "boxlite-shim"
+    shim.parent.mkdir(parents=True)
+    shim.write_text("#!/bin/sh\n", encoding="utf-8")
+    shim.chmod(0o644)
+
+    calls = 0
+
+    def _create_box(_sandbox_id: str):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("shim not executable")
+        return "ok"
+
+    monkeypatch.setattr(bench, "_boxlite_version", lambda: "0.9.7")
+
+    result = bench._create_box_with_097_shim_workaround(
+        _create_box,
+        "sandbox-id",
+        boxes_dir=str(boxes_dir),
+    )
+
+    assert result == "ok"
+    assert calls == 2
+    assert shim.stat().st_mode & 0o111
+
+
+def test_boxlite_shim_workaround_loud_fails_for_other_versions(monkeypatch, tmp_path):
+    boxes_dir = tmp_path / "boxes"
+    monkeypatch.setattr(bench, "_boxlite_version", lambda: "0.9.8")
+
+    def _create_box(_sandbox_id: str):
+        raise RuntimeError("shim not executable")
+
+    with __import__("pytest").raises(RuntimeError, match="only supports boxlite 0.9.7"):
+        bench._create_box_with_097_shim_workaround(
+            _create_box,
+            "sandbox-id",
+            boxes_dir=str(boxes_dir),
+        )
diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py
index 98c1abe08..dff07e888 100644
--- a/backend/tests/test_boxlite_provider.py
+++ b/backend/tests/test_boxlite_provider.py
@@ -7,6 +7,7 @@ which need a live box.
 from __future__ import annotations
 
 import asyncio
+import logging
 import sys
 import threading
 import time
@@ -158,6 +159,99 @@ def test_execute_command_forwards_timeout_to_sdk_and_loop_runner() -> None:
     assert run_timeouts == [5]
 
 
+def test_execute_command_invalidates_box_on_terminal_transport_error() -> None:
+    invalidated: list[tuple[str, str]] = []
+
+    def _failing_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("vsock disconnected")
+
+    box = BoxliteBox(
+        "box-id",
+        box=_FakeBox(name="box-id"),
+        run=_failing_run,
+        on_terminal_failure=lambda sandbox_id, reason: invalidated.append((sandbox_id, reason)),
+    )
+
+    output = box.execute_command("echo hi")
+
+    assert output == "Error: vsock disconnected"
+    assert invalidated == [("box-id", "vsock disconnected")]
+
+
+def test_execute_command_does_not_invalidate_on_regular_command_error() -> None:
+    invalidated: list[tuple[str, str]] = []
+
+    def _failing_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("user command failed")
+
+    box = BoxliteBox(
+        "box-id",
+        box=_FakeBox(name="box-id"),
+        run=_failing_run,
+        on_terminal_failure=lambda sandbox_id, reason: invalidated.append((sandbox_id, reason)),
+    )
+
+    output = box.execute_command("echo hi")
+
+    assert output == "Error: user command failed"
+    assert invalidated == []
+
+
+def test_execute_command_does_not_invalidate_on_retryable_transport_message() -> None:
+    invalidated: list[tuple[str, str]] = []
+
+    def _failing_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("transport not ready, retry later")
+
+    box = BoxliteBox(
+        "box-id",
+        box=_FakeBox(name="box-id"),
+        run=_failing_run,
+        on_terminal_failure=lambda sandbox_id, reason: invalidated.append((sandbox_id, reason)),
+    )
+
+    output = box.execute_command("echo hi")
+
+    assert output == "Error: transport not ready, retry later"
+    assert invalidated == []
+
+
+def test_execute_command_uses_overridable_terminal_markers(monkeypatch: pytest.MonkeyPatch) -> None:
+    invalidated: list[tuple[str, str]] = []
+    monkeypatch.setattr(BoxliteBox, "TERMINAL_ERROR_MARKERS", ("custom terminal marker",))
+    monkeypatch.setattr(BoxliteBox, "RETRYABLE_ERROR_MARKERS", ())
+
+    def _failing_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("custom terminal marker")
+
+    box = BoxliteBox(
+        "box-id",
+        box=_FakeBox(name="box-id"),
+        run=_failing_run,
+        on_terminal_failure=lambda sandbox_id, reason: invalidated.append((sandbox_id, reason)),
+    )
+
+    output = box.execute_command("echo hi")
+
+    assert output == "Error: custom terminal marker"
+    assert invalidated == [("box-id", "custom terminal marker")]
+
+
+def test_execute_command_closed_box_returns_without_error_log(caplog) -> None:
+    box = BoxliteBox("box-id", box=_FakeBox(name="box-id"), run=_fake_run)
+    box.close()
+
+    with caplog.at_level(logging.ERROR, logger="deerflow.community.boxlite.box"):
+        output = box.execute_command("echo hi")
+
+    assert output == "Error: sandbox has been closed"
+    assert "Failed to execute command in BoxLite box" not in caplog.text
+
+
 def test_sandbox_id_deterministic(monkeypatch):
     """_sandbox_id produces the same id for the same inputs."""
     monkeypatch.setattr(
@@ -343,6 +437,244 @@ def test_acquire_reclaims_from_warm_pool(monkeypatch):
     assert sid2 not in provider._warm_pool
 
 
+def test_explicit_recent_reclaim_skip_avoids_health_check(monkeypatch):
+    """A configured skip window can reclaim recently released boxes without a ping."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"health_check_skip_seconds": 5}),
+    )
+    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")
+    provider.release(sid)
+    box, _ = provider._warm_pool[sid]
+
+    def _fail_if_called(*args, **kwargs):
+        raise AssertionError("health check should be skipped for recently released boxes")
+
+    monkeypatch.setattr(box, "execute_command", _fail_if_called)
+
+    reclaimed = provider._reclaim_warm_pool(sid)
+    assert reclaimed == sid
+    assert sid in provider._boxes
+    assert sid not in provider._warm_pool
+
+    provider.shutdown()
+
+
+def test_recent_reclaim_validates_by_default(monkeypatch):
+    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")
+    provider.release(sid)
+    box, _ = provider._warm_pool[sid]
+    calls = 0
+    original_execute = box.execute_command
+
+    def _record_health_check(command: str, *args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return original_execute(command, *args, **kwargs)
+
+    monkeypatch.setattr(box, "execute_command", _record_health_check)
+
+    reclaimed = provider._reclaim_warm_pool(sid)
+    assert reclaimed == sid
+    assert calls == 1
+    assert sid in provider._boxes
+    assert sid not in provider._warm_pool
+
+    provider.shutdown()
+
+
+def test_default_recent_reclaim_drops_dead_warm_box(monkeypatch):
+    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")
+    provider.release(sid)
+    box, _ = provider._warm_pool[sid]
+
+    def _dead_health_check(command: str, *args, **kwargs):
+        assert command == "echo ok"
+        return "Error: vsock disconnected"
+
+    monkeypatch.setattr(box, "execute_command", _dead_health_check)
+
+    reclaimed = provider._reclaim_warm_pool(sid)
+    assert reclaimed is None
+    assert sid not in provider._boxes
+    assert sid not in provider._warm_pool
+    assert sid not in provider._skip_health_check_warm_ids
+    assert box.is_closed is True
+
+    provider.shutdown()
+
+
+def test_dead_active_box_invalidation_closes_adapter(monkeypatch):
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"health_check_skip_seconds": 5}),
+    )
+    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.get(sid)
+    assert box is not None
+
+    def _dead_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("vsock disconnected")
+
+    box._run = _dead_run
+
+    output = box.execute_command("echo hi")
+    assert output == "Error: vsock disconnected"
+    assert box._closed is True
+    assert provider.get(sid) is None
+
+    provider.shutdown()
+
+
+def test_adopted_warm_pool_box_still_health_checks(monkeypatch):
+    """Startup-adopted boxes must still pass a health check before reclaim."""
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"health_check_skip_seconds": 5}),
+    )
+
+    provider = BoxliteProvider()
+    adopted = BoxliteBox(
+        "adopted",
+        _FakeBox(name="deer-flow-boxlite-adopted"),
+        _fake_run,
+        default_env={},
+    )
+    provider._warm_pool["adopted"] = (adopted, time.time())
+    calls = 0
+    original_execute = adopted.execute_command
+
+    def _record_health_check(command: str, *args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return original_execute(command, *args, **kwargs)
+
+    monkeypatch.setattr(adopted, "execute_command", _record_health_check)
+
+    reclaimed = provider._reclaim_warm_pool("adopted")
+    assert reclaimed == "adopted"
+    assert calls == 1
+    assert "adopted" in provider._boxes
+    assert "adopted" not in provider._warm_pool
+
+    provider.shutdown()
+
+
+def test_dead_active_box_is_invalidated_after_command_failure(monkeypatch):
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"health_check_skip_seconds": 5}),
+    )
+    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.get(sid)
+    assert box is not None
+
+    def _dead_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("vsock disconnected")
+
+    box._run = _dead_run
+
+    output = box.execute_command("echo hi")
+    assert output == "Error: vsock disconnected"
+    assert provider.get(sid) is None
+
+    sid2 = provider.acquire("thread-1", user_id="u1")
+    assert sid2 == sid
+    assert provider.get(sid2) is not None
+
+    provider.shutdown()
+
+
+def test_stale_closed_adapter_cannot_invalidate_recreated_box(monkeypatch):
+    monkeypatch.setattr(
+        "deerflow.community.boxlite.provider.get_app_config",
+        lambda: _stub_config({"health_check_skip_seconds": 5}),
+    )
+    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")
+    stale_box = provider.get(sid)
+    assert stale_box is not None
+
+    def _dead_run(coro, *, timeout=None):
+        coro.close()
+        raise RuntimeError("vsock disconnected")
+
+    stale_box._run = _dead_run
+    stale_box.execute_command("echo hi")
+    assert provider.get(sid) is None
+
+    provider._loop.run = _fake_run
+    sid2 = provider.acquire("thread-1", user_id="u1")
+    replacement = provider.get(sid2)
+    assert sid2 == sid
+    assert replacement is not None
+    assert replacement is not stale_box
+
+    stale_box.execute_command("echo again")
+
+    assert provider.get(sid) is replacement
+    assert replacement._closed is False
+
+    provider.shutdown()
+
+
 def test_acquire_different_threads_dont_reclaim_each_other(monkeypatch):
     """Thread A's box can't be reclaimed by thread B."""
     monkeypatch.setattr(
@@ -394,6 +726,10 @@ def test_warm_pool_reclaim_failed_health_check_creates_new(monkeypatch):
     sid2 = provider.acquire("thread-1", user_id="u1")
     assert sid2 == sid1  # Same deterministic ID
     assert sid2 in provider._boxes
+    replacement = provider.get(sid2)
+    assert replacement is not None
+    assert replacement is not box
+    assert replacement._closed is False
 
 
 def test_concurrent_same_thread_acquire_creates_one_box(monkeypatch):
diff --git a/backend/tests/test_local_sandbox_command_timeout.py b/backend/tests/test_local_sandbox_command_timeout.py
index 984396dea..9fb8fdc1f 100644
--- a/backend/tests/test_local_sandbox_command_timeout.py
+++ b/backend/tests/test_local_sandbox_command_timeout.py
@@ -147,6 +147,11 @@ def test_sandbox_config_exposes_command_timeout_default():
     assert cfg.bash_command_timeout == 600
 
 
+def test_sandbox_config_exposes_health_check_skip_seconds_default():
+    cfg = SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider")
+    assert cfg.health_check_skip_seconds is None
+
+
 def test_bash_tool_description_guides_backgrounding_long_lived_processes():
     """The bash tool description (seen by the model) must tell it to background
     long-lived processes like servers, so it doesn't block the turn in the
diff --git a/config.example.yaml b/config.example.yaml
index 28a45523c..9b86608d2 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -15,7 +15,7 @@
 # ============================================================================
 # Bump this number when the config schema changes.
 # Run `make config-upgrade` to merge new fields into your local config.yaml.
-config_version: 20
+config_version: 21
 
 # ============================================================================
 # Logging
@@ -1149,6 +1149,10 @@ sandbox:
 #   # Set to 0 to keep warm VMs until shutdown or replica eviction.
 #   # idle_timeout: 600
 #
+#   # Optional: Skip the reclaim health check for very recently released VMs.
+#   # Default 0.0 keeps reliability-first validation before warm reuse.
+#   # health_check_skip_seconds: 0.0
+#
 #   # Optional: Environment variables to inject into every command.
 #   # environment:
 #   #   PYTHONUNBUFFERED: "1"
diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml
index 44153c25e..4f40f8a5b 100644
--- a/deploy/helm/deer-flow/values.yaml
+++ b/deploy/helm/deer-flow/values.yaml
@@ -221,7 +221,7 @@ ingress:
 # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
 #    inline literal secret values here. The default enables provisioner sandbox.
 config: |
-  config_version: 20
+  config_version: 21
   log_level: info
 
   models: []