From fa2afa6a390fe66966b36bbac38e5383fe030335 Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Sun, 27 Sep 2026 07:58:15 +0800 Subject: [PATCH] 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 --- .../harness/deerflow/sandbox/AGENTS.md | 13 +-- .../deerflow/sandbox/sandbox_provider.py | 17 ++- ...est_sandbox_network_policy_cancellation.py | 105 ++++++++++++++++++ 3 files changed, 125 insertions(+), 10 deletions(-) create mode 100644 backend/tests/test_sandbox_network_policy_cancellation.py diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 9646cc3f5b..9497851397 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py index 399cbd5512..9a5e1b2a68 100644 --- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py +++ b/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 diff --git a/backend/tests/test_sandbox_network_policy_cancellation.py b/backend/tests/test_sandbox_network_policy_cancellation.py new file mode 100644 index 0000000000..7c18de81be --- /dev/null +++ b/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)