fix(sandbox): stop multi-worker orphan reconcile from killing peer sandboxes (#4221)

* fix(sandbox): stop multi-worker orphan reconcile from killing peer sandboxes

Docker sandboxes are shared across gateway workers, but each worker kept its
own in-memory warm pool. Startup reconciliation adopted every running
container, so a peer idle reaper could destroy sandboxes another worker still
owned and tool calls hit 502 / Connection refused.

Add file-based ownership leases under sandbox-leases/, only adopt true
orphans, refuse idle/replica/shutdown destroy while a foreign lease is live,
and renew the lease on create/get/release/reclaim.

Fixes #4206

* fix(sandbox): close lease fail-open, hot-path IO, and check→destroy race

Address review of the multi-worker orphan lease (#4206):

- read_lease returns None only for a genuinely-absent lease and raises
  (CorruptLeaseError/OSError) when a lease is unreadable or corrupt, so the
  ownership check fails closed instead of mistaking an unprovable peer lease
  for a free container. clear_lease still removes a stuck/corrupt file.
- get() no longer renews the lease (blocking mkdir/fsync/os.replace on the
  event loop path used by ensure_sandbox_initialized_async); active leases are
  renewed off the event loop from the idle checker (_renew_active_leases).
- The ownership check and container stop run under a per-sandbox flock guard
  (lease_ownership_guard); every lease write takes the same guard so a peer's
  touch cannot interleave with a destroy. Same-host multi-worker scope, not a
  multi-pod distributed lock.

Also fixes the ruff format lint on the branch. Adds regression tests: corrupt
and unreadable lease fail closed, a tests/blocking_io anchor keeping get()
non-blocking on the event loop, and a peer-touch/destroy interleave test.

* fix(sandbox): share container ownership across gateway instances

Rework of the #4206 fix per review: ownership state is shared through a
third-party service instead of being maintained per gateway instance,
following the stream_bridge precedent (sandbox.ownership.type:
memory | redis). The file lease and its same-host flock guard are
deleted, not ported — they only covered workers on one host, while the
deployment that hits #4206 is a load-balanced multi-instance gateway.

A lease answers "who reaps this container", not "who may use it".
Containers are deterministic per (user, thread), so consecutive turns
legitimately land on different instances: take() transfers ownership on
acquire, while claim() gates every adopt/reap path.

Leases carry a state — own: or del: — so a takeover is refused against a
teardown in progress. Without it an unconditional take() would overwrite
a destroyer's claim and the peer's container stop would land on a
sandbox the new owner had already handed to an agent.

renew() distinguishes a lapsed lease from one a peer took; only the
latter drops the sandbox. Collapsing them meant a Redis restart evicted
every in-flight sandbox on every instance at once.

Renewal runs on its own thread with a TTL derived from its interval,
never from idle_timeout: renewal used to ride the idle checker, which
does not start at idle_timeout: 0, so leases silently lapsed on a
supported config.

Ownership establishment is fail-closed: a sandbox whose ownership cannot
be published is never handed out, and a just-created container is
destroyed rather than leaked as an adoptable orphan. Every destroy path
claims before untracking.

The memory store is single-instance only and says so; the resolver reads
app_config.stream_bridge and the env var in the bridge's own order, so
deployments already using Redis get a redis ownership store without
extra config.

* fix(sandbox): wait out a recovery grace before adopting a keyless container

An absent ownership lease meant two opposite things on two paths. Renewal
reads it as LAPSED and re-establishes it: nobody took the lease, so the
container is still ours. Reconciliation read the same absent key as "orphan"
and adopted on sight.

After the store loses its keys (a Redis restart without persistence, or
eviction under maxmemory) every owner is alive and merely pre-renewal-tick.
Whichever instance reconciled first therefore adopted every live container;
each real owner's next renewal reported LOST and dropped a sandbox it was
serving mid-turn, leaving it for the adopter to idle-destroy — #4206 through
the back door, in the very case the LAPSED handling was added to make safe.
Not limited to startup: an already-running instance hits the same window from
the idle checker's periodic reconcile.

_adoptable_after_grace requires an untracked container to be seen unowned
across a full lease TTL before it can be adopted. That rebuilds the delay the
state loss erased: a live owner republishes within one renewal interval,
shorter than the TTL by construction, while a crashed owner never does, so its
containers are still adopted one grace later rather than leaking. A republished
lease resets the grace; a pausing-only timer would still expire over a live
owner's lease. The peek is read-only — the atomic claim still gates adoption.

The grace is skipped when the store cannot coordinate across processes: no peer
can hold a lease such a store would show us, so single-instance deployments
keep instant orphan cleanup, and a grace could not help a multi-worker gateway
on memory anyway.

* 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.

* fix(sandbox): hold the teardown lease on every del: stop, and pin the claims that had no test

90936b49 said `_held_teardown_lease` wrapped "both" `_backend.destroy()` call
sites. There are three. `_drop_unhealthy_sandbox` marks `del:` and then blocks on
the same unbounded stop, and it untracks *before* claiming, so `_renew_owned_leases`
cannot see the id either — nothing refreshed the marker. Reproduced against a real
redis: the peer's `take()` succeeds 1.0s into a 2.5s stop. Same window, third path.

That miss came from the habit the rest of this commit addresses: a property
asserted in prose, with no test that could falsify it. Auditing every load-bearing
claim in this feature — AGENTS.md, the store docstrings, the provider's design
comments — against the test that would go red turned up several more, each
verified by mutating the code and watching the suite stay green.

Tests that could not fail:

- `test_reconcile_fails_closed_when_ownership_unknown` reached the grace gate, not
  the claim. A bare MagicMock answers `owner()` with a truthy mock, so the
  container read as peer-owned and deferred; `claim()` was never called. It stayed
  green with `_claim_ownership` failing open. Adding the grace ahead of the claim
  is what hollowed it out — inserting a gate can silently disarm the tests for
  the gate behind it.
- `test_adoption_grace_restarts_when_a_live_owner_republishes` never distinguished
  reset from pause. Those diverge only on a *second* lapse, which it never drove,
  so it passed with the reset deleted.

Claims with no test at all, each now pinned (mutation → red, per test):

- `destroy()`, `_evict_oldest_warm`, `_reclaim_warm_pool_sandbox`,
  `_register_created_sandbox` and `shutdown()`'s warm loop were each the one
  untested sibling of an "every path does X" enumeration. `shutdown()` was never
  driven with a non-empty warm pool, so a loop bypassing the ownership claim —
  stopping a live peer's container on our exit — went unnoticed.
- Renewal's unknown-is-not-lost rule, the single deliberate exception to
  fail-closed. Inverting it drops every active and warm sandbox on every instance
  the moment the store blinks.
- Both hops of the stream-bridge redis inference. Deleting either left the suite
  green while every config.yaml-native multi-instance deployment silently fell
  back to memory — #4206 reopened on exactly the deployments the inference exists
  for.

Claims narrowed instead, because they promised more than the code delivers:

- "run against both backends ... cannot drift" — CI provisions no redis, so the
  merge gate runs the memory tier only and the Lua never executes there.
- "Every destroy path claims before untracking" — `_drop_unhealthy_sandbox`
  untracks first, deliberately, under its `expected_info` TOCTOU guard.
- "Atomic: concurrent claims from different instances cannot both succeed" — true
  via Lua on redis, vacuous on the single-instance memory store, and pinned by
  neither, since the contract suite drives sequential calls. A concurrency test
  against the memory store would make the claim look covered while the mechanism
  that carries it still never runs in CI.

* fix(sandbox): release the teardown marker when a destroy() stop fails

The three `del:`-marked stop paths disagreed on failure. `_destroy_warm_entry`
releases on both outcomes and says why: the stop failed, so the container is
probably still up, and a marker left behind refuses its own thread's `take()`
until the TTL lapses. `_drop_unhealthy_sandbox` does the same. `destroy()` had no
such guard — a raising backend propagated straight past `_release_ownership`, and
the thread could not re-acquire for a full TTL.

Fails safe rather than fatal: a stuck marker stops peers from touching the
container, it is not the cross-instance kill. But the paths must agree, and this
one is the odd one out.

Release, then re-raise. Swallowing would be the easier symmetry with
`_destroy_warm_entry`'s `return False`, but `destroy()` has no failure return and
`shutdown()` logs per sandbox off the exception, so swallowing would silently
narrow what callers can see.

Found by comparing the three paths after @fancyboi999 asked for release to be
handled "consistently with the other destroy paths" on the unhealthy path — which
0d2377b2 already does. This is the sibling that wasn't.

* fix(deploy): bump chart config_version to 27 for sandbox.ownership

config.example.yaml went to 27 with the new sandbox.ownership section, but
the chart embeds its own copy and stayed at 26, so validate-chart failed.

A bare bump: the chart already sets stream_bridge.type=redis, which is what
resolve_ownership_config infers a redis ownership store from, so no field
change is needed.

* fix(sandbox): release the teardown lease from its heartbeat, not the caller

`_held_teardown_lease` joined its heartbeat only briefly and the caller
cleared the `del:` marker right after the stop. A refresh `claim` still in
flight (`RedisOwnershipStore` had no socket timeout, so a round trip could
block) could land *after* that release and rewrite `del:` on a container
whose stop had already completed — refusing a fresh `take()` (or rolling
back a fresh create) until the TTL.

Move the release into the heartbeat's own `finally`, after its loop stops,
so no refresh can run after it. The three destroy paths no longer release
after the `with` (`destroy()`'s no-container branch still does, since no
lease was held there). Bound every store round trip with a socket timeout
so the in-flight refresh — and thus the deferred release — stays finite,
and broaden the heartbeat's `except` so an unexpected error cannot strand
the marker during a long stop.

Also fold in the review follow-ups: stop re-resolving an already-resolved
ownership config in the factory, document the Redis-outage-vs-TTL boundary
in config.example.yaml, and add a tests/blocking_io anchor pinning that
`release()`'s store round trip stays off the event loop.

* fix(sandbox): refuse a non-destroy claim that would unwind our own teardown

`claim(for_destroy=False)` against our own `del:` lease fell through and
overwrote it with `own:`, cancelling a teardown that was already in flight.
The container stop cannot be recalled, so downgrading the marker would let a
`take()` hand out a container that is about to die -- #4206, self-inflicted.

No caller does this today: the two non-destroy callers run against an absent
key (the LAPSED re-claim) or an unowned one (post-grace reconcile). The
contract has to forbid it rather than rely on that staying true.

Fixed in both backends. The redis rule lives in Lua and the memory rule in
Python, so fixing one only would let them drift silently -- and the shared
contract suite is what is supposed to catch that drift, so it now covers this.

Also adds a contention test for `claim`. The suite drove sequential calls
only, so it pinned the exclusion predicate but not the atomicity that
predicate depends on; eight instances now race for one container and exactly
one must win.

* fix(sandbox): bound the container stop so it cannot outlive its teardown lease

`_stop_container` passed no `timeout` to `subprocess.run`, so a wedged
container runtime blocks it forever. The `del:` marker is what keeps a peer
from re-acquiring the container while the stop runs, but a marker is a lease
and a lease can lapse: a store outage longer than the TTL frees it, a peer's
`take()` succeeds against the still-running container, and the stop then
lands on the turn that was just handed it -- the exact #4206 failure.

The teardown heartbeat already covers the case where the store stays
reachable. This bounds the worst case independently of the ownership layer,
which is the point: it holds even when the ownership layer is the thing that
failed.

A timeout is not swallowed like a `CalledProcessError`. That error means the
runtime answered "I could not stop it"; a timeout means we do not know, and
the container is probably still running -- returning normally would let
`_destroy_warm_entry` report a clean stop and drop the warm entry, leaking a
running container nothing tracks.

* fix(sandbox): exclude this instance's own reapers from its acquire path

An ownership lease excludes peers and nothing else. `claim()` and `take()`
both succeed against our own `own:` lease by design -- that is what lets a
destroy path claim what it already owns -- so `del:` says nothing to this
process's other threads. Meanwhile every reaper decides outside `_lock`,
because a store round trip must not be held under the lock that guards every
acquire. So each reaper acts on a decision its own acquire path may already
have invalidated, and the store cannot see the difference.

Six paths end in an irreversible act (a container stop, or closing a
host-side client) on a decision made outside the lock. All six reproduce:

  _evict_oldest_warm      re-checks warm membership, then releases the lock
  _reap_expired_warm      no re-check at all
  _cleanup_idle_sandboxes re-verifies idle, then releases the lock
  _renew_owned_leases     acts on a stale renew() -> LOST
  release()               same staleness on its own refresh
  _drop_unhealthy_sandbox untracks before claiming, opening discovery

Both warm reapers are a regression from the deferred pop this branch
introduced: `WarmPoolLifecycleMixin` popped under the lock, so a reclaim's
membership check failed and the race could not occur. Deferring the pop is
still right (popping first loses the container on a refused claim), so the
exclusion has to be made explicit instead. The idle path is pre-existing in
shape, but this branch widened it from a few instructions to a network round
trip by claiming ownership before untracking.

Two guards, because the two directions want opposite answers:

Reaping -- nothing may promote it. The reaper reserves the id, and every
promote path refuses a reserved id exactly as it refuses a peer's `del:`
(drop and cold-start). The "is this still reapable?" test travels with the
reservation as a predicate and runs in the same critical section, because
checking first and reserving second is the window, not a narrower version of
it.

Forgetting -- the peer legitimately wins, so the promote is what to detect.
`_publish_ownership` bumps a per-id acquire epoch; the callers that decide
from a store round trip snapshot it first, and the pop is skipped if it
moved. Object identity cannot substitute: the reuse path re-publishes
ownership while handing out the same tracked `AioSandbox`, so an identity
check sees nothing and the pop closes a client mid-turn.

`still_reapable` is required rather than defaulting to unconditional -- the
safe default is the one that makes a new call site think about it. That
diverges from the mixin hook, which is safe because this provider overrides
both mixin callers, and loud rather than silent if those are ever dropped.

Also closes a client leak on the discover path: "nothing to roll back" was
true of the container but not of the HTTP client constructed before the
publish, which the sibling create path already closes.

The shared-store test view rebound `owner_id` outside the store's lock, so a
concurrent claim could execute under the wrong id and read its own lease as a
peer's. Serialized, so the heartbeat-hold tests stop flaking.

* fix(sandbox): mark acquire intent before the ownership round trip

A guard must become visible no later than the transition it guards. The
acquire epoch cannot manage that for `take()`: the takeover is durable before
`take()` returns -- redis has committed the SET while the reply is still in
flight -- and the epoch can only be written afterwards. In that interval the
store already says the container is ours while the epoch still reads as it
did when a renewal decided `LOST`, so the stale forget walks through, drops
the maps and closes the client the acquire is about to hand back. Acquire
then returns an id the provider no longer tracks and `get()` answers `None`
for the rest of the turn.

`_publish_ownership` now publishes an intent mark under `_lock` before the
round trip; the epoch keeps covering the other half, "an acquire completed
since you decided". `_forget_lost_sandbox` honours the intent mark
unconditionally rather than only when an epoch is supplied -- today's
epoch-less callers cannot reach the window, but "no epoch" reading as "no
guard" is how the next caller of a dangerous primitive gets written.

The same invariant had four more instances, all reproduced:

  reuse returns a decision the forget already invalidated -- before the mark
    is set a `LOST` is both current and correct, so the forget legitimately
    runs and the entry reuse decided to hand out is gone. Re-check after
    publishing and fall through to discovery instead.
  reclaim installs an entry a reaper reserved after its check -- the warm
    entry is still visible during the stop, and the reaper's claim succeeds
    because reclaim's own take() just made the lease ours. Re-check likewise.
  the reservation was released before the entry was removed -- the pop
    belonged to the caller, leaving a gap where the container is stopped, the
    entry is still in `_warm_pool`, and nothing marks it.
    `_destroy_warm_entry` removes it itself, inside the reservation; the pop
    stays deferred relative to the stop, just not to the reservation.
  reconcile adopts a container this instance is tearing down -- adoption is a
    promote and needs the same reservation check as the others. Neither
    existing guard excludes it: the claim succeeds because the lease is ours,
    and on `memory` the recovery grace is skipped outright.

The pre-round-trip checks in reuse and reclaim are kept as early-outs, since
they skip a health check and a store round trip on a doomed entry, and are
pinned to that job rather than to a correctness role they no longer hold.

The teardown reservation predicate runs under `_lock`, so it must not touch
the lock. Documented rather than engineered around: making the lock reentrant
to tolerate it would trade a loud hang for a quiet class of re-entrancy bugs
across the rest of the provider.

* fix(sandbox): honor local teardown after ownership publish

* fix(sandbox): clear a stale warm entry when an id becomes active

Active and warm are exclusive states, and the two register paths were the
only place that could hold both: they inserted into `_sandboxes` without
popping `_warm_pool`, so one container ended up with two reapers.
`_reap_expired_warm` judges an entry by its warm timestamp and never
consults `_last_activity`, so it stops a container an agent is actively
using while `_sandboxes` still hands out its client.

Reachable because `_reconcile_orphans` adopts an untracked-but-running
container into the warm pool inside the register's publish -> track
window, and on the `memory` store it adopts on sight:
`_adoptable_after_grace` short-circuits when `supports_cross_process` is
False, so an id carrying this process's own lease reads as adoptable.
That window is new to this branch -- on main the track was a single
locked insert with nothing before it.

Both register paths now pop the warm entry inside the same locked
section that installs the active one.

* fix(sandbox): harden ownership renewal teardown

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/AGENTS.md

@@ -372,7 +372,29 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti
 **Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved.
 **Implementations**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Legacy global-custom mounts are gated by the same user-scoped skill discovery rule used for prompt/list visibility; providers must not infer visibility from raw directory presence alone.
-- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths. Legacy global-custom mounts follow the same shared visibility helper as local and remote providers.
+- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths — it never touches the ownership store (that would be blocking IO on the event loop); ownership is published on acquire/reclaim and refreshed off the event loop by the dedicated renewal thread (`_renew_owned_leases`). Legacy global-custom mounts follow the same shared visibility helper as local and remote providers.
+  - **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 **every** `del:`-marked stop — `_destroy_warm_entry`, `destroy()`, and `_drop_unhealthy_sandbox` — and re-claims the marker every `renewal_interval_seconds` until the stop returns. `_drop_unhealthy_sandbox` needs it most: it untracks *before* claiming (under its `expected_info` TOCTOU guard), so `_renew_owned_leases` cannot see the id either. **The final release is the heartbeat's own last act, not the caller's** — a refresh `claim` still in flight when the context exits (the store's socket timeout bounds it, but it can be mid-round-trip) would otherwise land *after* a caller-side release and rewrite `del:` on a container whose stop already completed, stranding a fresh `take()` (or rolling back a fresh create) until the TTL. Releasing from inside the heartbeat, after its loop stops, sequences the release strictly after the last refresh, so no claim can follow it; the context join is bounded and, on a genuine wedge, defers the release to that thread rather than clearing the marker itself. This covers a **failed** stop too (the container is probably still up, and a marker left behind refuses its own thread's `take()` until the TTL lapses); `destroy()` still lets the error propagate out of the `with` — `shutdown()` logs per sandbox off it. `RedisOwnershipStore` sets a `socket_timeout` so no store round trip — and so no heartbeat refresh — can block unbounded, keeping that deferred release finite. 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. `LocalContainerBackend._stop_container` now passes a `timeout` to `subprocess.run` (`_STOP_TIMEOUT_SECONDS`) so a wedged daemon cannot block unbounded — that bounds the residual window independently of the ownership layer, for the case where the `del:` marker lapses mid-stop (a store outage longer than the TTL) and the stop then lands on a container a peer has been handed. A timed-out stop propagates rather than being swallowed like a `CalledProcessError`: the container is probably still running, so reporting a clean stop would drop the warm entry and leak it. 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. **Renewal is the deliberate exception**: an unanswerable store there means *unknown*, not lost, so `_refresh_ownership` keeps the sandbox and retries — failing closed on that path would evict every live sandbox on every instance the moment the store blinked. The TTL still bounds how long a genuinely dead owner holds a lease. Both paths that stop a container they still track — `destroy()` and `_destroy_warm_entry` — claim **before** untracking, so a refused claim cannot leave a container running and untracked. (`_drop_unhealthy_sandbox` untracks first, under its `expected_info` TOCTOU guard, then claims before the stop; a refused claim there leaves the container to the next reconcile, which re-adopts it after the grace.)
+    - **A lease excludes peers, never ourselves — same-process exclusion is the provider's job.** `claim()` and `take()` both succeed against this instance's own `own:` lease by design (that is what lets a destroy path claim what it already owns), so `del:` says nothing to this process's *other* threads. Every reaper — idle checker, renewal, warm eviction, unhealthy drop — decides outside `self._lock`, because a store round trip must not be held under the lock that guards every acquire; so each one acts on a decision its own acquire path may already have invalidated. Two guards cover the two directions, and both live in `AioSandboxProvider`, not the store:
+      - **Reaping** (`_reserve_local_teardown` / `_local_teardown`): the reaper marks the id, and every promote path — `_reuse_in_process_sandbox`, `_reclaim_warm_pool_sandbox`, `_register_discovered_sandbox` — refuses a marked id exactly as it refuses a peer's `del:` (drop and cold-start). The "is this still reapable?" check runs **in the same critical section as the mark**, passed down as a `still_reapable` predicate rather than run by the caller beforehand: checking first and marking second *is* the window, not a narrower version of it. This matters most where the entry deliberately stays visible during the stop — both warm reapers defer their pop so a refused claim cannot lose the container — and where the maps are cleared first (`_drop_unhealthy_sandbox`), which leaves backend discovery as the open path. On `main` the mixin's `_evict_oldest_warm` / `_reap_expired_warm` popped under the lock, so the deferred pop is what made this reachable.
+      - **Forgetting** (`_acquire_epoch`): when `renew()` reports `LOST` the peer legitimately wins, so here the *promote* is the thing to detect. `_publish_ownership` bumps a per-id acquire epoch; `_renew_owned_leases` and `release()` snapshot it before the round trip and hand it to `_forget_lost_sandbox`, which skips the pop if it moved. Object identity is not enough: the reuse path re-publishes ownership while handing out the **same** tracked `AioSandbox`, so an identity check sees nothing and the pop closes a client mid-turn.
+      - **A guard must become visible no later than the transition it guards.** The epoch cannot satisfy that for `take()`: the takeover is durable before `take()` returns (redis has committed the SET while the reply is in flight), and the epoch can only be written afterwards, so a renewal holding an older `LOST` walks through the gap, drops the maps, and closes the client the acquire is about to hand back — acquire then returns an id whose `get()` is `None`. `_publish_ownership` therefore publishes an **intent mark** (`_acquire_inflight`) under `_lock` *before* the round trip; the epoch covers the other half, "an acquire completed since you decided". `_forget_lost_sandbox` honours the intent mark unconditionally, not only when an epoch is supplied — "no epoch" must not read as "no guard".
+      - **A reservation must cover the removal, not just the stop.** `_destroy_warm_entry` pops the warm entry itself, inside the reservation. Releasing the reservation when the stop returns and letting the caller pop afterwards leaves a gap where the container is stopped, the entry is still in `_warm_pool`, and nothing marks it — a reclaim there hands out a dead container. The pop stays deferred relative to the *stop* (a refused or failed stop keeps the entry), just no longer relative to the reservation.
+      - **A check taken before a round trip must be retaken after it.** `_reuse_in_process_sandbox` re-verifies both its map entry and the local teardown reservation, `_reclaim_warm_pool_sandbox` re-checks the reservation, and `_register_discovered_sandbox` re-checks before installing its client, all after publishing ownership. Before the intent mark is set a renewal's `LOST` is both current and correct, so the forget can legitimately remove the entry the acquire decided to hand out; independently, a local reaper can reserve an id while reuse is outside `_lock` for its health/store calls and deliberately leaves the map entry present until its destroy claim succeeds. Falling through re-discovers or cold-starts instead of returning/installing a client for either stale decision. The pre-round-trip checks remain as early-outs that skip backend and store work on an already-doomed entry.
+      - **Adoption is a promote too.** `_reconcile_orphans` honours the reservation: a container being torn down is untracked and still running, which is exactly the shape that loop adopts, and neither the claim (ours) nor the recovery grace (skipped entirely on `memory`, where `supports_cross_process` is `False`) excludes it.
+      - **Active and warm are exclusive, and only a promote can violate it.** Both register paths pop `_warm_pool` inside the same locked section that inserts into `_sandboxes`: a warm entry for an id is stale the moment that id becomes active, and leaving it gives the container *two* reapers — `_reap_expired_warm` judges it by the warm timestamp and never consults `_last_activity`, so it stops a container an agent is using while `_sandboxes` still hands out its client. Reachable because reconciliation adopts into the warm pool inside the register's publish → track window, and on `memory` it adopts on sight (`_adoptable_after_grace` short-circuits when `supports_cross_process` is `False`, so an id carrying this process's own lease reads as adoptable). On `main` the track was a single locked insert with nothing before it, so the window did not exist.
+      A non-destroy `claim()` is the one case the store does police against its own owner: it refuses to overwrite our own `del:`, because the stop it marks is already in flight and downgrading the marker would let a `take()` hand out a container about to die. Enforced in both backends (Lua and Python) so they cannot drift.
+    - **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.
+      - A warm teardown is the local exception to that forget rule: `_destroy_warm_entry` deliberately keeps the entry in `_warm_pool` until the backend stop succeeds, while its own `del:` marker makes ordinary `renew()` report `LOST`. `_forget_lost_sandbox` therefore honours `_local_teardown`; otherwise the renewal thread can pop the retained entry mid-stop and a failed stop leaves a running container untracked.
+    - **`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.
+      - Renewal's fail-open rule covers both store round trips. If `renew()` returns `LAPSED` but the follow-up `claim()` cannot answer, ownership is still unknown rather than lost, so the provider keeps the sandbox and retries. The ordinary `_claim_ownership` helper remains fail-closed for adopt/reap callers and is intentionally not used for this re-claim.
+    - **Teardown join budget covers refresh plus release.** Redis bounds each ownership operation at five seconds, and context exit can catch the heartbeat in one final refresh before its `finally` performs the final release. `_TEARDOWN_JOIN_TIMEOUT_SECONDS` is therefore 12 seconds — greater than both sequential operation bounds — so a normal pair of socket timeouts does not emit the deferred-release warning; a still-running heartbeat continues to own the release safely.
+    - **An absent lease means the same thing on both paths, and reconciliation must say so too.** The `LAPSED` rule above only covers an owner renewing its *own* lease; on its own it does not make state loss safe, because reconciliation reads the same absent key as "orphan, adopt". After a Redis flush (restart without persistence, or eviction under `maxmemory`) every owner is alive and merely pre-renewal-tick, so whichever instance reconciles first would adopt every live container, each real owner's next renewal would report `LOST`, and it would drop a sandbox mid-turn for the adopter to idle-destroy — #4206 through the back door. `_adoptable_after_grace` closes it: an untracked container must be seen unowned (`owner()`, a read-only peek — the atomic `claim()` is still what actually gates adoption) across a full lease TTL before it can be adopted, tracked per container in `_unowned_since`. That rebuilds the delay the flush erased — a live owner republishes within one renewal interval, shorter than the TTL by construction (`ttl_multiplier >= 2`) — while a genuinely crashed owner never republishes, so its containers are still adopted one grace later rather than leaking. A republished lease **resets** the grace; a pausing-only timer would still expire over a live owner's lease. The grace is skipped when `supports_cross_process` is `False`: no peer can hold a lease such a store would show us, so single-instance deployments keep instant orphan cleanup, and a grace could not help a multi-worker gateway on `memory` anyway (peers are invisible to each other's leases with or without it).
+    - **The `memory` store is single-instance only** and says so via `supports_cross_process = False`; the provider logs a warning at startup when the configured store cannot see peers. A multi-worker gateway on `memory` has no cross-process coordination at all — same contract as `stream_bridge`'s memory backend. This is why the redis inference matters: it reads `app_config.stream_bridge` **and** the env var, in the same order the bridge's own resolver does, so any deployment already pointing the bridge at Redis (i.e. every multi-instance one) gets a redis ownership store without extra config.
+    - `get()` stays a pure in-memory lookup and must never call the store (that is blocking filesystem/network IO on the event loop); anchored by `tests/blocking_io/test_aio_sandbox_get.py`, which injects a deliberately-blocking probe store so the anchor keeps its teeth regardless of the configured backend. Tests: `tests/test_sandbox_ownership_store.py` (store contract, defined once for **both** backends — but the redis tier is `@pytest.mark.integration` + opt-in via `DEER_FLOW_TEST_REDIS_URL` and self-skips, and **CI provisions no redis**, so the merge gate runs the memory tier only and the Lua scripts never execute there; drift between the backends is caught only when the suite runs against a live redis. There is no fake-redis tier because the redis exclusion lives in Lua a fake would not execute) and `tests/test_sandbox_orphan_reconciliation.py` (provider behaviour, two providers sharing one store).
 - `BoxliteProvider` (`packages/harness/deerflow/community/boxlite/`) - BoxLite micro-VM isolation. The `boxlite` runtime is optional (`deerflow-harness[boxlite]`) and lazy-imported only when this provider is selected. The provider owns one private asyncio event loop on a daemon thread because BoxLite handles are loop-affine; sync `Sandbox` calls marshal onto that loop with `run_coroutine_threadsafe`.
   Boxes are named deterministically from `user_id:thread_id`, released into an in-process warm pool after each agent turn, and reclaimed only by the same user/thread. Warm-pool health checks use a short explicit timeout and forward that timeout through both BoxLite `exec(timeout=...)` and the private-loop `.result(timeout)` bridge so a hung VM cannot pin the per-thread acquire lock indefinitely.
   `sandbox.replicas` caps active + warm VMs per gateway process; if capacity is exhausted, only warm-pool VMs are evicted. `sandbox.idle_timeout` stops idle warm VMs after the configured seconds. `reset()` is intentionally a lightweight registry clear for `reset_sandbox_provider()` and does not close boxes, stop the idle reaper, or close the private loop; full teardown remains `shutdown()`.

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

@@ -200,6 +200,11 @@ class LocalContainerBackend(SandboxBackend):
     - Support for volume mounts and environment variables
     """
 
+    # Wall clock for a single `stop`. Comfortably above the runtime's own default
+    # SIGKILL escalation (10s for docker/podman), so this only fires when the
+    # daemon itself is wedged rather than truncating a slow-but-progressing stop.
+    _STOP_TIMEOUT_SECONDS = 120.0
+
     def __init__(
         self,
         *,
@@ -603,15 +608,30 @@ class LocalContainerBackend(SandboxBackend):
             raise RuntimeError(f"Failed to start sandbox container: {e.stderr}")
 
     def _stop_container(self, container_id: str) -> None:
-        """Stop a container (--rm ensures automatic removal)."""
+        """Stop a container (--rm ensures automatic removal).
+
+        The timeout bounds the worst case independently of the ownership layer.
+        The teardown lease keeps a peer from re-acquiring the container while
+        this runs, but that exclusion is a lease and can lapse (a store outage
+        longer than the TTL); an unbounded ``docker stop`` against a wedged
+        daemon could then outlive it and land on a peer's live container — #4206.
+        Bounding the stop caps how long that exposure can last even when the
+        store is perfectly healthy.
+        """
         try:
             subprocess.run(
                 [self._runtime, "stop", container_id],
                 capture_output=True,
                 text=True,
                 check=True,
+                timeout=self._STOP_TIMEOUT_SECONDS,
             )
             logger.info(f"Stopped container {container_id} using {self._runtime}")
+        except subprocess.TimeoutExpired:
+            # Deliberately not swallowed like a CalledProcessError: the container
+            # may still be running, so the caller must not report a clean stop.
+            logger.error(f"Timed out after {self._STOP_TIMEOUT_SECONDS}s stopping container {container_id} using {self._runtime}")
+            raise
         except subprocess.CalledProcessError as e:
             logger.warning(f"Failed to stop container {container_id}: {e.stderr}")
 

backend/packages/harness/deerflow/community/aio_sandbox/ownership/__init__.py

@@ -0,0 +1,23 @@
+"""Cross-instance ownership leases for shared sandbox containers (#4206)."""
+
+# NOTE: ``RedisOwnershipStore`` is intentionally NOT imported here. ``redis`` is an
+# optional extra, and this package is imported by ``aio_sandbox_provider`` at
+# provider construction. Importing ``.redis`` eagerly would couple every AIO
+# sandbox install to the redis package even when ownership is memory-only. It is
+# imported lazily inside ``make_sandbox_ownership_store`` only when
+# ``sandbox.ownership.type == "redis"``.
+
+from .base import OwnershipBackendError, RenewOutcome, SandboxOwnershipStore
+from .factory import compute_lease_ttl, generate_owner_id, make_sandbox_ownership_store, resolve_ownership_config
+from .memory import MemoryOwnershipStore
+
+__all__ = [
+    "MemoryOwnershipStore",
+    "OwnershipBackendError",
+    "RenewOutcome",
+    "SandboxOwnershipStore",
+    "compute_lease_ttl",
+    "generate_owner_id",
+    "make_sandbox_ownership_store",
+    "resolve_ownership_config",
+]

backend/packages/harness/deerflow/community/aio_sandbox/ownership/base.py

@@ -0,0 +1,188 @@
+"""Ownership store contract for shared sandbox containers (#4206).
+
+Gateway instances share sandbox containers but each keeps its own in-memory warm
+pool. Without shared ownership state, one instance's startup reconciliation
+adopts a container another instance is actively using and later idle-destroys it,
+so tool calls fail with 502 / connection refused.
+
+A lease answers "**which instance is responsible for reaping this container?**",
+not "which instance may use it". That distinction drives the whole interface:
+
+* A container is deterministic per (user, thread), so consecutive turns of one
+  thread legitimately land on different instances. The instance now serving the
+  thread :meth:`take` s the lease from whoever held it — refusing because a peer
+  still held it would strand the thread until that lease expired.
+* Reaping is the opposite. :meth:`claim` succeeds only when the container is
+  unowned or already ours, so an instance can never adopt (and later
+  idle-destroy) a container a live peer is responsible for. That is #4206.
+
+**A lease has two states, and that is what makes the destroy window safe.**
+`own:` means "I am responsible for this container"; `del:` means "I am tearing
+this container down". A takeover (:meth:`take`) is refused against a `del:`
+lease, so a container cannot be re-acquired between a destroy path's claim and
+its container stop — the window the deleted per-sandbox flock guard used to
+cover. Without the two states an unconditional `take` would silently overwrite a
+destroyer's claim and the peer's stop would land on a container the new owner had
+already handed to an agent.
+
+Contract notes for implementers:
+
+* Every method is **synchronous**. Unlike ``StreamBridge`` (whose async API exists
+  because it is driven from the event loop), ownership is driven from
+  ``AioSandboxProvider.__init__``, the background idle/renewal threads, and the
+  sync ``release()`` path. Sandbox tool paths that *do* run on the event loop
+  (``get()``) deliberately never touch the store, and async acquire paths offload
+  registration through ``asyncio.to_thread``.
+* Methods **raise** ``OwnershipBackendError`` on backend failure rather than
+  returning a falsy value. Callers must fail closed: a sandbox whose ownership
+  could not be published is not safe to hand out, and a container whose ownership
+  cannot be proven free is not safe to destroy. A ``False`` return means
+  "definitively not ours"; raising means "unknown".
+"""
+
+from __future__ import annotations
+
+import abc
+import enum
+
+
+class OwnershipBackendError(RuntimeError):
+    """The ownership backend could not answer.
+
+    Distinct from a definitive "not ours" (``False``): this means ownership is
+    *unknown*, so callers must fail closed rather than assume the container is
+    free.
+    """
+
+
+class RenewOutcome(enum.Enum):
+    """Why a renewal did or did not succeed.
+
+    ``LAPSED`` and ``LOST`` must not be collapsed into one falsy value. A lapsed
+    lease is *absent* — nobody took it, so re-establishing it is safe and is what
+    keeps a Redis restart from dropping every live sandbox fleet-wide. A lost
+    lease belongs to a peer, and re-taking it is the #4206 cross-instance kill.
+    """
+
+    #: Still ours; TTL refreshed.
+    RENEWED = "renewed"
+    #: No lease present (expired, or the store lost its state). Free to re-claim.
+    LAPSED = "lapsed"
+    #: A peer holds it, or it is being torn down. Do not re-take.
+    LOST = "lost"
+
+
+class SandboxOwnershipStore(abc.ABC):
+    """Cross-instance ownership leases for sandbox containers."""
+
+    #: Whether this store coordinates instances beyond the current process.
+    #: ``False`` means peers cannot see our leases, so every container looks like
+    #: an orphan to them — single-instance deployments only.
+    supports_cross_process: bool = False
+
+    @property
+    @abc.abstractmethod
+    def owner_id(self) -> str:
+        """This instance's owner id, as written into leases."""
+
+    @abc.abstractmethod
+    def take(self, sandbox_id: str) -> bool:
+        """Take responsibility for *sandbox_id* on the acquire path.
+
+        Takes over from a live peer: a turn for this container's thread has
+        routed here, and the previous owner learns to stop tracking it when its
+        next renewal reports ``LOST``. It must not destroy it — see
+        ``AioSandboxProvider._forget_lost_sandbox``.
+
+        Refuses only a container that is being torn down, which is what closes
+        the destroy → re-acquire window.
+
+        Returns:
+            ``True`` when this instance owns the lease afterwards.
+            ``False`` when the container is being destroyed and must not be used.
+
+        Raises:
+            OwnershipBackendError: ownership could not be published. Callers must
+                fail closed — an unpublished sandbox is not safe to hand out,
+                because peers will see it as an orphan.
+        """
+
+    @abc.abstractmethod
+    def claim(self, sandbox_id: str, *, for_destroy: bool = False) -> bool:
+        """Take ownership of *sandbox_id* only if it is unowned or already ours.
+
+        Exclusive: succeeds only when the container is unowned or already ours,
+        which is what gates every adopt/reap path.
+
+        Exclusive against **peers**, not against the caller's own process: a
+        claim against our own ``own:`` lease succeeds by design, which is what
+        lets a destroy path claim what it already owns. Same-process exclusion
+        between an instance's reaper threads and its own acquire path is the
+        provider's job, not this store's (``_reserve_local_teardown``).
+
+        One exception, so ``for_destroy`` cannot be silently unwound: a
+        **non**-destroy claim against our own ``del:`` lease is refused. The stop
+        it marks is already in flight and cannot be recalled, so downgrading the
+        marker would let a :meth:`take` hand out a container that is about to
+        die.
+
+        The read-modify-write must not interleave. On redis that is Lua (one
+        script, server-side); the memory store serializes on a process-local lock
+        and is single-instance anyway, so "different instances" cannot arise
+        there. Note what is *not* verified: the contract suite drives sequential
+        calls, so it pins the exclusion predicate, not the atomicity — and CI
+        runs the memory tier only, so the Lua that carries it never executes on
+        the merge gate.
+
+        Args:
+            for_destroy: mark the lease as a teardown in progress, so a
+                concurrent :meth:`take` is refused for as long as it is held.
+                Destroy paths must set this; the marker is cleared by
+                :meth:`release` once the container is stopped, and expires with
+                the TTL if the destroyer dies mid-stop.
+
+        Returns:
+            ``True`` when this instance owns the lease afterwards.
+            ``False`` when a live peer holds it.
+
+        Raises:
+            OwnershipBackendError: ownership could not be determined.
+        """
+
+    @abc.abstractmethod
+    def renew(self, sandbox_id: str) -> RenewOutcome:
+        """Refresh our lease on *sandbox_id*.
+
+        Deliberately does not re-acquire on its own — the caller decides, because
+        only the caller can tell a safe re-establish (``LAPSED``) from a
+        cross-instance steal (``LOST``).
+
+        Raises:
+            OwnershipBackendError: ownership could not be determined.
+        """
+
+    @abc.abstractmethod
+    def release(self, sandbox_id: str) -> None:
+        """Drop our lease on *sandbox_id*, in either state.
+
+        A no-op when the lease is not ours, so a peer's live lease is never
+        cleared. Best-effort: an expiring lease reaches the same state.
+
+        Raises:
+            OwnershipBackendError: the release could not be published.
+        """
+
+    @abc.abstractmethod
+    def owner(self, sandbox_id: str) -> str | None:
+        """Return the current owner id of *sandbox_id*, or ``None`` if unowned.
+
+        Read-only: unlike :meth:`claim`, this never takes ownership. Use it to
+        inspect (tests, logging) rather than to gate a destroy — a read is stale
+        the moment it returns, whereas a successful claim keeps peers out.
+
+        Raises:
+            OwnershipBackendError: ownership could not be read.
+        """
+
+    def close(self) -> None:
+        """Release backend resources. Default is a no-op."""

backend/packages/harness/deerflow/community/aio_sandbox/ownership/factory.py

@@ -0,0 +1,108 @@
+"""Resolve the configured sandbox ownership store.
+
+Mirrors ``stream_bridge``'s ``make_stream_bridge``: dispatch on ``config.type``,
+lazy per-branch imports so a memory-only install never imports ``redis``, and an
+env-var escape hatch so a container deployment can flip the backend without
+editing config.yaml.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import socket
+import uuid
+
+from deerflow.config.sandbox_config import SandboxOwnershipConfig
+from deerflow.config.stream_bridge_config import StreamBridgeConfig
+
+from .base import SandboxOwnershipStore
+
+logger = logging.getLogger(__name__)
+
+_ENV_OWNERSHIP_REDIS_URL = "DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL"
+_ENV_STREAM_BRIDGE_REDIS_URL = "DEER_FLOW_STREAM_BRIDGE_REDIS_URL"
+
+
+def generate_owner_id() -> str:
+    """Return a unique id for this provider instance: ``hostname:hex``.
+
+    Per-instance, not per-host: two gateway workers on one host must be able to
+    tell their leases apart.
+    """
+    return f"{socket.gethostname()}:{uuid.uuid4().hex}"
+
+
+def resolve_ownership_config(config: SandboxOwnershipConfig | None, *, stream_bridge: StreamBridgeConfig | None = None) -> SandboxOwnershipConfig:
+    """Fill in an omitted ownership section.
+
+    A deployment that already points the stream bridge at Redis is by definition
+    multi-instance, so it gets a redis ownership store rather than silently
+    falling back to memory (which cannot see peers and would leave #4206 open).
+
+    Both of the stream bridge's own redis triggers are honoured, and in its
+    order (``stream_bridge/async_provider.py::_resolve_config``): the config.yaml
+    section first, then the env var. Reading only the env var would miss the
+    config.yaml-native way of pointing the bridge at Redis — i.e. exactly the
+    multi-instance deployments this inference exists for.
+    """
+    if config is not None:
+        return config
+
+    if stream_bridge is not None and stream_bridge.type == "redis":
+        redis_url = stream_bridge.redis_url or os.getenv(_ENV_OWNERSHIP_REDIS_URL) or os.getenv(_ENV_STREAM_BRIDGE_REDIS_URL)
+        logger.info("Sandbox ownership: redis inferred from stream_bridge.type (multi-instance deployment)")
+        return SandboxOwnershipConfig(type="redis", redis_url=redis_url)
+
+    redis_url = os.getenv(_ENV_OWNERSHIP_REDIS_URL) or os.getenv(_ENV_STREAM_BRIDGE_REDIS_URL)
+    if redis_url:
+        logger.info("Sandbox ownership: redis inferred from environment (multi-instance deployment)")
+        return SandboxOwnershipConfig(type="redis", redis_url=redis_url)
+    return SandboxOwnershipConfig()
+
+
+def _resolve_redis_url(config: SandboxOwnershipConfig) -> str:
+    return config.redis_url or os.getenv(_ENV_OWNERSHIP_REDIS_URL) or os.getenv(_ENV_STREAM_BRIDGE_REDIS_URL) or os.getenv("REDIS_URL") or "redis://localhost:6379/0"
+
+
+def compute_lease_ttl(config: SandboxOwnershipConfig) -> float:
+    """Lease TTL in seconds.
+
+    Derived from the renewal interval, never from ``sandbox.idle_timeout``:
+    coupling liveness to the idle reaper is what let ownership lapse under
+    ``idle_timeout: 0``, where the idle checker never starts.
+    """
+    return config.renewal_interval_seconds * config.ttl_multiplier
+
+
+def make_sandbox_ownership_store(config: SandboxOwnershipConfig | None, *, owner_id: str | None = None) -> SandboxOwnershipStore:
+    """Build the ownership store for *config*.
+
+    Caller owns the returned store and must ``close()`` it.
+    """
+    # Trust an already-resolved config; only fill in an omitted section. The
+    # provider resolves once (with the stream_bridge inference this factory
+    # cannot do) and passes that in, so re-resolving here would be a no-op.
+    resolved = config if config is not None else resolve_ownership_config(None)
+    effective_owner_id = owner_id or generate_owner_id()
+    ttl = compute_lease_ttl(resolved)
+
+    if resolved.type == "memory":
+        from .memory import MemoryOwnershipStore
+
+        logger.info("Sandbox ownership store: memory (single-instance; ttl=%.1fs)", ttl)
+        return MemoryOwnershipStore(owner_id=effective_owner_id, ttl_seconds=ttl)
+
+    if resolved.type == "redis":
+        from .redis import RedisOwnershipStore
+
+        redis_url = _resolve_redis_url(resolved)
+        logger.info("Sandbox ownership store: redis (ttl=%.1fs, renewal=%.1fs)", ttl, resolved.renewal_interval_seconds)
+        return RedisOwnershipStore(
+            owner_id=effective_owner_id,
+            redis_url=redis_url,
+            ttl_seconds=ttl,
+            key_prefix=resolved.key_prefix,
+        )
+
+    raise ValueError(f"Unknown sandbox ownership type: {resolved.type!r}")

backend/packages/harness/deerflow/community/aio_sandbox/ownership/memory.py

@@ -0,0 +1,106 @@
+"""In-process ownership store for single-instance deployments.
+
+Correct only when one gateway process owns the container backend: nothing here is
+visible to another process, so a peer would see every container as unowned and
+adopt it. :attr:`supports_cross_process` is ``False`` to say so, and the provider
+warns at startup. Multi-worker / multi-instance gateways must use the redis
+store — the same rule `stream_bridge`'s memory backend carries.
+
+TTL and the two lease states are implemented for real rather than stubbed out, so
+one store-contract suite exercises both backends and a lapsed lease behaves
+identically whichever store is configured.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass
+
+from .base import RenewOutcome, SandboxOwnershipStore
+
+
+@dataclass(frozen=True, slots=True)
+class _Lease:
+    owner_id: str
+    expires_at: float
+    destroying: bool
+
+
+class MemoryOwnershipStore(SandboxOwnershipStore):
+    """Ownership leases held in this process only."""
+
+    supports_cross_process = False
+
+    def __init__(self, *, owner_id: str, ttl_seconds: float, time_source=time.monotonic) -> None:
+        self._owner_id = owner_id
+        self._ttl = float(ttl_seconds)
+        self._now = time_source
+        # sandbox_id -> _Lease. Guarded by _lock: the acquire path, the idle
+        # checker thread, and the renewal thread all touch it.
+        self._leases: dict[str, _Lease] = {}
+        self._lock = threading.Lock()
+
+    @property
+    def owner_id(self) -> str:
+        return self._owner_id
+
+    def _live_lease_locked(self, sandbox_id: str) -> _Lease | None:
+        lease = self._leases.get(sandbox_id)
+        if lease is None:
+            return None
+        if self._now() >= lease.expires_at:
+            del self._leases[sandbox_id]
+            return None
+        return lease
+
+    def _write_locked(self, sandbox_id: str, *, destroying: bool) -> None:
+        self._leases[sandbox_id] = _Lease(owner_id=self._owner_id, expires_at=self._now() + self._ttl, destroying=destroying)
+
+    def take(self, sandbox_id: str) -> bool:
+        with self._lock:
+            lease = self._live_lease_locked(sandbox_id)
+            # Refuse only a teardown in progress; a live peer's normal lease is
+            # taken over, which is the point of take().
+            if lease is not None and lease.destroying:
+                return False
+            self._write_locked(sandbox_id, destroying=False)
+            return True
+
+    def claim(self, sandbox_id: str, *, for_destroy: bool = False) -> bool:
+        with self._lock:
+            lease = self._live_lease_locked(sandbox_id)
+            if lease is not None and lease.owner_id != self._owner_id:
+                return False
+            if not for_destroy and lease is not None and lease.destroying:
+                # Never unwind our own teardown: the stop is already in flight
+                # and cannot be recalled, so downgrading to `own:` would let a
+                # `take()` hand out a container that is about to die.
+                return False
+            self._write_locked(sandbox_id, destroying=for_destroy)
+            return True
+
+    def renew(self, sandbox_id: str) -> RenewOutcome:
+        with self._lock:
+            lease = self._live_lease_locked(sandbox_id)
+            if lease is None:
+                return RenewOutcome.LAPSED
+            if lease.owner_id != self._owner_id or lease.destroying:
+                return RenewOutcome.LOST
+            self._write_locked(sandbox_id, destroying=False)
+            return RenewOutcome.RENEWED
+
+    def release(self, sandbox_id: str) -> None:
+        with self._lock:
+            lease = self._live_lease_locked(sandbox_id)
+            if lease is not None and lease.owner_id == self._owner_id:
+                del self._leases[sandbox_id]
+
+    def owner(self, sandbox_id: str) -> str | None:
+        with self._lock:
+            lease = self._live_lease_locked(sandbox_id)
+            return None if lease is None else lease.owner_id
+
+    def close(self) -> None:
+        with self._lock:
+            self._leases.clear()

backend/packages/harness/deerflow/community/aio_sandbox/ownership/redis.py

@@ -0,0 +1,214 @@
+"""Redis-backed ownership store for multi-instance gateways (#4206).
+
+Ownership is a single key per sandbox whose value encodes both the owner and the
+lease state — ``own:<owner_id>`` (responsible for this container) or
+``del:<owner_id>`` (tearing it down) — with a TTL the owning instance refreshes.
+
+The state prefix is what makes the destroy window safe without a lock: a
+takeover is refused against a ``del:`` lease, so a container cannot be
+re-acquired between a destroy path's claim and its container stop.
+
+The sync client is deliberate: this store is driven from provider construction
+and from background threads, never from the event loop (see ``base`` for the
+contract). ``redis.asyncio`` would be the wrong client here.
+
+Every mutation goes through a Lua script so the read and the write cannot be
+interleaved by a peer. ``SET NX`` alone is not enough: it fails on a key we
+already own, and a GET-then-SET fallback in Python reopens the race the script
+closes.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from .base import OwnershipBackendError, RenewOutcome, SandboxOwnershipStore
+
+try:
+    from redis import Redis
+    from redis.exceptions import RedisError
+except ImportError:  # pragma: no cover - only hit when the optional extra is missing
+    # ``redis`` is an optional extra (mirrors the stream_bridge redis path). This
+    # module is imported lazily from ``make_sandbox_ownership_store`` only when
+    # ``sandbox.ownership.type == "redis"``, so this hint surfaces exactly when a
+    # redis ownership store is requested without the package.
+    raise ImportError(
+        "sandbox.ownership.type is set to 'redis' but the redis package is not installed.\n"
+        "Install it with:\n"
+        "    cd backend && uv sync --all-packages --extra redis\n"
+        "On the next `make dev` the redis extra is auto-detected from config.yaml\n"
+        "(sandbox.ownership.type: redis) and reinstalled, so it will not be wiped again.\n"
+        "Or switch to sandbox.ownership.type: memory in config.yaml for single-instance deployment."
+    ) from None
+
+logger = logging.getLogger(__name__)
+
+_OWN = "own:"
+_DEL = "del:"
+
+# Bound every store round-trip so a stalled Redis cannot wedge a caller. This
+# matters most for the teardown heartbeat: its exit — and the final lease
+# release that exit performs — must stay finite, otherwise a refresh blocked on
+# a black-holed connection could hold a destroy path (and its deferred release)
+# open indefinitely. Without a socket timeout redis-py blocks forever.
+_STORE_SOCKET_TIMEOUT_SECONDS = 5.0
+
+# Acquire-path takeover. Overwrites a live peer's normal lease on purpose — a
+# thread's turn has routed here — but refuses a teardown in progress, which is
+# what stops us handing out a container a peer is about to stop.
+_TAKE_SCRIPT = """
+local current = redis.call('GET', KEYS[1])
+if current ~= false and string.sub(current, 1, 4) == 'del:' then
+    return 0
+end
+redis.call('SET', KEYS[1], 'own:' .. ARGV[1], 'PX', ARGV[2])
+return 1
+"""
+
+# Adopt/reap gate: only if unowned or already ours (in either state).
+# ARGV[3] selects the state written: '1' marks a teardown in progress.
+#
+# A non-destroy claim never unwinds our *own* teardown: a stop is already in
+# flight and cannot be recalled, so downgrading the marker to `own:` would let a
+# `take()` hand out a container that is about to die. No caller does this today
+# (the `for_destroy=false` callers run against an absent or unowned key), but the
+# contract has to forbid it rather than rely on that staying true.
+_CLAIM_SCRIPT = """
+local current = redis.call('GET', KEYS[1])
+local mine_own = 'own:' .. ARGV[1]
+local mine_del = 'del:' .. ARGV[1]
+if ARGV[3] == '0' and current == mine_del then
+    return 0
+end
+if current == false or current == mine_own or current == mine_del then
+    local value = mine_own
+    if ARGV[3] == '1' then
+        value = mine_del
+    end
+    redis.call('SET', KEYS[1], value, 'PX', ARGV[2])
+    return 1
+end
+return 0
+"""
+
+# Three-way so the caller can tell an absent lease (safe to re-establish) from a
+# peer's (re-taking it is the #4206 kill). Collapsing them is what let a Redis
+# restart drop every live sandbox fleet-wide.
+#    1 = renewed, -1 = lapsed/absent, 0 = held by a peer or being torn down
+_RENEW_SCRIPT = """
+local current = redis.call('GET', KEYS[1])
+if current == false then
+    return -1
+end
+if current == 'own:' .. ARGV[1] then
+    redis.call('PEXPIRE', KEYS[1], ARGV[2])
+    return 1
+end
+return 0
+"""
+
+# Drop only our own lease, in either state, so a peer's is never cleared.
+_RELEASE_SCRIPT = """
+local current = redis.call('GET', KEYS[1])
+if current == 'own:' .. ARGV[1] or current == 'del:' .. ARGV[1] then
+    return redis.call('DEL', KEYS[1])
+end
+return 0
+"""
+
+
+class RedisOwnershipStore(SandboxOwnershipStore):
+    """Ownership leases shared across gateway instances via Redis."""
+
+    supports_cross_process = True
+
+    def __init__(
+        self,
+        *,
+        owner_id: str,
+        redis_url: str,
+        ttl_seconds: float,
+        key_prefix: str = "deerflow:sandbox:owner",
+        client: Redis | None = None,
+    ) -> None:
+        self._owner_id = owner_id
+        self._ttl_ms = max(1, int(float(ttl_seconds) * 1000))
+        self._key_prefix = key_prefix.rstrip(":")
+        # Redis.from_url is lazy, so an unreachable Redis does not block provider
+        # construction; the first claim raises instead. socket_timeout bounds
+        # every round-trip (see _STORE_SOCKET_TIMEOUT_SECONDS) so no store call —
+        # in particular a teardown-heartbeat refresh — can block unbounded.
+        self._redis = (
+            client
+            if client is not None
+            else Redis.from_url(
+                redis_url,
+                decode_responses=True,
+                socket_timeout=_STORE_SOCKET_TIMEOUT_SECONDS,
+                socket_connect_timeout=_STORE_SOCKET_TIMEOUT_SECONDS,
+            )
+        )
+        self._owns_client = client is None
+        self._take = self._redis.register_script(_TAKE_SCRIPT)
+        self._claim = self._redis.register_script(_CLAIM_SCRIPT)
+        self._renew = self._redis.register_script(_RENEW_SCRIPT)
+        self._release = self._redis.register_script(_RELEASE_SCRIPT)
+
+    @property
+    def owner_id(self) -> str:
+        return self._owner_id
+
+    def _key(self, sandbox_id: str) -> str:
+        return f"{self._key_prefix}:{sandbox_id}"
+
+    def take(self, sandbox_id: str) -> bool:
+        try:
+            result = self._take(keys=[self._key(sandbox_id)], args=[self._owner_id, self._ttl_ms])
+        except RedisError as e:
+            raise OwnershipBackendError(f"failed to publish sandbox ownership for {sandbox_id}: {e}") from e
+        return bool(result)
+
+    def claim(self, sandbox_id: str, *, for_destroy: bool = False) -> bool:
+        try:
+            result = self._claim(keys=[self._key(sandbox_id)], args=[self._owner_id, self._ttl_ms, "1" if for_destroy else "0"])
+        except RedisError as e:
+            raise OwnershipBackendError(f"failed to claim sandbox ownership for {sandbox_id}: {e}") from e
+        return bool(result)
+
+    def renew(self, sandbox_id: str) -> RenewOutcome:
+        try:
+            result = int(self._renew(keys=[self._key(sandbox_id)], args=[self._owner_id, self._ttl_ms]))
+        except RedisError as e:
+            raise OwnershipBackendError(f"failed to renew sandbox ownership for {sandbox_id}: {e}") from e
+        if result == 1:
+            return RenewOutcome.RENEWED
+        if result == -1:
+            return RenewOutcome.LAPSED
+        return RenewOutcome.LOST
+
+    def release(self, sandbox_id: str) -> None:
+        try:
+            self._release(keys=[self._key(sandbox_id)], args=[self._owner_id])
+        except RedisError as e:
+            raise OwnershipBackendError(f"failed to release sandbox ownership for {sandbox_id}: {e}") from e
+
+    def owner(self, sandbox_id: str) -> str | None:
+        try:
+            value = self._redis.get(self._key(sandbox_id))
+        except RedisError as e:
+            raise OwnershipBackendError(f"failed to read sandbox ownership for {sandbox_id}: {e}") from e
+        if value is None:
+            return None
+        # An injected client may not set decode_responses.
+        text = value.decode("utf-8") if isinstance(value, bytes) else value
+        if text.startswith(_OWN) or text.startswith(_DEL):
+            return text[4:]
+        return text
+
+    def close(self) -> None:
+        if not self._owns_client:
+            return
+        try:
+            self._redis.close()
+        except Exception as e:  # pragma: no cover - teardown best effort
+            logger.warning("Error closing sandbox ownership redis client: %s", e)

backend/packages/harness/deerflow/config/sandbox_config.py

@@ -1,5 +1,48 @@
+from typing import Literal
+
 from pydantic import BaseModel, ConfigDict, Field
 
+SandboxOwnershipType = Literal["memory", "redis"]
+
+
+class SandboxOwnershipConfig(BaseModel):
+    """Configuration for cross-instance sandbox container ownership (#4206).
+
+    Gateway instances share sandbox containers but each keeps its own in-memory
+    warm pool. Without shared ownership state, one instance's reconciliation
+    adopts another's live container and later idle-destroys it. This selects
+    where that ownership state lives.
+    """
+
+    type: SandboxOwnershipType = Field(
+        default="memory",
+        description=(
+            "Sandbox ownership store backend. 'memory' keeps ownership in-process (single-instance deployments only, where cross-instance adoption cannot occur). "
+            "'redis' shares ownership across gateway instances and is required for load-balanced / multi-worker deployments that share a container backend."
+        ),
+    )
+    redis_url: str | None = Field(
+        default=None,
+        description="Redis URL for the redis ownership type. If omitted, DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL, DEER_FLOW_STREAM_BRIDGE_REDIS_URL, REDIS_URL, or redis://localhost:6379/0 is used.",
+    )
+    renewal_interval_seconds: float = Field(
+        default=30.0,
+        gt=0,
+        description=(
+            "How often an owning instance refreshes its leases. The lease TTL is derived from this (interval x ttl_multiplier), so ownership liveness is independent of sandbox.idle_timeout: "
+            "renewal keeps running even when idle cleanup is disabled (idle_timeout: 0)."
+        ),
+    )
+    ttl_multiplier: float = Field(
+        default=4.0,
+        ge=2,
+        description="Lease TTL as a multiple of renewal_interval_seconds. At least 2, so a single missed renewal (slow host, brief Redis blip) cannot expire a live owner's lease. Default 4 tolerates three consecutive misses.",
+    )
+    key_prefix: str = Field(
+        default="deerflow:sandbox:owner",
+        description="Redis key prefix for ownership leases. Only applies to the redis ownership type.",
+    )
+
 
 class VolumeMountConfig(BaseModel):
     """Configuration for a volume mount."""
@@ -43,6 +86,8 @@ class SandboxConfig(BaseModel):
         port: Base port for sandbox containers (default: 8080)
         container_prefix: Prefix for container names (default: deer-flow-sandbox)
         mounts: List of volume mounts to share directories with the container
+        ownership: Cross-instance container ownership store (memory | redis). Multi-instance
+            deployments sharing a container backend need redis; see SandboxOwnershipConfig.
     """
 
     use: str = Field(
@@ -78,6 +123,13 @@ class SandboxConfig(BaseModel):
         ge=0,
         description="BoxLite-only reclaim skip window in seconds for boxes recently released by this provider instance. Set to 0 to always validate before warm reuse.",
     )
+    ownership: SandboxOwnershipConfig | None = Field(
+        default=None,
+        description=(
+            "AioSandboxProvider-only: where cross-instance container ownership is tracked (#4206). Omitted = memory (single-instance). "
+            "Multi-worker / load-balanced gateways sharing one container backend must set type: redis, or peers will adopt and idle-destroy each other's live sandboxes."
+        ),
+    )
     mounts: list[VolumeMountConfig] = Field(
         default_factory=list,
         description="List of volume mounts to share directories between host and container",

backend/tests/blocking_io/test_aio_sandbox_get.py

@@ -0,0 +1,155 @@
+"""Regression: ``AioSandboxProvider.get()`` must not do blocking IO.
+
+``ensure_sandbox_initialized_async`` (``sandbox/tools.py``) calls
+``provider.get()`` directly on the LangGraph event loop for every sandbox tool
+lookup. A prior change renewed the cross-process lease inside ``get()``
+(``mkdir`` + temp-file write + ``fsync`` + ``os.replace``), which blocks the loop
+— reported on PR #4221.
+
+Under the strict Blockbuster context (this directory's conftest), any blocking IO
+reached from ``deerflow.*`` while on the event loop raises ``BlockingError``.
+
+The ownership store is injected here as a **blocking probe**: every store method
+does real file IO. That keeps the anchor honest across backends — the configured
+store may be in-memory (no IO to catch), but the redis store does network IO and a
+future store could do anything, so what must be pinned is that ``get()`` performs
+*no store call at all*, not merely that today's default store happens to be cheap.
+If ownership work is put back on this path, this test fails.
+"""
+
+from __future__ import annotations
+
+import threading
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+
+class _BlockingProbeStore:
+    """Ownership store whose every operation does real blocking file IO."""
+
+    supports_cross_process = True
+
+    def __init__(self, probe_path: Path):
+        self._probe_path = probe_path
+        self._probe_path.write_text("owner", encoding="utf-8")
+
+    @property
+    def owner_id(self) -> str:
+        return "worker-blockingio"
+
+    def _blocking_touch(self) -> str:
+        # Mirrors what a real store does on this call: sync IO the strict gate sees.
+        return self._probe_path.read_text(encoding="utf-8")
+
+    def take(self, sandbox_id: str) -> bool:
+        self._blocking_touch()
+        return True
+
+    def claim(self, sandbox_id: str, *, for_destroy: bool = False) -> bool:
+        self._blocking_touch()
+        return True
+
+    def renew(self, sandbox_id: str):
+        from deerflow.community.aio_sandbox.ownership import RenewOutcome
+
+        self._blocking_touch()
+        return RenewOutcome.RENEWED
+
+    def release(self, sandbox_id: str) -> None:
+        self._blocking_touch()
+
+    def owner(self, sandbox_id: str) -> str | None:
+        return self._blocking_touch()
+
+    def close(self) -> None:
+        pass
+
+
+def _make_provider(tmp_path: Path):
+    """Build an ``AioSandboxProvider`` without ``__init__`` (no Docker, no threads)."""
+    from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider
+    from deerflow.config.sandbox_config import SandboxOwnershipConfig
+
+    provider = AioSandboxProvider.__new__(AioSandboxProvider)
+    provider._lock = threading.Lock()
+    provider._sandboxes = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._thread_locks = {}
+    provider._last_activity = {}
+    provider._warm_pool = {}
+    provider._local_teardown = set()
+    provider._acquire_epoch = {}
+    provider._acquire_epoch_counter = 0
+    provider._acquire_inflight = {}
+    provider._shutdown_called = False
+    provider._idle_checker_stop = threading.Event()
+    provider._idle_checker_thread = None
+    provider._renewal_stop = threading.Event()
+    provider._renewal_thread = None
+    provider._config = {"idle_timeout": 600, "replicas": 3}
+    provider._backend = MagicMock()
+    provider._owner_id = "worker-blockingio"
+    provider._ownership_config = SandboxOwnershipConfig()
+    provider._ownership = _BlockingProbeStore(tmp_path / "ownership-probe")
+    return provider
+
+
+async def test_get_does_no_blocking_io_on_event_loop(tmp_path):
+    provider = _make_provider(tmp_path)
+    provider._sandboxes["sb-blockingio"] = MagicMock()
+
+    # If get() touches the ownership store, the probe's file read trips the gate.
+    assert provider.get("sb-blockingio") is not None
+
+
+async def test_blocking_probe_store_actually_trips_the_gate(tmp_path):
+    """Meta-check: prove the probe has teeth, so the test above is not vacuous.
+
+    Without this, a store that silently stopped doing IO would make the anchor
+    pass for the wrong reason.
+    """
+    from blockbuster import BlockingError
+
+    provider = _make_provider(tmp_path)
+
+    with pytest.raises(BlockingError):
+        provider._publish_ownership("sb-blockingio")
+
+
+async def test_async_acquire_offloads_ownership_publish(tmp_path, monkeypatch):
+    """The async acquire paths must offload registration, not just discovery.
+
+    ``_register_discovered_sandbox`` / ``_register_created_sandbox`` publish
+    ownership, which is blocking store IO. Every other blocking step in
+    ``_discover_or_create_with_lock_async`` is wrapped in ``asyncio.to_thread``;
+    these two were called directly, putting a Redis round trip on the event loop
+    for every discover/create.
+    """
+    import deerflow.community.aio_sandbox.aio_sandbox_provider as aio_mod
+    from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
+
+    provider = _make_provider(tmp_path)
+    info = SandboxInfo(
+        sandbox_id="sb-async",
+        sandbox_url="http://localhost:8080",
+        container_name="deer-flow-sandbox-sb-async",
+        created_at=1.0,
+    )
+    provider._backend.discover = MagicMock(return_value=info)
+
+    # Stub the path layer: `get_paths()` resolves the base dir via os.getcwd on
+    # the event loop, which is a pre-existing blocking call in this coroutine and
+    # not what this anchor is about. Scoping it out keeps the test pinned to the
+    # ownership publish this diff added.
+    fake_paths = MagicMock()
+    fake_paths.thread_dir.return_value = tmp_path
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: fake_paths)
+
+    sandbox_id = await provider._discover_or_create_with_lock_async("t-async", "sb-async", user_id="u1")
+
+    assert sandbox_id == "sb-async"

backend/tests/blocking_io/test_sandbox_release.py

@@ -0,0 +1,141 @@
+"""Regression anchor: sandbox release must not block the event loop.
+
+``AioSandboxProvider.release()`` refreshes the ownership lease
+(``_refresh_ownership`` -> store ``renew``/``claim``), which is blocking
+filesystem or network IO depending on the backend. It runs from
+``SandboxMiddleware`` at the end of every turn: the async gateway path
+(``aafter_agent``) offloads it with ``asyncio.to_thread``, so the store round
+trip stays off the loop. This pins that offload — a refactor that dropped it
+(or wired ``aafter_agent`` to call ``release`` directly) would put a Redis round
+trip on the event loop for sync graph execution, as flagged in review of
+PR #4221.
+
+The ownership store is injected as a **blocking probe** whose every method does
+real file IO, so the anchor keeps its teeth regardless of the configured backend
+(the default ``memory`` store does no IO to catch; redis does network IO).
+"""
+
+from __future__ import annotations
+
+import threading
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+
+class _BlockingProbeStore:
+    """Ownership store whose every operation does real blocking file IO."""
+
+    supports_cross_process = True
+
+    def __init__(self, probe_path: Path):
+        self._probe_path = probe_path
+        self._probe_path.write_text("owner", encoding="utf-8")
+
+    @property
+    def owner_id(self) -> str:
+        return "worker-blockingio"
+
+    def _blocking_touch(self) -> str:
+        return self._probe_path.read_text(encoding="utf-8")
+
+    def take(self, sandbox_id: str) -> bool:
+        self._blocking_touch()
+        return True
+
+    def claim(self, sandbox_id: str, *, for_destroy: bool = False) -> bool:
+        self._blocking_touch()
+        return True
+
+    def renew(self, sandbox_id: str):
+        from deerflow.community.aio_sandbox.ownership import RenewOutcome
+
+        self._blocking_touch()
+        return RenewOutcome.RENEWED
+
+    def release(self, sandbox_id: str) -> None:
+        self._blocking_touch()
+
+    def owner(self, sandbox_id: str) -> str | None:
+        return self._blocking_touch()
+
+    def close(self) -> None:
+        pass
+
+
+def _make_provider_with_active_sandbox(tmp_path: Path, sandbox_id: str):
+    """A real provider (no ``__init__``) holding one active sandbox to release."""
+    from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider
+    from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
+    from deerflow.config.sandbox_config import SandboxOwnershipConfig
+
+    provider = AioSandboxProvider.__new__(AioSandboxProvider)
+    provider._lock = threading.Lock()
+    provider._sandboxes = {sandbox_id: MagicMock()}
+    provider._sandbox_infos = {
+        sandbox_id: SandboxInfo(
+            sandbox_id=sandbox_id,
+            sandbox_url="http://localhost:8080",
+            container_name=f"deer-flow-sandbox-{sandbox_id}",
+            created_at=1.0,
+        )
+    }
+    provider._thread_sandboxes = {}
+    provider._thread_locks = {}
+    provider._last_activity = {sandbox_id: 1.0}
+    provider._warm_pool = {}
+    provider._unowned_since = {}
+    provider._local_teardown = set()
+    provider._acquire_epoch = {}
+    provider._acquire_epoch_counter = 0
+    provider._acquire_inflight = {}
+    provider._shutdown_called = False
+    provider._idle_checker_stop = threading.Event()
+    provider._idle_checker_thread = None
+    provider._renewal_stop = threading.Event()
+    provider._renewal_thread = None
+    provider._config = {"idle_timeout": 600, "replicas": 3}
+    provider._backend = MagicMock()
+    provider._owner_id = "worker-blockingio"
+    provider._ownership_config = SandboxOwnershipConfig()
+    provider._ownership = _BlockingProbeStore(tmp_path / "ownership-probe")
+    return provider
+
+
+async def test_aafter_agent_offloads_release_off_the_event_loop(tmp_path, monkeypatch):
+    """The async release hook must keep the ownership-store round trip off-loop.
+
+    If it regresses to calling ``release`` directly, the probe's file IO trips
+    the strict Blockbuster gate.
+    """
+    import deerflow.sandbox.middleware as mw_mod
+
+    provider = _make_provider_with_active_sandbox(tmp_path, "sb-release")
+    monkeypatch.setattr(mw_mod, "get_sandbox_provider", lambda: provider)
+
+    mw = mw_mod.SandboxMiddleware()
+    state = {"sandbox": {"sandbox_id": "sb-release"}}
+
+    # Offloaded via asyncio.to_thread, so no BlockingError under the strict gate.
+    await mw.aafter_agent(state, MagicMock())
+
+    # The release actually happened (parked in the warm pool), so the anchor is
+    # exercising the real path, not a no-op.
+    assert "sb-release" in provider._warm_pool
+
+
+async def test_release_on_loop_trips_the_gate(tmp_path):
+    """Meta-check: prove the probe has teeth, so the test above is not vacuous.
+
+    Calling ``release`` directly on the event loop must raise, otherwise the
+    offload anchor could pass because the store quietly stopped doing IO.
+    """
+    from blockbuster import BlockingError
+
+    provider = _make_provider_with_active_sandbox(tmp_path, "sb-onloop")
+
+    with pytest.raises(BlockingError):
+        provider.release("sb-onloop")

backend/tests/test_aio_sandbox_local_backend.py

@@ -333,3 +333,44 @@ def test_is_container_running_raises_on_unrelated_not_found_error(monkeypatch):
 
     with pytest.raises(RuntimeError, match="Failed to inspect container sandbox-busy"):
         backend._is_container_running("sandbox-busy")
+
+
+def test_stop_container_passes_a_timeout(monkeypatch):
+    """An unbounded `stop` can outlive the teardown lease that guards it.
+
+    The `del:` marker keeps a peer from re-acquiring the container during the
+    stop, but a lease can lapse (a store outage longer than the TTL) while a
+    wedged daemon leaves `docker stop` blocked forever — and the stop then lands
+    on a container the peer has since been handed. Bounding the call caps that
+    exposure independently of the ownership layer.
+    """
+    backend = _backend_for_inspect_tests()
+    seen = {}
+
+    def fake_run(cmd, **kwargs):
+        seen.update(kwargs)
+        return SimpleNamespace(stdout="", stderr="", returncode=0)
+
+    monkeypatch.setattr("subprocess.run", fake_run)
+    backend._stop_container("sandbox-slow")
+
+    assert seen.get("timeout") == backend._STOP_TIMEOUT_SECONDS
+
+
+def test_stop_container_propagates_a_timeout_instead_of_reporting_success(monkeypatch):
+    """A timed-out stop must not be swallowed like a failed one.
+
+    `CalledProcessError` means the runtime answered "I could not stop it"; a
+    timeout means we do not know, and the container is probably still running.
+    Returning normally would let `_destroy_warm_entry` report a clean stop and
+    drop the warm entry, leaking a running container nothing tracks.
+    """
+    backend = _backend_for_inspect_tests()
+
+    def fake_run(cmd, **kwargs):
+        raise subprocess.TimeoutExpired(cmd=cmd, timeout=kwargs["timeout"])
+
+    monkeypatch.setattr("subprocess.run", fake_run)
+
+    with pytest.raises(subprocess.TimeoutExpired):
+        backend._stop_container("sandbox-wedged")

backend/tests/test_aio_sandbox_provider.py

@@ -45,14 +45,33 @@ def test_host_thread_dir_rejects_invalid_thread_id(tmp_path):
 
 
 def _make_provider(tmp_path):
-    """Build a minimal AioSandboxProvider instance without starting the idle checker."""
+    """Build a minimal AioSandboxProvider instance without starting the idle checker.
+
+    ``tmp_path`` is accepted and ignored: ownership no longer lives on disk. Each
+    provider gets its own in-process ownership store, so it owns every sandbox it
+    tracks — cross-instance behaviour is covered in
+    ``test_sandbox_orphan_reconciliation.py`` (shared store) and
+    ``test_sandbox_ownership_store.py`` (store contract).
+    """
+    from deerflow.community.aio_sandbox.ownership.memory import MemoryOwnershipStore
+    from deerflow.config.sandbox_config import SandboxOwnershipConfig
+
     aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
     with patch.object(aio_mod.AioSandboxProvider, "_start_idle_checker"):
         provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
-        provider._config = {}
+        provider._config = {"idle_timeout": 600, "replicas": 3}
         provider._sandboxes = {}
+        provider._local_teardown = set()
+        provider._acquire_epoch = {}
+        provider._acquire_epoch_counter = 0
+        provider._acquire_inflight = {}
         provider._lock = MagicMock()
         provider._idle_checker_stop = MagicMock()
+        provider._renewal_stop = MagicMock()
+        provider._renewal_thread = None
+        provider._owner_id = "test-worker"
+        provider._ownership_config = SandboxOwnershipConfig()
+        provider._ownership = MemoryOwnershipStore(owner_id="test-worker", ttl_seconds=600)
     return provider
 
 
@@ -430,6 +449,10 @@ def _make_provider_with_active_sandbox(tmp_path, sandbox_id: str):
     }
     provider._thread_sandboxes = {}
     provider._last_activity = {sandbox_id: 0.0}
+    provider._local_teardown = set()
+    provider._acquire_epoch = {}
+    provider._acquire_epoch_counter = 0
+    provider._acquire_inflight = {}
     provider._shutdown_called = False
     provider._idle_checker_thread = None
     provider._backend = SimpleNamespace(destroy=MagicMock())
@@ -651,12 +674,19 @@ def test_cleanup_idle_sandboxes_keeps_active_cleanup_and_delegates_warm_expiry(t
     }
 
     calls = []
-    provider.destroy = MagicMock(side_effect=lambda _sandbox_id: calls.append("active"))
+    # The idle path destroys through `_destroy_tracked`, not `destroy()`: its
+    # "still idle?" re-check has to run in the same critical section that
+    # reserves the teardown, so it is passed down as a predicate. Asserting on
+    # `destroy` here would pass vacuously — it is no longer on this path.
+    provider._destroy_tracked = MagicMock(side_effect=lambda _sandbox_id, **_kw: calls.append("active"))
     provider._reap_expired_warm = MagicMock(side_effect=lambda _idle_timeout: calls.append("warm"))
 
     provider._cleanup_idle_sandboxes(1.0)
 
-    provider.destroy.assert_called_once_with("active-old")
+    assert provider._destroy_tracked.call_count == 1
+    assert provider._destroy_tracked.call_args.args == ("active-old",)
+    # The gate must actually be a live predicate, not a constant-true placeholder.
+    assert provider._destroy_tracked.call_args.kwargs["still_reapable"]() is True
     provider._reap_expired_warm.assert_called_once_with(1.0)
     assert calls == ["active", "warm"]
 

backend/tests/test_sandbox_ownership_store.py

@@ -0,0 +1,512 @@
+"""Contract tests for the sandbox ownership store (#4206).
+
+Every behavioural test here is **backend-agnostic**: it runs against each store
+implementation through the same fixture, so the memory and redis backends cannot
+drift apart on the semantics the provider depends on.
+
+Redis coverage is opt-in and self-skipping, mirroring the stream-bridge
+integration tier: point at a server with ``DEER_FLOW_TEST_REDIS_URL`` (defaults
+to redis://localhost:6379/15 — DB 15 to avoid clobbering real data). There is no
+fake-redis tier on purpose — the redis backend's exclusion lives in Lua scripts
+that a hand-rolled fake would not execute, so a fake would pin the mock rather
+than the script. When no server is reachable these skip and the memory backend
+still covers the contract.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+import uuid
+
+import pytest
+
+from deerflow.community.aio_sandbox.ownership import (
+    MemoryOwnershipStore,
+    OwnershipBackendError,
+    RenewOutcome,
+    compute_lease_ttl,
+    generate_owner_id,
+    make_sandbox_ownership_store,
+    resolve_ownership_config,
+)
+from deerflow.config.sandbox_config import SandboxOwnershipConfig
+from deerflow.config.stream_bridge_config import StreamBridgeConfig
+
+REDIS_TEST_URL = os.environ.get("DEER_FLOW_TEST_REDIS_URL", "redis://localhost:6379/15")
+
+
+def _redis_available() -> bool:
+    try:
+        import redis
+    except ImportError:
+        return False
+    try:
+        client = redis.Redis.from_url(REDIS_TEST_URL, socket_connect_timeout=0.5)
+        try:
+            client.ping()
+        finally:
+            client.close()
+        return True
+    except Exception:
+        return False
+
+
+requires_redis = pytest.mark.skipif(not _redis_available(), reason=f"Redis not reachable at {REDIS_TEST_URL}")
+
+
+class _StoreFactory:
+    """Builds stores for one backend that all share the same keyspace."""
+
+    def __init__(self, kind: str, ttl_seconds: float = 60.0):
+        self.kind = kind
+        self.ttl = ttl_seconds
+        self._shared_leases: dict = {}
+        self._key_prefix = f"deerflow:test:{uuid.uuid4().hex}"
+        self._made: list = []
+
+    def make(self, owner_id: str, *, ttl_seconds: float | None = None):
+        ttl = self.ttl if ttl_seconds is None else ttl_seconds
+        if self.kind == "memory":
+            store = MemoryOwnershipStore(owner_id=owner_id, ttl_seconds=ttl)
+            # Share one dict so separate store objects model separate instances
+            # talking to one backend, as redis clients naturally do.
+            store._leases = self._shared_leases
+        else:
+            from deerflow.community.aio_sandbox.ownership.redis import RedisOwnershipStore
+
+            store = RedisOwnershipStore(
+                owner_id=owner_id,
+                redis_url=REDIS_TEST_URL,
+                ttl_seconds=ttl,
+                key_prefix=self._key_prefix,
+            )
+        self._made.append(store)
+        return store
+
+    def cleanup(self):
+        if self.kind == "redis" and self._made:
+            client = self._made[0]._redis
+            for key in client.scan_iter(f"{self._key_prefix}:*"):
+                client.delete(key)
+        for store in self._made:
+            store.close()
+
+
+@pytest.fixture(params=["memory", pytest.param("redis", marks=[requires_redis, pytest.mark.integration])])
+def stores(request):
+    factory = _StoreFactory(request.param)
+    try:
+        yield factory
+    finally:
+        factory.cleanup()
+
+
+# ── The #4206 invariant: a peer cannot claim a live owner's container ─────────
+
+
+def test_claim_is_exclusive_across_instances(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+
+    assert a.claim("s1") is True
+    assert b.claim("s1") is False, "a peer claimed a container A already owns — #4206"
+    assert a.owner("s1") == "A"
+
+
+def test_failed_claim_does_not_steal_the_lease(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1")
+
+    b.claim("s1")
+
+    assert a.owner("s1") == "A"
+
+
+def test_claim_refreshes_our_own_lease(stores):
+    a = stores.make("A")
+    assert a.claim("s1") is True
+    assert a.claim("s1") is True
+
+
+def test_claim_succeeds_once_a_lease_expires(stores):
+    """The crash path: a dead owner's container must become adoptable."""
+    a = stores.make("A", ttl_seconds=0.2)
+    b = stores.make("B")
+    assert a.claim("s1") is True
+    assert b.claim("s1") is False
+
+    time.sleep(0.35)
+
+    assert a.owner("s1") is None
+    assert b.claim("s1") is True
+
+
+# ── take(): ownership transfers when a thread moves instance ─────────────────
+
+
+def test_take_transfers_ownership_from_a_live_peer(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1")
+
+    assert b.take("s1") is True
+
+    assert b.owner("s1") == "B"
+
+
+def test_take_makes_the_previous_owners_renew_report_lost(stores):
+    """How the previous owner learns to stop tracking the container."""
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1")
+    b.take("s1")
+
+    assert a.renew("s1") is RenewOutcome.LOST
+    assert b.renew("s1") is RenewOutcome.RENEWED
+
+
+# ── The destroy window: take() must not overrun a teardown ──────────────────
+
+
+def test_take_is_refused_while_a_peer_is_destroying(stores):
+    """#4206's remaining window: an unconditional take would overwrite a
+    destroyer's claim, and the peer's container stop would then land on a
+    container this instance had already handed to an agent."""
+    a = stores.make("A")
+    b = stores.make("B")
+    assert a.claim("s1", for_destroy=True) is True
+
+    assert b.take("s1") is False, "took over a container that is being destroyed"
+
+
+def test_take_is_allowed_once_the_teardown_marker_is_released(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1", for_destroy=True)
+    a.release("s1")
+
+    assert b.take("s1") is True
+
+
+def test_a_destroyers_own_claim_is_idempotent(stores):
+    a = stores.make("A")
+    assert a.claim("s1", for_destroy=True) is True
+    assert a.claim("s1", for_destroy=True) is True
+
+
+def test_claim_for_destroy_is_still_refused_against_a_peer(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1")
+
+    assert b.claim("s1", for_destroy=True) is False
+
+
+def test_a_stale_teardown_marker_expires(stores):
+    """A destroyer that dies mid-stop must not block the container forever."""
+    a = stores.make("A", ttl_seconds=0.2)
+    b = stores.make("B")
+    a.claim("s1", for_destroy=True)
+    assert b.take("s1") is False
+
+    time.sleep(0.35)
+
+    assert b.take("s1") is True
+
+
+def test_renew_does_not_keep_a_teardown_alive(stores):
+    """A teardown marker is not a normal lease; renewal must not extend it."""
+    a = stores.make("A")
+    a.claim("s1", for_destroy=True)
+
+    assert a.renew("s1") is RenewOutcome.LOST
+
+
+# ── renew(): distinguishes lapsed from stolen ───────────────────────────────
+
+
+def test_renew_reports_lapsed_not_lost_for_an_expired_lease(stores):
+    """Collapsing these is what dropped every live sandbox on a Redis restart.
+
+    Nobody took the lease — it is simply gone — so the caller is free to
+    re-establish it. Reporting LOST would make the provider evict a container it
+    is actively using.
+    """
+    a = stores.make("A", ttl_seconds=0.2)
+    a.claim("s1")
+
+    time.sleep(0.35)
+
+    assert a.renew("s1") is RenewOutcome.LAPSED
+    assert a.owner("s1") is None
+
+
+def test_renew_does_not_reacquire_on_its_own(stores):
+    """The caller decides; renew() never silently re-takes."""
+    a = stores.make("A", ttl_seconds=0.2)
+    a.claim("s1")
+    time.sleep(0.35)
+
+    a.renew("s1")
+
+    assert a.owner("s1") is None, "renew() re-acquired a lapsed lease by itself"
+
+
+def test_renew_extends_the_lease(stores):
+    a = stores.make("A", ttl_seconds=0.4)
+    b = stores.make("B")
+    a.claim("s1")
+
+    for _ in range(3):
+        time.sleep(0.15)
+        assert a.renew("s1") is RenewOutcome.RENEWED
+
+    assert b.claim("s1") is False, "a renewed lease must keep peers out"
+
+
+def test_renew_of_unknown_sandbox_is_lapsed(stores):
+    a = stores.make("A")
+    assert a.renew("never-claimed") is RenewOutcome.LAPSED
+
+
+# ── release(): only ever drops our own ──────────────────────────────────────
+
+
+def test_release_frees_the_container_for_a_peer(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1")
+
+    a.release("s1")
+
+    assert a.owner("s1") is None
+    assert b.claim("s1") is True
+
+
+def test_release_does_not_clear_a_peers_lease(stores):
+    a = stores.make("A")
+    b = stores.make("B")
+    a.claim("s1")
+
+    b.release("s1")
+
+    assert a.owner("s1") == "A", "B released a lease it does not hold"
+
+
+def test_release_of_unowned_sandbox_is_a_noop(stores):
+    a = stores.make("A")
+    a.release("never-claimed")
+
+
+# ── owner(): read-only ──────────────────────────────────────────────────────
+
+
+def test_owner_returns_none_when_unowned(stores):
+    a = stores.make("A")
+    assert a.owner("nobody") is None
+
+
+def test_owner_does_not_take_ownership(stores):
+    """Unlike claim(), a read must leave ownership untouched."""
+    a = stores.make("A")
+    b = stores.make("B")
+
+    assert b.owner("s1") is None
+
+    assert a.claim("s1") is True, "owner() took the lease as a side effect"
+
+
+# ── Factory / config ────────────────────────────────────────────────────────
+
+
+def test_memory_store_declares_it_cannot_see_peers():
+    store = make_sandbox_ownership_store(SandboxOwnershipConfig(type="memory"), owner_id="A")
+    assert store.supports_cross_process is False
+
+
+def test_default_config_is_memory():
+    store = make_sandbox_ownership_store(None, owner_id="A")
+    assert isinstance(store, MemoryOwnershipStore)
+
+
+def test_unknown_type_raises():
+    config = SandboxOwnershipConfig()
+    object.__setattr__(config, "type", "bogus")
+    with pytest.raises(ValueError, match="Unknown sandbox ownership type"):
+        make_sandbox_ownership_store(config, owner_id="A")
+
+
+def test_ttl_derives_from_renewal_interval_not_idle_timeout():
+    """The coupling that let leases lapse under idle_timeout: 0 must stay broken."""
+    config = SandboxOwnershipConfig(renewal_interval_seconds=30, ttl_multiplier=4)
+    assert compute_lease_ttl(config) == 120
+
+
+def test_ttl_tolerates_missed_renewals():
+    """A single slow renewal cycle must not expire a live owner's lease."""
+    config = SandboxOwnershipConfig()
+    assert compute_lease_ttl(config) > config.renewal_interval_seconds * 2
+
+
+def test_ttl_multiplier_below_two_is_rejected():
+    with pytest.raises(ValueError):
+        SandboxOwnershipConfig(ttl_multiplier=1.0)
+
+
+def test_owner_ids_are_unique_per_instance():
+    """Two workers on one host must not share an owner id."""
+    assert generate_owner_id() != generate_owner_id()
+
+
+def test_stream_bridge_redis_env_implies_redis_ownership(monkeypatch):
+    """A deployment already using redis for the stream bridge is multi-instance.
+
+    Defaulting it to memory ownership would leave #4206 open on exactly the
+    deployments that hit it.
+    """
+    monkeypatch.setenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "redis://somewhere:6379/0")
+    resolved = resolve_ownership_config(None)
+    assert resolved.type == "redis"
+    assert resolved.redis_url == "redis://somewhere:6379/0"
+
+
+def test_stream_bridge_redis_in_config_yaml_implies_redis_ownership(monkeypatch):
+    """The config.yaml-native way of using redis must trigger the inference too.
+
+    The stream bridge's own resolver reads `app_config.stream_bridge` *before*
+    the env var, so inferring only from the env var missed every deployment that
+    configures the bridge in config.yaml — i.e. exactly the multi-instance
+    deployments this inference exists for.
+    """
+    monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False)
+    monkeypatch.delenv("DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL", raising=False)
+
+    resolved = resolve_ownership_config(None, stream_bridge=StreamBridgeConfig(type="redis", redis_url="redis://in-yaml:6379/0"))
+
+    assert resolved.type == "redis"
+    assert resolved.redis_url == "redis://in-yaml:6379/0"
+
+
+def test_memory_stream_bridge_does_not_imply_redis_ownership(monkeypatch):
+    """The other direction: a single-process bridge must not force redis."""
+    monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False)
+    monkeypatch.delenv("DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL", raising=False)
+
+    resolved = resolve_ownership_config(None, stream_bridge=StreamBridgeConfig(type="memory"))
+
+    assert resolved.type == "memory"
+
+
+def test_explicit_config_wins_over_env(monkeypatch):
+    monkeypatch.setenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "redis://somewhere:6379/0")
+    resolved = resolve_ownership_config(SandboxOwnershipConfig(type="memory"))
+    assert resolved.type == "memory"
+
+
+def test_no_env_defaults_to_memory(monkeypatch):
+    monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False)
+    monkeypatch.delenv("DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL", raising=False)
+    assert resolve_ownership_config(None).type == "memory"
+
+
+# ── Redis-specific: failure surfaces as OwnershipBackendError ───────────────
+
+
+def test_redis_backend_error_is_wrapped_not_leaked():
+    """Callers fail closed on OwnershipBackendError; a raw RedisError would escape that.
+
+    Deliberately **not** marked `integration`/`requires_redis`: it points at a
+    dead port, so it needs no server — only the `redis` package, which is pinned
+    in the dev group. Gating it on a live Redis would mean the fail-closed
+    contract was never exercised in CI, which is the one place it matters.
+    """
+    from deerflow.community.aio_sandbox.ownership.redis import RedisOwnershipStore
+
+    store = RedisOwnershipStore(
+        owner_id="A",
+        redis_url="redis://127.0.0.1:1/0",  # nothing listening
+        ttl_seconds=60,
+        key_prefix=f"deerflow:test:{uuid.uuid4().hex}",
+    )
+    with pytest.raises(OwnershipBackendError):
+        store.claim("s1")
+    with pytest.raises(OwnershipBackendError):
+        store.claim("s1", for_destroy=True)
+    with pytest.raises(OwnershipBackendError):
+        store.take("s1")
+    with pytest.raises(OwnershipBackendError):
+        store.renew("s1")
+    with pytest.raises(OwnershipBackendError):
+        store.release("s1")
+    with pytest.raises(OwnershipBackendError):
+        store.owner("s1")
+
+
+def test_non_destroy_claim_does_not_unwind_our_own_teardown(stores):
+    """A `for_destroy=False` claim must not downgrade our own `del:` marker.
+
+    The stop it marks is already in flight and cannot be recalled, so turning the
+    lease back into `own:` would let a `take()` hand out a container that is
+    about to die — the #4206 failure, self-inflicted. No caller does this today
+    (the non-destroy callers run against an absent or unowned key), but the
+    contract has to forbid it rather than rely on that staying true.
+
+    Runs against both backends on purpose: the redis rule lives in Lua and the
+    memory rule in Python, so a fix applied to one only would drift silently.
+    """
+    a = stores.make("A")
+
+    assert a.claim("s1", for_destroy=True) is True
+    assert a.claim("s1") is False, "a non-destroy claim unwound our own in-flight teardown"
+    # Still a teardown: the marker survived the refused claim intact.
+    assert a.renew("s1") is RenewOutcome.LOST
+    b = stores.make("B")
+    assert b.take("s1") is False, "the teardown marker stopped refusing takes"
+
+    # Refreshing the teardown itself is still allowed — that is the heartbeat.
+    assert a.claim("s1", for_destroy=True) is True
+
+
+def test_concurrent_claims_serialize_to_one_winner(stores):
+    """The exclusion must hold under contention, not just in sequence.
+
+    The rest of this suite drives sequential calls, so it pins the predicate and
+    not the atomicity the predicate depends on — redis carries it in Lua, the
+    memory store in a process lock. Eight instances race for one container; the
+    read-modify-write is only atomic if exactly one wins.
+    """
+    barrier = threading.Barrier(8)
+    results = {}
+    lock = threading.Lock()
+
+    def contend(name):
+        store = stores.make(name)
+        barrier.wait(timeout=5)
+        won = store.claim("s1")
+        with lock:
+            results[name] = won
+
+    threads = [threading.Thread(target=contend, args=(f"W{i}",), daemon=True) for i in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=10)
+        assert not t.is_alive(), "a contending claim never finished"
+
+    winners = [name for name, won in results.items() if won]
+    assert len(winners) == 1, f"claim is not atomic under contention: {len(winners)} winners ({winners})"
+
+
+@pytest.mark.integration
+@requires_redis
+def test_redis_store_declares_cross_process_support():
+    from deerflow.community.aio_sandbox.ownership.redis import RedisOwnershipStore
+
+    store = RedisOwnershipStore(owner_id="A", redis_url=REDIS_TEST_URL, ttl_seconds=60, key_prefix=f"deerflow:test:{uuid.uuid4().hex}")
+    try:
+        assert store.supports_cross_process is True
+    finally:
+        store.close()

config.example.yaml

@@ -1185,6 +1185,54 @@ sandbox:
 #   #   DEBUG: "false"
 #   #   API_KEY: $MY_API_KEY        # Reads from host's MY_API_KEY env var
 #   #   DATABASE_URL: $DATABASE_URL  # Reads from host's DATABASE_URL env var
+#
+#   # Optional: Cross-instance container ownership (issue #4206).
+#   #
+#   # Gateway instances share sandbox containers but each keeps its own in-memory
+#   # warm pool. Without shared ownership state, one instance's startup
+#   # reconciliation adopts a container another instance is actively using and
+#   # later idle-destroys it — tool calls then fail with 502 / connection refused.
+#   #
+#   # Single gateway instance? Leave this out; `memory` is the default and the
+#   # cross-instance kill cannot happen.
+#   #
+#   # MULTIPLE gateway instances / workers sharing one container backend
+#   # (load-balanced deployments, Docker Compose) MUST set type: redis. Docker
+#   # Compose already sets DEER_FLOW_STREAM_BRIDGE_REDIS_URL, which is taken as
+#   # proof the deployment is multi-instance, so redis ownership is inferred even
+#   # if this section is omitted.
+#   #
+#   # The redis ownership store requires the optional `redis` extra. It is
+#   # auto-detected from this section on `make dev` and always installed in the
+#   # Docker image. To install it manually:
+#   #     cd backend && uv sync --all-packages --extra redis
+#   #
+#   # ownership:
+#   #   type: memory                 # single gateway instance only
+#   #
+#   # ownership:
+#   #   type: redis                  # required for multi-instance / load-balanced
+#   #   redis_url: redis://redis:6379/0
+#   #   renewal_interval_seconds: 30 # how often an owner refreshes its leases
+#   #   ttl_multiplier: 4            # lease TTL = interval x this (min 2, so a
+#   #                                # single missed renewal cannot expire a live
+#   #                                # owner). Liveness is deliberately independent
+#   #                                # of idle_timeout: renewal keeps running even
+#   #                                # at idle_timeout: 0.
+#   #   key_prefix: deerflow:sandbox:owner
+#   #
+#   # NOTE: the redis ownership store is fail-closed, matching the stream bridge's
+#   # fail-hard policy. Redis.from_url is lazy so a down Redis does not block
+#   # startup, but a sandbox whose ownership cannot be published is not handed out
+#   # — acquiring raises instead. The alternative (proceed unowned) is exactly the
+#   # #4206 cross-instance kill. Run Redis with HA / a restart policy.
+#   #
+#   # NOTE: the other boundary is the lease TTL (renewal_interval_seconds x
+#   # ttl_multiplier). A Redis outage longer than the TTL can let a reconciling
+#   # instance adopt a live owner's container: a lapsed lease is indistinguishable
+#   # from a dead owner, so one TTL of adoption grace is the whole safety margin.
+#   # Size the TTL against your Redis availability target (HA Redis keeps this
+#   # window out of reach).
 
 # Option 3: BoxLite micro-VM Sandbox
 # Runs each sandbox as a BoxLite micro-VM. Released boxes stay in an in-process

scripts/detect_uv_extras.py

@@ -11,8 +11,10 @@ Order of resolution:
    - database.backend == postgres        -> postgres
    - checkpointer.type == postgres       -> postgres
    - stream_bridge.type == redis         -> redis
+   - sandbox.ownership.type == redis     -> redis
 3. Runtime environment toggles that enable optional backends:
    - DEER_FLOW_STREAM_BRIDGE_REDIS_URL   -> redis
+   - DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL -> redis
 
 Each extra name is validated against ``^[A-Za-z][A-Za-z0-9_-]*$`` (the same
 shape uv enforces for `[project.optional-dependencies]` keys). Anything else
@@ -233,6 +235,8 @@ def detect_from_config(path: Path) -> list[str]:
         extras.add("postgres")
     if (section_value(lines, "stream_bridge", "type") or "").lower() == "redis":
         extras.add("redis")
+    if (nested_section_value(lines, "sandbox.ownership", "type") or "").lower() == "redis":
+        extras.add("redis")
     if (nested_section_value(lines, "channels.discord", "enabled") or "").lower() == "true":
         extras.add("discord")
     return sorted(extras)
@@ -242,6 +246,8 @@ def detect_from_runtime_env() -> list[str]:
     extras: set[str] = set()
     if os.environ.get("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "").strip():
         extras.add("redis")
+    if os.environ.get("DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL", "").strip():
+        extras.add("redis")
     return sorted(extras)