mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 23:47:54 +08:00
core/tools: restrict edit_lineno to lines read with line numbers
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user