Skip to content
Open
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
35 changes: 35 additions & 0 deletions src/google/adk/environment/_base_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,41 @@ async def read_file(self, path: Path) -> bytes:
FileNotFoundError: If the file does not exist.
"""

async def read_file_lines(
self,
path: str | Path,
start_line: int = 1,
end_line: int | None = None,
) -> tuple[list[bytes], int]:
"""Read a range of lines from a file.

The default implementation reads the whole file via ``read_file`` and
slices it in memory. Subclasses backed by a real filesystem should
override this to stream the file instead, so that requesting a small
range from a large file does not require buffering it entirely.

Args:
path: Absolute or working-dir-relative path to the file.
start_line: First line to return (1-based, inclusive).
end_line: Last line to return (1-based, inclusive). ``None`` means
through the end of the file.

Returns:
A tuple of (selected lines as bytes including line endings, total
number of lines in the file).

Raises:
FileNotFoundError: If the file does not exist.
"""
data = await self.read_file(path)
lines = data.splitlines(keepends=True)
total = len(lines)
start = max(1, start_line)
end = total if end_line is None else min(total, end_line)
if start > end:
return [], total
return lines[start - 1 : end], total

@abstractmethod
async def write_file(self, path: Path, content: str | bytes) -> None:
"""Write content to a file in the environment's filesystem.
Expand Down
77 changes: 77 additions & 0 deletions src/google/adk/environment/_local_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import shutil
import signal
import tempfile
from typing import BinaryIO
from typing import Iterator

from typing_extensions import override

Expand All @@ -37,6 +39,50 @@
# down cannot itself block forever.
_TERMINATE_GRACE_SECONDS = 5

# Chunk size for streaming a file while looking for line boundaries, so a
# single very long line cannot force reading the rest of the file at once.
_LINE_SCAN_CHUNK_SIZE = 65536


def _iter_binary_lines(f: BinaryIO) -> Iterator[bytes]:
"""Yields lines the same way `bytes.splitlines(keepends=True)` would.

Unlike iterating a binary file object directly (which only splits on
`b'\\n'`), this also splits on a bare `b'\\r'` and on `b'\\r\\n'`, matching
`bytes.splitlines()` so streamed reads agree with the whole-file path other
environments use.
"""
pending = b''
while True:
chunk = f.read(_LINE_SCAN_CHUNK_SIZE)
if not chunk:
break
buf = pending + chunk
pos = 0
end = len(buf)
while pos < end:
idx_n = buf.find(b'\n', pos)
idx_r = buf.find(b'\r', pos)
if idx_n == -1 and idx_r == -1:
break
if idx_r != -1 and (idx_n == -1 or idx_r < idx_n):
if idx_r == end - 1:
# A lone `\r` at the end of this chunk might be the start of a
# `\r\n` pair split across the chunk boundary; defer it.
break
if buf[idx_r + 1 : idx_r + 2] == b'\n':
yield buf[pos : idx_r + 2]
pos = idx_r + 2
else:
yield buf[pos : idx_r + 1]
pos = idx_r + 1
else:
yield buf[pos : idx_n + 1]
pos = idx_n + 1
pending = buf[pos:]
if pending:
yield pending


def _signal_group(group: int, sig: int) -> None:
"""Signals every process left in a group, tolerating an empty one."""
Expand Down Expand Up @@ -207,6 +253,21 @@ async def read_file(self, path: str | Path) -> bytes:
resolved = self._resolve_path(path)
return await asyncio.to_thread(self._sync_read, resolved)

@override
async def read_file_lines(
self,
path: str | Path,
start_line: int = 1,
end_line: int | None = None,
) -> tuple[list[bytes], int]:
if self._working_dir is None:
raise RuntimeError('`working_dir` is not set. Call initialize() first.')

resolved = self._resolve_path(path)
return await asyncio.to_thread(
self._sync_read_lines, resolved, start_line, end_line
)

@override
async def write_file(self, path: str | Path, content: str | bytes) -> None:
if self._working_dir is None:
Expand All @@ -232,6 +293,22 @@ def _sync_read(path: Path) -> bytes:
with open(path, 'rb') as f:
return f.read()

@staticmethod
def _sync_read_lines(
path: Path, start_line: int, end_line: int | None
) -> tuple[list[bytes], int]:
"""Streams *path* line-by-line, keeping only the requested range."""
start = max(1, start_line)
selected: list[bytes] = []
total = 0
with open(path, 'rb') as f:
for line in _iter_binary_lines(f):
total += 1
if total < start or (end_line is not None and total > end_line):
continue
selected.append(line)
return selected, total

