adding `edit` Action to File_System, allows search & replace for editing files and creating code diffs

Modules/File_System.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import difflib
 import json
 import os
+import re
 import shutil
 from typing import Annotated, Optional
 
@@ -20,7 +22,7 @@ from ._core import (
 
 TOOL_SUMMARY = (
     "Browse, search, and manage files within a safe root. "
-    "Actions: list, read, write, append, mkdir, move, copy, delete, info, search, help. "
+    "Actions: list, read, write, append, edit, mkdir, move, copy, delete, info, search, help. "
     "Fill other fields as needed. "
     "Use paths like `/` or `/notes/todo.txt` because all paths are relative to the root (`/`). "
     "Use 'help' to see action-specific required fields and examples."
@@ -36,6 +38,7 @@ HELP_TEXT = (
     "- read: path (e.g., /notes/todo.txt), offset=0, max_chars=4000 (shows next_cursor when truncated)\n"
     "- write: path, content (UTF-8), create_dirs=true\n"
     "- append: path, content (UTF-8), create_dirs=true\n"
+    "- edit: path, content (SEARCH/REPLACE blocks, see format below)\n"
     "- mkdir: path (directory), exist_ok=true\n"
     "- move: path (src), dest_path (dst), overwrite=false\n"
     "- copy: path (src), dest_path (dst), overwrite=false\n"
@@ -43,12 +46,22 @@ HELP_TEXT = (
     "- info: path\n"
     "- search: path (dir or file), content=query text, recursive=false, show_hidden=false, max_entries=20, case_sensitive=false, offset=0\n"
     "- help: show this guide\n\n"
+    "Edit format (SEARCH/REPLACE blocks):\n"
+    "<<<<<<< SEARCH\n"
+    "[exact content to find]\n"
+    "=======\n"
+    "[new content to replace with]\n"
+    ">>>>>>> REPLACE\n\n"
+    "Multiple blocks can be included; each is applied in order. "
+    "Search text must match exactly (whitespace, indentation). "
+    "Only the first occurrence of each search text is replaced.\n\n"
     "Errors are returned as JSON with fields: {status:'error', code, message, path?, hint?, data?}.\n\n"
     "Examples:\n"
     "- list current: action=list, path='/'\n"
     "- make folder: action=mkdir, path='/notes'\n"
     "- write file: action=write, path='/notes/todo.txt', content='hello'\n"
     "- read file: action=read, path='/notes/todo.txt', max_chars=200\n"
+    "- edit file: action=edit, path='/notes/todo.txt', content='<<<<<<< SEARCH\\nhello\\n=======\\nhi\\n>>>>>>> REPLACE'\n"
     "- move file: action=move, path='/notes/todo.txt', dest_path='/notes/todo-old.txt', overwrite=true\n"
     "- delete dir: action=delete, path='/notes', recursive=true\n"
     "- search text: action=search, path='/notes', content='TODO', recursive=true, max_entries=50\n"
@@ -156,6 +169,277 @@ def _delete(abs_path: str, *, recursive: bool) -> str:
     except Exception as exc:
         return _err("delete_failed", "Failed to delete path.", path=_display_path(abs_path), data={"error": _safe_err(exc)})
 
+
+# ---------------------------------------------------------------------------
+# Edit (search/replace) operation
+# ---------------------------------------------------------------------------
+
+# Regex to parse SEARCH/REPLACE blocks (5+ markers required)
+SEARCH_REPLACE_BLOCK_RE = re.compile(
+    r"<{5,} SEARCH\r?\n(.*?)\r?\n?={5,}\r?\n(.*?)\r?\n?>{5,} REPLACE", flags=re.DOTALL
+)
+
+
+def _parse_search_replace_blocks(content: str) -> list[tuple[str, str]]:
+    """Parse SEARCH/REPLACE blocks from content. Returns list of (search, replace) tuples."""
+    matches = SEARCH_REPLACE_BLOCK_RE.findall(content)
+    return [
+        (search.rstrip("\r\n"), replace.rstrip("\r\n"))
+        for search, replace in matches
+    ]
+
+
+def _find_search_context(content: str, search_text: str, max_context: int = 5) -> str:
+    """Find where parts of the search text appear in the content for debugging."""
+    lines = content.split("\n")
+    search_lines = search_text.split("\n")
+
+    if not search_lines:
+        return "Search text is empty"
+
+    first_search_line = search_lines[0].strip()
+    if not first_search_line:
+        return "First line of search text is empty or whitespace only"
+
+    matches = []
+    for i, line in enumerate(lines):
+        if first_search_line in line:
+            matches.append(i)
+
+    if not matches:
+        return f"First search line '{first_search_line[:60]}...' not found anywhere in file"
+
+    context_lines = []
+    for match_idx in matches[:3]:
+        start = max(0, match_idx - max_context)
+        end = min(len(lines), match_idx + max_context + 1)
+
+        context_lines.append(f"\nPotential match area around line {match_idx + 1}:")
+        for i in range(start, end):
+            marker = ">>>" if i == match_idx else "   "
+            context_lines.append(f"{marker} {i + 1:3d}: {lines[i]}")
+
+    return "\n".join(context_lines)
+
+
+def _find_best_fuzzy_match(
+    content: str, search_text: str, threshold: float = 0.9
+) -> tuple[float, int, int, str] | None:
+    """
+    Find the best fuzzy match for search_text within content.
+    Returns (similarity, start_line, end_line, matched_text) or None if no match above threshold.
+    """
+    content_lines = content.split("\n")
+    search_lines = search_text.split("\n")
+    window_size = len(search_lines)
+
+    if window_size == 0:
+        return None
+
+    non_empty_search = [line for line in search_lines if line.strip()]
+    if not non_empty_search:
+        return None
+
+    first_anchor = non_empty_search[0]
+    last_anchor = non_empty_search[-1] if len(non_empty_search) > 1 else first_anchor
+
+    # Find candidate starting positions near anchor lines
+    candidate_starts = set()
+    spread = 5
+
+    for i, line in enumerate(content_lines):
+        if first_anchor in line or last_anchor in line:
+            start_min = max(0, i - spread)
+            start_max = min(len(content_lines) - window_size + 1, i + spread + 1)
+            for s in range(start_min, start_max):
+                candidate_starts.add(s)
+
+    # Fall back to first N positions if no anchors found
+    if not candidate_starts:
+        max_positions = min(len(content_lines) - window_size + 1, 100)
+        candidate_starts = set(range(0, max(0, max_positions)))
+
+    best_match = None
+    best_similarity = 0.0
+
+    for start in candidate_starts:
+        end = start + window_size
+        window_text = "\n".join(content_lines[start:end])
+
+        matcher = difflib.SequenceMatcher(None, search_text, window_text)
+        similarity = matcher.ratio()
+
+        if similarity >= threshold and similarity > best_similarity:
+            best_similarity = similarity
+            best_match = (similarity, start + 1, end, window_text)  # 1-based line numbers
+
+    return best_match
+
+
+def _create_unified_diff(text1: str, text2: str, label1: str = "SEARCH", label2: str = "CLOSEST MATCH") -> str:
+    """Create a unified diff between two texts."""
+    lines1 = text1.splitlines(keepends=True)
+    lines2 = text2.splitlines(keepends=True)
+
+    lines1 = [line if line.endswith("\n") else line + "\n" for line in lines1]
+    lines2 = [line if line.endswith("\n") else line + "\n" for line in lines2]
+
+    diff = difflib.unified_diff(lines1, lines2, fromfile=label1, tofile=label2, lineterm="", n=3)
+    diff_lines = list(diff)
+
+    if diff_lines and not diff_lines[0].startswith("==="):
+        diff_lines.insert(2, "=" * 67 + "\n")
+
+    result = "".join(diff_lines)
+
+    # Truncate long diffs
+    max_chars = 2000
+    if len(result) > max_chars:
+        result = result[:max_chars] + "\n...(diff truncated)"
+
+    return result.rstrip()
+
+
+def _edit_file(abs_path: str, content: str) -> str:
+    """
+    Apply SEARCH/REPLACE blocks to a file.
+    
+    Returns a success message with stats, or an error message with debugging context.
+    """
+    display = _display_path(abs_path)
+    
+    # Validate file exists
+    if not os.path.exists(abs_path):
+        return _err("file_not_found", f"File not found: {display}", path=display)
+    if os.path.isdir(abs_path):
+        return _err("is_directory", f"Path is a directory, not a file: {display}", path=display, hint="Provide a file path.")
+
+    # Validate content has blocks
+    content = (content or "").strip()
+    if not content:
+        return _err(
+            "empty_content",
+            "No content provided for edit action.",
+            hint="Provide SEARCH/REPLACE blocks in the Content field."
+        )
+
+    blocks = _parse_search_replace_blocks(content)
+    if not blocks:
+        return _err(
+            "no_blocks_found",
+            "No valid SEARCH/REPLACE blocks found in content.",
+            hint=(
+                "Expected format:\n"
+                "<<<<<<< SEARCH\n"
+                "[exact content to find]\n"
+                "=======\n"
+                "[new content to replace with]\n"
+                ">>>>>>> REPLACE"
+            )
+        )
+
+    # Read file
+    try:
+        with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
+            file_content = f.read()
+    except Exception as exc:
+        return _err("read_failed", "Failed to read file.", path=display, data={"error": _safe_err(exc)})
+
+    # Apply blocks
+    current_content = file_content
+    applied = 0
+    errors: list[str] = []
+    warnings: list[str] = []
+
+    for i, (search, replace) in enumerate(blocks, 1):
+        if search not in current_content:
+            # Search text not found - provide helpful debugging context
+            context = _find_search_context(current_content, search)
+            
+            error_msg = (
+                f"Block {i} failed: Search text not found in {display}\n"
+                f"Search text was:\n{search!r}\n"
+                f"Context analysis:\n{context}"
+            )
+            
+            # Try fuzzy matching
+            fuzzy = _find_best_fuzzy_match(current_content, search, 0.9)
+            if fuzzy:
+                similarity, start_line, end_line, matched_text = fuzzy
+                diff = _create_unified_diff(search, matched_text)
+                error_msg += (
+                    f"\n\nClosest fuzzy match (similarity {similarity * 100:.1f}%) "
+                    f"at lines {start_line}–{end_line}:\n```diff\n{diff}\n```"
+                )
+            
+            error_msg += (
+                "\n\nDebugging tips:\n"
+                "1. Check for exact whitespace/indentation match\n"
+                "2. Verify line endings match the file exactly (\\r\\n vs \\n)\n"
+                "3. Ensure the search text hasn't been modified by previous blocks\n"
+                "4. Check for typos or case sensitivity issues"
+            )
+            
+            errors.append(error_msg)
+            continue
+
+        # Check for multiple occurrences
+        occurrences = current_content.count(search)
+        if occurrences > 1:
+            warnings.append(
+                f"Block {i}: Search text appears {occurrences} times. "
+                f"Only the first occurrence will be replaced."
+            )
+
+        # Apply replacement (first occurrence only)
+        current_content = current_content.replace(search, replace, 1)
+        applied += 1
+
+    # If all blocks failed, return error
+    if errors and applied == 0:
+        error_message = "All SEARCH/REPLACE blocks failed:\n\n" + "\n\n---\n\n".join(errors)
+        if warnings:
+            error_message += "\n\nWarnings:\n" + "\n".join(warnings)
+        return _err("edit_failed", error_message)
+
+    # If some blocks failed, still try to apply successful ones but report errors
+    if errors:
+        partial_msg = f"Partial edit: {applied} of {len(blocks)} blocks applied.\n\n"
+        partial_msg += "Failed blocks:\n\n" + "\n\n---\n\n".join(errors)
+        if warnings:
+            partial_msg += "\n\nWarnings:\n" + "\n".join(warnings)
+        
+        # Write partial changes
+        try:
+            with open(abs_path, "w", encoding="utf-8") as f:
+                f.write(current_content)
+        except Exception as exc:
+            return _err("write_failed", "Failed to write file after partial edit.", path=display, data={"error": _safe_err(exc)})
+        
+        return partial_msg
+
+    # All blocks succeeded - write file
+    try:
+        with open(abs_path, "w", encoding="utf-8") as f:
+            f.write(current_content)
+    except Exception as exc:
+        return _err("write_failed", "Failed to write file.", path=display, data={"error": _safe_err(exc)})
+
+    # Build success message
+    original_lines = len(file_content.splitlines())
+    new_lines = len(current_content.splitlines())
+    line_delta = new_lines - original_lines
+    delta_str = f"+{line_delta}" if line_delta > 0 else str(line_delta)
+
+    result = f"Edited {display}: {applied} block{'s' if applied != 1 else ''} applied"
+    if line_delta != 0:
+        result += f" ({delta_str} lines)"
+    
+    if warnings:
+        result += "\n\nWarnings:\n" + "\n".join(warnings)
+    
+    return result
+
 # ---------------------------------------------------------------------------
 # Read-only operations delegated to sandbox
 # ---------------------------------------------------------------------------
@@ -200,7 +484,7 @@ def _info(abs_path: str) -> str:
 
 @autodoc(summary=TOOL_SUMMARY)
 def File_System(
-    action: Annotated[str, "Operation to perform: 'list', 'read', 'write', 'append', 'mkdir', 'move', 'copy', 'delete', 'info', 'search'."],
+    action: Annotated[str, "Operation to perform: 'list', 'read', 'write', 'append', 'edit', 'mkdir', 'move', 'copy', 'delete', 'info', 'search'."],
     path: Annotated[str, "Target path, relative to root unless UNSAFE_ALLOW_ABS_PATHS=1."] = "/",
     content: Annotated[Optional[str], "Content for write/append actions or search query (UTF-8)."] = None,
     dest_path: Annotated[Optional[str], "Destination for move/copy (relative to root unless unsafe absolute allowed)."] = None,
@@ -228,11 +512,11 @@ def File_System(
         case_sensitive=case_sensitive,
     )
     action = (action or "").strip().lower()
-    if action not in {"list", "read", "write", "append", "mkdir", "move", "copy", "delete", "info", "search", "help"}:
+    if action not in {"list", "read", "write", "append", "edit", "mkdir", "move", "copy", "delete", "info", "search", "help"}:
         result = _err(
             "invalid_action",
             "Invalid action.",
-            hint="Choose from: list, read, write, append, mkdir, move, copy, delete, info, search, help."
+            hint="Choose from: list, read, write, append, edit, mkdir, move, copy, delete, info, search, help."
         )
         _log_call_end("File_System", _truncate_for_log(result))
         return result
@@ -263,6 +547,8 @@ def File_System(
                 )
             else:
                 result = _write_file(abs_path, content or "", append=(action == "append"), create_dirs=create_dirs)
+        elif action == "edit":
+            result = _edit_file(abs_path, content or "")
         elif action == "mkdir":
             result = _mkdir(abs_path, exist_ok=True)
         elif action in {"move", "copy"}:
@@ -309,12 +595,12 @@ def build_interface() -> gr.Interface:
         inputs=[
             gr.Radio(
                 label="Action",
-                choices=["list", "read", "write", "append", "mkdir", "move", "copy", "delete", "info", "search", "help"],
+                choices=["list", "read", "write", "append", "edit", "mkdir", "move", "copy", "delete", "info", "search", "help"],
                 value="help",
                 info="Operation to perform",
             ),
             gr.Textbox(label="Path", placeholder="/ or /src/file.txt", max_lines=1, value="/", info="Target path (relative to root)"),
-            gr.Textbox(label="Content", lines=6, placeholder="Text to write or search for...", info="Content for write/append actions or search query"),
+            gr.Textbox(label="Content", lines=6, placeholder="Text to write, SEARCH/REPLACE blocks for edit, or search query...", value="<<<<<<< SEARCH\n[exact text to find in the file]\n=======\n[exact text to replace it with]\n>>>>>>> REPLACE", info="Content for write/append/edit actions or search query"),
             gr.Textbox(label="Destination", max_lines=1, info="Destination path (Move/Copy only)"),
             gr.Checkbox(label="Recursive", value=False, info="Recurse into subfolders (List/Delete/Search)"),
             gr.Checkbox(label="Show hidden", value=False, info="Include hidden files (List/Search)"),