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 <yinx65588-boop@users.noreply.github.com>
Co-authored-by: yinx65588-boop <272180046+yinx65588-boop@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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

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

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

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 ──────────────────────────────────────────────────

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`

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

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"),
     ],
 )