fix(sandbox): enforce image byte limits during downloads (#6712)

* fix(sandbox): enforce image byte limits during downloads

* docs: keep image download guidance within inherited budget

* docs(sandbox): specify download overflow error contract

* refactor(sandbox): consolidate download budgets and explain AIO locking

Signed-off-by: ZJPex <3258236335@qq.com>

---------

Signed-off-by: ZJPex <3258236335@qq.com>

README.md

@@ -2277,6 +2277,8 @@ On Windows, `st_ctime_ns` may represent creation time; a same-size rewrite that
 restores the original `mtime` can evade validation. Timestamp checks are
 conservative metadata validation, not a guarantee of content equality.
 
+Sandbox image downloads enforce the 20 MiB vision limit while reading, including model-context rereads. Oversized streams stop at the first over-limit chunk. Custom sandbox providers must accept and enforce the `download_file(..., max_bytes=...)` keyword; older implementations fail the image read instead of falling back to an unrestricted download.
+
 Image bytes loaded for a vision-model call are transient: DeerFlow removes the hidden base64 message after the model consumes it so later checkpoints do not keep duplicating that payload.
 
 After each run, DeerFlow records a workspace change summary for the run-owned `workspace` and `outputs` directories. The Web UI shows a compact "files changed" badge on the assistant turn; opening it reveals created, modified, and deleted files with text diffs when safe to display. Uploads are excluded because they are user inputs, not agent-generated changes, and stdio MCP temporary/debug files under the DeerFlow-owned `.mcp/` namespace are excluded because they are process-internal state (like `.git/` and `node_modules/`, any directory named `.mcp` is excluded at any depth). Large, binary, or sensitive-looking files are shown as metadata only.

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -119,7 +119,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
     TODO attribution skips non-string statuses before membership; preserve args/TODO state for normal tool validation.
 24. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint.
 25. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses); captures the runtime-resolved user so standalone LangGraph Server reads and writes stay in the same bucket
-26. **ViewImageMiddleware** - *(vision models only)* Rebuilds hidden base64 image context in `ModelRequest.messages`, never in new checkpoint state; state retains lightweight `viewed_images` metadata. Blob refs resolve first and fail over to existing paths. `_image_injection_plan` first sweeps stale server-marked context (including payloads stranded by the old `before_model`/`after_model` lifecycle), then shares one read-eligibility predicate between sync injection and async authorization. The sweep requires both the reserved ID prefix and server-owned marker, preserving user messages and unmarked legacy payloads. Before reading images, both paths recheck `sandbox:execute`, including restored views after role changes; denial skips I/O and removes marked stale context. Host-copy reads additionally require the saved `actual_path` to match the current user/thread's canonical virtual image path, including for old checkpoints and replacement sandboxes; a missing or mismatched thread identity skips that host read. Only the async wrapper passes `authorization_checked=True` after its gate, avoiding a duplicate synchronous provider call in its worker.
+26. **ViewImageMiddleware** - *(vision models only)* Enforces the shared vision cap on remote rereads (see [download contract](../../sandbox/AGENTS.md)). Rebuilds hidden base64 context in `ModelRequest.messages`, keeping lightweight `viewed_images` in checkpoints. Validated blob refs precede compatibility paths. `_image_injection_plan` removes stale server-marked context, including old `before_model`/`after_model` payloads, then shares eligibility for sync/async injection. Removal requires both the reserved ID prefix and marker, preserving user and unmarked legacy messages. Both paths recheck `sandbox:execute` before I/O. Host copies require `actual_path` to match the current user/thread's canonical virtual path, even for old checkpoints or replacement sandboxes; absent/mismatched identity skips reads. Only the async wrapper passes `authorization_checked=True` after its gate to avoid duplicate checks in its worker.
 27. **McpRoutingMiddleware** - Auto-promotes deferred schemas matching the latest real user message before `DeferredToolFilterMiddleware`, without executing tools. New names emit `middleware:tool_promotion` with `source=routing_hint`; repeat passes emit nothing
 28. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, derives subagent attribution from the server-installed recorder, returns the original `Command`, and omits private payloads
 29. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)

backend/packages/harness/deerflow/agents/middlewares/view_image_middleware.py

@@ -14,14 +14,11 @@ from langchain.agents.middleware.types import ModelCallResult, ModelRequest, Mod
 from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
 
 from deerflow.agents.thread_state import ThreadState
+from deerflow.sandbox.download import MAX_IMAGE_BYTES as _MAX_IMAGE_BYTES
 from deerflow.sandbox.lease import run_sync_lifecycle_operation
 
 logger = logging.getLogger(__name__)
 
-# Mirror the tool-side size cap as a defense-in-depth check. The tool
-# enforces this at write time; the middleware re-checks at read time in
-# case the file grew between view and injection.
-_MAX_IMAGE_BYTES = 20 * 1024 * 1024
 _IMAGE_CONTEXT_MESSAGE_ID_PREFIX = "view-image-context:"
 _IMAGE_CONTEXT_MESSAGE_MARKER_KEY = "deerflow_view_image_context"
 
@@ -266,7 +263,7 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
                     if host_data_url is not None:
                         return host_data_url
                 try:
