"""File tools: read_file, write_file, edit_file, list_directory.""" from __future__ import annotations import difflib import os from pathlib import Path from typing import Any from ycode.errors import PermissionDenied, ToolError from ycode.tools.base import Tool, ToolContext, ToolResult MAX_READ_BYTES = 2_000_000 DEFAULT_READ_LINES = 2000 MAX_LINE_CHARS = 2000 def _is_binary(data: bytes) -> bool: return b"\x00" in data[:8192] def _read_text(path: Path) -> str: if not path.exists(): raise ToolError(f"File not found: {path}") if path.is_dir(): raise ToolError(f"{path} is a directory; use list_directory instead") size = path.stat().st_size if size > MAX_READ_BYTES: raise ToolError( f"{path} is {size:,} bytes, too large to read whole. Use search_files to find " "the relevant region, then read_file with offset/limit." ) data = path.read_bytes() if _is_binary(data): raise ToolError(f"{path} looks like a binary file") return data.decode("utf-8", errors="replace") def make_diff(before: str, after: str, rel: str, context: int = 3) -> str: return "".join( difflib.unified_diff( before.splitlines(keepends=True), after.splitlines(keepends=True), fromfile=f"a/{rel}", tofile=f"b/{rel}", n=context, ) ) def _diff_stats(diff: str) -> tuple[int, int]: added = sum(1 for line in diff.splitlines() if line.startswith("+") and not line.startswith("+++")) removed = sum(1 for line in diff.splitlines() if line.startswith("-") and not line.startswith("---")) return added, removed class ReadFileTool(Tool): name = "read_file" description = ( "Read a text file from the project. Returns numbered lines. For large files, " "pass offset (1-based line number) and limit to read a window." ) parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "Path relative to the project root."}, "offset": {"type": "integer", "description": "1-based line to start from (default 1)."}, "limit": {"type": "integer", "description": f"Max lines to return (default {DEFAULT_READ_LINES})."}, }, "required": ["path"], } def describe(self, args: dict[str, Any]) -> str: return f"Reading {args.get('path', '?')}" def run(self, args: dict[str, Any], ctx: ToolContext) -> ToolResult: path = ctx.permissions.paths.resolve(args["path"]) if not ctx.permissions.check_read(path): raise PermissionDenied(f"user declined reading {args['path']}") text = _read_text(path) lines = text.splitlines() offset = max(1, int(args.get("offset") or 1)) limit = max(1, int(args.get("limit") or DEFAULT_READ_LINES)) window = lines[offset - 1 : offset - 1 + limit] width = len(str(offset + len(window))) body = "\n".join( f"{n:>{width}}\t{line[:MAX_LINE_CHARS]}{' ...[truncated]' if len(line) > MAX_LINE_CHARS else ''}" for n, line in enumerate(window, start=offset) ) rel = ctx.permissions.paths.relative(path) end = offset + len(window) - 1 note = "" if end < len(lines): note = f"\n\n[Showing lines {offset}-{end} of {len(lines)}. Use offset to read more.]" if not lines: body = "[empty file]" return ToolResult( content=body + note, summary=f"Read {rel} ({len(window)} of {len(lines)} lines)", ) class WriteFileTool(Tool): name = "write_file" description = ( "Create a new file or completely overwrite an existing one. Prefer edit_file for " "changing part of an existing file. Parent directories are created as needed." ) parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "Path relative to the project root."}, "content": {"type": "string", "description": "Full file content."}, }, "required": ["path", "content"], } def describe(self, args: dict[str, Any]) -> str: return f"Writing {args.get('path', '?')}" def run(self, args: dict[str, Any], ctx: ToolContext) -> ToolResult: path = ctx.permissions.paths.resolve(args["path"], for_write=True) rel = ctx.permissions.paths.relative(path) if path.is_dir(): raise ToolError(f"{rel} is a directory") existed = path.exists() before = _read_text(path) if existed else "" if not ctx.permissions.check_write(path): raise PermissionDenied(f"user declined writing {rel}") content: str = args["content"] path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") ctx.changed_files.add(rel) diff = make_diff(before, content, rel) added, removed = _diff_stats(diff) verb = "Updated" if existed else "Created" return ToolResult( content=f"{verb} {rel} ({len(content.splitlines())} lines).", summary=f"{verb} {rel} (+{added} -{removed})", display={"diff": diff, "path": rel, "new_file": not existed}, ) class EditFileTool(Tool): name = "edit_file" description = ( "Replace an exact string in a file. old_string must match the file exactly " "(including whitespace/indentation) and be unique unless replace_all is true. " "Include enough surrounding context to make it unique. Read the file first." ) parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "Path relative to the project root."}, "old_string": {"type": "string", "description": "Exact text to replace."}, "new_string": {"type": "string", "description": "Replacement text."}, "replace_all": {"type": "boolean", "description": "Replace every occurrence (default false)."}, }, "required": ["path", "old_string", "new_string"], } def describe(self, args: dict[str, Any]) -> str: return f"Editing {args.get('path', '?')}" def run(self, args: dict[str, Any], ctx: ToolContext) -> ToolResult: path = ctx.permissions.paths.resolve(args["path"], for_write=True) rel = ctx.permissions.paths.relative(path) old, new = args["old_string"], args["new_string"] if old == new: raise ToolError("old_string and new_string are identical; nothing to change") if old == "": raise ToolError("old_string must not be empty; use write_file to create files") before = _read_text(path) count = before.count(old) if count == 0: hint = "" stripped = old.strip() if stripped and stripped in before: hint = " (a whitespace-trimmed version exists - check indentation and line endings)" raise ToolError(f"old_string not found in {rel}{hint}. Re-read the file and try again.") if count > 1 and not args.get("replace_all"): raise ToolError( f"old_string occurs {count} times in {rel}. Add more surrounding context " "to make it unique, or set replace_all to true." ) if not ctx.permissions.check_write(path): raise PermissionDenied(f"user declined editing {rel}") after = before.replace(old, new) if args.get("replace_all") else before.replace(old, new, 1) path.write_text(after, encoding="utf-8") ctx.changed_files.add(rel) diff = make_diff(before, after, rel) added, removed = _diff_stats(diff) replaced = count if args.get("replace_all") else 1 return ToolResult( content=f"Edited {rel}: replaced {replaced} occurrence(s).", summary=f"Edited {rel} (+{added} -{removed})", display={"diff": diff, "path": rel}, ) class ListDirectoryTool(Tool): name = "list_directory" description = ( "List files and folders as a tree (ignoring node_modules, .git, build outputs, ...). " "Use depth to control recursion (default 2)." ) parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "Directory relative to the project root (default '.')."}, "depth": {"type": "integer", "description": "Recursion depth, 1-6 (default 2)."}, }, "required": [], } MAX_ENTRIES = 500 def describe(self, args: dict[str, Any]) -> str: return f"Listing {args.get('path') or '.'}" def run(self, args: dict[str, Any], ctx: ToolContext) -> ToolResult: root = ctx.permissions.paths.resolve(args.get("path") or ".") if not root.exists(): raise ToolError(f"Directory not found: {args.get('path')}") if not root.is_dir(): raise ToolError(f"{args.get('path')} is a file; use read_file") depth = min(max(int(args.get("depth") or 2), 1), 6) ignore = set(ctx.ignore_dirs) lines: list[str] = [] count = 0 truncated = False def walk(directory: Path, level: int, prefix: str) -> None: nonlocal count, truncated try: entries = sorted(directory.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())) except OSError as exc: lines.append(f"{prefix}[unreadable: {exc.strerror}]") return for entry in entries: if entry.name in ignore: continue if count >= self.MAX_ENTRIES: truncated = True return count += 1 if entry.is_dir() and not entry.is_symlink(): lines.append(f"{prefix}{entry.name}/") if level < depth: walk(entry, level + 1, prefix + " ") else: lines.append(f"{prefix}{entry.name}") walk(root, 1, "") rel = ctx.permissions.paths.relative(root) body = "\n".join(lines) if lines else "[empty directory]" if truncated: body += f"\n... [truncated at {self.MAX_ENTRIES} entries; list a subdirectory]" return ToolResult(content=f"{rel}/\n{body}", summary=f"Listed {rel} ({count} entries)") def iter_files(root: Path, ignore_dirs: tuple[str, ...]): """Walk files under ``root`` skipping ignored directories.""" ignore = set(ignore_dirs) for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = sorted(d for d in dirnames if d not in ignore) for filename in sorted(filenames): yield Path(dirpath) / filename