@staticmethod
def _sync_write(path: Path, content: str | bytes) -> None:
os.makedirs(path.parent, exist_ok=True)
Expand Down
9 changes: 3 additions & 6 deletions src/google/adk/tools/environment/_read_file_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,12 +114,10 @@ async def run_async(
}

try:
# TODO: Avoid loading the entire file into memory to prevent OOM on large files.
data_bytes = await self._environment.read_file(path)
# Slice data_bytes by line boundaries before decoding.
lines_bytes = data_bytes.splitlines(keepends=True)
total = len(lines_bytes)
start = max(1, start_line or 1)
selected_bytes, total = await self._environment.read_file_lines(
path, start_line=start, end_line=end_line
)
end = min(total, end_line or total)
if start > total:
return {
Expand All @@ -135,7 +133,6 @@ async def run_async(
'error': f'`start_line` ({start}) is after `end_line` ({end}).',
'total_lines': total,
}
selected_bytes = lines_bytes[start - 1 : end]
lines = [
line_bytes.decode('utf-8', errors='replace')
for line_bytes in selected_bytes
Expand Down
61 changes: 61 additions & 0 deletions tests/unittests/environment/test_local_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,67 @@ async def test_read_nonexistent_raises(self, env: LocalEnvironment):
await env.read_file(Path("does_not_exist.txt"))


class TestReadFileLines:
"""Verify read_file_lines streams a range without buffering the file."""

@pytest.mark.asyncio
async def test_returns_requested_range_and_total(self, env: LocalEnvironment):
"""Selects the requested 1-based, inclusive line range."""
await env.write_file("lines.txt", "one\ntwo\nthree\nfour\n")

selected, total = await env.read_file_lines(
"lines.txt", start_line=2, end_line=3
)

assert selected == [b"two\n", b"three\n"]
assert total == 4

@pytest.mark.asyncio
@pytest.mark.parametrize(
"content",
[
pytest.param(b"one\rtwo\rthree\rfour\r", id="lone_cr"),
pytest.param(b"one\r\ntwo\rthree\nfour\r\n", id="mixed"),
],
)
async def test_line_splitting_matches_splitlines_semantics(
self, env: LocalEnvironment, content: bytes
):
"""Streamed splitting agrees with `bytes.splitlines()`, CR-only included."""
await env.write_file("lines.bin", content)

selected, total = await env.read_file_lines(
"lines.bin", start_line=2, end_line=3
)

expected_lines = content.splitlines(keepends=True)
assert selected == expected_lines[1:3]
assert total == len(expected_lines)

@pytest.mark.asyncio
async def test_does_not_read_the_whole_file_at_once(
self, env: LocalEnvironment, monkeypatch: pytest.MonkeyPatch
):
"""A ranged read must not go through the full-buffer `read_file` path."""
await env.write_file("lines.txt", "one\ntwo\nthree\nfour\n")

def _fail_full_read(path: Path) -> bytes:
raise AssertionError(
"read_file_lines must not buffer the whole file into memory"
)

monkeypatch.setattr(
LocalEnvironment, "_sync_read", staticmethod(_fail_full_read)
)

selected, total = await env.read_file_lines(
"lines.txt", start_line=2, end_line=2
)

assert selected == [b"two\n"]
assert total == 4


class TestExecuteTimeout:
"""Timeout and cancellation must reach the whole process tree."""

Expand Down
21 changes: 21 additions & 0 deletions tests/unittests/tools/environment/test_read_file_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ class _StubEnvironment(BaseEnvironment):
def __init__(self, files: dict[str, bytes]):
self._files = files
self.execute_calls: list[str] = []
self.read_file_calls: list[str] = []

@property
def working_dir(self) -> Path:
Expand All @@ -48,10 +49,29 @@ async def execute(

async def read_file(self, path: Path) -> bytes:
key = str(path)
self.read_file_calls.append(key)
if key not in self._files:
raise FileNotFoundError(key)
return self._files[key]

async def read_file_lines(
self,
path: Path,
start_line: int = 1,
end_line: Optional[int] = None,
) -> tuple[list[bytes], int]:
"""Slices the in-memory file without ever calling `read_file`."""
key = str(path)
if key not in self._files:
raise FileNotFoundError(key)
lines = self._files[key].splitlines(keepends=True)
total = len(lines)
start = max(1, start_line)
end = total if end_line is None else min(total, end_line)
if start > end:
return [], total
return lines[start - 1 : end], total

async def write_file(self, path: Path, content: str | bytes) -> None:
del path, content
raise NotImplementedError
Expand All @@ -75,6 +95,7 @@ async def test_read_file_with_line_range_uses_direct_file_read():
'total_lines': 4,
}
assert environment.execute_calls == []
assert environment.read_file_calls == []


@pytest.mark.asyncio
Expand Down