fix(sandbox): scope restored references to authenticated threads (#5736)

* fix(sandbox): scope restored references to authenticated threads

* docs(sandbox): trim inherited guidance

* fix(sandbox): reject restored references without thread scope

---------

Co-authored-by: YxinMiracle <“939157765@qq.com”>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -591,6 +591,12 @@ DeerFlow supports multiple sandbox execution modes:
 - **Docker Execution** (runs sandbox code in isolated Docker containers)
 - **Docker Execution with Kubernetes** (runs sandbox code in Kubernetes pods via provisioner service)
 
+Sandbox references in conversation state are server-owned. External run and
+thread-state APIs reject caller-supplied `sandbox` values; when restoring a
+checkpoint, the runtime resolves the reference against the authenticated user
+and thread before a tool can reuse it. A missing runtime thread ID raises an
+error even when the referenced sandbox is cached.
+
 When host Bash is enabled for Local Execution, DeerFlow starts OS detection with `uname -s`, then uses `sw_vers` on Darwin. On Linux, it reads host system files such as `/etc/os-release` only when the active sandbox policy permits it. Host filesystem path checks still apply; after a blocked path, the agent is directed to use a permitted command-only probe or virtual path instead of repeating the rejected command.
 
 For Docker development, service startup follows `config.yaml` sandbox mode. In Local/Docker modes, `provisioner` is not started.

backend/app/gateway/services.py

@@ -451,7 +451,8 @@ def _normalize_input_messages(
 def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, Any]:
     """Validate and sanitize caller-supplied state values before checkpointing.
 
-    The ``messages`` channel is canonicalized to a list of ``BaseMessage``
+    The server-owned ``sandbox`` channel is rejected. The ``messages`` channel
+    is canonicalized to a list of ``BaseMessage``
     objects, rejects external system/developer roles with HTTP 400, and strips
     server-owned metadata. Other channels keep their existing shapes while
     forged metadata and delegation verdicts are removed. ``normalize_input``
@@ -462,6 +463,12 @@ def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, An
     transform trails, or privileged message roles. Every channel is walked
     because middleware-contributed channels can also carry message-like values.
     """
+    if "sandbox" in values:
+        raise HTTPException(
+            status_code=400,
+            detail="External sandbox state is not allowed",
+        )
+
     stripped: dict[str, Any] = {}
     for channel, value in values.items():
         if channel == "messages" and value is not None:
@@ -491,6 +498,10 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool
     of bubbling up as a 500.  The gateway is a system boundary, so per-entry
     validation errors are the right shape for clients to retry against.
 
+    The ``sandbox`` channel is also server-owned. External callers cannot select
+    a provider resource by id; trusted internal run admission may carry the
+    server's own restored value.
+
     ``original_user_content``, dynamic-context reminder markers, the transient
     view-image context marker, the execution-only knowledge-scope marker, tool
     receipts, delegated receipt metadata/verdicts, and ``untrusted_input`` are
@@ -513,6 +524,11 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool
     """
     if raw_input is None:
         return {}
+    if not trusted_internal and "sandbox" in raw_input:
+        raise HTTPException(
+            status_code=400,
+            detail="External sandbox state is not allowed",
+        )
     result = raw_input
     messages = raw_input.get("messages")
     if messages is not None:

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

@@ -2398,6 +2398,25 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 self._last_activity[sandbox_id] = time.time()
         return sandbox
 
+    def get_scoped(
+        self,
+        sandbox_id: str,
+        *,
+        thread_id: str,
+        user_id: str,
+    ) -> Sandbox | None:
+        """Return a cached client only for its recorded user/thread identity."""
+        key = self._thread_key(thread_id, user_id)
+        with self._lock:
+            if self._thread_sandboxes.get(key) != sandbox_id:
+                return None
+            if self._active_sandbox_identity.get(sandbox_id) != key:
+                return None
+            sandbox = self._sandboxes.get(sandbox_id)
+            if sandbox is not None:
+                self._last_activity[sandbox_id] = time.time()
+            return sandbox
+
     def release(self, sandbox_id: str) -> None:
         """Release a sandbox from active use.
 

backend/packages/harness/deerflow/sandbox/AGENTS.md

@@ -1,5 +1,7 @@
 ### Sandbox System (`packages/harness/deerflow/sandbox/`)
 
+Sandbox restore requires a thread ID, even forks.
+
 **Network approval policy**: Sync/async `SandboxMiddleware` wrappers use
 `resolve_run_interaction_policy()` to gate lead-run network approval cards.
 Explicit `autonomous`, `webhook`, and `scheduled` modes auto-deny pending requests,
@@ -9,9 +11,9 @@ interactive runs; never consume events into a human-input card without a human
 to respond.
 
 **Interface**: `Sandbox`: `execute_command(command, env=None)`, additive `execute_command_in_scope(..., scope_id=...)` / `release_command_scope(scope_id)`, `read_file`, `write_file`, `list_dir`, `glob`, `grep`. Scoped hooks default to pass-through without server-side sessions, preserving third-party subclasses. `grep` accepts a text file or directory tree. Per-call `env` injects secrets: `LocalSandbox` merges into the subprocess environment; `AioSandbox` uses fresh `bash.exec(env=...)` sessions. `list_dir`: missing path → `FileNotFoundError`; command/client failure → `OSError`, never `[]` (`ls_tool`: `(empty)`). Remote `glob`/`grep` share it via `sandbox/remote_search.py`: missing root → `FileNotFoundError`, failed search → `OSError`; only a genuine no-match returns `[]`. The parser takes the command's `limit` and reports `truncated` when output passed it, which providers return after Python-side filtering; tools call an empty truncated result incomplete. Remote `grep(glob=...)` scopes like `glob()` (root-relative `path_matches`), never by basename alone. Remotes use `sandbox/remote_list_dir.py`: capture `find`'s status, not `| head`'s (`sh -lc` lacks `pipefail`); missing binary → `OSError`; truncation SIGPIPE → success.
-**Provider Pattern**: `SandboxProvider` exposes `acquire`, `acquire_async`, `get`, `release`. Async agent/tool paths use async hooks to keep Docker creation, discovery, cross-process locking, readiness polling, and release off-loop. Set `supports_agent_skill_isolation=True` only when the whole tool surface enforces explicit lead Agent policy: bind mounts use prepared thread roots; upload providers implement `sync_agent_skills`. Host-backed providers report false if an enabled shell bypasses path mappings. Under explicit policy, middleware rejects unsupported providers before acquire.
+**Provider Pattern**: `SandboxProvider`: `acquire`, `acquire_async`, `get`, `get_scoped`, `release`. `get_scoped` checks the identity-scoped cache without blocking; a miss uses canonical `acquire(user_id, thread_id)`, never a checkpoint id. AIO reads active identity maps. Async hooks keep Docker creation/discovery, cross-process locking, readiness checks, and release off-loop. Set `supports_agent_skill_isolation=True` only if every tool enforces explicit lead Agent policy: prepared thread-root bind mounts plus upload `sync_agent_skills`. Host-backed providers report false when enabled shells bypass path mappings; middleware rejects them before explicit-policy acquire.
 **Shared components** (RFC #4741): remote IDs use `derive_sandbox_scope_token` (`sandbox/identity.py`); preserve its keyword-only SHA-256/16-hex contract to avoid orphaning containers. `AcquireSerializer` (`sandbox/acquire_serialization.py`) serializes selected acquire/release transitions with a bounded, refcounted per-key `threading.Lock` table and dedicated bounded executor (no event-loop/default-executor blocking). Workers own cancellation cleanup without waiting for cancelled tasks to resume; provider `shutdown()`/`reset()` calls idempotent `close()`. Keys: AIO `(user_id, thread_id)`, E2B `(user_id, thread_id, skills_root)`, BoxLite/Tenki/OpenSandbox derived id. Random-UUID `thread_id=None` acquires bypass serialization.
-**Execution leases** (`sandbox/lease.py`, #5128): cross-instance ownership decides which Gateway may reap a container; process-local `SandboxLeaseManager` tracks concurrent lead, subagent, Gateway-request, and channel-upload users of one client. Runs get ephemeral owners, persisted sandboxes are retained idempotently, and the last holder performs any pending `SandboxProvider.release`. Outer lifecycle fences repeat idempotent release after their complete graph/tool/request batch drains; per-tool terminal `Command` wrappers never release because sibling handlers may still run. Fork-restored children and upload syncs use non-releasing holders: they fence the client and own scope cleanup without themselves requesting a park; an earlier normal-owner request waits for them, and a missing fork client is replaced by a normal owner. Persisted lookup plus retention is serialized per `(user_id, thread_id)`; stale bindings fall through to acquire, and a post-acquire lookup miss rolls back before raising. Provider I/O does not hold the metadata lock. Repeated cancellation cannot interrupt acquire/rollback/release reconciliation or let a `to_thread` sandbox operation outlive its enclosing execution/request holder; failures are logged without replacing the original cancellation. Lease/scope context IDs are server-owned: Gateway and worker scrub caller values; only the internal subagent path assigns a task ID. Managers are registered by provider object identity, not hash/equality, so unhashable custom providers remain valid. Subagent owners also serve as `sandbox_command_scope_id`: AIO gives each scope one ordered persistent shell session, replaces it after `ErrorObservation` or a missing-session 404, and cleans it on lease release. Registry identity is revalidated after every scope-lock wait, preventing queued commands from resurrecting released sessions. Env-bearing commands use fresh `bash.exec` sessions so secrets do not persist.
+**Execution leases** (`sandbox/lease.py`, #5128): cross-instance ownership selects which Gateway may reap a container; process-local `SandboxLeaseManager` tracks a client's concurrent lead, subagent, Gateway-request, and channel-upload users. Runs have ephemeral owners. Restore persisted sandboxes only after `get_scoped(user_id, thread_id)` succeeds; otherwise canonical `acquire` replaces the id. Last holder performs pending `SandboxProvider.release`. External run/state boundaries reject server-owned `sandbox`. Outer lifecycle fences repeat idempotent release after each graph/tool/request batch; per-tool terminal `Command` wrappers never release while siblings run. Fork-restored children and upload syncs hold without requesting release: server-created forks may borrow the parent client; ordinary checkpoint ids stay identity-scoped. Serialize persisted lookup and retention by `(user_id, thread_id)`; stale or mismatched bindings reacquire, and a post-acquire miss rolls back then raises. Provider I/O runs outside the metadata lock. Repeated cancellation cannot interrupt acquire/rollback/release reconciliation or let `to_thread` work outlive its execution/request holder; failures are logged without replacing the original cancellation. Lease/scope context IDs are server-owned: Gateway/worker scrub caller values; only internal subagents assign task IDs. Managers key by provider object identity (not hash/equality), so unhashable custom providers work. Subagent owners double as `sandbox_command_scope_id`: AIO gives each scope an ordered persistent shell, replaces it after `ErrorObservation` or a missing-session 404, and cleans it on lease release. Revalidate registry identity after each scope-lock wait so queued commands cannot resurrect released sessions. Env-bearing commands use fresh `bash.exec` sessions so secrets do not persist.
 **Authorization gate** (`sandbox:execute`, RFC #4063 Phase 3): every sandbox-backed tool call passes through the gate in `deerflow/authz/sandbox_authz.py` - a binary `authorize(principal, "sandbox", "execute", target="*")` check before either reusing a persisted sandbox id or calling `provider.acquire`. Rechecking reuse is required because authorization config and user roles can change while the sandbox remains cached. Sync tool invocations call `authorize_sandbox_execution`; async tool invocations await `authorize_sandbox_execution_async` exactly once. A task-local `ContextVar` scopes that single decision across the complete composed tool invocation, including `ReadBeforeWriteMiddleware`'s pre-write inspection, tool body, and post-read mark; the value is copied into `asyncio.to_thread` workers. Authorization denial is converted to the normal error `ToolMessage` at the composed middleware boundary and is explicitly excluded from the gate's generic fail-open handlers. Async config loading and provider class discovery/import are offloaded before `aauthorize()` so reused sandbox calls do not hash config files or import custom modules on the event loop; provider construction remains on the running event loop because async providers may initialize loop-affine clients. The gate lives at the single tool initialization entry point (`ensure_sandbox_initialized` / `ensure_sandbox_initialized_async` in `tools.py`), while `SandboxMiddleware.before_agent` / `abefore_agent` apply the matching sync/async check to eager acquisition. Deny raises `SandboxAuthorizationError` (`sandbox/exceptions.py`), which propagates out of ordinary tool execution as a friendly error `ToolMessage` ("sandbox execution is not permitted for your role") - the eager path catches it and skips acquisition instead, deferring the deny to the first sandbox-touching tool call so both paths share the same semantics. Provider errors (authorization calls and provider resolution) follow `authorization.fail_closed` / `fail_open`; no readable `config.yaml` or `authorization.enabled: false` makes the gate a no-op (`safe_app_config` tolerates missing config). Gateway upload/artifact sync calls `try_acquire_sandbox_for_request` (`app/gateway/authz.py`), which gates, returns a request lease, and skips sync on deny while preserving the primary operation. Callers release after their last sandbox operation; artifacts request normal parking, uploads do not. Tests: `tests/test_sandbox_authorization.py` and `tests/blocking_io/test_sandbox_authorization.py`.
 **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**:

backend/packages/harness/deerflow/sandbox/lease.py

@@ -436,12 +436,15 @@ class SandboxLeaseManager:
         user_id: str,
         release_on_last: bool = True,
         acquire_release_on_last: bool = True,
+        allow_unscoped_borrow: bool = False,
     ) -> str:
-        """Atomically retain a live persisted sandbox or acquire a replacement.
+        """Atomically restore a scoped sandbox or acquire a replacement.
 
-        A fork-restored live client is borrowed with ``release_on_last=False``;
-        if that persisted client is gone, its freshly acquired replacement is
-        owned normally unless ``acquire_release_on_last`` is also disabled.
+        Ordinary checkpoint ids must match ``user_id`` and ``thread_id``.
+        Server-created fork wrappers may set ``allow_unscoped_borrow`` with
+        ``release_on_last=False`` to share a parent's live client. If that
+        client is gone, the replacement is owned normally unless
+        ``acquire_release_on_last`` is also disabled.
         """
         key = self._thread_key(thread_id, user_id)
         with self._serializer.hold(key):
@@ -453,7 +456,16 @@ class SandboxLeaseManager:
             if existing_sandbox_id is not None:
                 return existing_sandbox_id
 
-            if self._provider.get(sandbox_id) is not None:
+            sandbox = (
+                self._provider.get(sandbox_id)
+                if allow_unscoped_borrow
+                else self._provider.get_scoped(
+                    sandbox_id,
+                    thread_id=thread_id,
+                    user_id=user_id,
+                )
+            )
+            if sandbox is not None:
                 with self._metadata_lock:
                     previous, release_previous = self._bind_locked(
                         owner_id,
@@ -508,6 +520,7 @@ class SandboxLeaseManager:
         user_id: str,
         release_on_last: bool = True,
         acquire_release_on_last: bool = True,
+        allow_unscoped_borrow: bool = False,
     ) -> str:
         """Async atomic retain-or-replace transition for a persisted sandbox."""
         key = self._thread_key(thread_id, user_id)
@@ -520,7 +533,16 @@ class SandboxLeaseManager:
             if existing_sandbox_id is not None:
                 return existing_sandbox_id
 
-            if self._provider.get(sandbox_id) is not None:
+            sandbox = (
+                self._provider.get(sandbox_id)
+                if allow_unscoped_borrow
+                else self._provider.get_scoped(
+                    sandbox_id,
+                    thread_id=thread_id,
+                    user_id=user_id,
+                )
+            )
+            if sandbox is not None:
                 with self._metadata_lock:
                     previous, release_previous = self._bind_locked(
                         owner_id,

backend/packages/harness/deerflow/sandbox/middleware.py

@@ -178,7 +178,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         sandbox_id = sandbox.get("sandbox_id")
         if isinstance(sandbox_id, str):
             provider = get_sandbox_provider()
-            get_sandbox_lease_manager(provider).retain(
+            sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire(
                 owner_id,
                 sandbox_id,
                 thread_id=thread_id,
@@ -203,7 +203,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         sandbox_id = sandbox.get("sandbox_id")
         if isinstance(sandbox_id, str):
             provider = get_sandbox_provider()
-            await get_sandbox_lease_manager(provider).retain_async(
+            sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async(
                 owner_id,
                 sandbox_id,
                 thread_id=thread_id,
@@ -306,8 +306,11 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
             user_id=user_id,
             owner_id=owner_id,
         )
-        if retained_id is not None and runtime.context is not None:
-            runtime.context["sandbox_id"] = retained_id
+        if retained_id is not None:
+            if runtime.context is not None:
+                runtime.context["sandbox_id"] = retained_id
+            if retained_id != existing_sandbox_id:
+                return {"sandbox": Overwrite({"sandbox_id": retained_id})}
         return super().before_agent(state, runtime)
 
     def _apply_network_policy_response(self, state: SandboxMiddlewareState, runtime: Runtime) -> None:
@@ -416,8 +419,11 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
             user_id=user_id,
             owner_id=owner_id,
         )
-        if retained_id is not None and runtime.context is not None:
-            runtime.context["sandbox_id"] = retained_id
+        if retained_id is not None:
+            if runtime.context is not None:
+                runtime.context["sandbox_id"] = retained_id
+            if retained_id != existing_sandbox_id:
+                return {"sandbox": Overwrite({"sandbox_id": retained_id})}
         return await super().abefore_agent(state, runtime)
 
     @override
@@ -508,7 +514,12 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         return sandbox_id if isinstance(sandbox_id, str) else None
 
     @staticmethod
-    def _attach_sandbox_update(result: ToolMessage | Command, sandbox_id: str) -> ToolMessage | Command:
+    def _attach_sandbox_update(
+        result: ToolMessage | Command,
+        sandbox_id: str,
+        *,
+        overwrite: bool = False,
+    ) -> ToolMessage | Command:
         """Wrap or merge ``result`` so that ``sandbox.sandbox_id`` is persisted.
 
         - ``ToolMessage`` -> ``Command(update={"sandbox": ..., "messages": [msg]})``
@@ -517,7 +528,10 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         - ``Command`` with non-dict / None update -> leave it untouched to
           avoid silent data loss on unknown update shapes.
         """
-        sandbox_update = {"sandbox": {"sandbox_id": sandbox_id}}
+        sandbox_value: object = {"sandbox_id": sandbox_id}
+        if overwrite:
+            sandbox_value = Overwrite(sandbox_value)
+        sandbox_update = {"sandbox": sandbox_value}
 
         if isinstance(result, ToolMessage):
             return Command(update={**sandbox_update, "messages": [result]})
@@ -545,8 +559,12 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         prev_sandbox_id = self._read_sandbox_id_from_request(request)
         result = handler(request)
         curr_sandbox_id = self._read_sandbox_id_from_request(request)
-        if prev_sandbox_id is None and curr_sandbox_id is not None:
-            result = self._attach_sandbox_update(result, curr_sandbox_id)
+        if curr_sandbox_id is not None and curr_sandbox_id != prev_sandbox_id:
+            result = self._attach_sandbox_update(
+                result,
+                curr_sandbox_id,
+                overwrite=prev_sandbox_id is not None,
+            )
         return self._maybe_request_network_approval(request, result, curr_sandbox_id or prev_sandbox_id)
 
     @override
@@ -558,8 +576,12 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         prev_sandbox_id = self._read_sandbox_id_from_request(request)
         result = await handler(request)
         curr_sandbox_id = self._read_sandbox_id_from_request(request)
-        if prev_sandbox_id is None and curr_sandbox_id is not None:
-            result = self._attach_sandbox_update(result, curr_sandbox_id)
+        if curr_sandbox_id is not None and curr_sandbox_id != prev_sandbox_id:
+            result = self._attach_sandbox_update(
+                result,
+                curr_sandbox_id,
+                overwrite=prev_sandbox_id is not None,
+            )
         sandbox_id = curr_sandbox_id or prev_sandbox_id
         if sandbox_id is None:
             return result
@@ -644,5 +666,6 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         )
         update: dict = {"messages": [message], "sandbox": {"sandbox_id": sandbox_id}}
         if isinstance(result, Command) and isinstance(result.update, dict):
-            update = {**result.update, **update}
+            update = {**result.update, "messages": [message]}
+            update.setdefault("sandbox", {"sandbox_id": sandbox_id})
         return Command(update=update, goto=END)

backend/packages/harness/deerflow/sandbox/sandbox_provider.py

@@ -81,6 +81,22 @@ class SandboxProvider(ABC):
         """
         pass
 
+    def get_scoped(
+        self,
+        sandbox_id: str,
+        *,
+        thread_id: str,
+        user_id: str,
+    ) -> Sandbox | None:
+        """Return an active sandbox only when it belongs to this identity.
+
+        This hook must remain a non-blocking in-memory lookup. Providers that
+        do not implement identity-aware lookup fail closed; the caller then
+        resolves the canonical sandbox through ``acquire``.
+        """
+        del sandbox_id, thread_id, user_id
+        return None
+
     @abstractmethod
     def release(self, sandbox_id: str) -> None:
         """Release a sandbox environment.

backend/packages/harness/deerflow/sandbox/tools.py

@@ -1545,22 +1545,44 @@ def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox:
             provider = get_sandbox_provider()
             owner_id = sandbox_lease_owner(runtime.context)
             thread_id = _resolve_runtime_thread_id(runtime)
-            if owner_id is not None and thread_id is not None:
-                sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire(
-                    owner_id,
-                    sandbox_id,
-                    thread_id=thread_id,
-                    user_id=resolve_runtime_user_id(runtime),
-                    release_on_last=not fork_restored,
-                )
+            user_id = resolve_runtime_user_id(runtime)
+            if thread_id is not None:
+                if owner_id is None:
+                    if not fork_restored:
+                        scoped = provider.get_scoped(
+                            sandbox_id,
+                            thread_id=thread_id,
+                            user_id=user_id,
+                        )
+                        if scoped is None:
+                            sandbox_id = provider.acquire(thread_id, user_id=user_id)
+                elif fork_restored:
+                    # Only the server-created fork wrapper may borrow a sandbox
+                    # from a different thread identity. Ordinary checkpoint ids
+                    # are resolved again from the authenticated user/thread.
+                    sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire(
+                        owner_id,
+                        sandbox_id,
+                        thread_id=thread_id,
+                        user_id=user_id,
+                        release_on_last=False,
+                        allow_unscoped_borrow=True,
+                    )
+                else:
+                    sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire(
+                        owner_id,
+                        sandbox_id,
+                        thread_id=thread_id,
+                        user_id=user_id,
+                    )
                 if not fork_restored:
                     runtime.state["sandbox"] = {"sandbox_id": sandbox_id}
-            sandbox = provider.get(sandbox_id)
-            if sandbox is not None:
-                if runtime.context is not None:
-                    runtime.context["sandbox_id"] = sandbox_id  # Ensure sandbox_id is in context for releasing in after_agent
-                return sandbox
-            # Sandbox was released, fall through to acquire new one
+                sandbox = provider.get(sandbox_id)
+                if sandbox is not None:
+                    if runtime.context is not None:
+                        runtime.context["sandbox_id"] = sandbox_id  # Ensure sandbox_id is in context for releasing in after_agent
+                    return sandbox
+            # Missing thread scope or released sandbox: use the lazy path below.
 
     # Lazy acquisition: get thread_id and acquire sandbox
     thread_id = _resolve_runtime_thread_id(runtime)
@@ -1623,21 +1645,40 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa
             provider = get_sandbox_provider()
             owner_id = sandbox_lease_owner(runtime.context)
             thread_id = _resolve_runtime_thread_id(runtime)
-            if owner_id is not None and thread_id is not None:
-                sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async(
-                    owner_id,
-                    sandbox_id,
-                    thread_id=thread_id,
-                    user_id=resolve_runtime_user_id(runtime),
-                    release_on_last=not fork_restored,
-                )
+            user_id = resolve_runtime_user_id(runtime)
+            if thread_id is not None:
+                if owner_id is None:
+                    if not fork_restored:
+                        scoped = provider.get_scoped(
+                            sandbox_id,
+                            thread_id=thread_id,
+                            user_id=user_id,
+                        )
+                        if scoped is None:
+                            sandbox_id = await provider.acquire_async(thread_id, user_id=user_id)
+                elif fork_restored:
+                    sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async(
+                        owner_id,
+                        sandbox_id,
+                        thread_id=thread_id,
+                        user_id=user_id,
+                        release_on_last=False,
+                        allow_unscoped_borrow=True,
+                    )
+                else:
+                    sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async(
+                        owner_id,
+                        sandbox_id,
+                        thread_id=thread_id,
+                        user_id=user_id,
+                    )
                 if not fork_restored:
                     runtime.state["sandbox"] = {"sandbox_id": sandbox_id}
-            sandbox = provider.get(sandbox_id)
-            if sandbox is not None:
-                if runtime.context is not None:
-                    runtime.context["sandbox_id"] = sandbox_id
-                return sandbox
+                sandbox = provider.get(sandbox_id)
+                if sandbox is not None:
+                    if runtime.context is not None:
+                        runtime.context["sandbox_id"] = sandbox_id
+                    return sandbox
 
     thread_id = _resolve_runtime_thread_id(runtime)
     if thread_id is None:

backend/tests/test_ensure_sandbox_initialized.py

@@ -80,6 +80,10 @@ class _RecordingProvider(SandboxProvider):
             return self.sandbox
         return None
 
+    def get_scoped(self, sandbox_id: str, *, thread_id: str, user_id: str) -> Sandbox | None:
+        del thread_id, user_id
+        return self.get(sandbox_id)
+
     def release(self, sandbox_id: str) -> None:
         self.released.append(sandbox_id)
 
@@ -193,6 +197,7 @@ def test_ensure_sandbox_initialized_unwraps_overwrite_state() -> None:
     set_sandbox_provider(provider)
     try:
         runtime = _make_runtime({"sandbox": Overwrite({"sandbox_id": "parent-sandbox"})})
+        runtime.context["thread_id"] = "thread-1"
         sandbox = ensure_sandbox_initialized(runtime)
     finally:
         reset_sandbox_provider()
@@ -210,6 +215,7 @@ async def test_ensure_sandbox_initialized_async_unwraps_overwrite_state() -> Non
     set_sandbox_provider(provider)
     try:
         runtime = _make_runtime({"sandbox": Overwrite({"sandbox_id": "parent-sandbox"})})
+        runtime.context["thread_id"] = "thread-1"
         sandbox = await ensure_sandbox_initialized_async(runtime)
     finally:
         reset_sandbox_provider()
@@ -280,6 +286,7 @@ def test_ensure_sandbox_initialized_plain_state_unchanged() -> None:
     set_sandbox_provider(provider)
     try:
         runtime = _make_runtime({"sandbox": {"sandbox_id": "parent-sandbox"}})
+        runtime.context["thread_id"] = "thread-1"
         sandbox = ensure_sandbox_initialized(runtime)
     finally:
         reset_sandbox_provider()
@@ -366,6 +373,7 @@ async def test_ensure_sandbox_initialized_async_plain_state_unchanged() -> None:
     set_sandbox_provider(provider)
     try:
         runtime = _make_runtime({"sandbox": {"sandbox_id": "parent-sandbox"}})
+        runtime.context["thread_id"] = "thread-1"
         sandbox = await ensure_sandbox_initialized_async(runtime)
     finally:
         reset_sandbox_provider()

backend/tests/test_sandbox_leases.py

@@ -71,6 +71,10 @@ class _LeaseProvider(SandboxProvider):
     def get(self, sandbox_id):
         return self.sandbox if sandbox_id == self.sandbox.id else None
 
+    def get_scoped(self, sandbox_id, *, thread_id, user_id):
+        del thread_id, user_id
+        return self.get(sandbox_id)
+
     def release(self, sandbox_id):
         self.release_calls.append(sandbox_id)
 
@@ -89,6 +93,10 @@ class _UnhashableLeaseProvider(SandboxProvider):
     def get(self, sandbox_id):
         return self.sandbox if sandbox_id == self.sandbox.id else None
 
+    def get_scoped(self, sandbox_id, *, thread_id, user_id):
+        del thread_id, user_id
+        return self.get(sandbox_id)
+
     def release(self, sandbox_id):
         self.release_calls.append(sandbox_id)
 

backend/tests/test_sandbox_middleware.py

@@ -158,6 +158,10 @@ class _AsyncOnlyProvider(SandboxProvider):
             return self.sandbox
         return None
 
+    def get_scoped(self, sandbox_id: str, *, thread_id: str, user_id: str) -> Sandbox | None:
+        del thread_id, user_id
+        return self.get(sandbox_id)
+
     def release(self, sandbox_id: str) -> None:
         self.released_ids.append(sandbox_id)
         return None
@@ -335,7 +339,7 @@ def test_explicit_skill_policy_does_not_reuse_checkpointed_sandbox_after_auth_de
     [
         (SandboxMiddleware(lazy_init=True), {}, Runtime(context={"thread_id": "thread-lazy"})),
         (SandboxMiddleware(lazy_init=False), {}, Runtime(context={})),
-        (SandboxMiddleware(lazy_init=False), {"sandbox": {"sandbox_id": "existing"}}, Runtime(context={"thread_id": "thread-existing"})),
+        (SandboxMiddleware(lazy_init=False), {"sandbox": {"sandbox_id": "async-sandbox"}}, Runtime(context={"thread_id": "thread-existing"})),
     ],
 )
 async def test_abefore_agent_delegates_to_super_when_not_acquiring(
@@ -599,6 +603,46 @@ def test_wrap_tool_call_passthrough_when_sandbox_already_in_state() -> None:
     assert result is original
 
 
+def test_wrap_tool_call_overwrites_a_repaired_checkpoint_sandbox() -> None:
+    middleware = SandboxMiddleware()
+    state: dict = {"sandbox": {"sandbox_id": "foreign"}}
+    request = _make_tool_call_request(state)
+
+    def handler(req: ToolCallRequest) -> ToolMessage:
+        req.runtime.state["sandbox"] = {"sandbox_id": "canonical"}
+        return ToolMessage(content="ok", tool_call_id="call-1", name="bash")
+
+    result = middleware.wrap_tool_call(request, handler)
+
+    assert isinstance(result, Command)
+    assert isinstance(result.update, dict)
+    assert isinstance(result.update["sandbox"], Overwrite)
+    assert result.update["sandbox"].value == {"sandbox_id": "canonical"}
+
+
+def test_network_prompt_preserves_repaired_checkpoint_overwrite() -> None:
+    provider = _NetworkPolicyProvider()
+    provider.events = [{"request_id": "req-1", "host": "example.com", "port": 443, "method": "CONNECT"}]
+    state: dict = {"sandbox": {"sandbox_id": "foreign"}}
+    request = _make_tool_call_request(state)
+
+    def handler(req: ToolCallRequest) -> ToolMessage:
+        req.runtime.state["sandbox"] = {"sandbox_id": "canonical"}
+        return ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash")
+
+    set_sandbox_provider(provider)
+    try:
+        result = SandboxMiddleware().wrap_tool_call(request, handler)
+    finally:
+        reset_sandbox_provider()
+
+    assert isinstance(result, Command)
+    assert result.goto == END
+    assert isinstance(result.update, dict)
+    assert isinstance(result.update["sandbox"], Overwrite)
+    assert result.update["sandbox"].value == {"sandbox_id": "canonical"}
+
+
 @pytest.mark.parametrize("async_path", [False, True])
 @pytest.mark.parametrize(
     "context",

backend/tests/test_sandbox_reference_ownership.py

@@ -0,0 +1,288 @@
+"""Sandbox references are server-owned and restored from authenticated scope."""
+
+from __future__ import annotations
+
+import threading
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from _router_auth_helpers import make_authed_test_app
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+from langgraph.types import Overwrite
+
+from app.gateway.routers import runs, thread_runs, threads
+from app.gateway.services import normalize_input, strip_server_owned_state_metadata
+from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider
+from deerflow.sandbox.exceptions import SandboxRuntimeError
+from deerflow.sandbox.lease import SANDBOX_LEASE_OWNER_CONTEXT_KEY
+from deerflow.sandbox.sandbox import Sandbox
+from deerflow.sandbox.sandbox_provider import SandboxProvider, reset_sandbox_provider, set_sandbox_provider
+from deerflow.sandbox.search import GrepMatch
+from deerflow.sandbox.tools import ensure_sandbox_initialized, ensure_sandbox_initialized_async
+
+FOREIGN_SANDBOX_ID = "sandbox-user-b-thread-b"
+OWN_SANDBOX_ID = "sandbox-user-a-thread-a"
+
+
+class _ScopedSandbox(Sandbox):
+    def execute_command(self, command, env=None, timeout=None):
+        return command
+
+    def read_file(self, path, start_line=None, end_line=None):
+        return self.id
+
+    def download_file(self, path):
+        return self.id.encode()
+
+    def list_dir(self, path, max_depth=2):
+        return []
+
+    def write_file(self, path, content, append=False):
+        return None
+
+    def glob(self, path, pattern, *, include_dirs=False, max_results=200):
+        return [], False
+
+    def grep(
+        self,
+        path: str,
+        pattern: str,
+        *,
+        glob: str | None = None,
+        literal: bool = False,
+        case_sensitive: bool = False,
+        max_results: int = 100,
+    ) -> tuple[list[GrepMatch], bool]:
+        return [], False
+
+    def update_file(self, path, content):
+        return None
+
+
+class _IdentityScopedProvider(SandboxProvider):
+    """Models an active foreign sandbox plus canonical current-scope lookup."""
+
+    def __init__(self) -> None:
+        self.sandboxes = {
+            FOREIGN_SANDBOX_ID: _ScopedSandbox(FOREIGN_SANDBOX_ID),
+            OWN_SANDBOX_ID: _ScopedSandbox(OWN_SANDBOX_ID),
+        }
+        self.acquire_calls: list[tuple[str | None, str | None]] = []
+        self.get_calls: list[str] = []
+        self.scoped_get_calls: list[tuple[str, str, str]] = []
+
+    def acquire(self, thread_id=None, *, user_id=None):
+        self.acquire_calls.append((thread_id, user_id))
+        assert (user_id, thread_id) == ("user-a", "thread-a")
+        return OWN_SANDBOX_ID
+
+    async def acquire_async(self, thread_id=None, *, user_id=None):
+        return self.acquire(thread_id, user_id=user_id)
+
+    def get(self, sandbox_id):
+        self.get_calls.append(sandbox_id)
+        return self.sandboxes.get(sandbox_id)
+
+    def get_scoped(self, sandbox_id, *, thread_id, user_id):
+        self.scoped_get_calls.append((sandbox_id, thread_id, user_id))
+        if (sandbox_id, user_id, thread_id) == (OWN_SANDBOX_ID, "user-a", "thread-a"):
+            return self.sandboxes[sandbox_id]
+        return None
+
+    def release(self, sandbox_id):
+        return None
+
+
+def _runtime_with_foreign_checkpoint():
+    return SimpleNamespace(
+        state={"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}},
+        context={
+            SANDBOX_LEASE_OWNER_CONTEXT_KEY: "run-owner-a",
+            "thread_id": "thread-a",
+            "user_id": "user-a",
+        },
+        config={"configurable": {"thread_id": "thread-a"}},
+    )
+
+
+@pytest.mark.parametrize(
+    ("boundary", "payload"),
+    [
+        (normalize_input, {"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}),
+        (strip_server_owned_state_metadata, {"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}),
+    ],
+)
+def test_external_sandbox_state_is_rejected_at_gateway_boundaries(boundary, payload):
+    with pytest.raises(HTTPException) as error:
+        boundary(payload)
+
+    assert error.value.status_code == 400
+    assert FOREIGN_SANDBOX_ID not in str(error.value.detail)
+
+
+def test_trusted_internal_run_input_can_restore_server_owned_sandbox_state():
+    payload = {"sandbox": {"sandbox_id": OWN_SANDBOX_ID}}
+    assert normalize_input(payload, trusted_internal=True)["sandbox"] == payload["sandbox"]
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/threads/sandbox-owner-http/runs",
+        "/api/threads/sandbox-owner-http/runs/stream",
+        "/api/threads/sandbox-owner-http/runs/wait",
+        "/api/runs/stream",
+        "/api/runs/wait",
+    ],
+)
+def test_all_external_run_entrypoints_reject_sandbox_before_worker(monkeypatch, path):
+    from app.gateway import services
+
+    app = make_authed_test_app()
+    app.include_router(runs.router)
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = SimpleNamespace()
+    app.state.run_manager = SimpleNamespace(create_or_reject=AsyncMock())
+    monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store))
+    monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant: object())
+    worker = AsyncMock()
+    monkeypatch.setattr(services, "run_agent", worker)
+
+    with TestClient(app) as client:
+        response = client.post(
+            path,
+            json={
+                "input": {
+                    "messages": [{"role": "user", "content": "synthetic"}],
+                    "sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID},
+                }
+            },
+        )
+
+    assert response.status_code == 400, response.text
+    assert FOREIGN_SANDBOX_ID not in response.text
+    app.state.run_manager.create_or_reject.assert_not_awaited()
+    worker.assert_not_awaited()
+
+
+def test_external_state_update_rejects_sandbox_before_checkpoint_access():
+    app = make_authed_test_app()
+    app.include_router(threads.router)
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/threads/sandbox-owner-http/state",
+            json={"values": {"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}},
+        )
+
+    assert response.status_code == 400, response.text
+    assert FOREIGN_SANDBOX_ID not in response.text
+
+
+def test_checkpoint_sandbox_is_resolved_against_current_identity_before_sync_use():
+    provider = _IdentityScopedProvider()
+    set_sandbox_provider(provider)
+    runtime = _runtime_with_foreign_checkpoint()
+    try:
+        sandbox = ensure_sandbox_initialized(runtime)
+    finally:
+        reset_sandbox_provider()
+
+    assert sandbox.id == OWN_SANDBOX_ID
+    assert runtime.state["sandbox"] == {"sandbox_id": OWN_SANDBOX_ID}
+    assert provider.acquire_calls == [("thread-a", "user-a")]
+    assert provider.scoped_get_calls == [(FOREIGN_SANDBOX_ID, "thread-a", "user-a")]
+    assert FOREIGN_SANDBOX_ID not in provider.get_calls
+
+
+@pytest.mark.asyncio
+async def test_checkpoint_sandbox_is_resolved_against_current_identity_before_async_use():
+    provider = _IdentityScopedProvider()
+    set_sandbox_provider(provider)
+    runtime = _runtime_with_foreign_checkpoint()
+    try:
+        sandbox = await ensure_sandbox_initialized_async(runtime)
+    finally:
+        reset_sandbox_provider()
+
+    assert sandbox.id == OWN_SANDBOX_ID
+    assert runtime.state["sandbox"] == {"sandbox_id": OWN_SANDBOX_ID}
+    assert provider.acquire_calls == [("thread-a", "user-a")]
+    assert provider.scoped_get_calls == [(FOREIGN_SANDBOX_ID, "thread-a", "user-a")]
+    assert FOREIGN_SANDBOX_ID not in provider.get_calls
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("async_path", [False, True], ids=["sync", "async"])
+@pytest.mark.parametrize("fork_restored", [False, True], ids=["checkpoint", "fork"])
+@pytest.mark.parametrize("with_lease_owner", [False, True], ids=["unleased", "leased"])
+async def test_checkpoint_sandbox_without_thread_id_fails_closed(async_path, fork_restored, with_lease_owner):
+    provider = _IdentityScopedProvider()
+    set_sandbox_provider(provider)
+    runtime = _runtime_with_foreign_checkpoint()
+    runtime.context.pop("thread_id")
+    runtime.config = {}
+    if not with_lease_owner:
+        runtime.context.pop(SANDBOX_LEASE_OWNER_CONTEXT_KEY)
+    if fork_restored:
+        runtime.state["sandbox"] = Overwrite(runtime.state["sandbox"])
+    original_state = runtime.state["sandbox"]
+    try:
+        with pytest.raises(SandboxRuntimeError, match="Thread ID not available"):
+            if async_path:
+                await ensure_sandbox_initialized_async(runtime)
+            else:
+                ensure_sandbox_initialized(runtime)
+    finally:
+        reset_sandbox_provider()
+
+    assert provider.get_calls == []
+    assert provider.scoped_get_calls == []
+    assert provider.acquire_calls == []
+    assert runtime.state["sandbox"] is original_state
+    assert "sandbox_id" not in runtime.context
+
+
+def test_matching_checkpoint_sandbox_is_reused_without_acquire():
+    provider = _IdentityScopedProvider()
+    set_sandbox_provider(provider)
+    runtime = _runtime_with_foreign_checkpoint()
+    runtime.state["sandbox"] = {"sandbox_id": OWN_SANDBOX_ID}
+    try:
+        sandbox = ensure_sandbox_initialized(runtime)
+    finally:
+        reset_sandbox_provider()
+
+    assert sandbox.id == OWN_SANDBOX_ID
+    assert provider.acquire_calls == []
+    assert provider.scoped_get_calls == [(OWN_SANDBOX_ID, "thread-a", "user-a")]
+
+
+def test_aio_cached_lookup_requires_matching_user_and_thread_identity():
+    provider = object.__new__(AioSandboxProvider)
+    provider._lock = threading.Lock()
+    foreign = _ScopedSandbox(FOREIGN_SANDBOX_ID)
+    provider._sandboxes = {FOREIGN_SANDBOX_ID: foreign}
+    provider._thread_sandboxes = {("user-b", "thread-b"): FOREIGN_SANDBOX_ID}
+    provider._active_sandbox_identity = {FOREIGN_SANDBOX_ID: ("user-b", "thread-b")}
+    provider._last_activity = {}
+
+    assert (
+        provider.get_scoped(
+            FOREIGN_SANDBOX_ID,
+            thread_id="thread-a",
+            user_id="user-a",
+        )
+        is None
+    )
+    assert (
+        provider.get_scoped(
+            FOREIGN_SANDBOX_ID,
+            thread_id="thread-b",
+            user_id="user-b",
+        )
+        is foreign
+    )
+    assert FOREIGN_SANDBOX_ID in provider._last_activity