fix(sandbox): hold the teardown lease for as long as the container stop runs

claim(..., for_destroy=True) wrote the del: marker with the ordinary lease TTL
and nothing refreshed it. renew() extends only own: and deliberately reports a
teardown as LOST, and the destroy paths drop the sandbox from the maps the
renewal loop iterates — so a container stop that outlived the TTL let the marker
lapse, a peer's take() succeeded against the still-running container, and the
stop then landed on the turn that had just been handed it. That is the exact
window the del: state exists to close, reopened by its own expiry.

The two lease states alone never made the per-sandbox flock redundant, as I
claimed when deleting it: a held lock cannot expire, a lease can. The exclusion
has to be held deliberately rather than assumed to outlast the work it guards.

_held_teardown_lease wraps both _backend.destroy() call sites and re-claims the
marker every renewal_interval_seconds until the stop returns. No store change is
needed: claim(for_destroy=True) already refreshes an existing del: marker on
both backends.

Reachable without an abnormal backend. The schema bounds only
renewal_interval_seconds (> 0) and ttl_multiplier (>= 2), so a legal config puts
the TTL below a normal container stop; and LocalContainerBackend._stop_container
passes no timeout to subprocess.run, so a wedged daemon blocks unbounded even at
the default 120s TTL.

The TTL stays finite on purpose: the heartbeat dies with the process, so a
destroyer that crashes mid-stop still releases the container one TTL later
instead of marking it undestroyable forever.

backend/AGENTS.md

