| 1 | """Search tools: grep and glob.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import fnmatch |
| 6 | import os |
| 7 | import re |
| 8 | from pathlib import Path, PurePosixPath |
| 9 | from typing import Any, Iterable, TypeVar |
| 10 | |
| 11 | from nanobot.agent.tools.filesystem import ListDirTool, _FsTool |
| 12 | |
| 13 | _DEFAULT_HEAD_LIMIT = 250 |
| 14 | T = TypeVar("T") |
| 15 | _TYPE_GLOB_MAP = { |
| 16 | "py": ("*.py", "*.pyi"), |
| 17 | "python": ("*.py", "*.pyi"), |
| 18 | "js": ("*.js", "*.jsx", "*.mjs", "*.cjs"), |
| 19 | "ts": ("*.ts", "*.tsx", "*.mts", "*.cts"), |
| 20 | "tsx": ("*.tsx",), |
| 21 | "jsx": ("*.jsx",), |
| 22 | "json": ("*.json",), |
| 23 | "md": ("*.md", "*.mdx"), |
| 24 | "markdown": ("*.md", "*.mdx"), |
| 25 | "go": ("*.go",), |
| 26 | "rs": ("*.rs",), |
| 27 | "rust": ("*.rs",), |
| 28 | "java": ("*.java",), |
| 29 | "sh": ("*.sh", "*.bash"), |
| 30 | "yaml": ("*.yaml", "*.yml"), |
| 31 | "yml": ("*.yaml", "*.yml"), |
| 32 | "toml": ("*.toml",), |
| 33 | "sql": ("*.sql",), |
| 34 | "html": ("*.html", "*.htm"), |
| 35 | "css": ("*.css", "*.scss", "*.sass"), |
| 36 | } |
| 37 | |
| 38 | |
| 39 | def _normalize_pattern(pattern: str) -> str: |
| 40 | return pattern.strip().replace("\\", "/") |
| 41 | |
| 42 | |
| 43 | def _match_glob(rel_path: str, name: str, pattern: str) -> bool: |
| 44 | normalized = _normalize_pattern(pattern) |
| 45 | if not normalized: |
| 46 | return False |
| 47 | if "/" in normalized or normalized.startswith("**"): |
| 48 | return PurePosixPath(rel_path).match(normalized) |
| 49 | return fnmatch.fnmatch(name, normalized) |
| 50 | |
| 51 | |
| 52 | def _is_binary(raw: bytes) -> bool: |
| 53 | if b"\x00" in raw: |
| 54 | return True |
| 55 | sample = raw[:4096] |
| 56 | if not sample: |
| 57 | return False |
| 58 | non_text = sum(byte < 9 or 13 < byte < 32 for byte in sample) |
| 59 | return (non_text / len(sample)) > 0.2 |
| 60 | |
| 61 | |
| 62 | def _paginate(items: list[T], limit: int | None, offset: int) -> tuple[list[T], bool]: |
| 63 | if limit is None: |
| 64 | return items[offset:], False |
| 65 | sliced = items[offset : offset + limit] |
| 66 | truncated = len(items) > offset + limit |
| 67 | return sliced, truncated |
| 68 | |
| 69 | |
| 70 | def _pagination_note(limit: int | None, offset: int, truncated: bool) -> str | None: |
| 71 | if truncated: |
| 72 | if limit is None: |
| 73 | return f"(pagination: offset={offset})" |
| 74 | return f"(pagination: limit={limit}, offset={offset})" |
| 75 | if offset > 0: |
| 76 | return f"(pagination: offset={offset})" |
| 77 | return None |
| 78 | |
| 79 | |
| 80 | def _matches_type(name: str, file_type: str | None) -> bool: |
| 81 | if not file_type: |
| 82 | return True |
| 83 | lowered = file_type.strip().lower() |
| 84 | if not lowered: |
| 85 | return True |
| 86 | patterns = _TYPE_GLOB_MAP.get(lowered, (f"*.{lowered}",)) |
| 87 | return any(fnmatch.fnmatch(name.lower(), pattern.lower()) for pattern in patterns) |
| 88 | |
| 89 | |
| 90 | class _SearchTool(_FsTool): |
| 91 | _IGNORE_DIRS = set(ListDirTool._IGNORE_DIRS) |
| 92 | |
| 93 | def _display_path(self, target: Path, root: Path) -> str: |
| 94 | if self._workspace: |
| 95 | try: |
| 96 | return target.relative_to(self._workspace).as_posix() |
| 97 | except ValueError: |
| 98 | pass |
| 99 | return target.relative_to(root).as_posix() |
| 100 | |
| 101 | def _iter_files(self, root: Path) -> Iterable[Path]: |
| 102 | if root.is_file(): |
| 103 | yield root |
| 104 | return |
| 105 | |
| 106 | for dirpath, dirnames, filenames in os.walk(root): |
| 107 | dirnames[:] = sorted(d for d in dirnames if d not in self._IGNORE_DIRS) |
| 108 | current = Path(dirpath) |
| 109 | for filename in sorted(filenames): |
| 110 | yield current / filename |
| 111 | |
| 112 | def _iter_entries( |
| 113 | self, |
| 114 | root: Path, |
| 115 | *, |
| 116 | include_files: bool, |
| 117 | include_dirs: bool, |
| 118 | ) -> Iterable[Path]: |
| 119 | if root.is_file(): |
| 120 | if include_files: |
| 121 | yield root |
| 122 | return |
| 123 | |
| 124 | for dirpath, dirnames, filenames in os.walk(root): |
| 125 | dirnames[:] = sorted(d for d in dirnames if d not in self._IGNORE_DIRS) |
| 126 | current = Path(dirpath) |
| 127 | if include_dirs: |
| 128 | for dirname in dirnames: |
| 129 | yield current / dirname |
| 130 | if include_files: |
| 131 | for filename in sorted(filenames): |
| 132 | yield current / filename |
| 133 | |
| 134 | |
| 135 | class GlobTool(_SearchTool): |
| 136 | """Find files matching a glob pattern.""" |
| 137 | |
| 138 | @property |
| 139 | def name(self) -> str: |
| 140 | return "glob" |
| 141 | |
| 142 | @property |
| 143 | def description(self) -> str: |
| 144 | return ( |
| 145 | "Find files matching a glob pattern (e.g. '*.py', 'tests/**/test_*.py'). " |
| 146 | "Results are sorted by modification time (newest first). " |
| 147 | "Skips .git, node_modules, __pycache__, and other noise directories." |
| 148 | ) |
| 149 | |
| 150 | @property |
| 151 | def read_only(self) -> bool: |
| 152 | return True |
| 153 | |
| 154 | @property |
| 155 | def parameters(self) -> dict[str, Any]: |
| 156 | return { |
| 157 | "type": "object", |
| 158 | "properties": { |
| 159 | "pattern": { |
| 160 | "type": "string", |
| 161 | "description": "Glob pattern to match, e.g. '*.py' or 'tests/**/test_*.py'", |
| 162 | "minLength": 1, |
| 163 | }, |
| 164 | "path": { |
| 165 | "type": "string", |
| 166 | "description": "Directory to search from (default '.')", |
| 167 | }, |
| 168 | "max_results": { |
| 169 | "type": "integer", |
| 170 | "description": "Legacy alias for head_limit", |
| 171 | "minimum": 1, |
| 172 | "maximum": 1000, |
| 173 | }, |
| 174 | "head_limit": { |
| 175 | "type": "integer", |
| 176 | "description": "Maximum number of matches to return (default 250)", |
| 177 | "minimum": 0, |
| 178 | "maximum": 1000, |
| 179 | }, |
| 180 | "offset": { |
| 181 | "type": "integer", |
| 182 | "description": "Skip the first N matching entries before returning results", |
| 183 | "minimum": 0, |
| 184 | "maximum": 100000, |
| 185 | }, |
| 186 | "entry_type": { |
| 187 | "type": "string", |
| 188 | "enum": ["files", "dirs", "both"], |
| 189 | "description": "Whether to match files, directories, or both (default files)", |
| 190 | }, |
| 191 | }, |
| 192 | "required": ["pattern"], |
| 193 | } |
| 194 | |
| 195 | async def execute( |
| 196 | self, |
| 197 | pattern: str, |
| 198 | path: str = ".", |
| 199 | max_results: int | None = None, |
| 200 | head_limit: int | None = None, |
| 201 | offset: int = 0, |
| 202 | entry_type: str = "files", |
| 203 | **kwargs: Any, |
| 204 | ) -> str: |
| 205 | try: |
| 206 | root = self._resolve(path or ".") |
| 207 | if not root.exists(): |
| 208 | return f"Error: Path not found: {path}" |
| 209 | if not root.is_dir(): |
| 210 | return f"Error: Not a directory: {path}" |
| 211 | |
| 212 | if head_limit is not None: |
| 213 | limit = None if head_limit == 0 else head_limit |
| 214 | elif max_results is not None: |
| 215 | limit = max_results |
| 216 | else: |
| 217 | limit = _DEFAULT_HEAD_LIMIT |
| 218 | include_files = entry_type in {"files", "both"} |
| 219 | include_dirs = entry_type in {"dirs", "both"} |
| 220 | matches: list[tuple[str, float]] = [] |
| 221 | for entry in self._iter_entries( |
| 222 | root, |
| 223 | include_files=include_files, |
| 224 | include_dirs=include_dirs, |
| 225 | ): |
| 226 | rel_path = entry.relative_to(root).as_posix() |
| 227 | if _match_glob(rel_path, entry.name, pattern): |
| 228 | display = self._display_path(entry, root) |
| 229 | if entry.is_dir(): |
| 230 | display += "/" |
| 231 | try: |
| 232 | mtime = entry.stat().st_mtime |
| 233 | except OSError: |
| 234 | mtime = 0.0 |
| 235 | matches.append((display, mtime)) |
| 236 | |
| 237 | if not matches: |
| 238 | return f"No paths matched pattern '{pattern}' in {path}" |
| 239 | |
| 240 | matches.sort(key=lambda item: (-item[1], item[0])) |
| 241 | ordered = [name for name, _ in matches] |
| 242 | paged, truncated = _paginate(ordered, limit, offset) |
| 243 | result = "\n".join(paged) |
| 244 | if note := _pagination_note(limit, offset, truncated): |
| 245 | result += f"\n\n{note}" |
| 246 | return result |
| 247 | except PermissionError as e: |
| 248 | return f"Error: {e}" |
| 249 | except Exception as e: |
| 250 | return f"Error finding files: {e}" |
| 251 | |
| 252 | |
| 253 | class GrepTool(_SearchTool): |
| 254 | """Search file contents using a regex-like pattern.""" |
| 255 | _MAX_RESULT_CHARS = 128_000 |
| 256 | _MAX_FILE_BYTES = 2_000_000 |
| 257 | |
| 258 | @property |
| 259 | def name(self) -> str: |
| 260 | return "grep" |
| 261 | |
| 262 | @property |
| 263 | def description(self) -> str: |
| 264 | return ( |
| 265 | "Search file contents with a regex pattern. " |
| 266 | "Default output_mode is files_with_matches (file paths only); " |
| 267 | "use content mode for matching lines with context. " |
| 268 | "Skips binary and files >2 MB. Supports glob/type filtering." |
| 269 | ) |
| 270 | |
| 271 | @property |
| 272 | def read_only(self) -> bool: |
| 273 | return True |
| 274 | |
| 275 | @property |
| 276 | def parameters(self) -> dict[str, Any]: |
| 277 | return { |
| 278 | "type": "object", |
| 279 | "properties": { |
| 280 | "pattern": { |
| 281 | "type": "string", |
| 282 | "description": "Regex or plain text pattern to search for", |
| 283 | "minLength": 1, |
| 284 | }, |
| 285 | "path": { |
| 286 | "type": "string", |
| 287 | "description": "File or directory to search in (default '.')", |
| 288 | }, |
| 289 | "glob": { |
| 290 | "type": "string", |
| 291 | "description": "Optional file filter, e.g. '*.py' or 'tests/**/test_*.py'", |
| 292 | }, |
| 293 | "type": { |
| 294 | "type": "string", |
| 295 | "description": "Optional file type shorthand, e.g. 'py', 'ts', 'md', 'json'", |
| 296 | }, |
| 297 | "case_insensitive": { |
| 298 | "type": "boolean", |
| 299 | "description": "Case-insensitive search (default false)", |
| 300 | }, |
| 301 | "fixed_strings": { |
| 302 | "type": "boolean", |
| 303 | "description": "Treat pattern as plain text instead of regex (default false)", |
| 304 | }, |
| 305 | "output_mode": { |
| 306 | "type": "string", |
| 307 | "enum": ["content", "files_with_matches", "count"], |
| 308 | "description": ( |
| 309 | "content: matching lines with optional context; " |
| 310 | "files_with_matches: only matching file paths; " |
| 311 | "count: matching line counts per file. " |
| 312 | "Default: files_with_matches" |
| 313 | ), |
| 314 | }, |
| 315 | "context_before": { |
| 316 | "type": "integer", |
| 317 | "description": "Number of lines of context before each match", |
| 318 | "minimum": 0, |
| 319 | "maximum": 20, |
| 320 | }, |
| 321 | "context_after": { |
| 322 | "type": "integer", |
| 323 | "description": "Number of lines of context after each match", |
| 324 | "minimum": 0, |
| 325 | "maximum": 20, |
| 326 | }, |
| 327 | "max_matches": { |
| 328 | "type": "integer", |
| 329 | "description": ( |
| 330 | "Legacy alias for head_limit in content mode" |
| 331 | ), |
| 332 | "minimum": 1, |
| 333 | "maximum": 1000, |
| 334 | }, |
| 335 | "max_results": { |
| 336 | "type": "integer", |
| 337 | "description": ( |
| 338 | "Legacy alias for head_limit in files_with_matches or count mode" |
| 339 | ), |
| 340 | "minimum": 1, |
| 341 | "maximum": 1000, |
| 342 | }, |
| 343 | "head_limit": { |
| 344 | "type": "integer", |
| 345 | "description": ( |
| 346 | "Maximum number of results to return. In content mode this limits " |
| 347 | "matching line blocks; in other modes it limits file entries. " |
| 348 | "Default 250" |
| 349 | ), |
| 350 | "minimum": 0, |
| 351 | "maximum": 1000, |
| 352 | }, |
| 353 | "offset": { |
| 354 | "type": "integer", |
| 355 | "description": "Skip the first N results before applying head_limit", |
| 356 | "minimum": 0, |
| 357 | "maximum": 100000, |
| 358 | }, |
| 359 | }, |
| 360 | "required": ["pattern"], |
| 361 | } |
| 362 | |
| 363 | @staticmethod |
| 364 | def _format_block( |
| 365 | display_path: str, |
| 366 | lines: list[str], |
| 367 | match_line: int, |
| 368 | before: int, |
| 369 | after: int, |
| 370 | ) -> str: |
| 371 | start = max(1, match_line - before) |
| 372 | end = min(len(lines), match_line + after) |
| 373 | block = [f"{display_path}:{match_line}"] |
| 374 | for line_no in range(start, end + 1): |
| 375 | marker = ">" if line_no == match_line else " " |
| 376 | block.append(f"{marker} {line_no}| {lines[line_no - 1]}") |
| 377 | return "\n".join(block) |
| 378 | |
| 379 | async def execute( |
| 380 | self, |
| 381 | pattern: str, |
| 382 | path: str = ".", |
| 383 | glob: str | None = None, |
| 384 | type: str | None = None, |
| 385 | case_insensitive: bool = False, |
| 386 | fixed_strings: bool = False, |
| 387 | output_mode: str = "files_with_matches", |
| 388 | context_before: int = 0, |
| 389 | context_after: int = 0, |
| 390 | max_matches: int | None = None, |
| 391 | max_results: int | None = None, |
| 392 | head_limit: int | None = None, |
| 393 | offset: int = 0, |
| 394 | **kwargs: Any, |
| 395 | ) -> str: |
| 396 | try: |
| 397 | target = self._resolve(path or ".") |
| 398 | if not target.exists(): |
| 399 | return f"Error: Path not found: {path}" |
| 400 | if not (target.is_dir() or target.is_file()): |
| 401 | return f"Error: Unsupported path: {path}" |
| 402 | |
| 403 | flags = re.IGNORECASE if case_insensitive else 0 |
| 404 | try: |
| 405 | needle = re.escape(pattern) if fixed_strings else pattern |
| 406 | regex = re.compile(needle, flags) |
| 407 | except re.error as e: |
| 408 | return f"Error: invalid regex pattern: {e}" |
| 409 | |
| 410 | if head_limit is not None: |
| 411 | limit = None if head_limit == 0 else head_limit |
| 412 | elif output_mode == "content" and max_matches is not None: |
| 413 | limit = max_matches |
| 414 | elif output_mode != "content" and max_results is not None: |
| 415 | limit = max_results |
| 416 | else: |
| 417 | limit = _DEFAULT_HEAD_LIMIT |
| 418 | blocks: list[str] = [] |
| 419 | result_chars = 0 |
| 420 | seen_content_matches = 0 |
| 421 | truncated = False |
| 422 | size_truncated = False |
| 423 | skipped_binary = 0 |
| 424 | skipped_large = 0 |
| 425 | matching_files: list[str] = [] |
| 426 | counts: dict[str, int] = {} |
| 427 | file_mtimes: dict[str, float] = {} |
| 428 | root = target if target.is_dir() else target.parent |
| 429 | |
| 430 | for file_path in self._iter_files(target): |
| 431 | rel_path = file_path.relative_to(root).as_posix() |
| 432 | if glob and not _match_glob(rel_path, file_path.name, glob): |
| 433 | continue |
| 434 | if not _matches_type(file_path.name, type): |
| 435 | continue |
| 436 | |
| 437 | raw = file_path.read_bytes() |
| 438 | if len(raw) > self._MAX_FILE_BYTES: |
| 439 | skipped_large += 1 |
| 440 | continue |
| 441 | if _is_binary(raw): |
| 442 | skipped_binary += 1 |
| 443 | continue |
| 444 | try: |
| 445 | mtime = file_path.stat().st_mtime |
| 446 | except OSError: |
| 447 | mtime = 0.0 |
| 448 | try: |
| 449 | content = raw.decode("utf-8") |
| 450 | except UnicodeDecodeError: |
| 451 | skipped_binary += 1 |
| 452 | continue |
| 453 | |
| 454 | lines = content.splitlines() |
| 455 | display_path = self._display_path(file_path, root) |
| 456 | file_had_match = False |
| 457 | for idx, line in enumerate(lines, start=1): |
| 458 | if not regex.search(line): |
| 459 | continue |
| 460 | file_had_match = True |
| 461 | |
| 462 | if output_mode == "count": |
| 463 | counts[display_path] = counts.get(display_path, 0) + 1 |
| 464 | continue |
| 465 | if output_mode == "files_with_matches": |
| 466 | if display_path not in matching_files: |
| 467 | matching_files.append(display_path) |
| 468 | file_mtimes[display_path] = mtime |
| 469 | break |
| 470 | |
| 471 | seen_content_matches += 1 |
| 472 | if seen_content_matches <= offset: |
| 473 | continue |
| 474 | if limit is not None and len(blocks) >= limit: |
| 475 | truncated = True |
| 476 | break |
| 477 | block = self._format_block( |
| 478 | display_path, |
| 479 | lines, |
| 480 | idx, |
| 481 | context_before, |
| 482 | context_after, |
| 483 | ) |
| 484 | extra_sep = 2 if blocks else 0 |
| 485 | if result_chars + extra_sep + len(block) > self._MAX_RESULT_CHARS: |
| 486 | size_truncated = True |
| 487 | break |
| 488 | blocks.append(block) |
| 489 | result_chars += extra_sep + len(block) |
| 490 | if output_mode == "count" and file_had_match: |
| 491 | if display_path not in matching_files: |
| 492 | matching_files.append(display_path) |
| 493 | file_mtimes[display_path] = mtime |
| 494 | if output_mode in {"count", "files_with_matches"} and file_had_match: |
| 495 | continue |
| 496 | if truncated or size_truncated: |
| 497 | break |
| 498 | |
| 499 | if output_mode == "files_with_matches": |
| 500 | if not matching_files: |
| 501 | result = f"No matches found for pattern '{pattern}' in {path}" |
| 502 | else: |
| 503 | ordered_files = sorted( |
| 504 | matching_files, |
| 505 | key=lambda name: (-file_mtimes.get(name, 0.0), name), |
| 506 | ) |
| 507 | paged, truncated = _paginate(ordered_files, limit, offset) |
| 508 | result = "\n".join(paged) |
| 509 | elif output_mode == "count": |
| 510 | if not counts: |
| 511 | result = f"No matches found for pattern '{pattern}' in {path}" |
| 512 | else: |
| 513 | ordered_files = sorted( |
| 514 | matching_files, |
| 515 | key=lambda name: (-file_mtimes.get(name, 0.0), name), |
| 516 | ) |
| 517 | ordered, truncated = _paginate(ordered_files, limit, offset) |
| 518 | lines = [f"{name}: {counts[name]}" for name in ordered] |
| 519 | result = "\n".join(lines) |
| 520 | else: |
| 521 | if not blocks: |
| 522 | result = f"No matches found for pattern '{pattern}' in {path}" |
| 523 | else: |
| 524 | result = "\n\n".join(blocks) |
| 525 | |
| 526 | notes: list[str] = [] |
| 527 | if output_mode == "content" and truncated: |
| 528 | notes.append( |
| 529 | f"(pagination: limit={limit}, offset={offset})" |
| 530 | ) |
| 531 | elif output_mode == "content" and size_truncated: |
| 532 | notes.append("(output truncated due to size)") |
| 533 | elif truncated and output_mode in {"count", "files_with_matches"}: |
| 534 | notes.append( |
| 535 | f"(pagination: limit={limit}, offset={offset})" |
| 536 | ) |
| 537 | elif output_mode in {"count", "files_with_matches"} and offset > 0: |
| 538 | notes.append(f"(pagination: offset={offset})") |
| 539 | elif output_mode == "content" and offset > 0 and blocks: |
| 540 | notes.append(f"(pagination: offset={offset})") |
| 541 | if skipped_binary: |
| 542 | notes.append(f"(skipped {skipped_binary} binary/unreadable files)") |
| 543 | if skipped_large: |
| 544 | notes.append(f"(skipped {skipped_large} large files)") |
| 545 | if output_mode == "count" and counts: |
| 546 | notes.append( |
| 547 | f"(total matches: {sum(counts.values())} in {len(counts)} files)" |
| 548 | ) |
| 549 | if notes: |
| 550 | result += "\n\n" + "\n".join(notes) |
| 551 | return result |
| 552 | except PermissionError as e: |
| 553 | return f"Error: {e}" |
| 554 | except Exception as e: |
| 555 | return f"Error searching files: {e}" |
| 556 |