fix(sandbox): bound AIO list_dir with a directory deadline (#5662)

* fix(sandbox): bound remaining shell session cleanup requests

release_command_scope(), close()'s scoped-session drain, and close()'s recovery-session cleanup called shell.cleanup_session() without a request budget, so a stalled cleanup could hold the thread-key serializer, scoped.lock, or the sandbox lock for the SDK's default transport budget. Pass the existing _bounded_cleanup_request_options() (5s, max_retries=0) at those three sites; _cleanup_session_best_effort() and its swallow-and-log contract are unchanged, as are create_session and list_dir lifetimes.

* fix(sandbox): bound AIO list_dir with a directory deadline

Give AioSandbox.list_dir its own 60s directory deadline with a 65s no-retry
host envelope, independent of bash_command_timeout. Preserve #5634's
shell-generation selection: list_dir runs on the current recovery session once
the implicit shell is fenced, and an ambiguous list_dir outcome - transport
timeout or an ambiguous returned status - fences whichever generation actually
executed the request, dropping local recovery ownership and attempting bounded
best-effort cleanup instead of leaving that session reusable. hard_timeout
remains definite termination and keeps the targeted session reusable. Only
completed/None results are parsed, so a partial find is never returned as a
complete listing.

Session creation RPC lifetime remains out of scope.

* fix(sandbox): recover sessions after transport failures

---------

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

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

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
+request is bounded there. On supported semver AIO images, `list_dir` uses a 60
-replayed.
+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

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:
+                    except httpx.TransportError as exc:
-                        return self._transport_timeout_error(effective_timeout)
+                        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:
+                    except httpx.TransportError as exc:
-                        return self._transport_timeout_error(effective_timeout)
+                        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:
+                    except httpx.TransportError as exc:
-                        return self._transport_timeout_error(effective_timeout)
+                        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:
+                    except httpx.TransportError as exc:
-                        return self._transport_timeout_error(effective_timeout)
+                        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:

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.
+- `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`, including when no entries were printed. Do not infer a missing path from status 1 alone.
+- `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

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

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