fix(sandbox): drain network policy mutations across cancellation (#5899)

* fix(sandbox): drain network policy mutations across cancellation

* test(sandbox): pin network policy mutation ownership

* docs(sandbox): pin network policy mutation ownership

* style(sandbox): sort network policy test imports

* fix: resolve sandbox cancellation lint checks

---------

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

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

@@ -2,13 +2,12 @@
 
 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,
-as do legacy unattended flags and the GitHub channel fallback. Explicit
-`interactive` overrides legacy hints. Subagents always auto-deny, even in
-interactive runs; never consume events into a human-input card without a human
-to respond.
+**Network approval policy**: `SandboxMiddleware` uses
+`resolve_run_interaction_policy()`. Autonomous, webhook, scheduled, legacy
+unattended and GitHub-fallback runs auto-deny; explicit interactive overrides
+legacy hints. Subagents always auto-deny. Never show approval cards without a
+human responder. Async claim/deny/decision workers drain through repeated
+cancellation before propagating it.
 
 **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`: `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.

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

@@ -129,7 +129,10 @@ class SandboxProvider(ABC):
         return []
 
     async def consume_network_policy_events_async(self, sandbox_id: str) -> list[dict[str, object]]:
-        return await asyncio.to_thread(self.consume_network_policy_events, sandbox_id)
+        return await run_sync_lifecycle_operation(
+            self.consume_network_policy_events,
+            sandbox_id,
+        )
 
     def deny_pending_network_policy_events(self, sandbox_id: str) -> bool:
         """Atomically deny all unsurfaced trusted-proxy events for a sandbox."""
@@ -137,7 +140,10 @@ class SandboxProvider(ABC):
         return False
 
     async def deny_pending_network_policy_events_async(self, sandbox_id: str) -> bool:
-        return await asyncio.to_thread(self.deny_pending_network_policy_events, sandbox_id)
+        return await run_sync_lifecycle_operation(
+            self.deny_pending_network_policy_events,
+            sandbox_id,
+        )
 
     def decide_network_policy_request(self, sandbox_id: str, request_id: str, decision: str) -> bool:
         """Apply a user decision to one trusted-proxy event."""
@@ -145,7 +151,12 @@ class SandboxProvider(ABC):
         return False
 
     async def decide_network_policy_request_async(self, sandbox_id: str, request_id: str, decision: str) -> bool:
-        return await asyncio.to_thread(self.decide_network_policy_request, sandbox_id, request_id, decision)
+        return await run_sync_lifecycle_operation(
+            self.decide_network_policy_request,
+            sandbox_id,
+            request_id,
+            decision,
+        )
 
 
 _default_sandbox_provider: SandboxProvider | None = None

backend/tests/test_sandbox_network_policy_cancellation.py

@@ -0,0 +1,105 @@
+"""Cancellation ownership for sandbox network-policy mutations."""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+
+import pytest
+
+from deerflow.sandbox.sandbox_provider import SandboxProvider
+
+
+class _BlockingPolicyProvider(SandboxProvider):
+    def __init__(self) -> None:
+        self.entered: dict[str, threading.Event] = {
+            "consume": threading.Event(),
+            "deny": threading.Event(),
+            "decide": threading.Event(),
+        }
+        self.release = threading.Event()
+        self.completed: list[str] = []
+
+    def acquire(
+        self,
+        thread_id: str | None = None,
+        *,
+        user_id: str | None = None,
+    ) -> str:
+        del thread_id, user_id
+        return "sandbox-1"
+
+    def get(self, sandbox_id: str):
+        del sandbox_id
+        return None
+
+    def release(self, sandbox_id: str) -> None:
+        del sandbox_id
+
+    def _block(self, kind: str, result):
+        self.entered[kind].set()
+        if not self.release.wait(5):
+            raise TimeoutError("test did not release network-policy mutation")
+        self.completed.append(kind)
+        return result
+
+    def consume_network_policy_events(self, sandbox_id: str) -> list[dict[str, object]]:
+        del sandbox_id
+        return self._block("consume", [{"request_id": "request-1"}])
+
+    def deny_pending_network_policy_events(self, sandbox_id: str) -> bool:
+        del sandbox_id
+        return self._block("deny", True)
+
+    def decide_network_policy_request(
+        self,
+        sandbox_id: str,
+        request_id: str,
+        decision: str,
+    ) -> bool:
+        del sandbox_id, request_id, decision
+        return self._block("decide", True)
+
+
+async def _deliver_repeated_cancellation(task: asyncio.Task) -> None:
+    task.cancel("first host cancellation")
+    await asyncio.sleep(0)
+    task.cancel("second host cancellation")
+    await asyncio.sleep(0)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("method_name", "args", "kind"),
+    [
+        ("consume_network_policy_events_async", ("sandbox-1",), "consume"),
+        ("deny_pending_network_policy_events_async", ("sandbox-1",), "deny"),
+        (
+            "decide_network_policy_request_async",
+            ("sandbox-1", "request-1", "deny"),
+            "decide",
+        ),
+    ],
+)
+async def test_async_network_policy_mutation_drains_before_cancellation(
+    method_name: str,
+    args: tuple[str, ...],
+    kind: str,
+) -> None:
+    provider = _BlockingPolicyProvider()
+    task = asyncio.create_task(getattr(provider, method_name)(*args))
+
+    try:
+        assert await asyncio.to_thread(provider.entered[kind].wait, 1)
+        await _deliver_repeated_cancellation(task)
+
+        assert not task.done(), "caller cancellation escaped while the policy mutation worker was still running"
+        assert provider.completed == []
+
+        provider.release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert provider.completed == [kind]
+    finally:
+        provider.release.set()
+        await asyncio.gather(task, return_exceptions=True)