From 81a30d0e4d2ac9a245a03d0d80a93173fe5a6594 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 14 Jul 2026 23:24:34 +0800 Subject: [PATCH] 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. --- CLAUDE.md | 7 +- src/plyngent/tools/__init__.py | 1 + src/plyngent/tools/file/__init__.py | 2 + src/plyngent/tools/file/edit_lineno.py | 97 ++++++++++++++++++++++++++ tests/test_tools/test_edit_lineno.py | 37 ++++++++++ 5 files changed, 141 insertions(+), 3 deletions(-) create mode 100644 src/plyngent/tools/file/edit_lineno.py create mode 100644 tests/test_tools/test_edit_lineno.py diff --git a/CLAUDE.md b/CLAUDE.md index 554d51a..4eb4620 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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. diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 1231005..0251020 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -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 diff --git a/src/plyngent/tools/file/__init__.py b/src/plyngent/tools/file/__init__.py index a8594f3..784192f 100644 --- a/src/plyngent/tools/file/__init__.py +++ b/src/plyngent/tools/file/__init__.py @@ -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, diff --git a/src/plyngent/tools/file/edit_lineno.py b/src/plyngent/tools/file/edit_lineno.py new file mode 100644 index 0000000..2b81fd0 --- /dev/null +++ b/src/plyngent/tools/file/edit_lineno.py @@ -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) diff --git a/tests/test_tools/test_edit_lineno.py b/tests/test_tools/test_edit_lineno.py new file mode 100644 index 0000000..4aa8d65 --- /dev/null +++ b/tests/test_tools/test_edit_lineno.py @@ -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")