From 4ec0fe1bb4967be9844ad2b1a2ee27da4890b7eb Mon Sep 17 00:00:00 2001 From: yinx65588-boop Date: Sat, 24 Oct 2026 08:49:35 +0800 Subject: [PATCH] fix(tenki): retain cleanup ownership across provider retirement (#7311) * fix(tenki): retain sandbox ownership through provider retirement * fix(tenki): fence terminal callbacks by sandbox generation * docs(sandbox): keep Tenki lifecycle guidance within the chain budget * fix(tenki): report pending exit cleanup without a traceback --------- Co-authored-by: yinx65588-boop Co-authored-by: yinx65588-boop <272180046+yinx65588-boop@users.noreply.github.com> Co-authored-by: Willem Jiang --- README.md | 6 + .../deerflow/community/tenki/README.md | 21 +- .../deerflow/community/tenki/provider.py | 190 ++++++-- .../deerflow/community/tenki/sandbox.py | 29 +- .../harness/deerflow/sandbox/AGENTS.md | 7 +- backend/tests/test_tenki_lifecycle.py | 458 ++++++++++++++++++ backend/tests/test_warm_pool_lifecycle.py | 3 +- 7 files changed, 643 insertions(+), 71 deletions(-) create mode 100644 backend/tests/test_tenki_lifecycle.py diff --git a/README.md b/README.md index 17337e8d27..21ce86899c 100644 --- a/README.md +++ b/README.md @@ -885,6 +885,12 @@ DeerFlow supports multiple sandbox execution modes: - **Docker Execution** (runs sandbox code in isolated Docker containers) - **Docker Execution with Kubernetes** (runs sandbox code in Kubernetes pods via provisioner service) +The optional [Tenki provider](backend/packages/harness/deerflow/community/tenki/README.md) +retains failed termination handles for retry and rejects late acquisitions after +reset or shutdown. If shutdown reports outstanding work, retry after its SDK +workers finish or the remote service recovers. +At process exit, outstanding cleanup is logged without an exception traceback. + In Local Execution, `glob` and `grep` can search the skills root (default `/mnt/skills`) across its mounted categories. Patterns stay relative to that root, and one result limit covers all categories. User isolation, disabled-skill diff --git a/backend/packages/harness/deerflow/community/tenki/README.md b/backend/packages/harness/deerflow/community/tenki/README.md index e1ccbcdeb9..ddb687bf8b 100644 --- a/backend/packages/harness/deerflow/community/tenki/README.md +++ b/backend/packages/harness/deerflow/community/tenki/README.md @@ -84,13 +84,32 @@ working via the remap. Warm-pool capacity is governed by `sandbox.replicas` across active + warm sandboxes. `sandbox.idle_timeout` controls how long released warm sandboxes stay running; `0` disables idle reaping. Active sandboxes are never evicted to satisfy -the cap. +the cap. Handles awaiting a failed termination also count toward capacity and +cannot be reused. Idle sweeps retry their termination; with idle reaping disabled, +call `shutdown()` again to retry cleanup. + +`reset()` retires the provider and parks its existing sandboxes; `shutdown()` +permanently fences admission and requests termination. A synchronous SDK creation +already in progress remains owned by its worker, which requests termination +instead of publishing its late result. Shutdown raises an actionable `RuntimeError` +while acquires or terminations remain outstanding; retry after the workers finish +or the service recovers. An async caller cancelling its wait does not stop the +SDK worker. If the provider remains open, the completed acquire can still be reused +by the same scope. Accepted termination requests do not prove that a remote VM +has physically stopped. Pending handles are process-local and do not survive a +Gateway crash; cross-process orphan recovery remains outside this contract. +The process-exit hook logs outstanding cleanup without an exception traceback; +explicit `shutdown()` calls still raise so callers can retry. The warm-pool health probe accepts an exact `ok` line alongside login-shell profile output or stderr warnings. Timeout, nonzero-exit, and command-error diagnostics still prevent reuse. Failed probes log the sandbox ID and returned output before the unhealthy microVM is terminated and replaced. +Terminal failures are bound to the adapter that issued the operation. If another +operation already replaced that sandbox, a delayed failure from the old adapter +cannot terminate its healthy replacement, whether active or waiting in the warm pool. + ## Scope: stable features only Only the stable Tenki surface is used — sandbox create/terminate plus diff --git a/backend/packages/harness/deerflow/community/tenki/provider.py b/backend/packages/harness/deerflow/community/tenki/provider.py index 1a36a1ab21..0daa207c84 100644 --- a/backend/packages/harness/deerflow/community/tenki/provider.py +++ b/backend/packages/harness/deerflow/community/tenki/provider.py @@ -23,6 +23,8 @@ import shlex import threading import time import uuid +from collections.abc import Iterator +from contextlib import contextmanager from functools import partial from typing import TYPE_CHECKING, Any @@ -97,6 +99,10 @@ def _import_client() -> type[Client]: return Client +class _ShutdownIncompleteError(RuntimeError): + """Cleanup still owns acquisitions or remote terminations that need retry.""" + + class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider): """Run each DeerFlow sandbox as a Tenki cloud microVM.""" @@ -128,9 +134,17 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider self._idle_checker_stop = threading.Event() self._idle_checker_thread: threading.Thread | None = None self._shutdown_called = False + self._shutdown_complete = False + self._shutdown_lock = threading.Lock() + self._accepting = True + self._acquires = 0 + # Object identity matters: a replacement can have the same logical id + # while an older remote generation still needs termination. + self._pending_closes: dict[int, TenkiSandbox] = {} + self._closing: set[int] = set() self._client: Client | None = None self._config = self._load_config() - atexit.register(self.shutdown) + atexit.register(self._shutdown_at_exit) self._start_idle_checker() def _load_config(self) -> dict[str, Any]: @@ -225,20 +239,38 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider super()._start_idle_checker() def _active_count_locked(self) -> int: - return len(self._sandboxes) + return len(self._sandboxes) + len(self._pending_closes) + + def _cleanup_idle_resources(self, idle_timeout: float) -> None: + self._retry_pending_closes() + super()._cleanup_idle_resources(idle_timeout) def _destroy_warm_entry(self, sandbox_id: str, entry: TenkiSandbox, *, reason: str) -> None: self._close_quietly(entry, context=f"warm pool, reason={reason}") - def _invalidate_sandbox(self, sandbox_id: str, reason: str) -> None: + def _invalidate_sandbox(self, sandbox_id: str, reason: str, *, expected_sandbox: TenkiSandbox | None = None) -> None: """Destroy and deregister a sandbox after a terminal command-path failure.""" to_close: TenkiSandbox | None = None with self._lock: + active = self._sandboxes.get(sandbox_id) + warm_entry = self._warm_pool.get(sandbox_id) + current = active or (warm_entry[0] if warm_entry is not None else None) + if expected_sandbox is not None and current is not expected_sandbox: + # Operations already in flight can fail after another failure + # has removed their adapter and acquire rebuilt this scope. + # The logical id is stable; only the originating adapter owns + # the right to invalidate this active or warm generation. + logger.debug("Ignoring late terminal failure from replaced Tenki sandbox %s: %s", sandbox_id, reason) + return active = self._sandboxes.pop(sandbox_id, None) warm_entry = self._warm_pool.pop(sandbox_id, None) for key in [k for k, sid in self._thread_sandboxes.items() if sid == sandbox_id]: self._thread_sandboxes.pop(key, None) to_close = active or (warm_entry[0] if warm_entry is not None else None) + if to_close is not None: + # Transfer ownership atomically with deregistration. Shutdown + # must see this handle even before its close worker starts. + self._pending_closes[id(to_close)] = to_close if to_close is None: logger.warning("Tenki sandbox %s failed terminally but was not tracked: %s", sandbox_id, reason) @@ -248,18 +280,41 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider # ── Acquire / release ──────────────────────────────────────────────── - def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: - if thread_id is None: - sandbox_id = str(uuid.uuid4())[:8] - sandbox = self._create_sandbox(sandbox_id) + @contextmanager + def _acquiring(self) -> Iterator[None]: + with self._lock: + if not self._accepting: + raise RuntimeError("Tenki provider is retired or shutting down") + self._acquires += 1 + try: + yield + finally: with self._lock: - self._sandboxes[sandbox.id] = sandbox - return sandbox.id + self._acquires -= 1 - key = self._thread_key(thread_id, user_id) - sandbox_id = self._sandbox_id(thread_id, user_id or "") - with self._acquire_serializer.hold(sandbox_id): - return self._acquire_scope_locked(key, sandbox_id) + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + with self._acquiring(): + if thread_id is None: + sandbox_id = str(uuid.uuid4())[:8] + sandbox = self._create_sandbox(sandbox_id) + return self._publish_sandbox(sandbox) + + key = self._thread_key(thread_id, user_id) + sandbox_id = self._sandbox_id(thread_id, user_id or "") + with self._acquire_serializer.hold(sandbox_id): + return self._acquire_scope_locked(key, sandbox_id) + + def _publish_sandbox(self, sandbox: TenkiSandbox, key: tuple[str, str] | None = None) -> str: + with self._lock: + if self._accepting: + self._sandboxes[sandbox.id] = sandbox + if key is not None: + self._thread_sandboxes[key] = sandbox.id + return sandbox.id + # The synchronous SDK worker owns its handle through this cleanup even + # when the async caller abandoned its await while create was running. + self._close_quietly(sandbox, context="acquired after provider retirement") + raise RuntimeError("Tenki provider retired or shut down during acquire") async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: """Acquire without blocking the event loop. @@ -275,6 +330,8 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider def _acquire_scope_locked(self, key: tuple[str, str], sandbox_id: str) -> str: with self._lock: + if not self._accepting: + raise RuntimeError("Tenki provider is retired or shutting down") existing = self._thread_sandboxes.get(key) if existing is not None and existing in self._sandboxes: return existing @@ -282,14 +339,13 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider reclaimed = self._reclaim_warm_pool(sandbox_id) if reclaimed is not None: with self._lock: + if not self._accepting: + raise RuntimeError("Tenki provider retired or shut down during acquire") self._thread_sandboxes[key] = reclaimed return reclaimed sandbox = self._create_sandbox(sandbox_id) - with self._lock: - self._sandboxes[sandbox.id] = sandbox - self._thread_sandboxes[key] = sandbox.id - return sandbox.id + return self._publish_sandbox(sandbox, key) def _create_sandbox(self, sandbox_id: str) -> TenkiSandbox: # Enforce replica soft cap: evict the oldest warm sandbox if active + warm @@ -341,30 +397,45 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider except Exception as e: logger.warning("Tenki bootstrap for %s raised: %s", sandbox_id, e) logger.info("Created Tenki sandbox %s (name=%s)", sandbox_id, self._sandbox_name(sandbox_id)) - return TenkiSandbox( + sandbox = TenkiSandbox( sandbox_id, remote, default_env=self._config["environment"], home_dir=self._config["home_dir"], - on_terminal_failure=self._invalidate_sandbox, + on_terminal_failure=lambda failed_id, reason: self._invalidate_sandbox(failed_id, reason, expected_sandbox=sandbox), ) + return sandbox - @staticmethod - def _terminate_orphan(sandbox_id: str, remote: Any) -> None: + def _terminate_orphan(self, sandbox_id: str, remote: Any) -> None: """Terminate a microVM that was created but never handed to the adapter.""" - try: - remote.close() - logger.warning("Terminated Tenki sandbox %s after it failed to become ready", sandbox_id) - except Exception as e: - logger.error("Leaked Tenki sandbox %s (id=%s): could not terminate after readiness failure: %s", sandbox_id, getattr(remote, "id", "?"), e) + self._close_quietly(TenkiSandbox(sandbox_id, remote), context="readiness failure") - @staticmethod - def _close_quietly(sandbox: TenkiSandbox, *, context: str) -> None: - """Close a sandbox where the caller has no way to act on a failure.""" + def _close_quietly(self, sandbox: TenkiSandbox, *, context: str) -> None: + """Own termination until accepted, retaining failures for a later sweep.""" + token = id(sandbox) + with self._lock: + self._pending_closes[token] = sandbox + if token in self._closing: + return + self._closing.add(token) + closed = False try: sandbox.close() except Exception as e: logger.warning("Error closing Tenki sandbox %s (%s): %s", sandbox.id, context, e) + else: + closed = True + finally: + with self._lock: + if closed: + self._pending_closes.pop(token, None) + self._closing.discard(token) + + def _retry_pending_closes(self) -> None: + with self._lock: + pending = list(self._pending_closes.values()) + for sandbox in pending: + self._close_quietly(sandbox, context="retry termination") def get(self, sandbox_id: str) -> Sandbox | None: with self._lock: @@ -386,12 +457,13 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider return if self._shutdown_called: close_sandbox = sandbox + self._pending_closes[id(sandbox)] = sandbox else: self._warm_pool[sandbox_id] = (sandbox, time.monotonic()) if close_sandbox is not None: self._close_quietly(close_sandbox, context="released during shutdown") - logger.info("Closed released Tenki sandbox %s because shutdown is in progress", sandbox_id) + logger.info("Requested termination of released Tenki sandbox %s because shutdown is in progress", sandbox_id) else: logger.info("Released Tenki sandbox %s to warm pool (microVM still running)", sandbox_id) @@ -429,7 +501,13 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider warm_entry = self._warm_pool.pop(sandbox_id, None) if warm_entry is None: return None # Raced with another thread - self._sandboxes[sandbox_id] = warm_entry[0] + admitted = self._accepting + if admitted: + self._sandboxes[sandbox_id] = warm_entry[0] + + if not admitted: + self._close_quietly(warm_entry[0], context="reclaimed after provider retirement") + raise RuntimeError("Tenki provider retired or shut down during acquire") logger.info("Reclaimed warm-pool Tenki sandbox %s", sandbox_id) return sandbox_id @@ -443,6 +521,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider instance's idle reaper and atexit shutdown instead of orphaning them. """ with self._lock: + self._accepting = False now = time.monotonic() for sandbox_id, sandbox in self._sandboxes.items(): self._warm_pool.setdefault(sandbox_id, (sandbox, now)) @@ -450,26 +529,35 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider self._thread_sandboxes.clear() self._acquire_serializer.close() - def shutdown(self) -> None: - with self._lock: - if self._shutdown_called: - return - self._shutdown_called = True - + def _shutdown_at_exit(self) -> None: try: - self._stop_idle_checker() - except Exception: + self.shutdown() + except _ShutdownIncompleteError as e: + logger.warning("%s", e) + + def shutdown(self) -> None: + # Shutdown retries cleanup but never reopens admission. An SDK call + # cannot be forcibly cancelled: its worker retains ownership and closes + # a late handle. Report unfinished work instead of pretending to drain + # it or blocking shutdown on an unbounded remote call. + with self._shutdown_lock: with self._lock: - self._shutdown_called = False - raise - - with self._lock: - active = list(self._sandboxes.values()) - warm = [sandbox for sandbox, _ in self._warm_pool.values()] - self._sandboxes.clear() - self._warm_pool.clear() - self._thread_sandboxes.clear() + if self._shutdown_complete: + return + self._shutdown_called = True + self._accepting = False self._acquire_serializer.close() - - for sandbox in active + warm: - self._close_quietly(sandbox, context="shutdown") + self._stop_idle_checker() + with self._lock: + active = list(self._sandboxes.values()) + warm = [sandbox for sandbox, _ in self._warm_pool.values()] + self._sandboxes.clear() + self._warm_pool.clear() + self._thread_sandboxes.clear() + previous_pending = list(self._pending_closes.values()) + for sandbox in {id(box): box for box in active + warm + previous_pending}.values(): + self._close_quietly(sandbox, context="shutdown") + with self._lock: + if self._acquires or self._pending_closes: + raise _ShutdownIncompleteError(f"Tenki shutdown incomplete: {self._acquires} acquires and {len(self._pending_closes)} terminations pending; retry shutdown") + self._shutdown_complete = True diff --git a/backend/packages/harness/deerflow/community/tenki/sandbox.py b/backend/packages/harness/deerflow/community/tenki/sandbox.py index a7859519b4..3651775a6e 100644 --- a/backend/packages/harness/deerflow/community/tenki/sandbox.py +++ b/backend/packages/harness/deerflow/community/tenki/sandbox.py @@ -133,24 +133,25 @@ class TenkiSandbox(Sandbox): def close(self) -> None: """Terminate the underlying Tenki session (idempotent). - The microVM is terminated *first*; the adapter is only marked closed once - the session is actually gone, so a failed termination stays retryable - instead of silently leaking a running (billed) sandbox. A terminal - session error means it is already gone, which counts as closed; anything - else is raised so the caller can retry or alert. + Mark closed only after the termination RPC is accepted or the service + confirms the session is already gone. This does not wait for physical + VM termination. Ambiguous transport failures stay retryable instead of + silently abandoning a possibly running (billed) sandbox. + Holding the adapter lock across the RPC serializes close attempts; + other operations checking this lock wait for that network call too. """ with self._lock: if self._closed: return - sandbox = self._sandbox - try: - sandbox.close() - except Exception as e: - if not self._is_terminal_failure(e): - logger.error("Error terminating Tenki sandbox %s: %s", self.id, e) - raise - logger.info("Tenki sandbox %s was already gone at close: %s", self.id, e) - with self._lock: + try: + self._sandbox.close() + except Exception as e: + # A broken transport can make an operational handle unusable + # without confirming that the termination RPC was accepted. + if type(e).__name__ not in {"SessionTerminatedError", "SessionNotFoundError"}: + logger.error("Error terminating Tenki sandbox %s: %s", self.id, e) + raise + logger.info("Tenki sandbox %s was already gone at close: %s", self.id, e) self._closed = True # ── bridge helpers ────────────────────────────────────────────────── diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 57bf40fdfe..93e9ef443c 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -111,11 +111,10 @@ Fork lookups reject invalidated clients; cancelled acquires leave borrowers a pe - `BoxliteProvider` (`community/boxlite/`): optional micro-VMs. One private daemon asyncio loop owns SDK handles; sync calls use `run_coroutine_threadsafe`. Deterministic `user_id:thread_id` boxes permit owner-only warm reuse. Health checks bound SDK `exec(timeout=...)` and bridge `.result(timeout)`; bridge timeouts cancel coroutines to release work/acquire ownership. `sandbox.replicas` caps active + warm VMs per process; evict only warm VMs. `sandbox.idle_timeout` expires warm VMs. `reset()` retains VMs, reaper, and loop; `shutdown()` performs full teardown. Fence registration under `_lock`; stop late VMs before loop close. Shutdown drains acquires for 5s; timeout retains resources for retry. `boxlite-loop` join must confirm exit before loop close; a join timeout keeps admission fenced and retries only loop cleanup. -- `TenkiSandboxProvider` (`community/tenki/`): cloud microVMs. Optional `tenki` SDK (`deerflow-harness[tenki]`, module `tenki_sandbox`) loads via `_import_client` only on provider selection; calls are synchronous, without an event-loop bridge. Native `sandbox.fs` (`read_text`/`read_stream`/`write_stream`/`mkdir`/`stat`) streams binary-safe files without shell/base64. Only search (`list_dir`/`glob`/`grep`) uses busybox-portable `find`/`grep` and shared `deerflow.sandbox.search` parsing, like E2B. The unprivileged `tenki` user needs `/mnt/user-data` remapped to writable HOME (`_resolve_path`), with a best-effort bootstrap `sudo` symlink. Deterministic names and warm-pool keys use `sha256(user_id:thread_id)[:16]` (64-bit, matching E2B), with no full-seed fallback. Release parks VMs in-process; only the same user/thread reclaims them after a health check. Named terminal SDK errors and builtin `ConnectionError`/`BrokenPipeError`/`EOFError` evict via `_invalidate_sandbox`. Cross-process orphan reconciliation remains a follow-up. - Health probes require an exact `ok` line and reject `Error:`/`Exit Code:` diagnostic lines; other login-shell stdout/stderr is tolerated. Failures log sandbox id and output before eviction. +- `TenkiSandboxProvider` (`community/tenki/`): optional `tenki` SDK (`tenki_sandbox`), loaded lazily by `_import_client`. Sync SDK calls; native streaming file APIs, shared shell-search parsing, HOME remap for `/mnt/user-data`, best-effort bootstrap symlink. IDs use `sha256(user_id:thread_id)[:16]`; only that scope reclaims warm VMs. Health probes require exact `ok`, reject `Error:`/`Exit Code:`, tolerate login-shell output, and log failures. + Reset parks existing VMs but permanently fences acquisition; shutdown requests termination. SDK workers own late results through cleanup, including abandoned async waits and random scopes. Keep admission closure separate from shutdown completion; outstanding work raises for retry. Atomically transfer removed handles into identity-keyed pending cleanup, exclude them from reuse, count them toward capacity, and retry on idle sweeps/shutdown. Bind terminal callbacks to the originating adapter and check identity under `_lock` before invalidation: a late old command must never evict its replacement. Operational transport failures invalidate a handle, but only accepted termination or authoritative already-gone errors acknowledge close. No physical-stop or crash-recovery guarantee. [Tenki configuration and lifecycle](../community/tenki/README.md). - -**Shared warm-pool lifecycle:** `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin` owns `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, the idle-checker loop, warm expiry, oldest-warm eviction, replica counts and soft-cap logs. Providers own active registries, creation/discovery, health checks and `_destroy_warm_entry`: AIO destroys `SandboxInfo` via its backend; Boxlite closes loop-affine `BoxliteBox`; Tenki terminates the remote VM with `TenkiSandbox.close`. AIO delegates only warm expiry; active-idle cleanup stays outside the mixin. Shutdown must await the idle-checker thread before tearing down provider resources. `_stop_idle_checker()` uses a bounded join and raises on timeout; reset the provider's shutdown guard on failure so shutdown can retry. +**Shared warm-pool lifecycle:** `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin` owns `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, the idle-checker loop, warm expiry, oldest-warm eviction, replica counts and soft-cap logs. Providers own active registries, creation/discovery, health checks and `_destroy_warm_entry`: AIO destroys `SandboxInfo` via its backend; Boxlite closes loop-affine `BoxliteBox`; Tenki terminates the remote VM with `TenkiSandbox.close`. AIO delegates only warm expiry; active-idle cleanup stays outside the mixin. Shutdown must await the idle-checker thread before tearing down provider resources. `_stop_idle_checker()` uses a bounded join and raises on timeout; keep provider shutdown retryable after failure. Tenki keeps its admission fence set and uses a separate completion flag rather than reopening admission on retry; its atexit wrapper logs only incomplete-cleanup errors, while explicit shutdown still raises. **Virtual Path System**: - Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills` diff --git a/backend/tests/test_tenki_lifecycle.py b/backend/tests/test_tenki_lifecycle.py new file mode 100644 index 0000000000..79d1bb18c9 --- /dev/null +++ b/backend/tests/test_tenki_lifecycle.py @@ -0,0 +1,458 @@ +"""Offline ownership regressions using the real provider and adapter lifecycle.""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest + +from deerflow.community.tenki.provider import TenkiSandboxProvider +from deerflow.community.tenki.sandbox import TenkiSandbox + + +class Remote: + def __init__(self, errors=()): + self.errors = list(errors) + self.calls = 0 + self.closed = False + + def close(self): + self.calls += 1 + if self.errors: + raise self.errors.pop(0) + self.closed = True + + def wait_ready(self): + pass + + def exec(self, *args, **kwargs): + return SimpleNamespace(timed_out=False, exit_code=0, stdout_text="BOOTSTRAP_OK", stderr_text="") + + +@pytest.fixture +def exit_callbacks(monkeypatch): + callbacks = [] + monkeypatch.setattr("deerflow.community.tenki.provider.atexit.register", callbacks.append) + return callbacks + + +@pytest.fixture +def provider(monkeypatch, exit_callbacks): + config = dict(idle_timeout=0, replicas=3, api_key=None, base_url=None, workspace_id="synthetic", sticky=False, max_duration=60, image=None, cpu_cores=None, memory_mb=None, environment={}, home_dir="/home/tenki") + monkeypatch.setattr(TenkiSandboxProvider, "_load_config", lambda self: config) + value = TenkiSandboxProvider() + yield value + value.shutdown() + + +def test_atexit_logs_incomplete_cleanup_without_losing_retry(provider, exit_callbacks, caplog): + remote = Remote([OSError("synthetic termination unavailable") for _ in range(2)]) + box = TenkiSandbox("synthetic", remote) + provider._sandboxes[box.id] = box + with pytest.raises(RuntimeError, match="Tenki shutdown incomplete"): + provider.shutdown() + + caplog.clear() + exit_callbacks[0]() + assert remote.calls == 2 + assert provider._pending_closes[id(box)] is box + assert not provider._shutdown_complete + assert not provider._accepting + assert "Tenki shutdown incomplete: 0 acquires and 1 terminations pending" in caplog.text + assert all(record.exc_info is None for record in caplog.records) + + provider.shutdown() + assert remote.closed + assert remote.calls == 3 + assert provider._shutdown_complete + assert not provider._pending_closes + + +def test_atexit_success_is_quiet(provider, exit_callbacks, caplog): + remote = Remote() + box = TenkiSandbox("synthetic", remote) + provider._sandboxes[box.id] = box + exit_callbacks[0]() + exit_callbacks[0]() + assert remote.closed + assert remote.calls == 1 + assert provider._shutdown_complete + assert not caplog.records + + +def test_atexit_preserves_unexpected_shutdown_errors(provider, exit_callbacks, monkeypatch): + with monkeypatch.context() as patch: + patch.setattr(provider, "_stop_idle_checker", lambda: (_ for _ in ()).throw(RuntimeError("reaper still running"))) + with pytest.raises(RuntimeError, match="reaper still running"): + exit_callbacks[0]() + + +@pytest.mark.parametrize("replacement_location", ["active", "warm"]) +@pytest.mark.parametrize("first_close_fails", [False, True]) +def test_late_terminal_failure_cannot_evict_replacement(provider, replacement_location, first_close_fails): + started, release = threading.Event(), threading.Event() + old_remote = Remote([OSError("termination unavailable")] if first_close_fails else []) + new_remote = Remote() + remotes = iter([old_remote, new_remote]) + provider._client = SimpleNamespace(create=lambda **kwargs: next(remotes)) + sandbox_id = provider.acquire("thread", user_id="user") + old = provider.get(sandbox_id) + + def fail_old_command(*args, **kwargs): + if args[-1] == "old-in-flight": + started.set() + assert release.wait(5) + raise ConnectionError("old connection lost") + + def healthy_command(*args, **kwargs): + text = "BOOTSTRAP_OK\n" if "BOOTSTRAP_OK" in args[-1] else "ok\n" + return SimpleNamespace(timed_out=False, exit_code=0, stdout_text=text, stderr_text="") + + old_remote.exec = fail_old_command + new_remote.exec = healthy_command + with ThreadPoolExecutor(max_workers=1) as executor: + pending = executor.submit(old.execute_command, "old-in-flight") + try: + assert started.wait(5) + assert "old connection lost" in old.execute_command("first-failure") + assert provider.get(sandbox_id) is None + assert provider.acquire("thread", user_id="user") == sandbox_id + replacement = provider.get(sandbox_id) + assert replacement is not old + if replacement_location == "warm": + provider.release(sandbox_id) + finally: + release.set() + assert "old connection lost" in pending.result(timeout=5) + + tracked = provider.get(sandbox_id) if replacement_location == "active" else provider._warm_pool[sandbox_id][0] + assert tracked is replacement + assert not replacement.is_closed + assert new_remote.calls == 0 + assert old_remote.calls == 1 + assert (id(old) in provider._pending_closes) is first_close_fails + # The retained replacement still serves the same scope, including a real + # warm health probe. Fencing old failures must not reject healthy reuse. + assert provider.acquire("thread", user_id="user") == sandbox_id + assert provider.get(sandbox_id) is replacement + assert replacement.execute_command("echo ok") == "ok\n" + + +def test_current_generation_terminal_callback_still_evicts(provider): + remote = Remote() + provider._client = SimpleNamespace(create=lambda **kwargs: remote) + sandbox_id = provider.acquire("thread", user_id="user") + sandbox = provider.get(sandbox_id) + + def fail_command(*args, **kwargs): + raise ConnectionError("current connection lost") + + remote.exec = fail_command + assert "current connection lost" in sandbox.execute_command("fail") + assert provider.get(sandbox_id) is None + assert not provider._thread_sandboxes + assert sandbox.is_closed + assert remote.calls == 1 + + +@pytest.mark.parametrize("thread_id", ["thread", None]) +def test_shutdown_fences_late_creation(provider, monkeypatch, thread_id): + started, release = threading.Event(), threading.Event() + remote = Remote() + + def create(sandbox_id): + started.set() + assert release.wait(5) + return TenkiSandbox(sandbox_id, remote) + + monkeypatch.setattr(provider, "_create_sandbox", create) + with ThreadPoolExecutor(max_workers=1) as executor: + work = executor.submit(provider.acquire, thread_id, user_id="user") + try: + assert started.wait(5) + with pytest.raises(RuntimeError, match="acquir"): + provider.shutdown() + finally: + release.set() + with pytest.raises(RuntimeError, match="shut|retir"): + work.result(timeout=5) + provider.shutdown() + assert remote.closed + assert not provider._sandboxes + assert not provider._thread_sandboxes + + +@pytest.mark.parametrize("thread_id", ["thread", None]) +def test_reset_fences_late_creation_without_reparking(provider, monkeypatch, thread_id): + started, release = threading.Event(), threading.Event() + remote = Remote() + + def create(sandbox_id): + started.set() + assert release.wait(5) + return TenkiSandbox(sandbox_id, remote) + + monkeypatch.setattr(provider, "_create_sandbox", create) + with ThreadPoolExecutor(max_workers=1) as executor: + work = executor.submit(provider.acquire, thread_id, user_id="user") + try: + assert started.wait(5) + provider.reset() + finally: + release.set() + with pytest.raises(RuntimeError, match="shut|retir"): + work.result(timeout=5) + assert remote.closed + assert not provider._sandboxes + assert not provider._warm_pool + + +def test_shutdown_rejects_new_random_acquire(provider, monkeypatch): + created = [] + monkeypatch.setattr(provider, "_create_sandbox", lambda sandbox_id: created.append(sandbox_id)) + provider.shutdown() + with pytest.raises(RuntimeError, match="shut|retir"): + provider.acquire() + assert created == [] + + +@pytest.mark.parametrize("origin", ["idle", "replicas", "shutdown", "invalidation"]) +def test_failed_termination_retains_ownership_for_retry(provider, origin): + remote = Remote([OSError("synthetic termination unavailable")]) + box = TenkiSandbox("synthetic", remote) + if origin in {"idle", "replicas"}: + provider._warm_pool[box.id] = (box, 0) + if origin == "idle": + provider._reap_expired_warm(1) + else: + provider._evict_oldest_warm() + else: + provider._sandboxes[box.id] = box + if origin == "shutdown": + with pytest.raises(RuntimeError, match="terminat|cleanup"): + provider.shutdown() + else: + provider._invalidate_sandbox(box.id, "synthetic failure") + assert not remote.closed + assert provider.get(box.id) is None + provider.shutdown() + assert remote.closed + assert remote.calls == 2 + + +@pytest.mark.parametrize("error", [ConnectionError("lost ack"), BrokenPipeError("broken pipe"), EOFError("closed response")]) +def test_ambiguous_termination_is_retryable(error): + remote = Remote([error]) + box = TenkiSandbox("synthetic", remote) + with pytest.raises(type(error)): + box.close() + assert not box.is_closed + box.close() + assert box.is_closed + assert remote.calls == 2 + + +def test_readiness_failure_does_not_forget_failed_cleanup(provider): + remote = Remote([OSError("termination unavailable")]) + + def fail_ready(): + raise ValueError("synthetic readiness failure") + + remote.wait_ready = fail_ready + provider._client = SimpleNamespace(create=lambda **kwargs: remote) + with pytest.raises(ValueError, match="readiness"): + provider.acquire("thread", user_id="user") + provider.shutdown() + assert remote.closed + assert remote.calls == 2 + + +@pytest.mark.asyncio +async def test_cancelled_async_acquire_still_owns_late_creation(provider, monkeypatch): + started, release, finished = threading.Event(), threading.Event(), threading.Event() + remote = Remote() + + def create(sandbox_id): + started.set() + try: + assert release.wait(5) + return TenkiSandbox(sandbox_id, remote) + finally: + finished.set() + + monkeypatch.setattr(provider, "_create_sandbox", create) + work = asyncio.create_task(provider.acquire_async("thread", user_id="user")) + try: + assert await asyncio.to_thread(started.wait, 5) + work.cancel() + with pytest.raises(asyncio.CancelledError): + await work + with pytest.raises(RuntimeError, match="acquir"): + provider.shutdown() + finally: + release.set() + assert await asyncio.to_thread(finished.wait, 5) + # Join the already-started SDK worker before the fixture's retry. + await asyncio.to_thread(provider._acquire_serializer.executor.shutdown, wait=True) + provider.shutdown() + assert remote.closed + assert not provider._sandboxes + + +def test_idle_sweep_retries_failed_close_and_counts_pending_capacity(provider): + remote = Remote([OSError("try later")]) + box = TenkiSandbox("synthetic", remote) + provider._warm_pool[box.id] = (box, 0) + provider._cleanup_idle_resources(1) + assert provider._replica_count() == (3, 1) + provider._cleanup_idle_resources(1) + assert remote.closed + assert provider._replica_count() == (3, 0) + + +def test_replacement_does_not_erase_old_generation_cleanup(provider): + old_remote = Remote([OSError("old generation unavailable")]) + new_remote = Remote() + sandbox_id = provider._sandbox_id("thread", "user") + old_box = TenkiSandbox(sandbox_id, old_remote) + provider._sandboxes[sandbox_id] = old_box + provider._invalidate_sandbox(sandbox_id, "old transport broke") + provider._client = SimpleNamespace(create=lambda **kwargs: new_remote) + assert provider.acquire("thread", user_id="user") == sandbox_id + assert provider.get(sandbox_id) is not old_box + assert provider._replica_count() == (3, 2) + provider.shutdown() + assert old_remote.closed + assert new_remote.closed + assert old_remote.calls == 2 + assert new_remote.calls == 1 + + +def test_shutdown_and_reaper_share_single_termination_owner(provider): + started, release = threading.Event(), threading.Event() + remote = Remote() + original_close = remote.close + + def close(): + started.set() + assert release.wait(5) + original_close() + + remote.close = close + box = TenkiSandbox("synthetic", remote) + provider._warm_pool[box.id] = (box, 0) + with ThreadPoolExecutor(max_workers=1) as executor: + work = executor.submit(provider._reap_expired_warm, 1) + try: + assert started.wait(5) + with pytest.raises(RuntimeError, match="terminations pending"): + provider.shutdown() + finally: + release.set() + work.result(timeout=5) + provider.shutdown() + assert remote.calls == 1 + + +@pytest.mark.parametrize("origin", ["invalidation", "release"]) +def test_retirement_transfers_ownership_before_releasing_registry_lock(provider, monkeypatch, origin): + entered, resume = threading.Event(), threading.Event() + remote = Remote([OSError("temporary termination failure")]) + box = TenkiSandbox("synthetic", remote) + provider._sandboxes[box.id] = box + close = provider._close_quietly + main_thread = threading.current_thread() + + def paused_close(*args, **kwargs): + if threading.current_thread() is not main_thread: + entered.set() + assert resume.wait(5) + return close(*args, **kwargs) + + monkeypatch.setattr(provider, "_close_quietly", paused_close) + if origin == "release": + provider._shutdown_called = True + operation = (lambda: provider._invalidate_sandbox(box.id, "synthetic")) if origin == "invalidation" else (lambda: provider.release(box.id)) + with ThreadPoolExecutor(max_workers=1) as executor: + work = executor.submit(operation) + try: + assert entered.wait(5) + with pytest.raises(RuntimeError, match="terminations pending"): + provider.shutdown() + finally: + resume.set() + work.result(timeout=5) + provider.shutdown() + assert remote.closed + assert remote.calls == 2 + + +@pytest.mark.parametrize("thread_id", ["thread", None]) +def test_reset_rejects_new_acquires(provider, monkeypatch, thread_id): + created = [] + monkeypatch.setattr(provider, "_create_sandbox", lambda sandbox_id: created.append(sandbox_id)) + provider.reset() + with pytest.raises(RuntimeError, match="retired"): + provider.acquire(thread_id, user_id="user") + assert not created + + +def test_reaper_shutdown_failure_does_not_reopen_admission(provider, monkeypatch): + original = provider._stop_idle_checker + monkeypatch.setattr(provider, "_stop_idle_checker", lambda: (_ for _ in ()).throw(RuntimeError("reaper still running"))) + try: + with pytest.raises(RuntimeError, match="reaper"): + provider.shutdown() + with pytest.raises(RuntimeError, match="shut|retir"): + provider.acquire() + finally: + monkeypatch.setattr(provider, "_stop_idle_checker", original) + provider.shutdown() + + +def test_reset_preserves_existing_warm_resource_until_shutdown(provider): + remote = Remote() + box = TenkiSandbox("synthetic", remote) + provider._sandboxes[box.id] = box + provider.reset() + assert not remote.closed + assert box.id in provider._warm_pool + provider.shutdown() + assert remote.closed + + +@pytest.mark.parametrize("name", ["InvalidStateError", "StreamClosedError"]) +def test_non_terminal_sdk_state_error_does_not_acknowledge_termination(name): + error = type(name, (RuntimeError,), {})("termination not confirmed") + remote = Remote([error]) + box = TenkiSandbox("synthetic", remote) + with pytest.raises(type(error)): + box.close() + assert not box.is_closed + box.close() + assert remote.calls == 2 + + +@pytest.mark.asyncio +async def test_cancelled_async_acquire_remains_reusable_while_provider_open(provider, monkeypatch): + started, release = threading.Event(), threading.Event() + remote = Remote() + + def create(sandbox_id): + started.set() + assert release.wait(5) + return TenkiSandbox(sandbox_id, remote) + + monkeypatch.setattr(provider, "_create_sandbox", create) + work = asyncio.create_task(provider.acquire_async("thread", user_id="user")) + try: + assert await asyncio.to_thread(started.wait, 5) + work.cancel() + with pytest.raises(asyncio.CancelledError): + await work + finally: + release.set() + sandbox_id = await provider.acquire_async("thread", user_id="user") + assert provider.get(sandbox_id) is not None + assert not remote.closed diff --git a/backend/tests/test_warm_pool_lifecycle.py b/backend/tests/test_warm_pool_lifecycle.py index e0e2d53b3e..3b6b84d03b 100644 --- a/backend/tests/test_warm_pool_lifecycle.py +++ b/backend/tests/test_warm_pool_lifecycle.py @@ -100,7 +100,8 @@ def test_stop_idle_checker_rejects_live_thread_after_timeout() -> None: [ ("deerflow.community.aio_sandbox.aio_sandbox_provider", "AioSandboxProvider"), ("deerflow.community.boxlite.provider", "BoxliteProvider"), - ("deerflow.community.tenki.provider", "TenkiSandboxProvider"), + # Tenki separates the permanent admission fence from cleanup completion; + # test_tenki_lifecycle exercises its retry through the real constructor. ("deerflow.community.opensandbox.provider", "OpenSandboxProvider"), ], )