125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
from abc import ABC, abstractmethod
|
|
|
|
from deerflow.sandbox.search import GrepMatch
|
|
|
|
|
|
class Sandbox(ABC):
|
|
"""Abstract base class for sandbox environments"""
|
|
|
|
_id: str
|
|
|
|
def __init__(self, id: str):
|
|
self._id = id
|
|
|
|
@property
|
|
def id(self) -> str:
|
|
return self._id
|
|
|
|
@abstractmethod
|
|
def execute_command(self, command: str) -> str:
|
|
"""Execute bash command in sandbox.
|
|
|
|
Args:
|
|
command: The command to execute.
|
|
|
|
Returns:
|
|
The standard or error output of the command.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def read_file(self, path: str) -> str:
|
|
"""Read the content of a file.
|
|
|
|
Args:
|
|
path: The absolute path of the file to read.
|
|
|
|
Returns:
|
|
The content of the file.
|
|
"""
|
|
pass
|
|
|
|
def read_file_range(
|
|
self,
|
|
path: str,
|
|
*,
|
|
start_line: int = 1,
|
|
end_line: int | None = None,
|
|
max_chars: int = 32_000,
|
|
) -> tuple[str, int | None, bool]:
|
|
"""Read a bounded line range.
|
|
|
|
Providers should override this method to avoid loading the entire file.
|
|
The compatibility implementation remains bounded at the tool boundary,
|
|
though an older remote provider may still materialize its response.
|
|
"""
|
|
content = self.read_file(path)
|
|
selected: list[str] = []
|
|
chars = 0
|
|
next_line: int | None = None
|
|
for number, line in enumerate(content.splitlines(), start=1):
|
|
if number < start_line:
|
|
continue
|
|
if end_line is not None and number > end_line:
|
|
break
|
|
addition = len(line) + (1 if selected else 0)
|
|
if chars + addition > max_chars:
|
|
next_line = number
|
|
break
|
|
selected.append(line)
|
|
chars += addition
|
|
return "\n".join(selected), next_line, next_line is not None
|
|
|
|
@abstractmethod
|
|
def list_dir(self, path: str, max_depth=2) -> list[str]:
|
|
"""List the contents of a directory.
|
|
|
|
Args:
|
|
path: The absolute path of the directory to list.
|
|
max_depth: The maximum depth to traverse. Default is 2.
|
|
|
|
Returns:
|
|
The contents of the directory.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def write_file(self, path: str, content: str, append: bool = False) -> None:
|
|
"""Write content to a file.
|
|
|
|
Args:
|
|
path: The absolute path of the file to write to.
|
|
content: The text content to write to the file.
|
|
append: Whether to append the content to the file. If False, the file will be created or overwritten.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def glob(self, path: str, pattern: str, *, include_dirs: bool = False, max_results: int = 200) -> tuple[list[str], bool]:
|
|
"""Find paths that match a glob pattern under a root directory."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def grep(
|
|
self,
|
|
path: str,
|
|
pattern: str,
|
|
*,
|
|
glob: str | None = None,
|
|
literal: bool = False,
|
|
case_sensitive: bool = False,
|
|
max_results: int = 100,
|
|
) -> tuple[list[GrepMatch], bool]:
|
|
"""Search for matches inside text files under a directory."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def update_file(self, path: str, content: bytes) -> None:
|
|
"""Update a file with binary content.
|
|
|
|
Args:
|
|
path: The absolute path of the file to update.
|
|
content: The binary content to write to the file.
|
|
"""
|
|
pass
|