9f2ba1b74d
Branch semantics kept where main and PM disagree: update_cmd_deps.py, constraints-termux.txt, the Electron update-api-check module and the post-swap hand-off test stay deleted; the pending-fleet-restart catch-up and the local_runtime tag/download ladder stay retired (PM owns engines). Ported from main onto the branch's shape: profile_scoped_chore for the auto-archive and plugin-update housekeeping chores, the local-runtime cross-process boot lock and residency cap, the checkpoint tmp_pack sweep, the cua daemon-liveness status probe, the remote-served Desktop update flag (posix.sh / windows.ps1), sign-in for env-pinned remote gateways (urlDisabled on RemoteSetupFields), the uvloop extra split (uvicorn without [standard]), and the umask-scoping spawn test. uv.lock regenerated with pm.build_env --lock-only; new utf-8 reads from main switched to utf-8-sig (check-windows-footguns).
966 lines
51 KiB
Python
966 lines
51 KiB
Python
"""Content/file search tier for ``tools.file_operations``.
|
|
|
|
``ShellFileOperations`` inherits ``SearchMixin``; module-level helpers are pure
|
|
(no I/O).
|
|
"""
|
|
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, List, Optional
|
|
|
|
from agent.search_policy import SEARCH_PRUNE_DIR_NAMES
|
|
from tools import interrupt as tool_interrupt
|
|
from tools.file_operations_common import ExecuteResult, SearchMatch, SearchResult
|
|
|
|
_MACOS_TCC_PROTECTED_HOME_DIRS = (
|
|
"Desktop", "Documents", "Downloads", "Library", "Movies", "Music", "Pictures",
|
|
)
|
|
|
|
|
|
def _macos_protected_search_exclusions(
|
|
path: str, *, cwd: Optional[str] = None, home: Optional[str] = None, platform: Optional[str] = None,
|
|
) -> List[str]:
|
|
"""Protected home dirs (relative to ``path``) below a broad macOS search root.
|
|
|
|
Only an ANCESTOR search (``$HOME``, ``/Users``) gets exclusions, so recursive
|
|
tools never trigger unattended TCC prompts; a search rooted inside a
|
|
protected dir stays allowed.
|
|
"""
|
|
if (platform or sys.platform) != "darwin":
|
|
return []
|
|
root = Path(path).expanduser()
|
|
if not root.is_absolute():
|
|
root = Path(cwd or os.getcwd()) / root
|
|
root = Path(os.path.normpath(str(root)))
|
|
home_path = Path(os.path.normpath(str(Path(home or Path.home()).expanduser())))
|
|
exclusions: List[str] = []
|
|
for dirname in _MACOS_TCC_PROTECTED_HOME_DIRS:
|
|
try:
|
|
relative = (home_path / dirname).relative_to(root)
|
|
except ValueError:
|
|
continue
|
|
if relative.parts:
|
|
exclusions.append(relative.as_posix())
|
|
return exclusions
|
|
|
|
|
|
# --- Filename-walk admission: one walk per (backend, root) at a time --------------
|
|
|
|
_FILENAME_SEARCH_ADMISSION = threading.Condition()
|
|
_ACTIVE_FILENAME_SEARCH_ROOTS: set[tuple[str, str, str]] = set()
|
|
_FILENAME_SEARCH_WAIT_SECONDS = 0.05
|
|
|
|
|
|
def _normalized_filename_search_root(env: Any, root: str, fallback_cwd: str) -> str:
|
|
"""Normalize a filename-walk root without resolving remote paths locally."""
|
|
from tools.environments.local import LocalEnvironment, _IS_WINDOWS, _msys_to_windows_path
|
|
|
|
cwd = getattr(env, "cwd", None) or fallback_cwd
|
|
if isinstance(env, LocalEnvironment):
|
|
if _IS_WINDOWS:
|
|
root = _msys_to_windows_path(root)
|
|
cwd = _msys_to_windows_path(cwd)
|
|
if not os.path.isabs(root):
|
|
root = os.path.join(cwd, root)
|
|
return os.path.normcase(os.path.abspath(os.path.normpath(root)))
|
|
if not posixpath.isabs(root):
|
|
root = posixpath.join(cwd, root)
|
|
return posixpath.normpath(root)
|
|
|
|
|
|
def _filename_search_root_keys(env: Any, roots: List[str], fallback_cwd: str) -> tuple[tuple[str, str, str], ...]:
|
|
"""Unique backend/root admission keys in deterministic order."""
|
|
env_type = type(env)
|
|
return tuple(sorted({
|
|
(env_type.__module__, env_type.__qualname__, _normalized_filename_search_root(env, root, fallback_cwd))
|
|
for root in roots}))
|
|
|
|
|
|
def _acquire_filename_search_roots(keys: tuple[tuple[str, str, str], ...]) -> bool:
|
|
"""Atomically claim every key, polling for thread-scoped interruption."""
|
|
with _FILENAME_SEARCH_ADMISSION:
|
|
while any(key in _ACTIVE_FILENAME_SEARCH_ROOTS for key in keys):
|
|
if tool_interrupt.is_interrupted():
|
|
return False
|
|
_FILENAME_SEARCH_ADMISSION.wait(_FILENAME_SEARCH_WAIT_SECONDS)
|
|
if tool_interrupt.is_interrupted():
|
|
return False
|
|
if tool_interrupt.is_interrupted():
|
|
return False
|
|
return tool_interrupt.run_if_not_interrupted(lambda: _ACTIVE_FILENAME_SEARCH_ROOTS.update(keys))
|
|
|
|
|
|
def _release_filename_search_roots(keys: tuple[tuple[str, str, str], ...]) -> None:
|
|
"""Release a completed walk and leave no idle per-root state behind."""
|
|
with _FILENAME_SEARCH_ADMISSION:
|
|
_ACTIVE_FILENAME_SEARCH_ROOTS.difference_update(keys)
|
|
_FILENAME_SEARCH_ADMISSION.notify_all()
|
|
|
|
|
|
_ADMISSION_INTERRUPTED_ERROR = (
|
|
"File search was interrupted while waiting for another filename "
|
|
"search on the same root. Retry when ready.")
|
|
|
|
_SEARCH_TIMEOUT_MARKER_RE = re.compile(r"\n?\[Command timed out after \d+s\]\s*$")
|
|
|
|
|
|
def _search_stdout_and_limit(result: ExecuteResult) -> tuple[str, Optional[str]]:
|
|
"""Return stdout cleaned for parsing and a limit reason for search timeouts."""
|
|
if result.exit_code == 124:
|
|
return _SEARCH_TIMEOUT_MARKER_RE.sub("", result.stdout), "search_timeout"
|
|
return result.stdout, None
|
|
|
|
|
|
# A real rg/grep output line is a whitespace-free path token followed by ``:``
|
|
# (match/count), ``-`` (context), or nothing (files_only); tool diagnostics
|
|
# ("rg: ...", indented carets) never match.
|
|
_SEARCH_OUTPUT_RE = re.compile(r'^([A-Za-z]:)?[^\s:][^\n]*?[:\-]\d|^[^\s:][^\s]*$')
|
|
|
|
|
|
def _split_tool_diagnostics(output: str) -> tuple[str, str]:
|
|
"""Separate rg/grep diagnostic lines from real match output → ``(diagnostics, payload)``.
|
|
``_exec`` merges stderr into stdout; classifying by SHAPE lets the exit-2 guard
|
|
tell a pure failure (no payload) from a partial one (one unreadable file, others
|
|
matched) and guarantees error text is never parsed as a match."""
|
|
diagnostics: list[str] = []
|
|
payload: list[str] = []
|
|
for line in output.split('\n'):
|
|
if not line.strip():
|
|
continue
|
|
# Prefix check first: a match path can contain "-<digit>" (".../pytest-686/...").
|
|
if line.lstrip().startswith(("rg: ", "grep: ")):
|
|
diagnostics.append(line)
|
|
elif line == "--" or _SEARCH_OUTPUT_RE.match(line):
|
|
payload.append(line)
|
|
else:
|
|
diagnostics.append(line)
|
|
return '\n'.join(diagnostics), '\n'.join(payload)
|
|
|
|
|
|
def _parse_search_context_line(line: str) -> tuple[str, int, str] | None:
|
|
"""Parse a ``path-line-content`` context line using the RIGHTMOST numeric
|
|
separator (filenames may contain ``-<digits>-`` segments):
|
|
``dir/file-12-name.py-8-context`` → (``dir/file-12-name.py``, 8, ``context``)."""
|
|
if not line or line == "--":
|
|
return None
|
|
match = None
|
|
for candidate in re.finditer(r'-(\d+)-', line):
|
|
match = candidate
|
|
if match is None or match.start() == 0:
|
|
return None
|
|
return line[:match.start()], int(match.group(1)), line[match.end():]
|
|
|
|
|
|
_REGEX_NEWLINE_ESCAPE_RE = re.compile(r"(?<!\\)(?:\\\\)*\\n")
|
|
|
|
|
|
def _pattern_has_regex_newline(pattern: str) -> bool:
|
|
"""True when a content regex wants to match a newline: a literal newline or a
|
|
``\\n`` escape with an ODD number of backslashes (``\\\\n`` is a literal
|
|
backslash+n and must not count)."""
|
|
return "\n" in pattern or bool(_REGEX_NEWLINE_ESCAPE_RE.search(pattern))
|
|
|
|
|
|
def _is_line_oriented_newline_error(error: Optional[str]) -> bool:
|
|
"""Return True for rg's hard error when multiline mode is required."""
|
|
return bool(error) and "literal \"\\n\" is not allowed" in error and "--multiline" in error
|
|
|
|
|
|
def _maybe_warn_line_oriented_newline_pattern(result: SearchResult, pattern: str) -> SearchResult:
|
|
"""Attach a newline-regex warning only when search found no usable results."""
|
|
if result.total_count != 0 or not _pattern_has_regex_newline(pattern):
|
|
return result
|
|
if result.error and not _is_line_oriented_newline_error(result.error):
|
|
return result
|
|
result.error = None
|
|
result.warning = (
|
|
"0 results found. Note: search_files content search is line-oriented "
|
|
"and does not run ripgrep with -U/--multiline, so `\\n` in the regex "
|
|
"does not match line breaks. Use context=N to inspect neighboring "
|
|
"lines, or escape as `\\\\n` when searching for a literal backslash+n.")
|
|
return result
|
|
|
|
|
|
# Match lines are "file:lineno:content". Windows paths carry a drive letter
|
|
# ("C:\path"), so a naive split(":") breaks — the regex handles both.
|
|
_MATCH_LINE_RE = re.compile(r'^([A-Za-z]:)?(.*?):(\d+):(.*)$')
|
|
|
|
# Output-mode → engine flag (identical for rg and grep).
|
|
_OUTPUT_MODE_FLAGS = {"files_only": "-l", "count": "-c"}
|
|
|
|
|
|
def _parse_search_output(result, output_mode: str, limit: int, offset: int,
|
|
context: int, warning: Optional[str] = None) -> SearchResult:
|
|
"""Parse rg/grep ``| head`` output into a SearchResult (shared by both engines).
|
|
Exit codes: 0=matches, 1=none, 2=error — but both tools return 2 on PARTIAL
|
|
errors (one unreadable file), so an error is surfaced only when exit==2 AND no
|
|
usable payload remains. ``warning`` is attached to files_only/content results."""
|
|
stdout, limit_reason = _search_stdout_and_limit(result)
|
|
diagnostics, payload = _split_tool_diagnostics(stdout)
|
|
if result.exit_code == 2 and not payload.strip():
|
|
error_msg = diagnostics.strip() or result.stdout.strip() or "Search error"
|
|
return SearchResult(error=f"Search failed: {error_msg}", total_count=0)
|
|
lines = [ln for ln in payload.strip().split('\n') if ln]
|
|
if output_mode == "files_only":
|
|
return SearchResult(
|
|
files=lines[offset:offset + limit], total_count=len(lines),
|
|
truncated=bool(limit_reason), limit_reason=limit_reason, warning=warning)
|
|
if output_mode == "count":
|
|
counts = {}
|
|
for line in lines:
|
|
if ':' in line:
|
|
path, n = line.rsplit(':', 1)
|
|
try:
|
|
counts[path] = int(n)
|
|
except ValueError:
|
|
pass
|
|
return SearchResult(
|
|
counts=counts, total_count=sum(counts.values()),
|
|
truncated=bool(limit_reason), limit_reason=limit_reason)
|
|
matches = []
|
|
for line in lines:
|
|
if line == "--":
|
|
continue
|
|
m = _MATCH_LINE_RE.match(line)
|
|
if m:
|
|
matches.append(SearchMatch(
|
|
path=(m.group(1) or '') + m.group(2), line_number=int(m.group(3)), content=m.group(4)[:500],
|
|
))
|
|
continue
|
|
# Context lines only when requested, to avoid false positives on dashy paths.
|
|
if context > 0:
|
|
parsed = _parse_search_context_line(line)
|
|
if parsed:
|
|
matches.append(SearchMatch(path=parsed[0], line_number=parsed[1], content=parsed[2][:500]))
|
|
total = len(matches)
|
|
return SearchResult(
|
|
matches=matches[offset:offset + limit], total_count=total,
|
|
truncated=total > offset + limit or bool(limit_reason), limit_reason=limit_reason, warning=warning,
|
|
)
|
|
|
|
|
|
def _posix_roots(roots: List[str]) -> bool:
|
|
"""Darwin-only: every root is POSIX-shaped (no drive letter / backslash)."""
|
|
return sys.platform == "darwin" and all(
|
|
not re.match(r"^[A-Za-z]:[\\/]", root) and "\\" not in root for root in roots)
|
|
|
|
|
|
class SearchMixin:
|
|
"""File-name and content search via rg with find/grep fallbacks. Requires
|
|
``_exec``, ``_has_command``, ``_expand_path``, ``_escape_shell_arg``,
|
|
``_escape_native_tool_arg``, ``env``, ``cwd``, ``_command_cache``,
|
|
``_rg_resolution_cache`` and ``_rg_modified_capability`` from the host class."""
|
|
|
|
# --- rg resolution --------------------------------------------------------
|
|
|
|
def _resolve_command(self, cmd: str) -> Optional[str]:
|
|
"""Resolve an executable in the command host's namespace. Ordinary commands
|
|
keep the bool hit/miss cache; rg alone caches successful resolved paths and
|
|
re-probes misses so a mid-session install becomes visible (with off-PATH
|
|
Windows candidates: cargo, scoop, winget)."""
|
|
if cmd != "rg":
|
|
return cmd if self._has_command(cmd) else None
|
|
cached = self._rg_resolution_cache.get(cmd)
|
|
if cached:
|
|
return cached
|
|
result = self._exec("command -v rg 2>/dev/null")
|
|
if result.exit_code == 0 and result.stdout.strip():
|
|
resolved = result.stdout.strip().splitlines()[0]
|
|
if resolved == "yes": # compatibility with old boolean-probe fakes
|
|
resolved = "rg"
|
|
self._rg_resolution_cache[cmd] = resolved
|
|
return resolved
|
|
from tools.environments.local import LocalEnvironment, _IS_WINDOWS
|
|
|
|
if _IS_WINDOWS and isinstance(self.env, LocalEnvironment):
|
|
user_profile = os.environ.get("USERPROFILE") or str(Path.home())
|
|
local_app_data = os.environ.get("LOCALAPPDATA")
|
|
scoop = os.environ.get("SCOOP") or os.path.join(user_profile, "scoop")
|
|
candidates = [
|
|
os.path.join(user_profile, ".cargo", "bin", "rg.exe"),
|
|
os.path.join(scoop, "shims", "rg.exe"),
|
|
]
|
|
if local_app_data:
|
|
candidates.append(os.path.join(local_app_data, "Microsoft", "WinGet", "Links", "rg.exe"))
|
|
for candidate in candidates:
|
|
if os.path.isfile(candidate):
|
|
resolved = candidate.replace("\\", "/")
|
|
self._rg_resolution_cache[cmd] = resolved
|
|
return resolved
|
|
return None
|
|
|
|
_RG_VERSION_RE = re.compile(
|
|
r"(?m)^ripgrep\s+((?:0|[1-9]\d*))\."
|
|
r"(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)"
|
|
r"(?:-(?:(?:0|[1-9]\d*)|(?:[0-9A-Za-z-]*[A-Za-z-]"
|
|
r"[0-9A-Za-z-]*))(?:\.(?:(?:0|[1-9]\d*)|"
|
|
r"(?:[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*)))*)?"
|
|
r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?"
|
|
r"(?:\s+\(rev [^)]+\))?\s*$")
|
|
|
|
def _modified_rg_capability_error(self, executable: str) -> Optional[str]:
|
|
"""Cached actionable error unless rg can sort exactly (full SemVer, >= 14)."""
|
|
if executable in self._rg_modified_capability:
|
|
return self._rg_modified_capability[executable]
|
|
result = self._exec(f"{self._quote_executable(executable)} --version", timeout=10)
|
|
match = self._RG_VERSION_RE.search(result.stdout or "")
|
|
if result.exit_code == 0 and match and int(match.group(1)) >= 14:
|
|
error = None
|
|
else:
|
|
error = ("Exact modification-time order requires ripgrep 14 or newer; "
|
|
"upgrade ripgrep or use order='discovery'.")
|
|
self._rg_modified_capability[executable] = error
|
|
return error
|
|
|
|
# --- native rg transport (local POSIX) --------------------------------------
|
|
|
|
def _run_rg_native(self, argv: List[str], fetch_limit: int, timeout: int,
|
|
merge_stderr: bool = False) -> ExecuteResult:
|
|
"""Run ``argv`` (shell-quoted rg words) natively and stop reading after
|
|
``fetch_limit`` lines — the ``| head -n`` of the shell pipeline without the
|
|
two bash spawns. ``shlex.split`` undoes the escaping the builders apply for
|
|
the shell path, so both transports see identical arguments. Exit code and
|
|
stdout follow the shell contract (rg 0/1/2; 124 on timeout with partial
|
|
output; 130 on interrupt), so ``_parse_search_output`` is shared. Once the
|
|
bound is reached rg is killed like ``head`` closing the pipe would.
|
|
``merge_stderr`` mirrors the shell path's stderr handling: merged for content
|
|
search (diagnostics feed the error message), discarded (``2>/dev/null``) for
|
|
file lists and probes."""
|
|
from tools.environments.local import _kill_process_group_posix, _make_run_env
|
|
cwd = getattr(self.env, "cwd", None) or self.cwd
|
|
args = shlex.split(" ".join(argv))
|
|
try:
|
|
proc = subprocess.Popen(
|
|
args, cwd=cwd, env=_make_run_env(self.env.env), stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT if merge_stderr else subprocess.DEVNULL,
|
|
start_new_session=True)
|
|
except OSError as exc:
|
|
return ExecuteResult(stdout=f"rg: {exc}", exit_code=2)
|
|
|
|
# Drain on a thread so a silent rg (huge tree, no hits yet) cannot pin the
|
|
# caller past the deadline or past a /stop; the waiter below owns both.
|
|
lines: List[bytes] = []
|
|
bounded = threading.Event()
|
|
|
|
def _drain() -> None:
|
|
for raw in proc.stdout:
|
|
lines.append(raw)
|
|
if len(lines) >= fetch_limit:
|
|
bounded.set()
|
|
break
|
|
|
|
drainer = threading.Thread(target=_drain, daemon=True)
|
|
drainer.start()
|
|
deadline = time.monotonic() + timeout
|
|
exit_code: Optional[int] = None
|
|
while True:
|
|
drainer.join(0.05)
|
|
if not drainer.is_alive() or bounded.is_set():
|
|
break
|
|
if tool_interrupt.is_interrupted():
|
|
exit_code = 130
|
|
break
|
|
if time.monotonic() > deadline:
|
|
exit_code = 124
|
|
break
|
|
if proc.poll() is None:
|
|
_kill_process_group_posix(proc) # native lane is POSIX-only (gate above)
|
|
proc.wait()
|
|
drainer.join()
|
|
proc.stdout.close()
|
|
stdout = b"".join(lines).decode("utf-8", errors="replace")
|
|
if exit_code == 124:
|
|
return ExecuteResult(stdout=stdout + f"\n[Command timed out after {timeout}s]", exit_code=124)
|
|
if exit_code == 130:
|
|
return ExecuteResult(stdout=stdout + "\n[Command interrupted]", exit_code=130)
|
|
# A killed-at-bound rg reports a signal (negative returncode); head would have
|
|
# left the pipeline at 0 unless rg itself already failed.
|
|
return ExecuteResult(stdout=stdout, exit_code=0 if bounded.is_set() else proc.returncode)
|
|
|
|
def _run_rg_bounded(self, words: List[str], fetch_limit: int, timeout: int, *,
|
|
merge_stderr: bool = False, native_ok: bool = True,
|
|
shell_prefix: str = "") -> ExecuteResult:
|
|
"""Run an rg command (shell-quoted words) and keep the first ``fetch_limit``
|
|
lines: natively on a local POSIX host, else through the backend shell as
|
|
``<prefix><words> | head -n N``. ``native_ok=False`` keeps a form the native
|
|
lane cannot express (the multi-root ``cd`` prefix); ``shell_prefix`` is shell-only."""
|
|
if native_ok and self._native_read_enabled():
|
|
return self._run_rg_native(words, fetch_limit, timeout, merge_stderr=merge_stderr)
|
|
stderr = "" if merge_stderr else " 2>/dev/null"
|
|
return self._exec(f"{shell_prefix}{' '.join(words)}{stderr} | head -n {fetch_limit}", timeout=timeout)
|
|
|
|
def _quote_executable(self, executable: str) -> str:
|
|
"""Quote an executable without leaking controller path semantics."""
|
|
if re.fullmatch(r"[A-Za-z0-9_.-]+", executable):
|
|
return executable
|
|
from tools.environments.local import LocalEnvironment
|
|
|
|
if isinstance(self.env, LocalEnvironment):
|
|
return self._escape_native_tool_arg(executable)
|
|
return "'" + executable.replace("'", "'\"'\"'") + "'"
|
|
|
|
# --- macOS protected-folder exclusions --------------------------------------
|
|
|
|
def _macos_search_exclusions(self, path: str) -> List[str]:
|
|
"""Protected descendants to prune for this search root, if any. Gated on
|
|
``env.is_local``: ``sys.platform``/``_HOME`` describe the CONTROLLER, but the
|
|
search runs on ``env``'s host. Envs without the flag default to local
|
|
semantics; pruning is a warning-carrying skip, never data loss."""
|
|
env = getattr(self, "env", None)
|
|
if env is not None and getattr(env, "is_local", True) is False:
|
|
return []
|
|
from tools import file_operations as _fo # lazy: _HOME is monkeypatched there
|
|
cwd = getattr(self.env, "cwd", None) or self.cwd
|
|
return _macos_protected_search_exclusions(path, cwd=cwd, home=_fo._HOME, platform=sys.platform)
|
|
|
|
def _protected_prune_paths(self, path: str) -> List[str]:
|
|
"""Absolute-ish protected paths for find's ``-path ... -prune``."""
|
|
return [os.path.normpath(os.path.join(path, item)) for item in self._macos_search_exclusions(path)]
|
|
|
|
def _effective_macos_search_exclusions(self, roots: List[str]) -> List[tuple[str, str, str]]:
|
|
"""Unique ``(root, relative, absolute)`` exclusions across ``roots``, never
|
|
pruning a root the caller chose explicitly."""
|
|
cwd = getattr(self.env, "cwd", None) or self.cwd
|
|
use_posix_paths = _posix_roots(roots)
|
|
|
|
def normalized(root: str) -> str:
|
|
if use_posix_paths:
|
|
return posixpath.normpath(root if posixpath.isabs(root) else posixpath.join(cwd, root))
|
|
return os.path.normcase(os.path.abspath(os.path.normpath(root)))
|
|
|
|
normalized_roots = [normalized(root) for root in roots]
|
|
explicit_roots = set(normalized_roots)
|
|
seen = set()
|
|
effective = []
|
|
for root, normalized_root in zip(roots, normalized_roots):
|
|
for relative in self._macos_search_exclusions(root):
|
|
if use_posix_paths:
|
|
absolute = key = posixpath.normpath(posixpath.join(normalized_root, relative))
|
|
else:
|
|
absolute = os.path.normpath(os.path.join(root, relative))
|
|
key = os.path.normcase(os.path.abspath(absolute))
|
|
if key in explicit_roots or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
effective.append((root, relative, absolute))
|
|
return effective
|
|
|
|
@staticmethod
|
|
def _macos_protected_search_warning(paths: List[str]) -> str:
|
|
skipped = ", ".join(os.path.basename(item) for item in paths)
|
|
return ("Skipped macOS protected folders during broad search to avoid "
|
|
f"an unattended privacy prompt: {skipped}. Search a protected "
|
|
"folder directly when access is intentional.")
|
|
|
|
@staticmethod
|
|
def _hidden_prune_expr(q_roots: List[str]) -> str:
|
|
"""find clause pruning hidden dirs while keeping an explicitly selected dot-named root
|
|
(dir or single file) — find echoes each start point as given, so ``! -path`` matches it."""
|
|
exemptions = "".join(f" ! -path {root}" for root in q_roots)
|
|
return f"\\( -type d -name '.*'{exemptions} \\) -prune"
|
|
|
|
def _prune_expr(self, protected_paths: List[str]) -> str:
|
|
"""find ``\\( -path A -o -path B \\) -prune`` clause for the protected dirs."""
|
|
terms = " -o ".join(f"-path {self._escape_shell_arg(item)}" for item in protected_paths)
|
|
return f"\\( {terms} \\) -prune"
|
|
|
|
def _root_under_hidden_dir(self, path: str) -> bool:
|
|
"""True when the search root or any ancestor is dot-named (``~/.hermes/skills``)."""
|
|
root = _normalized_filename_search_root(self.env, path or ".", self.cwd)
|
|
return any(part.startswith(".") and part not in (".", "..") for part in root.replace("\\", "/").split("/"))
|
|
|
|
def _rg_exclusion_globs(self, path: str) -> List[str]:
|
|
"""``--glob '!<dir>/**'`` pairs excluding protected dirs from an rg run."""
|
|
out: List[str] = []
|
|
for item in self._macos_search_exclusions(path):
|
|
out.extend(["--glob", self._escape_shell_arg(f"!{item}/**")])
|
|
return out
|
|
|
|
def _path_exists_probe(self, path: str) -> ExecuteResult:
|
|
"""Existence probe; stdout contains "exists" or "not_found" (or the probe's
|
|
``cwd_error`` when the exec wrapper itself failed)."""
|
|
if self._native_read_enabled():
|
|
full = path if os.path.isabs(path) else os.path.join(getattr(self.env, "cwd", None) or self.cwd, path)
|
|
return ExecuteResult(stdout="exists" if os.path.exists(full) else "not_found")
|
|
return self._exec(f"test -e {self._escape_shell_arg(path)} && echo exists || echo not_found")
|
|
|
|
def _dispatch_search(self, pattern: str, path: str, target: str,
|
|
file_glob: Optional[str], limit: int, offset: int,
|
|
output_mode: str, context: int, order: str = "discovery") -> SearchResult:
|
|
if target == "files":
|
|
return self._search_files(pattern, path, limit, offset, order)
|
|
return self._search_content(pattern, path, file_glob, limit, offset, output_mode, context)
|
|
|
|
def _path_not_found_result(self, path: str) -> SearchResult:
|
|
"""Error result for a missing search root, with nearby-entry suggestions."""
|
|
parent = os.path.dirname(path) or "."
|
|
basename_query = os.path.basename(path)
|
|
hint_parts = [f"Path not found: {path}"]
|
|
parent_check = self._exec(f"test -d {self._escape_shell_arg(parent)} && echo yes || echo no")
|
|
if "yes" in parent_check.stdout and basename_query:
|
|
ls_result = self._exec(f"ls -1 {self._escape_shell_arg(parent)} 2>/dev/null | head -20")
|
|
if ls_result.exit_code == 0 and ls_result.stdout.strip():
|
|
lq = basename_query.lower()
|
|
candidates = [
|
|
os.path.join(parent, e) for e in ls_result.stdout.strip().split('\n')
|
|
if e and (lq in e.lower() or e.lower() in lq or e.lower().startswith(lq[:3]))]
|
|
if candidates:
|
|
hint_parts.append("Similar paths: " + ", ".join(candidates[:5]))
|
|
return SearchResult(error=". ".join(hint_parts), total_count=0)
|
|
|
|
def _try_multi_path_search(self, pattern: str, path: str, target: str,
|
|
file_glob: Optional[str], limit: int, offset: int,
|
|
output_mode: str, context: int,
|
|
order: str = "discovery") -> Optional[SearchResult]:
|
|
"""Recover a not-found ``path`` that is really several paths in one string.
|
|
Commas explicitly delimit paths (internal spaces preserved); without commas
|
|
split on whitespace. Search every existing part, merge, and note skipped
|
|
parts. None when it doesn't look like a multi-path string."""
|
|
if "," in path:
|
|
parts = [part.strip() for part in path.split(",") if part.strip()]
|
|
else:
|
|
parts = path.split()
|
|
if len(parts) < 2:
|
|
return None
|
|
existing, missing = [], []
|
|
for p in parts:
|
|
expanded = self._expand_path(p)
|
|
(existing if "exists" in self._path_exists_probe(expanded).stdout else missing).append(expanded)
|
|
if not existing:
|
|
return None
|
|
if target == "files":
|
|
# One global traversal across roots so modified ordering and pagination
|
|
# are exact; root admission wraps the actual rg/find invocation.
|
|
merged = self._search_files(pattern, existing, limit, offset, order)
|
|
else:
|
|
merged = SearchResult()
|
|
for root in existing:
|
|
sub = self._search_content(pattern, root, file_glob, limit, offset, output_mode, context)
|
|
if sub.error:
|
|
return sub
|
|
merged.matches.extend(sub.matches)
|
|
merged.files.extend(sub.files)
|
|
merged.counts.update(sub.counts)
|
|
merged.total_count += sub.total_count
|
|
merged.truncated = merged.truncated or sub.truncated
|
|
merged.matches = merged.matches[:limit]
|
|
merged.files = merged.files[:limit]
|
|
note = f"path contained {len(parts)} entries; searched {len(existing)} that exist"
|
|
if missing:
|
|
note += "; skipped missing: " + ", ".join(missing[:3])
|
|
if len(missing) > 3:
|
|
note += f" (+{len(missing) - 3} more)"
|
|
warning_parts = [note]
|
|
if not merged.error:
|
|
protected_paths = [absolute for _r, _rel, absolute in self._effective_macos_search_exclusions(existing)]
|
|
if protected_paths:
|
|
warning_parts.append(self._macos_protected_search_warning(protected_paths))
|
|
merged.warning = " ".join(warning_parts)
|
|
return merged
|
|
|
|
def _search_prune_glob_args(self) -> str:
|
|
"""rg globs pruning known heavyweight recursive subtrees. Both forms are
|
|
needed: globs are relative to each rg root, so ``**/name/**`` alone misses an
|
|
explicitly selected ``name/`` root. Names come from the shared scan policy —
|
|
no second search-only list."""
|
|
globs = []
|
|
for dirname in sorted(SEARCH_PRUNE_DIR_NAMES):
|
|
for prefix in ("", "**/"):
|
|
globs.extend(("--glob", self._escape_shell_arg(f"!{prefix}{dirname}/**")))
|
|
return " ".join(globs)
|
|
|
|
# (rg flags, message template) probes for a 0-match content search, in order.
|
|
# The fixed-string probe only runs when the pattern has regex metacharacters.
|
|
_ZERO_MATCH_PROBES = (
|
|
("-i", "0 exact matches, but {total} case-insensitive match(es) in {n} file(s): "
|
|
"{paths} — the pattern's casing may be wrong."),
|
|
# rg skips dotdirs and .gitignore'd files by default; say so instead of a bare zero.
|
|
("--hidden --no-ignore", "0 matches in visible files, but {total} match(es) in {n} "
|
|
"hidden or gitignored file(s): {paths} — these are excluded by default."),
|
|
("-F", "0 regex matches, but {total} literal match(es) in {n} file(s): {paths} — the "
|
|
"pattern contains regex metacharacters that likely need escaping "
|
|
"(or pass a simpler substring)."),
|
|
)
|
|
|
|
def _zero_match_probe(self, pattern: str, path: str, file_glob: Optional[str]) -> Optional[str]:
|
|
"""Steering hint for a 0-match content search, or None: a bare zero gives the
|
|
model nothing to act on, so run cheap count-only rg probes (case-insensitive,
|
|
hidden/ignored, fixed-string) and report the first that hits."""
|
|
rg_executable = self._resolve_command('rg')
|
|
if not rg_executable:
|
|
return None
|
|
rg = self._quote_executable(rg_executable)
|
|
has_meta = bool(re.search(r"[.\[\](){}?*+^$\\|]", pattern))
|
|
glob_expr = f" --glob {self._escape_shell_arg(file_glob)}" if file_glob else ""
|
|
for flags, template in self._ZERO_MATCH_PROBES:
|
|
if flags == "-F" and not has_meta:
|
|
continue
|
|
# The hidden/ignored probe keeps --no-ignore so project-local ignored
|
|
# files stay diagnosable, but prunes heavyweight trees before rg recurses.
|
|
if flags.startswith("--hidden"):
|
|
glob_expr_probe = f"{glob_expr} {self._search_prune_glob_args()}"
|
|
else:
|
|
glob_expr_probe = glob_expr
|
|
probe_words = [rg, flags, "--count-matches", glob_expr_probe,
|
|
self._escape_shell_arg(pattern, translate_path=False), self._escape_native_tool_arg(path)]
|
|
probe = self._run_rg_bounded(probe_words, 50, timeout=30)
|
|
total, per_file = 0, []
|
|
for line in (probe.stdout or "").strip().splitlines():
|
|
p, _sep, n = line.rpartition(":")
|
|
if n.isdigit():
|
|
total += int(n)
|
|
per_file.append(p)
|
|
if total > 0:
|
|
extra = len(per_file) - 5
|
|
paths = ", ".join(per_file[:5]) + (f" (+{extra} more)" if extra > 0 else "")
|
|
return template.format(total=total, n=len(per_file), paths=paths)
|
|
return None
|
|
|
|
def _is_broad_local_search_root(self, path: str) -> bool:
|
|
"""Whether a no-rg LOCAL root (filesystem root, $HOME or an ancestor of it) is
|
|
unsafe for recursive find. Controller paths never classify remotes."""
|
|
from tools.environments.local import LocalEnvironment, _IS_WINDOWS, _msys_to_windows_path
|
|
|
|
if not isinstance(self.env, LocalEnvironment):
|
|
return False
|
|
|
|
def normalized(value: str) -> str:
|
|
if _IS_WINDOWS:
|
|
value = _msys_to_windows_path(value).replace("\\", "/")
|
|
if not os.path.isabs(value):
|
|
value = os.path.join(getattr(self.env, "cwd", None) or self.cwd, value)
|
|
# Classify the linked target, not the link: ``find -H`` now follows an
|
|
# operand symlink, so a link pointing at $HOME (or at the filesystem root)
|
|
# must not slip a recursive find past this guard (#116270). Local-only by
|
|
# the isinstance check above, so this resolves on the host that runs find.
|
|
return os.path.normcase(os.path.realpath(value))
|
|
|
|
from tools import file_operations as _fo # lazy: _HOME is monkeypatched there
|
|
root = normalized(path)
|
|
home = normalized(_fo._HOME)
|
|
drive = os.path.splitdrive(root)[0]
|
|
anchor = drive + os.sep if drive else os.path.abspath(os.sep)
|
|
if root == os.path.normcase(anchor):
|
|
return True
|
|
try:
|
|
common = os.path.commonpath([root, home])
|
|
except ValueError:
|
|
return False
|
|
return root == home or common == root
|
|
|
|
def _search_files(self, pattern: str, path: str | List[str], limit: int, offset: int,
|
|
order: str = "discovery") -> SearchResult:
|
|
"""Search for files by name (glob-like) across one or more roots: rg --files,
|
|
else a bounded find. ``order``: "discovery" (fast, bounded) or "modified"
|
|
(exact global newest-first; needs rg 14+ or GNU find)."""
|
|
search_pattern = pattern if (not pattern.startswith('**/') and '/' not in pattern) \
|
|
else pattern.split('/')[-1]
|
|
roots = [path] if isinstance(path, str) else path
|
|
if not roots:
|
|
return SearchResult(error="File search requires at least one search root in 'path'.")
|
|
|
|
# Prefer ripgrep: bounded parallel traversal with ignore semantics. Resolve
|
|
# the engine and exact-order capability BEFORE admission so a queued request
|
|
# does not occupy a root while doing command discovery.
|
|
if self._has_command("rg"):
|
|
rg_executable = self._resolve_command("rg") or "rg"
|
|
if order == "modified":
|
|
capability_error = self._modified_rg_capability_error(rg_executable)
|
|
if capability_error:
|
|
return SearchResult(error=capability_error)
|
|
keys = _filename_search_root_keys(self.env, roots, self.cwd)
|
|
if not _acquire_filename_search_roots(keys):
|
|
return SearchResult(error=_ADMISSION_INTERRUPTED_ERROR)
|
|
try:
|
|
return self._search_files_rg(search_pattern, path, limit, offset, order,
|
|
rg_executable=rg_executable)
|
|
finally:
|
|
_release_filename_search_roots(keys)
|
|
|
|
# A local find rooted at/above $HOME or a filesystem root can take minutes and
|
|
# prompt on protected paths: refuse before invoking find.
|
|
if any(self._is_broad_local_search_root(root) for root in roots):
|
|
return SearchResult(error=(
|
|
"Broad local file search without ripgrep is disabled because "
|
|
"find cannot keep this traversal safely bounded. Install "
|
|
"ripgrep or search a narrower directory."))
|
|
if not self._has_command("find"):
|
|
return SearchResult(
|
|
error="File search requires 'rg' (ripgrep) or 'find'. "
|
|
"Install ripgrep for best results: "
|
|
"https://github.com/BurntSushi/ripgrep#installation")
|
|
|
|
# Prune hidden descendant dirs (and hidden files, matching rg's default) while
|
|
# still allowing an explicitly selected hidden root; dash-prefixed roots get
|
|
# ``./`` so find doesn't parse them as options.
|
|
find_roots = [f"./{root}" if root.startswith("-") else root for root in roots]
|
|
q_roots = [self._escape_shell_arg(root) for root in find_roots]
|
|
hidden_prune = f" {self._hidden_prune_expr(q_roots)} -o"
|
|
protected_paths = [absolute for _r, _rel, absolute in self._effective_macos_search_exclusions(roots)]
|
|
protected_prune = f" {self._prune_expr(protected_paths)} -o" if protected_paths else ""
|
|
fetch_limit = offset + limit + 1
|
|
# ``-H`` follows a symlink handed in as an OPERAND, and only an operand: without
|
|
# it ``find <link> -type f`` tests the link itself, so ``target="files"`` listed
|
|
# nothing at all for a symlinked root - total_count: 0, no error, no warning,
|
|
# indistinguishable from an empty directory - while ``rg --files`` followed the
|
|
# same argument (#116270). Following the operand inside the command is also what
|
|
# covers a link that only exists on the execution host (SSH/container), with no
|
|
# probe of its own.
|
|
base = (f"find -H {' '.join(q_roots)}{protected_prune}{hidden_prune} -type f "
|
|
f"! -name '.*' -name {self._escape_shell_arg(search_pattern)}")
|
|
if order == "modified":
|
|
cmd = "set -o pipefail; " + base + f" -printf '%T@ %p\\n' 2>/dev/null | sort -rn | head -n {fetch_limit}"
|
|
else:
|
|
cmd = "set -o pipefail; " + base + f" -print 2>/dev/null | head -n {fetch_limit}"
|
|
|
|
keys = _filename_search_root_keys(self.env, roots, self.cwd)
|
|
if not _acquire_filename_search_roots(keys):
|
|
return SearchResult(error=_ADMISSION_INTERRUPTED_ERROR)
|
|
try:
|
|
result = self._exec(cmd, timeout=60)
|
|
finally:
|
|
_release_filename_search_roots(keys)
|
|
stdout, limit_reason = _search_stdout_and_limit(result)
|
|
|
|
# Parse BEFORE classifying exit 141: under pipefail a bounded producer gets
|
|
# SIGPIPE when head closes after fetch_limit rows — benign only when the
|
|
# payload proves the bound was reached; a shorter payload is a hard failure.
|
|
raw_files: List[str] = []
|
|
for line in stdout.splitlines():
|
|
if order == "modified":
|
|
parts = line.split(" ", 1)
|
|
if len(parts) != 2 or not parts[0].replace(".", "", 1).isdigit():
|
|
continue
|
|
raw_files.append(parts[1])
|
|
elif line:
|
|
raw_files.append(line)
|
|
bounded_sigpipe = result.exit_code == 141 and len(raw_files) >= fetch_limit
|
|
if result.exit_code not in {0, 124} and not bounded_sigpipe:
|
|
if order == "modified":
|
|
return SearchResult(error=(
|
|
"Exact modification-time order requires GNU find with "
|
|
"-printf support; install ripgrep 14+ or use order='discovery'."))
|
|
return SearchResult(error="File search failed while running bounded find traversal.")
|
|
|
|
from tools.environments.local import LocalEnvironment, _IS_WINDOWS, _msys_to_windows_path
|
|
if _IS_WINDOWS and isinstance(self.env, LocalEnvironment):
|
|
raw_files = [_msys_to_windows_path(file_path) for file_path in raw_files]
|
|
return SearchResult(
|
|
files=raw_files[offset:offset + limit], total_count=len(raw_files),
|
|
truncated=len(raw_files) > offset + limit or bool(limit_reason), limit_reason=limit_reason)
|
|
|
|
def _search_files_rg(self, pattern: str, path: str | List[str], limit: int, offset: int,
|
|
order: str = "discovery", rg_executable: Optional[str] = None) -> SearchResult:
|
|
"""File-name search via ``rg --files`` (respects .gitignore, skips hidden dirs,
|
|
parallel walk). Discovery order stays bounded and fast; exact modification-time
|
|
ordering is explicit because it scans globally."""
|
|
# Wrap bare names so -g matches at any depth (equivalent to find -name).
|
|
glob_pattern = f"*{pattern}" if ('/' not in pattern and not pattern.startswith('*')) else pattern
|
|
roots = [path] if isinstance(path, str) else path
|
|
fetch_limit = limit + offset + 1
|
|
effective_exclusions = self._effective_macos_search_exclusions(roots)
|
|
scoped_common = None
|
|
command_roots = roots
|
|
if len(roots) > 1 and effective_exclusions and _posix_roots(roots):
|
|
# Several roots: rg globs are root-relative, so cd to the common ancestor
|
|
# and express roots + exclusions relative to it.
|
|
cwd = getattr(self.env, "cwd", None) or self.cwd
|
|
absolute_roots = [
|
|
posixpath.normpath(root if posixpath.isabs(root) else posixpath.join(cwd, root))
|
|
for root in roots]
|
|
scoped_common = posixpath.commonpath(absolute_roots)
|
|
command_roots = [posixpath.relpath(root, scoped_common) for root in absolute_roots]
|
|
exclusion_terms = [
|
|
f"--glob {self._escape_shell_arg(f'!{posixpath.relpath(absolute, scoped_common)}/**')}"
|
|
for _r, _rel, absolute in effective_exclusions]
|
|
else:
|
|
exclusion_terms = [
|
|
f"--glob {self._escape_shell_arg(f'!{relative}/**')}"
|
|
for _r, relative, _abs in effective_exclusions]
|
|
exclusion_globs = " ".join(dict.fromkeys(exclusion_terms))
|
|
exclusion_args = f" {exclusion_globs}" if exclusion_globs else ""
|
|
rg_executable = rg_executable or self._resolve_command("rg")
|
|
if not rg_executable:
|
|
return SearchResult(error="File search requires ripgrep (rg).")
|
|
if order == "modified":
|
|
capability_error = self._modified_rg_capability_error(rg_executable)
|
|
if capability_error:
|
|
return SearchResult(error=capability_error)
|
|
rg = self._quote_executable(rg_executable)
|
|
sort_arg = " --sortr=modified" if order == "modified" else ""
|
|
root_args = " ".join(self._escape_native_tool_arg(root) for root in command_roots)
|
|
cd_prefix = f"cd {self._escape_shell_arg(scoped_common)} && " if scoped_common else ""
|
|
# ``--`` terminates options so a dash-prefixed root is never parsed as a flag.
|
|
rg_cmd = (f"{rg} --files{sort_arg} -g {self._escape_shell_arg(glob_pattern)}"
|
|
f"{exclusion_args} -- {root_args}")
|
|
result = self._run_rg_bounded([rg_cmd], fetch_limit, timeout=60, native_ok=not scoped_common,
|
|
shell_prefix=f"set -o pipefail; {cd_prefix}")
|
|
stdout, limit_reason = _search_stdout_and_limit(result)
|
|
all_files = [f for f in stdout.splitlines() if f]
|
|
if scoped_common:
|
|
all_files = [
|
|
f if posixpath.isabs(f) else posixpath.normpath(posixpath.join(scoped_common, f))
|
|
for f in all_files]
|
|
bounded_sigpipe = result.exit_code == 141 and len(all_files) >= fetch_limit
|
|
if result.exit_code not in {0, 1, 124} and not bounded_sigpipe:
|
|
if order == "modified":
|
|
return SearchResult(error=(
|
|
"Exact modification-time order failed; ripgrep 14+ is "
|
|
"required. Upgrade ripgrep or use order='discovery'."))
|
|
return SearchResult(error="File search failed while running ripgrep.")
|
|
return SearchResult(
|
|
files=all_files[offset:offset + limit], total_count=len(all_files),
|
|
truncated=len(all_files) > offset + limit or bool(limit_reason), limit_reason=limit_reason)
|
|
|
|
def _search_content(self, pattern: str, path: str, file_glob: Optional[str],
|
|
limit: int, offset: int, output_mode: str, context: int) -> SearchResult:
|
|
"""Content search: rg, else grep; attaches zero-match steering hints."""
|
|
used_rg = self._has_command('rg')
|
|
if used_rg:
|
|
result = self._search_with_rg(pattern, path, file_glob, limit, offset, output_mode, context,
|
|
rg_executable=self._resolve_command("rg") or "rg")
|
|
elif self._has_command('grep'):
|
|
result = self._search_with_grep(pattern, path, file_glob, limit, offset, output_mode, context)
|
|
else:
|
|
return SearchResult(
|
|
error="Content search requires ripgrep (rg) or grep. "
|
|
"Install ripgrep: https://github.com/BurntSushi/ripgrep#installation")
|
|
if (not result.error and result.total_count == 0
|
|
and not result.matches and not result.files and not result.counts):
|
|
try:
|
|
hint = self._zero_match_probe(pattern, path, file_glob)
|
|
except Exception:
|
|
hint = None
|
|
if hint:
|
|
result.warning = hint if not result.warning else f"{result.warning} {hint}"
|
|
# rg auto-enables --multiline for \n patterns, so the line-oriented
|
|
# explanation only applies to the grep fallback.
|
|
if used_rg:
|
|
return result
|
|
return _maybe_warn_line_oriented_newline_pattern(result, pattern)
|
|
|
|
def _run_search_pipeline(self, cmd_parts: List[str], output_mode: str, limit: int,
|
|
offset: int, context: int, warning: Optional[str] = None,
|
|
line_cap: bool = False) -> SearchResult:
|
|
"""Run ``cmd_parts | head -n <fetch_limit>`` under pipefail and parse. Extra
|
|
rows report the true total (context mode also emits "--" separators, so
|
|
grab 200 more). pipefail keeps the engine's exit 2 alive across ``| head``
|
|
(a truncating head makes rg exit 0 / grep 141, which the ==2 guard ignores).
|
|
``line_cap`` appends ``| cut -c1-2000`` for engines without --max-columns
|
|
(grep): bounds giant single-line matches at the pipe layer; skipped for
|
|
files_only/count where lines are paths/counts."""
|
|
fetch_limit = limit + offset + (200 if context > 0 else 0)
|
|
if line_cap: # grep/find pipelines: shell only, with the column cap
|
|
parts = cmd_parts + ["|", "head", "-n", str(fetch_limit)]
|
|
if output_mode not in ("files_only", "count"):
|
|
parts += ["|", "cut", "-c1-2000"]
|
|
result = self._exec("set -o pipefail; " + " ".join(parts), timeout=60)
|
|
else:
|
|
result = self._run_rg_bounded(cmd_parts, fetch_limit, timeout=60, merge_stderr=True,
|
|
shell_prefix="set -o pipefail; ")
|
|
return _parse_search_output(result, output_mode, limit, offset, context, warning=warning)
|
|
|
|
def _search_with_rg(self, pattern: str, path: str, file_glob: Optional[str],
|
|
limit: int, offset: int, output_mode: str, context: int,
|
|
rg_executable: Optional[str] = None) -> SearchResult:
|
|
"""Search using ripgrep."""
|
|
rg_executable = rg_executable or self._resolve_command("rg")
|
|
if not rg_executable:
|
|
return SearchResult(error="Content search requires ripgrep (rg).")
|
|
cmd_parts = [self._quote_executable(rg_executable), "--line-number", "--no-heading", "--with-filename"]
|
|
# Giant-single-line containment (cline#13525): a match inside a multi-MB
|
|
# single-line dump makes rg emit the ENTIRE line (``head -n`` counts lines).
|
|
# --max-columns bounds each printed line at the rg layer; --max-columns-preview
|
|
# keeps a truncated prefix so the model still sees the hit. 2000 cols exceeds
|
|
# the 500-char content clamp, so nothing previously visible is lost.
|
|
if output_mode not in ("files_only", "count"):
|
|
cmd_parts.extend(["--max-columns", "2000", "--max-columns-preview"])
|
|
# A regex \n hard-errors in line-oriented mode; enable -U up front and say so.
|
|
multiline = _pattern_has_regex_newline(pattern)
|
|
if multiline:
|
|
cmd_parts.append("--multiline")
|
|
if context > 0:
|
|
cmd_parts.extend(["-C", str(context)])
|
|
cmd_parts.extend(self._rg_exclusion_globs(path))
|
|
if file_glob:
|
|
cmd_parts.extend(["--glob", self._escape_shell_arg(file_glob)])
|
|
if output_mode in _OUTPUT_MODE_FLAGS:
|
|
cmd_parts.append(_OUTPUT_MODE_FLAGS[output_mode])
|
|
cmd_parts.append(self._escape_shell_arg(pattern, translate_path=False))
|
|
# rg is a native Windows binary (winget/cargo/choco): needs C:/... not MSYS /c/...
|
|
cmd_parts.append(self._escape_native_tool_arg(path))
|
|
ml_note = (
|
|
"Pattern contains \\n — multiline mode (-U) was enabled automatically "
|
|
"so the regex can match across line boundaries."
|
|
) if multiline else None
|
|
return self._run_search_pipeline(cmd_parts, output_mode, limit, offset, context, warning=ml_note)
|
|
|
|
def _grep_cmd(self, head: List[str], pattern: str, output_mode: str, context: int,
|
|
file_glob: Optional[str] = None) -> List[str]:
|
|
"""``head`` + context/include/mode flags + quoted pattern (argument order is fixed)."""
|
|
parts = list(head)
|
|
if context > 0:
|
|
parts.extend(["-C", str(context)])
|
|
if file_glob:
|
|
parts.extend(["--include", self._escape_shell_arg(file_glob)])
|
|
if output_mode in _OUTPUT_MODE_FLAGS:
|
|
parts.append(_OUTPUT_MODE_FLAGS[output_mode])
|
|
parts.append(self._escape_shell_arg(pattern, translate_path=False))
|
|
return parts
|
|
|
|
def _search_with_grep(self, pattern: str, path: str, file_glob: Optional[str],
|
|
limit: int, offset: int, output_mode: str, context: int) -> SearchResult:
|
|
"""Fallback search using grep."""
|
|
# grep's --exclude-dir matches BASENAMES anywhere, so it can't express "only
|
|
# the home-level Downloads"; route pruning through find's path-scoped -prune.
|
|
protected_paths = self._protected_prune_paths(path)
|
|
# grep applies --exclude-dir='.*' to the command-line root too (GNU grep: to
|
|
# every component of it), so a search rooted under a hidden dir such as
|
|
# ~/.hermes returns nothing (#18473); find's -prune only sees descendants.
|
|
if protected_paths or self._root_under_hidden_dir(path):
|
|
return self._search_with_grep_pruned(
|
|
pattern, path, file_glob, limit, offset, output_mode, context, protected_paths)
|
|
# -H forces filenames; -E matches rg regex behavior; --exclude-dir='.*'
|
|
# mirrors rg's hidden-dir default (.git/, .hub/index-cache/, ...).
|
|
cmd_parts = self._grep_cmd(["grep", "-rnHE", "--exclude-dir='.*'"], pattern, output_mode, context, file_glob)
|
|
# --exclude-dir applies to the root too, so "." would be excluded by '.*';
|
|
# anchor relative paths at the shell's live $PWD.
|
|
is_absolute = path.startswith(("/", "\\\\")) or bool(re.match(r"^[A-Za-z]:[\\/]", path))
|
|
if is_absolute:
|
|
search_root = self._escape_shell_arg(path)
|
|
else:
|
|
relative_path = path[2:] if path.startswith("./") else path
|
|
search_root = '"$PWD"'
|
|
if relative_path not in {"", "."}:
|
|
search_root += f"/{self._escape_shell_arg(relative_path)}"
|
|
cmd_parts.append(search_root)
|
|
return self._run_search_pipeline(cmd_parts, output_mode, limit, offset, context, line_cap=True)
|
|
|
|
def _search_with_grep_pruned(self, pattern: str, path: str, file_glob: Optional[str],
|
|
limit: int, offset: int, output_mode: str, context: int,
|
|
protected_paths: List[str]) -> SearchResult:
|
|
"""grep fallback via ``find ... -prune -exec grep {} +``, used when the root needs
|
|
path-scoped pruning (macOS protected dirs) or is itself under a dot-directory
|
|
(#18473: grep's ``--exclude-dir='.*'`` would drop the root). Trade-off: find folds
|
|
grep's exit code, so a hard grep error surfaces as an empty result."""
|
|
grep_parts = self._grep_cmd(["grep", "-nHE"], pattern, output_mode, context)
|
|
q_root = self._escape_shell_arg(path or ".")
|
|
# ``-H``: follow a symlink handed in as the OPERAND (and only the operand). Without
|
|
# it ``find <link> -type f`` tests the link itself and hands grep nothing, so a
|
|
# symlinked root answered a confident ``total_count: 0`` on every platform (#116270).
|
|
find_parts = ["find", "-H", q_root]
|
|
if protected_paths:
|
|
find_parts.extend([self._prune_expr(protected_paths), "-o"])
|
|
find_parts.extend([self._hidden_prune_expr([q_root]), "-o", "-type f"])
|
|
if file_glob:
|
|
find_parts.extend(["-name", self._escape_shell_arg(file_glob)])
|
|
find_parts.extend(["-exec", *grep_parts, "{}", "+", "2>/dev/null"])
|
|
return self._run_search_pipeline(find_parts, output_mode, limit, offset, context, line_cap=True)
|