Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 12 additions & 20 deletions libs/deepagents/deepagents/backends/sandbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,9 +380,6 @@ def write(
) -> WriteResult:
"""Create a new file, failing if it already exists.

Runs a small preflight command to check existence and create parent
directories, then transfers content via `upload_files()`.

Args:
file_path: Absolute path for the new file.
content: UTF-8 text content to write.
Expand All @@ -391,31 +388,23 @@ def write(
`WriteResult` with `path` on success or `error` on failure.
"""
# Existence check + mkdir. There is a TOCTOU window between this check
# and the upload below a concurrent process could create the file in
# and the upload below - a concurrent process could create the file in
# between. This is an inherent limitation of splitting the operation;
# the risk is minimal in single-agent sandbox environments.
path_b64 = base64.b64encode(file_path.encode("utf-8")).decode("ascii")
check_cmd = _WRITE_CHECK_TEMPLATE.format(path_b64=path_b64)
try:
result = self.execute(check_cmd)
except Exception as exc: # noqa: BLE001 # defense-in-depth for buggy subclass execute()
msg = f"Failed to write file '{file_path}': {exc}"
return WriteResult(error=msg)

result = self.execute(check_cmd)
if result.exit_code != 0 or "Error:" in result.output:
error_msg = result.output.strip() or f"Failed to write file '{file_path}'"
return WriteResult(error=error_msg)

# Transfer content via upload_files()
try:
responses = self.upload_files([(file_path, content.encode("utf-8"))])
except Exception as exc: # noqa: BLE001 # defense-in-depth for buggy subclass upload_files()
msg = f"Failed to write file '{file_path}': {exc}"
return WriteResult(error=msg)
responses = self.upload_files([(file_path, content.encode("utf-8"))])
if not responses:
return WriteResult(error=f"Failed to write file '{file_path}': upload returned no response")
if responses[0].error:
return WriteResult(error=f"Failed to write file '{file_path}': {responses[0].error}")
# An unreachable condition was reached
msg = f"Responses was expected to return 1 result, but it returned {len(responses)} with type {type(responses)}"
raise AssertionError(msg)
response = responses[0]
if response.error:
return WriteResult(error=f"Failed to write file '{file_path}': {response.error}")

return WriteResult(path=file_path, files_update=None)

Expand Down Expand Up @@ -676,6 +665,9 @@ def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadRespons

Implementations must support partial success - catch exceptions per-file
and return errors in `FileUploadResponse` objects rather than raising.

Upload files is responsible for ensuring that the parent path exists
(if user permissions allow the user to write to the given directory)
"""

@abstractmethod
Expand Down
11 changes: 0 additions & 11 deletions libs/deepagents/tests/unit_tests/backends/test_sandbox_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -686,17 +686,6 @@ def test_sandbox_write_returns_correct_result_on_success() -> None:
assert result.files_update is None


def test_sandbox_write_returns_error_on_empty_upload_response() -> None:
"""Test that write() handles upload_files returning an empty list."""
sandbox = MockSandbox()
sandbox.upload_files = lambda _files: [] # type: ignore[assignment]

result = sandbox.write("/test/file.txt", "content")

assert result.error is not None
assert "no response" in result.error


def test_sandbox_edit_upload_returns_error_on_empty_upload_response() -> None:
"""Test that upload-path edit handles upload_files returning empty list."""
sandbox = MockSandbox()
Expand Down
Loading