diff --git a/README.md b/README.md index ecba7c541..63942d95c 100644 --- a/README.md +++ b/README.md @@ -1685,6 +1685,10 @@ The built-in `grep` tool searches either one text file or all matching text file Remote `ls` excludes ignored descendants before applying its 500-entry listing limit, so dependency and build trees do not crowd out visible files. Explicitly listing an ignored directory still lists its contents; normal depth and output limits remain in effect. +AIO directory listings discard missing shell sessions so the next request can recover. +After a dropped connection, directory listings and persistent shell commands report an +unknown outcome without replaying the operation; later calls use a fresh session. + Uploaded Markdown outlines recognize ATX heading syntax, clean closing markers with a linear suffix scan, and skip fenced code examples, so hashtags and code comments do not crowd out real document sections from the agent's heading preview. UTF-8 Markdown files with or without a byte-order mark (BOM) produce the same diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index fa0f787a9..8e058f49a 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -637,8 +637,11 @@ For AIO images on the supported semver line (`1.9.3` through the recommended `1.11.0` image), `sandbox.bash_command_timeout` is enforced server-side through the `hard_timeout` API when the image exposes it. DeerFlow's legacy frozen `all-in-one-sandbox:latest` image predates that API, so only the host-side -request is bounded there. Timed-out or otherwise ambiguous commands are never -replayed. +request is bounded there. On supported semver AIO images, `list_dir` uses a 60 +second server-side hard timeout with a 65 second no-retry host envelope; the +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. **BoxLite micro-VM Sandbox** (runs sandbox code in daemonless OCI micro-VMs): ```yaml 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 e659407ec..8144e3277 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py @@ -154,6 +154,7 @@ class AioSandbox(Sandbox): self._client, scoped.session_id, context=f"execution scope {scope_id}", + request_options=self._bounded_cleanup_request_options(), ) scoped.session_id = None @@ -163,6 +164,7 @@ class AioSandbox(Sandbox): self._client, self._recovery_session_id, context="default recovery session", + request_options=self._bounded_cleanup_request_options(), ) self._recovery_session_id = None client = self._client @@ -383,17 +385,17 @@ class AioSandbox(Sandbox): session_id=scoped.session_id, timeout=effective_timeout, ) - except httpx.TimeoutException: + except httpx.TransportError as exc: session_id = scoped.session_id scoped.session_id = None if session_id is not None: self._cleanup_session_best_effort( client, session_id, - context="execution scope after transport timeout", + context="execution scope after transport failure", request_options=self._bounded_cleanup_request_options(), ) - return self._transport_timeout_error(effective_timeout) + return self._transport_failure_error(exc, effective_timeout) except ApiError as error: if not self._is_missing_shell_session_error(error): raise @@ -407,8 +409,8 @@ class AioSandbox(Sandbox): context="execution scope after missing session", timeout=effective_timeout, ) - except httpx.TimeoutException: - return self._transport_timeout_error(effective_timeout) + except httpx.TransportError as exc: + return self._transport_failure_error(exc, effective_timeout) if self._is_session_invalidating_shell_status(status): session_id = scoped.session_id scoped.session_id = None @@ -437,8 +439,8 @@ class AioSandbox(Sandbox): context="execution scope", timeout=effective_timeout, ) - except httpx.TimeoutException: - return self._transport_timeout_error(effective_timeout) + except httpx.TransportError as exc: + return self._transport_failure_error(exc, effective_timeout) return self._render_shell_output( output, exit_code, @@ -462,6 +464,7 @@ class AioSandbox(Sandbox): self._client, scoped.session_id, context=f"execution scope {scope_id}", + request_options=self._bounded_cleanup_request_options(), ) scoped.session_id = None @@ -485,6 +488,12 @@ class AioSandbox(Sandbox): request_timeout = cls._command_request_options(timeout)["timeout_in_seconds"] return f"Error: Sandbox command response timed out after {request_timeout} seconds; command outcome is unknown and the command was not retried." + @classmethod + def _transport_failure_error(cls, error: httpx.TransportError, timeout: float) -> str: + if isinstance(error, httpx.TimeoutException): + return cls._transport_timeout_error(timeout) + return "Error: Sandbox command transport failed; command outcome is unknown and the command was not retried." + @staticmethod def _is_unexpected_shell_status(status: str | None) -> bool: return status not in (None, "completed", "hard_timeout", "no_change_timeout", "terminated") @@ -554,6 +563,14 @@ class AioSandbox(Sandbox): _REQUEST_TIMEOUT_GRACE_SECONDS = 5.0 _CLEANUP_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 + # ``bash_command_timeout`` (600s default), or a wedged ``find`` holds + # ``self._lock`` for the full SDK budget. 60s is far above a real + # ``max_depth=2`` traversal and far below the SDK's 600s, and stays a + # private constant rather than new operator config. + _LIST_DIR_TIMEOUT_SECONDS = 60.0 + def _effective_command_timeout(self, timeout: float | None) -> float: return timeout if timeout is not None else (getattr(self, "_default_command_timeout", None) or self._DEFAULT_HARD_TIMEOUT) @@ -630,7 +647,7 @@ class AioSandbox(Sandbox): session_id=session_id, timeout=effective_timeout, ) - except httpx.TimeoutException: + except httpx.TransportError as exc: session_id = self._recovery_session_id self._recovery_session_id = None self._default_shell_corrupted = True @@ -638,10 +655,10 @@ class AioSandbox(Sandbox): self._cleanup_session_best_effort( client, session_id, - context="default shell after transport timeout", + context="default shell after transport failure", request_options=self._bounded_cleanup_request_options(), ) - return self._transport_timeout_error(effective_timeout) + return self._transport_failure_error(exc, effective_timeout) except ApiError as error: if not self._is_missing_shell_session_error(error): raise @@ -657,8 +674,8 @@ class AioSandbox(Sandbox): context="default shell after missing session", timeout=effective_timeout, ) - except httpx.TimeoutException: - return self._transport_timeout_error(effective_timeout) + except httpx.TransportError as exc: + return self._transport_failure_error(exc, effective_timeout) if not recovered_missing_session and status in (None, "completed") and output and _ERROR_OBSERVATION_SIGNATURE in output: self._default_shell_corrupted = True @@ -673,8 +690,8 @@ class AioSandbox(Sandbox): context="default shell", timeout=effective_timeout, ) - except httpx.TimeoutException: - return self._transport_timeout_error(effective_timeout) + except httpx.TransportError as exc: + return self._transport_failure_error(exc, effective_timeout) if self._is_session_invalidating_shell_status(status): session_id = self._recovery_session_id @@ -908,28 +925,82 @@ class AioSandbox(Sandbox): The contents of the directory. """ resolved = path + timeout = self._LIST_DIR_TIMEOUT_SECONDS with self._lock: + client = self._client + session_id: str | None = None try: - client = self._client session_id = self._ensure_default_shell_session_id(client) kwargs = { "command": remote_list_dir_command(resolved, max_depth), - "no_change_timeout": self._DEFAULT_NO_CHANGE_TIMEOUT, + "no_change_timeout": self._effective_no_change_timeout(timeout), + "hard_timeout": timeout, + "request_options": self._command_request_options(timeout), } if session_id is not None: kwargs["id"] = session_id result = client.shell.exec_command(**kwargs) + except httpx.TransportError as exc: + # The response never arrived, so we cannot tell whether ``find`` + # is still running on the targeted shell generation. Fence that + # generation and replay nothing. Local recovery ownership is + # dropped before the cleanup attempt, so a failed cleanup can + # never leave the ambiguous session reusable. + self._default_shell_corrupted = True + if session_id is not None: + self._recovery_session_id = None + self._cleanup_session_best_effort( + client, + session_id, + context="list_dir after transport failure", + request_options=self._bounded_cleanup_request_options(), + ) + logger.error(f"Failed to list directory in sandbox: {exc}") + reason = "request timed out" if isinstance(exc, httpx.TimeoutException) else "transport failed" + raise OSError(f"Failed to list directory '{resolved}': {reason}; directory result is unknown") from exc + except ApiError as exc: + if self._is_missing_shell_session_error(exc): + # The server has lost this generation. Forget it without replaying + # the listing; the next call creates a fresh recovery session. + self._default_shell_corrupted = True + self._recovery_session_id = None + logger.error(f"Failed to list directory in sandbox: {exc}") + raise OSError(f"Failed to list directory '{resolved}' in sandbox: {exc}") from exc except Exception as e: logger.error(f"Failed to list directory in sandbox: {e}") raise OSError(f"Failed to list directory '{resolved}' in sandbox: {e}") from e - if result.data is None: + + data = result.data if result else None + if data is None: raise OSError(f"Failed to list directory '{resolved}' in sandbox: empty response") + + # Only a completed listing is a listing. ``list_dir`` returns + # ``list[str]`` or raises; it never surfaces a partial ``find`` as + # the directory's contents. + status = getattr(data, "status", None) + if status == "hard_timeout": + raise TimeoutError(f"Failed to list directory '{resolved}': find timed out after {timeout:g} seconds; directory result may be incomplete") + if self._is_session_invalidating_shell_status(status): + # Same contract as an ambiguous transport timeout: fence the + # generation that actually executed this listing, dropping local + # recovery ownership before the bounded cleanup attempt. + self._default_shell_corrupted = True + if session_id is not None: + self._recovery_session_id = None + self._cleanup_session_best_effort( + client, + session_id, + context=f"list_dir after ambiguous status {status}", + request_options=self._bounded_cleanup_request_options(), + ) + raise OSError(f"Failed to list directory '{resolved}' in sandbox: command status '{status}'; directory result is unknown") + return parse_remote_list_dir_output( - result.data.output or "", + data.output or "", resolved, - pipeline_exit_code=getattr(result.data, "exit_code", None), + pipeline_exit_code=getattr(data, "exit_code", None), ) def write_file(self, path: str, content: str, append: bool = False) -> None: diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 69ee306f9..c172007dd 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -110,8 +110,8 @@ 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 command request cannot hold the sandbox lock for the SDK's full 600s budget; session-creation and file/list RPCs are not bounded by it. 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`, including when no entries were printed. Do not infer a missing path from status 1 alone. +- `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. +- `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 - `read_file` - Read file contents with optional line range diff --git a/backend/tests/test_aio_sandbox.py b/backend/tests/test_aio_sandbox.py index 35e1ecb37..7c3a04063 100644 --- a/backend/tests/test_aio_sandbox.py +++ b/backend/tests/test_aio_sandbox.py @@ -1266,6 +1266,22 @@ class TestScopedShellSessions: assert sandbox._scoped_shell_sessions == {} client.shell.create_session.assert_not_called() + def test_release_command_scope_uses_bounded_cleanup(self, sandbox): + from deerflow.community.aio_sandbox.aio_sandbox import _ScopedShellSession + + sandbox._scoped_shell_sessions["scope-a"] = _ScopedShellSession(session_id="session-a") + sandbox._client.shell.cleanup_session = MagicMock() + + sandbox.release_command_scope("scope-a") + + sandbox._client.shell.cleanup_session.assert_called_once_with( + "session-a", + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + class TestBashExecUnsupportedFailFast: """Regression tests for #3921: sandbox images older than all-in-one-sandbox @@ -1558,6 +1574,179 @@ class TestListDirSerialization: sandbox._client.shell.exec_command.assert_not_called() +class TestListDirTimeout: + """list_dir owns a directory-operation deadline, independent of bash_command_timeout (#5644). + + ``list_dir`` shells out to ``find`` while holding ``self._lock``, on whichever + shell generation #5634 selects: the implicit persistent session, or the + explicit recovery session once the implicit one has been fenced. Before this + contract it sent only the SDK's 600s ``no_change_timeout`` and used the SDK + client's 600s transport budget, so a wedged ``find`` held the sandbox lock for + the full SDK timeout. These tests pin the directory deadline, the + returned-status matrix, the target-generation fencing, and the "exception + must release the lock" liveness invariant. + """ + + def test_list_dir_passes_hard_timeout_and_bounded_request_options(self, sandbox): + """find runtime <= 60s, request wait <= 65s, no retry; idle guard cannot preempt it.""" + calls = [] + + def exec_command(command, **kwargs): + calls.append(kwargs) + return SimpleNamespace( + data=SimpleNamespace( + output="/a\n/b\n\n__DF_FIND_STATUS__:0\n", + exit_code=0, + status="completed", + ) + ) + + sandbox._client.shell.exec_command = exec_command + + assert sandbox.list_dir("/test") == ["/a", "/b"] + + budget = type(sandbox)._LIST_DIR_TIMEOUT_SECONDS + assert budget == 60.0 + assert len(calls) == 1 + assert calls[0]["hard_timeout"] == budget + assert calls[0]["request_options"] == {"timeout_in_seconds": 65, "max_retries": 0} + # no_change_timeout must not be the binding constraint: it stays strictly above + # the hard timeout, so a find that is merely quiet still dies on hard_timeout. + assert calls[0]["no_change_timeout"] > budget + + def test_list_dir_transport_timeout_releases_lock_and_fences_implicit_shell(self, sandbox): + """A host transport timeout is ambiguous: fence the implicit shell, replay nothing, free the lock.""" + calls = [] + + def exec_command(command, **kwargs): + calls.append(kwargs) + raise httpx.ReadTimeout("response stalled") + + sandbox._client.shell.exec_command = exec_command + + with pytest.raises(OSError, match="Failed to list directory") as exc: + sandbox.list_dir("/test") + + assert "unknown" in str(exc.value) + assert len(calls) == 1, "an ambiguous list_dir outcome must never be replayed" + assert sandbox._default_shell_corrupted is True + assert sandbox._lock.acquire(blocking=False) is True, "list_dir must release the lock after a transport timeout" + sandbox._lock.release() + + def test_list_dir_transport_timeout_fences_targeted_recovery_session(self, sandbox): + """An ambiguous list_dir on the recovery session must drop that generation, not keep it.""" + sandbox._default_shell_corrupted = True + sandbox._recovery_session_id = "recovery-session" + cleanup_session = MagicMock() + sandbox._client.shell.cleanup_session = cleanup_session + sandbox._client.shell.exec_command = MagicMock(side_effect=httpx.ConnectTimeout("connect stalled")) + + with pytest.raises(OSError): + sandbox.list_dir("/test") + + kwargs = sandbox._client.shell.exec_command.call_args.kwargs + assert kwargs["id"] == "recovery-session" + assert sandbox._default_shell_corrupted is True + assert sandbox._recovery_session_id is None + cleanup_session.assert_called_once_with( + "recovery-session", + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + + def test_list_dir_hard_timeout_raises_without_fencing_shell(self, sandbox): + """hard_timeout is a definite termination: raise, but keep the targeted generation reusable.""" + sandbox._default_shell_corrupted = True + sandbox._recovery_session_id = "recovery-session" + sandbox._client.shell.cleanup_session = MagicMock() + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace( + output="/a\n/b\n\n__DF_FIND_STATUS__:0\n", + exit_code=None, + status="hard_timeout", + ) + ) + ) + + with pytest.raises(TimeoutError) as exc: + sandbox.list_dir("/test") + + assert type(exc.value) is TimeoutError + kwargs = sandbox._client.shell.exec_command.call_args.kwargs + assert kwargs["id"] == "recovery-session" + assert sandbox._default_shell_corrupted is True + assert sandbox._recovery_session_id == "recovery-session" + sandbox._client.shell.cleanup_session.assert_not_called() + assert sandbox._lock.acquire(blocking=False) is True + sandbox._lock.release() + + def test_list_dir_partial_output_with_hard_timeout_is_not_a_listing(self, sandbox): + """Partial find stdout under hard_timeout must never be returned as a complete listing.""" + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace(output="/a\n/b\n", exit_code=None, status="hard_timeout"), + ) + ) + + with pytest.raises(TimeoutError): + sandbox.list_dir("/test") + + @pytest.mark.parametrize("status", ["no_change_timeout", "terminated", "running", "pending", "weird_future_status"]) + def test_list_dir_ambiguous_status_does_not_return_partial_listing(self, sandbox, status): + """Ambiguous statuses raise, never parse, and fence the generation that ran the listing.""" + sandbox._default_shell_corrupted = True + sandbox._recovery_session_id = "recovery-session" + cleanup_session = MagicMock() + sandbox._client.shell.cleanup_session = cleanup_session + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace(output="/a\n/b\n", exit_code=0, status=status), + ) + ) + + with pytest.raises(OSError, match="Failed to list directory") as exc: + sandbox.list_dir("/test") + + assert type(exc.value) is OSError + kwargs = sandbox._client.shell.exec_command.call_args.kwargs + assert kwargs["id"] == "recovery-session" + assert sandbox._default_shell_corrupted is True + assert sandbox._recovery_session_id is None + cleanup_session.assert_called_once_with( + "recovery-session", + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + + def test_list_dir_completed_status_preserves_existing_parsing(self, sandbox): + """A completed find still follows the shared stdout contract, including missing-path classification.""" + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace( + output="/test\n/test/sub\n\n__DF_FIND_STATUS__:0\n", + exit_code=0, + status="completed", + ) + ) + ) + + assert sandbox.list_dir("/test") == ["/test", "/test/sub"] + + sandbox._client.shell.exec_command = MagicMock( + return_value=SimpleNamespace( + data=SimpleNamespace(output="\n__DF_FIND_STATUS__:missing\n", exit_code=1, status="completed"), + ) + ) + + with pytest.raises(FileNotFoundError): + sandbox.list_dir("/missing") + + class TestNoChangeTimeout: """Verify that no_change_timeout is forwarded to every exec_command call.""" @@ -2061,6 +2250,38 @@ class TestClose: sandbox.close() # must not raise assert sandbox._client is None + def test_close_scoped_session_cleanup_uses_bounded_request(self, sandbox): + from deerflow.community.aio_sandbox.aio_sandbox import _ScopedShellSession + + sandbox._scoped_shell_sessions["scope-a"] = _ScopedShellSession(session_id="session-a") + cleanup_session = MagicMock() + sandbox._client.shell.cleanup_session = cleanup_session + + sandbox.close() + + cleanup_session.assert_called_once_with( + "session-a", + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + + def test_close_recovery_session_cleanup_uses_bounded_request(self, sandbox): + sandbox._recovery_session_id = "recovery-session" + cleanup_session = MagicMock() + sandbox._client.shell.cleanup_session = cleanup_session + + sandbox.close() + + cleanup_session.assert_called_once_with( + "recovery-session", + request_options={ + "timeout_in_seconds": sandbox._CLEANUP_REQUEST_TIMEOUT_SECONDS, + "max_retries": 0, + }, + ) + def test_list_dir_preserves_trailing_space_in_filename(sandbox): """ "notes.txt " (trailing space) is a legal Linux filename; find prints it diff --git a/backend/tests/test_aio_sandbox_recovery_errors.py b/backend/tests/test_aio_sandbox_recovery_errors.py new file mode 100644 index 000000000..ffd8434b9 --- /dev/null +++ b/backend/tests/test_aio_sandbox_recovery_errors.py @@ -0,0 +1,126 @@ +"""AIO transport failures must not leave a shell generation reusable.""" + +from types import SimpleNamespace +from unittest.mock import patch + +import httpx +import pytest +from agent_sandbox.core.api_error import ApiError + +from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox, _ScopedShellSession + + +@pytest.fixture +def sandbox(): + with patch("deerflow.community.aio_sandbox.aio_sandbox.AioSandboxClient"): + yield AioSandbox(id="recovery-test", base_url="http://localhost:8080") + + +def completed(output="/test\n/test/file\n\n__DF_FIND_STATUS__:0\n"): + return SimpleNamespace(data=SimpleNamespace(output=output, exit_code=0, status="completed")) + + +@pytest.mark.parametrize("session_id", [None, "old-session"]) +@pytest.mark.parametrize("error_type", [httpx.ReadError, httpx.WriteError, httpx.RemoteProtocolError]) +def test_list_dir_transport_failure_fences_before_cleanup_and_recovers(sandbox, session_id, error_type): + sandbox._default_shell_corrupted = session_id is not None + sandbox._recovery_session_id = session_id + client = sandbox._client + client.shell.exec_command.side_effect = [error_type("connection lost"), completed()] + + cleanup_states = [] + + def cleanup(*args, **kwargs): + cleanup_states.append((sandbox._recovery_session_id, sandbox._default_shell_corrupted)) + raise httpx.ReadError("cleanup also failed") + + client.shell.cleanup_session.side_effect = cleanup + with pytest.raises(OSError, match="unknown"): + sandbox.list_dir("/test") + assert client.shell.exec_command.call_count == 1 + assert sandbox._default_shell_corrupted + assert sandbox._recovery_session_id is None + assert sandbox._lock.acquire(blocking=False) + sandbox._lock.release() + if session_id: + assert cleanup_states == [(None, True)] + client.shell.cleanup_session.assert_called_once_with(session_id, request_options={"timeout_in_seconds": 5, "max_retries": 0}) + else: + client.shell.cleanup_session.assert_not_called() + assert sandbox.list_dir("/test") == ["/test", "/test/file"] + replacement = client.shell.create_session.call_args.kwargs["id"] + assert replacement != session_id + assert client.shell.exec_command.call_args.kwargs["id"] == replacement + + +@pytest.mark.parametrize("session_id", [None, "missing-session"]) +def test_list_dir_missing_session_is_forgotten_without_replaying(sandbox, session_id): + sandbox._default_shell_corrupted = session_id is not None + sandbox._recovery_session_id = session_id + client = sandbox._client + client.shell.exec_command.side_effect = [ApiError(status_code=404, body={"message": "Shell session not found"}), completed()] + with pytest.raises(OSError): + sandbox.list_dir("/test") + assert client.shell.exec_command.call_count == 1 + assert sandbox._default_shell_corrupted + assert sandbox._recovery_session_id is None + client.shell.cleanup_session.assert_not_called() + assert sandbox.list_dir("/test") == ["/test", "/test/file"] + replacement = client.shell.create_session.call_args.kwargs["id"] + assert replacement != session_id + assert client.shell.exec_command.call_args.kwargs["id"] == replacement + + +@pytest.mark.parametrize("status,body", [(404, {"message": "route not found"}), (503, {"message": "session not found"}), (404, "session not found")]) +def test_list_dir_other_api_errors_do_not_discard_recovery_session(sandbox, status, body): + sandbox._default_shell_corrupted = True + sandbox._recovery_session_id = "keep-session" + sandbox._client.shell.exec_command.side_effect = ApiError(status_code=status, body=body) + with pytest.raises(OSError): + sandbox.list_dir("/test") + assert sandbox._recovery_session_id == "keep-session" + sandbox._client.shell.create_session.assert_not_called() + sandbox._client.shell.cleanup_session.assert_not_called() + + +@pytest.mark.parametrize("scoped", [False, True]) +@pytest.mark.parametrize("error_type", [httpx.ReadError, httpx.WriteError, httpx.RemoteProtocolError]) +def test_command_transport_failure_fences_without_replay(sandbox, scoped, error_type): + client = sandbox._client + sandbox._default_shell_corrupted = True + sandbox._recovery_session_id = "default-session" + sandbox._scoped_shell_sessions["task"] = _ScopedShellSession(session_id="scoped-session") + client.shell.exec_command.side_effect = [error_type("connection lost"), completed("recovered")] + invoke = (lambda: sandbox.execute_command_in_scope("echo test", scope_id="task")) if scoped else (lambda: sandbox.execute_command("echo test")) + result = invoke() + assert "unknown" in result + assert "not retried" in result + assert client.shell.exec_command.call_count == 1 + if scoped: + assert sandbox._scoped_shell_sessions["task"].session_id is None + assert sandbox._recovery_session_id == "default-session" + else: + assert sandbox._recovery_session_id is None + assert sandbox._scoped_shell_sessions["task"].session_id == "scoped-session" + client.shell.cleanup_session.assert_called_once_with("scoped-session" if scoped else "default-session", request_options={"timeout_in_seconds": 5, "max_retries": 0}) + assert invoke() == "recovered" + assert client.shell.exec_command.call_args.kwargs["id"] == client.shell.create_session.call_args.kwargs["id"] + + +@pytest.mark.parametrize("scoped", [False, True]) +@pytest.mark.parametrize("missing", [False, True]) +def test_replacement_transport_failure_is_cleaned_up_without_third_attempt(sandbox, scoped, missing): + client = sandbox._client + sandbox._default_shell_corrupted = True + sandbox._recovery_session_id = "default-session" + sandbox._scoped_shell_sessions["task"] = _ScopedShellSession(session_id="scoped-session") + first = ApiError(status_code=404, body={"message": "Session not found"}) if missing else completed("'ErrorObservation' object has no attribute 'exit_code'") + client.shell.exec_command.side_effect = [first, httpx.RemoteProtocolError("connection dropped during recovery")] + result = sandbox.execute_command_in_scope("echo test", scope_id="task") if scoped else sandbox.execute_command("echo test") + assert "unknown" in result + assert "not retried" in result + assert client.shell.exec_command.call_count == 2 + replacement = client.shell.create_session.call_args.kwargs["id"] + assert client.shell.cleanup_session.call_args.args == (replacement,) + assert client.shell.cleanup_session.call_args.kwargs == {"request_options": {"timeout_in_seconds": 5, "max_retries": 0}} + assert (sandbox._scoped_shell_sessions["task"].session_id if scoped else sandbox._recovery_session_id) is None