From 8f08df9b48b8788551123b7185f7cf5274023a91 Mon Sep 17 00:00:00 2001 From: spud <92900806+jamespud@users.noreply.github.com> Date: Tue, 22 Sep 2026 20:48:22 +0800 Subject: [PATCH] fix(sandbox): fence ambiguous AIO session creation ownership (#5711) * fix(sandbox): quarantine ambiguous shell session creation * fix(sandbox): quarantine ambiguous bash session creation * fix(sandbox): recycle sandboxes with ambiguous session creates * fix(sandbox): keep quarantined release teardown non-raising * docs(sandbox): keep guidance within instruction budget --------- Co-authored-by: Willem Jiang --- backend/docs/CONFIGURATION.md | 9 + .../community/aio_sandbox/aio_sandbox.py | 146 +++++- .../aio_sandbox/aio_sandbox_provider.py | 36 +- .../harness/deerflow/sandbox/AGENTS.md | 16 +- .../tests/blocking_io/test_sandbox_release.py | 4 +- backend/tests/test_aio_sandbox.py | 428 ++++++++++++++++++ backend/tests/test_aio_sandbox_provider.py | 74 +++ backend/tests/test_github_token_plumbing.py | 28 +- 8 files changed, 711 insertions(+), 30 deletions(-) diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 8e058f49a..6aab4e0d2 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -643,6 +643,15 @@ frozen legacy image only gets the bounded host wait. Timed-out or otherwise ambiguous commands are never replayed, and a partial `list_dir` result is never returned as a complete listing. +Explicit AIO shell/bash session creation is a separate control-plane +operation. DeerFlow bounds those create requests to 5 seconds with SDK +retries disabled. If a response cannot prove whether creation committed, +DeerFlow does not replay the create or execute on that session id. The +affected creation plane is quarantined, bounded best-effort session cleanup +is attempted, and the container is recycled instead of being returned to the +warm pool. Session-level cleanup does not clear that quarantine because a +timed-out create may commit after cleanup has already returned. + **BoxLite micro-VM Sandbox** (runs sandbox code in daemonless OCI micro-VMs): ```yaml sandbox: diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py index 8144e3277..453155cde 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py @@ -45,6 +45,14 @@ class _ScopedShellSession: session_id: str | None = None +@dataclass +class _SessionCreationState: + """Process-local ownership state for one server-side session plane.""" + + pending: set[str] = field(default_factory=set) + ambiguous: set[str] = field(default_factory=set) + + class AioSandbox(Sandbox): """Sandbox implementation using the agent-infra/sandbox Docker container. @@ -112,6 +120,9 @@ class AioSandbox(Sandbox): # Set to True after bash.exec answers 404 (image predates /v1/bash/*), # so later env-bearing calls fail fast instead of re-hitting HTTP (#3921). self._bash_exec_unsupported = False + self._session_creation_state_lock = threading.Lock() + self._shell_session_creation_state = _SessionCreationState() + self._bash_session_creation_state = _SessionCreationState() @property def base_url(self) -> str: @@ -168,6 +179,20 @@ class AioSandbox(Sandbox): ) self._recovery_session_id = None client = self._client + if client is not None: + shell_tombstones, bash_tombstones = self._ambiguous_session_creation_snapshot() + for session_id in shell_tombstones: + self._cleanup_session_best_effort( + client, + session_id, + context="ambiguous shell session creation during close", + request_options=self._bounded_cleanup_request_options(), + ) + for session_id in bash_tombstones: + self._cleanup_bash_session_best_effort( + client, + session_id, + ) # Drop the reference under the lock for use-after-close safety: any # later command on this instance fails loudly instead of reusing a # half-closed client. @@ -244,9 +269,112 @@ class AioSandbox(Sandbox): cleanup_error, ) + @staticmethod + def _is_definite_session_creation_failure(error: Exception) -> bool: + if isinstance(error, httpx.ConnectError): + return True + if isinstance(error, ApiError): + return 400 <= error.status_code < 500 + return False + + def _session_creation_state(self, plane: str) -> _SessionCreationState: + if plane == "shell": + return self._shell_session_creation_state + if plane == "bash": + return self._bash_session_creation_state + raise ValueError(f"unknown session creation plane: {plane}") + + def _begin_session_creation(self, plane: str, session_id: str) -> None: + with self._session_creation_state_lock: + state = self._session_creation_state(plane) + if state.ambiguous: + raise RuntimeError(f"AIO {plane} session creation is quarantined after an earlier ambiguous create outcome; recycle the sandbox before creating another session") + state.pending.add(session_id) + + def _resolve_session_creation(self, plane: str, session_id: str) -> None: + with self._session_creation_state_lock: + self._session_creation_state(plane).pending.discard(session_id) + + def _mark_session_creation_ambiguous( + self, + plane: str, + session_id: str, + ) -> None: + with self._session_creation_state_lock: + state = self._session_creation_state(plane) + state.pending.discard(session_id) + state.ambiguous.add(session_id) + + def _ambiguous_session_creation_snapshot( + self, + ) -> tuple[tuple[str, ...], tuple[str, ...]]: + with self._session_creation_state_lock: + return ( + tuple(self._shell_session_creation_state.ambiguous), + tuple(self._bash_session_creation_state.ambiguous), + ) + + @property + def requires_container_recycle(self) -> bool: + with self._session_creation_state_lock: + shell = self._shell_session_creation_state + bash = self._bash_session_creation_state + return bool(shell.pending or shell.ambiguous or bash.pending or bash.ambiguous) + def _create_shell_session(self, client) -> str: session_id = str(uuid.uuid4()) - client.shell.create_session(id=session_id) + self._begin_session_creation("shell", session_id) + + try: + client.shell.create_session( + id=session_id, + request_options=self._session_create_request_options(), + ) + except Exception as error: + if self._is_definite_session_creation_failure(error): + self._resolve_session_creation("shell", session_id) + raise + + self._mark_session_creation_ambiguous("shell", session_id) + + # Best effort only. A successful cleanup does NOT clear the tombstone: + # the original create may still commit after this cleanup returns. + self._cleanup_session_best_effort( + client, + session_id, + context="ambiguous shell session creation", + request_options=self._bounded_cleanup_request_options(), + ) + raise RuntimeError("AIO shell session creation outcome is unknown; the sandbox is quarantined for recycle") from error + + self._resolve_session_creation("shell", session_id) + return session_id + + def _create_bash_session(self, client) -> str: + session_id = str(uuid.uuid4()) + self._begin_session_creation("bash", session_id) + + try: + client.bash.create_session( + session_id=session_id, + request_options=self._session_create_request_options(), + ) + except Exception as error: + if self._is_definite_session_creation_failure(error): + self._resolve_session_creation("bash", session_id) + raise + + self._mark_session_creation_ambiguous("bash", session_id) + + # Same rule as shell: compensation is bounded but cannot prove that + # the original create will not commit later. + self._cleanup_bash_session_best_effort( + client, + session_id, + ) + raise RuntimeError("AIO bash session creation outcome is unknown; the sandbox is quarantined for recycle") from error + + self._resolve_session_creation("bash", session_id) return session_id def _ensure_default_shell_session_id(self, client) -> str | None: @@ -483,6 +611,13 @@ class AioSandbox(Sandbox): "max_retries": 0, } + @classmethod + def _session_create_request_options(cls) -> dict[str, int]: + return { + "timeout_in_seconds": cls._SESSION_CREATE_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + } + @classmethod def _transport_timeout_error(cls, timeout: float) -> str: request_timeout = cls._command_request_options(timeout)["timeout_in_seconds"] @@ -562,6 +697,7 @@ class AioSandbox(Sandbox): _DEFAULT_HARD_TIMEOUT = 600.0 _REQUEST_TIMEOUT_GRACE_SECONDS = 5.0 _CLEANUP_REQUEST_TIMEOUT_SECONDS = 5 + _SESSION_CREATE_REQUEST_TIMEOUT_SECONDS = 5 # Directory-operation deadline for ``list_dir`` (#5644). ``list_dir`` is an # independent operation, not a shell command: it must not inherit @@ -776,11 +912,9 @@ class AioSandbox(Sandbox): """Single bash.exec invocation in an explicitly released fresh session.""" with self._lock: for attempt in range(2): - session_id = str(uuid.uuid4()) - session_created = False + session_id: str | None = None try: - self._client.bash.create_session(session_id=session_id) - session_created = True + session_id = self._create_bash_session(self._client) result = self._client.bash.exec( command=command, session_id=session_id, @@ -841,7 +975,7 @@ class AioSandbox(Sandbox): logger.error(f"Failed to execute command with injected env in sandbox: {e}") return f"Error: {e}", None finally: - if session_created: + if session_id is not None: self._cleanup_bash_session_best_effort(self._client, session_id) return "Error: bash.exec session disappeared after retry", None 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 9ec83e718..3fd103f24 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 @@ -2399,20 +2399,40 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): return sandbox def release(self, sandbox_id: str) -> None: - """Release a sandbox from active use into the warm pool. + """Release a sandbox from active use. - The container is kept running so it can be reclaimed quickly by the same - thread on its next turn without a cold-start. The container will only be - stopped when the replicas limit forces eviction or during shutdown. + Healthy sandboxes are parked in the warm pool for fast reuse. Sandboxes + quarantined after an ambiguous session-creation outcome are destroyed + instead so unresolved server-side session state is never deliberately + reused. - The host-side HTTP client owned by the cached ``AioSandbox`` instance is - closed before the instance is dropped (#2872). The warm-pool entry only - stores ``SandboxInfo``, so a fresh ``AioSandbox`` (and a fresh client) - is constructed if the container is later reclaimed. + Release is best-effort at turn teardown: recycle failures are logged + rather than propagated to the completed agent run. Args: sandbox_id: The ID of the sandbox to release. """ + with self._lock: + recycle_sandbox = self._sandboxes.get(sandbox_id) + + if recycle_sandbox is not None and recycle_sandbox.requires_container_recycle: + logger.warning( + "Recycling sandbox %s instead of returning it to the warm pool after ambiguous session creation", + sandbox_id, + ) + try: + self._destroy_tracked( + sandbox_id, + still_reapable=lambda: self._sandboxes.get(sandbox_id) is recycle_sandbox, + ) + except Exception: + logger.error( + "Failed to recycle sandbox %s after ambiguous session creation", + sandbox_id, + exc_info=True, + ) + return + info = None sandbox = None thread_keys_to_remove: list[tuple[str, str]] = [] diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index c172007dd..f8027bb00 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -1,12 +1,12 @@ ### Sandbox System (`packages/harness/deerflow/sandbox/`) -**Network approval interaction policy**: Sync and async `SandboxMiddleware` tool -wrappers use `resolve_run_interaction_policy()` to decide whether a lead run can -open a network approval card. Explicit `autonomous`, `webhook`, and `scheduled` -modes auto-deny pending requests, as do the legacy unattended flags and GitHub -channel fallback. An explicit `interactive` mode takes precedence over those -legacy hints. Subagents always auto-deny, including in interactive runs; never -consume events into a human-input card when no human can respond. +**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. **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. @@ -110,7 +110,7 @@ subshell. Keep parser filtering as a backstop; test ignored roots, metacharacter symlinks, and visible depth/size bounds. - Every sandbox tool keeps a model-visible `description` field for a human-readable progress label, but the field is optional and defaults to an empty string. Tool execution must depend only on its operational arguments; the frontend supplies localized fallback labels when a provider omits `description`. -- `bash` - Execute commands with path translation and error handling. For `LocalSandbox`, POSIX/Windows output uses bounded pipe-drain threads with `/dev/null` stdin; Windows decodes locale-code-page/UTF-8/CRLF/bare-CR output with universal-newline translation, while POSIX stays byte-decoded. POSIX background commands return without blocking on inherited pipes; unredirected output is drained without unbounded temp files. Commands that read stdin get immediate EOF. `bash_command_timeout` sets T (600s) for Local/AIO/OpenSandbox; others keep defaults unless explicit. Local: T is a wall-clock process-group deadline. Supported-semver AIO: server-side `hard_timeout`; frozen legacy `all-in-one-sandbox:latest`: command requests get a bounded host `T+5s` wait (`max_retries=0`), so a wedged request cannot hold the sandbox lock for the SDK's 600s budget; session-creation and other file RPCs keep that budget; `list_dir` is bounded separately: 60s hard + 65s host on supported semver, 65s host only on legacy. Never replay ambiguous outcomes (transport errors, terminated, no_change_timeout, unknown). `hard_timeout` => terminated + Exit Code: 124, keep session; `no_change_timeout` => may still be running, fence session generation. The description scopes host probes to LocalSandbox and tells the model to background long-lived processes. See `LocalSandbox.execute_command`, its platform runners, and `bash_tool`'s docstring. +- `bash` - Execute commands with path translation and error handling. For `LocalSandbox`, POSIX/Windows output uses bounded pipe-drain threads with `/dev/null` stdin; Windows decodes locale-code-page/UTF-8/CRLF/bare-CR output with universal-newline translation, while POSIX stays byte-decoded. POSIX background commands return without blocking on inherited pipes; unredirected output is drained without unbounded temp files. Commands that read stdin get immediate EOF. `bash_command_timeout` sets T (600s) for Local/AIO/OpenSandbox; others keep defaults unless explicit. Local: T is a wall-clock process-group deadline. Supported-semver AIO: server-side `hard_timeout`; frozen legacy `all-in-one-sandbox:latest`: command requests get a bounded host `T+5s` wait (`max_retries=0`), so a wedged command request cannot hold the sandbox lock for the SDK's full 600s budget; file/list RPCs are not bounded by it. Explicit AIO shell/bash session creation uses a separate 5s/no-retry host bound; an ambiguous create is never replayed and forces container recycle before warm reuse. Never replay ambiguous outcomes: transport timeout, terminated, no_change_timeout, unknown statuses. `hard_timeout` => terminated + Exit Code: 124, keep session; `no_change_timeout` => may still be running, fence session generation. The description scopes host-environment probes to LocalSandbox (`uname -s`, then `sw_vers` on Darwin; Linux files only when policy permits), gives conditional recovery for local path/`file://` rejections, and tells the model to background long-lived processes. See `LocalSandbox.execute_command`, its platform runners, and `bash_tool`'s docstring. - `ls` - Directory listing (tree format, max 2 levels). Remote commands precheck root existence and emit `__DF_FIND_STATUS__:missing`; `find` status 1 is always an incomplete-traversal `OSError`. `AioSandbox.list_dir` parses only `completed`/`None` on #5634's selected generation: `hard_timeout` raises `TimeoutError` and keeps that generation reusable; an ambiguous timeout or status raises `OSError`, replays nothing, and fences that generation. Missing-session 404s drop the stale id without replay. Persistent/scoped shells fence all `httpx.TransportError`s before cleanup. - `glob` - Find files or directories below a root directory with bounded results - `grep` - Search one text file or recursively search a directory, with optional glob filtering and bounded line-level results diff --git a/backend/tests/blocking_io/test_sandbox_release.py b/backend/tests/blocking_io/test_sandbox_release.py index ba75bd44c..07006eeed 100644 --- a/backend/tests/blocking_io/test_sandbox_release.py +++ b/backend/tests/blocking_io/test_sandbox_release.py @@ -75,7 +75,9 @@ def _make_provider_with_active_sandbox(tmp_path: Path, sandbox_id: str): provider = AioSandboxProvider.__new__(AioSandboxProvider) provider._lock = threading.Lock() - provider._sandboxes = {sandbox_id: MagicMock()} + sandbox = MagicMock() + sandbox.requires_container_recycle = False + provider._sandboxes = {sandbox_id: sandbox} provider._active_sandbox_identity = {sandbox_id: ("default", "thread-1")} provider._sandbox_infos = { sandbox_id: SandboxInfo( diff --git a/backend/tests/test_aio_sandbox.py b/backend/tests/test_aio_sandbox.py index 7c3a04063..feefb3982 100644 --- a/backend/tests/test_aio_sandbox.py +++ b/backend/tests/test_aio_sandbox.py @@ -672,6 +672,434 @@ class TestErrorObservationRetry: assert len(created_ids) == 2 +class TestShellSessionCreationOwnership: + """Shell session creation must be bounded and ambiguous outcomes quarantined.""" + + def test_shell_create_uses_bounded_no_retry_request(self, sandbox): + sandbox._default_shell_corrupted = True + + sandbox._client.shell.create_session = MagicMock(side_effect=lambda id, **kwargs: SimpleNamespace(data=SimpleNamespace(session_id=id))) + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace( + output="ok", + exit_code=0, + status="completed", + ) + ) + ) + + assert sandbox.execute_command("echo ok") == "ok" + + kwargs = sandbox._client.shell.create_session.call_args.kwargs + assert kwargs["request_options"] == { + "timeout_in_seconds": 5, + "max_retries": 0, + } + + def test_shell_ambiguous_create_is_quarantined_without_exec(self, sandbox): + sandbox._default_shell_corrupted = True + create_session = MagicMock(side_effect=httpx.ReadTimeout("response stalled")) + cleanup_session = MagicMock() + exec_command = MagicMock() + + sandbox._client.shell.create_session = create_session + sandbox._client.shell.cleanup_session = cleanup_session + sandbox._client.shell.exec_command = exec_command + + out = sandbox.execute_command("unsafe-to-run") + + assert "session creation outcome is unknown" in out + exec_command.assert_not_called() + + created_id = create_session.call_args.kwargs["id"] + cleanup_session.assert_called_once_with( + created_id, + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + assert sandbox.requires_container_recycle is True + + def test_shell_ambiguous_create_blocks_later_create_without_replay( + self, + sandbox, + ): + sandbox._default_shell_corrupted = True + create_session = MagicMock(side_effect=httpx.ReadTimeout("response stalled")) + sandbox._client.shell.create_session = create_session + sandbox._client.shell.cleanup_session = MagicMock() + + first = sandbox.execute_command("first") + assert "session creation outcome is unknown" in first + assert create_session.call_count == 1 + + create_session.reset_mock() + + second = sandbox.execute_command("second") + + assert "earlier ambiguous create outcome" in second + create_session.assert_not_called() + + def test_shell_connect_error_is_definite_and_does_not_quarantine( + self, + sandbox, + ): + sandbox._default_shell_corrupted = True + create_session = MagicMock(side_effect=httpx.ConnectError("connection refused")) + cleanup_session = MagicMock() + + sandbox._client.shell.create_session = create_session + sandbox._client.shell.cleanup_session = cleanup_session + + assert sandbox.execute_command("first").startswith("Error:") + assert sandbox.requires_container_recycle is False + cleanup_session.assert_not_called() + + sandbox.execute_command("second") + assert create_session.call_count == 2 + + def test_shell_client_error_is_definite_and_does_not_quarantine( + self, + sandbox, + ): + from agent_sandbox.core.api_error import ApiError + + sandbox._default_shell_corrupted = True + sandbox._client.shell.create_session = MagicMock( + side_effect=ApiError( + status_code=400, + body={"message": "invalid request"}, + ) + ) + sandbox._client.shell.cleanup_session = MagicMock() + + assert sandbox.execute_command("bad").startswith("Error:") + assert sandbox.requires_container_recycle is False + sandbox._client.shell.cleanup_session.assert_not_called() + + @pytest.mark.parametrize( + ("status_code", "body"), + [ + (500, {"message": "upstream failed after create"}), + (200, "not-json"), + ], + ) + def test_shell_api_error_without_definite_failure_quarantines( + self, + sandbox, + status_code, + body, + ): + from agent_sandbox.core.api_error import ApiError + + sandbox._default_shell_corrupted = True + sandbox._client.shell.create_session = MagicMock( + side_effect=ApiError( + status_code=status_code, + body=body, + ) + ) + sandbox._client.shell.cleanup_session = MagicMock() + sandbox._client.shell.exec_command = MagicMock() + + out = sandbox.execute_command("unsafe") + + assert "session creation outcome is unknown" in out + assert sandbox.requires_container_recycle is True + sandbox._client.shell.exec_command.assert_not_called() + sandbox._client.shell.cleanup_session.assert_called_once() + + def test_shell_create_already_in_flight_may_finish_but_no_new_create_starts_after_dirty( + self, + sandbox, + ): + first_entered = threading.Event() + release_first = threading.Event() + created_ids: list[str] = [] + exec_ids: list[str] = [] + create_calls = 0 + + def create_session(id, **kwargs): + nonlocal create_calls + create_calls += 1 + created_ids.append(id) + if create_calls == 1: + first_entered.set() + assert release_first.wait(timeout=2) + raise httpx.ReadTimeout("first create stalled") + return SimpleNamespace(data=SimpleNamespace(session_id=id)) + + sandbox._client.shell.create_session = create_session + sandbox._client.shell.cleanup_session = MagicMock() + sandbox._client.shell.exec_command = lambda command, **kwargs: ( + exec_ids.append(kwargs["id"]) + or SimpleNamespace( + data=SimpleNamespace( + output="ok", + exit_code=0, + status="completed", + ) + ) + ) + + first_results: list[str] = [] + + def first_scope() -> None: + first_results.append( + sandbox.execute_command_in_scope( + "first", + scope_id="scope-a", + ) + ) + + thread = threading.Thread(target=first_scope) + thread.start() + assert first_entered.wait(timeout=2) + + # This create began while the plane was still CLEAN. + assert ( + sandbox.execute_command_in_scope( + "second", + scope_id="scope-b", + ) + == "ok" + ) + owned_id = sandbox._scoped_shell_sessions["scope-b"].session_id + assert owned_id is not None + + release_first.set() + thread.join(timeout=2) + assert not thread.is_alive() + assert "session creation outcome is unknown" in first_results[0] + assert sandbox.requires_container_recycle is True + + # Existing confirmed ownership remains usable. + assert ( + sandbox.execute_command_in_scope( + "reuse", + scope_id="scope-b", + ) + == "ok" + ) + assert exec_ids[-1] == owned_id + + before = create_calls + blocked = sandbox.execute_command_in_scope( + "new", + scope_id="scope-c", + ) + assert "earlier ambiguous create outcome" in blocked + assert create_calls == before + + +class TestBashSessionCreationOwnership: + """Bash env-session creation follows the same bounded ownership contract.""" + + def test_bash_create_uses_bounded_no_retry_request(self, sandbox): + sandbox._client.bash.create_session = MagicMock() + sandbox._client.bash.exec = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace( + stdout="ok", + stderr="", + exit_code=0, + status="completed", + ) + ) + ) + + assert ( + sandbox.execute_command( + "echo $TOKEN", + env={"TOKEN": "secret"}, + ) + == "ok" + ) + + kwargs = sandbox._client.bash.create_session.call_args.kwargs + assert kwargs["request_options"] == { + "timeout_in_seconds": 5, + "max_retries": 0, + } + + @pytest.mark.parametrize( + "error_cls", + [ + httpx.ReadTimeout, + httpx.ReadError, + ], + ) + def test_bash_ambiguous_create_is_quarantined_without_exec( + self, + sandbox, + error_cls, + ): + create_session = MagicMock(side_effect=error_cls("response became ambiguous")) + close_session = MagicMock() + exec_command = MagicMock() + + sandbox._client.bash.create_session = create_session + sandbox._client.bash.close_session = close_session + sandbox._client.bash.exec = exec_command + + out = sandbox.execute_command( + "unsafe", + env={"TOKEN": "secret"}, + ) + + assert "session creation outcome is unknown" in out + exec_command.assert_not_called() + + created_id = create_session.call_args.kwargs["session_id"] + close_session.assert_called_once_with( + created_id, + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + assert sandbox.requires_container_recycle is True + + def test_bash_ambiguous_create_blocks_later_create(self, sandbox): + create_session = MagicMock(side_effect=httpx.ReadTimeout("response stalled")) + sandbox._client.bash.create_session = create_session + sandbox._client.bash.close_session = MagicMock() + + first = sandbox.execute_command( + "first", + env={"TOKEN": "secret"}, + ) + assert "session creation outcome is unknown" in first + assert create_session.call_count == 1 + + create_session.reset_mock() + + second = sandbox.execute_command( + "second", + env={"TOKEN": "secret"}, + ) + + assert "earlier ambiguous create outcome" in second + create_session.assert_not_called() + + def test_shell_creation_quarantine_does_not_block_bash_plane( + self, + sandbox, + ): + sandbox._default_shell_corrupted = True + sandbox._client.shell.create_session = MagicMock(side_effect=httpx.ReadTimeout("shell stalled")) + sandbox._client.shell.cleanup_session = MagicMock() + + assert "session creation outcome is unknown" in sandbox.execute_command("shell") + + sandbox._client.bash.create_session = MagicMock() + sandbox._client.bash.exec = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace( + stdout="bash-ok", + stderr="", + exit_code=0, + status="completed", + ) + ) + ) + + assert ( + sandbox.execute_command( + "echo $TOKEN", + env={"TOKEN": "secret"}, + ) + == "bash-ok" + ) + + def test_bash_creation_quarantine_does_not_block_shell_plane( + self, + sandbox, + ): + sandbox._client.bash.create_session = MagicMock(side_effect=httpx.ReadTimeout("bash stalled")) + sandbox._client.bash.close_session = MagicMock() + + assert "session creation outcome is unknown" in sandbox.execute_command( + "bash", + env={"TOKEN": "secret"}, + ) + + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace( + output="shell-ok", + exit_code=0, + status="completed", + ) + ) + ) + + assert sandbox.execute_command("shell") == "shell-ok" + + def test_bash_connect_error_is_definite_and_does_not_quarantine( + self, + sandbox, + ): + sandbox._client.bash.create_session = MagicMock(side_effect=httpx.ConnectError("connection refused")) + sandbox._client.bash.close_session = MagicMock() + + out = sandbox.execute_command( + "echo x", + env={"TOKEN": "secret"}, + ) + + assert out.startswith("Error:") + assert sandbox.requires_container_recycle is False + sandbox._client.bash.close_session.assert_not_called() + + def test_close_retries_ambiguous_creation_cleanup_without_clearing_tombstones( + self, + sandbox, + ): + shell_cleanup = MagicMock() + bash_close = MagicMock() + + sandbox._default_shell_corrupted = True + sandbox._client.shell.create_session = MagicMock(side_effect=httpx.ReadTimeout("shell stalled")) + sandbox._client.shell.cleanup_session = shell_cleanup + + sandbox.execute_command("shell") + shell_id = sandbox._client.shell.create_session.call_args.kwargs["id"] + assert shell_cleanup.call_count == 1 + + sandbox._client.bash.create_session = MagicMock(side_effect=httpx.ReadTimeout("bash stalled")) + sandbox._client.bash.close_session = bash_close + + sandbox.execute_command( + "bash", + env={"TOKEN": "secret"}, + ) + bash_id = sandbox._client.bash.create_session.call_args.kwargs["session_id"] + assert bash_close.call_count == 1 + assert sandbox.requires_container_recycle is True + + sandbox.close() + + assert shell_cleanup.call_count == 2 + assert shell_cleanup.call_args_list[-1].args == (shell_id,) + assert shell_cleanup.call_args_list[-1].kwargs["request_options"] == { + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + } + + assert bash_close.call_count == 2 + assert bash_close.call_args_list[-1].args == (bash_id,) + assert bash_close.call_args_list[-1].kwargs["request_options"] == { + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + } + + # Cleanup is compensation, not proof that a late create cannot commit. + assert sandbox.requires_container_recycle is True + + class TestScopedShellSessions: """Concurrent subagents use independent persistent shell sessions (#5128).""" diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index ed3aa2889..2e2b33d95 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -1280,6 +1280,7 @@ def _make_provider_with_active_sandbox(tmp_path, sandbox_id: str): sandbox = MagicMock() sandbox.id = sandbox_id sandbox.close = MagicMock() + sandbox.requires_container_recycle = False provider._sandboxes = {sandbox_id: sandbox} return provider, sandbox, aio_mod @@ -1338,6 +1339,79 @@ async def test_reset_closes_acquire_serializer_executor(tmp_path): pass +def test_release_dirty_sandbox_branches_before_warm_pool( + tmp_path, +): + provider, sandbox, _ = _make_provider_with_active_sandbox( + tmp_path, + "sandbox-dirty", + ) + sandbox.requires_container_recycle = True + + observed: dict[str, bool] = {} + + def destroy_tracked( + sandbox_id, + *, + still_reapable, + ): + observed["active_before_destroy"] = provider._sandboxes.get(sandbox_id) is sandbox + observed["warm_before_destroy"] = sandbox_id in provider._warm_pool + observed["still_reapable"] = still_reapable() + + provider._destroy_tracked = MagicMock(side_effect=destroy_tracked) + + provider.release("sandbox-dirty") + + assert observed == { + "active_before_destroy": True, + "warm_before_destroy": False, + "still_reapable": True, + } + + +def test_release_dirty_sandbox_destroys_container_instead_of_warming( + tmp_path, +): + provider, sandbox, _ = _make_provider_with_active_sandbox( + tmp_path, + "sandbox-dirty-destroy", + ) + sandbox.requires_container_recycle = True + info = provider._sandbox_infos["sandbox-dirty-destroy"] + + provider.release("sandbox-dirty-destroy") + + assert "sandbox-dirty-destroy" not in provider._warm_pool + assert "sandbox-dirty-destroy" not in provider._sandboxes + assert "sandbox-dirty-destroy" not in provider._sandbox_infos + + sandbox.close.assert_called_once_with() + provider._backend.destroy.assert_called_once_with(info) + + +def test_release_dirty_sandbox_destroy_failure_is_logged_without_warming( + tmp_path, + caplog, +): + provider, sandbox, _ = _make_provider_with_active_sandbox( + tmp_path, + "sandbox-dirty-fail", + ) + sandbox.requires_container_recycle = True + provider._backend.destroy.side_effect = RuntimeError("container stop failed") + + with caplog.at_level("ERROR"): + provider.release("sandbox-dirty-fail") + + assert "sandbox-dirty-fail" not in provider._warm_pool + assert "sandbox-dirty-fail" not in provider._sandboxes + assert "sandbox-dirty-fail" not in provider._sandbox_infos + assert "Failed to recycle sandbox sandbox-dirty-fail" in caplog.text + provider._backend.destroy.assert_called_once() + sandbox.close.assert_called_once_with() + + def test_release_swallows_close_errors(tmp_path, caplog): """A failure inside sandbox.close() must not break provider release().""" provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-rel-err") diff --git a/backend/tests/test_github_token_plumbing.py b/backend/tests/test_github_token_plumbing.py index cebec36f8..c9b2021e0 100644 --- a/backend/tests/test_github_token_plumbing.py +++ b/backend/tests/test_github_token_plumbing.py @@ -18,6 +18,7 @@ concurrent runs on different repos from clobbering each other's token. from __future__ import annotations import os +import threading from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -33,6 +34,17 @@ from deerflow.sandbox.local.local_sandbox import LocalSandbox from deerflow.sandbox.tools import _github_env_from_runtime, bash_tool +def _new_aio_sandbox_with_session_state(): + """Build a manually wired AioSandbox including creation-ownership state.""" + from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox, _SessionCreationState + + sbx = AioSandbox.__new__(AioSandbox) + sbx._session_creation_state_lock = threading.Lock() + sbx._shell_session_creation_state = _SessionCreationState() + sbx._bash_session_creation_state = _SessionCreationState() + return sbx + + def _make_conflict_error(detail: str = "thread_id already exists") -> ConflictError: """Mint a ConflictError that matches what langgraph_sdk would raise on a 409 from ``POST /threads``. Constructing the SDK error directly requires @@ -105,13 +117,13 @@ def test_aio_sandbox_env_routes_through_bash_exec() -> None: persistent-shell ``export … unset`` overlay, which could not keep secrets out of the command string. """ - from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox captured: dict = {} class _FakeBash: - def create_session(self, *, session_id): + def create_session(self, *, session_id, **kwargs): captured["created_session"] = session_id + captured["create_options"] = kwargs.get("request_options") def exec(self, *, command, env=None, **kwargs): captured["command"] = command @@ -122,7 +134,7 @@ def test_aio_sandbox_env_routes_through_bash_exec() -> None: def close_session(self, session_id, **kwargs): captured["closed_session"] = session_id - sbx = AioSandbox.__new__(AioSandbox) + sbx = _new_aio_sandbox_with_session_state() sbx._lock = __import__("threading").Lock() sbx._client = SimpleNamespace(bash=_FakeBash()) sbx._DEFAULT_NO_CHANGE_TIMEOUT = 30 @@ -135,10 +147,13 @@ def test_aio_sandbox_env_routes_through_bash_exec() -> None: assert captured["command"] == "gh pr create" assert captured["env"] == {"GH_TOKEN": "tok-123"} assert captured["created_session"] == captured["exec_session"] == captured["closed_session"] + assert captured["create_options"] == { + "timeout_in_seconds": 5, + "max_retries": 0, + } def test_aio_sandbox_no_env_leaves_command_unchanged() -> None: - from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox captured: dict = {} @@ -153,7 +168,7 @@ def test_aio_sandbox_no_env_leaves_command_unchanged() -> None: captured["command"] = command return _FakeResult() - sbx = AioSandbox.__new__(AioSandbox) + sbx = _new_aio_sandbox_with_session_state() sbx._lock = __import__("threading").Lock() sbx._client = SimpleNamespace(shell=_FakeShell()) sbx._DEFAULT_NO_CHANGE_TIMEOUT = 30 @@ -273,7 +288,6 @@ def test_aio_sandbox_rejects_invalid_env_key() -> None: """End-to-end on the AIO sandbox path — the injection vector flagged in the review never reaches the shell's ``exec_command``. """ - from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox exec_called = False @@ -283,7 +297,7 @@ def test_aio_sandbox_rejects_invalid_env_key() -> None: exec_called = True return SimpleNamespace(data=SimpleNamespace(output="ok")) - sbx = AioSandbox.__new__(AioSandbox) + sbx = _new_aio_sandbox_with_session_state() sbx._lock = __import__("threading").Lock() sbx._client = SimpleNamespace(shell=_FakeShell()) sbx._DEFAULT_NO_CHANGE_TIMEOUT = 30