@@ -362,6 +362,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti
   - **Cross-instance ownership store** (`aio_sandbox/ownership/`, #4206): gateway instances sharing a container backend coordinate container ownership through a pluggable lease store, selected by `sandbox.ownership.type` (`memory` | `redis`) and resolved like `stream_bridge` (`factory.py`, lazy per-branch import, `redis` optional extra, `DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL` env escape hatch; a set `DEER_FLOW_STREAM_BRIDGE_REDIS_URL` implies a multi-instance deployment and infers `redis`). `memory` is single-instance only and declares `supports_cross_process = False`.
     - **A lease answers "who reaps this container", not "who may use it".** That splits the interface in two: `take()` transfers ownership on the **acquire** path (a container is deterministic per user/thread, so consecutive turns legitimately land on different instances — a conditional claim there would strand the thread until the previous lease expired), while `claim()` succeeds only if the container is unowned or already ours and gates every **adopt/reap** path. `release()` never clears a peer's lease.
     - **A lease carries a state, and that is what makes the destroy window safe.** `own:` = responsible for this container; `del:` = tearing it down (`claim(..., for_destroy=True)`). `take()` is refused against a `del:` lease, so a container cannot be re-acquired between a destroy path's claim and its container stop. Without the two states an unconditional `take()` would silently overwrite the destroyer's claim and the peer's stop would land on a container the new owner had already handed to an agent — i.e. #4206 again. That pairing is what replaced the previous same-host `flock` guard, which is gone; Redis makes the scope genuinely multi-instance instead of same-host. A destroyer that dies mid-stop leaves a `del:` marker that lapses with the TTL. On the acquire path a refused take raises `SandboxBeingDestroyedError`: the reuse/reclaim paths drop the container and cold-start, and the discover path propagates (falling through to create would collide with the not-yet-removed container name).
+    - **The `del:` state has to be *held* for the stop, not just written before it.** The two states alone do not make `flock` redundant — a held lock cannot expire, whereas a lease can, and `claim(..., for_destroy=True)` writes the marker with the ordinary lease TTL. Nothing else refreshes it: `renew()` extends only `own:` and deliberately reports a teardown as `LOST`, and the destroy paths drop the sandbox from the maps `_renew_owned_leases` iterates. So a container stop that outlived the TTL let the marker lapse, a peer's `take()` succeeded against the still-running container, and the stop then landed on the turn that had just been handed it — the exact window `del:` exists to close, reopened by its own expiry. `_held_teardown_lease` wraps both `_backend.destroy()` call sites and re-claims the marker every `renewal_interval_seconds` until the stop returns. This needs no abnormal backend: the schema bounds only `renewal_interval_seconds` (> 0) and `ttl_multiplier` (>= 2), so a legal config puts the TTL below a normal container stop, and `LocalContainerBackend._stop_container` passes no `timeout` to `subprocess.run`, so a wedged daemon blocks unbounded even at the default 120s TTL. The TTL stays finite on purpose — the heartbeat dies with the process, so a destroyer that crashes mid-stop still releases the container one TTL later instead of marking it undestroyable forever. Raising a separate teardown TTL instead would only be sufficient if it were bounded above every backend's real stop deadline.
     - **Fail-closed both directions.** Establishment: a sandbox whose ownership cannot be published is never handed out (a just-created container is destroyed rather than leaked as an adoptable orphan) — acquiring raises `OwnershipBackendError`, matching the stream bridge's fail-hard v1 policy. Reaping: a store that cannot answer is treated as peer-owned, so an outage never turns live peer containers into orphans. Every destroy path claims **before** untracking, so a refused claim cannot leave a container running and invisible to every map.
     - **Renewal is independent of `idle_timeout`** (`_start_lease_renewal`, own daemon thread; TTL = `renewal_interval_seconds × ttl_multiplier`). Renewal used to ride on the idle checker, which `__init__` only starts when `idle_timeout > 0` — so `idle_timeout: 0` ("keep warm VMs until shutdown", a documented config) let every lease lapse. Liveness and reaping must not share a switch. Renewal covers warm entries as well as active ones; losing a lease drops the sandbox from this instance's maps **without touching the container** (`_forget_lost_sandbox`) — destroying it there would be the very cross-instance kill this store prevents.
     - **`renew()` distinguishes lapsed from lost** (`RenewOutcome`), and the two must not be collapsed. `LAPSED` means the lease is simply absent — nobody took it — so `_refresh_ownership` re-establishes it; `LOST` means a peer holds it and it is never re-taken. Treating an absent lease as lost meant a Redis restart without persistence (every key gone) evicted every in-flight sandbox on every instance at once.

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

@@ -12,6 +12,7 @@ The provider itself handles:
 
 import asyncio
 import atexit
+import contextlib
 import hashlib
 import logging
 import os
@@ -366,6 +367,61 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             return False
         return False
 
+    @contextlib.contextmanager
+    def _held_teardown_lease(self, sandbox_id: str):
+        """Keep *sandbox_id*'s teardown marker alive for as long as its stop runs.
+
+        ``claim(..., for_destroy=True)`` writes the ``del:`` marker with the
+        ordinary lease TTL, and nothing else refreshes it: ``renew()`` extends
+        only ``own:`` and deliberately reports a teardown as ``LOST``, and the
+        destroy paths drop the sandbox from the maps ``_renew_owned_leases``
+        iterates. A container stop that outlives the TTL therefore let the marker
+        lapse, a peer's ``take()`` succeeded against the still-running container,
+        and the stop landed on the turn that had just been handed it — the very
+        window the ``del:`` state exists to close, reopened by its own expiry.
+
+        This is what the per-sandbox ``flock`` used to cover for free: a held lock
+        cannot expire. A lease can, so the exclusion has to be held deliberately
+        rather than assumed to outlast the work it guards. Reachable without an
+        abnormal backend — the config schema bounds only ``renewal_interval_seconds``
+        (> 0) and ``ttl_multiplier`` (>= 2), so a legal setting puts the TTL below a
+        normal container stop, and ``LocalContainerBackend._stop_container`` passes
+        no ``timeout`` to ``subprocess.run``, so a wedged daemon blocks unbounded
+        even at the default 120s.
+
+        The TTL stays finite on purpose: the heartbeat dies with the process, so a
+        destroyer that crashes mid-stop still releases the container one TTL later
+        instead of marking it undestroyable forever.
+        """
+        stop = threading.Event()
+
+        def beat() -> None:
+            interval = self._ownership_config.renewal_interval_seconds
+            while not stop.wait(interval):
+                try:
+                    if not self._ownership.claim(sandbox_id, for_destroy=True):
+                        # Only reachable if the store lost our marker *and* a peer
+                        # took it (e.g. a flush mid-stop). The stop is already in
+                        # flight and cannot be recalled, so say so loudly rather
+                        # than let a peer's container die without a trace.
+                        logger.error(
+                            "Lost the teardown exclusion for %s while its container stop was still in flight; a peer may have taken it",
+                            sandbox_id,
+                        )
+                        return
+                except OwnershipBackendError as e:
+                    # Unknown, not lost: the marker may well still be live, and
+                    # the TTL bounds how long a stale one survives. Retry.
+                    logger.warning("Could not refresh the teardown lease for %s, will retry: %s", sandbox_id, e)
+
+        beater = threading.Thread(target=beat, name="sandbox-teardown-lease", daemon=True)
+        beater.start()
+        try:
+            yield
+        finally:
+            stop.set()
+            beater.join(timeout=5)
+
     # ── Startup reconciliation ────────────────────────────────────────────
 
     def _adoptable_after_grace(self, sandbox_id: str, now: float) -> bool:
@@ -1106,7 +1162,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             return False
 
         try:
-            self._backend.destroy(entry)
+            # The marker must outlast the stop, not the TTL it was written with.
+            with self._held_teardown_lease(sandbox_id):
+                self._backend.destroy(entry)
         except Exception as e:
             if reason == "idle_timeout":
                 logger.error(f"Failed to destroy idle warm-pool sandbox {sandbox_id}: {e}")
@@ -1446,7 +1504,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 logger.warning(f"Error closing sandbox {sandbox_id} during destroy: {e}")
 
         if info:
-            self._backend.destroy(info)
+            # The marker must outlast the stop, not the TTL it was written with.
+            with self._held_teardown_lease(sandbox_id):
+                self._backend.destroy(info)
             logger.info(f"Destroyed sandbox {sandbox_id}")
 
         # Clears the teardown marker whether or not there was a container to

backend/tests/test_sandbox_orphan_reconciliation.py

@@ -1120,6 +1120,100 @@ def test_acquire_refuses_a_container_a_peer_is_destroying():
     assert "dying01" not in worker_a._sandboxes
 
 
+def test_teardown_marker_is_held_for_a_stop_that_outlives_the_lease_ttl():
+    """The `del:` state must not expire out from under an in-flight stop.
+
+    `test_acquire_refuses_a_container_a_peer_is_destroying` above proves the
+    marker refuses a takeover — but never lets it expire. `claim(for_destroy)`
+    writes it with the ordinary lease TTL and nothing refreshes it: `renew()`
+    only extends `own:` and reports a teardown as LOST, and the destroy paths
+    drop the sandbox from the maps the renewal loop iterates. So a container
+    stop that outlives the TTL let the marker lapse, a peer's `take()` then
+    succeeded against the still-running container, and the stop landed on the
+    turn that had just been handed it — the very window `del:` exists to close,
+    reopened by its own expiry. The `flock` this replaced could not expire; a
+    lease can, so it has to be held on purpose.
+    """
+    from deerflow.config.sandbox_config import SandboxOwnershipConfig
+
+    lease_ttl = 0.15
+    shared = _make_shared_ownership_store(ttl_seconds=lease_ttl)
+    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
+    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
+    # A legal config: the schema bounds only renewal > 0 and multiplier >= 2.
+    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05, ttl_multiplier=3.0)
+    info = SandboxInfo(
+        sandbox_id="doomed1",
+        sandbox_url="http://localhost:8080",
+        container_name="deer-flow-sandbox-doomed1",
+        created_at=time.time(),
+    )
+
+    stop_entered = threading.Event()
+    release_stop = threading.Event()
+
+    def slow_destroy(entry):
+        stop_entered.set()
+        release_stop.wait(timeout=5)
+
+    worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
+    worker_a._warm_pool["doomed1"] = (info, time.time())
+
+    reaper = threading.Thread(
+        target=lambda: worker_a._destroy_warm_entry("doomed1", info, reason="idle_timeout"),
+        daemon=True,
+    )
+    reaper.start()
+    try:
+        assert stop_entered.wait(timeout=5), "the reaper never reached the backend stop"
+
+        # Across a span several times the lease TTL, a turn for this thread must
+        # keep being refused — the container is still being stopped.
+        deadline = time.time() + lease_ttl * 4
+        while time.time() < deadline:
+            assert not worker_b._ownership.take("doomed1"), "a peer took a container whose stop was still in flight"
+            time.sleep(0.02)
+    finally:
+        release_stop.set()
+        reaper.join(timeout=5)
+
+    # Once the stop returns the marker is dropped, so the thread can cold-start.
+    assert shared.owner("doomed1") is None, "the teardown marker outlived the stop that justified it"
+    assert worker_b._ownership.take("doomed1") is True
+
+
+def test_teardown_heartbeat_stops_when_the_stop_returns():
+    """A finite TTL must survive the fix, or a crashed destroyer leaks forever.
+
+    The heartbeat is what holds the exclusion, so it has to die with the stop:
+    if it outlived the destroy the marker would be refreshed indefinitely and no
+    peer could ever adopt or recreate the container.
+    """
+    from deerflow.config.sandbox_config import SandboxOwnershipConfig
+
+    shared = _make_shared_ownership_store(ttl_seconds=0.15)
+    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
+    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05, ttl_multiplier=3.0)
+    info = SandboxInfo(
+        sandbox_id="doomed2",
+        sandbox_url="http://localhost:8080",
+        container_name="deer-flow-sandbox-doomed2",
+        created_at=time.time(),
+    )
+    worker_a._warm_pool["doomed2"] = (info, time.time())
+
+    assert worker_a._destroy_warm_entry("doomed2", info, reason="idle_timeout") is True
+
+    # Named rather than counted: threading.active_count() is global and other
+    # tests' idle-checker/renewal threads make it noise, so a count comparison
+    # here passes straight through a leak.
+    assert [t for t in threading.enumerate() if t.name == "sandbox-teardown-lease"] == [], "a teardown heartbeat thread outlived its stop"
+
+    # And nothing keeps refreshing the marker past its TTL.
+    time.sleep(0.3)
+    assert shared.owner("doomed2") is None
+
+
 def test_cached_sandbox_being_destroyed_is_dropped_not_reused():
     """The same window on the warm/in-process reuse path falls through cleanly."""
     shared = _make_shared_ownership_store()