core/tools: mtime freshness replaces per-write lineno invalidation

edit_lineno now rejects edits when the file changed since the numbered
read (mtime recorded at read time), so stale line numbers are caught
even when the mutation came from run_argv / an external process — the
class the explicit invalidations could never cover.

This lets write_file / edit_replace / edit_lineno drop their explicit
invalidate calls (the write itself changes the mtime, and the error is
now the accurate "changed since it was read" instead of a misleading
"not read"). copy/move keep explicit invalidation because shutil
preserves the source mtime, which the freshness check cannot see.

src/plyngent/agent/__init__.py

@@ -2,6 +2,7 @@ from .chat import ChatAgent as ChatAgent
 from .chat import get_lineno_read_files as get_lineno_read_files
 from .chat import invalidate_lineno_read as invalidate_lineno_read
 from .chat import lineno_read_lines as lineno_read_lines
+from .chat import lineno_read_mtime as lineno_read_mtime
 from .chat import mark_lineno_read as mark_lineno_read
 from .chat import reset_lineno_tracker as reset_lineno_tracker
 from .chat import was_lineno_read as was_lineno_read

src/plyngent/agent/chat.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from contextvars import ContextVar, Token
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from msgspec import UNSET
@@ -47,14 +48,18 @@ if TYPE_CHECKING:
 
 
 # Per-conversation tracking of lines read with ``with_lineno=True``, keyed by
-# resolved file path. Set to a fresh ``dict`` at the start of each user turn via
+# resolved file path, with the file's mtime at read time (so stale line numbers
-# :func:`reset_lineno_tracker`; cleared after the assistant finishes.
+# after external/process mutations are detected). Set to a fresh ``dict`` at the
-# Tool handlers read/write via :func:`get_lineno_read_files` /
+# start of each user turn via :func:`reset_lineno_tracker`; cleared after the
-# :func:`lineno_read_lines` / :func:`mark_lineno_read`.
+# assistant finishes. Tool handlers read/write via :func:`get_lineno_read_files` /
-_lineno_read_files: ContextVar[dict[str, set[int]] | None] = ContextVar("_lineno_read_files", default=None)
+# :func:`lineno_read_lines` / :func:`lineno_read_mtime` / :func:`mark_lineno_read`.
+_lineno_read_files: ContextVar[dict[str, tuple[set[int], int]] | None] = ContextVar(
+    "_lineno_read_files",
+    default=None,
+)
 
 
-def reset_lineno_tracker() -> Token[dict[str, set[int]] | None]:
+def reset_lineno_tracker() -> Token[dict[str, tuple[set[int], int]] | None]:
     """Replace the tracker with a fresh empty mapping; return the previous token.
 
     Call at the start of a new user turn. Restore with
@@ -80,17 +85,38 @@ def lineno_read_lines(path: str) -> set[int]:
     val = _lineno_read_files.get()
     if val is None:
         return set()
-    return set(val.get(path, ()))
+    entry = val.get(path)
+    return set(entry[0]) if entry is not None else set()
+
+
+def lineno_read_mtime(path: str) -> int | None:
+    """Return the file mtime (ns) recorded when *path* was read, or None."""
+    val = _lineno_read_files.get()
+    if val is None:
+        return None
+    entry = val.get(path)
+    return entry[1] if entry is not None else None
 
 
 def mark_lineno_read(path: str, lines: set[int]) -> None:
     """Record that *path* was read with line numbers this turn.
 
     ``lines`` are 1-based absolute file line numbers (``read_file`` slice).
