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 fda656dcc1..3f6ae9a9cc 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 @@ -211,6 +211,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._idle_checker_thread: threading.Thread | None = None self._renewal_stop = threading.Event() self._renewal_thread: threading.Thread | None = None + self._health_check_thread: threading.Thread | None = None # Per-instance id used for cross-instance sandbox ownership leases (#4206). self._owner_id = generate_owner_id() @@ -1305,6 +1306,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): thread = self._renewal_thread if thread is not None and thread.is_alive() and thread is not threading.current_thread(): thread.join(timeout=5) + health_thread = getattr(self, "_health_check_thread", None) + if health_thread is not None and health_thread.is_alive() and health_thread is not threading.current_thread(): + health_thread.join(timeout=5) def _lease_renewal_loop(self) -> None: interval = self._ownership_config.renewal_interval_seconds @@ -1314,9 +1318,29 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): except Exception: logger.exception("Error in sandbox ownership renewal loop") try: - self._health_check_owned_sandboxes() + self._schedule_health_check() except Exception: - logger.exception("Error in sandbox health check loop") + logger.exception("Error scheduling sandbox health check") + + def _schedule_health_check(self) -> None: + """Start at most one health scan without delaying lease renewal.""" + if not self._config.get("auto_restart", True) or self._renewal_stop.is_set(): + return + thread = getattr(self, "_health_check_thread", None) + if thread is not None and thread.is_alive(): + return + self._health_check_thread = threading.Thread( + target=self._run_health_check, + name="sandbox-health-check", + daemon=True, + ) + self._health_check_thread.start() + + def _run_health_check(self) -> None: + 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. @@ -1324,10 +1348,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): ``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. + ``acquire``/``reclaim`` runs. A worker scheduled by the renewal loop + detects a crash between tool calls without delaying the next lease + refresh, then uses the usual ownership-fenced teardown. """ if not self._config.get("auto_restart", True): return @@ -1336,7 +1359,11 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): 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 getattr(self, "_shutdown_called", False): + return if self._check_tracked_sandbox_alive(sandbox_id, info) is False: + if getattr(self, "_shutdown_called", False): + return self._drop_unhealthy_sandbox( sandbox_id, "periodic health check", diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 71748f9cf9..2ba9b63a41 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -98,6 +98,8 @@ to respond. **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; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. +**AIO health polling:** The lease thread schedules at most one separate health worker, so slow active/warm container probes never delay lease renewal. + **Virtual Path System**: - Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills` - Physical: `backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/...`; raw skills stay under `deer-flow/skills/` and managed integration storage. Unrestricted sandboxes read `backend/.deer-flow/skills_view/public/` and `backend/.deer-flow/users/{user_id}/skills_view/{custom,legacy,integrations}/`; explicit lead Agent policies read `backend/.deer-flow/users/{user_id}/threads/{thread_id}/skills_view/{public,custom,legacy,integrations}/`. diff --git a/backend/tests/test_aio_sandbox_auto_restart.py b/backend/tests/test_aio_sandbox_auto_restart.py index 2d84af9448..fda47c1c30 100644 --- a/backend/tests/test_aio_sandbox_auto_restart.py +++ b/backend/tests/test_aio_sandbox_auto_restart.py @@ -58,7 +58,7 @@ def test_health_check_evicts_sandbox_that_crashed_between_tool_calls(tmp_path): ``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 + so the scheduled health worker 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. """ @@ -95,3 +95,44 @@ def test_health_check_respects_auto_restart_disabled(tmp_path): provider._backend.is_alive.assert_not_called() provider._backend.destroy.assert_not_called() assert provider.get("sandbox-crashed-disabled") is sandbox + + +def test_slow_health_check_does_not_delay_lease_renewal(): + """A blocked remote probe must not hold up the next ownership renewal.""" + module = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + provider = module.AioSandboxProvider.__new__(module.AioSandboxProvider) + provider._config = {"auto_restart": True} + provider._ownership_config = SimpleNamespace(renewal_interval_seconds=0.01, ttl_multiplier=4) + provider._renewal_stop = threading.Event() + provider._renewal_thread = None + provider._health_check_thread = None + + health_started = threading.Event() + allow_health = threading.Event() + second_renewal = threading.Event() + renewals: list[int] = [] + health_checks: list[int] = [] + + def renew() -> None: + renewals.append(1) + if len(renewals) >= 2: + second_renewal.set() + + def check_health() -> None: + health_checks.append(1) + health_started.set() + allow_health.wait(timeout=2) + + provider._renew_owned_leases = renew + provider._health_check_owned_sandboxes = check_health + + provider._start_lease_renewal() + try: + assert health_started.wait(timeout=1) + assert second_renewal.wait(timeout=0.5), "a slow health probe delayed lease renewal" + assert len(health_checks) == 1, "health checks must not spawn unbounded workers" + finally: + allow_health.set() + provider._stop_lease_renewal() + assert provider._health_check_thread is not None + assert not provider._health_check_thread.is_alive()