Evict dead cached sandboxes on the lease renewal thread

Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py

@@ -1313,6 +1313,35 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 self._renew_owned_leases()
             except Exception:
                 logger.exception("Error in sandbox ownership renewal loop")
+            try:
+                self._health_check_owned_sandboxes()
+            except Exception:
+                logger.exception("Error in sandbox health check loop")
+
+    def _health_check_owned_sandboxes(self) -> None:
+        """Evict cached sandboxes whose containers died since last use.
+
+        ``get()``/``get_scoped()`` are intentionally pure in-memory lookups
+        (async tool paths call them directly on the event loop), so a
+        mid-run container crash is otherwise only caught the next time
+        ``acquire``/``reclaim`` runs. Riding the existing renewal thread lets a
+        crash between two tool calls be detected and evicted — via the same
+        ownership-fenced teardown as every other reap path — within one
+        renewal interval instead of persisting for the rest of the run.
+        """
+        if not self._config.get("auto_restart", True):
+            return
+
+        with self._lock:
+            tracked = list(self._sandbox_infos.items()) + [(sandbox_id, info) for sandbox_id, (info, _) in self._warm_pool.items()]
+
+        for sandbox_id, info in tracked:
+            if self._check_tracked_sandbox_alive(sandbox_id, info) is False:
+                self._drop_unhealthy_sandbox(
+                    sandbox_id,
+                    "periodic health check",
+                    expected_info=info,
+                )
 
     def _renew_owned_leases(self) -> None:
         """Renew every container this instance believes it owns.

backend/tests/test_aio_sandbox_auto_restart.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import pytest
+from test_aio_sandbox_provider import _make_provider_with_active_sandbox
 
 from deerflow.config.sandbox_config import SandboxConfig
 
@@ -50,3 +51,47 @@ def test_get_remains_in_memory_lookup():
 
     assert provider.get("sandbox-1") is sandbox
     provider._backend.is_alive.assert_not_called()
+
+
+def test_health_check_evicts_sandbox_that_crashed_between_tool_calls(tmp_path):
+    """A container crash between two tool calls must not keep handing out a dead client.
+
+    ``get()``/``get_scoped()`` stay pure in-memory lookups (see
+    ``test_get_remains_in_memory_lookup`` / ``test_get_uses_in_memory_registry_only``),
+    so the renewal-thread health check is what catches a mid-run crash: it must
+    evict the cached sandbox so the *next* cache lookup misses and the caller
+    falls through to ``acquire()`` for a fresh container.
+    """
+    provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-crashed")
+    provider._thread_sandboxes = {("default", "thread-crashed"): "sandbox-crashed"}
+    info = provider._sandbox_infos["sandbox-crashed"]
+
+    # First tool call: container is healthy.
+    assert provider.get("sandbox-crashed") is sandbox
+
+    # Container crashes mid-run, before the next tool call.
+    provider._backend.is_alive = MagicMock(return_value=False)
+
+    provider._health_check_owned_sandboxes()
+
+    sandbox.close.assert_called_once_with()
+    provider._backend.destroy.assert_called_once_with(info)
+    assert "sandbox-crashed" not in provider._sandboxes
+    assert "sandbox-crashed" not in provider._sandbox_infos
+    assert ("default", "thread-crashed") not in provider._thread_sandboxes
+
+    # Next tool call's cache lookup must miss instead of returning the dead client.
+    assert provider.get("sandbox-crashed") is None
+
+
+def test_health_check_respects_auto_restart_disabled(tmp_path):
+    """With auto_restart disabled, a crashed container must not be evicted."""
+    provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-crashed-disabled")
+    provider._config = {"auto_restart": False}
+    provider._backend.is_alive = MagicMock(return_value=False)
+
+    provider._health_check_owned_sandboxes()
+
+    provider._backend.is_alive.assert_not_called()
+    provider._backend.destroy.assert_not_called()
+    assert provider.get("sandbox-crashed-disabled") is sandbox