+    The file's current mtime is recorded too, so edits can detect that the
+    file changed since the read (external/process mutations included).
     """
     val = _lineno_read_files.get()
-    if val is not None:
+    if val is None:
-        val.setdefault(path, set()).update(lines)
+        return
+    try:
+        mtime_ns = Path(path).stat().st_mtime_ns
+    except OSError:
+        mtime_ns = 0
+    entry = val.get(path)
+    if entry is None:
+        val[path] = (set(lines), mtime_ns)
+    else:
+        val[path] = (entry[0] | lines, mtime_ns)
 
 
 def invalidate_lineno_read(path: str) -> None:

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, invalidate_lineno_read, lineno_read_lines, tool
+from plyngent.agent import ToolTag, lineno_read_lines, lineno_read_mtime, tool
 from plyngent.tools.workspace import resolve_path
 
 if TYPE_CHECKING:
@@ -119,10 +119,22 @@ async def edit_lineno(path: str, start_line: int, end_line: int, new_content: st
             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"
 
+    recorded_mtime = lineno_read_mtime(str(target))
+    if recorded_mtime is not None:
+        try:
+            current_mtime = target.stat().st_mtime_ns
+        except OSError:
+            current_mtime = None
+        if current_mtime is not None and current_mtime != recorded_mtime:
+            return (
+                f"error: {path} changed since it was read (line numbers may be stale); "
+                "re-read with read_file(path, with_lineno=true)"
+            )
+
     if start_line == len(lines) + 1:
         result = _append_after(target, path, text, lines, 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.
+    # Line numbers are stale after a write; the fresh mtime makes the next
-    invalidate_lineno_read(str(target))
+    # edit_lineno reject until the file is re-read.
     return result

src/plyngent/tools/file/edit_replace.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from plyngent.agent import ToolTag, invalidate_lineno_read, tool
+from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import resolve_path
 
 
@@ -54,6 +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.
+    # The write changes the file's mtime, so edit_lineno's freshness check
-    invalidate_lineno_read(str(target))
+    # rejects stale line numbers until the model re-reads with_lineno.
     return _success_message(path, replaced=n, found=found)

src/plyngent/tools/file/fs_ops.py

@@ -68,10 +68,13 @@ 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.
+            # A copy does not change the source, but the destination content is
+            # new. copy2/copytree preserve the source mtime, so the freshness
+            # check alone could miss the change — invalidate explicitly.
             invalidate_lineno_read(str(dest))
             return f"copied {label} {src} -> {dest.relative_to(root)}"
         moved = Path(shutil.move(str(source), str(dest))).resolve()
+        # shutil.move preserves mtime too; drop read state for both paths.
         invalidate_lineno_read(str(source))
         invalidate_lineno_read(str(moved))
         return f"moved {_kind(moved)} {src} -> {moved.relative_to(root)}"

src/plyngent/tools/file/write.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from plyngent.agent import ToolTag, invalidate_lineno_read, tool
+from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import resolve_path
 
 
@@ -10,5 +10,6 @@ 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))
+    # The write changes the file's mtime, so edit_lineno's freshness check
+    # rejects stale line numbers until the model re-reads with_lineno.
     return f"wrote {len(content)} characters to {path}"

tests/test_tools/test_edit_lineno.py

@@ -108,7 +108,7 @@ def test_edit_lineno_requires_rerun_after_write(workspace: object) -> None:
     _ = _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
+    assert "changed" in out and "re-read" in out
 
 
 def test_edit_lineno_requires_rerun_after_edit_replace(workspace: object) -> None:
@@ -117,7 +117,7 @@ def test_edit_lineno_requires_rerun_after_edit_replace(workspace: object) -> Non
     _ = _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
+    assert "changed" in out and "re-read" in out
 
 
 def test_edit_lineno_invalidates_after_successful_edit(workspace: object) -> None:
@@ -127,4 +127,19 @@ def test_edit_lineno_invalidates_after_successful_edit(workspace: object) -> Non
     _ = 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
+    assert "changed" in out and "re-read" in out
+
+
+def test_edit_lineno_detects_external_change(workspace: object) -> None:
+    """A mutation outside the file tools (run_argv, another process) is caught."""
+    import os
+    import time
+
+    assert isinstance(workspace, Path)
+    _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\n")
+    _ = _read_lineno("i.txt")
+    # Simulate an external/process-tool mutation: rewrite content + new mtime.
+    _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\nCHANGED\n")
+    os.utime(workspace / "i.txt", (time.time() + 1, time.time() + 1))
+    out = call_sync(edit_lineno, "i.txt", 1, 1, "x\n")
+    assert "changed since it was read" in out