core/tools: invalidate lineno read state on file mutation

src/plyngent/tools/file/edit_lineno.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from plyngent.agent import ToolTag, lineno_read_lines, tool
+from plyngent.agent import ToolTag, invalidate_lineno_read, lineno_read_lines, tool
 from plyngent.tools.workspace import resolve_path
 
 if TYPE_CHECKING:
@@ -120,5 +120,9 @@ async def edit_lineno(path: str, start_line: int, end_line: int, new_content: st
         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)
+        result = _append_after(target, path, text, lines, new_content)
-    return _replace_range(target, path, lines, start_line, end_line, new_content)
+    else:
+        result = _replace_range(target, path, lines, start_line, end_line, new_content)
+    # Line numbers are stale after a write; force a re-read before next edit.
+    invalidate_lineno_read(str(target))
+    return result

src/plyngent/tools/file/edit_replace.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from plyngent.agent import ToolTag, tool
+from plyngent.agent import ToolTag, invalidate_lineno_read, tool
 from plyngent.tools.workspace import resolve_path
 
 
@@ -54,4 +54,6 @@ async def edit_replace(path: str, old_string: str, new_string: str, max_replaces
     n = min(max_replaces, found)
     updated = text.replace(old_string, new_string, n)
     _ = target.write_text(updated, encoding="utf-8")
+    # Any lineno read state is stale after a write; force a re-read.
+    invalidate_lineno_read(str(target))
     return _success_message(path, replaced=n, found=found)

src/plyngent/tools/file/fs_ops.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import shutil
 from pathlib import Path
 
-from plyngent.agent import ToolTag, tool
+from plyngent.agent import ToolTag, invalidate_lineno_read, tool
 from plyngent.tools.workspace import WorkspaceError, get_workspace_root, resolve_path
 
 
@@ -68,8 +68,12 @@ def _copy_or_move_validated(
             else:
                 _ = shutil.copy2(source, dest, follow_symlinks=False)
                 label = _kind(source)
+            # A copy does not change the source, but the destination content is new.
+            invalidate_lineno_read(str(dest))
             return f"copied {label} {src} -> {dest.relative_to(root)}"
         moved = Path(shutil.move(str(source), str(dest))).resolve()
+        invalidate_lineno_read(str(source))
+        invalidate_lineno_read(str(moved))
         return f"moved {_kind(moved)} {src} -> {moved.relative_to(root)}"
     except WorkspaceError as exc:
         return f"error: {exc}"
@@ -110,11 +114,13 @@ async def move_path(src: str, dst: str, *, overwrite: bool = False) -> str:
 def _delete_directory(target: Path, path: str, *, recursive: bool) -> str:
     if recursive:
         shutil.rmtree(target)
+        invalidate_lineno_read(str(target))
         return f"deleted directory {path} (recursive)"
     try:
         _ = next(target.iterdir())
     except StopIteration:
         target.rmdir()
+        invalidate_lineno_read(str(target))
         return f"deleted empty directory {path}"
     return f"error: directory not empty: {path} (set recursive=true)"
 
@@ -123,6 +129,7 @@ def _delete_target(target: Path, path: str, *, recursive: bool) -> str:
     if target.is_symlink() or target.is_file():
         kind = _kind(target)
         target.unlink()
+        invalidate_lineno_read(str(target))
         return f"deleted {kind} {path}"
     if target.is_dir():
         return _delete_directory(target, path, recursive=recursive)

src/plyngent/tools/file/write.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from plyngent.agent import ToolTag, tool
+from plyngent.agent import ToolTag, invalidate_lineno_read, tool
 from plyngent.tools.workspace import resolve_path
 
 
@@ -10,4 +10,5 @@ async def write_file(path: str, content: str) -> str:
     target = resolve_path(path)
     target.parent.mkdir(parents=True, exist_ok=True)
     _ = target.write_text(content, encoding="utf-8")
+    invalidate_lineno_read(str(target))
     return f"wrote {len(content)} characters to {path}"

tests/test_tools/test_edit_lineno.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from pathlib import Path
 
 from plyngent.agent import reset_lineno_tracker
-from plyngent.tools.file import edit_lineno, read_file, write_file
+from plyngent.tools.file import edit_lineno, edit_replace, read_file, write_file
 from tests.test_tools.helpers import call_sync
 
 
@@ -84,3 +84,31 @@ def test_edit_lineno_rejects_without_prior_lineno_read(workspace: object) -> Non
     _ = reset_lineno_tracker()
     out = call_sync(edit_lineno, "e.txt", 1, 1, "hi\n")
     assert "read" in out and "with_lineno=true" in out
+
+
+def test_edit_lineno_requires_rerun_after_write(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "f.txt", "one\ntwo\nthree\n")
+    _ = _read_lineno("f.txt")
+    _ = call_sync(write_file, "f.txt", "changed\n")
+    out = call_sync(edit_lineno, "f.txt", 1, 1, "x\n")
+    assert "not read" in out
+
+
+def test_edit_lineno_requires_rerun_after_edit_replace(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "g.txt", "one\ntwo\nthree\n")
+    _ = _read_lineno("g.txt")
+    _ = call_sync(edit_replace, "g.txt", "two", "TWO")
+    out = call_sync(edit_lineno, "g.txt", 2, 2, "x\n")
+    assert "not read" in out
+
+
+def test_edit_lineno_invalidates_after_successful_edit(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "h.txt", "a\nb\nc\n")
+    _ = _read_lineno("h.txt")
+    _ = call_sync(edit_lineno, "h.txt", 2, 2, "B\n")
+    # After the edit, line numbers are stale — must re-read before editing again.
+    out = call_sync(edit_lineno, "h.txt", 2, 2, "C\n")
+    assert "not read" in out