-                    image_bytes = sandbox.download_file(image_path)
+                    image_bytes = sandbox.download_file(image_path, max_bytes=_MAX_IMAGE_BYTES)
                 except Exception:
                     logger.warning(
                         "Failed to recover viewed image %s from replacement sandbox %s",
@@ -289,7 +286,7 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
                 return None
 
             try:
-                image_bytes = sandbox.download_file(image_path)
+                image_bytes = sandbox.download_file(image_path, max_bytes=_MAX_IMAGE_BYTES)
             except Exception:
                 logger.warning("Failed to read viewed image %s from sandbox %s", image_path, sandbox_id, exc_info=True)
                 return None

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py

@@ -1,5 +1,4 @@
 import base64
-import errno
 import logging
 import math
 import threading
@@ -11,6 +10,8 @@ from agent_sandbox import Sandbox as AioSandboxClient
 from agent_sandbox.core.api_error import ApiError
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT as _MAX_DOWNLOAD_SIZE
+from deerflow.sandbox.download import download_limit, read_download_stream
 from deerflow.sandbox.remote_list_dir import parse_remote_list_dir_output, remote_list_dir_command
 from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env
 from deerflow.sandbox.search import GrepMatch, path_matches, should_ignore_path_under_root, truncate_line
@@ -19,8 +20,6 @@ from .backend import sandbox_http_trust_env
 
 logger = logging.getLogger(__name__)
 
-_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024  # 100 MB
-
 _ERROR_OBSERVATION_SIGNATURE = "'ErrorObservation' object has no attribute 'exit_code'"
 
 # Env-bearing commands require the bash.exec API (POST /v1/bash/exec), which the
@@ -1034,7 +1033,7 @@ class AioSandbox(Sandbox):
             logger.error(f"Failed to read file in sandbox: {e}")
             return f"Error: {e}"
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         """Download file bytes from the sandbox.
 
         Raises:
@@ -1042,6 +1041,7 @@ class AioSandbox(Sandbox):
                 outside ``VIRTUAL_PATH_PREFIX``.
             OSError: If the file cannot be retrieved from the sandbox.
         """
+        limit = download_limit(max_bytes, _MAX_DOWNLOAD_SIZE)
         # Reject path traversal before sending to the container API.
         # LocalSandbox gets this implicitly via _resolve_path;
         # here the path is forwarded verbatim so we must check explicitly.
@@ -1057,20 +1057,14 @@ class AioSandbox(Sandbox):
             logger.error("Refused download outside allowed directory: path=%s, allowed_prefix=%s", path, VIRTUAL_PATH_PREFIX)
             raise PermissionError(f"Access denied: path must be under '{VIRTUAL_PATH_PREFIX}': '{path}'")
 
+        # Keep the lock through iteration and stream cleanup: the SDK iterator
+        # opens its HTTP response lazily, and close() detaches this client under
+        # the same lock before closing its pooled connections. Request-level
+        # concurrency does not make a download safe against concurrent close.
         with self._lock:
             try:
-                chunks: list[bytes] = []
-                total = 0
-                for chunk in self._client.file.download_file(path=path):
-                    total += len(chunk)
-                    if total > _MAX_DOWNLOAD_SIZE:
-                        raise OSError(
-                            errno.EFBIG,
-                            f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes",
-                            path,
-                        )
-                    chunks.append(chunk)
-                return b"".join(chunks)
+                stream = self._client.file.download_file(path=path)
+                return read_download_stream(stream, limit, path)
             except OSError:
                 raise
             except Exception as e:

backend/packages/harness/deerflow/community/boxlite/box.py

@@ -25,6 +25,8 @@ import threading
 from typing import TYPE_CHECKING, TypeVar
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT as _MAX_DOWNLOAD_SIZE
+from deerflow.sandbox.download import download_limit
 from deerflow.sandbox.read_file_contract import split_file_lines
 from deerflow.sandbox.remote_list_dir import parse_remote_list_dir_output, remote_list_dir_command
 from deerflow.sandbox.remote_search import parse_remote_search_output, remote_search_command
@@ -40,7 +42,6 @@ logger = logging.getLogger(__name__)
 
 T = TypeVar("T")
 
-_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024  # 100 MB
 # One base64 chunk stays well under Linux MAX_ARG_STRLEN (128 KiB per argv entry),
 # and 60000 is a multiple of 4 so each chunk is a self-contained base64 unit whose
 # decoded bytes concatenate losslessly.
@@ -267,31 +268,31 @@ class BoxliteBox(Sandbox):
                 raise OSError(f"write '{resolved}' failed: {(r.stderr or '').strip()}")
             first = False
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
+        limit = download_limit(max_bytes, _MAX_DOWNLOAD_SIZE)
         normalized = self._guard_traversal(path)
         stripped = normalized.lstrip("/")
         allowed = VIRTUAL_PATH_PREFIX.lstrip("/")
         if stripped != allowed and not stripped.startswith(f"{allowed}/"):
             raise PermissionError(f"Access denied: path must be under '{VIRTUAL_PATH_PREFIX}': '{path}'")
 
-        # Enforce the size cap before buffering the whole payload.
-        size_r = self._sh(f"wc -c < {shlex.quote(normalized)}")
-        if size_r.exit_code not in (0, None):
-            raise OSError(f"cannot read '{path}' from box: {(size_r.stderr or '').strip() or 'not found'}")
-        try:
-            size = int((size_r.stdout or "0").strip() or "0")
-        except ValueError:
-            size = 0
-        if size > _MAX_DOWNLOAD_SIZE:
-            raise OSError(errno.EFBIG, f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", path)
-
-        r = self._sh(f"base64 {shlex.quote(normalized)}")
+        # Limit the remote read itself, including a one-byte overflow probe.
+        # sh lacks pipefail: encode head's status with the payload so base64
+        # success cannot hide a missing file or a partial read failure.
+        script = f"{{ head -c {limit + 1} {shlex.quote(normalized)}; printf '\\n__DF_READ_STATUS__:%s\\n' \"$?\"; }} | base64"
+        r = self._sh(script)
         if r.exit_code not in (0, None):
             raise OSError(f"cannot read '{path}' from box: {(r.stderr or '').strip()}")
         try:
-            return base64.b64decode("".join((r.stdout or "").split()))
+            decoded = base64.b64decode("".join((r.stdout or "").split()), validate=True)
         except Exception as e:
             raise OSError(f"failed to decode '{path}' from box: {e}") from e
+        data, marker, status = decoded.rpartition(b"\n__DF_READ_STATUS__:")
+        if not marker or status != b"0\n":
+            raise OSError(f"cannot read '{path}' from box: {(r.stderr or '').strip() or 'incomplete read'}")
+        if len(data) > limit:
+            raise OSError(errno.EFBIG, f"File exceeds maximum download size of {limit} bytes", path)
+        return data
 
     def list_dir(self, path: str, max_depth: int = 2) -> list[str]:
         resolved = self._resolve_path(path)

backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py

@@ -11,6 +11,8 @@ from e2b import CommandExitException, FileNotFoundException
 from e2b_code_interpreter import Sandbox as E2BClientSandbox
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT as _MAX_DOWNLOAD_SIZE
+from deerflow.sandbox.download import download_limit, read_download_stream
 from deerflow.sandbox.read_file_contract import split_file_lines
 from deerflow.sandbox.remote_list_dir import parse_remote_list_dir_output, remote_list_dir_command
 from deerflow.sandbox.remote_search import parse_remote_search_output, remote_search_command
@@ -22,7 +24,6 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024  # 100 MB
 
 # Where DeerFlow's ``/mnt/user-data`` virtual prefix is materialised inside
 # the e2b sandbox.  e2b code-interpreter templates default to ``/home/user``
@@ -251,7 +252,8 @@ class E2BSandbox(Sandbox):
             logger.error("Failed to read file %s in e2b sandbox: %s", resolved, e)
             return f"Error: {e}"
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
+        limit = download_limit(max_bytes, _MAX_DOWNLOAD_SIZE)
         normalised = path.replace("\\", "/")
         for segment in normalised.split("/"):
             if segment == "..":
@@ -283,7 +285,9 @@ class E2BSandbox(Sandbox):
                 raise RuntimeError("sandbox client has been closed")
             try:
                 data = client.files.read(resolved, format="stream")
-            except TypeError:
+            except TypeError as error:
+                if max_bytes is not None:
+                    raise OSError("Bounded downloads require E2B streaming support") from error
                 try:
                     data = client.files.read(resolved, format="bytes")
                 except Exception as e:
@@ -293,58 +297,39 @@ class E2BSandbox(Sandbox):
                 logger.error("Failed to download file %s from e2b sandbox: %s", resolved, e)
                 raise OSError(f"Failed to download file '{path}' from sandbox: {e}") from e
 
+        if max_bytes is not None and isinstance(data, (bytes, bytearray, str)):
+            raise OSError("Bounded downloads require an E2B byte stream")
+
         if data is None:
             return b""
 
         # Buffered fallbacks (bytes/bytearray/str): apply the cap up front so
         # we still refuse oversize payloads even on this path.
         if isinstance(data, (bytes, bytearray)):
-            if len(data) > _MAX_DOWNLOAD_SIZE:
+            if len(data) > limit:
                 raise OSError(
                     errno.EFBIG,
-                    f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes",
+                    f"File exceeds maximum download size of {limit} bytes",
                     path,
                 )
             return bytes(data)
         if isinstance(data, str):
             encoded = data.encode("utf-8")
-            if len(encoded) > _MAX_DOWNLOAD_SIZE:
+            if len(encoded) > limit:
                 raise OSError(
                     errno.EFBIG,
-                    f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes",
+                    f"File exceeds maximum download size of {limit} bytes",
                     path,
                 )
             return encoded
 
-        chunks: list[bytes] = []
-        total = 0
-        close = getattr(data, "close", None)
         try:
-            try:
-                for chunk in data:
-                    if not chunk:
-                        continue
-                    chunk_bytes = chunk if isinstance(chunk, bytes) else bytes(chunk)
-                    total += len(chunk_bytes)
-                    if total > _MAX_DOWNLOAD_SIZE:
-                        raise OSError(
-                            errno.EFBIG,
-                            f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes",
-                            path,
-                        )
-                    chunks.append(chunk_bytes)
-            except OSError:
-                raise
-            except Exception as e:
-                logger.error("Failed to stream file %s from e2b sandbox: %s", resolved, e)
-                raise OSError(f"Failed to download file '{path}' from sandbox: {e}") from e
-        finally:
-            if callable(close):
-                try:
-                    close()
-                except Exception:
-                    pass
-        return b"".join(chunks)
+            return read_download_stream(data, limit, path)
+        except OSError:
+            raise
+        except Exception as e:
+            logger.error("Failed to stream file %s from e2b sandbox: %s", resolved, e)
+            raise OSError(f"Failed to download file '{path}' from sandbox: {e}") from e
 
     def list_dir(self, path: str, max_depth: int = 2) -> list[str]:
         resolved = self._resolve_path(path)

backend/packages/harness/deerflow/community/opensandbox/sandbox.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import errno
 import logging
 import posixpath
 import re
@@ -12,6 +11,8 @@ from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT as _MAX_DOWNLOAD_SIZE
+from deerflow.sandbox.download import download_limit, read_download_stream
 from deerflow.sandbox.read_file_contract import split_file_lines
 from deerflow.sandbox.remote_list_dir import parse_remote_list_dir_output, remote_list_dir_command
 from deerflow.sandbox.remote_search import parse_remote_search_output, remote_search_command
@@ -27,7 +28,6 @@ logger = logging.getLogger(__name__)
 
 _TERMINAL_ERROR_NAMES = frozenset({"SandboxUnhealthyException"})
 _COMMAND_TTL_GRACE = timedelta(seconds=30)
-_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024
 
 
 def _exception_chain(error: BaseException):
@@ -298,24 +298,12 @@ class OpenSandboxSandbox(Sandbox):
         resolved = self._resolve_path(path)
         self._file_op(lambda files: files.write_file(resolved, content, mode=644))
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
+        limit = download_limit(max_bytes, _MAX_DOWNLOAD_SIZE)
         resolved = self._resolve_download_path(path)
 
         def read_bounded(files) -> bytes:
-            chunks: list[bytes] = []
-            total = 0
-            stream = files.read_bytes_stream(resolved)
-            try:
-                for chunk in stream:
-                    total += len(chunk)
-                    if total > _MAX_DOWNLOAD_SIZE:
-                        raise OSError(errno.EFBIG, f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", path)
-                    chunks.append(chunk)
-            finally:
-                close = getattr(stream, "close", None)
-                if callable(close):
-                    close()
-            return b"".join(chunks)
+            return read_download_stream(files.read_bytes_stream(resolved), limit, path)
 
         try:
             return self._file_op(read_bounded)

backend/packages/harness/deerflow/community/tenki/sandbox.py

@@ -29,6 +29,8 @@ import threading
 from typing import TYPE_CHECKING, Any, TypeVar
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT as _MAX_DOWNLOAD_SIZE
+from deerflow.sandbox.download import download_limit, read_download_stream
 from deerflow.sandbox.read_file_contract import split_file_lines
 from deerflow.sandbox.remote_list_dir import parse_remote_list_dir_output, remote_list_dir_command
 from deerflow.sandbox.remote_search import parse_remote_search_output, remote_search_command
@@ -45,7 +47,6 @@ T = TypeVar("T")
 
 logger = logging.getLogger(__name__)
 
-_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024  # 100 MB
 # Tenki sandboxes run as the unprivileged ``tenki`` user (HOME=/home/tenki) and
 # ``/mnt`` is root-owned, so DeerFlow's ``/mnt/user-data`` virtual prefix is not
 # writable directly. Like ``community/e2b_sandbox``, file ops are remapped under
@@ -337,7 +338,8 @@ class TenkiSandbox(Sandbox):
                     raise
             self._fs_op(lambda fs: fs.write_stream(resolved, _frames(data)))
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
+        limit = download_limit(max_bytes, _MAX_DOWNLOAD_SIZE)
         normalized = self._guard_traversal(path)
         stripped = normalized.lstrip("/")
         allowed = VIRTUAL_PATH_PREFIX.lstrip("/")
@@ -361,14 +363,8 @@ class TenkiSandbox(Sandbox):
         #
         # The cap is enforced on bytes actually received, so a file that grows
         # mid-transfer still can't exceed it (a stat-then-read check could).
-        chunks: list[bytes] = []
-        total = 0
         try:
-            for chunk in fs.read_stream(resolved):
-                total += len(chunk)
-                if total > _MAX_DOWNLOAD_SIZE:
-                    raise OSError(errno.EFBIG, f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", path)
-                chunks.append(chunk)
+            return read_download_stream(fs.read_stream(resolved), limit, path)
         except OSError as e:
             # Our own EFBIG size-cap is not a session death — let it pass through
             # without evicting. Every other OSError is a real transport failure:
@@ -384,7 +380,6 @@ class TenkiSandbox(Sandbox):
         except Exception as e:
             self._note_failure(e)
             raise OSError(f"cannot read '{path}' from sandbox: {e}") from e
-        return b"".join(chunks)
 
     def list_dir(self, path: str, max_depth: int = 2) -> list[str]:
         resolved = self._resolve_path(path)

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

@@ -10,6 +10,8 @@ human responder. Async claim/deny/decision workers drain through repeated
 cancellation before propagating it.
 
 **Interface**: `Sandbox`: `execute_command(command, env=None)`, additive `execute_command_in_scope(..., scope_id=...)` / `release_command_scope(scope_id)`, `read_file`, `write_file`, `list_dir`, `glob`, `grep`. Scoped hooks default to pass-through without server-side sessions, preserving third-party subclasses. `grep` accepts a text file or directory tree. Per-call `env` injects secrets: `LocalSandbox` merges into the subprocess environment; `AioSandbox` uses fresh `bash.exec(env=...)` sessions. `list_dir`: missing path → `FileNotFoundError`; command/client failure → `OSError`, never `[]` (`ls_tool`: `(empty)`). Remote `glob`/`grep` share it via `sandbox/remote_search.py`: missing root → `FileNotFoundError`, failed search → `OSError`; only a genuine no-match returns `[]`. The parser takes the command's `limit` and reports `truncated` when output passed it, which providers return after Python-side filtering; tools call an empty truncated result incomplete. Remote `grep(glob=...)` scopes like `glob()` (root-relative `path_matches`), never by basename alone. Remotes use `sandbox/remote_list_dir.py`: capture `find`'s status, not `| head`'s (`sh -lc` lacks `pipefail`); missing binary → `OSError`; truncation SIGPIPE → success.
+**Bounded downloads**: `sandbox/download.py` owns `DEFAULT_DOWNLOAD_LIMIT` (100 MiB) and `MAX_IMAGE_BYTES` (20 MiB), shared by providers and vision readers. `download_file(path, max_bytes=...)` enforces the smaller of the caller and provider limits during reading. Streaming readers close on every exit and discard the first overflowing chunk without consuming later chunks; SDK chunk sizing still determines over-read. Local/BoxLite readers probe one extra byte. Oversize is `OSError(EFBIG)`, never a successful prefix. Explicit E2B limits require streaming; custom providers must implement the keyword for vision callers, which fail closed without an unbounded retry. AIO holds its transfer lock through stream cleanup so `close()` cannot close the shared HTTP client mid-read. Keep cancellation draining before lease release. Tests: `test_image_download_limit.py` and image cancellation tests.
+
 **Provider Pattern**: `SandboxProvider`: `acquire`, `acquire_async`, `get`, `get_scoped`, `release`. `get_scoped` checks the identity-scoped cache without blocking; a miss uses canonical `acquire(user_id, thread_id)`, never a checkpoint id. AIO reads active identity maps. Async hooks keep Docker creation/discovery, cross-process locking, readiness checks, and release off-loop. Set `supports_agent_skill_isolation=True` only if every tool enforces explicit lead Agent policy: prepared thread-root bind mounts plus upload `sync_agent_skills`. Host-backed providers report false when enabled shells bypass path mappings; middleware rejects them before explicit-policy acquire.
 **Shared components** (RFC #4741): remote IDs use `derive_sandbox_scope_token` (`sandbox/identity.py`); preserve its keyword-only SHA-256/16-hex contract to avoid orphaning containers. `AcquireSerializer` (`sandbox/acquire_serialization.py`) serializes selected acquire/release transitions with a bounded, refcounted per-key `threading.Lock` table and dedicated bounded executor (no event-loop/default-executor blocking). Workers own cancellation cleanup without waiting for cancelled tasks to resume; provider `shutdown()`/`reset()` calls idempotent `close()`. Keys: AIO `(user_id, thread_id)`, E2B `(user_id, thread_id, skills_root)`, BoxLite/Tenki/OpenSandbox derived id. Random-UUID `thread_id=None` acquires bypass serialization.
 **Execution leases** (`sandbox/lease.py`, #5128): cross-instance ownership selects which Gateway may reap a container; process-local `SandboxLeaseManager` tracks a client's concurrent lead, subagent, Gateway-request, and channel-upload users. Runs have ephemeral owners. Restore persisted sandboxes only after `get_scoped(user_id, thread_id)` succeeds; otherwise canonical `acquire` replaces the id. Last holder performs pending `SandboxProvider.release`. External run/state boundaries reject server-owned `sandbox`. Outer lifecycle fences repeat idempotent release after each graph/tool/request batch; per-tool terminal `Command` wrappers never release while siblings run. Fork-restored children and upload syncs hold without requesting release: server-created forks may borrow the parent client; ordinary checkpoint ids stay identity-scoped. Serialize persisted lookup and retention by `(user_id, thread_id)`; stale or mismatched bindings reacquire, and a post-acquire miss rolls back then raises. Provider I/O runs outside the metadata lock. Repeated cancellation cannot interrupt acquire/rollback/release reconciliation or let `to_thread` work outlive its execution/request holder; failures are logged without replacing the original cancellation. Lease/scope context IDs are server-owned: Gateway/worker scrub caller values; only internal subagents assign task IDs. Managers key by provider object identity (not hash/equality), so unhashable custom providers work. Subagent owners double as `sandbox_command_scope_id`: AIO gives each scope an ordered persistent shell, replaces it after `ErrorObservation` or a missing-session 404, and cleans it on lease release. Revalidate registry identity after each scope-lock wait so queued commands cannot resurrect released sessions. Env-bearing commands use fresh `bash.exec` sessions so secrets do not persist.

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

@@ -0,0 +1,39 @@
+"""Byte budgets shared by sandbox binary downloads."""
+
+import errno
+from collections.abc import Iterable
+from contextlib import suppress
+
+DEFAULT_DOWNLOAD_LIMIT = 100 * 1024 * 1024
+MAX_IMAGE_BYTES = 20 * 1024 * 1024
+
+
+def download_limit(max_bytes: int | None, provider_limit: int) -> int:
+    """A caller may tighten, but never raise, the provider's download ceiling."""
+    if max_bytes is None:
+        return provider_limit
+    if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes < 0:
+        raise ValueError("max_bytes must be a non-negative integer or None")
+    return min(max_bytes, provider_limit)
+
+
+def read_download_stream(stream: Iterable[bytes], limit: int, path: str) -> bytes:
+    """Stop at the first over-budget chunk and close even on interruption.
+
+    The transport owns chunk sizing; do not retain the first overflowing chunk
+    or consume any later chunks. Cleanup must not replace the original failure.
+    """
+    chunks: list[bytes] = []
+    total = 0
+    try:
+        for chunk in stream:
+            total += len(chunk)
+            if total > limit:
+                raise OSError(errno.EFBIG, f"File exceeds maximum download size of {limit} bytes", path)
+            chunks.append(chunk)
+        return b"".join(chunks)
+    finally:
+        close = getattr(stream, "close", None)
+        if callable(close):
+            with suppress(Exception):
+                close()

backend/packages/harness/deerflow/sandbox/local/local_sandbox.py

@@ -14,6 +14,7 @@ from pathlib import Path
 from typing import NamedTuple
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT, download_limit
 from deerflow.sandbox.env_policy import build_sandbox_env
 from deerflow.sandbox.local.list_dir import list_dir
 from deerflow.sandbox.path_patterns import replace_output_path_matches
@@ -921,7 +922,8 @@ class LocalSandbox(Sandbox):
             # Re-raise with the original path for clearer error messages, hiding internal resolved paths
             raise type(e)(e.errno, e.strerror, path) from None
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
+        max_download_size = download_limit(max_bytes, DEFAULT_DOWNLOAD_LIMIT)
         normalised = path.replace("\\", "/")
         stripped_path = normalised.lstrip("/")
         allowed_prefix = VIRTUAL_PATH_PREFIX.lstrip("/")
@@ -930,15 +932,15 @@ class LocalSandbox(Sandbox):
             raise PermissionError(errno.EACCES, f"Access denied: path must be under '{VIRTUAL_PATH_PREFIX}'", path)
 
         resolved_path = self._resolve_path(path)
-        max_download_size = 100 * 1024 * 1024
         try:
             file_size = os.path.getsize(resolved_path)
             if file_size > max_download_size:
                 raise OSError(errno.EFBIG, f"File exceeds maximum download size of {max_download_size} bytes", path)
-            # TOCTOU note: the file could grow between getsize() and read(); accepted
-            # tradeoff since this is a controlled sandbox environment.
             with open(resolved_path, "rb") as f:
-                return f.read()
+                data = f.read(max_download_size + 1)
+            if len(data) > max_download_size:
+                raise OSError(errno.EFBIG, f"File exceeds maximum download size of {max_download_size} bytes", path)
+            return data
         except OSError as e:
             # Re-raise with the original path for clearer error messages, hiding internal resolved paths
             raise type(e)(e.errno, e.strerror, path) from None

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

@@ -162,19 +162,29 @@ class Sandbox(ABC):
         pass
 
     @abstractmethod
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         """Download the binary content of a file.
 
         Args:
             path: The absolute path of the file to download.
+            max_bytes: Optional non-negative byte ceiling, capped by the
+                provider's own limit. Enforce it during reading, not after an
+                unbounded SDK read. Streaming providers stop on the first
+                overflowing chunk and close the stream on every exit; file and
+                exec readers may probe one extra byte. Never return a truncated
+                prefix as success. None preserves the provider's default cap.
+                Custom providers must support this keyword to serve image
+                reads; callers must not retry without it on TypeError.
 
         Returns:
             Raw file bytes.
 
         Raises:
+            ValueError: If max_bytes is not a non-negative integer or None.
             PermissionError: If path traversal is detected or the path is outside
                 the allowed virtual prefix.
-            OSError: If the file cannot be read or does not exist.  Both local
+            OSError: If the file cannot be read or does not exist, or with
+                errno.EFBIG if it exceeds the effective byte limit. Both local
                 and remote implementations must raise ``OSError`` so callers
                 have a single exception type to handle.
         """

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

@@ -58,7 +58,7 @@ Only standalone tool discovery without a model falls back to the base profile.
 3. **Built-in tools**:
    - `present_files` - Make output files visible to user (only `/mnt/user-data/outputs`); virtual paths use `resolve_runtime_user_id(runtime)` so validation resolves the same user-scoped outputs directory established by `ThreadDataMiddleware`
    - `ask_clarification` - Request clarification (intercepted by ClarificationMiddleware, which preserves text fallback and adds `artifact.human_input` for Web UI Human Input Cards). Beyond free text and single choice, the request-side v2 protocol supports `fields` (structured form card collecting several values at once; field types: text/textarea/number/select/multi_select/checkbox/date, validated and normalized server-side in the middleware — invalid entries are dropped, unknown types degrade to `text`; a standalone multi-select question is a one-field form). Replies stay on the v1 response protocol (`text`/`option`): the form card submits a readable text summary
-   - `view_image` - Read image bytes for vision-capable models; both sync and async entry points require `sandbox:execute` before any host or sandbox read. With blob storage enabled, validated bytes are persisted as `kind="viewed-image"` and the checkpoint carries a JSON-safe ref beside the compatibility host path; store initialization/write failures return a generic tool error. Blob-backed model reads validate the ref against the separate image metadata before use and fall through to validated compatibility paths on any store failure. Live sandbox bytes remain the fallback for the same sandbox generation, replacement-sandbox recovery uses only SHA-256-verified synchronized host bytes, and async tool invocation drains blocking reads before cancellation may release the sandbox lease
+   - `view_image` - Pass the 20 MiB caller limit to sandbox downloads (see [sandbox download contract](../sandbox/AGENTS.md)). Read image bytes for vision-capable models; both sync and async entry points require `sandbox:execute` before any host or sandbox read. With blob storage enabled, validated bytes are persisted as `kind="viewed-image"` and the checkpoint carries a JSON-safe ref beside the compatibility host path; store initialization/write failures return a generic tool error. Blob-backed model reads validate the ref against the separate image metadata before use and fall through to validated compatibility paths on any store failure. Live sandbox bytes remain the fallback for the same sandbox generation, replacement-sandbox recovery uses only SHA-256-verified synchronized host bytes, and async tool invocation drains blocking reads before cancellation may release the sandbox lease
    - `setup_agent` - Bootstrap-only: persist a custom agent's `SOUL.md` and `config.yaml`. Re-bootstrapping preserves the owner's existing `display_name`. Bound only when `is_bootstrap=True`. The default agent's global `SOUL.md` is published through a staged temporary plus `os.replace`, like the agent store's per-agent write, so an interrupted setup keeps the previous persona file instead of leaving it truncated.
    - `update_agent` - Custom-agent-only: persist self-updates to the current agent's `SOUL.md` / `config.yaml` from inside a normal chat (partial update + atomic write). Bound when `agent_name` is set and `is_bootstrap=False`.
 4. **Subagent tool** (if enabled):

backend/packages/harness/deerflow/tools/builtins/view_image_tool.py

@@ -12,6 +12,7 @@ from langgraph.types import Command
 
 from deerflow.agents.thread_state import ThreadDataState, ViewedImageData
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.download import MAX_IMAGE_BYTES as _MAX_IMAGE_BYTES
 from deerflow.tools.types import Runtime
 
 logger = logging.getLogger(__name__)
@@ -22,7 +23,6 @@ _ALLOWED_IMAGE_VIRTUAL_ROOTS = (
     f"{VIRTUAL_PATH_PREFIX}/outputs",
 )
 _ALLOWED_IMAGE_VIRTUAL_ROOTS_TEXT = ", ".join(_ALLOWED_IMAGE_VIRTUAL_ROOTS)
-_MAX_IMAGE_BYTES = 20 * 1024 * 1024
 _EXTENSION_TO_MIME = {
     ".jpg": "image/jpeg",
     ".jpeg": "image/jpeg",
@@ -200,7 +200,7 @@ def _view_image_authorized(runtime: Runtime, image_path: str, tool_call_id: str)
 
     if sandbox is not None:
         try:
-            image_data = sandbox.download_file(image_path)
+            image_data = sandbox.download_file(image_path, max_bytes=_MAX_IMAGE_BYTES)
             read_source_sandbox_id = sandbox_id
         except IsADirectoryError:
             return Command(

backend/tests/test_aio_sandbox.py

@@ -2901,6 +2901,77 @@ class TestDownloadFile:
 class TestClose:
     """Verify AioSandbox.close() tears down the host-side HTTP client (#2872)."""
 
+    def test_close_waits_for_download_stream_cleanup(self, sandbox):
+        reading = threading.Event()
+        allow_read = threading.Event()
+        closing = threading.Event()
+        stream_closed = threading.Event()
+        close_observed_cleanup = []
+        outcomes = []
+
+        class ObservedLock:
+            def __init__(self):
+                self.lock = threading.Lock()
+                self.waited_for_reader = False
+
+            def __enter__(self):
+                if threading.current_thread().name == "download-close":
+                    if self.lock.acquire(blocking=False):
+                        closing.set()
+                        return
+                    self.waited_for_reader = True
+                    closing.set()
+                self.lock.acquire()
+
+            def __exit__(self, *args):
+                self.lock.release()
+
+        def stream():
+            try:
+                reading.set()
+                assert allow_read.wait(timeout=5)
+                yield b"data"
+            finally:
+                stream_closed.set()
+
+        def close_client():
+            close_observed_cleanup.append(stream_closed.is_set())
+
+        http_client = MagicMock(spec=["close"])
+        http_client.close.side_effect = close_client
+        sandbox._client._client_wrapper = SimpleNamespace(httpx_client=SimpleNamespace(httpx_client=http_client))
+        sandbox._client.file.download_file.return_value = stream()
+        sandbox._lock = ObservedLock()
+
+        def download():
+            try:
+                outcomes.append(sandbox.download_file("/mnt/user-data/outputs/image.png", max_bytes=4))
+            except BaseException as error:
+                outcomes.append(error)
+
+        reader = threading.Thread(target=download)
+        closer = threading.Thread(target=sandbox.close, name="download-close")
+        reader.start()
+        try:
+            assert reading.wait(timeout=5)
+            closer.start()
+            assert closing.wait(timeout=5)
+            assert sandbox._lock.waited_for_reader
+            http_client.close.assert_not_called()
+            assert not stream_closed.is_set()
+        finally:
+            allow_read.set()
+            reader.join(timeout=5)
+            if closer.ident is not None:
+                closer.join(timeout=5)
+        assert not reader.is_alive()
+        assert not closer.is_alive()
+        assert outcomes == [b"data"]
+        assert stream_closed.is_set()
+        assert close_observed_cleanup == [True]
+        http_client.close.assert_called_once_with()
+        assert sandbox._client is None
+
     def test_close_calls_real_nested_httpx_client(self, sandbox):
         """close() must close the real httpx.Client at the bottom of the chain.
 

backend/tests/test_image_download_limit.py

@@ -0,0 +1,237 @@
+"""Caller limits must reach the provider before image bytes are buffered."""
+
+import errno
+import io
+import os
+import shlex
+import shutil
+import subprocess
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
+from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox
+from deerflow.community.boxlite.box import BoxliteBox
+from deerflow.community.e2b_sandbox.e2b_sandbox import E2BSandbox
+from deerflow.community.opensandbox.sandbox import OpenSandboxSandbox
+from deerflow.community.tenki.sandbox import TenkiSandbox
+from deerflow.sandbox.local.local_sandbox import LocalSandbox
+from deerflow.tools.builtins.view_image_tool import _MAX_IMAGE_BYTES, _view_image_authorized
+
+PATH = "/mnt/user-data/outputs/image.png"
+
+
+class Stream:
+    def __init__(self, chunks, error=None):
+        self.chunks = iter(chunks)
+        self.error = error
+        self.consumed = 0
+        self.closed = False
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        chunk = next(self.chunks)
+        self.consumed += len(chunk)
+        if self.error:
+            raise self.error
+        return chunk
+
+    def close(self):
+        self.closed = True
+
+
+def provider(kind, stream):
+    if kind == "e2b":
+        return E2BSandbox("remote", SimpleNamespace(files=SimpleNamespace(read=Mock(return_value=stream))))
+    if kind == "aio":
+        box = object.__new__(AioSandbox)
+        box._lock = threading.Lock()
+        box._client = SimpleNamespace(file=SimpleNamespace(download_file=Mock(return_value=stream)))
+        return box
+    if kind == "opensandbox":
+        box = object.__new__(OpenSandboxSandbox)
+        files = SimpleNamespace(read_bytes_stream=Mock(return_value=stream))
+        box._file_op = lambda operation: operation(files)
+        return box
+    box = object.__new__(TenkiSandbox)
+    box._lock = threading.Lock()
+    box._closed = False
+    box._sandbox = SimpleNamespace(fs=SimpleNamespace(read_stream=Mock(return_value=stream)))
+    box._resolve_path = lambda path: path
+    box._note_failure = Mock()
+    return box
+
+
+@pytest.mark.parametrize("kind", ["aio", "e2b", "opensandbox", "tenki"])
+@pytest.mark.parametrize("size", [0, 4, 5, 9])
+def test_provider_limit_stops_stream_and_closes(kind, size):
+    stream = Stream([b"x"] * size)
+    box = provider(kind, stream)
+    if size > 4:
+        with pytest.raises(OSError) as error:
+            box.download_file(PATH, max_bytes=4)
+        assert error.value.errno == errno.EFBIG
+    else:
+        assert box.download_file(PATH, max_bytes=4) == b"x" * size
+    assert stream.consumed == min(size, 5)
+    assert stream.closed
+
+
+@pytest.mark.parametrize("kind", ["aio", "e2b", "opensandbox", "tenki"])
+@pytest.mark.parametrize("error", [OSError("transport"), KeyboardInterrupt()])
+def test_provider_closes_stream_on_error_or_interruption(kind, error):
+    stream = Stream([b"x"], error)
+    with pytest.raises(type(error)):
+        provider(kind, stream).download_file(PATH, max_bytes=4)
+    assert stream.closed
+
+
+def test_tool_rejects_21_mib_without_consuming_the_complete_stream(tmp_path, monkeypatch):
+    stream = Stream([b"x" * 65536] * (21 * 16))
+    box = provider("aio", stream)
+    thread_data = {f"{name}_path": str(tmp_path / name) for name in ("workspace", "uploads", "outputs")}
+    monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: SimpleNamespace(get=lambda _: box))
+    runtime = SimpleNamespace(state={"thread_data": thread_data, "sandbox": {"sandbox_id": "remote"}}, context={"thread_id": "thread"}, config={})
+    result = _view_image_authorized(runtime, PATH, "call")
+    assert "Error" in result.update["messages"][0].content
+    assert "viewed_images" not in result.update
+    assert stream.consumed <= _MAX_IMAGE_BYTES + 65536
+    assert stream.closed
+
+
+def test_e2b_bounded_read_never_retries_buffered_sdk():
+    read = Mock(side_effect=TypeError("stream unavailable"))
+    box = E2BSandbox("remote", SimpleNamespace(files=SimpleNamespace(read=read)))
+    with pytest.raises(OSError):
+        box.download_file(PATH, max_bytes=4)
+    assert read.call_count == 1
+    assert read.call_args.kwargs == {"format": "stream"}
+
+
+def test_local_growth_after_stat_is_still_bounded(monkeypatch):
+    box = object.__new__(LocalSandbox)
+    box._resolve_path = lambda path: path
+    stream = io.BytesIO(b"x" * 100)
+    requests = []
+
+    class Reader(io.BytesIO):
+        def read(self, size=-1):
+            requests.append(size)
+            return stream.read(size)
+
+    monkeypatch.setattr("os.path.getsize", lambda _: 4)
+    monkeypatch.setattr("builtins.open", lambda *args, **kwargs: Reader())
+    with pytest.raises(OSError) as error:
+        box.download_file(PATH, max_bytes=4)
+    assert error.value.errno == errno.EFBIG
+    assert requests == [5]
+    assert stream.tell() == 5
+
+
+@pytest.mark.parametrize("kind", ["aio", "e2b", "opensandbox", "tenki", "local", "boxlite"])
+@pytest.mark.parametrize("limit", [-1, True, 1.5])
+def test_invalid_limit_rejected_before_io(kind, limit):
+    box = object.__new__({"local": LocalSandbox, "boxlite": BoxliteBox}[kind]) if kind in {"local", "boxlite"} else provider(kind, Stream([]))
+    with pytest.raises(ValueError):
+        box.download_file(PATH, max_bytes=limit)
+
+
+@pytest.mark.parametrize("size", [0, 4, 5, 100])
+def test_boxlite_bounds_actual_shell_read(tmp_path, size):
+    git = shutil.which("git")
+    shell = str(Path(git).parent.parent / "bin" / "bash.exe") if os.name == "nt" and git else shutil.which("sh")
+    if not shell or not Path(shell).exists():
+        pytest.skip("POSIX shell is required for the BoxLite transport probe")
+    image = tmp_path / "image ; quoted.png"
+    image.write_bytes(b"x" * size)
+    box = object.__new__(BoxliteBox)
+    outputs = []
+
+    def run(script):
+        script = script.replace(shlex.quote(PATH), shlex.quote(image.as_posix()))
+        if os.name == "nt":
+            script = "export PATH=/usr/bin:/bin\n" + script
+        result = subprocess.run([shell, "-c", script], capture_output=True, text=True, timeout=10)
+        outputs.append(result.stdout)
+        return SimpleNamespace(exit_code=result.returncode, stdout=result.stdout, stderr=result.stderr)
+
+    box._sh = run
+    if size > 4:
+        with pytest.raises(OSError) as error:
+            box.download_file(PATH, max_bytes=4)
+        assert error.value.errno == errno.EFBIG
+    else:
+        assert box.download_file(PATH, max_bytes=4) == b"x" * size
+    assert len(outputs[0]) < 80
+    image.unlink()
+    with pytest.raises(OSError):
+        box.download_file(PATH, max_bytes=4)
+
+
+@pytest.mark.parametrize("source", ["remote", "old-remote"])
+def test_model_injection_limits_remote_download(tmp_path, monkeypatch, source):
+    stream = Stream([b"x" * 65536] * (21 * 16))
+    box = provider("aio", stream)
+    monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: SimpleNamespace(get=lambda _: box))
+    result = ViewImageMiddleware._read_image_as_data_url({"sandbox": {"sandbox_id": "remote"}}, PATH, "", "image/png", 1, "digest", source)
+    assert result is None
+    assert stream.consumed <= _MAX_IMAGE_BYTES + 65536
+    assert stream.closed
+
+
+def test_custom_provider_without_limit_fails_closed(tmp_path, monkeypatch):
+    calls = []
+
+    class Legacy:
+        def download_file(self, path):
+            calls.append(path)
+            return b"legacy"
+
+    monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: SimpleNamespace(get=lambda _: Legacy()))
+    runtime = SimpleNamespace(
+        state={"thread_data": {f"{name}_path": str(tmp_path / name) for name in ("workspace", "uploads", "outputs")}, "sandbox": {"sandbox_id": "remote"}},
+        context={"thread_id": "thread"},
+        config={},
+    )
+    result = _view_image_authorized(runtime, PATH, "call")
+    assert "Error" in result.update["messages"][0].content
+    assert calls == []
+
+
+@pytest.mark.parametrize("kind", ["aio", "e2b", "opensandbox", "tenki"])
+@pytest.mark.parametrize("max_bytes", [None, 100 * 1024 * 1024 + 1])
+def test_caller_cannot_raise_default_provider_ceiling(kind, max_bytes):
+    from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT
+
+    class OversizedChunk:
+        def __len__(self):
+            return DEFAULT_DOWNLOAD_LIMIT + 1
+
+    stream = Stream([OversizedChunk(), b"unread"])
+    with pytest.raises(OSError) as error:
+        provider(kind, stream).download_file(PATH, max_bytes=max_bytes)
+    assert error.value.errno == errno.EFBIG
+    assert str(DEFAULT_DOWNLOAD_LIMIT) in str(error.value)
+    assert stream.consumed == DEFAULT_DOWNLOAD_LIMIT + 1
+    assert stream.closed
+
+
+@pytest.mark.parametrize("max_bytes", [None, 100 * 1024 * 1024 + 1])
+def test_local_caller_cannot_raise_default_provider_ceiling(monkeypatch, max_bytes):
+    from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT
+
+    box = object.__new__(LocalSandbox)
+    box._resolve_path = lambda path: path
+    monkeypatch.setattr("os.path.getsize", lambda _: DEFAULT_DOWNLOAD_LIMIT + 1)
+    read = Mock()
+    monkeypatch.setattr("builtins.open", read)
+    with pytest.raises(OSError) as error:
+        box.download_file(PATH, max_bytes=max_bytes)
+    assert error.value.errno == errno.EFBIG
+    read.assert_not_called()

backend/tests/test_view_image_provider_error_classification.py

@@ -26,7 +26,7 @@ class _HttpMissingSandbox:
     def __init__(self) -> None:
         self.downloads: list[str] = []
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         self.downloads.append(path)
         try:
             raise _HttpNotFoundError("not found")

backend/tests/test_view_image_remote_sandbox.py

@@ -31,7 +31,7 @@ class _RemoteSandbox:
         self.downloads: list[str] = []
         self.released_scopes: list[str] = []
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         self.downloads.append(path)
         return self.image_bytes
 
@@ -40,13 +40,13 @@ class _RemoteSandbox:
 
 
 class _FailingRemoteSandbox(_RemoteSandbox):
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         self.downloads.append(path)
         raise OSError("remote download failed")
 
 
 class _MissingRemoteSandbox(_RemoteSandbox):
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         self.downloads.append(path)
         try:
             raise _E2BFileNotFound("not found")
@@ -60,7 +60,7 @@ class _BlockingRemoteSandbox(_RemoteSandbox):
         self.download_started = threading.Event()
         self.allow_download = threading.Event()
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         self.downloads.append(path)
         self.download_started.set()
         assert self.allow_download.wait(timeout=5)

backend/tests/test_view_image_tool_cancellation.py

@@ -21,7 +21,7 @@ class _BlockingRemoteSandbox:
         self.downloads: list[str] = []
         self.released_scopes: list[str] = []
 
-    def download_file(self, path: str) -> bytes:
+    def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes:
         self.downloads.append(path)
         self.download_started.set()
         assert self.allow_download.wait(timeout=5)