fix(cache): implement racily-clean guard for file hashing; enhance stat signature checks

graphify/cache.py

@@ -7,6 +7,7 @@ import json
 import os
 import re
 import tempfile
+import time
 import warnings
 from collections.abc import Iterable
 from pathlib import Path
@@ -178,11 +179,12 @@ def _body_content(content: bytes) -> bytes:
     return text[closer.start() + 3:].encode()
 
 
-# Stat-based index: maps absolute path → {size, mtime_ns, hash}.
+# Stat-based index: maps absolute path → {size, mtime_ns, indexed_at_ns, ...}.
 # Loaded once per process, flushed via atexit. Skips full file reads when
 # size+mtime_ns are unchanged — same trade-off as make(1).
-# Correctness risks: `touch` causes a harmless extra re-hash; same-size edits
-# within NFS second-resolution mtime have a 1-second window (same as make).
+# Correctness risks: `touch` causes a harmless extra re-hash. Same-size edits
+# inside one mtime tick used to return the PREVIOUS content's digest; the
+# racily-clean guard below closes that hole (see _stat_sig_fresh).
 # `graphify extract --force` / `graphify update --force` (or GRAPHIFY_FORCE=1)
 # skip the cache reads and re-dispatch everything when needed (#1894).
 _stat_index: dict[str, dict] = {}
@@ -194,6 +196,84 @@ _stat_index_anchor: Path | None = None
 _stat_index_dirty: bool = False
 
 
+# Filesystem mtime granularity, in nanoseconds. A stat signature only proves a
+# file is unchanged when the clock that stamped its mtime is finer-grained than
+# the interval between two writes — which is false almost everywhere: NTFS
+# advances mtime on the ~15.6 ms system tick, FAT/exFAT on 2 s, and Linux
+# stamps from the coarse (jiffies) clock even though ext4 stores nanoseconds.
+# 2 s is the conservative default that covers all of them. It costs nothing in
+# practice: only files modified within the last 2 s lose the fastpath, and in a
+# real corpus those are exactly the handful of files that changed and have to be
+# read anyway. Override with GRAPHIFY_MTIME_GRANULARITY_MS (0 disables the
+# guard and restores the pre-fix behaviour).
+_MTIME_GRANULARITY_NS = 2_000_000_000
+
+
+def _mtime_granularity_ns() -> int:
+    """Return the assumed filesystem mtime granularity in nanoseconds.
+
+    Read fresh on every call so the env var can be set after import (and so
+    tests can flip it without reloading the module).
+    """
+    raw = os.environ.get("GRAPHIFY_MTIME_GRANULARITY_MS", "").strip()
+    if raw:
+        try:
+            ms = float(raw)
+        except ValueError:
+            return _MTIME_GRANULARITY_NS
+        if ms >= 0:
+            return int(ms * 1_000_000)
+    return _MTIME_GRANULARITY_NS
+
+
+def _stat_sig_fresh(entry: object, st: "os.stat_result") -> bool:
+    """True if ``entry`` provably describes the file's CURRENT content.
+
+    Beyond matching (size, mtime_ns), the entry must be *racily clean* in git's
+    sense: we must have read the content strictly after the file's mtime tick
+    had already closed. Otherwise a write that landed between our read and the
+    end of that tick would have left mtime (and, for a same-length edit, size)
+    untouched, and the stored digest would describe content that is no longer
+    on disk.
+
+    ``indexed_at_ns`` is the wall clock captured immediately BEFORE the content
+    was read. Requiring ``mtime + granularity <= indexed_at`` means any later
+    write necessarily lands in a new tick and so changes mtime, making it
+    visible to the next signature comparison.
+
+    Entries written by an older graphify carry no ``indexed_at_ns``; they are
+    treated as untrusted (one re-read each), and gain the field when rewritten.
+    """
+    if not isinstance(entry, dict):
+        return False
+    if entry.get("size") != st.st_size or entry.get("mtime_ns") != st.st_mtime_ns:
+        return False
+    indexed_at = entry.get("indexed_at_ns")
+    if not isinstance(indexed_at, int):
+        return False
+    return st.st_mtime_ns + _mtime_granularity_ns() <= indexed_at
+
+
+def _stat_entry_for(abs_key: str, st: "os.stat_result", observed_at_ns: int) -> dict:
+    """Get-or-reset the index entry for ``abs_key`` and stamp when it was read.
+
+    Reuses the existing dict when the stat signature still matches, so
+    co-located values (other salts' digests, ``word_count``) survive; resets it
+    otherwise, so a stale ``word_count`` cannot outlive the content it counted.
+
+    ``observed_at_ns`` must be the clock reading taken *before* the content was
+    read — see :func:`_stat_sig_fresh` for why the ordering matters.
+    """
+    entry = _stat_index.get(abs_key)
+    if (not isinstance(entry, dict)
+            or entry.get("size") != st.st_size
+            or entry.get("mtime_ns") != st.st_mtime_ns):
+        entry = {"size": st.st_size, "mtime_ns": st.st_mtime_ns}
+        _stat_index[abs_key] = entry
+    entry["indexed_at_ns"] = observed_at_ns
+    return entry
+
+
 def _stat_key_to_relative(key: str, anchor: Path) -> str:
     """Return ``key`` as a forward-slash relative path from ``anchor``.
 
@@ -325,8 +405,10 @@ def file_hash(path: Path, root: Path = Path("."), cache_root: "Path | None" = No
     """SHA256 of file contents + path relative to root.
 
     Uses a stat-based fastpath (size + mtime_ns) to skip full reads when the
