fix(sandbox): defer fresh AIO skill-policy acquisition (#7782)

* fix(sandbox): defer fresh AIO skill-policy acquisition

Prepare the authorized thread skill view before the model runs without
acquiring an AIO container for a fresh lazy run. Keep existing sandbox
bindings and upload-based providers on the eager rebind/sync path.

Add sync/async regressions for first-tool authorization, strict rebinding,
filtered read-only AIO mounts, and policy-scoped warm reuse.

AI assistance: Codex; the account owner confirmed human review of the
complete seven-file patch and responsibility for the change.

* docs(sandbox): keep lazy-isolation guidance within inherited budget

---------

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

CHANGELOG.md

@@ -24,6 +24,11 @@ This section accumulates work toward the **2.3.0** milestone
 
 ### Fixed
 
+- **sandbox:** Fresh lazy AIO runs with an explicit authorized skill view no
+  longer create or acquire a sandbox before the first sandbox-backed tool
+  call. Existing sandbox bindings and providers that require skill uploads
+  retain their eager synchronization behavior. ([#7782])
+  
 - **gateway:** A caller-supplied thread ID that is a Windows reserved device
   name (`NUL`, `CON`, `COM1`, ...) is now rejected at the shared
   `validate_thread_id()` boundary. Such IDs previously passed validation but
@@ -16026,4 +16031,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7781]: https://github.com/bytedance/deer-flow/pull/7781
 [#7792]: https://github.com/bytedance/deer-flow/pull/7792
 [#7831]: https://github.com/bytedance/deer-flow/pull/7831
+[#7782]: https://github.com/bytedance/deer-flow/pull/7782
 [#7892]: https://github.com/bytedance/deer-flow/pull/7892

CHANGELOG_zh.md

@@ -23,6 +23,10 @@
 
 ### 修复
 
+- **沙箱：** 使用显式授权技能视图的新 AIO 运行，在启用惰性初始化时，不再于首次调用
+  沙箱工具前创建或获取沙箱。已有沙箱绑定和需要上传技能的提供方继续保留原有的即时同步行为。
+  ([#7782])
+
 - **扩展：** 上下文压缩观察者现在能知道压缩发生在哪个任务里。在 Lead 运行或子 Agent
   内部，观察者收到的是该任务的存储，而不再是分离存储；`CompactionEvent` 新增
   `task_id`、`run_id`、`thread_id`（默认 `None`），取自宿主放进每个任务存储的
@@ -12758,3 +12762,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7781]: https://github.com/bytedance/deer-flow/pull/7781
 [#7792]: https://github.com/bytedance/deer-flow/pull/7792
 [#7831]: https://github.com/bytedance/deer-flow/pull/7831
+[#7782]: https://github.com/bytedance/deer-flow/pull/7782

backend/README.md

@@ -74,6 +74,12 @@ receive a transient hint on the next model request.
 
 ### Sandbox System
 
+With lazy initialization, a fresh AIO run prepares its authorized Agent skill
+view without acquiring a container until a sandbox-backed tool needs one.
+Existing sandbox state still requires eager policy rebinding; upload-based
+providers still acquire and synchronize skills eagerly. Explicit sandbox
+prewarming remains a separate opt-in behavior.
+
 Per-thread isolated execution with virtual path translation:
 
 - **Abstract interface**: `execute_command`, `read_file`, `write_file`, `list_dir`

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

@@ -237,6 +237,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
     """
 
     supports_agent_skill_isolation = True
+    supports_lazy_agent_skill_isolation = True
 
     # How long `_held_teardown_lease` waits for its heartbeat thread to exit
     # before deferring the final lease release to that (still-running) thread.

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

@@ -1,12 +1,13 @@
-### Sandbox System (`packages/harness/deerflow/sandbox/`)
+### Sandbox System
 
-Sandbox restore requires a thread ID, even forks.
+Lazy projections: [contract](../../../../README.md#sandbox-system).
+Restore needs a thread ID, even forks.
 
-Signal managed bash before draining cancellation. Stop only the lease owner’s
-processes before outer release; normal completion keeps previews. Replacement
-resets its command generation. Regression: `test_sandbox_command_cancellation.py`.
+Signal managed bash before cancellation drain; stop only the lease owner's
+processes before outer release. Keep previews on normal completion; replacement
+resets command generation. Test: `test_sandbox_command_cancellation.py`.
 
-On startup failure, close reads and kill/reap the child, preserving the error.
+On startup failure, close reads, kill/reap child; preserve the error.
 Only `finally` closes write ends.
 
 On Linux/macOS, LocalSandbox non-append text writes use `_publish_file_revision` and the shared atomic byte publisher. Check write access without truncation; retain the source descriptor through metadata transfer. Preserve symlink mapping; publish only regular files. Pre-replace failures retain the old file and clean staging. Append/other platforms write in place. Replacement needs parent-directory write and metadata permissions; hard-link aliases retain the old inode. No concurrent-writer arbitration or power-loss recovery. Tests: `tests/test_local_sandbox_atomic_writes.py`.

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

@@ -128,6 +128,21 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         if projection is not None and not provider.supports_agent_skill_isolation:
             raise SandboxRuntimeError(f"Sandbox provider {provider.__class__.__name__} cannot enforce per-Agent skill filesystem isolation")
 
+    def _can_defer_sandbox_acquisition(self, state: SandboxMiddlewareState, runtime: Runtime, projection) -> bool:
+        if not self._lazy_init:
+            return False
+        if projection is None:
+            return True
+        # Existing or fork-restored state must still be rebound before tools
+        # can reuse a sandbox that had a different Agent skill policy.
+        if state.get("sandbox") is not None or (runtime.context or {}).get("sandbox_id") is not None:
+            return False
+        provider = get_sandbox_provider()
+        self._require_projection_support(provider, projection)
+        # AIO mounts the stable projection roots during canonical acquisition.
+        # Upload-based providers still need eager acquire + strict skill sync.
+        return provider.supports_lazy_agent_skill_isolation
+
     def _acquire_sandbox(
         self,
         thread_id: str,
@@ -238,10 +253,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         projection = self._prepare_agent_skill_projection(thread_id, user_id=user_id)
         owner_id = ensure_sandbox_lease_owner(runtime.context)
 
-        # Preserve lazy initialization for threads that use the shared view.
-        # A policy-scoped view is acquired eagerly so an old shared-view
-        # sandbox cannot survive into this run through checkpoint state.
-        if self._lazy_init and projection is None:
+        if self._can_defer_sandbox_acquisition(state, runtime, projection):
             # Bind the execution lease only when a sandbox-backed tool actually
             # touches the persisted sandbox. Runs that only answer or return a
             # terminal Command must not leave an unused owner behind when the
@@ -365,7 +377,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
         )
         owner_id = ensure_sandbox_lease_owner(runtime.context)
 
-        if self._lazy_init and projection is None:
+        if self._can_defer_sandbox_acquisition(state, runtime, projection):
             return await super().abefore_agent(state, runtime)
 
         existing_sandbox_id = self._read_sandbox_id_from_state(state)

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

@@ -23,6 +23,9 @@ class SandboxProvider(ABC):
     # provider's current Agent-accessible tool surface. Host-backed providers
     # must return False whenever shell access can bypass managed path mappings.
     supports_agent_skill_isolation: bool = False
+    # Opt in only when acquire/reuse itself exposes the prepared thread view,
+    # without a separate sync_agent_skills upload before tools can access it.
+    supports_lazy_agent_skill_isolation: bool = False
 
     @abstractmethod
     def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:

backend/tests/test_lazy_aio_skill_projection.py

@@ -0,0 +1,192 @@
+"""Offline integration of lazy Agent skill views and AIO acquisition."""
+
+from __future__ import annotations
+
+import atexit
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from langchain.tools import ToolRuntime
+from langchain_core.messages import HumanMessage
+from langgraph.runtime import Runtime
+
+from deerflow.community.aio_sandbox import aio_sandbox_provider as aio_mod
+from deerflow.community.aio_sandbox.local_backend import LocalContainerBackend
+from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
+from deerflow.config.extensions_config import ExtensionsConfig
+from deerflow.config.paths import Paths
+from deerflow.config.sandbox_config import SandboxConfig
+from deerflow.config.skills_config import SkillsConfig
+from deerflow.sandbox.middleware import SandboxMiddleware
+from deerflow.sandbox.sandbox_provider import reset_sandbox_provider, set_sandbox_provider
+from deerflow.sandbox.tools import ensure_sandbox_initialized, ensure_sandbox_initialized_async
+
+
+class _RecordingContainerBackend(LocalContainerBackend):
+    """Replace container I/O, retaining the provider's real mount/identity code."""
+
+    def __init__(self) -> None:
+        self.created: list[tuple[str | None, str, dict]] = []
+        self.discovered: list[str] = []
+        self.running: dict[str, SandboxInfo] = {}
+
+    def discover(self, sandbox_id: str) -> SandboxInfo | None:
+        self.discovered.append(sandbox_id)
+        return self.running.get(sandbox_id)
+
+    def create(self, thread_id: str | None, sandbox_id: str, **kwargs) -> SandboxInfo:
+        self.created.append((thread_id, sandbox_id, kwargs))
+        info = SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox.invalid")
+        self.running[sandbox_id] = info
+        return info
+
+    def is_alive(self, info: SandboxInfo) -> bool:
+        return info.sandbox_id in self.running
+
+    def destroy(self, info: SandboxInfo) -> None:
+        self.running.pop(info.sandbox_id, None)
+
+    def is_absent(self, sandbox_id: str) -> bool:
+        return sandbox_id not in self.running
+
+    def complete_absent_teardown(self, sandbox_id: str) -> None:
+        assert self.is_absent(sandbox_id)
+
+
+def _write_skill(root: Path, name: str) -> None:
+    target = root / name / "SKILL.md"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(f"---\nname: {name}\ndescription: Test skill\n---\n\n# {name}\n", encoding="utf-8")
+
+
+@pytest.fixture
+def lazy_aio_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    paths = Paths(base_dir=tmp_path / "state")
+    skills_root = tmp_path / "skills"
+    config = SimpleNamespace(
+        sandbox=SandboxConfig(use="deerflow.community.aio_sandbox:AioSandboxProvider", idle_timeout=0),
+        skills=SkillsConfig(path=str(skills_root)),
+        stream_bridge=None,
+    )
+    extensions = ExtensionsConfig()
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
+    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: paths)
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: paths)
+    monkeypatch.setattr(aio_mod, "get_app_config", lambda: config)
+    monkeypatch.setattr("deerflow.config.extensions_config.ExtensionsConfig.from_file", lambda *a, **kw: extensions)
+    monkeypatch.setattr("deerflow.config.extensions_config.get_extensions_config", lambda: extensions)
+
+    for name in ("public-allowed", "public-denied"):
+        _write_skill(skills_root / "public", name)
+    for name in ("custom-allowed", "custom-denied"):
+        _write_skill(paths.user_custom_skills_dir("alice"), name)
+    for name in ("integration-allowed", "integration-denied"):
+        _write_skill(paths.integration_skills_dir() / "test-provider", name)
+    _write_skill(skills_root / "custom", "legacy-denied")
+
+    backend = _RecordingContainerBackend()
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_create_backend", lambda self: backend)
+    # No daemon, signal handler, or startup orphan probe is needed for the
+    # deterministic single-provider lifecycle under test.
+    for method in ("_register_signal_handlers", "_reconcile_orphans", "_start_lease_renewal"):
+        monkeypatch.setattr(aio_mod.AioSandboxProvider, method, lambda self: None)
+    monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda *a, **kw: True)
+
+    async def ready(*args, **kwargs):
+        return True
+
+    monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready_async", ready)
+    provider = aio_mod.AioSandboxProvider()
+    set_sandbox_provider(provider)
+    try:
+        yield SimpleNamespace(paths=paths, provider=provider, backend=backend)
+    finally:
+        try:
+            provider.shutdown()
+        finally:
+            atexit.unregister(provider.shutdown)
+            reset_sandbox_provider()
+
+
+def _tool_runtime(state: dict, context: dict) -> ToolRuntime:
+    return ToolRuntime(
+        state=state,
+        context=context,
+        config={"configurable": {}},
+        stream_writer=lambda _: None,
+        tools=[],
+        tool_call_id="first-sandbox-tool",
+        store=None,
+    )
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("async_init", [False, True], ids=["sync", "async"])
+async def test_lazy_aio_projects_before_first_tool_and_reuses_filtered_mounts(lazy_aio_env, async_init):
+    env = lazy_aio_env
+    thread_id = "thread-policy"
+    shared_id = env.provider._sandbox_id_for_thread(thread_id, "alice")
+    allowed = {"public-allowed", "custom-allowed", "integration-allowed"}
+    middleware = SandboxMiddleware(lazy_init=True, available_skills=allowed)
+    context = {"thread_id": thread_id, "user_id": "alice"}
+    state = {"messages": [HumanMessage(content="Hello, how are you?")]}
+    runtime = Runtime(context=context)
+
+    if async_init:
+        update = await middleware.abefore_agent(state, runtime)
+    else:
+        update = middleware.before_agent(state, runtime)
+
+    projection = env.paths.thread_skills_view_dir(thread_id, user_id="alice")
+    assert update is None
+    assert "sandbox" not in state
+    assert "sandbox_id" not in context
+    assert env.backend.created == []
+    assert env.backend.discovered == []
+    assert (projection / "public/public-allowed/SKILL.md").is_file()
+    assert (projection / "custom/custom-allowed/SKILL.md").is_file()
+    assert (projection / "integrations/test-provider/integration-allowed/SKILL.md").is_file()
+    assert not any(projection.rglob("*-denied"))
+
+    tool_runtime = _tool_runtime(state, context)
+    if async_init:
+        sandbox = await ensure_sandbox_initialized_async(tool_runtime)
+    else:
+        sandbox = ensure_sandbox_initialized(tool_runtime)
+
+    policy_id = env.provider._sandbox_id_for_thread(thread_id, "alice")
+    assert sandbox.id == policy_id != shared_id
+    assert env.backend.discovered == [policy_id]
+    assert len(env.backend.created) == 1
+    created_thread, created_id, kwargs = env.backend.created[0]
+    assert (created_thread, created_id) == (thread_id, policy_id)
+    skill_mounts = {destination: (Path(source), read_only) for source, destination, read_only in kwargs["extra_mounts"] if destination.startswith("/mnt/skills/")}
+    assert skill_mounts == {f"/mnt/skills/{category}": (projection / category, True) for category in ("public", "custom", "legacy", "integrations")}
+    assert kwargs["config_mount_exclusion_root"] == "/mnt/skills"
+    assert state["sandbox"] == {"sandbox_id": policy_id}
+    assert context["sandbox_id"] == policy_id
+
+    # A later run can narrow the stable mounted view before claiming this
+    # thread's warm sandbox. It must not fall back to shared-view discovery.
+    if async_init:
+        await middleware.aafter_agent(state, runtime)
+    else:
+        middleware.after_agent(state, runtime)
+    assert policy_id in env.provider._warm_pool
+    narrowed = SandboxMiddleware(lazy_init=True, available_skills={"public-allowed"})
+    next_context = {"thread_id": thread_id, "user_id": "alice"}
+    next_state = {"messages": [HumanMessage(content="Hello again")]}
+    next_runtime = Runtime(context=next_context)
+    if async_init:
+        assert await narrowed.abefore_agent(next_state, next_runtime) is None
+        reused = await ensure_sandbox_initialized_async(_tool_runtime(next_state, next_context))
+    else:
+        assert narrowed.before_agent(next_state, next_runtime) is None
+        reused = ensure_sandbox_initialized(_tool_runtime(next_state, next_context))
+    assert reused.id == policy_id
+    assert len(env.backend.created) == 1
+    assert env.backend.discovered == [policy_id]
+    assert (skill_mounts["/mnt/skills/public"][0] / "public-allowed/SKILL.md").is_file()
+    assert not (skill_mounts["/mnt/skills/custom"][0] / "custom-allowed").exists()
+    assert not (skill_mounts["/mnt/skills/integrations"][0] / "test-provider/integration-allowed").exists()

backend/tests/test_sandbox_middleware.py

@@ -265,10 +265,13 @@ def test_explicit_skill_policy_eagerly_acquires_and_syncs_existing_thread(
     assert provider.skill_syncs == [("sync-sandbox", "thread-policy", "owner-policy", projection)]
 
 
+@pytest.mark.parametrize("lazy_projection", [False, True])
 def test_explicit_skill_policy_fails_closed_for_unsupported_provider(
     monkeypatch: pytest.MonkeyPatch,
+    lazy_projection: bool,
 ) -> None:
     provider = _SyncProvider()
+    provider.supports_lazy_agent_skill_isolation = lazy_projection
     middleware = SandboxMiddleware(lazy_init=True, available_skills={"allowed"})
     monkeypatch.setattr(
         middleware,
@@ -291,6 +294,147 @@ def test_explicit_skill_policy_fails_closed_for_unsupported_provider(
     assert provider.thread_ids == []
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("async_hook", [False, True])
+async def test_fresh_skill_projection_defers_acquire_until_first_tool(monkeypatch, async_hook):
+    provider = _AgentSkillSyncProvider()
+    # Bind-mounted providers install the prepared projection during acquire.
+    provider.supports_lazy_agent_skill_isolation = True
+    projection = object()
+    middleware = SandboxMiddleware(available_skills={"allowed"})
+    prepared = []
+
+    def prepare(thread_id, *, user_id):
+        prepared.append((thread_id, user_id))
+        return projection
+
+    monkeypatch.setattr(middleware, "_prepare_agent_skill_projection", prepare)
+    context = {"thread_id": "thread-lazy-policy", "user_id": "owner-policy"}
+    set_sandbox_provider(provider)
+    try:
+        runtime = Runtime(context=context)
+        result = await middleware.abefore_agent({}, runtime) if async_hook else middleware.before_agent({}, runtime)
+        assert result is None
+        assert prepared == [("thread-lazy-policy", "owner-policy")]
+        assert provider.thread_ids == []
+        assert provider.skill_syncs == []
+        assert "sandbox_id" not in context
+
+        tool_runtime = ToolRuntime(state={}, context=context, config={}, stream_writer=lambda _: None, tools=[], tool_call_id="tool-1", store=None)
+        if async_hook:
+            sandbox = await ensure_sandbox_initialized_async(tool_runtime)
+        else:
+            sandbox = ensure_sandbox_initialized(tool_runtime)
+        assert sandbox is provider.sandbox
+        assert provider.thread_ids == ["thread-lazy-policy"]
+        assert provider.user_ids == ["owner-policy"]
+    finally:
+        reset_sandbox_provider()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("async_hook", [False, True])
+async def test_deferred_skill_projection_still_authorizes_first_tool(monkeypatch, async_hook):
+    provider = _AgentSkillSyncProvider()
+    provider.supports_lazy_agent_skill_isolation = True
+    middleware = SandboxMiddleware(available_skills=set())
+    monkeypatch.setattr(middleware, "_prepare_agent_skill_projection", lambda *_args, **_kwargs: object())
+
+    def deny(**_kwargs):
+        raise SandboxAuthorizationError("denied at first tool")
+
+    async def deny_async(**_kwargs):
+        deny()
+
+    monkeypatch.setattr("deerflow.sandbox.tools.authorize_sandbox_execution", deny)
+    monkeypatch.setattr("deerflow.sandbox.tools.authorize_sandbox_execution_async", deny_async)
+    context = {"thread_id": "thread-lazy-denied", "user_id": "owner-policy"}
+    set_sandbox_provider(provider)
+    try:
+        runtime = Runtime(context=context)
+        result = await middleware.abefore_agent({}, runtime) if async_hook else middleware.before_agent({}, runtime)
+        assert result is None
+        tool_runtime = ToolRuntime(state={}, context=context, config={}, stream_writer=lambda _: None, tools=[], tool_call_id="tool-1", store=None)
+        with pytest.raises(SandboxAuthorizationError, match="denied at first tool"):
+            if async_hook:
+                await ensure_sandbox_initialized_async(tool_runtime)
+            else:
+                ensure_sandbox_initialized(tool_runtime)
+        assert provider.thread_ids == []
+    finally:
+        reset_sandbox_provider()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("async_hook", [False, True])
+@pytest.mark.parametrize("existing", ["checkpoint", "fork", "context", "empty-state", "eager", "upload"])
+async def test_skill_projection_keeps_eager_rebind_and_upload_sync(monkeypatch, async_hook, existing):
+    provider = _AgentSkillSyncProvider()
+    if existing != "upload":
+        provider.supports_lazy_agent_skill_isolation = True
+    projection = object()
+    middleware = SandboxMiddleware(lazy_init=existing != "eager", available_skills=set())
+    monkeypatch.setattr(middleware, "_prepare_agent_skill_projection", lambda *_args, **_kwargs: projection)
+    context = {"thread_id": "thread-policy", "user_id": "owner-policy"}
+    state = {}
+    if existing == "checkpoint":
+        state["sandbox"] = {"sandbox_id": "shared-view-sandbox"}
+    elif existing == "fork":
+        state["sandbox"] = Overwrite({"sandbox_id": "shared-view-sandbox"})
+    elif existing == "context":
+        context["sandbox_id"] = "shared-view-sandbox"
+    elif existing == "empty-state":
+        state["sandbox"] = {}
+    set_sandbox_provider(provider)
+    try:
+        runtime = Runtime(context=context)
+        result = await middleware.abefore_agent(state, runtime) if async_hook else middleware.before_agent(state, runtime)
+        assert result is not None
+        assert context["sandbox_id"] == "sync-sandbox"
+        assert provider.thread_ids == ["thread-policy"]
+        assert provider.skill_syncs == [("sync-sandbox", "thread-policy", "owner-policy", projection)]
+    finally:
+        reset_sandbox_provider()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("async_hook", [False, True])
+@pytest.mark.parametrize("existing", ["checkpoint", "fork", "context"])
+async def test_lazy_projection_existing_binding_still_fails_closed_on_deny(monkeypatch, async_hook, existing):
+    provider = _AgentSkillSyncProvider()
+    provider.supports_lazy_agent_skill_isolation = True
+    middleware = SandboxMiddleware(available_skills=set())
+    monkeypatch.setattr(middleware, "_prepare_agent_skill_projection", lambda *_args, **_kwargs: object())
+
+    def deny(**_kwargs):
+        raise SandboxAuthorizationError("denied before rebinding")
+
+    async def deny_async(**_kwargs):
+        deny()
+
+    monkeypatch.setattr("deerflow.sandbox.middleware.authorize_sandbox_execution", deny)
+    monkeypatch.setattr("deerflow.sandbox.middleware.authorize_sandbox_execution_async", deny_async)
+    context = {"thread_id": "thread-policy", "user_id": "owner-policy"}
+    state = {}
+    if existing == "context":
+        context["sandbox_id"] = "shared-view-sandbox"
+    else:
+        binding = {"sandbox_id": "shared-view-sandbox"}
+        state["sandbox"] = Overwrite(binding) if existing == "fork" else binding
+    set_sandbox_provider(provider)
+    try:
+        runtime = Runtime(context=context)
+        with pytest.raises(SandboxAuthorizationError, match="denied before rebinding"):
+            if async_hook:
+                await middleware.abefore_agent(state, runtime)
+            else:
+                middleware.before_agent(state, runtime)
+        assert provider.thread_ids == []
+        assert provider.skill_syncs == []
+    finally:
+        reset_sandbox_provider()
+
+
 def test_non_owner_skill_policy_preserves_lazy_init_without_projection_or_acquire(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: