src/cli_textual/tools/grep.py

"""Content search tool — a structured, token-cheap alternative to `grep` via bash.

Pure stdlib (no ripgrep dependency); early-exits once `max_results` matches are
found so it stays fast even on large trees. Read-only and workspace-jailed.
"""
import re
from pathlib import Path

from cli_textual.tools.base import ToolResult
from cli_textual.tools.write_file import _resolve_in_workspace

MAX_CHARS = 8192
_SKIP_DIRS = {".git", "__pycache__", "node_modules", ".venv", ".mypy_cache", ".pytest_cache", "target", "dist", "build"}


async def grep(
    pattern: str,
    path: str = ".",
    ignore_case: bool = False,
    max_results: int = 100,
    workspace_root: Path | None = None,
) -> ToolResult:
    """Search file contents for a regular expression, returning ``file:line: text``.

    Recursively searches under ``path``. Skips VCS/build dirs and binary files.
    Capped at ``max_results`` matches. Path is jailed to the workspace.

    Args:
        pattern: Regular expression to search for.
        path: File or directory to search under (relative to workspace or absolute).
        ignore_case: Case-insensitive match when True.
        max_results: Stop after this many matching lines (default 100).
    """
    try:
        regex = re.compile(pattern, re.IGNORECASE if ignore_case else 0)
    except re.error as exc:
        return ToolResult(output=f"Error: invalid regex: {exc}", is_error=True)

    try:
        workspace, base = _resolve_in_workspace(path, workspace_root)
    except PermissionError as exc:
        return ToolResult(output=f"Error: {exc}", is_error=True)
    if not base.exists():
        return ToolResult(output=f"Error: path not found: {path}", is_error=True)

    files = [base] if base.is_file() else sorted(
        p for p in base.rglob("*")
        if p.is_file() and not any(part in _SKIP_DIRS for part in p.relative_to(base).parts)
    )

    matches: list[str] = []
    for f in files:
        try:
            text = f.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError):
            continue  # skip binary / unreadable
        rel = f.relative_to(workspace)
        for n, line in enumerate(text.splitlines(), 1):
            if regex.search(line):
                matches.append(f"{rel}:{n}: {line.strip()[:200]}")
                if len(matches) >= max_results:
                    break
        if len(matches) >= max_results:
            break

    if not matches:
        return ToolResult(output=f"No matches for {pattern!r}")
    out = "\n".join(matches)
    truncated = ""
    if len(out) > MAX_CHARS:
        out = out[:MAX_CHARS]
        truncated = "\n[truncated]"
    note = f"\n[stopped at {max_results} matches]" if len(matches) >= max_results else ""
    return ToolResult(output=out + truncated + note)