diff --git a/CHANGELOG.md b/CHANGELOG.md
index 735acb90b4..7150e0c628 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1397,6 +1397,16 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**.
 
 ### Fixed
 
+- **sandbox:** Stopping the Gateway no longer destroys AIO or E2B sandboxes
+  before in-flight runs drain. When the first sandbox use built the provider on
+  the event-loop thread, the provider's own SIGTERM/SIGINT handler destroyed
+  every sandbox synchronously as the signal arrived: the event loop froze for
+  the whole teardown and running tool calls lost their sandbox. The Gateway now
+  owns sandbox shutdown and closes the provider after the run drain and MCP
+  teardown; embedded clients and scripts keep the signal fallback, and SIGHUP
+  still cleans up. AIO acquire now refuses once shutdown has begun, as E2B
+  already did. ([#7560])
+
 - **frontend:** SSO sign-ins now return to the page that asked for them, such
   as a shared chat link, instead of always opening the default workspace. The
   sign-in pages used to send every signed-in request to `/workspace`, which
@@ -13125,3 +13135,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7521]: https://github.com/bytedance/deer-flow/pull/7521
 [#7530]: https://github.com/bytedance/deer-flow/pull/7530
 [#7557]: https://github.com/bytedance/deer-flow/pull/7557
+[#7560]: https://github.com/bytedance/deer-flow/pull/7560
diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md
index c679787dcc..193128ba4a 100644
--- a/CHANGELOG_zh.md
+++ b/CHANGELOG_zh.md
@@ -1112,6 +1112,13 @@
 
 ### 修复
 
+- **沙箱：** 停止 Gateway 时，AIO 和 E2B 沙箱不再在进行中的运行收尾之前被销毁。此前
+  如果第一次使用沙箱时在事件循环线程上构造了提供方，提供方自己的 SIGTERM/SIGINT
+  处理器会在信号到达时同步销毁所有沙箱：整个销毁期间事件循环冻结，正在执行的工具
+  调用失去沙箱。现在由 Gateway 负责沙箱关闭，在运行收尾和 MCP 关闭之后再关闭提供方；
+  嵌入式客户端和脚本保留信号兜底，SIGHUP 仍会清理。AIO 的 acquire 在关闭开始后会
+  拒绝请求，与 E2B 一致。([#7560])
+
 - **前端：** SSO 登录后现在会回到发起登录的页面（例如分享的对话链接），不再总是打开
   默认工作区。此前登录相关页面会把所有已登录的请求都跳转到 `/workspace`，导致
   `/auth/callback?next=...` 中的 `next` 路径丢失；已登录时访问 `/login?next=...`
@@ -10477,3 +10484,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7521]: https://github.com/bytedance/deer-flow/pull/7521
 [#7530]: https://github.com/bytedance/deer-flow/pull/7530
 [#7557]: https://github.com/bytedance/deer-flow/pull/7557
+[#7560]: https://github.com/bytedance/deer-flow/pull/7560
diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md
index e05751a237..681562e1fa 100644
--- a/backend/app/gateway/AGENTS.md
+++ b/backend/app/gateway/AGENTS.md
@@ -21,10 +21,10 @@ resources unwind; each service `stop()` has a 30-second `asyncio.timeout`.
 Routers with persistent writes drain through
 `app.gateway.persistent_writes.run_drained_write`.
 
-After `langgraph_runtime` exits (run drain and extension teardown), close MCP
-bounded and failure-isolated. Then drain off-thread
-`shutdown_sandbox_provider()` in `finally`, including cancellation. Never
-initialize for cleanup; log failures and retain native retryability.
+`lifespan` first holds `hold_sandbox_shutdown_ownership()`. After
+`langgraph_runtime` drains runs and extensions, close MCP (bounded, isolated),
+then drain `shutdown_sandbox_provider()` off-loop in `finally`, even on
+cancel. Never allocate for cleanup; log failures, stay retryable.
 
 Retention holds durable thread admission and the checkpoint lock before
 checkpoint reads. Inspect root head/parent and leaf writes only; protect selectors,
diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py
index 5b8ac282cb..a098929046 100644
--- a/backend/app/gateway/app.py
+++ b/backend/app/gateway/app.py
@@ -64,6 +64,7 @@ from app.gateway.routers import (
 from app.gateway.trace_middleware import TraceMiddleware
 from deerflow.config import app_config as deerflow_app_config
 from deerflow.logging_config import DEFAULT_LOG_DATE_FORMAT, DEFAULT_LOG_FORMAT, configure_logging
+from deerflow.sandbox.sandbox_provider import hold_sandbox_shutdown_ownership
 from deerflow.tracing.monocle import setup_monocle_tracing_if_enabled
 from deerflow.uploads.manager import cleanup_stale_upload_staging_files
 from deerflow.utils.file_io import await_drained
@@ -623,7 +624,17 @@ async def _runtime_with_mcp_pool_shutdown(app: FastAPI, startup_config: AppConfi
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan handler."""
+    # Claim sandbox shutdown before any startup step can build a provider: a
+    # provider built on the event-loop thread would otherwise install SIGTERM/
+    # SIGINT handlers that destroy every sandbox before uvicorn starts the run
+    # drain. `_runtime_with_mcp_pool_shutdown` closes the provider afterwards.
+    with hold_sandbox_shutdown_ownership():
+        async with _gateway_lifespan(app):
+            yield
 
+
+@asynccontextmanager
+async def _gateway_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # Load config and check necessary environment variables at startup.
     # `startup_config` is a local snapshot used only for one-shot bootstrap
     # work (logging level, langgraph_runtime engines, channels). Request-time
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md b/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md
index cffeb7e4ad..2a452393b3 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md
@@ -82,6 +82,24 @@ lock, only for the entry that passed reuse checks; a changed entry must retry.
 Partial shutdown raises and remains retryable; successful resources are not retried,
 and cleanup infrastructure closes only after every owned teardown completes.
 
+Signal handlers are a fallback for hosts without a lifespan (embedded
+`DeerFlowClient`, scripts). The handler runs `shutdown()` synchronously and
+force-destroys every sandbox, so on the Gateway's event-loop thread it would
+freeze the loop and kill in-flight runs before uvicorn starts the drain. Under
+`hold_sandbox_shutdown_ownership()` the constructor therefore installs only
+SIGHUP, which uvicorn does not handle; SIGTERM/SIGINT go to uvicorn and the
+lifespan hook shuts the provider down after the run drain and MCP teardown. A
+provider built before the declaration keeps its handlers: the Gateway declares
+first, and uvicorn already replaced any earlier handler and re-raises captured
+signals only after lifespan shutdown.
+
+`shutdown()` closes admission (`_admission_closed`) under `_lock` before its
+snapshot and never reopens it, even when a failed teardown resets
+`_shutdown_called` for retry. `acquire`/`acquire_async`, the post-serializer
+recheck and every active commit (warm reclaim, discovery, create) then raise
+`SandboxCapacityExceededError(reason="shutdown")`. A refused warm reclaim stays
+warm for shutdown to destroy; a refused create rolls back its container.
+
 Broker capability failures are unknown, never confirmed negatives. Cache only
 successful observations per provisioner endpoint/auth identity; a failed refresh
 may retain a confirmed broker requirement. A provisioner lark-configuration conflict marks its create response; the
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
index b12875ad99..2c777853c9 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
@@ -56,10 +56,11 @@ from deerflow.integrations.lark_cli import (
 )
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.acquire_serialization import AcquireSerializer
+from deerflow.sandbox.exceptions import SandboxCapacityExceededError
 from deerflow.sandbox.identity import derive_sandbox_scope_token
 from deerflow.sandbox.lease import discard_sandbox_lease_manager, get_sandbox_lease_manager, run_sync_lifecycle_operation
 from deerflow.sandbox.sandbox import Sandbox
-from deerflow.sandbox.sandbox_provider import SandboxProvider
+from deerflow.sandbox.sandbox_provider import SandboxProvider, host_owns_sandbox_shutdown
 from deerflow.skills.types import SkillCategory
 from deerflow.utils.file_io import await_drained
 
@@ -286,6 +287,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         self._acquire_epoch_counter = 0
         self._acquire_inflight: dict[str, int] = {}
         self._shutdown_called = False
+        # Unlike `_shutdown_called`, a failed teardown never reopens admission.
+        self._admission_closed = False
         self._cleanup_retry_ids: set[str] = set()
         self._idle_checker_stop = threading.Event()
         self._idle_checker_thread: threading.Thread | None = None
@@ -1599,10 +1602,17 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
         Handles SIGTERM, SIGINT, and SIGHUP (terminal close) to ensure
         sandbox containers are cleaned up even when the user closes the terminal.
+        A host that owns sandbox shutdown (the Gateway lifespan) keeps SIGTERM
+        and SIGINT: it drains in-flight runs before shutting the provider down,
+        and this synchronous teardown would run first on the event-loop thread.
+        SIGHUP keeps the fallback because uvicorn does not handle it.
         """
         self._original_sigterm = signal.getsignal(signal.SIGTERM)
         self._original_sigint = signal.getsignal(signal.SIGINT)
         self._original_sighup = signal.getsignal(signal.SIGHUP) if hasattr(signal, "SIGHUP") else None
+        signums = [] if host_owns_sandbox_shutdown() else [signal.SIGTERM, signal.SIGINT]
+        if hasattr(signal, "SIGHUP"):
+            signums.append(signal.SIGHUP)
 
         def signal_handler(signum, frame):
             try:
@@ -1622,10 +1632,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 signal.raise_signal(signum)
 
         try:
-            signal.signal(signal.SIGTERM, signal_handler)
-            signal.signal(signal.SIGINT, signal_handler)
-            if hasattr(signal, "SIGHUP"):
-                signal.signal(signal.SIGHUP, signal_handler)
+            for signum in signums:
+                signal.signal(signum, signal_handler)
         except ValueError:
             logger.debug("Could not register signal handlers (not main thread)")
 
@@ -1888,6 +1896,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 logger.info("Warm-pool sandbox %s was claimed for teardown while publishing ownership; not reclaiming it", sandbox_id)
                 return None
             self._assert_warm_identity_available_locked(sandbox_id, key)
+            # Leave the entry warm for shutdown's own warm-pool teardown.
+            self._raise_if_shutting_down()
             warm_item = self._warm_pool.get(sandbox_id)
             if warm_item is None or warm_item[0] is not info:
                 # A different entry has not passed the reuse checks above.
@@ -1972,6 +1982,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                     raise SandboxBeingDestroyedError(info.sandbox_id)
                 self._assert_active_identity_available_locked(info.sandbox_id, key)
                 self._assert_warm_identity_available_locked(info.sandbox_id, key)
+                self._raise_if_shutting_down()
                 # Active and warm are exclusive states, and only this insert can
                 # violate that: a warm entry for the same id is stale the moment
                 # the id becomes active. Leaving it there gives the container two
@@ -2025,6 +2036,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 if key is not None:
                     self._assert_active_identity_available_locked(sandbox_id, key)
                     self._assert_warm_identity_available_locked(sandbox_id, key)
+                # A create that outlived shutdown's snapshot rolls back below.
+                self._raise_if_shutting_down()
                 # Same exclusivity rule as the discover path.
                 self._warm_pool.pop(sandbox_id, None)
                 self._warm_pool_identity.pop(sandbox_id, None)
@@ -2239,6 +2252,19 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
     # ── Core: acquire / get / release / shutdown ─────────────────────────
 
+    def _raise_if_shutting_down(self) -> None:
+        """Refuse admission once shutdown began; safe with or without ``_lock`` held."""
+        if not getattr(self, "_admission_closed", False):
+            return
+        raise SandboxCapacityExceededError(
+            "Sandbox provider is shutting down; cannot acquire sandbox",
+            active=len(self._sandboxes),
+            warm=len(self._warm_pool),
+            replicas=int(self._config.get("replicas", DEFAULT_REPLICAS)),
+            retry_after_seconds=30.0,
+            reason="shutdown",
+        )
+
     def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
         """Acquire a sandbox environment and return its ID.
 
@@ -2254,6 +2280,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         Returns:
             The ID of the acquired sandbox environment.
         """
+        self._raise_if_shutting_down()
         effective_user_id = self._effective_acquire_user_id(user_id)
         if thread_id:
             with self._acquire_serializer.hold(self._thread_key(thread_id, effective_user_id)):
@@ -2267,6 +2294,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         event loop and using async-native readiness polling for newly created
         sandboxes.
         """
+        self._raise_if_shutting_down()
         effective_user_id = self._effective_acquire_user_id(user_id)
         if thread_id:
             async with self._acquire_serializer.hold_async(self._thread_key(thread_id, effective_user_id)):
@@ -2281,6 +2309,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                  sandbox_id is deterministic from thread_id so no shared state file
                  is needed — any process can derive the same container name)
         """
+        # The serializer may have queued this caller before shutdown started.
+        self._raise_if_shutting_down()
         self._ensure_skills_projection(user_id)
         cached_id = self._reuse_in_process_sandbox(thread_id, user_id=user_id)
         if cached_id is not None:
@@ -2309,6 +2339,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
     async def _acquire_internal_async(self, thread_id: str | None, *, user_id: str) -> str:
         """Async counterpart to ``_acquire_internal``."""
+        self._raise_if_shutting_down()
         await asyncio.to_thread(self._ensure_skills_projection, user_id)
         cached_id = await _run_started_acquire_worker(
             self._acquire_worker_executor,
@@ -2953,6 +2984,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
     def shutdown(self) -> None:
         """Shutdown all sandboxes. Thread-safe and idempotent."""
         with self._lock:
+            # Closed under the lock that guards every active/warm commit, so no
+            # acquire can install a sandbox after the snapshot below.
+            self._admission_closed = True
             if self._shutdown_called:
                 return
             self._shutdown_called = True
diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md b/backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md
index 191d44d3dd..dcc4914f85 100644
--- a/backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md
+++ b/backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md
@@ -57,7 +57,9 @@ and preserve all tracked state so a later `shutdown()` can retry cleanup.
 That fence covers cached sync/async acquisition too: recheck after serializer
 waits and immediately before exposing a reused client. Signal-triggered shutdown
 may report deferred cleanup, but it must still forward the process's original
-SIGTERM/SIGINT/SIGHUP action while retaining the pending E2B state.
+SIGTERM/SIGINT/SIGHUP action while retaining the pending E2B state. Under
+`hold_sandbox_shutdown_ownership()` (Gateway lifespan) only SIGHUP is
+installed; the lifespan shuts the provider down after the run drain.
 
 ## File search
 
diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py
index 418010a84e..04b34b2845 100644
--- a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py
@@ -68,7 +68,7 @@ from deerflow.sandbox.acquire_serialization import AcquireSerializer
 from deerflow.sandbox.exceptions import SandboxCapacityExceededError
 from deerflow.sandbox.identity import derive_sandbox_scope_token
 from deerflow.sandbox.sandbox import Sandbox
-from deerflow.sandbox.sandbox_provider import SandboxProvider
+from deerflow.sandbox.sandbox_provider import SandboxProvider, host_owns_sandbox_shutdown
 
 from ..aio_sandbox.ownership import (
     OwnershipBackendError,
@@ -584,7 +584,10 @@ class E2BSandboxProvider(SandboxProvider):
                 signal.signal(signum, signal.SIG_DFL)
                 signal.raise_signal(signum)
 
-        for sig_name in ("SIGTERM", "SIGINT", "SIGHUP"):
+        # A host that owns shutdown (the Gateway lifespan) drains runs before
+        # tearing sandboxes down; SIGHUP keeps the fallback uvicorn lacks.
+        sig_names = ("SIGHUP",) if host_owns_sandbox_shutdown() else ("SIGTERM", "SIGINT", "SIGHUP")
+        for sig_name in sig_names:
             sig = getattr(signal, sig_name, None)
             if sig is None:
                 continue
diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md
index a80dcb036f..53e07d6453 100644
--- a/backend/packages/harness/deerflow/sandbox/AGENTS.md
+++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md
@@ -36,6 +36,7 @@ cancellation before propagating it.
 **Environment policy** (`sandbox/env_policy.py`): `build_sandbox_env()` scrubs inherited credential names via `is_blocked_env_name`, preserving benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`). Explicit `sandbox.environment` injection survives scrubbing; request secrets override it. Local sandboxes validate keys at construction; generic and per-thread instances receive resolved config values verbatim. Local `bash` masks effective operator credentials and request secrets in output and command errors, preserving benign config values.
 Holders stay bound through scope cleanup; AIO reserves idle teardown atomically.
 Failed provider shutdown retains the singleton and lease manager for retry unless replaced concurrently.
+Hosts that drain runs first (Gateway lifespan) enter `hold_sandbox_shutdown_ownership()` before any provider exists; providers built inside skip SIGTERM/SIGINT handlers but keep SIGHUP.
 Fork lookups reject invalidated clients; cancelled acquires leave borrowers a pending release.
 **Implementations**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-user/thread `LocalSandbox` (id `local:{user_id}:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data` (file tools: `ls` lists only these three; search/read reject the root), its `{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`. Shared runs use category mappings; a policy-scoped run replaces them with one `/mnt/skills` root mapping to the coherent thread view, so structured file tools resolve through one managed boundary. This is not a host filesystem security boundary: an enabled host `bash` subprocess can use canonical paths without `PathMapping`, so `supports_agent_skill_isolation` is dynamic and explicit Agent policies fail closed while host bash is enabled; unrestricted runs then get the shared mappings, never a leftover thread view. Host-to-virtual output masking scans dynamic per-user/per-thread roots directly instead of compiling path-specific regexes, so evicted thread IDs do not remain in Python's global regex caches; a separate 256-entry root cache prevents repeated `realpath()` walks for every glob/grep match while bounding dynamic-path retention, and only the small process-stable skill/integration source set uses a bounded compiled cache. The shared `path_patterns.py` tail stops at `:`, so `$PATH`-style lists mask every entry; a mount symlink resolving outside all mounts keeps its mount path. On Windows, Git Bash/MSYS argument-conversion exclusions are limited to safe non-root virtual path prefixes; do not restore a blanket conversion disable, because host-native CLI launchers need normal MSYS path conversion for their own installation paths.
diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
index b608dec188..4c1b760951 100644
--- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
+++ b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
@@ -1,6 +1,8 @@
 import asyncio
 import threading
 from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from contextlib import contextmanager
 from typing import TYPE_CHECKING
 
 from deerflow.config import get_app_config
@@ -176,6 +178,37 @@ _default_sandbox_provider: SandboxProvider | None = None
 # slow teardown. Keeping callbacks off the lock avoids both.
 _provider_lock = threading.Lock()
 
+# Number of hosts (the Gateway lifespan) currently owning provider shutdown.
+# Providers built while it is positive leave SIGTERM/SIGINT to the host, which
+# drains in-flight runs before calling `shutdown_sandbox_provider()`. Guarded by
+# `_provider_lock`; read by provider constructors, never under it.
+_host_shutdown_owners = 0
+
+
+@contextmanager
+def hold_sandbox_shutdown_ownership() -> Iterator[None]:
+    """Declare that the host, not provider signal handlers, shuts sandboxes down.
+
+    Enter this before any provider can be built. Providers built inside it skip
+    their SIGTERM/SIGINT handlers and keep the SIGHUP fallback, since uvicorn
+    does not handle SIGHUP. A provider built earlier keeps the handlers it
+    already installed.
+    """
+    global _host_shutdown_owners
+    with _provider_lock:
+        _host_shutdown_owners += 1
+    try:
+        yield
+    finally:
+        with _provider_lock:
+            _host_shutdown_owners -= 1
+
+
+def host_owns_sandbox_shutdown() -> bool:
+    """Return whether a host currently owns sandbox provider shutdown."""
+    with _provider_lock:
+        return _host_shutdown_owners > 0
+
 
 def get_initialized_sandbox_provider() -> SandboxProvider | None:
     """Return the provider only when another lifecycle path initialized it."""
diff --git a/backend/tests/_gateway_shutdown_budget.py b/backend/tests/_gateway_shutdown_budget.py
index e8d100fa83..eb862d7621 100644
--- a/backend/tests/_gateway_shutdown_budget.py
+++ b/backend/tests/_gateway_shutdown_budget.py
@@ -65,6 +65,7 @@ def lifespan_shutdown_seconds() -> float:
 
     Unbounded steps (OIDC close, MCP task-service stop, backend close) are
     expected to be quick and are not modeled; deployments keep slack on top of
-    this figure for them.
+    this figure for them. Sandbox provider shutdown is unbounded too and is not
+    quick: it runs one backend destroy per tracked sandbox after the run drain.
     """
     return bounded_hooks_seconds() + retrieval_warm_wait_seconds() + run_drain_seconds() + memory_flush_seconds()
diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py
index b4e1735254..616632d79b 100644
--- a/backend/tests/test_aio_sandbox_provider.py
+++ b/backend/tests/test_aio_sandbox_provider.py
@@ -3769,6 +3769,120 @@ def test_signal_handler_forwards_original_signal_when_shutdown_fails(tmp_path, m
     assert "Sandbox shutdown failed while handling signal" in caplog.text
 
 
+@pytest.mark.parametrize(
+    ("host_owned", "expected"),
+    [(False, {"SIGTERM", "SIGINT", "SIGHUP"}), (True, {"SIGHUP"})],
+    ids=["standalone", "host-owned"],
+)
+def test_signal_handlers_leave_stop_signals_to_a_host_that_owns_shutdown(tmp_path, monkeypatch, host_owned, expected):
+    """The Gateway drains runs before sandbox shutdown; only SIGHUP, which uvicorn ignores, keeps the fallback."""
+    from deerflow.sandbox import sandbox_provider as registry
+
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    if not hasattr(aio_mod.signal, "SIGHUP"):
+        pytest.skip("SIGHUP not available on this platform")
+    provider = _make_provider(tmp_path)
+    provider.shutdown = MagicMock()
+    forwarded = MagicMock()
+    installed = {}
+    monkeypatch.setattr(aio_mod.signal, "getsignal", lambda _signum: forwarded)
+    monkeypatch.setattr(aio_mod.signal, "signal", lambda signum, handler: installed.__setitem__(signum, handler))
+
+    with registry.hold_sandbox_shutdown_ownership() if host_owned else contextlib.nullcontext():
+        provider._register_signal_handlers()
+
+    assert {aio_mod.signal.Signals(signum).name for signum in installed} == expected
+    installed[aio_mod.signal.SIGHUP](aio_mod.signal.SIGHUP, None)
+    provider.shutdown.assert_called_once_with()
+    forwarded.assert_called_once_with(aio_mod.signal.SIGHUP, None)
+
+
+def test_acquire_refuses_once_shutdown_began_even_after_a_failed_teardown(tmp_path):
+    """A retryable shutdown failure reopens shutdown, not admission."""
+    from deerflow.sandbox.exceptions import SandboxCapacityExceededError
+
+    provider = _make_provider(tmp_path)
+    provider._lock = threading.Lock()
+    provider._warm_pool = {}
+    provider._shutdown_called = False
+    provider._stop_idle_checker = MagicMock(side_effect=RuntimeError("reaper still alive"))
+    provider._ensure_skills_projection = MagicMock()
+
+    with pytest.raises(RuntimeError, match="reaper still alive"):
+        provider.shutdown()
+    assert provider._shutdown_called is False
+
+    with pytest.raises(SandboxCapacityExceededError) as sync_exc:
+        provider.acquire("thread-late", user_id="alice")
+    assert sync_exc.value.reason == "shutdown"
+    with pytest.raises(SandboxCapacityExceededError) as async_exc:
+        asyncio.run(provider.acquire_async("thread-late", user_id="alice"))
+    assert async_exc.value.reason == "shutdown"
+    # A caller already queued on the per-thread serializer rechecks after its wait.
+    with pytest.raises(SandboxCapacityExceededError):
+        provider._acquire_internal("thread-late", user_id="alice")
+    with pytest.raises(SandboxCapacityExceededError):
+        asyncio.run(provider._acquire_internal_async("thread-late", user_id="alice"))
+    provider._ensure_skills_projection.assert_not_called()
+    provider.reset()
+
+
+def test_sandbox_created_after_shutdown_began_is_rolled_back_not_tracked(tmp_path):
+    """An in-flight create that finishes after shutdown's snapshot must not leak its container."""
+    from deerflow.sandbox.exceptions import SandboxCapacityExceededError
+
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._lock = threading.Lock()
+    provider._warm_pool = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._last_activity = {}
+    provider._publish_ownership = MagicMock()
+    provider._destroy_unready_sandbox = MagicMock()
+    provider._admission_closed = True
+    info = aio_mod.SandboxInfo(sandbox_id="sandbox-late", sandbox_url="http://sandbox-late")
+
+    with pytest.raises(SandboxCapacityExceededError):
+        provider._register_created_sandbox("thread-late", "sandbox-late", info, user_id="alice")
+
+    provider._destroy_unready_sandbox.assert_called_once_with("sandbox-late", info)
+    assert "sandbox-late" not in provider._sandboxes
+    provider.reset()
+
+
+@pytest.mark.parametrize("path", ["warm-reclaim", "discovered"])
+def test_shutdown_starting_during_ownership_publish_blocks_activation(tmp_path, path):
+    """Shutdown that begins during the store round trip must leave nothing active behind."""
+    from deerflow.sandbox.exceptions import SandboxCapacityExceededError
+
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._lock = threading.Lock()
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._last_activity = {}
+    provider._check_sandbox_reuse_policy = MagicMock()
+    provider._check_tracked_sandbox_alive = MagicMock(return_value=True)
+    provider._publish_ownership = MagicMock(side_effect=lambda _sandbox_id: setattr(provider, "_admission_closed", True))
+    info = aio_mod.SandboxInfo(sandbox_id="sandbox-late", sandbox_url="http://sandbox-late")
+    provider._warm_pool = {"sandbox-late": (info, 1.0)} if path == "warm-reclaim" else {}
+
+    with pytest.raises(SandboxCapacityExceededError), patch.object(aio_mod, "AioSandbox") as sandbox_cls:
+        if path == "warm-reclaim":
+            provider._reclaim_warm_pool_sandbox("thread-late", "sandbox-late", user_id="alice")
+        else:
+            provider._register_discovered_sandbox("thread-late", info, user_id="alice")
+
+    assert provider._sandboxes == {} and provider._thread_sandboxes == {}
+    if path == "warm-reclaim":
+        # Still warm, so shutdown's warm-pool pass destroys it.
+        assert provider._warm_pool == {"sandbox-late": (info, 1.0)}
+    else:
+        sandbox_cls.return_value.close.assert_called_once_with()
+    provider.reset()
+
+
 def test_cleanup_idle_sandboxes_keeps_active_cleanup_and_delegates_warm_expiry(tmp_path):
     """AIO active-idle cleanup must remain local while warm expiry uses the shared lifecycle."""
     aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py
index b6f57d32f6..3d2408a834 100644
--- a/backend/tests/test_e2b_sandbox_provider.py
+++ b/backend/tests/test_e2b_sandbox_provider.py
@@ -3493,6 +3493,42 @@ def test_signal_handler_forwards_original_action_when_shutdown_cleanup_is_pendin
     assert p._shutdown_cleanup_pending is True
 
 
+@pytest.mark.parametrize(
+    ("host_owned", "expected"),
+    [(False, {"SIGTERM", "SIGINT", "SIGHUP"}), (True, {"SIGHUP"})],
+    ids=["standalone", "host-owned"],
+)
+def test_signal_handlers_leave_stop_signals_to_a_host_that_owns_shutdown(monkeypatch, host_owned, expected):
+    """The Gateway drains runs before sandbox shutdown; only SIGHUP, which uvicorn ignores, keeps the fallback."""
+    from deerflow.sandbox import sandbox_provider as registry
+
+    if not hasattr(signal, "SIGHUP"):
+        pytest.skip("SIGHUP not available on this platform")
+    p = _make_provider()
+    registered: dict[int, Any] = {}
+    forwarded: list[tuple[int, Any]] = []
+    shutdowns: list[None] = []
+
+    def original_handler(signum, frame):
+        forwarded.append((signum, frame))
+
+    monkeypatch.setattr(signal, "getsignal", lambda _signum: original_handler)
+    monkeypatch.setattr(signal, "signal", lambda signum, handler: registered.__setitem__(signum, handler))
+    monkeypatch.setattr(p, "shutdown", lambda: shutdowns.append(None))
+
+    if host_owned:
+        with registry.hold_sandbox_shutdown_ownership():
+            p._register_signal_handlers()
+    else:
+        p._register_signal_handlers()
+
+    assert {signal.Signals(signum).name for signum in registered} == expected
+    frame = object()
+    registered[signal.SIGHUP](signal.SIGHUP, frame)
+    assert shutdowns == [None]
+    assert forwarded == [(signal.SIGHUP, frame)]
+
+
 def test_shutdown_only_kills_sandboxes_owned_by_current_instance(monkeypatch):
     p = _make_provider()
     owned_client = FakeClient(sandbox_id="sb-owned")
diff --git a/backend/tests/test_gateway_sandbox_shutdown.py b/backend/tests/test_gateway_sandbox_shutdown.py
index 1686ef8b71..9e23b46232 100644
--- a/backend/tests/test_gateway_sandbox_shutdown.py
+++ b/backend/tests/test_gateway_sandbox_shutdown.py
@@ -1,10 +1,13 @@
 """Gateway owns sandbox cleanup even when lazy creation cannot bind signals."""
 
 import asyncio
+import importlib
+import signal
 import threading
+import time
 from contextlib import asynccontextmanager
 from types import SimpleNamespace
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from fastapi import FastAPI
@@ -195,3 +198,115 @@ def test_shutdown_drains_sandbox_worker_across_cancellation(monkeypatch, cancel_
             await asyncio.gather(task, return_exceptions=True)
 
     asyncio.run(scenario())
+
+
+@pytest.fixture
+def restore_signal_handlers():
+    signums = [signal.SIGTERM, signal.SIGINT] + ([signal.SIGHUP] if hasattr(signal, "SIGHUP") else [])
+    saved = {signum: signal.getsignal(signum) for signum in signums}
+    try:
+        yield
+    finally:
+        for signum, handler in saved.items():
+            signal.signal(signum, handler)
+
+
+def _build_aio_provider(monkeypatch, backend, sandbox_ids):
+    """Run the real AIO constructor (signal registration included) over a fake backend."""
+    from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
+
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    config = {"replicas": 10, "idle_timeout": 0, "ownership": None, "stream_bridge": None, "command_timeout": 600.0}
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_load_config", lambda _self: config)
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_create_backend", lambda _self: backend)
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_reconcile_orphans", lambda _self: None)
+    provider = aio_mod.AioSandboxProvider()
+    for sandbox_id in sandbox_ids:
+        sandbox = MagicMock(id=sandbox_id, requires_container_recycle=False, has_pending_session_creates=False)
+        with provider._lock:
+            provider._sandbox_infos[sandbox_id] = SandboxInfo(sandbox_id=sandbox_id, sandbox_url=f"http://fake/{sandbox_id}")
+            provider._sandboxes[sandbox_id] = sandbox
+            provider._last_activity[sandbox_id] = time.monotonic()
+    return provider
+
+
+@pytest.mark.parametrize("signum", [signal.SIGTERM, signal.SIGINT], ids=["SIGTERM", "SIGINT"])
+def test_stop_signal_with_main_thread_provider_destroys_sandboxes_only_after_run_drain(monkeypatch, restore_signal_handlers, signum):
+    """G6/S0: a provider built on the event-loop thread must not tear down from its signal handler."""
+    from app.gateway.app import _runtime_with_mcp_pool_shutdown
+    from deerflow.sandbox import sandbox_provider as registry
+
+    sandbox_ids = ["sbx-0", "sbx-1", "sbx-2"]
+    events = []
+
+    class Backend:
+        def destroy(self, info):
+            events.append(f"destroy:{info.sandbox_id}")
+
+        def list_running(self):
+            return []
+
+    def uvicorn_handle_exit(received, _frame):
+        events.append(f"handle_exit:{signal.Signals(received).name}")
+
+    # uvicorn installs its handlers before lifespan startup builds anything.
+    signal.signal(signal.SIGTERM, uvicorn_handle_exit)
+    signal.signal(signal.SIGINT, uvicorn_handle_exit)
+    monkeypatch.setattr(registry, "_default_sandbox_provider", None)
+    pool = _runtime(monkeypatch, events)
+
+    async def scenario():
+        provider = None
+        with registry.hold_sandbox_shutdown_ownership():
+            try:
+                provider = _build_aio_provider(monkeypatch, Backend(), sandbox_ids)
+                assert threading.current_thread() is threading.main_thread()
+                registry.set_sandbox_provider(provider)
+
+                @asynccontextmanager
+                async def draining_runtime(_app, _config):
+                    try:
+                        yield
+                    finally:
+                        # The in-flight run still reaches its sandbox while it drains.
+                        events.append("runtime_drained" if provider.get("sbx-0") is not None else "runtime_drained_without_sandbox")
+
+                monkeypatch.setattr("app.gateway.app.langgraph_runtime", draining_runtime)
+                async with _runtime_with_mcp_pool_shutdown(FastAPI(), SimpleNamespace()):
+                    signal.raise_signal(signum)
+                    await asyncio.sleep(0)
+                    assert events == [f"handle_exit:{signal.Signals(signum).name}"]
+            finally:
+                if provider is not None:
+                    provider.shutdown()
+
+        assert events[:3] == [f"handle_exit:{signal.Signals(signum).name}", "runtime_drained", "mcp_closed"]
+        assert sorted(events[3:]) == [f"destroy:{sandbox_id}" for sandbox_id in sandbox_ids]
+        assert registry.get_initialized_sandbox_provider() is None
+        pool.close_all.assert_awaited_once()
+
+    asyncio.run(scenario())
+
+
+def test_gateway_lifespan_owns_sandbox_shutdown_from_its_first_step(monkeypatch):
+    """The declaration must precede every startup step that can build a provider."""
+    from app.gateway import app as gateway_app
+    from deerflow.sandbox import sandbox_provider as registry
+
+    observed = []
+
+    def failing_config():
+        observed.append(registry.host_owns_sandbox_shutdown())
+        raise RuntimeError("stop startup here")
+
+    monkeypatch.setattr(gateway_app, "get_app_config", failing_config)
+
+    async def scenario():
+        assert registry.host_owns_sandbox_shutdown() is False
+        with pytest.raises(RuntimeError, match="stop startup here"):
+            async with gateway_app.lifespan(FastAPI()):
+                pytest.fail("startup should have failed")
+        assert observed == [True]
+        assert registry.host_owns_sandbox_shutdown() is False
+
+    asyncio.run(scenario())
diff --git a/backend/tests/test_sandbox_provider_lifecycle.py b/backend/tests/test_sandbox_provider_lifecycle.py
index 37b7d03db1..889ab4c841 100644
--- a/backend/tests/test_sandbox_provider_lifecycle.py
+++ b/backend/tests/test_sandbox_provider_lifecycle.py
@@ -333,3 +333,15 @@ def test_failed_shutdown_does_not_replace_a_concurrently_installed_provider():
     finally:
         sandbox_provider.reset_sandbox_provider()
         discard_sandbox_lease_manager(old)
+
+
+def test_sandbox_shutdown_ownership_nests_and_is_released_on_error():
+    """Overlapping hosts keep ownership until the last one exits, even on failure."""
+    assert sandbox_provider.host_owns_sandbox_shutdown() is False
+    with sandbox_provider.hold_sandbox_shutdown_ownership():
+        with pytest.raises(RuntimeError, match="startup failed"):
+            with sandbox_provider.hold_sandbox_shutdown_ownership():
+                assert sandbox_provider.host_owns_sandbox_shutdown() is True
+                raise RuntimeError("startup failed")
+        assert sandbox_provider.host_owns_sandbox_shutdown() is True
+    assert sandbox_provider.host_owns_sandbox_shutdown() is False
diff --git a/backend/tests/test_scheduler_shutdown_timeout.py b/backend/tests/test_scheduler_shutdown_timeout.py
index d3d091b6b8..5f6f154389 100644
--- a/backend/tests/test_scheduler_shutdown_timeout.py
+++ b/backend/tests/test_scheduler_shutdown_timeout.py
@@ -33,5 +33,5 @@ async def test_scheduled_task_service_shutdown_is_bounded(monkeypatch: pytest.Mo
 
 
 def test_lifespan_uses_bounded_scheduled_task_shutdown() -> None:
-    source = inspect.getsource(gateway_app.lifespan)
+    source = inspect.getsource(gateway_app._gateway_lifespan)
     assert "await _shutdown_scheduled_task_service(app)" in source
diff --git a/backend/tests/test_subagent_batch_shutdown_timeout.py b/backend/tests/test_subagent_batch_shutdown_timeout.py
index 12aac488b0..0a691d88a7 100644
--- a/backend/tests/test_subagent_batch_shutdown_timeout.py
+++ b/backend/tests/test_subagent_batch_shutdown_timeout.py
@@ -110,7 +110,7 @@ async def test_gateway_shutdown_timeout_does_not_cancel_batch_cleanup(
 
 
 def test_lifespan_uses_bounded_subagent_batch_shutdown() -> None:
-    source = inspect.getsource(gateway_app.lifespan)
+    source = inspect.getsource(gateway_app._gateway_lifespan)
     assert "await _shutdown_subagent_batch_service(app)" in source