langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-22 01:35:28 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

fix(sdk): route BaseSandbox async helpers through aexecute (#3996)

Browse Source 
Fixes #665

## Problem

`BaseSandbox`'s async helpers (`als`, `aread`, `agrep`, `aglob`,
`awrite`, `aedit`) previously wrapped the corresponding sync methods in
`asyncio.to_thread`, meaning overriding `aexecute` with a native async
implementation (e.g. wrapping an async SDK like Daytona's
`AsyncSandbox`) had no effect on those helpers, they still ran the sync
path in a thread pool. Subclasses were forced to override all 9
file-operation methods to get native async behavior.

## Fix

This PR overrides all six async helpers in `BaseSandbox` to call
`aexecute`/`aupload_files` directly. Subclasses now only need to
override those three core methods. To avoid duplicating command-building
and output-parsing logic between sync and async paths, each method's
pure logic was factored into private helpers shared by both. Sync
methods and their behavior are completely unchanged.

## Tests

Added a `NativeAsyncSandbox` test fixture where `exes `RuntimeError`, if
any async helper falls backto the sync path, the test fails immediately.
Seven new `@pytest.mark.asyncio` tests cover each overridden method
(`als`, `aread`, `agrep`, `aglob`, `awrite`, inline `aedit`, and
large-payload `aedit` via upload), asserting that
`aexecute`/`aupload_files` are
called the expected number of times and results are tests continue to
pass.

---------

Signed-off-by: Nishitha M <32355027+imnishitha@users.noreply.github.com>
Co-authored-by: Eugene Yurtsev <eugene@langchain.dev>
This commit is contained in:
Nishitha M
2026-06-17 09:41:19 -04:00
committed by GitHub
parent 0412009c54
commit 52dcf1a42c
2 changed files with 546 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/deepagents/deepagents/backends/sandbox.py
+340 -216
View File
@@ -14,6 +14,7 @@ operations are derived from those.
from __future__ import annotations
import asyncio
import base64
import json
import logging
@@ -23,6 +24,7 @@ from abc import ABC, abstractmethod
from typing import Final
from deepagents.backends.protocol import (
ASYNC_GREP_TIMEOUT,
EditResult,
ExecuteResponse,
FileData,
@@ -391,6 +393,196 @@ success or `{{"error": ...}}` on failure.
"""
def _build_ls_cmd(path: str) -> str:
path_b64 = base64.b64encode(path.encode("utf-8")).decode("ascii")
return f"""python3 -c "
import os
import json
import base64
path = base64.b64decode('{path_b64}').decode('utf-8')
try:
with os.scandir(path) as it:
for entry in it:
result = {{
'path': os.path.join(path, entry.name),
'is_dir': entry.is_dir(follow_symlinks=False)
}}
print(json.dumps(result))
except FileNotFoundError:
print(json.dumps({{'error': 'path_not_found'}}))
except NotADirectoryError:
print(json.dumps({{'error': 'not_a_directory'}}))
except PermissionError:
print(json.dumps({{'error': 'permission_denied'}}))
" 2>/dev/null"""
def _parse_ls_output(output: str, path: str) -> LsResult:
file_infos: list[FileInfo] = []
error: str | None = None
for line in output.strip().split("\n"):
if not line:
continue
try:
data = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(data, dict) and "error" in data:
error = data["error"]
continue
file_infos.append({"path": data["path"], "is_dir": data["is_dir"]})
if error is not None:
return LsResult(entries=None, error=f"Path '{path}': {error}")
return LsResult(entries=file_infos)
def _build_read_cmd(file_path: str, offset: int, limit: int) -> str:
file_type = _get_file_type(file_path)
path_b64 = base64.b64encode(file_path.encode("utf-8")).decode("ascii")
# Defensive int coercion in case callers bypass type checking.
return _READ_COMMAND_TEMPLATE.format(
path_b64=path_b64,
file_type=file_type,
offset=int(offset),
limit=int(limit),
)
def _parse_read_output(output: str, file_path: str) -> ReadResult:
output = output.rstrip()
try:
data = json.loads(output)
except (json.JSONDecodeError, ValueError):
detail = output[:200] if output else "(empty)"
return ReadResult(error=f"File '{file_path}': unexpected server response: {detail}")
if not isinstance(data, dict):
detail = output[:200] if output else "(empty)"
return ReadResult(error=f"File '{file_path}': unexpected server response: {detail}")
if "error" in data:
return ReadResult(error=f"File '{file_path}': {data['error']}")
return ReadResult(
file_data=FileData(
content=data["content"],
encoding=data.get("encoding", "utf-8"),
)
)
def _build_write_preflight_cmd(file_path: str) -> str:
path_b64 = base64.b64encode(file_path.encode("utf-8")).decode("ascii")
return _WRITE_CHECK_TEMPLATE.format(path_b64=path_b64)
def _check_preflight_result(result: ExecuteResponse, file_path: str) -> WriteResult | None:
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)
return None
def _build_grep_cmd(pattern: str, path: str | None, glob: str | None) -> str:
search_path = shlex.quote(path or ".")
# `-Z` separates the filename from line data with NUL, so filenames may
# contain `:` without making the output ambiguous.
grep_opts = "-rHnFZ"
glob_pattern = f"--include={shlex.quote(glob)}" if glob else ""
pattern_escaped = shlex.quote(pattern)
return f"grep {grep_opts} {glob_pattern} -e {pattern_escaped} {search_path} 2>/dev/null || true"
def _parse_grep_output(result: ExecuteResponse, path: str | None) -> GrepResult:
output = result.output.rstrip("\n")
if result.exit_code is not None and result.exit_code != 0:
detail = output.strip() if output else f"exit code {result.exit_code}"
return GrepResult(error=f"Path '{path or '.'}': {detail}")
if not output:
return GrepResult(matches=[])
matches: list[GrepMatch] = []
parse_error: str | None = None
for line in output.split("\n"):
# Format is: path\0line_number:text
try:
file_path, rest = line.split("\0", 1)
line_num_str, text = rest.split(":", 1)
matches.append({"path": file_path, "line": int(line_num_str), "text": text})
except ValueError:
parse_error = line
if parse_error is not None and not matches:
return GrepResult(error=f"Path '{path or '.'}': {parse_error}")
return GrepResult(matches=matches)
def _build_glob_cmd(pattern: str, search_path: str) -> str:
pattern_b64 = base64.b64encode(pattern.encode("utf-8")).decode("ascii")
path_b64 = base64.b64encode(search_path.encode("utf-8")).decode("ascii")
return _GLOB_COMMAND_TEMPLATE.format(path_b64=path_b64, pattern_b64=pattern_b64)
def _parse_glob_output(output: str, search_path: str) -> GlobResult:
output = output.strip()
if not output:
return GlobResult(matches=[])
file_infos: list[FileInfo] = []
error: str | None = None
for line in output.split("\n"):
try:
data = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(data, dict) and "error" in data:
error = data["error"]
continue
file_infos.append({"path": data["path"], "is_dir": data["is_dir"]})
if error is not None:
return GlobResult(matches=None, error=f"Path '{search_path}': {error}")
return GlobResult(matches=file_infos)
def _build_edit_inline_cmd(file_path: str, old_string: str, new_string: str, *, replace_all: bool) -> str:
payload = json.dumps({"path": file_path, "old": old_string, "new": new_string, "replace_all": replace_all})
payload_b64 = base64.b64encode(payload.encode("utf-8")).decode("ascii")
return _EDIT_COMMAND_TEMPLATE.format(payload_b64=payload_b64)
def _map_edit_error(error: str, file_path: str, old_string: str) -> EditResult:
"""Map server-side error codes to `EditResult` objects."""
messages: dict[str, str] = {
"file_not_found": f"Error: File '{file_path}' not found",
"permission_denied": f"Error: Permission denied editing file '{file_path}'",
"not_a_file": f"Error: '{file_path}' is not a regular file",
"not_a_text_file": f"Error: File '{file_path}' is not a text file",
"string_not_found": f"Error: String not found in file: '{old_string}'",
"multiple_occurrences": (f"Error: String '{old_string}' appears multiple times. Use replace_all=True to replace all occurrences."),
}
return EditResult(error=messages.get(error, f"Error editing file '{file_path}': {error}"))
def _parse_edit_output(output: str, file_path: str, old_string: str) -> EditResult:
output = output.rstrip()
try:
data = json.loads(output)
except (json.JSONDecodeError, ValueError):
detail = output[:200] if output else "(empty)"
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if not isinstance(data, dict):
detail = output[:200] if output else "(empty)"
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if "error" in data:
return _map_edit_error(data["error"], file_path, old_string)
return EditResult(path=file_path, occurrences=data.get("count", 1))
def _build_edit_tmpfile_cmd(file_path: str, old_tmp: str, new_tmp: str, *, replace_all: bool) -> str:
return _EDIT_TMPFILE_TEMPLATE.format(
old_path_b64=base64.b64encode(old_tmp.encode("utf-8")).decode("ascii"),
new_path_b64=base64.b64encode(new_tmp.encode("utf-8")).decode("ascii"),
target_b64=base64.b64encode(file_path.encode("utf-8")).decode("ascii"),
replace_all=replace_all,
)
class BaseSandbox(SandboxBackendProtocol, ABC):
"""Base sandbox implementation with `execute()` as the core abstract method.
@@ -434,49 +626,13 @@ class BaseSandbox(SandboxBackendProtocol, ABC):
def ls(self, path: str) -> LsResult:
"""Structured listing with file metadata using os.scandir."""
path_b64 = base64.b64encode(path.encode("utf-8")).decode("ascii")
cmd = f"""python3 -c "
import os
import json
import base64
result = self.execute(_build_ls_cmd(path))
return _parse_ls_output(result.output, path)
path = base64.b64decode('{path_b64}').decode('utf-8')
try:
with os.scandir(path) as it:
for entry in it:
result = {{
'path': os.path.join(path, entry.name),
'is_dir': entry.is_dir(follow_symlinks=False)
}}
print(json.dumps(result))
except FileNotFoundError:
print(json.dumps({{'error': 'path_not_found'}}))
except NotADirectoryError:
print(json.dumps({{'error': 'not_a_directory'}}))
except PermissionError:
print(json.dumps({{'error': 'permission_denied'}}))
" 2>/dev/null"""
result = self.execute(cmd)
file_infos: list[FileInfo] = []
error: str | None = None
for line in result.output.strip().split("\n"):
if not line:
continue
try:
data = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(data, dict) and "error" in data:
error = data["error"]
continue
file_infos.append({"path": data["path"], "is_dir": data["is_dir"]})
if error is not None:
return LsResult(entries=None, error=f"Path '{path}': {error}")
return LsResult(entries=file_infos)
async def als(self, path: str) -> LsResult:
"""Async version of `ls`, delegating to `aexecute`."""
result = await self.aexecute(_build_ls_cmd(path))
return _parse_ls_output(result.output, path)
def read(
self,
@@ -509,38 +665,18 @@ except PermissionError:
Returns:
`ReadResult` with `file_data` on success or `error` on failure.
"""
file_type = _get_file_type(file_path)
path_b64 = base64.b64encode(file_path.encode("utf-8")).decode("ascii")
result = self.execute(_build_read_cmd(file_path, offset, limit))
return _parse_read_output(result.output, file_path)
# Defensive int coercion in case callers bypass type checking.
cmd = _READ_COMMAND_TEMPLATE.format(
path_b64=path_b64,
file_type=file_type,
offset=int(offset),
limit=int(limit),
)
result = self.execute(cmd)
output = result.output.rstrip()
try:
data = json.loads(output)
except (json.JSONDecodeError, ValueError):
detail = output[:200] if output else "(empty)"
return ReadResult(error=f"File '{file_path}': unexpected server response: {detail}")
if not isinstance(data, dict):
detail = output[:200] if output else "(empty)"
return ReadResult(error=f"File '{file_path}': unexpected server response: {detail}")
if "error" in data:
return ReadResult(error=f"File '{file_path}': {data['error']}")
return ReadResult(
file_data=FileData(
content=data["content"],
encoding=data.get("encoding", "utf-8"),
)
)
async def aread(
self,
file_path: str,
offset: int = 0,
limit: int = 2000,
) -> ReadResult:
"""Async version of `read`, delegating to `aexecute`."""
result = await self.aexecute(_build_read_cmd(file_path, offset, limit))
return _parse_read_output(result.output, file_path)
def _write_preflight(self, file_path: str) -> WriteResult | None:
"""Run the existence check + parent-directory creation for `write()`.
@@ -559,13 +695,13 @@ except PermissionError:
created); a populated `WriteResult` with `error` set if the
check fails.
"""
path_b64 = base64.b64encode(file_path.encode("utf-8")).decode("ascii")
check_cmd = _WRITE_CHECK_TEMPLATE.format(path_b64=path_b64)
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)
return None
result = self.execute(_build_write_preflight_cmd(file_path))
return _check_preflight_result(result, file_path)
async def _awrite_preflight(self, file_path: str) -> WriteResult | None:
"""Async version of `_write_preflight`, delegating to `aexecute`."""
result = await self.aexecute(_build_write_preflight_cmd(file_path))
return _check_preflight_result(result, file_path)
def write(
self,
@@ -596,6 +732,20 @@ except PermissionError:
return WriteResult(path=file_path)
async def awrite(self, file_path: str, content: str) -> WriteResult:
"""Async version of `write`, delegating to `aexecute` and `aupload_files`."""
preflight_error = await self._awrite_preflight(file_path)
if preflight_error is not None:
return preflight_error
responses = await self.aupload_files([(file_path, content.encode("utf-8"))])
if not responses:
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)
def edit(
self,
file_path: str,
@@ -639,6 +789,19 @@ except PermissionError:
return self._edit_via_upload(file_path, old_string, new_string, replace_all)
async def aedit(
self,
file_path: str,
old_string: str,
new_string: str,
replace_all: bool = False, # noqa: FBT001, FBT002
) -> EditResult:
"""Async version of `edit`, delegating to `aexecute` and `aupload_files`."""
payload_size = len(old_string.encode("utf-8")) + len(new_string.encode("utf-8"))
if payload_size <= _EDIT_INLINE_MAX_BYTES:
return await self._aedit_inline(file_path, old_string, new_string, replace_all)
return await self._aedit_via_upload(file_path, old_string, new_string, replace_all)
def _edit_inline(
self,
file_path: str,
@@ -646,37 +809,20 @@ except PermissionError:
new_string: str,
replace_all: bool, # noqa: FBT001
) -> EditResult:
"""Server-side replace via `execute()` — single round-trip."""
payload = json.dumps(
{
"path": file_path,
"old": old_string,
"new": new_string,
"replace_all": replace_all,
}
)
payload_b64 = base64.b64encode(payload.encode("utf-8")).decode("ascii")
cmd = _EDIT_COMMAND_TEMPLATE.format(payload_b64=payload_b64)
result = self.execute(cmd)
output = result.output.rstrip()
"""Server-side replace via `execute()` (single round-trip)."""
result = self.execute(_build_edit_inline_cmd(file_path, old_string, new_string, replace_all=replace_all))
return _parse_edit_output(result.output, file_path, old_string)
try:
data = json.loads(output)
except (json.JSONDecodeError, ValueError):
detail = output[:200] if output else "(empty)"
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if not isinstance(data, dict):
detail = output[:200] if output else "(empty)"
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if "error" in data:
return self._map_edit_error(data["error"], file_path, old_string)
return EditResult(
path=file_path,
occurrences=data.get("count", 1),
)
async def _aedit_inline(
self,
file_path: str,
old_string: str,
new_string: str,
replace_all: bool, # noqa: FBT001
) -> EditResult:
"""Async version of `_edit_inline`, delegating to `aexecute`."""
result = await self.aexecute(_build_edit_inline_cmd(file_path, old_string, new_string, replace_all=replace_all))
return _parse_edit_output(result.output, file_path, old_string)
def _edit_via_upload(
self,
@@ -707,12 +853,7 @@ except PermissionError:
if r.error:
return EditResult(error=f"Error editing file '{file_path}': {r.error}")
cmd = _EDIT_TMPFILE_TEMPLATE.format(
old_path_b64=base64.b64encode(old_tmp.encode("utf-8")).decode("ascii"),
new_path_b64=base64.b64encode(new_tmp.encode("utf-8")).decode("ascii"),
target_b64=base64.b64encode(file_path.encode("utf-8")).decode("ascii"),
replace_all=replace_all,
)
cmd = _build_edit_tmpfile_cmd(file_path, old_tmp, new_tmp, replace_all=replace_all)
result = self.execute(cmd)
output = result.output.rstrip()
@@ -736,25 +877,62 @@ except PermissionError:
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if "error" in data:
return self._map_edit_error(data["error"], file_path, old_string)
return _map_edit_error(data["error"], file_path, old_string)
return EditResult(
path=file_path,
occurrences=data.get("count", 1),
)
@staticmethod
def _map_edit_error(error: str, file_path: str, old_string: str) -> EditResult:
"""Map server-side error codes to `EditResult` objects."""
messages: dict[str, str] = {
"file_not_found": f"Error: File '{file_path}' not found",
"permission_denied": f"Error: Permission denied editing file '{file_path}'",
"not_a_file": f"Error: '{file_path}' is not a regular file",
"not_a_text_file": f"Error: File '{file_path}' is not a text file",
"string_not_found": f"Error: String not found in file: '{old_string}'",
"multiple_occurrences": (f"Error: String '{old_string}' appears multiple times. Use replace_all=True to replace all occurrences."),
}
return EditResult(error=messages.get(error, f"Error editing file '{file_path}': {error}"))
async def _aedit_via_upload(
self,
file_path: str,
old_string: str,
new_string: str,
replace_all: bool, # noqa: FBT001
) -> EditResult:
"""Async version of `_edit_via_upload`, delegating to `aexecute` and `aupload_files`."""
uid = base64.b32encode(os.urandom(10)).decode("ascii").lower()
old_tmp = f"/tmp/.deepagents_edit_{uid}_old" # noqa: S108
new_tmp = f"/tmp/.deepagents_edit_{uid}_new" # noqa: S108
resps = await self.aupload_files(
[
(old_tmp, old_string.encode("utf-8")),
(new_tmp, new_string.encode("utf-8")),
]
)
if len(resps) < 2: # noqa: PLR2004
return EditResult(error=f"Error editing file '{file_path}': upload returned no response")
for r in resps:
if r.error:
return EditResult(error=f"Error editing file '{file_path}': {r.error}")
cmd = _build_edit_tmpfile_cmd(file_path, old_tmp, new_tmp, replace_all=replace_all)
result = await self.aexecute(cmd)
output = result.output.rstrip()
try:
data = json.loads(output)
except (json.JSONDecodeError, ValueError):
cleanup = await self.aexecute(f"rm -f {shlex.quote(old_tmp)} {shlex.quote(new_tmp)}")
if cleanup.exit_code != 0:
logger.warning(
"Failed to clean up temp files for edit %s: %s",
file_path,
cleanup.output[:200],
)
detail = output[:200] if output else "(empty)"
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if not isinstance(data, dict):
detail = output[:200] if output else "(empty)"
return EditResult(error=f"Error editing file '{file_path}': unexpected server response: {detail}")
if "error" in data:
return _map_edit_error(data["error"], file_path, old_string)
return EditResult(path=file_path, occurrences=data.get("count", 1))
def grep(
self,
@@ -767,7 +945,6 @@ except PermissionError:
Args:
pattern: Literal string to search for (not a regex).
path: Directory or file to search in.
Defaults to `"."`.
glob: Optional file-name glob to restrict the search
(e.g. `'*.py'`).
@@ -775,98 +952,45 @@ except PermissionError:
Returns:
`GrepResult` with a list of `GrepMatch` dicts, or `error` on failure.
"""
search_path = shlex.quote(path or ".")
result = self.execute(_build_grep_cmd(pattern, path, glob))
return _parse_grep_output(result, path)
# Build grep command to get structured output
# `-Z` separates the filename from line data with NUL, so filenames may
# contain `:` without making the output ambiguous.
grep_opts = "-rHnFZ"
# Add glob pattern if specified
glob_pattern = ""
if glob:
glob_pattern = f"--include={shlex.quote(glob)}"
# Escape pattern for shell
pattern_escaped = shlex.quote(pattern)
cmd = f"grep {grep_opts} {glob_pattern} -e {pattern_escaped} {search_path} 2>/dev/null || true"
result = self.execute(cmd)
output = result.output.rstrip("\n")
if result.exit_code is not None and result.exit_code != 0:
detail = output.strip() if output else f"exit code {result.exit_code}"
return GrepResult(error=f"Path '{path or '.'}': {detail}")
if not output:
return GrepResult(matches=[])
# Parse grep output into GrepMatch objects
matches: list[GrepMatch] = []
parse_error: str | None = None
for line in output.split("\n"):
# Format is: path\0line_number:text
parts = line.split("\0", 1)
if len(parts) != 2: # noqa: PLR2004 # Grep output field count
parse_error = line
continue
line_parts = parts[1].split(":", 1)
if len(line_parts) != 2: # noqa: PLR2004 # Grep output field count
parse_error = line
continue
try:
matches.append(
{
"path": parts[0],
"line": int(line_parts[0]),
"text": line_parts[1],
}
)
except ValueError:
parse_error = line
if parse_error is not None and not matches:
return GrepResult(error=f"Path '{path or '.'}': {parse_error}")
return GrepResult(matches=matches)
async def agrep(
self,
pattern: str,
path: str | None = None,
glob: str | None = None,
) -> GrepResult:
"""Async version of `grep`, delegating to `aexecute` with timeout guard."""
try:
result = await asyncio.wait_for(
self.aexecute(_build_grep_cmd(pattern, path, glob)),
timeout=ASYNC_GREP_TIMEOUT,
)
except TimeoutError:
logger.warning(
"agrep timed out after %ds (pattern=%r, path=%r, glob=%r)",
ASYNC_GREP_TIMEOUT,
pattern,
path,
glob,
)
return GrepResult(
error=f"Error: grep timed out after {ASYNC_GREP_TIMEOUT}s. Try a more specific pattern or a narrower path.",
)
return _parse_grep_output(result, path)
def glob(self, pattern: str, path: str | None = None) -> GlobResult:
"""Structured glob matching returning `GlobResult`."""
search_path = path or "/"
# Encode pattern and path as base64 to avoid escaping issues
pattern_b64 = base64.b64encode(pattern.encode("utf-8")).decode("ascii")
path_b64 = base64.b64encode(search_path.encode("utf-8")).decode("ascii")
result = self.execute(_build_glob_cmd(pattern, search_path))
return _parse_glob_output(result.output, search_path)
cmd = _GLOB_COMMAND_TEMPLATE.format(path_b64=path_b64, pattern_b64=pattern_b64)
result = self.execute(cmd)
output = result.output.strip()
if not output:
return GlobResult(matches=[])
# Parse JSON output into FileInfo dicts. Any error record (emitted when
# the search path itself is unreachable) wins over partial matches —
# mirrors read()/ls() convention: sandbox emits a short code, host wraps
# it with the search-path prefix.
file_infos: list[FileInfo] = []
error: str | None = None
for line in output.split("\n"):
try:
data = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(data, dict) and "error" in data:
error = data["error"]
continue
file_infos.append(
{
"path": data["path"],
"is_dir": data["is_dir"],
}
)
if error is not None:
return GlobResult(matches=None, error=f"Path '{search_path}': {error}")
return GlobResult(matches=file_infos)
async def aglob(self, pattern: str, path: str | None = None) -> GlobResult:
"""Async version of `glob`, delegating to `aexecute`."""
search_path = path or "/"
result = await self.aexecute(_build_glob_cmd(pattern, search_path))
return _parse_glob_output(result.output, search_path)
@property
@abstractmethod
libs/deepagents/tests/unit_tests/backends/test_sandbox_backend.py
+206 -2
View File
@@ -30,6 +30,9 @@ from deepagents.backends.sandbox import (
_READ_COMMAND_TEMPLATE,
_WRITE_CHECK_TEMPLATE,
BaseSandbox,
_check_preflight_result,
_map_edit_error,
_parse_grep_output,
)
@@ -1034,7 +1037,7 @@ def test_sandbox_edit_one_over_threshold_uses_upload() -> None:
def test_map_edit_error_unknown_code_falls_through() -> None:
"""Test that _map_edit_error returns a generic error for unrecognized codes."""
result = BaseSandbox._map_edit_error("temp_read_failed", "/test/file.txt", "old")
result = _map_edit_error("temp_read_failed", "/test/file.txt", "old")
assert result.error is not None
assert "temp_read_failed" in result.error
@@ -1294,7 +1297,7 @@ def test_glob_empty_returns_empty_matches() -> None:
def test_map_edit_error_permission_denied() -> None:
"""_map_edit_error returns a readable message for permission_denied."""
result = BaseSandbox._map_edit_error("permission_denied", "/test/file.txt", "old")
result = _map_edit_error("permission_denied", "/test/file.txt", "old")
assert result.error is not None
assert "permission" in result.error.lower()
assert "/test/file.txt" in result.error
@@ -1326,3 +1329,204 @@ def test_sandbox_edit_inline_permission_denied() -> None:
assert result.error is not None
assert "permission" in result.error.lower()
assert "/test/locked.txt" in result.error
# -- async override tests (issue #665) ----------------------------------------
#
# These tests verify that the async helpers on BaseSandbox call aexecute()
# rather than wrapping the sync methods with asyncio.to_thread. The fixture is
# a NativeAsyncSandbox that overrides aexecute() with a coroutine that records
# all calls, while execute() raises to prove it is never reached.
class NativeAsyncSandbox(BaseSandbox):
"""Sandbox where aexecute() is natively async; execute() always raises."""
def __init__(self) -> None:
self._aexecute_calls: list[str] = []
self._aupload_calls: list[list[tuple[str, bytes]]] = []
self._next_output: str = ""
self._next_exit_code: int = 0
@property
def id(self) -> str:
return "native-async-sandbox"
def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
msg = "sync execute() must not be called from async helpers"
raise RuntimeError(msg)
async def aexecute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse: # noqa: ASYNC109
self._aexecute_calls.append(command)
output = self._next_output
exit_code = self._next_exit_code
return ExecuteResponse(output=output, exit_code=exit_code, truncated=False)
def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
msg = "sync upload_files() must not be called from async helpers"
raise RuntimeError(msg)
async def aupload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
self._aupload_calls.append(files)
return [FileUploadResponse(path=f[0], error=None) for f in files]
def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
msg = "sync download_files() must not be called from async helpers"
raise RuntimeError(msg)
async def test_als_calls_aexecute() -> None:
"""als() must call aexecute(), not the sync execute()."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = json.dumps({"path": "/foo/bar.txt", "is_dir": False})
result = await sandbox.als("/foo")
assert len(sandbox._aexecute_calls) == 1
assert result.error is None
assert result.entries is not None
async def test_aread_calls_aexecute() -> None:
"""aread() must call aexecute(), not the sync execute()."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = json.dumps({"encoding": "utf-8", "content": "hello"})
result = await sandbox.aread("/foo/bar.txt")
assert len(sandbox._aexecute_calls) == 1
assert result.error is None
assert result.file_data is not None
async def test_agrep_calls_aexecute() -> None:
"""agrep() must call aexecute(), not the sync execute()."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = "/foo/bar.txt\x00" + "1:hello"
sandbox._next_exit_code = 0
result = await sandbox.agrep("hello")
assert len(sandbox._aexecute_calls) == 1
assert result.error is None
async def test_aglob_calls_aexecute() -> None:
"""aglob() must call aexecute(), not the sync execute()."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = json.dumps({"path": "/foo/bar.txt", "is_dir": False})
result = await sandbox.aglob("**/*.txt")
assert len(sandbox._aexecute_calls) == 1
assert result.error is None
assert result.matches is not None
async def test_awrite_calls_aexecute_and_aupload_files() -> None:
"""awrite() must call aexecute() for preflight and aupload_files(), not sync methods."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = ""
sandbox._next_exit_code = 0
result = await sandbox.awrite("/foo/new.txt", "content")
assert len(sandbox._aexecute_calls) == 1, "expected one aexecute call for preflight"
assert len(sandbox._aupload_calls) == 1, "expected one aupload_files call"
assert result.error is None
assert result.path == "/foo/new.txt"
async def test_aedit_inline_calls_aexecute() -> None:
"""aedit() (small payload) must call aexecute(), not the sync execute()."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = json.dumps({"count": 1})
result = await sandbox.aedit("/foo/bar.txt", "old", "new")
assert len(sandbox._aexecute_calls) == 1
assert result.error is None
assert result.occurrences == 1
async def test_aedit_via_upload_calls_aexecute_and_aupload_files() -> None:
"""aedit() (large payload) must call aexecute() and aupload_files(), not sync methods."""
sandbox = NativeAsyncSandbox()
sandbox._next_output = json.dumps({"count": 1})
large = "x" * (_EDIT_INLINE_MAX_BYTES + 1)
result = await sandbox.aedit("/foo/bar.txt", large, "new")
assert len(sandbox._aupload_calls) == 1, "expected one aupload_files call for temp files"
assert len(sandbox._aexecute_calls) == 1, "expected one aexecute call for server-side replace"
assert result.error is None
# -- direct unit tests for module-level helper functions ----------------------
def test_map_edit_error_file_not_found() -> None:
result = _map_edit_error("file_not_found", "/a/b.txt", "old")
assert result.error is not None
assert "not found" in result.error.lower()
assert "/a/b.txt" in result.error
def test_map_edit_error_not_a_file() -> None:
result = _map_edit_error("not_a_file", "/a/b.txt", "old")
assert result.error is not None
assert "not a regular file" in result.error.lower()
assert "/a/b.txt" in result.error
def test_map_edit_error_not_a_text_file() -> None:
result = _map_edit_error("not_a_text_file", "/a/b.bin", "old")
assert result.error is not None
assert "not a text file" in result.error.lower()
assert "/a/b.bin" in result.error
def test_map_edit_error_string_not_found() -> None:
result = _map_edit_error("string_not_found", "/a/b.txt", "needle")
assert result.error is not None
assert "not found" in result.error.lower()
assert "needle" in result.error
def test_map_edit_error_multiple_occurrences() -> None:
result = _map_edit_error("multiple_occurrences", "/a/b.txt", "needle")
assert result.error is not None
assert "multiple" in result.error.lower() or "replace_all" in result.error
assert "needle" in result.error
def test_check_preflight_result_nonzero_exit_returns_error() -> None:
resp = ExecuteResponse(output="Error: file exists", exit_code=1, truncated=False)
result = _check_preflight_result(resp, "/a/b.txt")
assert result is not None
assert result.error is not None
assert "Error: file exists" in result.error
def test_check_preflight_result_error_in_output_returns_error() -> None:
resp = ExecuteResponse(output="Error: parent dir missing", exit_code=0, truncated=False)
result = _check_preflight_result(resp, "/a/b.txt")
assert result is not None
assert result.error is not None
assert "Error: parent dir missing" in result.error
def test_check_preflight_result_success_returns_none() -> None:
resp = ExecuteResponse(output="", exit_code=0, truncated=False)
result = _check_preflight_result(resp, "/a/b.txt")
assert result is None
def test_parse_grep_output_non_integer_line_number_is_skipped() -> None:
# Format: path\0not_a_number:text — int() raises ValueError, line is treated as parse error.
output = "file.py\0not_a_number:some text"
resp = ExecuteResponse(output=output, exit_code=0, truncated=False)
result = _parse_grep_output(resp, ".")
# The only line has a bad line number; no valid matches → error is set.
assert result.matches is None or result.matches == []
assert result.error is not None