-    file hasn't changed. Falls through to full SHA256 on first encounter or
-    when stat changes. Index is flushed atomically at process exit.
+    file hasn't changed. Falls through to full SHA256 on first encounter, when
+    stat changes, and when the recorded signature is not yet provably stable
+    (see :func:`_stat_sig_fresh`) — so two different contents can never share a
+    digest. Index is flushed atomically at process exit.
 
     Using a relative path (not absolute) makes cache entries portable across
     machines and checkout directories, so shared caches and CI work correctly.
@@ -363,11 +445,8 @@ def file_hash(path: Path, root: Path = Path("."), cache_root: "Path | None" = No
     st: "os.stat_result | None" = None
     try:
         st = p.stat()
-        entry = _stat_index.get(abs_key)
-        if (isinstance(entry, dict)
-                and entry.get("size") == st.st_size
-                and entry.get("mtime_ns") == st.st_mtime_ns):
-            hashes = entry.get("hashes")
+        if _stat_sig_fresh(_stat_index.get(abs_key), st):
+            hashes = _stat_index[abs_key].get("hashes")
             if isinstance(hashes, dict):
                 cached = hashes.get(salt)
                 if isinstance(cached, str):
@@ -377,6 +456,9 @@ def file_hash(path: Path, root: Path = Path("."), cache_root: "Path | None" = No
     except OSError:
         pass
 
+    # Captured BEFORE the read so the stamp can never post-date content that
+    # changed while we were reading it (see _stat_sig_fresh).
+    observed_at_ns = time.time_ns()
     raw = p.read_bytes()
     content = _body_content(raw) if p.suffix.lower() == ".md" else raw
     h = hashlib.sha256()
@@ -386,19 +468,13 @@ def file_hash(path: Path, root: Path = Path("."), cache_root: "Path | None" = No
     digest = h.hexdigest()
 
     if st is not None:
-        entry = _stat_index.get(abs_key)
-        if (isinstance(entry, dict)
-                and entry.get("size") == st.st_size
-                and entry.get("mtime_ns") == st.st_mtime_ns):
-            hashes = entry.get("hashes")
-            if not isinstance(hashes, dict):
-                hashes = {}
-                entry["hashes"] = hashes
-            hashes[salt] = digest       # preserve a co-located word_count / other salts
-            entry.pop("hash", None)     # retire the un-salted legacy digest
-        else:
-            _stat_index[abs_key] = {"size": st.st_size, "mtime_ns": st.st_mtime_ns,
-                                    "hashes": {salt: digest}}
+        entry = _stat_entry_for(abs_key, st, observed_at_ns)
+        hashes = entry.get("hashes")
+        if not isinstance(hashes, dict):
+            hashes = {}
+            entry["hashes"] = hashes
+        hashes[salt] = digest       # preserve a co-located word_count / other salts
+        entry.pop("hash", None)     # retire the un-salted legacy digest
         _stat_index_dirty = True
 
     return digest
@@ -426,26 +502,18 @@ def cached_word_count(path: Path, root: Path, compute, cache_root: "Path | None"
     try:
         st = p.stat()
         entry = _stat_index.get(abs_key)
-        if (entry
-                and entry.get("size") == st.st_size
-                and entry.get("mtime_ns") == st.st_mtime_ns
-                and "word_count" in entry):
+        if _stat_sig_fresh(entry, st) and "word_count" in entry:
             return entry["word_count"]
     except OSError:
         pass
 
+    # Captured BEFORE compute() reads the file, for the same reason file_hash
+    # stamps before its read (see _stat_sig_fresh).
+    observed_at_ns = time.time_ns()
     wc = compute(Path(path))
 
     if st is not None:
-        entry = _stat_index.get(abs_key)
-        if (entry
-                and entry.get("size") == st.st_size
-                and entry.get("mtime_ns") == st.st_mtime_ns):
-            entry["word_count"] = wc  # augment the existing hash entry in place
-        else:
-            _stat_index[abs_key] = {
-                "size": st.st_size, "mtime_ns": st.st_mtime_ns, "word_count": wc,
-            }
+        _stat_entry_for(abs_key, st, observed_at_ns)["word_count"] = wc
         _stat_index_dirty = True
 
     return wc

tests/test_cache.py

@@ -1325,3 +1325,64 @@ def test_prompt_file_reflects_edited_spec(tmp_path):
     _os.utime(spec, ns=(0, 0))  # force a distinct stat signature
     _, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt_file=str(spec))
     assert uncached == [str(f)], "an edited spec must invalidate, not reuse the memo"
+
+
+# --- stat-fastpath racily-clean guard ---------------------------------------
+# (size, mtime_ns) alone cannot prove a file is unchanged: NTFS advances mtime
+# on a ~15.6 ms tick, so a same-length rewrite inside one tick leaves the
+# signature identical and the memo used to return the PREVIOUS content's
+# digest. These two tests pin both halves of the fix — the hole is closed, and
+# the fastpath still actually fires for files whose mtime tick has closed.
+
+def test_file_hash_detects_same_size_rewrite_within_one_mtime_tick(tmp_path):
+    """A same-length edit must change the digest even when the filesystem
+    reports an identical (size, mtime_ns) for both writes.
+
+    The collision is forced with utime rather than raced for: on a filesystem
+    with fine-grained timestamps the two writes would land in different ticks
+    and the memo would never be consulted, making the test vacuous. Pinning
+    both writes to one mtime models the coarse-granularity filesystem (NTFS,
+    FAT, NFS) on every host.
+    """
+    import os as _os
+
+    _reset_stat_index()
+    f = tmp_path / "mod.py"
+
+    f.write_text("x = 1  # aaa\n", encoding="utf-8")
+    st = f.stat()
+    h1 = file_hash(f, tmp_path)
+
+    f.write_text("x = 2  # bbb\n", encoding="utf-8")   # same length, new content
+    _os.utime(f, ns=(st.st_atime_ns, st.st_mtime_ns))  # ...inside the same tick
+
+    assert f.stat().st_size == st.st_size and f.stat().st_mtime_ns == st.st_mtime_ns, (
+        "test setup failed to reproduce an identical stat signature"
+    )
+
+    h2 = file_hash(f, tmp_path)
+    assert h1 != h2, "same-size rewrite returned the previous content's digest"
+
+
+def test_file_hash_fastpath_still_serves_a_settled_file(tmp_path, monkeypatch):
+    """The guard must not disable the cache: once a file's mtime tick has
+    closed, the digest is served from the index without re-reading."""
+    _reset_stat_index()
+    f = tmp_path / "mod.py"
+    f.write_text("x = 1\n", encoding="utf-8")
+
+    # Backdate well past the granularity window so the entry is provably clean.
+    import os as _os
+    old_ns = f.stat().st_mtime_ns - 60 * 1_000_000_000
+    _os.utime(f, ns=(old_ns, old_ns))
+
+    first = file_hash(f, tmp_path)
+
+    reads = []
+    real_read_bytes = Path.read_bytes
+    monkeypatch.setattr(Path, "read_bytes",
+                        lambda self: (reads.append(self), real_read_bytes(self))[1])
+
+    second = file_hash(f, tmp_path)
+    assert second == first
+    assert reads == [], "settled file was re-read; the stat fastpath is dead"

tests/test_stat_index_portability.py

@@ -58,6 +58,19 @@ def _fail_compute(p: Path) -> int:
     raise AssertionError(f"word-count compute invoked for {p}; expected a warm stat hit")
 
 
+def _settle(path: Path) -> None:
+    """Backdate mtime past the racily-clean window so the stat fastpath is
+    allowed to serve this file.
+
+    A just-written file is deliberately never trusted: its mtime tick may still
+    be open, so a same-length rewrite could hide behind an identical
+    (size, mtime_ns). See cache._stat_sig_fresh. Any test asserting a warm stat
+    hit therefore has to settle the file first.
+    """
+    old = path.stat().st_mtime_ns - 10 * 1_000_000_000
+    os.utime(path, ns=(old, old))
+
+
 def test_cache_hits_survive_corpus_move(tmp_path, monkeypatch):
     """Run A under tmp/a, copy the corpus (with graphify-out/) to tmp/b: run B
     must be 100% warm — zero content reads, zero word-count computes, digests
@@ -69,6 +82,8 @@ def test_cache_hits_survive_corpus_move(tmp_path, monkeypatch):
     sub = a / "sub"
     sub.mkdir()
     (sub / "f2.md").write_text("hello world one two\n")
+    _settle(a / "f1.py")
+    _settle(sub / "f2.md")
 
     digests_a = {
         "f1.py": cache.file_hash(a / "f1.py", a),
@@ -122,14 +137,20 @@ def test_deleted_entries_are_pruned_on_flush(tmp_path):
 
 
 def test_legacy_absolute_index_migrates_gracefully(tmp_path, monkeypatch):
-    """A pre-#2199 index keyed by absolute paths still HITS on the unmoved
-    root, and the first flush prunes dead entries and rewrites live keys
-    relative (self-heals)."""
+    """A pre-#2199 index keyed by absolute paths still resolves to the right
+    digest on the unmoved root, and the first flush prunes dead entries and
+    rewrites live keys relative (self-heals).
+
+    A legacy entry carries no ``indexed_at_ns``, so it cannot be proven racily
+    clean and costs exactly one re-read before it is healed with a stamp — the
+    same one-time cost every entry pays on the first run after upgrading.
+    """
     _reset_stat_index()
     a = tmp_path / "a"
     a.mkdir()
     f1 = a / "f1.py"
     f1.write_text("x = 1\n")
+    _settle(f1)
     st = f1.stat()
     salt = "f1.py"
     digest = hashlib.sha256(f1.read_bytes() + b"\x00" + salt.encode()).hexdigest()
@@ -147,16 +168,20 @@ def test_legacy_absolute_index_migrates_gracefully(tmp_path, monkeypatch):
 
     reads = _count_read_bytes(monkeypatch)
     assert cache.file_hash(f1, a) == digest
-    assert reads["n"] == 0, "legacy absolute key should still serve a warm hit"
+    assert reads["n"] == 1, "unstamped legacy entry should cost exactly one re-read"
+
+    # ...and having paid it once, the healed entry is warm from then on.
+    assert cache.file_hash(f1, a) == digest
+    assert reads["n"] == 1, "healed entry should serve from the stat index"
 
-    # Force a write so the self-heal is observable (a pure warm run leaves the
-    # index clean and flush is a no-op by design).
-    cache._stat_index_dirty = True
     cache._flush_stat_index()
 
     on_disk = _read_index(a)
     assert set(on_disk) == {"f1.py"}, "dead absolute keys should be pruned"
     assert on_disk["f1.py"]["hashes"][salt] == digest
+    assert isinstance(on_disk["f1.py"].get("indexed_at_ns"), int), (
+        "self-heal should stamp the entry so it is trustable next run"
+    )
 
 
 def test_out_of_root_key_round_trips_absolute(tmp_path, monkeypatch):
@@ -165,6 +190,7 @@ def test_out_of_root_key_round_trips_absolute(tmp_path, monkeypatch):
     a.mkdir()
     outside = tmp_path / "outside.txt"
     outside.write_text("out of root\n")
+    _settle(outside)
 
     d1 = cache.file_hash(outside, a)
     cache._flush_stat_index()

tests/test_word_count_cache.py

@@ -4,11 +4,25 @@ the corpus.
 """
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
 from graphify import cache
 
 
+def _settle(path: Path) -> None:
+    """Backdate mtime past the racily-clean window so the stat fastpath is
+    allowed to serve this file.
+
+    A just-written file is deliberately never trusted: its mtime tick may still
+    be open, so a same-length rewrite could hide behind an identical
+    (size, mtime_ns). See cache._stat_sig_fresh. Any test asserting a warm stat
+    hit therefore has to settle the file first.
+    """
+    old = path.stat().st_mtime_ns - 10 * 1_000_000_000
+    os.utime(path, ns=(old, old))
+
+
 def test_word_count_cached_until_file_changes(tmp_path, monkeypatch):
     # Isolate the stat index to this tmp root.
     monkeypatch.setattr(cache, "_stat_index", {})
@@ -16,6 +30,7 @@ def test_word_count_cached_until_file_changes(tmp_path, monkeypatch):
 
     f = tmp_path / "doc.txt"
     f.write_text("one two three four five")
+    _settle(f)
 
     calls = {"n": 0}
     def compute(p: Path) -> int: