From 56ac8d17d7d562f47b2acb229d43b240289b97cc Mon Sep 17 00:00:00 2001 From: worldmozara Date: Sat, 1 Aug 2026 01:21:00 +0800 Subject: [PATCH] core/tools: restrict edit_lineno to lines read with line numbers --- src/plyngent/tools/file/edit_lineno.py | 34 +++++++++++++++++++++----- tests/test_tools/test_edit_lineno.py | 30 +++++++++++++++++++++++ 2 files changed, 58 insertions(+), 6 deletions(-) diff --git a/src/plyngent/tools/file/edit_lineno.py b/src/plyngent/tools/file/edit_lineno.py index 9738d88..95958a2 100644 --- a/src/plyngent/tools/file/edit_lineno.py +++ b/src/plyngent/tools/file/edit_lineno.py @@ -2,12 +2,15 @@ from __future__ import annotations from typing import TYPE_CHECKING -from plyngent.agent import ToolTag, tool, was_lineno_read +from plyngent.agent import ToolTag, lineno_read_lines, tool from plyngent.tools.workspace import resolve_path if TYPE_CHECKING: from pathlib import Path +# How many unread line numbers to preview in the error message. +_UNREAD_PREVIEW = 8 + def _detect_newline(lines: list[str]) -> str: for sample in lines: @@ -48,6 +51,22 @@ def _validate_range(start_line: int, end_line: int, n: int) -> str | None: return None +def _unread_lines_in_range( + read_lines: set[int], + start_line: int, + end_line: int, + n: int, +) -> list[int]: + """Lines in ``start_line..end_line`` that were not read with line numbers. + + Appending (``start_line == n + 1``) requires the last line to be read, so + line numbers stay truthful; the append point itself is not a line. + """ + if start_line == n + 1: + return [] if n in read_lines else [n] + return sorted(set(range(start_line, end_line + 1)) - read_lines) + + def _append_after(target: Path, path: str, text: str, lines: list[str], new_content: str) -> str: n = len(lines) newline = _detect_newline(lines[-1:] if lines else []) @@ -85,11 +104,6 @@ async def edit_lineno(path: str, start_line: int, end_line: int, new_content: st target = resolve_path(path) if not target.is_file(): return f"error: not a file: {path}" - if not was_lineno_read(str(target)): - return ( - f"error: read {path!r} first with read_file(path, with_lineno=true) " - "to get correct line numbers" - ) text = target.read_text(encoding="utf-8", errors="replace") lines = text.splitlines(keepends=True) @@ -97,6 +111,14 @@ async def edit_lineno(path: str, start_line: int, end_line: int, new_content: st if err is not None: return err + unread = _unread_lines_in_range(lineno_read_lines(str(target)), start_line, end_line, len(lines)) + if unread: + preview = unread[:_UNREAD_PREVIEW] + head = ", ".join(str(i) for i in preview) + if len(unread) > _UNREAD_PREVIEW: + head += f", … ({len(unread)} total)" + return f"error: lines {head} not read; call read_file(path, with_lineno=true) first to see current line numbers" + if start_line == len(lines) + 1: return _append_after(target, path, text, lines, new_content) return _replace_range(target, path, lines, start_line, end_line, new_content) diff --git a/tests/test_tools/test_edit_lineno.py b/tests/test_tools/test_edit_lineno.py index e13262d..f9bbd79 100644 --- a/tests/test_tools/test_edit_lineno.py +++ b/tests/test_tools/test_edit_lineno.py @@ -22,6 +22,36 @@ def test_edit_lineno_replace_middle(workspace: object) -> None: assert call_sync(read_file, "a.txt") == "one\nTWO\nTHREE\nfour\n" +def test_edit_lineno_requires_all_lines_read(workspace: object) -> None: + del workspace + _ = call_sync(write_file, "r.txt", "a\nb\nc\nd\ne\n") + _ = reset_lineno_tracker() + # Only lines 2-4 are read; editing line 1 or 5 must be rejected. + _ = call_sync(read_file, "r.txt", offset=1, limit=3, with_lineno=True) + out = call_sync(edit_lineno, "r.txt", 1, 2, "X\n") + assert "not read" in out and "1" in out + out = call_sync(edit_lineno, "r.txt", 4, 5, "Y\n") + assert "not read" in out and "5" in out + # Fully read range still edits. + out = call_sync(edit_lineno, "r.txt", 2, 3, "B\nC\n") + assert "replaced lines 2-3" in out + + +def test_edit_lineno_append_requires_last_line_read(workspace: object) -> None: + del workspace + _ = call_sync(write_file, "ap.txt", "a\nb\nc\n") + _ = reset_lineno_tracker() + # Only lines 1-2 read; appending after line 3 requires line 3 read. + _ = call_sync(read_file, "ap.txt", offset=0, limit=2, with_lineno=True) + out = call_sync(edit_lineno, "ap.txt", 4, 4, "d\n") + assert "not read" in out and "3" in out + # Read the last line, then append succeeds. + _ = call_sync(read_file, "ap.txt", offset=2, limit=1, with_lineno=True) + out = call_sync(edit_lineno, "ap.txt", 4, 4, "d\n") + assert "appended" in out + assert call_sync(read_file, "ap.txt") == "a\nb\nc\nd\n" + + def test_edit_lineno_delete_range(workspace: object) -> None: del workspace _ = call_sync(write_file, "b.txt", "a\nb\nc\n")