| 1 | import mimetypes |
| 2 | import os |
| 3 | |
| 4 | from helpers.api import ApiHandler, Input, Output, Request |
| 5 | from helpers.file_browser import FileBrowser |
| 6 | from helpers import runtime, files, extension |
| 7 | |
| 8 | MAX_EDIT_FILE_SIZE = 1024 * 1024 |
| 9 | BINARY_SAMPLE_SIZE = 10 * 1024 |
| 10 | |
| 11 | |
| 12 | class EditWorkDirFile(ApiHandler): |
| 13 | @classmethod |
| 14 | def get_methods(cls): |
| 15 | return ["GET", "POST"] |
| 16 | |
| 17 | def _extract_error_message(self, error_str: str) -> str: |
| 18 | """Extract user-friendly error message from exception string.""" |
| 19 | for line in reversed(error_str.split('\n')): |
| 20 | if ': ' in line and ('Exception' in line or 'Error' in line): |
| 21 | return line.split(': ', 1)[1].strip() |
| 22 | return error_str.strip() |
| 23 | |
| 24 | async def process(self, input: Input, request: Request) -> Output: |
| 25 | try: |
| 26 | if request.method == "GET": |
| 27 | file_path = request.args.get("path", "") |
| 28 | if not file_path: |
| 29 | return {"error": "Path is required"} |
| 30 | if not file_path.startswith("/"): |
| 31 | file_path = f"/{file_path}" |
| 32 | |
| 33 | data = await runtime.call_development_function(load_file, file_path) |
| 34 | return {"data": data} |
| 35 | |
| 36 | file_path = input.get("path", "") |
| 37 | if not file_path: |
| 38 | return {"error": "Path is required"} |
| 39 | if not file_path.startswith("/"): |
| 40 | file_path = f"/{file_path}" |
| 41 | |
| 42 | content = input.get("content", "") |
| 43 | if not isinstance(content, str): |
| 44 | return {"error": "Content must be a string"} |
| 45 | |
| 46 | content_size = len(content.encode("utf-8")) |
| 47 | if content_size > MAX_EDIT_FILE_SIZE: |
| 48 | return {"error": "File exceeds 1 MB and cannot be edited"} |
| 49 | |
| 50 | res = await runtime.call_development_function(save_file, file_path, content) |
| 51 | if not res: |
| 52 | return {"error": "Failed to save file"} |
| 53 | |
| 54 | await extension.call_extensions_async( |
| 55 | "workdir_file_mutation_after", |
| 56 | agent=None, |
| 57 | data={ |
| 58 | "action": "edit", |
| 59 | "path": file_path, |
| 60 | "paths": [file_path], |
| 61 | }, |
| 62 | ) |
| 63 | return {"ok": True} |
| 64 | except Exception as e: |
| 65 | # Extract clean error message from exception |
| 66 | # RPC calls may return full tracebacks in exception strings |
| 67 | return {"error": self._extract_error_message(str(e))} |
| 68 | |
| 69 | |
| 70 | async def load_file(file_path: str) -> dict: |
| 71 | browser = FileBrowser() |
| 72 | full_path = browser.get_full_path(file_path) |
| 73 | |
| 74 | if os.path.isdir(full_path): |
| 75 | raise Exception("Path points to a directory") |
| 76 | |
| 77 | size = os.path.getsize(full_path) |
| 78 | if size > MAX_EDIT_FILE_SIZE: |
| 79 | raise Exception("File exceeds 1 MB and cannot be edited") |
| 80 | |
| 81 | # Binary detection: only sample the first ~10KB (per backend rules) |
| 82 | if files.is_probably_binary_file(full_path, sample_size=BINARY_SAMPLE_SIZE): |
| 83 | raise Exception("Binary file detected; editing is not supported") |
| 84 | |
| 85 | mime_type, _ = mimetypes.guess_type(full_path) |
| 86 | try: |
| 87 | with open(full_path, "r", encoding="utf-8", errors="strict") as file: |
| 88 | content = file.read() |
| 89 | except UnicodeDecodeError: |
| 90 | raise Exception("Unable to decode file as UTF-8; editing is not supported") |
| 91 | |
| 92 | return { |
| 93 | "path": file_path, |
| 94 | "name": os.path.basename(full_path), |
| 95 | "mime_type": mime_type or "text/plain", |
| 96 | "content": content, |
| 97 | } |
| 98 | |
| 99 | |
| 100 | def save_file(file_path: str, content: str) -> bool: |
| 101 | browser = FileBrowser() |
| 102 | return browser.save_text_file(file_path, content) |