fix(files): avoid false stale writes from Windows stat/fstat ctime differences

tests/tools/test_file_snapshot.py

@@ -0,0 +1,97 @@
+"""File snapshots tolerate stat/fstat ctime differences, not concurrent changes."""
+
+import hashlib
+import json
+import os
+import stat
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from tools import file_tools_read_tracking as tracking
+
+
+_FIELDS = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
+
+
+@pytest.mark.parametrize("ctime_offset", [0, 100])
+@pytest.mark.parametrize("change", [None, "identity", *(
+    f"{api}:{field}" for api in ("stat", "fstat") for field in _FIELDS
+)])
+def test_snapshot_requires_stable_path_and_descriptor(tmp_path, monkeypatch, ctime_offset, change):
+    path = tmp_path / "snapshot.txt"
+    content = b"raw bytes\r\nincluding \x1a\n"
+    path.write_bytes(content)
+    metadata = dict(st_mode=stat.S_IFREG, st_dev=1, st_ino=2,
+                    st_size=len(content), st_mtime_ns=3, st_ctime_ns=4)
+    path_before = SimpleNamespace(**metadata)
+    path_after = SimpleNamespace(**metadata)
+    fd_before = SimpleNamespace(**{**metadata, "st_ctime_ns": metadata["st_ctime_ns"] + ctime_offset})
+    fd_after = SimpleNamespace(**vars(fd_before))
+    if change == "identity":
+        # Replacement between path stat and open, stable thereafter on the fd.
+        fd_before.st_ino += 1
+        fd_after.st_ino += 1
+    elif change is not None:
+        api, field = change.split(":")
+        snapshot = path_after if api == "stat" else fd_after
+        setattr(snapshot, field, getattr(snapshot, field) + 1)
+
+    # Only this module sees synthetic metadata; open/read/hash use real bytes.
+    fake_os = SimpleNamespace(**vars(os))
+    fake_os.stat = Mock(side_effect=[path_before, path_after])
+    fake_os.fstat = Mock(side_effect=[fd_before, fd_after])
+    monkeypatch.setattr(tracking, "os", fake_os)
+    version = tracking._file_version(str(path))
+    if change is None:
+        assert version == (*tuple(metadata[field] for field in _FIELDS), hashlib.sha256(content).digest())
+    else:
+        assert version is None
+
+
+@pytest.mark.platforms("any")
+@pytest.mark.parametrize("external_change", [None, "edit", "replace"])
+def test_real_read_write_baseline_preserves_freshness(tmp_path, external_change):
+    from tools.file_tools import clear_file_ops_cache
+    from tools.registry import registry
+
+    path = tmp_path / "notes.txt"
+    path.write_bytes(b"original\n")
+    task = f"snapshot-{external_change}"
+
+    def call(name, **args):
+        return json.loads(registry.dispatch(name, {"path": str(path), **args}, task_id=task))
+
+    try:
+        read = call("read_file")
+        assert "error" not in read, read
+        assert "original" in read["content"]
+        tracking.reset_file_dedup(task)
+        assert tracking._has_full_write_baseline(str(path), task)
+        written = call("write_file", content="own edit\n")
+        assert "error" not in written, written
+        assert written.get("verified") is True
+        assert path.read_bytes() == b"own edit\n"
+        assert tracking._has_full_write_baseline(str(path), task)
+
+        if external_change is not None:
+            stamp = path.stat()
+            target = path if external_change == "edit" else tmp_path / "replacement.txt"
+            target.write_bytes(b"external\n")
+            os.utime(target, ns=(stamp.st_atime_ns, stamp.st_mtime_ns))
+            if external_change == "replace":
+                os.replace(target, path)
+            # Neither size nor mtime alone can spot this edit/replacement.
+            assert path.stat().st_size == stamp.st_size
+            assert path.stat().st_mtime_ns == stamp.st_mtime_ns
+            refused = call("write_file", content="clobber\n")
+            assert refused.get("stale_write_blocked"), refused
+            assert path.read_bytes() == b"external\n"
+        else:
+            written = call("write_file", content="next edit\n")
+            assert "error" not in written, written
+            assert path.read_bytes() == b"next edit\n"
+    finally:
+        clear_file_ops_cache(task)
+        tracking._read_tracker.pop(task, None)

tools/file_tools_read_tracking.py

@@ -253,18 +253,27 @@ def _file_metadata(resolved: str) -> tuple | 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):
+        path_before = os.stat(resolved)
+        if not stat.S_ISREG(path_before.st_mode):
             return None
         fd = os.open(resolved, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_BINARY", 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)
+            after = os.fstat(stream.fileno())
+            path_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):
+        version = tuple(getattr(path_before, name) for name in fields)
+        fd_version = tuple(getattr(before, name) for name in fields)
+        # Windows stat/fstat can report different ctimes for the same file.
+        # Compare each API to itself, retaining ctime change detection, and
+        # bind the hashed descriptor to the path by identity, size and mtime.
+        # Return path metadata so _file_metadata() comparisons stay compatible.
+        if (version == tuple(getattr(path_after, name) for name in fields)
+                and fd_version == tuple(getattr(after, name) for name in fields)
+                and version[:-1] == fd_version[:-1]):
             return (*version, digest)
         return None
     except OSError: