fix: preserve known file baselines across partial rereads

tests/tools/test_known_file_write_baseline.py

@@ -0,0 +1,76 @@
+"""A narrower read must not discard unchanged whole-file knowledge."""
+
+import json
+import os
+
+from tools.file_tools import clear_file_ops_cache
+from tools.registry import registry
+
+
+def _call(name, path, task_id, **arguments):
+    result = registry.dispatch(name, {"path": str(path), **arguments}, task_id=task_id)
+    assert isinstance(result, str)
+    return json.loads(result)
+
+
+def test_partial_reread_keeps_an_unchanged_full_read_or_write(tmp_path):
+    original = "first\nsecond\nthird\n"
+    for source in ("read", "write", "pages"):
+        task = f"known-{source}"
+        path = tmp_path / f"{source}.txt"
+        try:
+            if source == "write":
+                assert "error" not in _call("write_file", path, task, content=original)
+            else:
+                path.write_text(original, encoding="utf-8")
+                if source == "pages":
+                    for offset in (1, 2, 3):
+                        assert "error" not in _call("read_file", path, task, offset=offset, limit=1)
+                else:
+                    assert "error" not in _call("read_file", path, task)
+            assert "error" not in _call("read_file", path, task, offset=2, limit=2)
+            written = _call("write_file", path, task, content="replacement\n")
+            assert "error" not in written, (source, written)
+            assert path.read_text(encoding="utf-8") == "replacement\n"
+        finally:
+            clear_file_ops_cache(task)
+
+
+def test_partial_read_cannot_refresh_a_changed_full_baseline(tmp_path):
+    original = "first\nsecond\nthird\n"
+    for source in ("write", "pages"):
+        path = tmp_path / f"changed-{source}.txt"
+        task = f"snapshot-{source}"
+        try:
+            if source == "write":
+                assert "error" not in _call("write_file", path, task, content=original)
+            else:
+                path.write_text(original, encoding="utf-8")
+                assert "error" not in _call("read_file", path, task, offset=1, limit=1)
+            stamp = path.stat()
+            path.write_text("other\nsecond\nthird\n", encoding="utf-8")
+            # mtime alone cannot identify bytes: editors/copy tools can preserve it.
+            os.utime(path, ns=(stamp.st_atime_ns, stamp.st_mtime_ns))
+            assert "error" not in _call("read_file", path, task, offset=2, limit=2)
+            refused = _call("write_file", path, task, content="replacement\n")
+            assert refused.get("stale_write_blocked"), (source, refused)
+            assert path.read_text(encoding="utf-8") == "other\nsecond\nthird\n"
+            # Reading all of the new version is recovery, not a bypass.
+            assert "error" not in _call("read_file", path, task)
+            assert "error" not in _call("write_file", path, task, content="merged\n")
+            assert path.read_text(encoding="utf-8") == "merged\n"
+        finally:
+            clear_file_ops_cache(task)
+
+    # A page that hides part of a line never supplies whole-file knowledge.
+    from tools.tool_output_limits import get_max_line_length
+
+    path = tmp_path / "clamped.txt"
+    path.write_text("x" * (get_max_line_length() + 10) + "\nlast\n", encoding="utf-8")
+    try:
+        assert "error" not in _call("read_file", path, "clamped")
+        refused = _call("write_file", path, "clamped", content="replacement\n")
+        assert refused.get("stale_write_blocked"), refused
+        assert path.read_text(encoding="utf-8").startswith("x" * (get_max_line_length() + 10))
+    finally:
+        clear_file_ops_cache("clamped")

tools/file_operations.py

@@ -581,9 +581,13 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
             hint_parts.append(
                 f"Use offset={end_line + 1} to continue reading "
                 f"(showing {offset}-{end_line} of {total_lines} lines)")
+        from tools.tool_output_limits import get_max_line_length
+        max_line_length = get_max_line_length()
+        truncated_lines = any(len(line) > max_line_length for line in content.split('\n'))
         return ReadResult(
             content=self._add_line_numbers(content, offset), total_lines=total_lines,
-            file_size=file_size, truncated=truncated, hint=" ".join(hint_parts))
+            file_size=file_size, truncated=truncated, hint=" ".join(hint_parts),
+            truncated_lines=True if truncated_lines else None)
 
     def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult:
         """Read a file with pagination, binary detection, and line numbers.
@@ -724,6 +728,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
         kept = bytearray()      # first ``clamp`` bytes of that line
         have_partial = False    # that line has bytes but no newline yet
         last_byte = b""
+        digest = hashlib.sha256()
         try:
             with open(full, "rb") as fh:
                 sample = fh.read(1000)
@@ -735,6 +740,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
                     chunk = fh.read(1 << 20)
                     if not chunk:
                         break
+                    digest.update(chunk)
                     last_byte = chunk[-1:]
                     if lineno > end_line:
                         # Past the window: only the line count and trailing byte
@@ -767,10 +773,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
             page.append(bytes(kept) + b"\n")
 
         read_output = _strip_terminal_fence_leaks(b"".join(page).decode("utf-8", errors="replace"))
-        return self._assemble_read_result(
+        result = self._assemble_read_result(
             read_output, offset=offset, end_line=end_line, total_lines=total_lines,
             file_size=file_size,
             file_ends_with_newline=(last_byte == b"\n") if file_size else None)
+        result._snapshot = (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns, digest.digest())
+        return result
 
     @staticmethod
     def _image_redirect_result(file_size: int) -> ReadResult:
@@ -924,9 +932,13 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
                     f"Note: offset {offset} is beyond the end of the file "
                     f"({total_lines} lines total). Retry with offset <= "
                     f"{total_lines}."))
+        from tools.tool_output_limits import get_max_line_length
+        max_line_length = get_max_line_length()
+        truncated_lines = any(len(line) > max_line_length for line in read_output.split('\n'))
         return ReadResult(
             content=self._add_line_numbers(read_output, offset), total_lines=total_lines,
-            file_size=file_size, truncated=truncated, hint=hint)
+            file_size=file_size, truncated=truncated, hint=hint,
+            truncated_lines=True if truncated_lines else None)
 
     # Confusable characters seen in real filenames, collapsed after NFC.
     _CONFUSABLES = (
@@ -1295,6 +1307,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
             lsp_diagnostics = self._maybe_lsp_diagnostics(path, pre_content=pre_content, post_content=content) or None
         return WriteResult(
             bytes_written=len(content_bytes), dirs_created=dirs_created, verified=content_verified,
+            _content_sha256=hashlib.sha256(content_bytes).hexdigest(),
             lint=lint_result.to_dict() if lint_result else None, lsp_diagnostics=lsp_diagnostics)
 
     # --- PATCH (replace mode) -----------------------------------------------

tools/file_operations_common.py

@@ -17,6 +17,7 @@ class ReadResult:
     total_lines: int = 0
     file_size: int = 0
     truncated: bool = False
+    truncated_lines: Optional[bool] = None
     hint: Optional[str] = None
     is_binary: bool = False
     is_image: bool = False
@@ -25,9 +26,10 @@ class ReadResult:
     dimensions: Optional[str] = None  # For images: "WIDTHxHEIGHT"
     error: Optional[str] = None
     similar_files: List[str] = field(default_factory=list)
+    _snapshot: Optional[tuple] = None
 
     def to_dict(self) -> dict:
-        return {k: v for k, v in self.__dict__.items() if v is not None and v != []}
+        return {k: v for k, v in self.__dict__.items() if not k.startswith("_") and v is not None and v != []}
 
 
 @dataclass
@@ -38,6 +40,7 @@ class WriteResult:
     # True when the on-disk sha256 matched the intended content; None when the
     # backend couldn't verify (no sha256sum). A mismatch is a hard error, never a flag.
     verified: Optional[bool] = None
+    _content_sha256: Optional[str] = None
     lint: Optional[Dict[str, Any]] = None
     # LSP semantic diagnostics, kept separate from ``lint`` (syntax) so the model
     # reads the two as independent signals. None when LSP is off/inapplicable.
@@ -46,7 +49,7 @@ class WriteResult:
     warning: Optional[str] = None
 
     def to_dict(self) -> dict:
-        return {k: v for k, v in self.__dict__.items() if v is not None}
+        return {k: v for k, v in self.__dict__.items() if not k.startswith("_") and v is not None}
 
 
 @dataclass

tools/file_tools.py

@@ -35,6 +35,7 @@ from tools.file_tools_write_guards import (
     _is_internal_file_tool_content, _stale_overwrite_blocker, _stale_write_refusal)
 from tools.file_tools_read_tracking import (
     _bump_consecutive, _cap_read_tracker_data, _check_file_staleness, _check_not_found_cache,
+    _file_metadata, _file_version,
     _mark_full_write_baseline, _mark_verification_stale, _note_read_coverage, _patch_failure_lock,
     _patch_failure_tracker, _read_tracker, _read_tracker_lock, _record_not_found,
     _record_patch_failure, _reset_patch_failures, _task_data, _update_read_timestamp)
@@ -104,6 +105,7 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines
         f"{lines_kept} line(s) (showing lines {offset}-{next_offset - 1} of "
         f"{total_lines}). Use offset={next_offset} to continue.")
     if len(trimmed.split("\n", 1)[0]) >= max_chars:
+        result_dict["truncated_lines"] = True
         result_dict["hint"] += (
             " Note: the first line alone exceeded the budget and was "
             "clamped mid-line; its remainder is not retrievable via offset.")
@@ -455,6 +457,9 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
     total_lines = len(lines)
     end_line = offset + limit - 1
     page_text = "\n".join(lines[offset - 1:end_line])
+    from tools.tool_output_limits import get_max_line_length
+    max_line_length = get_max_line_length()
+    truncated_lines = any(len(line) > max_line_length for line in page_text.split('\n'))
     result_dict = {
         "content": file_ops._add_line_numbers(page_text, offset) if page_text else "",
         "total_lines": total_lines,
@@ -476,7 +481,8 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
         redacted = result_dict["content"] != rendered
     else:
         redacted = False
-    if offset == 1 and not result_dict["truncated"] and not redacted:
+    if (offset == 1 and not result_dict["truncated"] and not redacted
+            and not truncated_lines and not result_dict.get("truncated_lines")):
         # The whole document was shown, so a text-authorable format (.ipynb)
         # may later be overwritten by write_file; the binary-container guard
         # keeps refusing .docx/.xlsx/.pdf regardless of this baseline.
@@ -522,7 +528,7 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str:
 def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_str: str,
                             offset: int, limit: int, dedup_key: tuple, *, partial: bool,
                             redacted: bool = False, end_line: int | None = None,
-                            total_lines=None) -> int:
+                            total_lines=None, version_before=None, snapshot=None) -> int:
     """Bookkeeping after a real (non-stub) read; returns the consecutive-read count.
 
     Per-task tracker under the lock (stub counter, history, consecutive count,
@@ -535,7 +541,9 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
     background-review read-mark (a FULL read of a skill file counts like
     skill_view so a follow-up skill_manage(patch) is accepted).
     """
-    complete = not partial
+    version = (snapshot or _file_version(resolved_str)) if version_before is not None else None
+    stable = version is not None and version[:-1] == version_before == _file_metadata(resolved_str)
+    complete = False
     with _read_tracker_lock:
         task_data["dedup_hits"].pop(dedup_key, None)
         task_data["dedup_generation_reads"].add(dedup_key)
@@ -543,15 +551,28 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
         count = _bump_consecutive(task_data, ("read", path, offset, limit))
         try:
             _mtime_now = os.path.getmtime(resolved_str)
-            task_data["dedup"][dedup_key] = _mtime_now
             task_data.setdefault("read_timestamps", {})[resolved_str] = _mtime_now
-            if partial and end_line is not None:
-                complete, redacted = _note_read_coverage(
-                    task_data, resolved_str, _mtime_now, offset, end_line, total_lines, redacted)
         except OSError:
             pass
-        if complete and not redacted:
+        baselines = task_data["full_write_baselines"]
-            task_data.setdefault("full_write_baselines", set()).add(resolved_str)
+        if stable and version is not None and count < 4:
+            task_data["dedup"][dedup_key] = version_before
+            # A narrower view does not undo knowledge of these same bytes. Do
+            # not revive a baseline after a partial read of a different version.
+            complete = baselines.get(resolved_str) == version
+            if not complete:
+                complete = not partial
+                if partial and end_line is not None:
+                    complete, redacted = _note_read_coverage(
+                        task_data, resolved_str, version, offset, end_line, total_lines, redacted)
+                complete = complete and not redacted
+            if complete:
+                baselines[resolved_str] = version
+        if not complete:
+            baselines.pop(resolved_str, None)
+        if not stable or count >= 4:
+            task_data["dedup"].pop(dedup_key, None)
+            task_data["dedup_generation_reads"].discard(dedup_key)
         _cap_read_tracker_data(task_data)
 
     try:
@@ -642,27 +663,27 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
         dedup_key = (resolved_str, offset, limit)
         with _read_tracker_lock:
             task_data = _task_data(task_id)
-            cached_mtime = task_data["dedup"].get(dedup_key)
+            cached_version = task_data["dedup"].get(dedup_key)
             # First unchanged read after a compaction boundary serves full content
             # (the summary may have dropped exact bytes); later ones get the stub.
             content_served_in_generation = dedup_key in task_data["dedup_generation_reads"]
         # Same rule as skill_view: the review fork shares the parent's task_id and its
         # read-before-write guard needs a real read, which the stub path never records (#95976).
-        if cached_mtime is not None and not is_background_review():
+        file_ops = _get_file_ops(task_id)
-            try:
+        version_before = _file_metadata(resolved_str) if _file_ops_uses_host_paths(file_ops) else None
-                if os.path.getmtime(resolved_str) == cached_mtime and content_served_in_generation:
+        if (cached_version is not None and not is_background_review()
-                    return _dedup_stub_or_block(task_data, dedup_key, path)
+                and version_before == cached_version and content_served_in_generation):
-            except OSError:
+            return _dedup_stub_or_block(task_data, dedup_key, path)
-                pass  # stat failed — fall through to full read
 
-        result = _get_file_ops(task_id).read_file(path, offset, limit)
+        result = file_ops.read_file(path, offset, limit)
         result_dict = result.to_dict()
 
-        # Cache a not-found result for retries. Deliberately NO early return:
+        # Failed reads cannot establish whole-file knowledge.
-        # error results still flow through the tracking below unchanged.
         _err = result_dict.get("error") or ""
         if isinstance(_err, str) and _err.startswith("File not found:"):
             _record_not_found("read", resolved_str, task_id, json.dumps(result_dict, ensure_ascii=False))
+        if _err or result_dict.get("is_binary"):
+            return json.dumps(result_dict, ensure_ascii=False)
 
         # Char budget on the FORMATTED content (what enters context), BEFORE
         # redaction (skip the regex pass on huge content); truncate gracefully
@@ -705,7 +726,10 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
                 end_line = min(end_line, total_lines)
         count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit,
                                         dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")),
-                                        redacted=redacted, end_line=end_line, total_lines=total_lines)
+                                        redacted=redacted or bool(result_dict.get("truncated_lines")),
+                                        end_line=end_line, total_lines=total_lines,
+                                        version_before=version_before,
+                                        snapshot=getattr(result, "_snapshot", None))
         if count >= 4:
             return tool_error(
                 f"BLOCKED: You have read this exact file region {count} times in a row. "
@@ -865,7 +889,8 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
                 return json.dumps(_stale_write_refusal(path, blocker, _resolved), ensure_ascii=False)
             warnings = _edit_warnings([path], path_to_resolved, task_id)
             rewrite_hint = _whole_file_rewrite_hint(task_id, _resolved, content)
-            result_dict = _get_file_ops(task_id).write_file(_resolved or path, content).to_dict()
+            result = _get_file_ops(task_id).write_file(_resolved or path, content)
+            result_dict = result.to_dict()
             if warnings:
                 result_dict["_warning"] = warnings[0]
             if rewrite_hint and not result_dict.get("error"):
@@ -881,7 +906,7 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
                     result_dict["files_modified"] = [_resolved]
                     # Own write = current whole-file content: consecutive
                     # same-task writes stay unblocked. patch never does this.
-                    _mark_full_write_baseline(_resolved, task_id)
+                    _mark_full_write_baseline(_resolved, task_id, getattr(result, "_content_sha256", None))
                 _note_edited(task_id, [path], path_to_resolved, session_id)
         return json.dumps(result_dict, ensure_ascii=False)
     except Exception as e:

tools/file_tools_read_tracking.py

@@ -3,12 +3,12 @@
 Process-lifetime state behind read_file/search_files/write_file/patch.
 Per task_id ``_read_tracker``
 stores: ``last_key``/``consecutive`` (loop detection; reset by any OTHER tool
-call), ``read_history`` (diagnostics), ``dedup`` (key -> mtime; survives context
+call), ``read_history`` (diagnostics), ``dedup`` (key -> file metadata; survives context
 compression), ``dedup_generation_reads`` (keys whose full content was served since
 the last compaction boundary; cleared on compression so one recovery read returns
 full content), ``dedup_hits`` (stub-loop breaker), ``read_timestamps``
 (staleness warnings), ``read_coverage`` (per resolved path: the line ranges the
-task has paged through at one mtime — contiguous pages that reach the last line
+task has paged through at one file version — contiguous pages that reach the last line
 count as a whole-file read), ``full_write_baselines`` (resolved paths whose
 whole-file content this task saw via unredacted read_file page(s) or wrote via
 write_file; required before write_file may overwrite an existing file — patch
@@ -16,12 +16,14 @@ never qualifies) and ``not_found`` (short-TTL negative cache). Every
 container is hard-capped (``_cap_read_tracker_data``) so long sessions stay small.
 """
 
+import hashlib
 import logging
 import os
+import stat
 import threading
 import time
 
-from tools.file_state import _evict_oldest, _mtime_or_none
+from tools.file_state import _evict_oldest
 from tools.file_tools_paths import _authoritative_workspace_root, _resolve_path_for_task
 
 logger = logging.getLogger("tools.file_tools")
@@ -50,10 +52,9 @@ def _task_data(task_id: str) -> dict:
     (search_tool / tests create partial entries). Lock must be held."""
     task_data = _read_tracker.setdefault(task_id, {
         "last_key": None, "consecutive": 0, "read_history": set()})
-    for key in ("dedup", "dedup_hits", "read_timestamps", "read_coverage"):
+    for key in ("dedup", "dedup_hits", "read_timestamps", "read_coverage", "full_write_baselines"):
         task_data.setdefault(key, {})
-    for key in ("dedup_generation_reads", "full_write_baselines"):
+    task_data.setdefault("dedup_generation_reads", set())
-        task_data.setdefault(key, set())
     return task_data
 
 
@@ -155,13 +156,14 @@ def _bump_consecutive(task_data: dict, key: tuple) -> int:
 
 def reset_file_dedup(task_id: str = None):
     """Advance the read-dedup generation after context compression (one task, or all
-    when ``task_id`` is None). The per-key ``dedup`` mtime map is PRESERVED so unchanged
+    when ``task_id`` is None). The per-key ``dedup`` metadata map is PRESERVED so unchanged
     files keep returning stubs instead of re-bloating the reclaimed context; the
     generation-read set is cleared so the FIRST unchanged read of each key after
     compaction returns full content the summary may have dropped. Stub-hit counters
     are cleared so the hard block restarts fresh. write_file baselines survive
-    exactly like the dedup map does — for files whose mtime still matches the
+    exactly like the dedup map does — while the file metadata still matches the
-    stamp this task recorded; a baseline whose file changed underneath is dropped
+    stamp this task recorded; byte identity is checked before overwriting. A baseline
+    whose file changed underneath is dropped
     (the stat runs outside the lock so a hung mount cannot stall other tasks)."""
     with _read_tracker_lock:
         if task_id:
@@ -172,13 +174,15 @@ def reset_file_dedup(task_id: str = None):
             if "dedup_hits" in task_data:
                 task_data["dedup_hits"].clear()
             task_data.setdefault("dedup_generation_reads", set()).clear()
-        candidates = [(task_data, list(task_data.get("full_write_baselines", ())),
+        candidates = [(task_data, dict(task_data.get("full_write_baselines", {})))
-                       dict(task_data.get("read_timestamps", {}))) for task_data in targets]
+                      for task_data in targets]
-    for task_data, baselines, stamps in candidates:
+    for task_data, baselines in candidates:
-        changed = {p for p in baselines if _mtime_or_none(p) != stamps.get(p)}
+        changed = {p for p, version in baselines.items() if _file_metadata(p) != version[:-1]}
         if changed:
             with _read_tracker_lock:
-                task_data.get("full_write_baselines", set()).difference_update(changed)
+                for p in changed:
+                    if task_data["full_write_baselines"].get(p) == baselines[p]:
+                        task_data["full_write_baselines"].pop(p, None)
 
 
 def notify_other_tool_call(task_id: str = "default"):
@@ -238,37 +242,70 @@ def _update_read_timestamp(filepath: str, task_id: str) -> None:
             _cap_read_tracker_data(task_data)
 
 
-def _mark_full_write_baseline(resolved: str, task_id: str) -> None:
+def _file_metadata(resolved: str) -> tuple | None:
+    try:
+        st = os.stat(resolved)
+        return st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns
+    except OSError:
+        return None
+
+
+def _file_version(resolved: str) -> tuple | None:
+    """A byte snapshot, not just mtime (editors/copy tools can preserve that)."""
+    try:
+        if not stat.S_ISREG(os.stat(resolved).st_mode):
+            return None
+        fd = os.open(resolved, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0))
+        with os.fdopen(fd, "rb") as stream:
+            before = os.fstat(stream.fileno())
+            if not stat.S_ISREG(before.st_mode):
+                return None
+            digest = hashlib.file_digest(stream, "sha256").digest()
+            after = os.stat(resolved)
+        fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
+        version = tuple(getattr(before, name) for name in fields)
+        if version == tuple(getattr(after, name) for name in fields):
+            return (*version, digest)
+        return None
+    except OSError:
+        return None
+
+
+def _mark_full_write_baseline(resolved: str, task_id: str, expected_sha256: str | None = None) -> None:
     """Record that *task_id* saw the whole current content of *resolved* (full
     unredacted read_file, or its own successful write_file), so a later
     write_file may replace the file. Acquires the lock itself."""
+    version = _file_version(resolved)
+    if version is None or (expected_sha256 is not None and version[-1].hex() != expected_sha256):
+        return
     with _read_tracker_lock:
         task_data = _task_data(task_id)
-        task_data["full_write_baselines"].add(str(resolved))
+        task_data["full_write_baselines"][str(resolved)] = version
         _cap_read_tracker_data(task_data)
 
 
 def _has_full_write_baseline(resolved: str, task_id: str) -> bool:
     with _read_tracker_lock:
         task_data = _read_tracker.get(task_id) or {}
-        return str(resolved) in task_data.get("full_write_baselines", set())
+        baseline = task_data.get("full_write_baselines", {}).get(str(resolved))
+    return baseline is not None and _file_version(resolved) == baseline
 
 
 _READ_COVERAGE_RANGES_CAP = 256
 
 
-def _note_read_coverage(task_data: dict, resolved: str, mtime: float, start: int, end: int,
+def _note_read_coverage(task_data: dict, resolved: str, version: tuple, start: int, end: int,
                         total_lines, redacted: bool) -> tuple[bool, bool]:
     """Merge the page ``start..end`` into this task's coverage of *resolved* and return
-    ``(complete, redacted_any)``: whether pages taken at this same *mtime* now reach from
+    ``(complete, redacted_any)``: whether pages taken at this same *version* now reach from
     line 1 to *total_lines*, and whether any of them came back redacted. A file too large
     for one read_file page (>2000 lines / the char budget) can only ever be seen this
-    way, so paging through it must count as a whole-file read. A new mtime restarts the
+    way, so paging through it must count as a whole-file read. A new version restarts the
     coverage (the earlier pages describe a file that no longer exists). Lock must be held."""
     coverage = task_data.setdefault("read_coverage", {})
     entry = coverage.get(resolved)
-    if entry is None or entry["mtime"] != mtime or len(entry["ranges"]) > _READ_COVERAGE_RANGES_CAP:
+    if entry is None or entry["version"] != version or len(entry["ranges"]) > _READ_COVERAGE_RANGES_CAP:
-        entry = coverage[resolved] = {"mtime": mtime, "ranges": [], "redacted": False}
+        entry = coverage[resolved] = {"version": version, "ranges": [], "redacted": False}
     entry["redacted"] = entry["redacted"] or redacted
     merged: list[tuple[int, int]] = []
     for s, e in sorted(entry["ranges"] + [(start, end)]):

website/docs/reference/tools-reference.md

@@ -117,6 +117,8 @@ Scoped to the Feishu document-comment handler. Drives comment read/write operati
 | `search_files` | Search file contents or find files by name. Use this instead of grep/rg/find/ls in terminal. Ripgrep-backed, faster than shell equivalents. Content search (target='content'): Regex search inside files. Output modes: full matches with line… | — |
 | `write_file` | Write content to a file, completely replacing existing content. Use this instead of echo/cat heredoc in terminal. Creates parent directories automatically. OVERWRITES the entire file — use 'patch' for targeted edits. For an existing file, call read_file first: write_file refuses (file untouched) when the task has no current full read/write of the file or the file changed on disk since — on refusal, read_file, merge, retry. Auto-runs syntax checks on .py/.json/.yaml/.toml and other linted languages; only NEW errors introduced by the write are surfaced. | — |
 
+For local files, a full unredacted read (including all pages of the same file version) or a successful `write_file` supplies a whole-file baseline. Reading a smaller region afterward does not discard that baseline while the bytes remain unchanged. A changed file, an unread file, or a view with hidden/redacted or clamped content still needs a full current read before replacement; `patch` remains available for targeted edits. Writes made through terminal commands or `execute_code` do not establish a `write_file` baseline.
+
 ## `homeassistant` toolset
 
 | Tool | Description | Requires environment |