core/tools: edit_replace max_replaces and richer match messages

Default still replaces first match only. Report how many matches were
found, how many replaced, and remaining hits so agents can raise
max_replaces or narrow old_string.
This commit is contained in:
2026-07-19 21:55:16 +08:00
parent 0ec0d542e8
commit d528f65c4c
3 changed files with 69 additions and 7 deletions
+1 -1
View File
@@ -79,7 +79,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
Module-level `@tool` handlers. Call `set_workspace_root()` before use.
- **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist; clearer deny messages.
- **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace`, `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind).
- **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind).
- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); session limit/idle TTL/output budget; close terminate→kill.
- CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget.
+42 -5
View File
@@ -4,17 +4,54 @@ from plyngent.agent import tool
from plyngent.tools.workspace import resolve_path
def _count_non_overlapping(text: str, needle: str) -> int:
"""Count left-to-right non-overlapping matches (same as ``str.replace``)."""
if not needle:
return 0
n = 0
start = 0
while True:
index = text.find(needle, start)
if index < 0:
return n
n += 1
start = index + len(needle)
def _success_message(path: str, *, replaced: int, found: int) -> str:
remaining = found - replaced
unit = "occurrence" if replaced == 1 else "occurrences"
if remaining == 0:
if found == 1:
return f"replaced 1 occurrence in {path}"
return f"replaced {replaced} {unit} in {path} (all {found} matches)"
return (
f"replaced {replaced} {unit} in {path} "
f"({replaced} of {found} matches; {remaining} remain — "
f"raise max_replaces or use a more specific old_string)"
)
@tool
def edit_replace(path: str, old_string: str, new_string: str) -> str:
"""Replace the first occurrence of ``old_string`` with ``new_string`` in a file."""
def edit_replace(path: str, old_string: str, new_string: str, max_replaces: int = 1) -> str:
"""Replace occurrences of ``old_string`` with ``new_string`` in a file.
Replaces left-to-right, non-overlapping. Default ``max_replaces=1`` (first match
only). Raise ``max_replaces`` to change multiple identical hits; use a more
specific ``old_string`` when you need a particular occurrence.
"""
if not old_string:
return "error: old_string must not be empty"
if max_replaces < 1:
return "error: max_replaces must be >= 1"
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")
if old_string not in text:
found = _count_non_overlapping(text, old_string)
if found == 0:
return "error: old_string not found in file"
updated = text.replace(old_string, new_string, 1)
n = min(max_replaces, found)
updated = text.replace(old_string, new_string, n)
_ = target.write_text(updated, encoding="utf-8")
return f"replaced first occurrence in {path}"
return _success_message(path, replaced=n, found=found)
+26 -1
View File
@@ -19,8 +19,33 @@ def test_write_read_listdir_edit(workspace: object) -> None:
def test_edit_replace_first_only(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "t.txt", "aa aa")
_ = call_sync(edit_replace, "t.txt", "aa", "bb")
result = call_sync(edit_replace, "t.txt", "aa", "bb")
assert call_sync(read_file, "t.txt") == "bb aa"
assert "1 of 2" in result or "1 of 2 matches" in result
assert "remain" in result
def test_edit_replace_max_replaces(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "t.txt", "aa aa aa")
result = call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=2)
assert call_sync(read_file, "t.txt") == "bb bb aa"
assert "2 of 3" in result
assert "1 remain" in result
def test_edit_replace_all_matches(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "t.txt", "aa aa")
result = call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=10)
assert call_sync(read_file, "t.txt") == "bb bb"
assert "all 2 matches" in result
def test_edit_replace_max_replaces_invalid(workspace: object) -> None:
del workspace
_ = call_sync(write_file, "t.txt", "aa")
assert "max_replaces" in call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=0)
def test_edit_missing_old_string(workspace: object) -> None: