From 20fed8ab624aa19a003cb1d2043693a634b6c178 Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Tue, 7 Apr 2026 23:52:23 +0800 Subject: [PATCH] feat: implement file read tool with support for text and image files, including validation for large files --- astrbot/core/astr_main_agent.py | 10 +- astrbot/core/computer/booters/local.py | 3 +- .../booters/shipyard_search_file_util.py | 33 +- astrbot/core/computer/tools/fs.py | 24 +- .../computer/tools/tool_utils/file_read.py | 458 ++++++++++++++++++ tests/test_computer_fs_tools.py | 215 ++++++++ 6 files changed, 722 insertions(+), 21 deletions(-) create mode 100644 astrbot/core/computer/tools/tool_utils/file_read.py create mode 100644 tests/test_computer_fs_tools.py diff --git a/astrbot/core/astr_main_agent.py b/astrbot/core/astr_main_agent.py index 18383394f..266499e3d 100644 --- a/astrbot/core/astr_main_agent.py +++ b/astrbot/core/astr_main_agent.py @@ -1350,7 +1350,15 @@ async def build_main_agent( ) if config.computer_use_runtime == "local": - tool_prompt += f"\nCurrent workspace you can use: `{os.path.join(get_astrbot_workspaces_path(), event.unified_msg_origin)}`\n" + from astrbot.core.computer.tools.fs import _normalize_umo_for_workspace + + normalized_umo = _normalize_umo_for_workspace(event.unified_msg_origin) + tool_prompt += ( + f"\nCurrent workspace you can use: " + f"`{os.path.join(get_astrbot_workspaces_path(), normalized_umo)}`\n" + "Unless the user explicitly specifies a different directory, " + "perform all file-related operations in this workspace.\n" + ) req.system_prompt += f"\n{tool_prompt}\n" diff --git a/astrbot/core/computer/booters/local.py b/astrbot/core/computer/booters/local.py index 212f0129c..875a0f7e7 100644 --- a/astrbot/core/computer/booters/local.py +++ b/astrbot/core/computer/booters/local.py @@ -16,6 +16,7 @@ from astrbot.core.utils.astrbot_path import get_astrbot_root from ..olayer import FileSystemComponent, PythonComponent, ShellComponent from .base import ComputerBooter +from .shipyard_search_file_util import _truncate_long_lines _BLOCKED_COMMAND_PATTERNS = [ " rm -rf ", @@ -236,7 +237,7 @@ class LocalFileSystemComponent(FileSystemComponent): before_context=before_context, line_number=True, ) - return {"success": True, "content": "".join(results)} + return {"success": True, "content": _truncate_long_lines("".join(results))} return await asyncio.to_thread(_run) diff --git a/astrbot/core/computer/booters/shipyard_search_file_util.py b/astrbot/core/computer/booters/shipyard_search_file_util.py index a85998f98..1227244de 100644 --- a/astrbot/core/computer/booters/shipyard_search_file_util.py +++ b/astrbot/core/computer/booters/shipyard_search_file_util.py @@ -5,6 +5,27 @@ from typing import Any from ..olayer import ShellComponent +_MAX_SEARCH_LINE_COLUMNS = 1000 + + +def _truncate_long_lines(text: str) -> str: + output_lines: list[str] = [] + for line in text.splitlines(keepends=True): + line_ending = "" + line_body = line + if line.endswith("\r\n"): + line_body = line[:-2] + line_ending = "\r\n" + elif line.endswith("\n") or line.endswith("\r"): + line_body = line[:-1] + line_ending = line[-1] + + if len(line_body) > _MAX_SEARCH_LINE_COLUMNS: + line_body = line_body[:_MAX_SEARCH_LINE_COLUMNS] + + output_lines.append(f"{line_body}{line_ending}") + return "".join(output_lines) + def _build_rg_command( *, @@ -14,7 +35,15 @@ def _build_rg_command( after_context: int | None, before_context: int | None, ) -> list[str]: - command = ["rg", "--color=never", "-n", "-e", pattern] + command = [ + "rg", + "--color=never", + "-n", + "--max-columns", + str(_MAX_SEARCH_LINE_COLUMNS), + "-e", + pattern, + ] if glob: command.extend(["-g", glob]) if after_context is not None: @@ -104,7 +133,7 @@ async def search_files_via_shell( before_context=before_context, ) result = await shell.exec(command, timeout=timeout) - stdout = str(result.get("stdout", "") or "") + stdout = _truncate_long_lines(str(result.get("stdout", "") or "")) stderr = str(result.get("stderr", "") or "") exit_code = result.get("exit_code") if exit_code in (0, None): diff --git a/astrbot/core/computer/tools/fs.py b/astrbot/core/computer/tools/fs.py index 0d21d0d68..7595e543c 100644 --- a/astrbot/core/computer/tools/fs.py +++ b/astrbot/core/computer/tools/fs.py @@ -51,6 +51,7 @@ from astrbot.core.utils.astrbot_path import ( from ..computer_client import get_booter from .permissions import check_admin_permission +from .tool_utils.file_read import read_file_tool_result def _normalize_umo_for_workspace(umo: str) -> str: @@ -175,12 +176,12 @@ class FileReadTool(FunctionTool): }, "offset": { "type": "integer", - "description": "Line offset to start reading from. Defaults to 0.", + "description": "Optional line offset to start reading from. 0-based index.", "minimum": 0, }, "limit": { "type": "integer", - "description": "Maximum number of lines to read. Defaults to 4000.", + "description": "Optional maximum number of lines to read.", "minimum": 1, }, }, @@ -226,23 +227,12 @@ class FileReadTool(FunctionTool): context.context.context, context.context.event.unified_msg_origin, ) - result = await sb.fs.read_file( + return await read_file_tool_result( + sb, path=normalized_path, - encoding="utf-8", offset=offset, limit=limit, ) - if not result.get("success", False): - error_detail = str(result.get("error", "") or "").strip() - return ( - "Error reading file: " - f"{error_detail or 'unknown filesystem read error'}" - ) - - content = str(result.get("content", "") or "") - if not content: - return "No content found at the requested line offset." - return content or "" except PermissionError as exc: return f"Error: {exc}" except Exception as exc: @@ -438,7 +428,7 @@ class GrepTool(FunctionTool): }, "result_limit": { "type": "integer", - "description": "Maximum number of result groups returned by the tool. Defaults to 50.", + "description": "Maximum number of result groups returned by the tool. Defaults to 100.", "minimum": 1, }, }, @@ -544,7 +534,7 @@ class GrepTool(FunctionTool): pattern: str, path: str | None = None, glob: str | None = None, - result_limit: int = 50, + result_limit: int = 100, **kwargs, ) -> ToolExecResult: normalized_pattern = pattern.strip() diff --git a/astrbot/core/computer/tools/tool_utils/file_read.py b/astrbot/core/computer/tools/tool_utils/file_read.py new file mode 100644 index 000000000..e0a6f2142 --- /dev/null +++ b/astrbot/core/computer/tools/tool_utils/file_read.py @@ -0,0 +1,458 @@ +from __future__ import annotations + +import base64 +import json +from asyncio import to_thread +from dataclasses import dataclass +from pathlib import Path +from typing import Literal + +import mcp + +from astrbot.core.agent.context.token_counter import EstimateTokenCounter +from astrbot.core.agent.message import Message +from astrbot.core.agent.tool import ToolExecResult +from astrbot.core.computer.booters.base import ComputerBooter +from astrbot.core.computer.booters.local import LocalBooter +from astrbot.core.utils.astrbot_path import get_astrbot_temp_path +from astrbot.core.utils.media_utils import ( + IMAGE_COMPRESS_DEFAULT_MAX_SIZE, + IMAGE_COMPRESS_DEFAULT_OPTIMIZE, + IMAGE_COMPRESS_DEFAULT_QUALITY, + _compress_image_sync, +) + +_MAX_FILE_READ_BYTES = 128 * 1024 +_MAX_FILE_READ_TOKENS = 25_000 +_MAX_TEXT_FILE_FULL_READ_BYTES = 256 * 1024 +_FILE_SNIFF_BYTES = 512 +_TOKEN_COUNTER = EstimateTokenCounter() +_TEXT_ENCODINGS = ( + "utf-8-sig", + "utf-8", + "gb18030", + "utf-16", + "utf-16-le", + "utf-16-be", + "utf-32", + "utf-32-le", + "utf-32-be", +) +_UTF_BOMS = ( + b"\xef\xbb\xbf", + b"\xff\xfe", + b"\xfe\xff", + b"\xff\xfe\x00\x00", + b"\x00\x00\xfe\xff", +) +_BINARY_MAGIC_PREFIXES = ( + b"%PDF-", + b"PK\x03\x04", + b"PK\x05\x06", + b"PK\x07\x08", + b"\x1f\x8b", + b"7z\xbc\xaf\x27\x1c", + b"Rar!\x1a\x07", + b"\x7fELF", + b"MZ", +) + + +@dataclass(frozen=True) +class FileProbe: + kind: Literal["text", "image", "binary"] + encoding: str | None + mime_type: str | None + size_bytes: int + + +def _build_probe_script(path: str) -> str: + return f""" +import base64 +import json +from pathlib import Path + +path = Path({path!r}) +with path.open("rb") as file_obj: + sample = file_obj.read({_FILE_SNIFF_BYTES}) +print( + json.dumps( + {{ + "size_bytes": path.stat().st_size, + "sample_b64": base64.b64encode(sample).decode("ascii"), + }} + ) +) +""".strip() + + +def _build_text_read_script( + path: str, + *, + encoding: str, + offset: int | None, + limit: int | None, +) -> str: + start_expr = "0" if offset is None else str(offset) + limit_expr = "None" if limit is None else str(limit) + return f""" +import json +from pathlib import Path + +path = Path({path!r}) +start = {start_expr} +limit = {limit_expr} +end = None if limit is None else start + limit +lines = [] +with path.open("r", encoding={encoding!r}, newline="") as file_obj: + for index, line in enumerate(file_obj): + if index < start: + continue + if end is not None and index >= end: + break + lines.append(line) +content = "".join(lines) +print(json.dumps({{"content": content}}, ensure_ascii=False)) +""".strip() + + +def _build_image_read_script(path: str) -> str: + return f""" +import base64 +import json +from pathlib import Path + +path = Path({path!r}) +data = path.read_bytes() +print( + json.dumps( + {{ + "size_bytes": len(data), + "base64": base64.b64encode(data).decode("ascii"), + }} + ) +) +""".strip() + + +async def _exec_python_json( + booter: ComputerBooter, + script: str, + *, + action: str, +) -> dict: + result = await booter.python.exec(script) + data = result.get("data") if isinstance(result.get("data"), dict) else {} + if not isinstance(data, dict): + raise RuntimeError(f"{action} failed: invalid result format") + output = data.get("output") if isinstance(data.get("output"), dict) else {} + if not isinstance(output, dict): + raise RuntimeError(f"{action} failed: invalid output format") + error_text = str(data.get("error", "") or result.get("error", "") or "").strip() + if error_text: + raise RuntimeError(f"{action} failed: {error_text}") + + text = str(output.get("text", "") or "").strip() + if not text: + raise RuntimeError(f"{action} failed: empty output") + + try: + payload = json.loads(text) + except json.JSONDecodeError as exc: + raise RuntimeError(f"{action} failed: invalid JSON output") from exc + + if not isinstance(payload, dict): + raise RuntimeError(f"{action} failed: invalid JSON payload") + return payload + + +async def _probe_local_file(path: str) -> dict[str, str | int]: + def _run() -> dict[str, str | int]: + file_path = Path(path) + with file_path.open("rb") as file_obj: + sample = file_obj.read(_FILE_SNIFF_BYTES) + return { + "size_bytes": file_path.stat().st_size, + "sample_b64": base64.b64encode(sample).decode("ascii"), + } + + return await to_thread(_run) + + +async def _read_local_text_range( + path: str, + *, + encoding: str, + offset: int | None, + limit: int | None, +) -> str: + def _run() -> str: + lines: list[str] = [] + start = 0 if offset is None else offset + end = None if limit is None else start + limit + with Path(path).open("r", encoding=encoding, newline="") as file_obj: + for index, line in enumerate(file_obj): + if index < start: + continue + if end is not None and index >= end: + break + lines.append(line) + return "".join(lines) + + return await to_thread(_run) + + +async def _read_local_image_base64(path: str) -> dict[str, str | int]: + def _run() -> dict[str, str | int]: + data = Path(path).read_bytes() + return { + "size_bytes": len(data), + "base64": base64.b64encode(data).decode("ascii"), + } + + return await to_thread(_run) + + +async def _compress_image_bytes_to_base64(data: bytes) -> dict[str, str | int]: + def _run() -> dict[str, str | int]: + temp_dir = Path(get_astrbot_temp_path()) + temp_dir.mkdir(parents=True, exist_ok=True) + compressed_path = Path( + _compress_image_sync( + data, + temp_dir, + IMAGE_COMPRESS_DEFAULT_MAX_SIZE, + IMAGE_COMPRESS_DEFAULT_QUALITY, + IMAGE_COMPRESS_DEFAULT_OPTIMIZE, + ) + ) + try: + compressed_bytes = compressed_path.read_bytes() + finally: + compressed_path.unlink(missing_ok=True) + + return { + "size_bytes": len(compressed_bytes), + "base64": base64.b64encode(compressed_bytes).decode("ascii"), + "mime_type": "image/jpeg", + } + + return await to_thread(_run) + + +def _detect_image_mime(sample: bytes) -> str | None: + if sample.startswith(b"\x89PNG\r\n\x1a\n"): + return "image/png" + if sample.startswith(b"\xff\xd8\xff"): + return "image/jpeg" + if sample.startswith((b"GIF87a", b"GIF89a")): + return "image/gif" + if sample.startswith(b"BM"): + return "image/bmp" + if sample.startswith((b"II*\x00", b"MM\x00*")): + return "image/tiff" + if sample.startswith(b"\x00\x00\x01\x00"): + return "image/x-icon" + if len(sample) >= 12 and sample[:4] == b"RIFF" and sample[8:12] == b"WEBP": + return "image/webp" + if len(sample) >= 12 and sample[4:12] in (b"ftypavif", b"ftypavis"): + return "image/avif" + return None + + +def _looks_like_known_binary(sample: bytes) -> bool: + return any(sample.startswith(prefix) for prefix in _BINARY_MAGIC_PREFIXES) + + +def _looks_like_text(decoded: str) -> bool: + if not decoded: + return True + + disallowed = 0 + printable = 0 + for char in decoded: + if char in "\n\r\t\f\b": + printable += 1 + continue + if char.isprintable(): + printable += 1 + code = ord(char) + if (0 <= code < 32) or (127 <= code < 160): + disallowed += 1 + + total = max(len(decoded), 1) + return disallowed / total <= 0.02 and printable / total >= 0.85 + + +def _detect_text_encoding(sample: bytes) -> str | None: + if not sample: + return "utf-8" + + if b"\x00" in sample and not sample.startswith(_UTF_BOMS): + odd_bytes = sample[1::2] + even_bytes = sample[0::2] + odd_zero_ratio = odd_bytes.count(0) / max(len(odd_bytes), 1) + even_zero_ratio = even_bytes.count(0) / max(len(even_bytes), 1) + if odd_zero_ratio < 0.8 and even_zero_ratio < 0.8: + return None + + for encoding in _TEXT_ENCODINGS: + try: + decoded = sample.decode(encoding) + except UnicodeDecodeError: + continue + if _looks_like_text(decoded): + return encoding + + return None + + +def _probe_file(sample: bytes, *, size_bytes: int) -> FileProbe: + if image_mime := _detect_image_mime(sample): + return FileProbe( + kind="image", + encoding=None, + mime_type=image_mime, + size_bytes=size_bytes, + ) + + if _looks_like_known_binary(sample): + return FileProbe( + kind="binary", + encoding=None, + mime_type=None, + size_bytes=size_bytes, + ) + + if encoding := _detect_text_encoding(sample): + return FileProbe( + kind="text", + encoding=encoding, + mime_type="text/plain", + size_bytes=size_bytes, + ) + + return FileProbe( + kind="binary", + encoding=None, + mime_type=None, + size_bytes=size_bytes, + ) + + +def _validate_text_output(content: str) -> str | None: + content_bytes = len(content.encode("utf-8")) + if content_bytes > _MAX_FILE_READ_BYTES: + return ( + "Error reading file: " + f"output exceeds {_MAX_FILE_READ_BYTES} bytes " + f"({content_bytes} bytes). Use `offset`, `limit` to narrow the read window." + ) + + content_tokens = _TOKEN_COUNTER.count_tokens( + [Message(role="user", content=content)] + ) + if content_tokens > _MAX_FILE_READ_TOKENS: + return ( + "Error reading file: " + f"output exceeds {_MAX_FILE_READ_TOKENS} tokens " + f"({content_tokens} tokens). Use `offset`, `limit` to narrow the read window." + ) + + return None + + +def _validate_full_text_read_request(probe: FileProbe) -> str | None: + if probe.size_bytes > _MAX_TEXT_FILE_FULL_READ_BYTES: + return ( + "Error reading file: " + f"text file exceeds {_MAX_TEXT_FILE_FULL_READ_BYTES} bytes " + f"({probe.size_bytes} bytes). Use `offset` and `limit` to narrow the read window." + ) + return None + + +async def read_file_tool_result( + booter: ComputerBooter, + *, + path: str, + offset: int | None, + limit: int | None, +) -> ToolExecResult: + if isinstance(booter, LocalBooter): + probe_payload = await _probe_local_file(path) + else: + probe_payload = await _exec_python_json( + booter, + _build_probe_script(path), + action="file probe", + ) + sample_b64 = str(probe_payload.get("sample_b64", "") or "") + sample = base64.b64decode(sample_b64) if sample_b64 else b"" + size_bytes = int(probe_payload.get("size_bytes", 0) or 0) + probe = _probe_file(sample, size_bytes=size_bytes) + + if probe.kind == "binary": + return "Error reading file: binary files are not supported by this tool." + + if probe.kind == "image": + if isinstance(booter, LocalBooter): + image_payload = await _read_local_image_base64(path) + else: + image_payload = await _exec_python_json( + booter, + _build_image_read_script(path), + action="image read", + ) + raw_base64_data = str(image_payload.get("base64", "") or "") + if not raw_base64_data: + return "Error reading file: image payload is empty." + raw_bytes = base64.b64decode(raw_base64_data) + compressed_payload = await _compress_image_bytes_to_base64(raw_bytes) + base64_data = str(compressed_payload.get("base64", "") or "") + if not base64_data: + return "Error reading file: compressed image payload is empty." + return mcp.types.CallToolResult( + content=[ + mcp.types.ImageContent( + type="image", + data=base64_data, + mimeType=str( + compressed_payload.get("mime_type", "") or "image/jpeg" + ), + ) + ] + ) + + if isinstance(booter, LocalBooter): + if offset is None and limit is None: + if validation_error := _validate_full_text_read_request(probe): + return validation_error + content = await _read_local_text_range( + path, + encoding=probe.encoding or "utf-8", + offset=offset, + limit=limit, + ) + else: + if offset is None and limit is None: + if validation_error := _validate_full_text_read_request(probe): + return validation_error + text_payload = await _exec_python_json( + booter, + _build_text_read_script( + path, + encoding=probe.encoding or "utf-8", + offset=offset, + limit=limit, + ), + action="text read", + ) + content = str(text_payload.get("content", "") or "") + if not content: + return "No content found at the requested line offset." + + if validation_error := _validate_text_output(content): + return validation_error + + return content diff --git a/tests/test_computer_fs_tools.py b/tests/test_computer_fs_tools.py new file mode 100644 index 000000000..857c73ff0 --- /dev/null +++ b/tests/test_computer_fs_tools.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import base64 +from types import SimpleNamespace +from typing import Any + +import pytest +from mcp.types import CallToolResult, ImageContent +from PIL import Image + +from astrbot.core.agent.run_context import ContextWrapper +from astrbot.core.computer.booters.local import LocalBooter +from astrbot.core.computer.tools import fs as fs_tools +from astrbot.core.computer.tools.tool_utils import file_read as file_read_utils + + +def _make_context( + *, + require_admin: bool = True, + role: str = "admin", + runtime: str = "local", + umo: str = "qq:friend:user-1", +) -> ContextWrapper: + config_holder = SimpleNamespace( + get_config=lambda umo=None: { + "provider_settings": { + "computer_use_require_admin": require_admin, + "computer_use_runtime": runtime, + } + } + ) + event = SimpleNamespace( + role=role, + unified_msg_origin=umo, + get_sender_id=lambda: "user-1", + ) + astr_ctx = SimpleNamespace(context=config_holder, event=event) + return ContextWrapper(context=astr_ctx) + + +def _setup_local_fs_tools( + monkeypatch: pytest.MonkeyPatch, + tmp_path, + *, + umo: str = "qq:friend:user-1", +) -> Any: + workspaces_root = tmp_path / "workspaces" + skills_root = tmp_path / "skills" + temp_root = tmp_path / "temp" + workspaces_root.mkdir() + skills_root.mkdir() + temp_root.mkdir() + + monkeypatch.setattr( + fs_tools, + "get_astrbot_workspaces_path", + lambda: str(workspaces_root), + ) + monkeypatch.setattr( + fs_tools, + "get_astrbot_skills_path", + lambda: str(skills_root), + ) + monkeypatch.setattr( + fs_tools, + "get_astrbot_temp_path", + lambda: str(temp_root), + ) + monkeypatch.setattr( + file_read_utils, + "get_astrbot_temp_path", + lambda: str(temp_root), + ) + + booter = LocalBooter() + + async def _fake_get_booter(_ctx, _umo): + return booter + + monkeypatch.setattr(fs_tools, "get_booter", _fake_get_booter) + + normalized_umo = fs_tools._normalize_umo_for_workspace(umo) + workspace = workspaces_root / normalized_umo + workspace.mkdir(parents=True, exist_ok=True) + return workspace + + +def _make_large_text() -> str: + return "".join(f"line-{index:05d}-{'x' * 48}\n" for index in range(6000)) + + +@pytest.mark.asyncio +async def test_file_read_tool_rejects_large_full_text_read_before_local_stream_read( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + workspace = _setup_local_fs_tools(monkeypatch, tmp_path) + large_file = workspace / "large.txt" + large_file.write_text(_make_large_text(), encoding="utf-8") + + async def _unexpected_read(*args, **kwargs): + raise AssertionError("full file read should be rejected before streaming") + + monkeypatch.setattr(file_read_utils, "_read_local_text_range", _unexpected_read) + + result = await fs_tools.FileReadTool().call( + _make_context(), + path="large.txt", + ) + + assert "text file exceeds 262144 bytes" in result + assert "Use `offset` and `limit`" in result + + +@pytest.mark.asyncio +async def test_file_read_tool_allows_partial_read_for_large_text_file( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + workspace = _setup_local_fs_tools(monkeypatch, tmp_path) + large_file = workspace / "large.txt" + lines = [f"line-{index:05d}\n" for index in range(50000)] + large_file.write_text("".join(lines), encoding="utf-8") + + result = await fs_tools.FileReadTool().call( + _make_context(), + path="large.txt", + offset=1000, + limit=3, + ) + + assert result == "".join(lines[1000:1003]) + + +@pytest.mark.asyncio +async def test_file_read_tool_returns_image_call_tool_result_for_images( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + workspace = _setup_local_fs_tools(monkeypatch, tmp_path) + image_path = workspace / "sample.png" + Image.new("RGB", (32, 16), color=(255, 0, 0)).save(image_path, format="PNG") + + result = await fs_tools.FileReadTool().call( + _make_context(), + path="sample.png", + ) + + assert isinstance(result, CallToolResult) + assert len(result.content) == 1 + assert isinstance(result.content[0], ImageContent) + assert result.content[0].mimeType == "image/jpeg" + assert base64.b64decode(result.content[0].data).startswith(b"\xff\xd8\xff") + + +@pytest.mark.asyncio +async def test_file_read_tool_treats_svg_as_text( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + workspace = _setup_local_fs_tools(monkeypatch, tmp_path) + svg_path = workspace / "shape.svg" + svg_text = ( + "" + ) + svg_path.write_text(svg_text, encoding="utf-8") + + result = await fs_tools.FileReadTool().call( + _make_context(), + path="shape.svg", + ) + + assert result == svg_text + + +@pytest.mark.asyncio +async def test_file_read_tool_rejects_pdf_as_binary( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + workspace = _setup_local_fs_tools(monkeypatch, tmp_path) + pdf_path = workspace / "doc.pdf" + pdf_path.write_bytes(b"%PDF-1.7\n%\xe2\xe3\xcf\xd3\n1 0 obj\n<<>>\nendobj\n") + + result = await fs_tools.FileReadTool().call( + _make_context(), + path="doc.pdf", + ) + + assert result == "Error reading file: binary files are not supported by this tool." + + +@pytest.mark.asyncio +async def test_grep_tool_applies_result_limit( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + workspace = _setup_local_fs_tools(monkeypatch, tmp_path) + text_path = workspace / "grep.txt" + text_path.write_text( + "match-1\nmatch-2\nmatch-3\nmatch-4\n", + encoding="utf-8", + ) + + result = await fs_tools.GrepTool().call( + _make_context(), + pattern="match", + path="grep.txt", + result_limit=2, + ) + + assert "match-1" in result + assert "match-2" in result + assert "match-3" not in result + assert "[Truncated to first 2 result groups.]" in result