core/tools/file: add edit_lineno line-range edit tool

1-based inclusive range replace/delete/append; document Phase B agent
and tool updates in CLAUDE.md.
This commit is contained in:
2026-07-14 23:24:34 +08:00
parent 3363ab928b
commit 81a30d0e4d
5 changed files with 141 additions and 3 deletions
+4 -3
View File
@@ -54,16 +54,17 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init, default local user, se
- **`ChatClient`** Protocol for `chat_completions`.
- **`@tool` / `ToolRegistry`**: decorator infers JSON Schema from type hints; execute tools by name.
- **`run_chat_loop`**: multi-round tool loop; default **streaming** text deltas + raw-SSE tool-call merge; optional `on_limit`.
- **`ChatAgent`**: optional `MemoryStore` (persist on success only); `stream` flag; `pending_retry_text` + `retry()`.
- **`run_chat_loop`**: multi-round tool loop; default **streaming** text deltas + stream tool-call merge; parallel tools; tool-result char budget; optional `on_limit`.
- **`ChatAgent`**: optional `MemoryStore` (persist on success only); `stream`; system prompt; `pending_retry_text` + `retry()`.
- Events: text_delta, assistant_message, tool_call/result, max_rounds, **error**, **cancelled**.
- Config ``[agent]``: `system_prompt`, `max_tool_result_chars`, `parallel_tools`.
### Tools (`tools/`)
Module-level `@tool` handlers. Call `set_workspace_root()` before use.
- **`workspace`**: path resolve under root; path substring denylist; command basename denylist.
- **`file`**: `read_file`, `write_file`, `listdir`, `tree` (depth/entry caps, skip VCS + optional hidden dirs), `edit_replace` (first occurrence), `copy_path` / `move_path` / `delete_path` (shutil; dirs supported).
- **`file`**: `read_file`, `write_file`, `listdir`, `tree`, `edit_replace`, `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`.
- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); PTY `open_pty` / `read_pty` / `write_pty` / `close_pty` (**Unix only**: `pty`+`fork`).
- PTY: structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)`; session limit/idle TTL/output budget; close SIGTERM→SIGKILL.
- CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget.
+1
View File
@@ -1,6 +1,7 @@
from .file import FILE_TOOLS as FILE_TOOLS
from .file import copy_path as copy_path
from .file import delete_path as delete_path
from .file import edit_lineno as edit_lineno
from .file import edit_replace as edit_replace
from .file import listdir as listdir
from .file import move_path as move_path
+2
View File
@@ -1,3 +1,4 @@
from .edit_lineno import edit_lineno as edit_lineno
from .edit_replace import edit_replace as edit_replace
from .fs_ops import copy_path as copy_path
from .fs_ops import delete_path as delete_path
@@ -13,6 +14,7 @@ FILE_TOOLS = [
listdir,
tree,
edit_replace,
edit_lineno,
copy_path,
move_path,
delete_path,
+97
View File
@@ -0,0 +1,97 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from plyngent.agent import tool
from plyngent.tools.workspace import resolve_path
if TYPE_CHECKING:
from pathlib import Path
def _detect_newline(lines: list[str]) -> str:
for sample in lines:
if sample.endswith("\r\n"):
return "\r\n"
if sample.endswith("\n"):
return "\n"
return "\n"
def _to_keepends(new_content: str, newline: str, *, force_trailing: bool) -> list[str]:
if new_content == "":
return []
body = new_content.replace("\r\n", "\n").replace("\r", "\n")
parts = body.split("\n")
# Drop trailing empty part from a final newline
if parts and parts[-1] == "" and body.endswith("\n"):
parts = parts[:-1]
out: list[str] = []
for i, part in enumerate(parts):
is_last = i == len(parts) - 1
if is_last and not body.endswith("\n") and not force_trailing:
out.append(part)
else:
out.append(part + newline)
return out
def _validate_range(start_line: int, end_line: int, n: int) -> str | None:
if start_line < 1:
return "error: start_line must be >= 1"
if end_line < start_line:
return "error: end_line must be >= start_line"
if start_line > n + 1:
return f"error: start_line {start_line} past end of file ({n} lines)"
if start_line == n + 1 and end_line != start_line:
return "error: when appending, end_line must equal start_line"
return None
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 [])
block_lines = _to_keepends(new_content, newline, force_trailing=False)
if block_lines and not block_lines[-1].endswith(("\n", "\r\n")):
block_lines[-1] = block_lines[-1] + newline
_ = target.write_text(text + "".join(block_lines), encoding="utf-8")
return f"appended content after line {n} in {path}"
def _replace_range(
target: Path,
path: str,
lines: list[str],
start_line: int,
end_line: int,
new_content: str,
) -> str:
n = len(lines)
end = min(end_line, n)
newline = _detect_newline(lines[start_line - 1 : end] or lines)
replacement = _to_keepends(new_content, newline, force_trailing=end < n)
new_lines = lines[: start_line - 1] + replacement + lines[end:]
_ = target.write_text("".join(new_lines), encoding="utf-8")
removed = end - start_line + 1
return f"replaced lines {start_line}-{end} ({removed} lines) with {len(replacement)} lines in {path}"
@tool
def edit_lineno(path: str, start_line: int, end_line: int, new_content: str) -> str:
"""Replace lines ``start_line``..``end_line`` (1-based, inclusive) in a file.
``new_content`` may be multi-line. Use an empty string to delete the range.
"""
target = resolve_path(path)
if not target.is_file():
return f"error: not a file: {path}"
text = target.read_text(encoding="utf-8", errors="replace")
lines = text.splitlines(keepends=True)
err = _validate_range(start_line, end_line, len(lines))
if err is not None:
return err
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)
+37
View File
@@ -0,0 +1,37 @@
from __future__ import annotations
from pathlib import Path
from plyngent.tools.file import edit_lineno, read_file, write_file
from tests.test_tools.helpers import call_sync
def test_edit_lineno_replace_middle(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "a.txt", "one\ntwo\nthree\nfour\n")
out = call_sync(edit_lineno, "a.txt", 2, 3, "TWO\nTHREE\n")
assert "replaced lines 2-3" in out
assert call_sync(read_file, "a.txt") == "one\nTWO\nTHREE\nfour\n"
def test_edit_lineno_delete_range(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "b.txt", "a\nb\nc\n")
out = call_sync(edit_lineno, "b.txt", 2, 2, "")
assert "replaced" in out
assert call_sync(read_file, "b.txt") == "a\nc\n"
def test_edit_lineno_append(workspace: object) -> None:
assert isinstance(workspace, Path)
_ = call_sync(write_file, "c.txt", "only\n")
out = call_sync(edit_lineno, "c.txt", 2, 2, "more\n")
assert "appended" in out
assert call_sync(read_file, "c.txt") == "only\nmore\n"
def test_edit_lineno_invalid(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "d.txt", "x\n")
assert "start_line" in call_sync(edit_lineno, "d.txt", 0, 1, "y")
assert "end_line" in call_sync(edit_lineno, "d.txt", 2, 1, "y")