| 1 | """ |
| 2 | Pure file operations for the text_editor plugin. |
| 3 | |
| 4 | No agent/tool dependencies — only stdlib + tokens helper. |
| 5 | """ |
| 6 | |
| 7 | import os |
| 8 | import shutil |
| 9 | import tempfile |
| 10 | from typing import TypedDict |
| 11 | |
| 12 | from helpers import tokens |
| 13 | from plugins._text_editor.helpers.context_patch import ( |
| 14 | apply_context_patch_with_metadata, |
| 15 | ) |
| 16 | |
| 17 | _BINARY_PEEK = 8192 |
| 18 | |
| 19 | |
| 20 | # ------------------------------------------------------------------ |
| 21 | # Binary detection |
| 22 | # ------------------------------------------------------------------ |
| 23 | |
| 24 | def is_binary(path: str) -> bool: |
| 25 | """Detect binary file by checking for null bytes.""" |
| 26 | try: |
| 27 | with open(path, "rb") as f: |
| 28 | chunk = f.read(_BINARY_PEEK) |
| 29 | return b"\x00" in chunk |
| 30 | except OSError: |
| 31 | return False |
| 32 | |
| 33 | |
| 34 | # ------------------------------------------------------------------ |
| 35 | # File metadata |
| 36 | # ------------------------------------------------------------------ |
| 37 | |
| 38 | class FileInfo(TypedDict): |
| 39 | exists: bool |
| 40 | is_file: bool |
| 41 | realpath: str |
| 42 | expanded: str |
| 43 | mtime: float | None |
| 44 | |
| 45 | |
| 46 | def file_info(path: str) -> FileInfo: |
| 47 | """Return file metadata for mtime tracking and path resolution.""" |
| 48 | path = os.path.expanduser(path) |
| 49 | rp = os.path.realpath(path) |
| 50 | exists = os.path.exists(path) |
| 51 | is_file = os.path.isfile(path) |
| 52 | mtime = None |
| 53 | if exists: |
| 54 | try: |
| 55 | mtime = os.path.getmtime(path) |
| 56 | except OSError: |
| 57 | pass |
| 58 | return FileInfo( |
| 59 | exists=exists, |
| 60 | is_file=is_file, |
| 61 | realpath=rp, |
| 62 | expanded=path, |
| 63 | mtime=mtime, |
| 64 | ) |
| 65 | |
| 66 | |
| 67 | # ------------------------------------------------------------------ |
| 68 | # Read |
| 69 | # ------------------------------------------------------------------ |
| 70 | |
| 71 | class ReadResult(TypedDict): |
| 72 | content: str |
| 73 | total_lines: int |
| 74 | warnings: str |
| 75 | error: str |
| 76 | |
| 77 | |
| 78 | def read_file( |
| 79 | path: str, |
| 80 | line_from: int = 1, |
| 81 | line_to: int | None = None, |
| 82 | max_line_tokens: int = 500, |
| 83 | default_line_count: int = 100, |
| 84 | max_total_read_tokens: int = 4000, |
| 85 | ) -> ReadResult: |
| 86 | """ |
| 87 | Read a text file and return numbered lines with token budgeting. |
| 88 | |
| 89 | Line numbers are 1-based (matching grep, sed, editors). |
| 90 | line_from and line_to are both inclusive. |
| 91 | None line_to defaults to line_from + default_line_count - 1. |
| 92 | """ |
| 93 | path = os.path.expanduser(path) |
| 94 | |
| 95 | if not os.path.isfile(path): |
| 96 | return ReadResult( |
| 97 | content="", total_lines=0, warnings="", |
| 98 | error="file not found", |
| 99 | ) |
| 100 | |
| 101 | if is_binary(path): |
| 102 | return ReadResult( |
| 103 | content="", total_lines=0, warnings="", |
| 104 | error="file appears binary, use terminal instead", |
| 105 | ) |
| 106 | |
| 107 | try: |
| 108 | with open(path, "r", encoding="utf-8", errors="replace") as f: |
| 109 | all_lines = f.readlines() |
| 110 | except OSError as exc: |
| 111 | return ReadResult( |
| 112 | content="", total_lines=0, warnings="", |
| 113 | error=str(exc), |
| 114 | ) |
| 115 | |
| 116 | total_lines = len(all_lines) |
| 117 | line_from = max(line_from, 1) |
| 118 | if line_to is None: |
| 119 | line_to = line_from + default_line_count - 1 |
| 120 | line_to = min(line_to, total_lines) |
| 121 | |
| 122 | # Convert 1-based inclusive range to 0-based slice |
| 123 | idx_from = line_from - 1 |
| 124 | idx_to = line_to # slice is exclusive, line_to is inclusive 1-based |
| 125 | selected = all_lines[idx_from:idx_to] |
| 126 | num_width = len(str(line_to)) |
| 127 | |
| 128 | warn_parts: list[str] = [] |
| 129 | cropped_lines: list[int] = [] |
| 130 | output_lines: list[str] = [] |
| 131 | running_tokens = 0 |
| 132 | trimmed_by_total = False |
| 133 | |
| 134 | for i, raw_line in enumerate(selected): |
| 135 | line_no = line_from + i # 1-based |
| 136 | stripped = raw_line.rstrip("\n").rstrip("\r") |
| 137 | line_tok = tokens.count_tokens(stripped) |
| 138 | |
| 139 | if line_tok > max_line_tokens: |
| 140 | chars_per_tok = max(len(stripped) / line_tok, 1) |
| 141 | keep_chars = int(max_line_tokens * chars_per_tok * tokens.TRIM_BUFFER) |
| 142 | stripped = stripped[:keep_chars] + "..." |
| 143 | cropped_lines.append(line_no) |
| 144 | line_tok = max_line_tokens |
| 145 | |
| 146 | if running_tokens + line_tok > max_total_read_tokens: |
| 147 | trimmed_by_total = True |
| 148 | break |
| 149 | |
| 150 | running_tokens += line_tok |
| 151 | output_lines.append(f"{line_no:>{num_width}} {stripped}") |
| 152 | |
| 153 | if cropped_lines: |
| 154 | nums = " ".join(str(n) for n in cropped_lines) |
| 155 | warn_parts.append( |
| 156 | f"long lines {nums} cropped - use terminal for precise manipulation" |
| 157 | ) |
| 158 | if trimmed_by_total: |
| 159 | actual_end = line_from + len(output_lines) |
| 160 | warn_parts.append( |
| 161 | f"output trimmed at line {actual_end} due to token limit" |
| 162 | " - use line_from/line_to for remaining" |
| 163 | ) |
| 164 | |
| 165 | warn_str = "" |
| 166 | if warn_parts: |
| 167 | warn_str = "\nwarning: " + "; ".join(warn_parts) |
| 168 | |
| 169 | return ReadResult( |
| 170 | content="\n".join(output_lines), |
| 171 | total_lines=total_lines, |
| 172 | warnings=warn_str, |
| 173 | error="", |
| 174 | ) |
| 175 | |
| 176 | |
| 177 | # ------------------------------------------------------------------ |
| 178 | # Write |
| 179 | # ------------------------------------------------------------------ |
| 180 | |
| 181 | class WriteResult(TypedDict): |
| 182 | total_lines: int |
| 183 | error: str |
| 184 | |
| 185 | |
| 186 | def write_file(path: str, content: str | None) -> WriteResult: |
| 187 | """Create or overwrite a file.""" |
| 188 | if content is None: |
| 189 | content = "" |
| 190 | path = os.path.expanduser(path) |
| 191 | try: |
| 192 | os.makedirs(os.path.dirname(path) or ".", exist_ok=True) |
| 193 | with open(path, "w", encoding="utf-8") as f: |
| 194 | f.write(content) |
| 195 | except OSError as exc: |
| 196 | return WriteResult(total_lines=0, error=str(exc)) |
| 197 | |
| 198 | total = content.count("\n") + ( |
| 199 | 1 if content and not content.endswith("\n") else 0 |
| 200 | ) |
| 201 | return WriteResult(total_lines=total, error="") |
| 202 | |
| 203 | |
| 204 | # ------------------------------------------------------------------ |
| 205 | # Patch |
| 206 | # ------------------------------------------------------------------ |
| 207 | |
| 208 | class PatchResult(TypedDict): |
| 209 | total_lines: int |
| 210 | edit_count: int |
| 211 | error: str |
| 212 | |
| 213 | |
| 214 | class ContextPatchFileResult(TypedDict): |
| 215 | total_lines: int |
| 216 | hunk_count: int |
| 217 | line_from: int |
| 218 | line_to: int |
| 219 | |
| 220 | |
| 221 | class ExactReplaceFileResult(TypedDict): |
| 222 | total_lines: int |
| 223 | replacement_count: int |
| 224 | line_from: int |
| 225 | line_to: int |
| 226 | |
| 227 | |
| 228 | def validate_edits(edits: list | None) -> tuple[list[dict], str]: |
| 229 | """ |
| 230 | Normalise and validate an edits array. |
| 231 | |
| 232 | Line numbers are 1-based (matching grep, sed, editors). |
| 233 | Semantics (to is inclusive): |
| 234 | {from:2, to:2, content:"x\\n"} - replace line 2 |
| 235 | {from:1, to:3, content:"x\\n"} - replace lines 1-3 |
| 236 | {from:2, to:2} - delete line 2 |
| 237 | {from:5} or {from:5, to:-1} - insert before line 5 (no deletion) |
| 238 | |
| 239 | Returns (parsed_edits, error_string). error_string is empty on success. |
| 240 | """ |
| 241 | if not edits or not isinstance(edits, list): |
| 242 | return [], "edits array is required" |
| 243 | |
| 244 | parsed: list[dict] = [] |
| 245 | for e in edits: |
| 246 | if not isinstance(e, dict): |
| 247 | return [], f"invalid edit entry: {e}" |
| 248 | frm = int(e.get("from", 0)) |
| 249 | if frm < 1: |
| 250 | return [], f"edit missing or invalid from (must be >= 1): {e}" |
| 251 | # to == -1 or absent means pure insert (no lines removed) |
| 252 | to = int(e.get("to", -1)) |
| 253 | is_insert = to < 0 or to < frm |
| 254 | if is_insert: |
| 255 | to = frm - 1 # normalise: marks zero-width range |
| 256 | parsed.append({ |
| 257 | "from": frm, |
| 258 | "to": to, |
| 259 | "content": e.get("content", ""), |
| 260 | "insert": is_insert, |
| 261 | }) |
| 262 | |
| 263 | parsed.sort(key=lambda x: (x["from"], 0 if x["insert"] else 1)) |
| 264 | for i in range(1, len(parsed)): |
| 265 | prev, cur = parsed[i - 1], parsed[i] |
| 266 | # Inserts at the same line don't overlap with each other or |
| 267 | # with a replace that starts at the same line. |
| 268 | if prev["insert"]: |
| 269 | continue |
| 270 | # prev is a replace/delete: its range is [from..to] inclusive |
| 271 | if cur["from"] <= prev["to"]: |
| 272 | return [], ( |
| 273 | f"overlapping edits: edit at {prev['from']}" |
| 274 | f" (to {prev['to']}) and {cur['from']}" |
| 275 | f" (to {cur['to']})" |
| 276 | ) |
| 277 | |
| 278 | return parsed, "" |
| 279 | |
| 280 | |
| 281 | def apply_patch(path: str, edits: list[dict]) -> int: |
| 282 | """ |
| 283 | Apply sorted, validated edits by streaming to a temp file. |
| 284 | |
| 285 | Line numbers are 1-based. Edits use inclusive 'to'. |
| 286 | Inserts have 'insert': True. |
| 287 | Returns total line count after patching. |
| 288 | """ |
| 289 | # Ensure content always ends with newline to prevent line merging |
| 290 | for e in edits: |
| 291 | if e["content"] and not e["content"].endswith("\n"): |
| 292 | e["content"] += "\n" |
| 293 | |
| 294 | dir_name = os.path.dirname(path) or "." |
| 295 | fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp") |
| 296 | try: |
| 297 | with ( |
| 298 | open(path, "r", encoding="utf-8", errors="replace") as src, |
| 299 | os.fdopen(fd, "w", encoding="utf-8") as dst, |
| 300 | ): |
| 301 | edit_idx = 0 |
| 302 | line_no = 1 # 1-based |
| 303 | total_written = 0 |
| 304 | |
| 305 | for raw_line in src: |
| 306 | # Process all inserts targeting this line first |
| 307 | while ( |
| 308 | edit_idx < len(edits) |
| 309 | and edits[edit_idx]["insert"] |
| 310 | and edits[edit_idx]["from"] == line_no |
| 311 | ): |
| 312 | edit = edits[edit_idx] |
| 313 | if edit["content"]: |
| 314 | dst.write(edit["content"]) |
| 315 | total_written += _count_content_lines(edit["content"]) |
| 316 | edit_idx += 1 |
| 317 | |
| 318 | # Check if current line falls in a replace/delete range |
| 319 | if edit_idx < len(edits) and not edits[edit_idx]["insert"]: |
| 320 | edit = edits[edit_idx] |
| 321 | if edit["from"] <= line_no <= edit["to"]: |
| 322 | # Write replacement content once at range start |
| 323 | if line_no == edit["from"] and edit["content"]: |
| 324 | dst.write(edit["content"]) |
| 325 | total_written += _count_content_lines( |
| 326 | edit["content"] |
| 327 | ) |
| 328 | # Skip original line; advance edit at range end |
| 329 | if line_no == edit["to"]: |
| 330 | edit_idx += 1 |
| 331 | line_no += 1 |
| 332 | continue |
| 333 | |
| 334 | dst.write(raw_line) |
| 335 | total_written += 1 |
| 336 | line_no += 1 |
| 337 | |
| 338 | # Remaining edits past end of file |
| 339 | while edit_idx < len(edits): |
| 340 | edit = edits[edit_idx] |
| 341 | if edit["content"]: |
| 342 | dst.write(edit["content"]) |
| 343 | total_written += _count_content_lines(edit["content"]) |
| 344 | edit_idx += 1 |
| 345 | |
| 346 | shutil.move(tmp_path, path) |
| 347 | return total_written |
| 348 | except Exception: |
| 349 | if os.path.exists(tmp_path): |
| 350 | os.unlink(tmp_path) |
| 351 | raise |
| 352 | |
| 353 | |
| 354 | def patch_file(path: str, edits: list | None) -> PatchResult: |
| 355 | """Validate and apply edits to a file.""" |
| 356 | path = os.path.expanduser(path) |
| 357 | if not os.path.isfile(path): |
| 358 | return PatchResult(total_lines=0, edit_count=0, error="file not found") |
| 359 | |
| 360 | parsed, err = validate_edits(edits) |
| 361 | if err: |
| 362 | return PatchResult(total_lines=0, edit_count=0, error=err) |
| 363 | |
| 364 | try: |
| 365 | total = apply_patch(path, parsed) |
| 366 | except Exception as exc: |
| 367 | return PatchResult(total_lines=0, edit_count=0, error=str(exc)) |
| 368 | |
| 369 | return PatchResult(total_lines=total, edit_count=len(parsed), error="") |
| 370 | |
| 371 | |
| 372 | def apply_context_patch_file(path: str, patch_text: str) -> ContextPatchFileResult: |
| 373 | """Apply a context patch to an existing text file.""" |
| 374 | path = os.path.expanduser(path) |
| 375 | if not os.path.isfile(path): |
| 376 | raise FileNotFoundError("file not found") |
| 377 | |
| 378 | with open(path, "r", encoding="utf-8", errors="replace") as src: |
| 379 | content = src.read() |
| 380 | |
| 381 | result = apply_context_patch_with_metadata(content, patch_text) |
| 382 | dir_name = os.path.dirname(path) or "." |
| 383 | fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp") |
| 384 | try: |
| 385 | with os.fdopen(fd, "w", encoding="utf-8") as dst: |
| 386 | dst.write(result.content) |
| 387 | shutil.move(tmp_path, path) |
| 388 | except Exception: |
| 389 | if os.path.exists(tmp_path): |
| 390 | os.unlink(tmp_path) |
| 391 | raise |
| 392 | |
| 393 | return ContextPatchFileResult( |
| 394 | total_lines=_count_content_lines(result.content), |
| 395 | hunk_count=result.hunk_count, |
| 396 | line_from=result.line_from, |
| 397 | line_to=result.line_to, |
| 398 | ) |
| 399 | |
| 400 | |
| 401 | def apply_exact_replace_file( |
| 402 | path: str, old_text: str, new_text: str |
| 403 | ) -> ExactReplaceFileResult: |
| 404 | """Replace exactly one text span in an existing text file.""" |
| 405 | path = os.path.expanduser(path) |
| 406 | if not os.path.isfile(path): |
| 407 | raise FileNotFoundError("file not found") |
| 408 | if not old_text: |
| 409 | raise ValueError("old_text is required for exact replace") |
| 410 | |
| 411 | with open(path, "r", encoding="utf-8", errors="replace") as src: |
| 412 | content = src.read() |
| 413 | |
| 414 | match_count = content.count(old_text) |
| 415 | if match_count == 0: |
| 416 | raise ValueError("old_text not found") |
| 417 | if match_count > 1: |
| 418 | raise ValueError( |
| 419 | f"old_text matched {match_count} times; provide a longer exact span" |
| 420 | ) |
| 421 | if old_text == new_text: |
| 422 | raise ValueError("old_text and new_text are identical") |
| 423 | |
| 424 | start = content.index(old_text) |
| 425 | line_from = content[:start].count("\n") + 1 |
| 426 | line_to = line_from + max(_count_content_lines(old_text) - 1, 0) |
| 427 | new_content = content.replace(old_text, new_text, 1) |
| 428 | |
| 429 | dir_name = os.path.dirname(path) or "." |
| 430 | fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp") |
| 431 | try: |
| 432 | with os.fdopen(fd, "w", encoding="utf-8") as dst: |
| 433 | dst.write(new_content) |
| 434 | shutil.move(tmp_path, path) |
| 435 | except Exception: |
| 436 | if os.path.exists(tmp_path): |
| 437 | os.unlink(tmp_path) |
| 438 | raise |
| 439 | |
| 440 | return ExactReplaceFileResult( |
| 441 | total_lines=_count_content_lines(new_content), |
| 442 | replacement_count=1, |
| 443 | line_from=line_from, |
| 444 | line_to=line_to, |
| 445 | ) |
| 446 | |
| 447 | |
| 448 | # ------------------------------------------------------------------ |
| 449 | # Internal |
| 450 | # ------------------------------------------------------------------ |
| 451 | |
| 452 | def _count_content_lines(content: str) -> int: |
| 453 | return content.count("\n") + ( |
| 454 | 1 if content and not content.endswith("\n") else 0 |
| 455 | ) |