mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-31 00:03:01 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: add demand-driven document retrieval (#5525)

Browse Source
This commit is contained in:
chengyongru
2026-08-25 15:34:48 +08:00
committed by GitHub
parent 5cf78540a4
commit 4b2965c8f3
9 changed files with 1097 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/agent/tools/filesystem.py
+83 -32
View File
@@ -251,16 +251,16 @@ def _builtin_skill_read_path(path: str) -> Path | None:
tool_parameters_schema(
path=StringSchema("The file path to read"),
offset=IntegerSchema(
description="Line number to start reading from (1-indexed, default 1)",
description="1-based text or extracted-document line (default 1)",
minimum=1,
),
limit=IntegerSchema(
description="Maximum number of lines to read (default 2000)",
description="Maximum lines to return (default 2000)",
minimum=1,
),
pages=StringSchema("Page range for PDF files, e.g. '1-5' (default: all, max 20 pages)"),
pages=StringSchema("PDF page number or range, e.g. '7' or '1-5' (max 20 pages)"),
force=BooleanSchema(
description="Bypass same-file read deduplication and return content again.",
description="Return an unchanged range again",
default=False,
),
required=["path"],
@@ -282,18 +282,8 @@ class ReadFileTool(_FsTool):
@property
def description(self) -> str:
return (
"Read a file (text, image, or document). "
"Text output format: LINE_NUM|CONTENT. "
"Images return visual content for analysis. "
"Supports PDF, DOCX, XLSX, PPTX documents. "
"Uploaded non-image attachments are referenced by path; read them "
"with this tool only when their contents are needed. "
"Use find_files/list_dir first when the path is uncertain. "
"Read the relevant range before editing so replacements or patches "
"are based on current content. "
"Use offset and limit for large text files. "
"Use force=true to re-read content even if unchanged. "
"Reads exceeding ~128K chars are truncated."
"Read text, images, PDFs, and Office documents by path. "
"Text is line-numbered; use offset/limit or pages for targeted ranges."
)
@property
@@ -342,7 +332,7 @@ class ReadFileTool(_FsTool):
# Office document support
if fp.suffix.lower() in {".docx", ".xlsx", ".pptx"}:
return self._read_office_doc(fp)
return self._read_office_doc(fp, offset, limit)
raw = fp.read_bytes()
if not raw:
@@ -464,8 +454,8 @@ class ReadFileTool(_FsTool):
max_pages=self._MAX_PDF_PAGES,
max_chars=self._MAX_CHARS,
)
except PdfPageRangeError:
return ToolResult.error(f"Error: Invalid page range '{pages}'. Use format like '1-5'.")
except PdfPageRangeError as e:
return ToolResult.error(f"Error: Invalid page range '{pages}': {e!s}.")
except PdfSafetyError as e:
return ToolResult.error(f"Error reading PDF: {e}")
except Exception as e:
@@ -484,24 +474,85 @@ class ReadFileTool(_FsTool):
)
return result
def _read_office_doc(self, fp: Path) -> str:
from nanobot.utils.document import extract_text
def _read_office_doc(
self,
fp: Path,
offset: int,
limit: int | None,
) -> str:
from nanobot.utils.document import open_document_line_source
result = extract_text(fp)
offset = max(1, offset)
requested_limit = limit or self._DEFAULT_LIMIT
source_iterator = None
try:
source = open_document_line_source(fp)
if source is None:
return ToolResult.error(f"Error: Unsupported file format: {fp.suffix}")
source_iterator = source.lines
numbered: list[str] = []
output_chars = 0
total_seen = 0
end = offset - 1
has_more = False
line_was_clipped = False
if result is None:
return ToolResult.error(f"Error: Unsupported file format: {fp.suffix}")
for line in source_iterator:
total_seen = line.extracted_line
if line.extracted_line < offset:
continue
if len(numbered) >= requested_limit:
has_more = True
break
if result.startswith("[error:"):
return ToolResult.error(f"Error reading {fp.suffix.upper()} file: {result}")
rendered = f"{line.extracted_line}| {line.text}"
extra = 1 if numbered else 0
if output_chars + extra + len(rendered) > self._MAX_CHARS:
if numbered:
has_more = True
break
prefix = f"{line.extracted_line}| "
available = max(0, self._MAX_CHARS - len(prefix) - 3)
rendered = f"{prefix}{line.text[:available]}..."
line_was_clipped = True
has_more = True
numbered.append(rendered)
output_chars += extra + len(rendered)
end = line.extracted_line
if line_was_clipped:
break
if not result:
return f"({fp.suffix.upper().lstrip('.')} has no extractable text: {fp})"
if not numbered:
if total_seen == 0:
return (
f"({fp.suffix.upper().lstrip('.')} has no extractable text: {fp})"
)
return ToolResult.error(
f"Error: offset {offset} is beyond end of extracted document "
f"({total_seen} lines)"
)
if len(result) > self._MAX_CHARS:
result = result[:self._MAX_CHARS] + "\n\n(Document text truncated at ~128K chars)"
return result
output = "\n".join(numbered)
if has_more:
if line_was_clipped:
output += (
"\n\n(Document text truncated at ~128K chars; line clipped. "
f"Use offset={end + 1} to continue.)"
)
else:
output += (
f"\n\n(Showing extracted lines {offset}-{end}. "
f"Use offset={end + 1} to continue.)"
)
else:
output += f"\n\n(End of document — {total_seen} extracted lines total)"
return output
except Exception as e:
return ToolResult.error(f"Error reading {fp.suffix.upper()} file: {e!s}")
finally:
close = getattr(source_iterator, "close", None)
if close is not None:
close()
# ---------------------------------------------------------------------------
nanobot/agent/tools/search.py
+253 -120
View File
@@ -7,15 +7,23 @@ from __future__ import annotations
import fnmatch
import os
import re
from collections import deque
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any, Iterable, TypeVar
from typing import Any, Iterable, Iterator, TypeVar
from nanobot.agent.tools.base import ToolResult
from nanobot.agent.tools.filesystem import ListDirTool, _FsTool
from nanobot.utils.document import (
LocatedDocumentLine,
PdfPageRangeError,
open_document_line_source,
)
_DEFAULT_HEAD_LIMIT = 250
_DEFAULT_FILE_HEAD_LIMIT = 200
_DOCUMENT_EXTENSIONS = frozenset({".pdf", ".docx", ".xlsx", ".pptx"})
T = TypeVar("T")
_TYPE_GLOB_MAP = {
"py": ("*.py", "*.pyi"),
@@ -41,6 +49,14 @@ _TYPE_GLOB_MAP = {
}
@dataclass(slots=True)
class _PendingContextMatch:
lines: list[LocatedDocumentLine]
match_index: int
match_start: int
remaining_after: int
def _normalize_pattern(pattern: str) -> str:
return pattern.strip().replace("\\", "/")
@@ -64,6 +80,15 @@ def _is_binary(raw: bytes) -> bool:
return (non_text / len(sample)) > 0.2
def _excel_column(index: int) -> str:
"""Return a 1-indexed spreadsheet column label without importing openpyxl."""
label = ""
while index > 0:
index, remainder = divmod(index - 1, 26)
label = chr(ord("A") + remainder) + label
return label
def _paginate(items: list[T], limit: int | None, offset: int) -> tuple[list[T], bool]:
if limit is None:
return items[offset:], False
@@ -133,11 +158,8 @@ class FindFilesTool(_SearchTool):
@property
def description(self) -> str:
return (
"Find files by path fragment, glob, or file type. "
"Use this before read_file when you need to locate files, and "
"prefer it over shell find/ls for ordinary workspace discovery. "
"Returns workspace-relative paths and skips common dependency/build "
"directories."
"Find workspace paths by name, glob, or file type. "
"Returns relative paths and skips dependency/build directories."
)
@property
@@ -151,41 +173,38 @@ class FindFilesTool(_SearchTool):
"properties": {
"path": {
"type": "string",
"description": "Directory or file to search in (default '.')",
"description": "Search root (default '.')",
},
"query": {
"type": "string",
"description": (
"Optional case-insensitive path fragment search. "
"Whitespace-separated terms must all be present."
),
"description": "Case-insensitive path terms; all must match",
},
"glob": {
"type": "string",
"description": "Optional file filter, e.g. '*.py' or 'tests/**/test_*.py'",
"description": "Path filter, e.g. '*.py' or 'tests/**/test_*.py'",
},
"type": {
"type": "string",
"description": "Optional file type shorthand, e.g. 'py', 'ts', 'md', 'json'",
"description": "File type, e.g. 'py', 'ts', 'md', or 'json'",
},
"include_dirs": {
"type": "boolean",
"description": "Include matching directories as well as files (default false)",
"description": "Include directories (default false)",
},
"sort": {
"type": "string",
"enum": ["path", "modified"],
"description": "Sort by path or most recently modified first (default path)",
"description": "Sort order (default path)",
},
"head_limit": {
"type": "integer",
"description": "Maximum number of paths to return (default 200, 0 for all, max 1000)",
"description": "Maximum paths (default 200; 0 for all)",
"minimum": 0,
"maximum": 1000,
},
"offset": {
"type": "integer",
"description": "Skip the first N results before applying head_limit",
"description": "Paths to skip before head_limit",
"minimum": 0,
"maximum": 100000,
},
@@ -280,10 +299,11 @@ class FindFilesTool(_SearchTool):
class GrepTool(_SearchTool):
"""Search file contents using a regex-like pattern."""
"""Search text and document contents using a regex-like pattern."""
_scopes = {"core", "subagent"}
_MAX_RESULT_CHARS = 128_000
_MAX_RENDERED_LINE_CHARS = 2_000
_MAX_FILE_BYTES = 2_000_000
_MAX_EXPLICIT_FILE_BYTES = 100_000_000
@@ -294,12 +314,8 @@ class GrepTool(_SearchTool):
@property
def description(self) -> str:
return (
"Search file contents with a regex pattern. "
"Default output_mode is files_with_matches (file paths only); "
"use content mode for matching lines with context. Prefer this "
"over shell grep for ordinary workspace searches. "
"Binary and file-size limits are enforced by the tool; explicit file paths "
"use a larger bounded limit than directory searches. Supports glob/type filtering."
"Search text, PDF, DOCX, XLSX, and PPTX content. "
"Returns matches with five context lines and source locators by default."
)
@property
@@ -313,80 +329,62 @@ class GrepTool(_SearchTool):
"properties": {
"pattern": {
"type": "string",
"description": "Regex or plain text pattern to search for",
"description": "Regex, or literal text when fixed_strings=true",
"minLength": 1,
},
"path": {
"type": "string",
"description": "File or directory to search in (default '.')",
"description": "Search root (default '.')",
},
"glob": {
"type": "string",
"description": "Optional file filter, e.g. '*.py' or 'tests/**/test_*.py'",
"description": "Path filter, e.g. '*.py' or 'tests/**/test_*.py'",
},
"type": {
"type": "string",
"description": "Optional file type shorthand, e.g. 'py', 'ts', 'md', 'json'",
"description": "File type, e.g. 'py', 'ts', 'md', or 'json'",
},
"pages": {
"type": "string",
"description": "PDF page number or range, e.g. '7' or '101-200' (max 100 pages)",
},
"case_insensitive": {
"type": "boolean",
"description": "Case-insensitive search (default false)",
"description": "Ignore case (default false)",
},
"fixed_strings": {
"type": "boolean",
"description": "Treat pattern as plain text instead of regex (default false)",
"description": "Treat pattern literally (default false)",
},
"output_mode": {
"type": "string",
"enum": ["content", "files_with_matches", "count"],
"description": (
"content: matching lines with optional context; "
"files_with_matches: only matching file paths; "
"count: matching line counts per file. "
"Default: files_with_matches"
"content: matches with context (default); "
"files_with_matches: paths; count: matches per file"
),
},
"context_before": {
"type": "integer",
"description": "Number of lines of context before each match",
"description": "Context lines before a match (default 5)",
"minimum": 0,
"maximum": 20,
},
"context_after": {
"type": "integer",
"description": "Number of lines of context after each match",
"description": "Context lines after a match (default 5)",
"minimum": 0,
"maximum": 20,
},
"max_matches": {
"type": "integer",
"description": (
"Legacy alias for head_limit in content mode"
),
"minimum": 1,
"maximum": 1000,
},
"max_results": {
"type": "integer",
"description": (
"Legacy alias for head_limit in files_with_matches or count mode"
),
"minimum": 1,
"maximum": 1000,
},
"head_limit": {
"type": "integer",
"description": (
"Maximum number of results to return. In content mode this limits "
"matching line blocks; in other modes it limits file entries. "
"Default 250"
),
"description": "Maximum matches or file entries (default 250; 0 for all)",
"minimum": 0,
"maximum": 1000,
},
"offset": {
"type": "integer",
"description": "Skip the first N results before applying head_limit",
"description": "Matches or file entries to skip before head_limit",
"minimum": 0,
"maximum": 100000,
},
@@ -395,19 +393,96 @@ class GrepTool(_SearchTool):
}
@staticmethod
def _format_block(
display_path: str,
lines: list[str],
match_line: int,
def _clip_rendered_line(text: str, match_start: int | None = None) -> str:
limit = GrepTool._MAX_RENDERED_LINE_CHARS
if len(text) <= limit:
return text
marker = "..."
available = limit - len(marker)
if match_start is None:
return text[:available] + marker
start = max(0, match_start - available // 3)
start = min(start, len(text) - available)
end = start + available
prefix = marker if start else ""
suffix = marker if end < len(text) else ""
visible = text[start:end]
if prefix and suffix:
visible = visible[: available - len(marker)]
return prefix + visible + suffix
@staticmethod
def _matching_contexts(
lines: Iterable[LocatedDocumentLine],
regex: re.Pattern[str],
before: int,
after: int,
) -> Iterable[tuple[list[LocatedDocumentLine], int, int]]:
history: deque[LocatedDocumentLine] = deque(maxlen=before)
pending: list[_PendingContextMatch] = []
for line in lines:
if not line.searchable:
continue
still_pending: list[_PendingContextMatch] = []
for item in pending:
item.lines.append(line)
item.remaining_after -= 1
if item.remaining_after == 0:
yield item.lines, item.match_index, item.match_start
else:
still_pending.append(item)
pending = still_pending
match = regex.search(line.text)
if match is not None:
context_lines = [*history, line]
item = _PendingContextMatch(
lines=context_lines,
match_index=len(context_lines) - 1,
match_start=match.start(),
remaining_after=after,
)
if after == 0:
yield item.lines, item.match_index, item.match_start
else:
pending.append(item)
history.append(line)
for item in pending:
yield item.lines, item.match_index, item.match_start
@staticmethod
def _format_block(
display_path: str,
lines: list[LocatedDocumentLine],
match_index: int,
match_start: int = 0,
) -> str:
start = max(1, match_line - before)
end = min(len(lines), match_line + after)
block = [f"{display_path}:{match_line}"]
for line_no in range(start, end + 1):
marker = ">" if line_no == match_line else " "
block.append(f"{marker} {line_no}| {lines[line_no - 1]}")
match_line = lines[match_index]
source_line = match_line.extracted_line
match_locator = match_line.locator
if match_locator.startswith("sheet="):
column = _excel_column(match_line.text[:match_start].count("\t") + 1)
row_match = re.search(r",row=(\d+)$", match_locator)
if row_match:
match_locator += f",cell={column}{row_match.group(1)}"
suffix = f" [{match_locator}]" if match_locator else ""
block = [f"{display_path}:{source_line}{suffix}"]
for index, line in enumerate(lines):
is_match = index == match_index
marker = ">" if is_match else " "
coordinate = str(line.extracted_line)
if line.locator:
coordinate += f" [{line.locator}]"
rendered = GrepTool._clip_rendered_line(
line.text,
match_start if is_match else None,
)
block.append(f"{marker} {coordinate}| {rendered}")
return "\n".join(block)
async def execute(
@@ -416,11 +491,12 @@ class GrepTool(_SearchTool):
path: str = ".",
glob: str | None = None,
type: str | None = None,
pages: str | None = None,
case_insensitive: bool = False,
fixed_strings: bool = False,
output_mode: str = "files_with_matches",
context_before: int = 0,
context_after: int = 0,
output_mode: str = "content",
context_before: int = 5,
context_after: int = 5,
max_matches: int | None = None,
max_results: int | None = None,
head_limit: int | None = None,
@@ -456,6 +532,8 @@ class GrepTool(_SearchTool):
size_truncated = False
skipped_binary = 0
skipped_large = 0
document_errors: list[str] = []
document_continuations: list[str] = []
matching_files: list[str] = []
counts: dict[str, int] = {}
file_mtimes: dict[str, float] = {}
@@ -470,61 +548,109 @@ class GrepTool(_SearchTool):
continue
if not _matches_type(file_path.name, type):
continue
display_path = self._display_path(file_path, root)
with file_path.open("rb") as file:
raw = file.read(max_file_bytes + 1)
if len(raw) > max_file_bytes:
skipped_large += 1
continue
if _is_binary(raw):
try:
file_size = file_path.stat().st_size
except OSError:
skipped_binary += 1
continue
if file_size > max_file_bytes:
skipped_large += 1
continue
try:
mtime = file_path.stat().st_mtime
except OSError:
mtime = 0.0
source_iterator: Iterator[LocatedDocumentLine] | None = None
is_document = file_path.suffix.lower() in _DOCUMENT_EXTENSIONS
try:
content = raw.decode("utf-8")
except UnicodeDecodeError:
if is_document:
source = open_document_line_source(file_path, pages=pages)
if source is None:
skipped_binary += 1
continue
source_iterator = source.lines
source_lines: Iterable[LocatedDocumentLine] = source_iterator
if source.continuation:
document_continuations.append(
f"({display_path}: continue PDF search with "
f"{source.continuation})"
)
else:
with file_path.open("rb") as file:
raw = file.read(max_file_bytes + 1)
if _is_binary(raw):
skipped_binary += 1
continue
try:
content = raw.decode("utf-8")
except UnicodeDecodeError:
skipped_binary += 1
continue
source_lines = (
LocatedDocumentLine(text, line_no, "")
for line_no, text in enumerate(content.splitlines(), 1)
)
file_had_match = False
if output_mode == "content":
contexts = self._matching_contexts(
source_lines,
regex,
context_before,
context_after,
)
for context_lines, match_index, match_start in contexts:
file_had_match = True
seen_content_matches += 1
if seen_content_matches <= offset:
continue
if limit is not None and len(blocks) >= limit:
truncated = True
break
block = self._format_block(
display_path,
context_lines,
match_index,
match_start,
)
extra_sep = 2 if blocks else 0
if result_chars + extra_sep + len(block) > self._MAX_RESULT_CHARS:
size_truncated = True
break
blocks.append(block)
result_chars += extra_sep + len(block)
else:
for line in source_lines:
if not line.searchable or regex.search(line.text) is None:
continue
file_had_match = True
if output_mode == "count":
counts[display_path] = counts.get(display_path, 0) + 1
continue
if display_path not in matching_files:
matching_files.append(display_path)
file_mtimes[display_path] = mtime
break
except Exception as e:
if not is_document:
raise
if target.is_file():
if isinstance(e, PdfPageRangeError):
return ToolResult.error(
f"Error: Invalid PDF page range '{pages}': {e!s}."
)
return ToolResult.error(
f"Error searching document {display_path}: {e!s}"
)
skipped_binary += 1
document_errors.append(f"{display_path}: {e!s}")
continue
lines = content.splitlines()
display_path = self._display_path(file_path, root)
file_had_match = False
for idx, line in enumerate(lines, start=1):
if not regex.search(line):
continue
file_had_match = True
if output_mode == "count":
counts[display_path] = counts.get(display_path, 0) + 1
continue
if output_mode == "files_with_matches":
if display_path not in matching_files:
matching_files.append(display_path)
file_mtimes[display_path] = mtime
break
seen_content_matches += 1
if seen_content_matches <= offset:
continue
if limit is not None and len(blocks) >= limit:
truncated = True
break
block = self._format_block(
display_path,
lines,
idx,
context_before,
context_after,
)
extra_sep = 2 if blocks else 0
if result_chars + extra_sep + len(block) > self._MAX_RESULT_CHARS:
size_truncated = True
break
blocks.append(block)
result_chars += extra_sep + len(block)
finally:
close = getattr(source_iterator, "close", None)
if close is not None:
close()
if output_mode == "count" and file_had_match:
if display_path not in matching_files:
matching_files.append(display_path)
@@ -553,8 +679,8 @@ class GrepTool(_SearchTool):
key=lambda name: (-file_mtimes.get(name, 0.0), name),
)
ordered, truncated = _paginate(ordered_files, limit, offset)
lines = [f"{name}: {counts[name]}" for name in ordered]
result = "\n".join(lines)
count_lines = [f"{name}: {counts[name]}" for name in ordered]
result = "\n".join(count_lines)
else:
if not blocks:
result = f"No matches found for pattern '{pattern}' in {path}"
@@ -564,10 +690,14 @@ class GrepTool(_SearchTool):
notes: list[str] = []
if output_mode == "content" and truncated:
notes.append(
f"(pagination: limit={limit}, offset={offset})"
f"(pagination: limit={limit}, offset={offset}; "
f"use offset={offset + len(blocks)} to continue)"
)
elif output_mode == "content" and size_truncated:
notes.append("(output truncated due to size)")
notes.append(
"(output truncated due to size; "
f"use offset={offset + len(blocks)} to continue)"
)
elif truncated and output_mode in {"count", "files_with_matches"}:
notes.append(
f"(pagination: limit={limit}, offset={offset})"
@@ -580,6 +710,9 @@ class GrepTool(_SearchTool):
notes.append(f"(skipped {skipped_binary} binary/unreadable files)")
if skipped_large:
notes.append(f"(skipped {skipped_large} large files)")
if document_errors:
notes.append(f"(first document error: {document_errors[0]})")
notes.extend(document_continuations[:10])
if output_mode == "count" and counts:
notes.append(
f"(total matches: {sum(counts.values())} in {len(counts)} files)"
nanobot/templates/agent/tool_contract.md
+2 -4
View File
@@ -18,11 +18,9 @@
## Discovery and Reading
- Use `find_files` or `list_dir` to locate workspace paths before `read_file` when a path is uncertain.
- Use `grep` for content search inside the workspace; prefer it over shell grep for ordinary searches.
- `grep` defaults to `output_mode="files_with_matches"`; use `output_mode="content"` for matching lines with context.
- Use `find_files` or `list_dir` for uncertain paths, `grep` for content, and `read_file` for a known path.
- `grep` returns matches with five context lines by default; use `files_with_matches` for paths or `count` for totals.
- Use `fixed_strings=true` for literal keywords containing regex characters.
- Use `output_mode="count"` to size a broad search before reading full matches.
- Use `head_limit` and `offset` to page across large result sets.
- Search tools enforce binary and file-size limits and report skipped files in the result.
nanobot/utils/document.py
+360 -117
View File
@@ -66,6 +66,10 @@ class DocxSafetyError(Exception):
"""Raised when a DOCX table exceeds a parser safety boundary."""
class DocumentExtractionError(Exception):
"""Raised when a document cannot be opened for incremental extraction."""
@dataclass(frozen=True, slots=True)
class PdfExtraction:
text: str
@@ -74,6 +78,24 @@ class PdfExtraction:
end_page: int
@dataclass(frozen=True, slots=True)
class LocatedDocumentLine:
"""One searchable document line with a stable, human-readable locator."""
text: str
extracted_line: int
locator: str
searchable: bool = True
@dataclass(frozen=True, slots=True)
class DocumentLineSource:
"""Incremental document lines plus an optional next PDF page range."""
lines: Iterator[LocatedDocumentLine]
continuation: str | None = None
def extract_text(path: str | Path) -> str | None:
"""Extract text from a file.
@@ -85,13 +107,8 @@ def extract_text(path: str | Path) -> str | None:
or error string for failures.
"""
path = Path(path)
if not path.exists():
return f"[error: file not found: {path}]"
try:
if path.stat().st_size > _MAX_EXTRACT_FILE_SIZE:
return f"[error: file exceeds {_MAX_EXTRACT_FILE_SIZE // (1024 * 1024)} MB limit]"
except OSError as e:
return f"[error: failed to inspect file: {e!s}]"
if error := _extraction_path_error(path):
return error
ext = path.suffix.lower()
@@ -115,6 +132,303 @@ def extract_text(path: str | Path) -> str | None:
return None
def open_document_line_source(
path: str | Path,
*,
pages: str | None = None,
) -> DocumentLineSource | None:
"""Open a document as an incremental stream of extracted lines.
Unlike :func:`extract_text`, this interface does not apply the attachment
text preview limit. Parser/file safety limits still apply. Lines that are
useful only for the rendered document view (for example sheet headers and
blank separators) have ``searchable=False`` so range reads can retain them
without making grep match synthetic text.
"""
path = Path(path)
ext = path.suffix.lower()
if ext not in {".pdf", ".docx", ".xlsx", ".pptx"}:
return None
if error := _extraction_path_error(path):
raise DocumentExtractionError(_clean_extraction_error(error))
if ext == ".pdf":
return _open_pdf_line_source(path, pages)
if ext == ".docx":
return _open_docx_line_source(path)
if ext == ".xlsx":
return _open_xlsx_line_source(path)
return _open_pptx_line_source(path)
def _clean_extraction_error(error: str) -> str:
if error.startswith("[error:") and error.endswith("]"):
return error[len("[error:") : -1].strip()
return error
def _check_office_archive(path: Path) -> None:
if error := _office_archive_error(path):
raise DocumentExtractionError(_clean_extraction_error(error))
def _open_pdf_line_source(path: Path, pages: str | None) -> DocumentLineSource:
try:
from pypdf import PdfReader
reader = PdfReader(path, strict=False)
total_pages = len(reader.pages)
if total_pages == 0:
return DocumentLineSource(iter(()))
start, requested_end = _parse_pdf_page_range(pages, total_pages)
except PdfPageRangeError:
raise
except Exception as e:
raise DocumentExtractionError(f"failed to open PDF: {e!s}") from e
end = min(requested_end, start + _MAX_PDF_ATTACHMENT_PAGES - 1)
continuation = None
if end < total_pages - 1:
next_start = end + 2
next_end = min(end + 1 + _MAX_PDF_ATTACHMENT_PAGES, total_pages)
continuation = f"pages='{next_start}-{next_end}'"
def iter_lines() -> Iterator[LocatedDocumentLine]:
extracted_line = 0
wrote_page = False
for index in range(start, end + 1):
page = reader.pages[index]
contents = page.get_contents()
if contents is not None:
stream_size = len(contents.get_data())
if stream_size > _MAX_PDF_CONTENT_STREAM_SIZE:
raise PdfSafetyError(
f"page {index + 1} content stream exceeds "
f"{_MAX_PDF_CONTENT_STREAM_SIZE // (1024 * 1024)} MB limit"
)
text = (page.extract_text() or "").strip()
if not text:
continue
if wrote_page:
extracted_line += 1
yield LocatedDocumentLine("", extracted_line, "", searchable=False)
extracted_line += 1
yield LocatedDocumentLine(
f"--- Page {index + 1} ---",
extracted_line,
"",
searchable=False,
)
page_line = 0
for text_line in text.splitlines():
extracted_line += 1
if not text_line:
yield LocatedDocumentLine("", extracted_line, "", searchable=False)
continue
page_line += 1
yield LocatedDocumentLine(
text_line,
extracted_line,
f"page={index + 1},line={page_line}",
)
wrote_page = True
return DocumentLineSource(iter_lines(), continuation=continuation)
def _open_xlsx_line_source(path: Path) -> DocumentLineSource:
_check_office_archive(path)
try:
from openpyxl import load_workbook
except ImportError as e:
raise DocumentExtractionError("openpyxl not installed") from e
try:
workbook = load_workbook(path, read_only=True, data_only=True)
except Exception as e:
raise DocumentExtractionError(f"failed to open XLSX: {e!s}") from e
def iter_lines() -> Iterator[LocatedDocumentLine]:
extracted_line = 0
wrote_document_content = False
try:
for sheet_name in workbook.sheetnames:
worksheet = workbook[sheet_name]
wrote_header = False
for row_index, row in enumerate(worksheet.iter_rows(values_only=True), 1):
row_text = "\t".join(
str(cell) if cell is not None else "" for cell in row
)
if not row_text.strip():
continue
if not wrote_header:
if wrote_document_content:
extracted_line += 1
yield LocatedDocumentLine(
"", extracted_line, "", searchable=False
)
extracted_line += 1
yield LocatedDocumentLine(
f"--- Sheet: {sheet_name} ---",
extracted_line,
"",
searchable=False,
)
wrote_header = True
wrote_document_content = True
extracted_line += 1
yield LocatedDocumentLine(
row_text,
extracted_line,
f"sheet={sheet_name!r},row={row_index}",
)
finally:
workbook.close()
return DocumentLineSource(iter_lines())
def _open_pptx_line_source(path: Path) -> DocumentLineSource:
_check_office_archive(path)
try:
from pptx import Presentation as PptxPresentation
except ImportError as e:
raise DocumentExtractionError("python-pptx not installed") from e
try:
presentation = PptxPresentation(str(path))
except Exception as e:
raise DocumentExtractionError(f"failed to open PPTX: {e!s}") from e
def iter_lines() -> Iterator[LocatedDocumentLine]:
extracted_line = 0
wrote_slide = False
for slide_number, slide in enumerate(presentation.slides, 1):
slide_text: list[str] = []
for shape in slide.shapes:
_collect_pptx_shape_text(shape, slide_text)
rendered_lines = [line for text in slide_text for line in text.splitlines()]
if not rendered_lines:
continue
if wrote_slide:
extracted_line += 1
yield LocatedDocumentLine("", extracted_line, "", searchable=False)
extracted_line += 1
yield LocatedDocumentLine(
f"--- Slide {slide_number} ---",
extracted_line,
"",
searchable=False,
)
slide_line = 0
for text_line in rendered_lines:
extracted_line += 1
if not text_line:
yield LocatedDocumentLine("", extracted_line, "", searchable=False)
continue
slide_line += 1
yield LocatedDocumentLine(
text_line,
extracted_line,
f"slide={slide_number},line={slide_line}",
)
wrote_slide = True
return DocumentLineSource(iter_lines())
def _open_docx_line_source(path: Path) -> DocumentLineSource:
_check_office_archive(path)
try:
from docx import Document as DocxDocument
from docx.table import Table, _Cell # pyright: ignore[reportPrivateUsage]
from docx.text.paragraph import Paragraph
except ImportError as e:
raise DocumentExtractionError("python-docx not installed") from e
try:
document = DocxDocument(str(path))
except Exception as e:
raise DocumentExtractionError(f"failed to open DOCX: {e!s}") from e
def iter_lines() -> Iterator[LocatedDocumentLine]:
table_cell_count = 0
def cell_text(cell: _Cell, depth: int) -> str:
parts: list[str] = []
for block in cell.iter_inner_content():
if isinstance(block, Paragraph):
text = " ".join(block.text.split())
if text:
parts.append(text)
elif isinstance(block, Table): # pyright: ignore[reportUnnecessaryIsInstance]
parts.extend(
row.replace("\t", " | ") for row in table_rows(block, depth + 1)
)
return " ".join(parts)
def table_rows(table: Table, depth: int) -> Iterator[str]:
nonlocal table_cell_count
if depth > _MAX_DOCX_TABLE_DEPTH:
raise DocxSafetyError(
f"table nesting exceeds {_MAX_DOCX_TABLE_DEPTH} levels"
)
for row in table.rows:
cells: list[str] = []
for tc in row._tr.tc_lst: # pyright: ignore[reportPrivateUsage]
table_cell_count += 1
if table_cell_count > _MAX_DOCX_TABLE_CELLS:
raise DocxSafetyError(
f"document contains more than {_MAX_DOCX_TABLE_CELLS} table cells"
)
cells.append(cell_text(_Cell(tc, table), depth))
if any(cells):
yield "\t".join(cells)
def blocks() -> Iterator[tuple[str, bool]]:
for block in document.iter_inner_content():
if isinstance(block, Paragraph):
text = block.text.strip()
if text:
yield text, True
continue
if not isinstance(block, Table): # pyright: ignore[reportUnnecessaryIsInstance]
continue
first_row = True
for row_text in table_rows(block, 1):
yield row_text, first_row
first_row = False
extracted_line = 0
paragraph = 0
wrote_content = False
for text, separate in blocks():
if wrote_content and separate:
extracted_line += 1
yield LocatedDocumentLine("", extracted_line, "", searchable=False)
for text_line in text.splitlines():
extracted_line += 1
if not text_line:
yield LocatedDocumentLine("", extracted_line, "", searchable=False)
continue
paragraph += 1
yield LocatedDocumentLine(
text_line,
extracted_line,
f"paragraph={paragraph}",
)
wrote_content = True
return DocumentLineSource(iter_lines())
def _extraction_path_error(path: Path) -> str | None:
if not path.exists():
return f"[error: file not found: {path}]"
try:
if path.stat().st_size > _MAX_EXTRACT_FILE_SIZE:
return f"[error: file exceeds {_MAX_EXTRACT_FILE_SIZE // (1024 * 1024)} MB limit]"
except OSError as e:
return f"[error: failed to inspect file: {e!s}]"
return None
def _extract_pdf(path: Path) -> str:
"""Extract text from PDF using pypdf."""
try:
@@ -170,144 +484,73 @@ def extract_pdf_pages(
def _parse_pdf_page_range(pages: str | None, total_pages: int) -> tuple[int, int]:
if not pages:
return 0, total_pages - 1
page_word = "page" if total_pages == 1 else "pages"
guidance = (
f"document has {total_pages} {page_word}; "
f"use a page number or range within 1-{total_pages}"
)
values = pages.strip().split("-")
if len(values) not in {1, 2}:
raise PdfPageRangeError(f"invalid page range: {pages}")
raise PdfPageRangeError(guidance)
try:
start = int(values[0])
end = int(values[-1])
except ValueError as e:
raise PdfPageRangeError(f"invalid page range: {pages}") from e
raise PdfPageRangeError(guidance) from e
if start < 1 or end < start or start > total_pages:
raise PdfPageRangeError(f"invalid page range: {pages}")
raise PdfPageRangeError(guidance)
return start - 1, min(end, total_pages) - 1
def _extract_docx(path: Path) -> str:
"""Extract text from DOCX using python-docx."""
def _render_document_preview(source: DocumentLineSource) -> str:
"""Render a bounded attachment preview from the canonical line stream."""
collector = _TextCollector(_MAX_TEXT_LENGTH)
iterator = source.lines
first_line = True
try:
from docx import Document as DocxDocument
from docx.table import Table, _Cell # pyright: ignore[reportPrivateUsage]
from docx.text.paragraph import Paragraph
except ImportError:
return "[error: python-docx not installed]"
try:
if error := _office_archive_error(path):
return error
doc = DocxDocument(str(path))
collector = _TextCollector(_MAX_TEXT_LENGTH)
table_cell_count = 0
def cell_text(cell: _Cell, depth: int) -> str:
parts: list[str] = []
for block in cell.iter_inner_content():
if isinstance(block, Paragraph):
text = " ".join(block.text.split())
if text:
parts.append(text)
elif isinstance(block, Table): # pyright: ignore[reportUnnecessaryIsInstance]
parts.extend(row.replace("\t", " | ") for row in table_rows(block, depth + 1))
return " ".join(parts)
def table_rows(table: Table, depth: int) -> Iterator[str]:
nonlocal table_cell_count
if depth > _MAX_DOCX_TABLE_DEPTH:
raise DocxSafetyError(
f"table nesting exceeds {_MAX_DOCX_TABLE_DEPTH} levels"
)
for row in table.rows:
cells: list[str] = []
# row.cells expands w:gridSpan before callers can apply a bound.
# Physical w:tc elements keep malformed documents proportional to XML size.
for tc in row._tr.tc_lst: # pyright: ignore[reportPrivateUsage]
table_cell_count += 1
if table_cell_count > _MAX_DOCX_TABLE_CELLS:
raise DocxSafetyError(
f"document contains more than {_MAX_DOCX_TABLE_CELLS} table cells"
)
cells.append(cell_text(_Cell(tc, table), depth))
if any(cells):
yield "\t".join(cells)
for block in doc.iter_inner_content():
if isinstance(block, Paragraph):
text = block.text.strip()
if text and not collector.add(text, separator="\n\n"):
break
continue
if not isinstance(block, Table): # pyright: ignore[reportUnnecessaryIsInstance]
continue
first_row = True
for row_text in table_rows(block, 1):
separator = "\n\n" if first_row else "\n"
first_row = False
if not collector.add(row_text, separator=separator):
return collector.render()
for line in iterator:
if not first_line and not collector.add("\n"):
break
first_line = False
if line.text and not collector.add(line.text):
break
return collector.render()
finally:
close = getattr(iterator, "close", None)
if close is not None:
close()
def _extract_docx(path: Path) -> str:
"""Extract a bounded DOCX attachment preview."""
try:
return _render_document_preview(_open_docx_line_source(path))
except DocxSafetyError as e:
return f"[error: unsafe DOCX: {e!s}]"
except DocumentExtractionError as e:
return f"[error: {e!s}]"
except Exception as e:
logger.exception("Failed to extract DOCX {}", path)
return f"[error: failed to extract DOCX: {e!s}]"
def _extract_xlsx(path: Path) -> str:
"""Extract text from XLSX using openpyxl."""
"""Extract a bounded XLSX attachment preview."""
try:
from openpyxl import load_workbook
except ImportError:
return "[error: openpyxl not installed]"
try:
if error := _office_archive_error(path):
return error
wb = load_workbook(path, read_only=True, data_only=True)
try:
collector = _TextCollector(_MAX_TEXT_LENGTH)
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
wrote_header = False
for row in ws.iter_rows(values_only=True):
row_text = "\t".join(str(cell) if cell is not None else "" for cell in row)
if row_text.strip():
if not wrote_header:
if not collector.add(
f"--- Sheet: {sheet_name} ---",
separator="\n\n",
):
return collector.render()
wrote_header = True
if not collector.add(row_text, separator="\n"):
return collector.render()
return collector.render()
finally:
wb.close()
return _render_document_preview(_open_xlsx_line_source(path))
except DocumentExtractionError as e:
return f"[error: {e!s}]"
except Exception as e:
logger.exception("Failed to extract XLSX {}", path)
return f"[error: failed to extract XLSX: {e!s}]"
def _extract_pptx(path: Path) -> str:
"""Extract text from PPTX using python-pptx."""
"""Extract a bounded PPTX attachment preview."""
try:
from pptx import Presentation as PptxPresentation
except ImportError:
return "[error: python-pptx not installed]"
try:
if error := _office_archive_error(path):
return error
prs = PptxPresentation(str(path))
collector = _TextCollector(_MAX_TEXT_LENGTH)
for i, slide in enumerate(prs.slides, 1):
slide_text: list[str] = []
for shape in slide.shapes:
_collect_pptx_shape_text(shape, slide_text)
if slide_text:
if not collector.add(
f"--- Slide {i} ---\n" + "\n".join(slide_text),
separator="\n\n",
):
break
return collector.render()
return _render_document_preview(_open_pptx_line_source(path))
except DocumentExtractionError as e:
return f"[error: {e!s}]"
except Exception as e:
logger.exception("Failed to extract PPTX {}", path)
return f"[error: failed to extract PPTX: {e!s}]"
tests/agent/test_context_builder.py
+2
View File
@@ -223,6 +223,8 @@ class TestBundledToolContract:
assert "Use the narrowest structured tool" in content
assert "Do not use `exec` as a universal workaround" in content
assert "## File and Coding Workflows" in content
assert "`grep` returns matches with five context lines by default" in content
assert 'defaults to `output_mode="files_with_matches"`' not in content
assert "apply_patch" in content
assert "acceptance criteria into concrete checks" in content
assert "visual evidence reaches the model" in content
tests/test_document_parsing.py
+34
View File
@@ -10,6 +10,7 @@ from nanobot.utils.document import (
_is_text_extension,
extract_pdf_pages,
extract_text,
open_document_line_source,
)
@@ -87,6 +88,39 @@ class TestExtractText:
result = extract_text(json_file)
assert result == content
def test_pdf_search_lines_expose_page_continuation(
self,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
):
pdf_file = tmp_path / "large.pdf"
pdf_file.write_bytes(b"%PDF")
class _Page:
@staticmethod
def get_contents():
return None
@staticmethod
def extract_text():
return "needle"
class _Reader:
def __init__(self, *_args, **_kwargs):
self.pages = [_Page() for _ in range(250)]
monkeypatch.setattr("pypdf.PdfReader", _Reader)
source = open_document_line_source(pdf_file, pages="101-200")
assert source is not None
iterator = source.lines
next(iterator)
line = next(iterator)
iterator.close()
assert line.locator == "page=101,line=1"
assert source.continuation == "pages='201-250'"
def test_extract_text_xlsx(self, tmp_path: Path):
"""Test extracting text from an .xlsx file."""
from openpyxl import Workbook
tests/tools/test_read_enhancements.py
+105 -10
View File
@@ -8,6 +8,19 @@ import pytest
from nanobot.agent.tools import file_state
from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool
from nanobot.utils.document import (
DocumentExtractionError,
DocumentLineSource,
LocatedDocumentLine,
)
def _document_source(text: str) -> DocumentLineSource:
lines = (
LocatedDocumentLine(line, line_no, "")
for line_no, line in enumerate(text.splitlines(), 1)
)
return DocumentLineSource(lines)
@pytest.fixture(autouse=True)
@@ -220,6 +233,12 @@ class TestReadPdf:
assert "Invalid page range" in result
out_of_bounds = await tool.execute(path=str(pdf_path), pages="99")
assert out_of_bounds == (
"Error: Invalid page range '99': document has 1 page; "
"use a page number or range within 1-1."
)
@pytest.mark.asyncio
async def test_pdf_file_not_found_error(self, tool, tmp_path):
result = await tool.execute(path=str(tmp_path / "nope.pdf"))
@@ -345,7 +364,10 @@ class TestReadOfficeDocuments:
@pytest.mark.asyncio
async def test_docx_returns_extracted_text(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="Title\n\nParagraph 1"):
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source("Title\n\nParagraph 1"),
):
f = tmp_path / "test.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -355,16 +377,69 @@ class TestReadOfficeDocuments:
@pytest.mark.asyncio
async def test_xlsx_returns_extracted_text(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="--- Sheet: Sheet1 ---\nName\tAge\nAlice\t30"):
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source("--- Sheet: Sheet1 ---\nName\tAge\nAlice\t30"),
):
f = tmp_path / "test.xlsx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
assert "Sheet1" in result
assert "Alice" in result
@pytest.mark.asyncio
async def test_office_documents_support_extracted_line_ranges(self, tool, tmp_path):
extracted = "--- Sheet: Sheet1 ---\nName\tAge\nAlice\t30\nBob\t25"
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source(extracted),
):
f = tmp_path / "test.xlsx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f), offset=3, limit=1)
assert "3| Alice\t30" in result
assert "Name\tAge" not in result
assert "Use offset=4 to continue" in result
@pytest.mark.asyncio
async def test_office_range_reaches_beyond_attachment_preview_limit(
self,
tool,
tmp_path,
monkeypatch,
):
from openpyxl import Workbook
from nanobot.utils import document as document_utils
workbook_path = tmp_path / "long.xlsx"
workbook = Workbook()
sheet = workbook.active
for row in range(1, 20):
sheet.append([f"ordinary-row-{row}"])
sheet.append(["late-content"])
workbook.save(workbook_path)
workbook.close()
monkeypatch.setattr(document_utils, "_MAX_TEXT_LENGTH", 50)
preview = document_utils.extract_text(workbook_path)
assert preview is not None
assert "late-content" not in preview
result = await tool.execute(path=str(workbook_path), offset=21, limit=1)
assert "21| late-content" in result
assert "beyond end" not in result
@pytest.mark.asyncio
async def test_pptx_returns_extracted_text(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="--- Slide 1 ---\nWelcome\n--- Slide 2 ---\nContent"):
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source(
"--- Slide 1 ---\nWelcome\n--- Slide 2 ---\nContent"
),
):
f = tmp_path / "test.pptx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -373,7 +448,10 @@ class TestReadOfficeDocuments:
@pytest.mark.asyncio
async def test_docx_missing_library(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="[error: python-docx not installed]"):
with patch(
"nanobot.utils.document.open_document_line_source",
side_effect=DocumentExtractionError("python-docx not installed"),
):
f = tmp_path / "test.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -382,7 +460,10 @@ class TestReadOfficeDocuments:
@pytest.mark.asyncio
async def test_docx_corrupt_file(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="[error: failed to extract DOCX: bad zip]"):
with patch(
"nanobot.utils.document.open_document_line_source",
side_effect=DocumentExtractionError("failed to extract DOCX: bad zip"),
):
f = tmp_path / "test.docx"
f.write_bytes(b"not-a-zip")
result = await tool.execute(path=str(f))
@@ -391,7 +472,7 @@ class TestReadOfficeDocuments:
@pytest.mark.asyncio
async def test_unsupported_extension(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value=None):
with patch("nanobot.utils.document.open_document_line_source", return_value=None):
f = tmp_path / "test.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -400,7 +481,10 @@ class TestReadOfficeDocuments:
@pytest.mark.asyncio
async def test_empty_document_returns_descriptive_message(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value=""):
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source(""),
):
f = tmp_path / "empty.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -415,7 +499,10 @@ class TestOfficeDocTruncation:
@pytest.mark.asyncio
async def test_large_document_truncated(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="x" * 200_000):
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source("x" * 200_000),
):
f = tmp_path / "large.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -424,7 +511,10 @@ class TestOfficeDocTruncation:
@pytest.mark.asyncio
async def test_small_document_not_truncated(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="Hello world"):
with patch(
"nanobot.utils.document.open_document_line_source",
return_value=_document_source("Hello world"),
):
f = tmp_path / "small.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
@@ -433,7 +523,12 @@ class TestOfficeDocTruncation:
@pytest.mark.asyncio
async def test_error_response_not_truncated(self, tool, tmp_path):
with patch("nanobot.utils.document.extract_text", return_value="[error: failed to extract DOCX: something went wrong]"):
with patch(
"nanobot.utils.document.open_document_line_source",
side_effect=DocumentExtractionError(
"failed to extract DOCX: something went wrong"
),
):
f = tmp_path / "bad.docx"
f.write_bytes(b"PK")
result = await tool.execute(path=str(f))
tests/tools/test_search_tools.py
+240 -6
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import os
import re
import time
from pathlib import Path
from types import SimpleNamespace
@@ -124,9 +125,14 @@ async def test_grep_respects_glob_filter_and_context(tmp_path: Path) -> None:
@pytest.mark.asyncio
async def test_grep_defaults_to_files_with_matches(tmp_path: Path) -> None:
async def test_grep_defaults_to_match_context(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "src" / "main.py").write_text("match_here\n", encoding="utf-8")
(tmp_path / "src" / "main.py").write_text(
"\n".join(f"line {line}" for line in range(1, 6))
+ "\nmatch_here\n"
+ "\n".join(f"line {line}" for line in range(7, 13)),
encoding="utf-8",
)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
@@ -134,8 +140,230 @@ async def test_grep_defaults_to_files_with_matches(tmp_path: Path) -> None:
path="src",
)
assert result.splitlines() == ["src/main.py"]
assert "1|" not in result
assert "src/main.py:6" in result
assert " 1| line 1" in result
assert "> 6| match_here" in result
assert " 11| line 11" in result
assert "line 12" not in result
@pytest.mark.asyncio
async def test_grep_searches_xlsx_with_sheet_cell_locator(tmp_path: Path) -> None:
from openpyxl import Workbook
workbook_path = tmp_path / "people.xlsx"
workbook = Workbook()
sheet = workbook.active
sheet.title = "People"
sheet.append(["Name", "Role"])
sheet.append(["Ada", "Engineer"])
workbook.save(workbook_path)
workbook.close()
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="Engineer",
path="people.xlsx",
fixed_strings=True,
)
assert "people.xlsx:3" in result
assert "sheet='People',row=2,cell=B2" in result
assert "Ada\tEngineer" in result
@pytest.mark.asyncio
async def test_grep_searches_docx_with_paragraph_locator(tmp_path: Path) -> None:
from docx import Document
document_path = tmp_path / "notes.docx"
document = Document()
document.add_paragraph("Introduction")
document.add_paragraph("late needle")
document.save(document_path)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="late needle",
path="notes.docx",
fixed_strings=True,
)
assert "notes.docx:3 [paragraph=2]" in result
assert "late needle" in result
@pytest.mark.asyncio
async def test_grep_searches_pptx_with_slide_locator(tmp_path: Path) -> None:
from pptx import Presentation
from pptx.util import Inches
presentation_path = tmp_path / "deck.pptx"
presentation = Presentation()
slide = presentation.slides.add_slide(presentation.slide_layouts[6])
textbox = slide.shapes.add_textbox(
Inches(1), Inches(1), Inches(4), Inches(1)
)
textbox.text_frame.text = "slide needle"
presentation.save(presentation_path)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="slide needle",
path="deck.pptx",
fixed_strings=True,
)
assert "deck.pptx:2 [slide=1,line=1]" in result
assert "slide needle" in result
@pytest.mark.asyncio
async def test_grep_searches_pdf_with_page_locator(tmp_path: Path) -> None:
import fitz
pdf_path = tmp_path / "notes.pdf"
document = fitz.open()
page = document.new_page()
page.insert_text((72, 72), "pdf needle")
document.save(pdf_path)
document.close()
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="pdf needle",
path="notes.pdf",
fixed_strings=True,
)
assert "notes.pdf:2 [page=1,line=1]" in result
assert "pdf needle" in result
@pytest.mark.asyncio
async def test_grep_searches_xlsx_beyond_attachment_preview_limit(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from openpyxl import Workbook
from nanobot.utils import document as document_utils
workbook_path = tmp_path / "long.xlsx"
workbook = Workbook()
sheet = workbook.active
sheet.title = "Data"
for row in range(1, 20):
sheet.append([f"ordinary-row-{row}"])
sheet.append(["late-needle"])
workbook.save(workbook_path)
workbook.close()
monkeypatch.setattr(document_utils, "_MAX_TEXT_LENGTH", 50)
preview = document_utils.extract_text(workbook_path)
assert preview is not None
assert "late-needle" not in preview
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="late-needle",
path="long.xlsx",
fixed_strings=True,
)
assert "late-needle" in result
assert "sheet='Data',row=20,cell=A20" in result
assert "No matches found" not in result
@pytest.mark.asyncio
async def test_grep_keeps_an_oversized_matching_line_visible(tmp_path: Path) -> None:
long_line = "x" * 130_000 + "needle" + "y" * 10_000
(tmp_path / "huge-line.txt").write_text(long_line, encoding="utf-8")
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="needle",
path="huge-line.txt",
fixed_strings=True,
context_before=0,
context_after=0,
)
assert "huge-line.txt:1" in result
assert "needle" in result
assert "No matches found" not in result
assert len(result) < GrepTool._MAX_RESULT_CHARS
@pytest.mark.asyncio
async def test_grep_size_limit_returns_a_resumable_offset(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
content = "\n".join(f"needle-{line}-" + "x" * 80 for line in range(1, 11))
(tmp_path / "many.txt").write_text(content, encoding="utf-8")
monkeypatch.setattr(GrepTool, "_MAX_RESULT_CHARS", 350)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
first = await tool.execute(
pattern="needle",
path="many.txt",
fixed_strings=True,
context_before=0,
context_after=0,
head_limit=10,
)
continuation = re.search(r"use offset=(\d+) to continue", first)
assert continuation is not None
next_offset = int(continuation.group(1))
assert next_offset > 0
second = await tool.execute(
pattern="needle",
path="many.txt",
fixed_strings=True,
context_before=0,
context_after=0,
head_limit=10,
offset=next_offset,
)
first_headers = {
line for line in first.splitlines() if line.startswith("many.txt:")
}
second_headers = {
line for line in second.splitlines() if line.startswith("many.txt:")
}
assert second_headers
assert first_headers.isdisjoint(second_headers)
@pytest.mark.asyncio
async def test_grep_reports_an_invalid_pdf_page_range(tmp_path: Path) -> None:
from pypdf import PdfWriter
pdf_path = tmp_path / "one-page.pdf"
writer = PdfWriter()
writer.add_blank_page(width=100, height=100)
with pdf_path.open("wb") as output:
writer.write(output)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(pattern="needle", path="one-page.pdf", pages="bad")
assert result.startswith("Error: Invalid PDF page range 'bad'")
assert "binary/unreadable" not in result
out_of_bounds = await tool.execute(
pattern="needle",
path="one-page.pdf",
pages="99",
)
assert out_of_bounds == (
"Error: Invalid PDF page range '99': document has 1 page; "
"use a page number or range within 1-1."
)
@pytest.mark.asyncio
@@ -169,6 +397,7 @@ async def test_grep_type_filter_limits_files(tmp_path: Path) -> None:
pattern="needle",
path="src",
type="py",
output_mode="files_with_matches",
)
assert result.splitlines() == ["src/a.py"]
@@ -224,6 +453,7 @@ async def test_grep_files_with_matches_supports_head_limit_and_offset(tmp_path:
result = await tool.execute(
pattern="needle",
path="src",
output_mode="files_with_matches",
head_limit=1,
offset=1,
)
@@ -319,12 +549,16 @@ async def test_grep_uses_a_larger_bounded_limit_for_an_explicit_file(
assert "skipped 1 large files" in capped_result
def test_grep_description_keeps_size_thresholds_implementation_specific(tmp_path: Path) -> None:
def test_grep_schema_is_concise_and_keeps_legacy_aliases_hidden(tmp_path: Path) -> None:
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
properties = tool.parameters["properties"]
assert "limits are enforced by the tool" in tool.description
assert len(tool.description) < 150
assert "2 MB" not in tool.description
assert "100 MB" not in tool.description
assert "head_limit" in properties
assert "max_matches" not in properties
assert "max_results" not in properties
@pytest.mark.asyncio
tests/tools/test_tool_descriptions.py
+18 -6
View File
@@ -30,12 +30,24 @@ def test_coding_tool_descriptions_steer_discovery() -> None:
find_files = FindFilesTool().description.lower()
grep = GrepTool().description.lower()
assert "find_files/list_dir first" in read_file
assert "before editing" in read_file
assert "uploaded non-image attachments are referenced by path" in read_file
assert "only when their contents are needed" in read_file
assert "prefer it over shell find/ls" in find_files
assert "prefer this over shell grep" in grep
assert "text, images, pdfs, and office documents" in read_file
assert "line-numbered" in read_file
assert "targeted ranges" in read_file
assert len(read_file) < 160
assert "workspace paths" in find_files
assert "relative paths" in find_files
assert len(find_files) < 140
assert "pdf, docx, xlsx, and pptx" in grep
assert "five context lines" in grep
assert "source locators" in grep
assert len(grep) < 150
read_pages = ReadFileTool().parameters["properties"]["pages"]["description"].lower()
grep_pages = GrepTool().parameters["properties"]["pages"]["description"].lower()
assert "page number or range" in read_pages
assert "page number or range" in grep_pages
def test_exec_tool_descriptions_are_concise() -> None: