diff --git a/navi/tools/_filesystem/__init__.py b/navi/tools/_filesystem/__init__.py new file mode 100644 index 0000000..c07eb55 --- /dev/null +++ b/navi/tools/_filesystem/__init__.py @@ -0,0 +1,9 @@ +"""Filesystem tool implementation — `helpers` (pure) + `tool` (the `Tool`). + +`navi.tools.filesystem` remains as a compatibility shim; new code should import +from here. +""" + +from .tool import FilesystemTool + +__all__ = ["FilesystemTool"] diff --git a/navi/tools/_filesystem/helpers.py b/navi/tools/_filesystem/helpers.py new file mode 100644 index 0000000..5da1293 --- /dev/null +++ b/navi/tools/_filesystem/helpers.py @@ -0,0 +1,322 @@ +"""Pure filesystem helpers: path checks, formatting, diffs, AI text plumbing. + +Nothing here touches a `Tool` instance — `tool` composes these. +""" + +import difflib +import re +from datetime import datetime +from pathlib import Path + +from navi.config import settings + +from .._internal.areas import allowed_areas, resolve_in_areas +from .._internal.base import ( + current_session_id, + current_user_id, + current_user_role, + current_working_directory, +) + +_READ_WARN_BYTES = 100_000 # 100 KB — add size warning in output +_READ_HARD_BYTES = 1_000_000 # 1 MB — refuse full read without offset/limit +_LIST_MAX_ENTRIES = 500 +_FIND_MAX_RESULTS = 200 + +# AI actions: ~20k tokens of file content per chunk (4 chars ≈ 1 token) +_AI_CHUNK_CHARS = 80_000 +_AI_OVERLAP_LINES = 30 +# smart_edit: refuse files larger than ~50k tokens (full file must fit in one call) +_AI_EDIT_MAX_CHARS = 200_000 + +# ── System prompts ──────────────────────────────────────────────────────────── + +_QUERY_SINGLE_SYSTEM = ( + "You are a precise file analysis assistant. " + "Answer the question based strictly on the file content shown. " + "Be specific and concise. Include line numbers when relevant." +) + +_QUERY_CHUNK_SYSTEM = ( + "You are analyzing one section of a larger file. " + "Answer the question using only the lines shown. " + "If the answer is not present in this section, respond with exactly: NOT_FOUND\n" + "Otherwise be specific and include line numbers." +) + +_QUERY_SYNTHESIS_SYSTEM = ( + "Combine these partial findings from different sections of a file into one clear answer. " + "Remove duplicates. Be direct and concise." +) + +_EDIT_SYSTEM = ( + "You are a precise file editor. " + "Given file content with line numbers and an instruction, output ONLY a JSON array of edit operations. " + "No explanation, no markdown — just the JSON.\n\n" + "Format:\n" + "[\n" + ' {"op": "replace", "start": LINE, "end": LINE, "content": "new text\\nmore lines"},\n' + ' {"op": "delete", "start": LINE, "end": LINE},\n' + ' {"op": "insert", "after": LINE, "content": "text to insert"}\n' + "]\n\n" + "Rules:\n" + "- Line numbers are 1-based and inclusive\n" + "- Use \\n in content strings for embedded newlines\n" + "- Make MINIMAL changes to accomplish the instruction\n" + "- 'insert' after=0 inserts before the first line\n" + "- If no changes are needed, return []" +) + + +# ── Path helpers ────────────────────────────────────────────────────────────── + + +def _resolve_relative_path( + p: Path, user_id: str | None, role: str | None, cwd: str | None = None +) -> Path: + """Resolve a relative path against the session cwd when available. + + In single-user/legacy/admin mode, uses the client-provided working directory + so that `filesystem read src/main.py` works from the project directory the + user launched Navi Code in, regardless of the server's cwd. + """ + if user_id and role != "admin": + # Sandbox mode keeps its own resolution rules. + return p + + cwd = cwd or current_working_directory.get(None) + if cwd: + return (Path(cwd) / p).resolve() + return p.resolve() + + +def _check_path( + path_str: str, + user_id: str | None = None, + role: str | None = None, + cwd: str | None = None, +) -> Path | None: + """Return resolved Path if access is allowed, else None. + + When a user_id is active (multi-user mode), all paths are resolved + inside user_data// or inside the current session directory. + This prevents users from accessing each other's files or random OS + directories, while still letting a user read what an upload put in the + session and what share_file hands back. + """ + if not path_str or path_str.strip() == "": + return None + try: + p = Path(path_str) + except Exception: + return None + + user_id = user_id or current_user_id.get(None) + role = role or current_user_role.get() + + # Admins bypass sandbox and use FS_ALLOWED_PATHS directly + if user_id and role != "admin": + # Sandboxed mode: the user's own sandbox, plus the current session + # directory. Relative paths keep resolving inside the sandbox. + areas = allowed_areas(user_id, current_session_id.get(None)) + return resolve_in_areas(p, areas) + + # Fallback to FS_ALLOWED_PATHS for single-user / legacy mode / admin + try: + if not p.is_absolute(): + p = _resolve_relative_path(p, user_id, role, cwd=cwd) + else: + p = p.expanduser().resolve() + except Exception: + return None + + if settings.fs_allowed_paths.strip() == "*": + return p + + allowed = [Path(r).expanduser().resolve() for r in settings.fs_allowed_paths_list] + for root in allowed: + try: + p.relative_to(root) + return p + except ValueError: + continue + return None + + +def _fmt_size(n: int) -> str: + if n < 1024: + return f"{n} B" + if n < 1024**2: + return f"{n / 1024:.1f} KB" + if n < 1024**3: + return f"{n / 1024**2:.1f} MB" + return f"{n / 1024**3:.1f} GB" + + +def _fmt_time(ts: float) -> str: + return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") + + +# ── AI helpers (module-level, no self) ──────────────────────────────────────── + + +def _number_lines(lines: list[str], start: int = 1) -> str: + """Return file lines with 1-based line numbers, right-aligned.""" + width = len(str(start + len(lines) - 1)) + return "\n".join(f"{start + i:>{width}}: {line}" for i, line in enumerate(lines)) + + +def _make_chunks(lines: list[str], target_chars: int, overlap: int) -> list[tuple[int, int]]: + """ + Split lines into (start_idx, end_idx) chunks of at most target_chars characters. + Consecutive chunks overlap by `overlap` lines to preserve boundary context. + """ + if not lines: + return [(0, 0)] + total = len(lines) + total_chars = sum(len(line) + 1 for line in lines) + if total_chars <= target_chars: + return [(0, total)] + + chunks: list[tuple[int, int]] = [] + start = 0 + while start < total: + chars = 0 + end = start + while end < total and chars < target_chars: + chars += len(lines[end]) + 1 + end += 1 + chunks.append((start, end)) + if end >= total: + break + start = max(start + 1, end - overlap) # always make progress + + return chunks or [(0, total)] + + +def _validate_ops(ops: list, total_lines: int) -> list[str]: + errors: list[str] = [] + for i, op in enumerate(ops): + if not isinstance(op, dict): + errors.append(f"op[{i}] is not a dict") + continue + kind = op.get("op") + if kind not in ("replace", "delete", "insert"): + errors.append(f"op[{i}] unknown type {kind!r}") + continue + if kind in ("replace", "delete"): + s, e = op.get("start"), op.get("end") + if not isinstance(s, int) or not isinstance(e, int): + errors.append(f"op[{i}] start/end must be integers") + elif s < 1 or e > total_lines or s > e: + errors.append(f"op[{i}] range {s}-{e} out of bounds (file has {total_lines} lines)") + elif kind == "insert": + after = op.get("after") + if not isinstance(after, int): + errors.append(f"op[{i}] 'after' must be integer") + elif after < 0 or after > total_lines: + errors.append(f"op[{i}] 'after'={after} out of bounds (0–{total_lines})") + return errors + + +def _apply_ops(lines: list[str], ops: list[dict]) -> list[str]: + """Apply edit operations bottom-up (highest line first) to preserve line numbers.""" + sorted_ops = sorted( + ops, + key=lambda o: o.get("start", o.get("after", 0)), + reverse=True, + ) + result = list(lines) + for op in sorted_ops: + kind = op["op"] + if kind == "replace": + s = op["start"] - 1 # 0-based + e = op["end"] # exclusive (1-based end = exclusive 0-based end) + new = op.get("content", "").split("\n") + result[s:e] = new + elif kind == "delete": + s = op["start"] - 1 + e = op["end"] + del result[s:e] + elif kind == "insert": + after = op["after"] # insert after this 1-based line (0 = before line 1) + new = op.get("content", "").split("\n") + result[after:after] = new + return result + + +_DIFF_HUNK_RE = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@") + + +def _number_diff(diff_lines: list[str]) -> list[str]: + """Prefix each unified-diff content line with its real file line number. + + Numbers come from the enclosing ``@@ -old_start,n +new_start,n @@`` hunk: + removed and context lines use the old (from) line number, added lines use the + new (to) line number. The format ``{marker} {num:>width}│ {content}`` keeps the + ``+``/``-``/`` `` marker first so existing ``startswith``-based highlighting + stays valid; a fixed space separates the marker from the right-aligned number + column. Hunk headers, file headers (``---``/``+++``) and the + ``\\ No newline`` marker are left unchanged. + """ + # First pass: assign numbers, track the max for column width. + numbered: list[tuple[str, int | None, str]] = [] # (marker, num, content) + max_num = 0 + old_line = 0 + new_line = 0 + for raw_line in diff_lines: + # difflib.unified_diff keeps a trailing "\n" on content lines when the + # input was fed with explicit newlines (see _unified_diff, which adds + # "\n" so difflib does not emit a "\ No newline at end of file" marker). + # Strip it here so the later "\n".join does not produce a blank line + # between every diff line. No-op for the _diff action, whose lines come + # from splitlines() without trailing newlines. + line = raw_line.rstrip("\n") + if line.startswith("@@"): + m = _DIFF_HUNK_RE.match(line) + if m: + old_line = int(m.group(1)) + new_line = int(m.group(3)) + numbered.append(("", None, line)) + continue + if line.startswith("---") or line.startswith("+++") or line.startswith("\\"): + numbered.append(("", None, line)) + continue + marker = line[0] if line else " " + content = line[1:] + if marker == "+": + num = new_line + new_line += 1 + elif marker == "-": + num = old_line + old_line += 1 + else: # " " context — present in both old and new + num = old_line + old_line += 1 + new_line += 1 + if num > max_num: + max_num = num + numbered.append((marker, num, content)) + + width = len(str(max_num)) if max_num else 1 + out: list[str] = [] + for marker, num, content in numbered: + if num is None: + out.append(content) + else: + out.append(f"{marker} {num:>{width}}│ {content}") + return out + + +def _unified_diff(original: list[str], modified: list[str], path: Path) -> str: + diff = list( + difflib.unified_diff( + [line + "\n" for line in original], + [line + "\n" for line in modified], + fromfile=f"a/{path.name}", + tofile=f"b/{path.name}", + lineterm="", + ) + ) + return "\n".join(_number_diff(diff)) + diff --git a/navi/tools/_filesystem/tool.py b/navi/tools/_filesystem/tool.py new file mode 100644 index 0000000..a7a9d45 --- /dev/null +++ b/navi/tools/_filesystem/tool.py @@ -0,0 +1,847 @@ +"""`FilesystemTool` — the dispatcher and its read/write/AI action handlers.""" + +import asyncio +import difflib +import re +import shutil +import stat +from pathlib import Path + +from navi.config import settings + +from .._internal.base import Tool, ToolContext, ToolResult, current_user_id, current_user_role +from .helpers import ( + _AI_CHUNK_CHARS, + _AI_EDIT_MAX_CHARS, + _AI_OVERLAP_LINES, + _EDIT_SYSTEM, + _FIND_MAX_RESULTS, + _LIST_MAX_ENTRIES, + _QUERY_CHUNK_SYSTEM, + _QUERY_SINGLE_SYSTEM, + _QUERY_SYNTHESIS_SYSTEM, + _READ_HARD_BYTES, + _READ_WARN_BYTES, + _apply_ops, + _check_path, + _fmt_size, + _fmt_time, + _make_chunks, + _number_diff, + _number_lines, + _unified_diff, + _validate_ops, +) + + +class FilesystemTool(Tool): + name = "filesystem" + description = ( + "Operate the local filesystem: read, write, edit, search, and manage files. " + "Editing policy — cheapest deterministic method first, smart_edit LAST: use " + "'edit' (exact text, the default) or 'edit_lines' (known line numbers); both " + "are deterministic. Reach for 'smart_edit' ONLY when the change cannot be " + "expressed as text or line numbers — it costs an LLM call and reads the whole " + "file. Call 'info' before reading an unknown file. Full action list, argument " + 'rules and access limits: tool_manual("filesystem").' + ) + parameters = { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": [ + "read", + "write", + "append", + "edit", + "edit_lines", + "list", + "find", + "find_up", + "info", + "move", + "copy", + "delete", + "exists", + "mkdir", + "query", + "smart_edit", + "grep", + "diff", + ], + "description": "Operation to perform.", + }, + "path": { + "type": "string", + "description": "Absolute or relative file/directory path (~ is expanded).", + }, + "content": { + "type": "string", + "description": "Text to write or append (required for write/append).", + }, + "old": { + "type": "string", + "description": "Exact text to replace (required for edit). Must occur exactly once.", + }, + "new": { + "type": "string", + "description": "Replacement text for edit.", + }, + "operations": { + "type": "array", + "description": ( + "Line-based edits (required for edit_lines). 1-based, inclusive. " + "'after' applies to insert." + ), + "items": { + "type": "object", + "properties": { + "op": {"type": "string", "enum": ["replace", "delete", "insert"]}, + "start": {"type": "integer"}, + "end": {"type": "integer"}, + "after": {"type": "integer"}, + "content": {"type": "string"}, + }, + "required": ["op"], + }, + }, + "numbered": { + "type": "boolean", + "description": "Number the lines in read output (default true). Set false to copy text into 'edit'.", + }, + "destination": { + "type": "string", + "description": "Second path: the target for move/copy, or the file to compare against for diff.", + }, + "pattern": { + "type": "string", + "description": "Glob pattern for find (e.g. '*.log'), or exact filename for find_up.", + }, + "offset": { + "type": "integer", + "description": "First line to read, 1-based (for read action).", + }, + "limit": { + "type": "integer", + "description": "Max lines to return (for read action).", + }, + "recursive": { + "type": "boolean", + "description": "Full recursive directory tree (for list, default false).", + }, + "regex": { + "type": "boolean", + "description": "Use regex matching for grep (default false = literal substring search).", + }, + "question": { + "type": "string", + "description": ( + "Natural language question about the file (for query) — the answer comes " + "back instead of the file." + ), + }, + "instruction": { + "type": "string", + "description": "Natural language edit instruction (for smart_edit).", + }, + }, + "required": ["action", "path"], + } + + def __init__(self, ai_helper=None) -> None: + # ai_helper is optional — standard actions work without it + self._ai = ai_helper + + async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult: + action = params.get("action", "") + raw_path = params.get("path", "") + user_id = ctx.user_id if ctx else current_user_id.get(None) + role = ctx.user_role if ctx else current_user_role.get() + path = _check_path(raw_path, user_id, role, cwd=ctx.cwd if ctx else None) + + if path is None: + if not raw_path or raw_path.strip() == "": + return ToolResult( + success=False, + output="'path' is required. For 'write' action use 'path' (not 'destination').", + error="missing_path", + ) + return ToolResult( + success=False, + output=( + f"Access denied: '{raw_path}' is outside allowed paths " + f"({settings.fs_allowed_paths}). " + "Set FS_ALLOWED_PATHS=* in .env to allow all paths." + ), + error="access_denied", + ) + + try: + match action: + case "read": + return await asyncio.to_thread(self._read, path, params) + case "write": + return await asyncio.to_thread(self._write, path, params) + case "append": + return await asyncio.to_thread(self._append, path, params) + case "edit": + return await asyncio.to_thread(self._edit, path, params) + case "edit_lines": + return await asyncio.to_thread(self._edit_lines, path, params) + case "list": + return await asyncio.to_thread(self._list, path, params) + case "find": + return await asyncio.to_thread(self._find, path, params) + case "find_up": + return await asyncio.to_thread(self._find_up, path, params) + case "info": + return await asyncio.to_thread(self._info, path) + case "move": + return await asyncio.to_thread(self._move, path, params) + case "copy": + return await asyncio.to_thread(self._copy, path, params) + case "delete": + return await asyncio.to_thread(self._delete, path) + case "exists": + return ToolResult(success=True, output="true" if path.exists() else "false") + case "mkdir": + return await asyncio.to_thread(self._mkdir, path) + case "query": + return await self._query(path, params) + case "smart_edit": + return await self._smart_edit(path, params) + case "grep": + return await asyncio.to_thread(self._grep, path, params) + case "diff": + return await asyncio.to_thread(self._diff, path, params) + case _: + return ToolResult( + success=False, output=f"Unknown action: {action}", error="invalid_action" + ) + + except PermissionError as e: + return ToolResult(success=False, output=f"Permission denied: {e}", error=str(e)) + except Exception as e: + return ToolResult(success=False, output=f"Filesystem error: {e}", error=str(e)) + + # ── Standard action handlers ────────────────────────────────────────────── + + def _read(self, path: Path, params: dict) -> ToolResult: + if not path.exists(): + return ToolResult(success=False, output=f"File not found: {path}", error="not_found") + if path.is_dir(): + return ToolResult( + success=False, + output=f"Path is a directory, use 'list': {path}", + error="is_directory", + ) + + file_size = path.stat().st_size + offset = params.get("offset") + limit = params.get("limit") + + if file_size > _READ_HARD_BYTES and offset is None and limit is None: + return ToolResult( + success=False, + output=( + f"File too large to read in full: {_fmt_size(file_size)} — {path}\n" + "Use offset/limit to read specific line ranges " + "(e.g. offset=1, limit=100), or use 'query' to ask a question about it." + ), + error="file_too_large", + ) + + text = path.read_text(encoding="utf-8", errors="replace") + lines = text.splitlines(keepends=True) + total_lines = len(lines) + + if offset is not None or limit is not None: + start = max(0, (offset or 1) - 1) + end = (start + limit) if limit is not None else total_lines + selected = lines[start:end] + actual_end = min(end, total_lines) + header = ( + f"[{path} | lines {start + 1}–{actual_end} of {total_lines}" + f" | {_fmt_size(file_size)}]\n" + ) + numbered = params.get("numbered", True) + body = ( + _number_lines([l.rstrip("\r\n") for l in selected], start=start + 1) + if numbered + else "".join(selected) + ) + return ToolResult(success=True, output=header + body) + + warn = ( + f"⚠ Large file ({_fmt_size(file_size)}) — consider offset/limit next time.\n" + if file_size > _READ_WARN_BYTES + else "" + ) + numbered = params.get("numbered", True) + header = f"[{path} | {total_lines} lines | {_fmt_size(file_size)}]\n" + if numbered: + return ToolResult( + success=True, + output=header + warn + _number_lines(text.splitlines(keepends=False), 1), + ) + return ToolResult(success=True, output=header + warn + text) + + def _write(self, path: Path, params: dict) -> ToolResult: + content = params.get("content", "") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + lines = len(content.splitlines()) + return ToolResult( + success=True, + output=f"Written {_fmt_size(len(content.encode()))} ({lines} lines) → {path}", + ) + + def _append(self, path: Path, params: dict) -> ToolResult: + content = params.get("content", "") + if not content: + return ToolResult( + success=False, output="'content' is required for append", error="missing_content" + ) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as f: + f.write(content) + return ToolResult( + success=True, + output=f"Appended {_fmt_size(len(content.encode()))} to {path} (file now {_fmt_size(path.stat().st_size)})", + ) + + def _edit(self, path: Path, params: dict) -> ToolResult: + old = params.get("old") + new = params.get("new") + if old is None: + return ToolResult( + success=False, output="'old' is required for edit", error="missing_old" + ) + if new is None: + return ToolResult( + success=False, output="'new' is required for edit", error="missing_new" + ) + if old == "": + return ToolResult(success=False, output="'old' must not be empty", error="empty_old") + if not path.exists(): + return ToolResult(success=False, output=f"File not found: {path}", error="not_found") + if path.is_dir(): + return ToolResult( + success=False, output="edit works on files, not directories.", error="is_directory" + ) + + text = path.read_text(encoding="utf-8", errors="replace") + count = text.count(old) + if count == 0: + return ToolResult( + success=False, + output=( + "Exact text not found. Read the relevant file section and call edit again " + "with text copied exactly from the file." + ), + error="old_not_found", + ) + if count > 1: + return ToolResult( + success=False, + output=( + f"Exact text occurs {count} times. Prefer 'edit' with a larger " + "unique 'old' block (include surrounding lines so it matches once) " + "or 'edit_lines' with line numbers. Fall back to 'smart_edit' only if " + "the change is genuinely semantic and cannot be expressed as exact text." + ), + error="old_not_unique", + ) + + new_text = text.replace(old, new, 1) + path.write_text(new_text, encoding="utf-8") + delta = len(new_text.encode()) - len(text.encode()) + summary = ( + f"Edited {path}: replaced {len(old.encode())} B with " + f"{len(new.encode())} B (delta {delta:+d} B)." + ) + # Keep the model-facing output short; carry the unified diff in metadata + # for the TUI to render with red/green highlighting (the model does not + # see metadata, so its context stays unchanged). + diff = _unified_diff(text.splitlines(), new_text.splitlines(), path) + return ToolResult( + success=True, + output=summary, + metadata={"diff": diff} if diff else {}, + ) + + def _edit_lines(self, path: Path, params: dict) -> ToolResult: + ops = params.get("operations") + if not ops: + return ToolResult( + success=False, + output="'operations' is required for edit_lines", + error="missing_operations", + ) + if not path.exists(): + return ToolResult(success=False, output=f"File not found: {path}", error="not_found") + if path.is_dir(): + return ToolResult( + success=False, + output="edit_lines works on files, not directories.", + error="is_directory", + ) + + text = path.read_text(encoding="utf-8", errors="replace") + lines = text.splitlines() + errors = _validate_ops(ops, len(lines)) + if errors: + return ToolResult( + success=False, + output="Invalid operations:\n" + "\n".join(f" • {e}" for e in errors), + error="invalid_ops", + ) + new_lines = _apply_ops(lines, ops) + diff = _unified_diff(lines, new_lines, path) + + new_text = "\n".join(new_lines) + ("\n" if text.endswith("\n") else "") + tmp = path.with_suffix(path.suffix + ".tmp") + try: + tmp.write_text(new_text, encoding="utf-8") + tmp.replace(path) + finally: + if tmp.exists(): + tmp.unlink(missing_ok=True) + + summary = ( + f"Applied {len(ops)} operation(s) to {path.name} " + f"({len(lines)} → {len(new_lines)} lines)." + ) + return ToolResult(success=True, output=f"{summary}\n\n{diff}" if diff else summary) + + def _list(self, path: Path, params: dict) -> ToolResult: + if not path.exists(): + return ToolResult(success=False, output=f"Path not found: {path}", error="not_found") + if path.is_file(): + return self._info(path) + + recursive = params.get("recursive", False) + raw_entries = list(path.rglob("*") if recursive else path.iterdir()) + raw_entries.sort(key=lambda e: (e.is_file(), str(e).lower())) + + truncated = len(raw_entries) > _LIST_MAX_ENTRIES + entries = raw_entries[:_LIST_MAX_ENTRIES] + lines = [] + for e in entries: + try: + s = e.stat() + rel = e.relative_to(path) + if e.is_dir(): + if not recursive: + try: + n = sum(1 for _ in e.iterdir()) + lines.append(f"d {rel}/ ({n} items)") + except PermissionError: + lines.append(f"d {rel}/") + else: + lines.append(f"d {rel}/") + else: + lines.append( + f" {str(rel):<48} {_fmt_size(s.st_size):>10} {_fmt_time(s.st_mtime)}" + ) + except Exception: + lines.append(f"? {e.name}") + + note = " ⚠ truncated" if truncated else "" + header = f"[{path} | {len(entries)} entries{note}]\n" + return ToolResult(success=True, output=header + ("\n".join(lines) or "(empty directory)")) + + def _find(self, path: Path, params: dict) -> ToolResult: + pattern = params.get("pattern") + if not pattern: + return ToolResult( + success=False, output="'pattern' is required for find", error="missing_pattern" + ) + if not path.exists(): + return ToolResult(success=False, output=f"Path not found: {path}", error="not_found") + + matches: list[Path] = [] + try: + for p in path.rglob(pattern): + matches.append(p) + if len(matches) >= _FIND_MAX_RESULTS: + break + except Exception as e: + return ToolResult(success=False, output=f"Find error: {e}", error=str(e)) + + if not matches: + return ToolResult(success=True, output=f"No matches for '{pattern}' in {path}") + + matches.sort() + lines = [] + for m in matches: + try: + size = _fmt_size(m.stat().st_size) if m.is_file() else "" + lines.append(f"{m} ({size})") + except Exception: + lines.append(str(m)) + + extra = ( + f" ⚠ showing first {_FIND_MAX_RESULTS}" if len(matches) == _FIND_MAX_RESULTS else "" + ) + header = f"[{len(matches)} matches for '{pattern}' in {path}{extra}]\n" + return ToolResult(success=True, output=header + "\n".join(lines)) + + def _find_up(self, path: Path, params: dict) -> ToolResult: + filename = params.get("pattern") + if not filename: + return ToolResult(success=False, output="", error="pattern is required for find_up") + current = path if path.is_dir() else path.parent + checked = [] + while True: + target = current / filename + checked.append(str(target)) + if target.exists(): + return ToolResult(success=True, output=str(target)) + parent = current.parent + if parent == current: + return ToolResult( + success=True, output=f"not found (searched: {', '.join(checked)})" + ) + current = parent + + def _info(self, path: Path) -> ToolResult: + if not path.exists(): + return ToolResult(success=False, output=f"Not found: {path}", error="not_found") + + s = path.stat() + kind = "symlink" if path.is_symlink() else ("directory" if path.is_dir() else "file") + lines = [ + f"path: {path}", + f"type: {kind}", + f"size: {_fmt_size(s.st_size)}", + f"modified: {_fmt_time(s.st_mtime)}", + f"created: {_fmt_time(s.st_ctime)}", + f"mode: {stat.filemode(s.st_mode)}", + ] + if path.is_file(): + try: + text = path.read_text(encoding="utf-8", errors="replace") + lines.append(f"lines: {len(text.splitlines())}") + except Exception: + lines.append("lines: (binary or unreadable)") + elif path.is_dir(): + try: + children = list(path.iterdir()) + lines.append( + f"contents: {sum(c.is_file() for c in children)} files, {sum(c.is_dir() for c in children)} dirs (top level)" + ) + except Exception: + pass + return ToolResult(success=True, output="\n".join(lines)) + + def _move(self, path: Path, params: dict) -> ToolResult: + dest_raw = params.get("destination") + if not dest_raw: + return ToolResult( + success=False, + output="'destination' is required for move", + error="missing_destination", + ) + dest = _check_path(dest_raw) + if dest is None: + return ToolResult( + success=False, + output=f"Access denied: destination '{dest_raw}' outside allowed paths.", + error="access_denied", + ) + if not path.exists(): + return ToolResult(success=False, output=f"Not found: {path}", error="not_found") + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(path), str(dest)) + return ToolResult(success=True, output=f"Moved: {path} → {dest}") + + def _mkdir(self, path: Path) -> ToolResult: + path.mkdir(parents=False, exist_ok=True) + return ToolResult(success=True, output=f"Directory created: {path}") + + def _delete(self, path: Path) -> ToolResult: + if not path.exists(): + return ToolResult(success=False, output=f"Not found: {path}", error="not_found") + if path.is_dir(): + shutil.rmtree(path) + else: + path.unlink() + return ToolResult(success=True, output=f"Deleted: {path}") + + def _copy(self, path: Path, params: dict) -> ToolResult: + dest_raw = params.get("destination") + if not dest_raw: + return ToolResult( + success=False, + output="'destination' is required for copy", + error="missing_destination", + ) + dest = _check_path(dest_raw) + if dest is None: + return ToolResult( + success=False, + output=f"Access denied: destination '{dest_raw}' outside allowed paths.", + error="access_denied", + ) + if not path.exists(): + return ToolResult(success=False, output=f"Not found: {path}", error="not_found") + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(str(path), str(dest)) + return ToolResult(success=True, output=f"Copied: {path} → {dest}") + + def _grep(self, path: Path, params: dict) -> ToolResult: + pattern = params.get("pattern", "").strip() + if not pattern: + return ToolResult( + success=False, output="'pattern' is required for grep", error="missing_pattern" + ) + use_regex = params.get("regex", False) + glob_filter = params.get("glob", "") + + try: + if use_regex: + compiled = re.compile(pattern, re.IGNORECASE) + else: + compiled = re.compile(re.escape(pattern), re.IGNORECASE) + except re.error as e: + return ToolResult( + success=False, output=f"Invalid pattern: {e}", error="invalid_pattern" + ) + + max_results = 200 + matches: list[str] = [] + files_searched = 0 + + if path.is_file(): + files_to_search = [path] + elif path.is_dir(): + if glob_filter: + files_to_search = list(path.rglob(glob_filter)) + else: + files_to_search = list(path.rglob("*")) + files_to_search = [f for f in files_to_search if f.is_file()] + else: + return ToolResult(success=False, output=f"Not found: {path}", error="not_found") + + for fpath in files_to_search: + if len(matches) >= max_results: + break + files_searched += 1 + try: + text = fpath.read_text(encoding="utf-8", errors="replace") + for i, line in enumerate(text.splitlines(), 1): + if compiled.search(line): + rel = fpath.relative_to(path) if fpath != path else fpath.name + matches.append(f"{rel}:{i}: {line}") + if len(matches) >= max_results: + break + except Exception: + continue + + if not matches: + return ToolResult(success=True, output=f"No matches for '{pattern}' in {path}") + + note = f" ⚠ showing first {max_results}" if len(matches) == max_results else "" + header = f"[{len(matches)} matches for '{pattern}' in {path} ({files_searched} files searched){note}]\n" + return ToolResult(success=True, output=header + "\n".join(matches)) + + def _diff(self, path: Path, params: dict) -> ToolResult: + dest_raw = params.get("destination") + if not dest_raw: + return ToolResult( + success=False, + output="'destination' is required for diff (path to second file)", + error="missing_destination", + ) + dest = _check_path(dest_raw) + if dest is None: + return ToolResult( + success=False, + output=f"Access denied: destination '{dest_raw}' outside allowed paths.", + error="access_denied", + ) + if not path.exists(): + return ToolResult(success=False, output=f"Not found: {path}", error="not_found") + if not dest.exists(): + return ToolResult(success=False, output=f"Not found: {dest}", error="not_found") + if path.is_dir() or dest.is_dir(): + return ToolResult( + success=False, output="diff works on files, not directories.", error="is_directory" + ) + + try: + a_lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + b_lines = dest.read_text(encoding="utf-8", errors="replace").splitlines() + except Exception as e: + return ToolResult(success=False, output=f"Read error: {e}", error=str(e)) + + diff = list( + difflib.unified_diff( + a_lines, + b_lines, + fromfile=f"a/{path.name}", + tofile=f"b/{dest.name}", + lineterm="", + ) + ) + if not diff: + return ToolResult(success=True, output="Files are identical.") + return ToolResult(success=True, output="\n".join(_number_diff(diff))) + + # ── AI action handlers ──────────────────────────────────────────────────── + + def _require_ai(self) -> ToolResult | None: + if not self._ai: + return ToolResult( + success=False, + output="AI helper is not available for this action.", + error="no_ai_helper", + ) + return None + + async def _query(self, path: Path, params: dict) -> ToolResult: + if (err := self._require_ai()) is not None: + return err + + question = params.get("question", "").strip() + if not question: + return ToolResult( + success=False, output="'question' is required for query.", error="missing_question" + ) + if not path.exists(): + return ToolResult(success=False, output=f"File not found: {path}", error="not_found") + if path.is_dir(): + return ToolResult( + success=False, output="query works on files, not directories.", error="is_directory" + ) + + text = await asyncio.to_thread(path.read_text, "utf-8", "replace") + lines = text.splitlines() + total = len(lines) + chunks = _make_chunks(lines, _AI_CHUNK_CHARS, _AI_OVERLAP_LINES) + + if len(chunks) == 1: + s, e = chunks[0] + numbered = _number_lines(lines[s:e], s + 1) + answer = await self._ai.ask( + _QUERY_SINGLE_SYSTEM, + f"File: {path}\n\nQuestion: {question}\n\nContent:\n{numbered}", + ) + return ToolResult(success=True, output=answer) + + # Multi-chunk: accumulate partial answers + partials: list[str] = [] + for s, e in chunks: + numbered = _number_lines(lines[s:e], s + 1) + partial = await self._ai.ask( + _QUERY_CHUNK_SYSTEM, + f"File: {path} (lines {s + 1}–{e} of {total})\nQuestion: {question}\n\nContent:\n{numbered}", + ) + if partial and "NOT_FOUND" not in partial.upper(): + partials.append(f"[lines {s + 1}–{e}] {partial}") + + if not partials: + return ToolResult( + success=True, + output=f"No information found in '{path.name}' relevant to: {question}", + ) + + if len(partials) == 1: + # Single finding — strip range prefix, return directly + answer = partials[0].split("] ", 1)[-1] + return ToolResult(success=True, output=answer) + + answer = await self._ai.ask( + _QUERY_SYNTHESIS_SYSTEM, + f"Question: {question}\n\nFindings from {len(partials)} sections:\n\n" + + "\n\n".join(partials), + ) + return ToolResult(success=True, output=answer) + + async def _smart_edit(self, path: Path, params: dict) -> ToolResult: + if (err := self._require_ai()) is not None: + return err + + instruction = params.get("instruction", "").strip() + if not instruction: + return ToolResult( + success=False, + output="'instruction' is required for smart_edit.", + error="missing_instruction", + ) + if not path.exists(): + return ToolResult(success=False, output=f"File not found: {path}", error="not_found") + if path.is_dir(): + return ToolResult( + success=False, + output="smart_edit works on files, not directories.", + error="is_directory", + ) + + text = await asyncio.to_thread(path.read_text, "utf-8", "replace") + if len(text) > _AI_EDIT_MAX_CHARS: + return ToolResult( + success=False, + output=( + f"File too large for smart_edit ({_fmt_size(len(text.encode()))}, " + f"limit {_fmt_size(_AI_EDIT_MAX_CHARS)}). " + "Use read with offset/limit to locate the relevant section, then write it back." + ), + error="file_too_large", + ) + + lines = text.splitlines() + numbered = _number_lines(lines, 1) + + raw_ops = await self._ai.ask_json( + _EDIT_SYSTEM, + f"File: {path}\nTotal lines: {len(lines)}\n\nInstruction: {instruction}\n\nContent:\n{numbered}", + ) + + if raw_ops is None: + return ToolResult( + success=False, + output="AI could not produce valid edit operations. Try rephrasing the instruction.", + error="invalid_ai_response", + ) + + if not isinstance(raw_ops, list): + raw_ops = [raw_ops] if isinstance(raw_ops, dict) else [] + + if not raw_ops: + return ToolResult(success=True, output="No changes needed — file unchanged.") + + errors = _validate_ops(raw_ops, len(lines)) + if errors: + return ToolResult( + success=False, + output="AI returned invalid operations:\n" + "\n".join(f" • {e}" for e in errors), + error="invalid_ops", + ) + + new_lines = _apply_ops(lines, raw_ops) + diff = _unified_diff(lines, new_lines, path) + + # Preserve trailing newline — write atomically to avoid partial writes on failure + import os + + new_text = "\n".join(new_lines) + ("\n" if text.endswith("\n") else "") + tmp = path.with_suffix(path.suffix + ".tmp") + + def _atomic_write() -> None: + try: + tmp.write_text(new_text, encoding="utf-8") + os.replace(tmp, path) + finally: + if tmp.exists(): + tmp.unlink(missing_ok=True) + + await asyncio.to_thread(_atomic_write) + + summary = ( + f"Applied {len(raw_ops)} operation(s) to {path.name} " + f"({len(lines)} → {len(new_lines)} lines)." + ) + return ToolResult(success=True, output=f"{summary}\n\n{diff}" if diff else summary) diff --git a/navi/tools/filesystem.py b/navi/tools/filesystem.py index 9396e92..ace2e7d 100644 --- a/navi/tools/filesystem.py +++ b/navi/tools/filesystem.py @@ -1,1145 +1,57 @@ -"""Filesystem tool — read/write/append/list/find/info/move/delete/exists + AI query/smart_edit. +"""Compatibility shim — the filesystem tool lives in `navi.tools._filesystem`. -If FS_ALLOWED_PATHS=* (default), any path is accessible. -Otherwise set a comma-separated list of allowed root paths, e.g.: - FS_ALLOWED_PATHS=/home/user,/var/www +Existing imports (`from navi.tools.filesystem import FilesystemTool`) keep +working; new code should import from `navi.tools._filesystem`. """ -import asyncio -import difflib -import re -import shutil -import stat -from datetime import datetime -from pathlib import Path - -from navi.config import settings - -from ._internal.areas import allowed_areas, resolve_in_areas -from ._internal.base import ( - Tool, - ToolContext, - ToolResult, - current_session_id, - current_user_id, - current_user_role, - current_working_directory, +from ._filesystem import FilesystemTool +from ._filesystem.helpers import ( + _AI_CHUNK_CHARS, + _AI_EDIT_MAX_CHARS, + _AI_OVERLAP_LINES, + _DIFF_HUNK_RE, + _EDIT_SYSTEM, + _FIND_MAX_RESULTS, + _LIST_MAX_ENTRIES, + _QUERY_CHUNK_SYSTEM, + _QUERY_SINGLE_SYSTEM, + _QUERY_SYNTHESIS_SYSTEM, + _READ_HARD_BYTES, + _READ_WARN_BYTES, + _apply_ops, + _check_path, + _fmt_size, + _fmt_time, + _make_chunks, + _number_diff, + _number_lines, + _resolve_relative_path, + _unified_diff, + _validate_ops, ) -_READ_WARN_BYTES = 100_000 # 100 KB — add size warning in output -_READ_HARD_BYTES = 1_000_000 # 1 MB — refuse full read without offset/limit -_LIST_MAX_ENTRIES = 500 -_FIND_MAX_RESULTS = 200 - -# AI actions: ~20k tokens of file content per chunk (4 chars ≈ 1 token) -_AI_CHUNK_CHARS = 80_000 -_AI_OVERLAP_LINES = 30 -# smart_edit: refuse files larger than ~50k tokens (full file must fit in one call) -_AI_EDIT_MAX_CHARS = 200_000 - -# ── System prompts ──────────────────────────────────────────────────────────── - -_QUERY_SINGLE_SYSTEM = ( - "You are a precise file analysis assistant. " - "Answer the question based strictly on the file content shown. " - "Be specific and concise. Include line numbers when relevant." -) - -_QUERY_CHUNK_SYSTEM = ( - "You are analyzing one section of a larger file. " - "Answer the question using only the lines shown. " - "If the answer is not present in this section, respond with exactly: NOT_FOUND\n" - "Otherwise be specific and include line numbers." -) - -_QUERY_SYNTHESIS_SYSTEM = ( - "Combine these partial findings from different sections of a file into one clear answer. " - "Remove duplicates. Be direct and concise." -) - -_EDIT_SYSTEM = ( - "You are a precise file editor. " - "Given file content with line numbers and an instruction, output ONLY a JSON array of edit operations. " - "No explanation, no markdown — just the JSON.\n\n" - "Format:\n" - "[\n" - ' {"op": "replace", "start": LINE, "end": LINE, "content": "new text\\nmore lines"},\n' - ' {"op": "delete", "start": LINE, "end": LINE},\n' - ' {"op": "insert", "after": LINE, "content": "text to insert"}\n' - "]\n\n" - "Rules:\n" - "- Line numbers are 1-based and inclusive\n" - "- Use \\n in content strings for embedded newlines\n" - "- Make MINIMAL changes to accomplish the instruction\n" - "- 'insert' after=0 inserts before the first line\n" - "- If no changes are needed, return []" -) - - -# ── Path helpers ────────────────────────────────────────────────────────────── - - -def _resolve_relative_path( - p: Path, user_id: str | None, role: str | None, cwd: str | None = None -) -> Path: - """Resolve a relative path against the session cwd when available. - - In single-user/legacy/admin mode, uses the client-provided working directory - so that `filesystem read src/main.py` works from the project directory the - user launched Navi Code in, regardless of the server's cwd. - """ - if user_id and role != "admin": - # Sandbox mode keeps its own resolution rules. - return p - - cwd = cwd or current_working_directory.get(None) - if cwd: - return (Path(cwd) / p).resolve() - return p.resolve() - - -def _check_path( - path_str: str, - user_id: str | None = None, - role: str | None = None, - cwd: str | None = None, -) -> Path | None: - """Return resolved Path if access is allowed, else None. - - When a user_id is active (multi-user mode), all paths are resolved - inside user_data// or inside the current session directory. - This prevents users from accessing each other's files or random OS - directories, while still letting a user read what an upload put in the - session and what share_file hands back. - """ - if not path_str or path_str.strip() == "": - return None - try: - p = Path(path_str) - except Exception: - return None - - user_id = user_id or current_user_id.get(None) - role = role or current_user_role.get() - - # Admins bypass sandbox and use FS_ALLOWED_PATHS directly - if user_id and role != "admin": - # Sandboxed mode: the user's own sandbox, plus the current session - # directory. Relative paths keep resolving inside the sandbox. - areas = allowed_areas(user_id, current_session_id.get(None)) - return resolve_in_areas(p, areas) - - # Fallback to FS_ALLOWED_PATHS for single-user / legacy mode / admin - try: - if not p.is_absolute(): - p = _resolve_relative_path(p, user_id, role, cwd=cwd) - else: - p = p.expanduser().resolve() - except Exception: - return None - - if settings.fs_allowed_paths.strip() == "*": - return p - - allowed = [Path(r).expanduser().resolve() for r in settings.fs_allowed_paths_list] - for root in allowed: - try: - p.relative_to(root) - return p - except ValueError: - continue - return None - - -def _fmt_size(n: int) -> str: - if n < 1024: - return f"{n} B" - if n < 1024**2: - return f"{n / 1024:.1f} KB" - if n < 1024**3: - return f"{n / 1024**2:.1f} MB" - return f"{n / 1024**3:.1f} GB" - - -def _fmt_time(ts: float) -> str: - return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") - - -# ── AI helpers (module-level, no self) ──────────────────────────────────────── - - -def _number_lines(lines: list[str], start: int = 1) -> str: - """Return file lines with 1-based line numbers, right-aligned.""" - width = len(str(start + len(lines) - 1)) - return "\n".join(f"{start + i:>{width}}: {line}" for i, line in enumerate(lines)) - - -def _make_chunks(lines: list[str], target_chars: int, overlap: int) -> list[tuple[int, int]]: - """ - Split lines into (start_idx, end_idx) chunks of at most target_chars characters. - Consecutive chunks overlap by `overlap` lines to preserve boundary context. - """ - if not lines: - return [(0, 0)] - total = len(lines) - total_chars = sum(len(line) + 1 for line in lines) - if total_chars <= target_chars: - return [(0, total)] - - chunks: list[tuple[int, int]] = [] - start = 0 - while start < total: - chars = 0 - end = start - while end < total and chars < target_chars: - chars += len(lines[end]) + 1 - end += 1 - chunks.append((start, end)) - if end >= total: - break - start = max(start + 1, end - overlap) # always make progress - - return chunks or [(0, total)] - - -def _validate_ops(ops: list, total_lines: int) -> list[str]: - errors: list[str] = [] - for i, op in enumerate(ops): - if not isinstance(op, dict): - errors.append(f"op[{i}] is not a dict") - continue - kind = op.get("op") - if kind not in ("replace", "delete", "insert"): - errors.append(f"op[{i}] unknown type {kind!r}") - continue - if kind in ("replace", "delete"): - s, e = op.get("start"), op.get("end") - if not isinstance(s, int) or not isinstance(e, int): - errors.append(f"op[{i}] start/end must be integers") - elif s < 1 or e > total_lines or s > e: - errors.append(f"op[{i}] range {s}-{e} out of bounds (file has {total_lines} lines)") - elif kind == "insert": - after = op.get("after") - if not isinstance(after, int): - errors.append(f"op[{i}] 'after' must be integer") - elif after < 0 or after > total_lines: - errors.append(f"op[{i}] 'after'={after} out of bounds (0–{total_lines})") - return errors - - -def _apply_ops(lines: list[str], ops: list[dict]) -> list[str]: - """Apply edit operations bottom-up (highest line first) to preserve line numbers.""" - sorted_ops = sorted( - ops, - key=lambda o: o.get("start", o.get("after", 0)), - reverse=True, - ) - result = list(lines) - for op in sorted_ops: - kind = op["op"] - if kind == "replace": - s = op["start"] - 1 # 0-based - e = op["end"] # exclusive (1-based end = exclusive 0-based end) - new = op.get("content", "").split("\n") - result[s:e] = new - elif kind == "delete": - s = op["start"] - 1 - e = op["end"] - del result[s:e] - elif kind == "insert": - after = op["after"] # insert after this 1-based line (0 = before line 1) - new = op.get("content", "").split("\n") - result[after:after] = new - return result - - -_DIFF_HUNK_RE = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@") - - -def _number_diff(diff_lines: list[str]) -> list[str]: - """Prefix each unified-diff content line with its real file line number. - - Numbers come from the enclosing ``@@ -old_start,n +new_start,n @@`` hunk: - removed and context lines use the old (from) line number, added lines use the - new (to) line number. The format ``{marker} {num:>width}│ {content}`` keeps the - ``+``/``-``/`` `` marker first so existing ``startswith``-based highlighting - stays valid; a fixed space separates the marker from the right-aligned number - column. Hunk headers, file headers (``---``/``+++``) and the - ``\\ No newline`` marker are left unchanged. - """ - # First pass: assign numbers, track the max for column width. - numbered: list[tuple[str, int | None, str]] = [] # (marker, num, content) - max_num = 0 - old_line = 0 - new_line = 0 - for raw_line in diff_lines: - # difflib.unified_diff keeps a trailing "\n" on content lines when the - # input was fed with explicit newlines (see _unified_diff, which adds - # "\n" so difflib does not emit a "\ No newline at end of file" marker). - # Strip it here so the later "\n".join does not produce a blank line - # between every diff line. No-op for the _diff action, whose lines come - # from splitlines() without trailing newlines. - line = raw_line.rstrip("\n") - if line.startswith("@@"): - m = _DIFF_HUNK_RE.match(line) - if m: - old_line = int(m.group(1)) - new_line = int(m.group(3)) - numbered.append(("", None, line)) - continue - if line.startswith("---") or line.startswith("+++") or line.startswith("\\"): - numbered.append(("", None, line)) - continue - marker = line[0] if line else " " - content = line[1:] - if marker == "+": - num = new_line - new_line += 1 - elif marker == "-": - num = old_line - old_line += 1 - else: # " " context — present in both old and new - num = old_line - old_line += 1 - new_line += 1 - if num > max_num: - max_num = num - numbered.append((marker, num, content)) - - width = len(str(max_num)) if max_num else 1 - out: list[str] = [] - for marker, num, content in numbered: - if num is None: - out.append(content) - else: - out.append(f"{marker} {num:>{width}}│ {content}") - return out - - -def _unified_diff(original: list[str], modified: list[str], path: Path) -> str: - diff = list( - difflib.unified_diff( - [line + "\n" for line in original], - [line + "\n" for line in modified], - fromfile=f"a/{path.name}", - tofile=f"b/{path.name}", - lineterm="", - ) - ) - return "\n".join(_number_diff(diff)) - - -# ── Tool class ──────────────────────────────────────────────────────────────── - - -class FilesystemTool(Tool): - name = "filesystem" - description = ( - "Operate the local filesystem: read, write, edit, search, and manage files. " - "Editing policy — cheapest deterministic method first, smart_edit LAST: use " - "'edit' (exact text, the default) or 'edit_lines' (known line numbers); both " - "are deterministic. Reach for 'smart_edit' ONLY when the change cannot be " - "expressed as text or line numbers — it costs an LLM call and reads the whole " - "file. Call 'info' before reading an unknown file. Full action list, argument " - 'rules and access limits: tool_manual("filesystem").' - ) - parameters = { - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": [ - "read", - "write", - "append", - "edit", - "edit_lines", - "list", - "find", - "find_up", - "info", - "move", - "copy", - "delete", - "exists", - "mkdir", - "query", - "smart_edit", - "grep", - "diff", - ], - "description": "Operation to perform.", - }, - "path": { - "type": "string", - "description": "Absolute or relative file/directory path (~ is expanded).", - }, - "content": { - "type": "string", - "description": "Text to write or append (required for write/append).", - }, - "old": { - "type": "string", - "description": "Exact text to replace (required for edit). Must occur exactly once.", - }, - "new": { - "type": "string", - "description": "Replacement text for edit.", - }, - "operations": { - "type": "array", - "description": ( - "Line-based edits (required for edit_lines). 1-based, inclusive. " - "'after' applies to insert." - ), - "items": { - "type": "object", - "properties": { - "op": {"type": "string", "enum": ["replace", "delete", "insert"]}, - "start": {"type": "integer"}, - "end": {"type": "integer"}, - "after": {"type": "integer"}, - "content": {"type": "string"}, - }, - "required": ["op"], - }, - }, - "numbered": { - "type": "boolean", - "description": "Number the lines in read output (default true). Set false to copy text into 'edit'.", - }, - "destination": { - "type": "string", - "description": "Second path: the target for move/copy, or the file to compare against for diff.", - }, - "pattern": { - "type": "string", - "description": "Glob pattern for find (e.g. '*.log'), or exact filename for find_up.", - }, - "offset": { - "type": "integer", - "description": "First line to read, 1-based (for read action).", - }, - "limit": { - "type": "integer", - "description": "Max lines to return (for read action).", - }, - "recursive": { - "type": "boolean", - "description": "Full recursive directory tree (for list, default false).", - }, - "regex": { - "type": "boolean", - "description": "Use regex matching for grep (default false = literal substring search).", - }, - "question": { - "type": "string", - "description": ( - "Natural language question about the file (for query) — the answer comes " - "back instead of the file." - ), - }, - "instruction": { - "type": "string", - "description": "Natural language edit instruction (for smart_edit).", - }, - }, - "required": ["action", "path"], - } - - def __init__(self, ai_helper=None) -> None: - # ai_helper is optional — standard actions work without it - self._ai = ai_helper - - async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult: - action = params.get("action", "") - raw_path = params.get("path", "") - user_id = ctx.user_id if ctx else current_user_id.get(None) - role = ctx.user_role if ctx else current_user_role.get() - path = _check_path(raw_path, user_id, role, cwd=ctx.cwd if ctx else None) - - if path is None: - if not raw_path or raw_path.strip() == "": - return ToolResult( - success=False, - output="'path' is required. For 'write' action use 'path' (not 'destination').", - error="missing_path", - ) - return ToolResult( - success=False, - output=( - f"Access denied: '{raw_path}' is outside allowed paths " - f"({settings.fs_allowed_paths}). " - "Set FS_ALLOWED_PATHS=* in .env to allow all paths." - ), - error="access_denied", - ) - - try: - match action: - case "read": - return await asyncio.to_thread(self._read, path, params) - case "write": - return await asyncio.to_thread(self._write, path, params) - case "append": - return await asyncio.to_thread(self._append, path, params) - case "edit": - return await asyncio.to_thread(self._edit, path, params) - case "edit_lines": - return await asyncio.to_thread(self._edit_lines, path, params) - case "list": - return await asyncio.to_thread(self._list, path, params) - case "find": - return await asyncio.to_thread(self._find, path, params) - case "find_up": - return await asyncio.to_thread(self._find_up, path, params) - case "info": - return await asyncio.to_thread(self._info, path) - case "move": - return await asyncio.to_thread(self._move, path, params) - case "copy": - return await asyncio.to_thread(self._copy, path, params) - case "delete": - return await asyncio.to_thread(self._delete, path) - case "exists": - return ToolResult(success=True, output="true" if path.exists() else "false") - case "mkdir": - return await asyncio.to_thread(self._mkdir, path) - case "query": - return await self._query(path, params) - case "smart_edit": - return await self._smart_edit(path, params) - case "grep": - return await asyncio.to_thread(self._grep, path, params) - case "diff": - return await asyncio.to_thread(self._diff, path, params) - case _: - return ToolResult( - success=False, output=f"Unknown action: {action}", error="invalid_action" - ) - - except PermissionError as e: - return ToolResult(success=False, output=f"Permission denied: {e}", error=str(e)) - except Exception as e: - return ToolResult(success=False, output=f"Filesystem error: {e}", error=str(e)) - - # ── Standard action handlers ────────────────────────────────────────────── - - def _read(self, path: Path, params: dict) -> ToolResult: - if not path.exists(): - return ToolResult(success=False, output=f"File not found: {path}", error="not_found") - if path.is_dir(): - return ToolResult( - success=False, - output=f"Path is a directory, use 'list': {path}", - error="is_directory", - ) - - file_size = path.stat().st_size - offset = params.get("offset") - limit = params.get("limit") - - if file_size > _READ_HARD_BYTES and offset is None and limit is None: - return ToolResult( - success=False, - output=( - f"File too large to read in full: {_fmt_size(file_size)} — {path}\n" - "Use offset/limit to read specific line ranges " - "(e.g. offset=1, limit=100), or use 'query' to ask a question about it." - ), - error="file_too_large", - ) - - text = path.read_text(encoding="utf-8", errors="replace") - lines = text.splitlines(keepends=True) - total_lines = len(lines) - - if offset is not None or limit is not None: - start = max(0, (offset or 1) - 1) - end = (start + limit) if limit is not None else total_lines - selected = lines[start:end] - actual_end = min(end, total_lines) - header = ( - f"[{path} | lines {start + 1}–{actual_end} of {total_lines}" - f" | {_fmt_size(file_size)}]\n" - ) - numbered = params.get("numbered", True) - body = ( - _number_lines([l.rstrip("\r\n") for l in selected], start=start + 1) - if numbered - else "".join(selected) - ) - return ToolResult(success=True, output=header + body) - - warn = ( - f"⚠ Large file ({_fmt_size(file_size)}) — consider offset/limit next time.\n" - if file_size > _READ_WARN_BYTES - else "" - ) - numbered = params.get("numbered", True) - header = f"[{path} | {total_lines} lines | {_fmt_size(file_size)}]\n" - if numbered: - return ToolResult( - success=True, - output=header + warn + _number_lines(text.splitlines(keepends=False), 1), - ) - return ToolResult(success=True, output=header + warn + text) - - def _write(self, path: Path, params: dict) -> ToolResult: - content = params.get("content", "") - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(content, encoding="utf-8") - lines = len(content.splitlines()) - return ToolResult( - success=True, - output=f"Written {_fmt_size(len(content.encode()))} ({lines} lines) → {path}", - ) - - def _append(self, path: Path, params: dict) -> ToolResult: - content = params.get("content", "") - if not content: - return ToolResult( - success=False, output="'content' is required for append", error="missing_content" - ) - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("a", encoding="utf-8") as f: - f.write(content) - return ToolResult( - success=True, - output=f"Appended {_fmt_size(len(content.encode()))} to {path} (file now {_fmt_size(path.stat().st_size)})", - ) - - def _edit(self, path: Path, params: dict) -> ToolResult: - old = params.get("old") - new = params.get("new") - if old is None: - return ToolResult( - success=False, output="'old' is required for edit", error="missing_old" - ) - if new is None: - return ToolResult( - success=False, output="'new' is required for edit", error="missing_new" - ) - if old == "": - return ToolResult(success=False, output="'old' must not be empty", error="empty_old") - if not path.exists(): - return ToolResult(success=False, output=f"File not found: {path}", error="not_found") - if path.is_dir(): - return ToolResult( - success=False, output="edit works on files, not directories.", error="is_directory" - ) - - text = path.read_text(encoding="utf-8", errors="replace") - count = text.count(old) - if count == 0: - return ToolResult( - success=False, - output=( - "Exact text not found. Read the relevant file section and call edit again " - "with text copied exactly from the file." - ), - error="old_not_found", - ) - if count > 1: - return ToolResult( - success=False, - output=( - f"Exact text occurs {count} times. Prefer 'edit' with a larger " - "unique 'old' block (include surrounding lines so it matches once) " - "or 'edit_lines' with line numbers. Fall back to 'smart_edit' only if " - "the change is genuinely semantic and cannot be expressed as exact text." - ), - error="old_not_unique", - ) - - new_text = text.replace(old, new, 1) - path.write_text(new_text, encoding="utf-8") - delta = len(new_text.encode()) - len(text.encode()) - summary = ( - f"Edited {path}: replaced {len(old.encode())} B with " - f"{len(new.encode())} B (delta {delta:+d} B)." - ) - # Keep the model-facing output short; carry the unified diff in metadata - # for the TUI to render with red/green highlighting (the model does not - # see metadata, so its context stays unchanged). - diff = _unified_diff(text.splitlines(), new_text.splitlines(), path) - return ToolResult( - success=True, - output=summary, - metadata={"diff": diff} if diff else {}, - ) - - def _edit_lines(self, path: Path, params: dict) -> ToolResult: - ops = params.get("operations") - if not ops: - return ToolResult( - success=False, - output="'operations' is required for edit_lines", - error="missing_operations", - ) - if not path.exists(): - return ToolResult(success=False, output=f"File not found: {path}", error="not_found") - if path.is_dir(): - return ToolResult( - success=False, - output="edit_lines works on files, not directories.", - error="is_directory", - ) - - text = path.read_text(encoding="utf-8", errors="replace") - lines = text.splitlines() - errors = _validate_ops(ops, len(lines)) - if errors: - return ToolResult( - success=False, - output="Invalid operations:\n" + "\n".join(f" • {e}" for e in errors), - error="invalid_ops", - ) - new_lines = _apply_ops(lines, ops) - diff = _unified_diff(lines, new_lines, path) - - new_text = "\n".join(new_lines) + ("\n" if text.endswith("\n") else "") - tmp = path.with_suffix(path.suffix + ".tmp") - try: - tmp.write_text(new_text, encoding="utf-8") - tmp.replace(path) - finally: - if tmp.exists(): - tmp.unlink(missing_ok=True) - - summary = ( - f"Applied {len(ops)} operation(s) to {path.name} " - f"({len(lines)} → {len(new_lines)} lines)." - ) - return ToolResult(success=True, output=f"{summary}\n\n{diff}" if diff else summary) - - def _list(self, path: Path, params: dict) -> ToolResult: - if not path.exists(): - return ToolResult(success=False, output=f"Path not found: {path}", error="not_found") - if path.is_file(): - return self._info(path) - - recursive = params.get("recursive", False) - raw_entries = list(path.rglob("*") if recursive else path.iterdir()) - raw_entries.sort(key=lambda e: (e.is_file(), str(e).lower())) - - truncated = len(raw_entries) > _LIST_MAX_ENTRIES - entries = raw_entries[:_LIST_MAX_ENTRIES] - lines = [] - for e in entries: - try: - s = e.stat() - rel = e.relative_to(path) - if e.is_dir(): - if not recursive: - try: - n = sum(1 for _ in e.iterdir()) - lines.append(f"d {rel}/ ({n} items)") - except PermissionError: - lines.append(f"d {rel}/") - else: - lines.append(f"d {rel}/") - else: - lines.append( - f" {str(rel):<48} {_fmt_size(s.st_size):>10} {_fmt_time(s.st_mtime)}" - ) - except Exception: - lines.append(f"? {e.name}") - - note = " ⚠ truncated" if truncated else "" - header = f"[{path} | {len(entries)} entries{note}]\n" - return ToolResult(success=True, output=header + ("\n".join(lines) or "(empty directory)")) - - def _find(self, path: Path, params: dict) -> ToolResult: - pattern = params.get("pattern") - if not pattern: - return ToolResult( - success=False, output="'pattern' is required for find", error="missing_pattern" - ) - if not path.exists(): - return ToolResult(success=False, output=f"Path not found: {path}", error="not_found") - - matches: list[Path] = [] - try: - for p in path.rglob(pattern): - matches.append(p) - if len(matches) >= _FIND_MAX_RESULTS: - break - except Exception as e: - return ToolResult(success=False, output=f"Find error: {e}", error=str(e)) - - if not matches: - return ToolResult(success=True, output=f"No matches for '{pattern}' in {path}") - - matches.sort() - lines = [] - for m in matches: - try: - size = _fmt_size(m.stat().st_size) if m.is_file() else "" - lines.append(f"{m} ({size})") - except Exception: - lines.append(str(m)) - - extra = ( - f" ⚠ showing first {_FIND_MAX_RESULTS}" if len(matches) == _FIND_MAX_RESULTS else "" - ) - header = f"[{len(matches)} matches for '{pattern}' in {path}{extra}]\n" - return ToolResult(success=True, output=header + "\n".join(lines)) - - def _find_up(self, path: Path, params: dict) -> ToolResult: - filename = params.get("pattern") - if not filename: - return ToolResult(success=False, output="", error="pattern is required for find_up") - current = path if path.is_dir() else path.parent - checked = [] - while True: - target = current / filename - checked.append(str(target)) - if target.exists(): - return ToolResult(success=True, output=str(target)) - parent = current.parent - if parent == current: - return ToolResult( - success=True, output=f"not found (searched: {', '.join(checked)})" - ) - current = parent - - def _info(self, path: Path) -> ToolResult: - if not path.exists(): - return ToolResult(success=False, output=f"Not found: {path}", error="not_found") - - s = path.stat() - kind = "symlink" if path.is_symlink() else ("directory" if path.is_dir() else "file") - lines = [ - f"path: {path}", - f"type: {kind}", - f"size: {_fmt_size(s.st_size)}", - f"modified: {_fmt_time(s.st_mtime)}", - f"created: {_fmt_time(s.st_ctime)}", - f"mode: {stat.filemode(s.st_mode)}", - ] - if path.is_file(): - try: - text = path.read_text(encoding="utf-8", errors="replace") - lines.append(f"lines: {len(text.splitlines())}") - except Exception: - lines.append("lines: (binary or unreadable)") - elif path.is_dir(): - try: - children = list(path.iterdir()) - lines.append( - f"contents: {sum(c.is_file() for c in children)} files, {sum(c.is_dir() for c in children)} dirs (top level)" - ) - except Exception: - pass - return ToolResult(success=True, output="\n".join(lines)) - - def _move(self, path: Path, params: dict) -> ToolResult: - dest_raw = params.get("destination") - if not dest_raw: - return ToolResult( - success=False, - output="'destination' is required for move", - error="missing_destination", - ) - dest = _check_path(dest_raw) - if dest is None: - return ToolResult( - success=False, - output=f"Access denied: destination '{dest_raw}' outside allowed paths.", - error="access_denied", - ) - if not path.exists(): - return ToolResult(success=False, output=f"Not found: {path}", error="not_found") - dest.parent.mkdir(parents=True, exist_ok=True) - shutil.move(str(path), str(dest)) - return ToolResult(success=True, output=f"Moved: {path} → {dest}") - - def _mkdir(self, path: Path) -> ToolResult: - path.mkdir(parents=False, exist_ok=True) - return ToolResult(success=True, output=f"Directory created: {path}") - - def _delete(self, path: Path) -> ToolResult: - if not path.exists(): - return ToolResult(success=False, output=f"Not found: {path}", error="not_found") - if path.is_dir(): - shutil.rmtree(path) - else: - path.unlink() - return ToolResult(success=True, output=f"Deleted: {path}") - - def _copy(self, path: Path, params: dict) -> ToolResult: - dest_raw = params.get("destination") - if not dest_raw: - return ToolResult( - success=False, - output="'destination' is required for copy", - error="missing_destination", - ) - dest = _check_path(dest_raw) - if dest is None: - return ToolResult( - success=False, - output=f"Access denied: destination '{dest_raw}' outside allowed paths.", - error="access_denied", - ) - if not path.exists(): - return ToolResult(success=False, output=f"Not found: {path}", error="not_found") - dest.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(str(path), str(dest)) - return ToolResult(success=True, output=f"Copied: {path} → {dest}") - - def _grep(self, path: Path, params: dict) -> ToolResult: - pattern = params.get("pattern", "").strip() - if not pattern: - return ToolResult( - success=False, output="'pattern' is required for grep", error="missing_pattern" - ) - use_regex = params.get("regex", False) - glob_filter = params.get("glob", "") - - try: - if use_regex: - compiled = re.compile(pattern, re.IGNORECASE) - else: - compiled = re.compile(re.escape(pattern), re.IGNORECASE) - except re.error as e: - return ToolResult( - success=False, output=f"Invalid pattern: {e}", error="invalid_pattern" - ) - - max_results = 200 - matches: list[str] = [] - files_searched = 0 - - if path.is_file(): - files_to_search = [path] - elif path.is_dir(): - if glob_filter: - files_to_search = list(path.rglob(glob_filter)) - else: - files_to_search = list(path.rglob("*")) - files_to_search = [f for f in files_to_search if f.is_file()] - else: - return ToolResult(success=False, output=f"Not found: {path}", error="not_found") - - for fpath in files_to_search: - if len(matches) >= max_results: - break - files_searched += 1 - try: - text = fpath.read_text(encoding="utf-8", errors="replace") - for i, line in enumerate(text.splitlines(), 1): - if compiled.search(line): - rel = fpath.relative_to(path) if fpath != path else fpath.name - matches.append(f"{rel}:{i}: {line}") - if len(matches) >= max_results: - break - except Exception: - continue - - if not matches: - return ToolResult(success=True, output=f"No matches for '{pattern}' in {path}") - - note = f" ⚠ showing first {max_results}" if len(matches) == max_results else "" - header = f"[{len(matches)} matches for '{pattern}' in {path} ({files_searched} files searched){note}]\n" - return ToolResult(success=True, output=header + "\n".join(matches)) - - def _diff(self, path: Path, params: dict) -> ToolResult: - dest_raw = params.get("destination") - if not dest_raw: - return ToolResult( - success=False, - output="'destination' is required for diff (path to second file)", - error="missing_destination", - ) - dest = _check_path(dest_raw) - if dest is None: - return ToolResult( - success=False, - output=f"Access denied: destination '{dest_raw}' outside allowed paths.", - error="access_denied", - ) - if not path.exists(): - return ToolResult(success=False, output=f"Not found: {path}", error="not_found") - if not dest.exists(): - return ToolResult(success=False, output=f"Not found: {dest}", error="not_found") - if path.is_dir() or dest.is_dir(): - return ToolResult( - success=False, output="diff works on files, not directories.", error="is_directory" - ) - - try: - a_lines = path.read_text(encoding="utf-8", errors="replace").splitlines() - b_lines = dest.read_text(encoding="utf-8", errors="replace").splitlines() - except Exception as e: - return ToolResult(success=False, output=f"Read error: {e}", error=str(e)) - - diff = list( - difflib.unified_diff( - a_lines, - b_lines, - fromfile=f"a/{path.name}", - tofile=f"b/{dest.name}", - lineterm="", - ) - ) - if not diff: - return ToolResult(success=True, output="Files are identical.") - return ToolResult(success=True, output="\n".join(_number_diff(diff))) - - # ── AI action handlers ──────────────────────────────────────────────────── - - def _require_ai(self) -> ToolResult | None: - if not self._ai: - return ToolResult( - success=False, - output="AI helper is not available for this action.", - error="no_ai_helper", - ) - return None - - async def _query(self, path: Path, params: dict) -> ToolResult: - if (err := self._require_ai()) is not None: - return err - - question = params.get("question", "").strip() - if not question: - return ToolResult( - success=False, output="'question' is required for query.", error="missing_question" - ) - if not path.exists(): - return ToolResult(success=False, output=f"File not found: {path}", error="not_found") - if path.is_dir(): - return ToolResult( - success=False, output="query works on files, not directories.", error="is_directory" - ) - - text = await asyncio.to_thread(path.read_text, "utf-8", "replace") - lines = text.splitlines() - total = len(lines) - chunks = _make_chunks(lines, _AI_CHUNK_CHARS, _AI_OVERLAP_LINES) - - if len(chunks) == 1: - s, e = chunks[0] - numbered = _number_lines(lines[s:e], s + 1) - answer = await self._ai.ask( - _QUERY_SINGLE_SYSTEM, - f"File: {path}\n\nQuestion: {question}\n\nContent:\n{numbered}", - ) - return ToolResult(success=True, output=answer) - - # Multi-chunk: accumulate partial answers - partials: list[str] = [] - for s, e in chunks: - numbered = _number_lines(lines[s:e], s + 1) - partial = await self._ai.ask( - _QUERY_CHUNK_SYSTEM, - f"File: {path} (lines {s + 1}–{e} of {total})\nQuestion: {question}\n\nContent:\n{numbered}", - ) - if partial and "NOT_FOUND" not in partial.upper(): - partials.append(f"[lines {s + 1}–{e}] {partial}") - - if not partials: - return ToolResult( - success=True, - output=f"No information found in '{path.name}' relevant to: {question}", - ) - - if len(partials) == 1: - # Single finding — strip range prefix, return directly - answer = partials[0].split("] ", 1)[-1] - return ToolResult(success=True, output=answer) - - answer = await self._ai.ask( - _QUERY_SYNTHESIS_SYSTEM, - f"Question: {question}\n\nFindings from {len(partials)} sections:\n\n" - + "\n\n".join(partials), - ) - return ToolResult(success=True, output=answer) - - async def _smart_edit(self, path: Path, params: dict) -> ToolResult: - if (err := self._require_ai()) is not None: - return err - - instruction = params.get("instruction", "").strip() - if not instruction: - return ToolResult( - success=False, - output="'instruction' is required for smart_edit.", - error="missing_instruction", - ) - if not path.exists(): - return ToolResult(success=False, output=f"File not found: {path}", error="not_found") - if path.is_dir(): - return ToolResult( - success=False, - output="smart_edit works on files, not directories.", - error="is_directory", - ) - - text = await asyncio.to_thread(path.read_text, "utf-8", "replace") - if len(text) > _AI_EDIT_MAX_CHARS: - return ToolResult( - success=False, - output=( - f"File too large for smart_edit ({_fmt_size(len(text.encode()))}, " - f"limit {_fmt_size(_AI_EDIT_MAX_CHARS)}). " - "Use read with offset/limit to locate the relevant section, then write it back." - ), - error="file_too_large", - ) - - lines = text.splitlines() - numbered = _number_lines(lines, 1) - - raw_ops = await self._ai.ask_json( - _EDIT_SYSTEM, - f"File: {path}\nTotal lines: {len(lines)}\n\nInstruction: {instruction}\n\nContent:\n{numbered}", - ) - - if raw_ops is None: - return ToolResult( - success=False, - output="AI could not produce valid edit operations. Try rephrasing the instruction.", - error="invalid_ai_response", - ) - - if not isinstance(raw_ops, list): - raw_ops = [raw_ops] if isinstance(raw_ops, dict) else [] - - if not raw_ops: - return ToolResult(success=True, output="No changes needed — file unchanged.") - - errors = _validate_ops(raw_ops, len(lines)) - if errors: - return ToolResult( - success=False, - output="AI returned invalid operations:\n" + "\n".join(f" • {e}" for e in errors), - error="invalid_ops", - ) - - new_lines = _apply_ops(lines, raw_ops) - diff = _unified_diff(lines, new_lines, path) - - # Preserve trailing newline — write atomically to avoid partial writes on failure - import os - - new_text = "\n".join(new_lines) + ("\n" if text.endswith("\n") else "") - tmp = path.with_suffix(path.suffix + ".tmp") - - def _atomic_write() -> None: - try: - tmp.write_text(new_text, encoding="utf-8") - os.replace(tmp, path) - finally: - if tmp.exists(): - tmp.unlink(missing_ok=True) - - await asyncio.to_thread(_atomic_write) - - summary = ( - f"Applied {len(raw_ops)} operation(s) to {path.name} " - f"({len(lines)} → {len(new_lines)} lines)." - ) - return ToolResult(success=True, output=f"{summary}\n\n{diff}" if diff else summary) +__all__ = [ + "_AI_CHUNK_CHARS", + "_AI_EDIT_MAX_CHARS", + "_AI_OVERLAP_LINES", + "_DIFF_HUNK_RE", + "_EDIT_SYSTEM", + "_FIND_MAX_RESULTS", + "_LIST_MAX_ENTRIES", + "_QUERY_CHUNK_SYSTEM", + "_QUERY_SINGLE_SYSTEM", + "_QUERY_SYNTHESIS_SYSTEM", + "_READ_HARD_BYTES", + "_READ_WARN_BYTES", + "FilesystemTool", + "_apply_ops", + "_check_path", + "_fmt_size", + "_fmt_time", + "_make_chunks", + "_number_diff", + "_number_lines", + "_resolve_relative_path", + "_unified_diff", + "_validate_ops", +] diff --git a/tests/unit/tools/test_filesystem.py b/tests/unit/tools/test_filesystem.py index 57c16ec..c718efa 100644 --- a/tests/unit/tools/test_filesystem.py +++ b/tests/unit/tools/test_filesystem.py @@ -9,10 +9,14 @@ class TestCheckPath: @pytest.fixture(autouse=True) def _allow_all(self, monkeypatch): - import navi.tools.filesystem as _fs_mod + # `settings` is read in both implementation modules of the split tool. + import navi.tools._filesystem.helpers as _helpers + import navi.tools._filesystem.tool as _tool from navi.config import Settings - monkeypatch.setattr(_fs_mod, "settings", Settings(fs_allowed_paths="*")) + new = Settings(fs_allowed_paths="*") + for mod in (_helpers, _tool): + monkeypatch.setattr(mod, "settings", new) def test_resolves_relative(self): p = _check_path(".") @@ -27,14 +31,17 @@ assert _check_path("") is None def test_restricted_paths(self, monkeypatch, tmp_path): - import navi.tools.filesystem as _fs_mod + import navi.tools._filesystem.helpers as _helpers + import navi.tools._filesystem.tool as _tool from navi.config import Settings allowed = tmp_path / "allowed" allowed.mkdir() blocked = tmp_path / "blocked" blocked.mkdir() - monkeypatch.setattr(_fs_mod, "settings", Settings(fs_allowed_paths=str(allowed))) + new = Settings(fs_allowed_paths=str(allowed)) + for mod in (_helpers, _tool): + monkeypatch.setattr(mod, "settings", new) assert _check_path(str(allowed / "file.txt")) is not None assert _check_path(str(blocked / "file.txt")) is None @@ -107,14 +114,17 @@ class TestFilesystemToolBasic: @pytest.fixture(autouse=True) def _allow_all(self, monkeypatch): - import navi.tools.filesystem as _fs_mod + import navi.tools._filesystem.helpers as _helpers + import navi.tools._filesystem.tool as _tool from navi.config import Settings async def _to_thread(func, *args, **kwargs): return func(*args, **kwargs) - monkeypatch.setattr(_fs_mod, "settings", Settings(fs_allowed_paths="*")) - monkeypatch.setattr(_fs_mod.asyncio, "to_thread", _to_thread) + new = Settings(fs_allowed_paths="*") + for mod in (_helpers, _tool): + monkeypatch.setattr(mod, "settings", new) + monkeypatch.setattr(_tool.asyncio, "to_thread", _to_thread) @pytest.fixture def tool(self):