(Claude) Leave a refresh's derived rows unwritten when another refresh moved a file under it

CONTRIBUTING.md

@@ -657,7 +657,10 @@ The tool implements a SQLite-based caching system for performance:
   mark (the project's message total at 0) that its last transaction lifts and
   that otherwise makes the next run refresh the project in full. A cold
   build, of a project with nothing cached, writes per file without a spool,
-  its total at 0 until it finishes. Only the refresh writes entry rows: a run
+  its total at 0 until it finishes. Either reads the project's file records
+  as it opens and writes the derived rows only if no other refresh has
+  rewritten a file since; otherwise it leaves the total at 0 for the next
+  run. Only the refresh writes entry rows: a run
   renders what its refresh committed, so lines written to a session while it
   renders appear on the next run. A date-filtered run refreshes the
   cache as an unfiltered one would and leaves the files it wrote from its

claude_code_log/cache.py

@@ -1221,6 +1221,11 @@ def _forget_leftover_spool(conn: sqlite3.Connection) -> None:
             pass
 
 
+# A file's record identity (`CacheManager._file_record_identity`), by
+# file name, for the files of one project.
+_FileRecords = Dict[str, Tuple[Any, ...]]
+
+
 class _PendingRefresh:
     """What an `atomic_refresh` scope accumulates until its merge.
 
@@ -1232,16 +1237,32 @@ class _PendingRefresh:
     rows are no longer what the tail expected to follow, it rewrites
     the file whole from those entries, as the direct path's refused
     append falls back to a full save.
+
+    ``records`` is the project's file records as the scope opened
+    (`CacheManager._project_file_records`): the merge writes the derived
+    rows only if every file it did not write itself still carries its
+    own (`CacheManager._moved_files`).
     """
 
-    def __init__(self, spool: RefreshSpool) -> None:
+    def __init__(self, spool: RefreshSpool, records: _FileRecords) -> None:
         self.spool = spool
+        self.records = records
         self.deferred: List[Callable[[sqlite3.Connection], None]] = []
         self.appends: Dict[
             str, Tuple[Path, List[TranscriptEntry], Optional[str], os.stat_result]
         ] = {}
 
 
+class _ColdBuild:
+    """What a `CacheManager.cold_build` scope remembers until its
+    aggregates: the project's file records as it opened, and the
+    record of each file it has written since."""
+
+    def __init__(self, records: _FileRecords) -> None:
+        self.records = records
+        self.written: Dict[str, Optional[Tuple[Any, ...]]] = {}
+
+
 class CacheManager:
     """SQLite-based cache manager for Claude Code Log."""
 
@@ -1289,6 +1310,7 @@ def __init__(
         # Inside an `atomic_refresh` scope: the spool and the deferred
         # derived writes; None means every write lands at once.
         self._refresh: Optional[_PendingRefresh] = None
+        self._cold: Optional[_ColdBuild] = None
 
         self._project_id: Optional[int] = None
         if read_only:
@@ -1417,10 +1439,29 @@ def atomic_refresh(self) -> Generator[None, None, None]:
         So a refresh that stops at any point (a lock-out, a kill, a
         full disk) never records a file as current over session rows
         that predate its entries, and a reader on another connection
-        sees either the whole refresh or none of it. Two refreshes of
-        one project overlapping commit two self-consistent snapshots:
-        the later wins, and if it parsed older files its
-        ``cached_files`` rows say so.
+        sees either the whole refresh or none of it.
+
+        Another refresh of the project may commit while this one runs
+        (only `process_projects_hierarchy` holds the pass lock). A file
+        this one parsed is written with the record of that parse, so if
+        the other's was newer the file reads modified and the next run
+        reparses it. A file it took from the cache is not written, and
+        its record would read current over derived rows computed from
+        the rows it replaced: so the scope reads every file record of
+        the project as it opens, before the body's first read, and the
+        merge writes the derived rows only if each file it did not
+        write still carries that record (`_moved_files`). If one moved,
+        the merge commits the rows and leaves the project marked
+        unfinished (its total at 0), and the next run refreshes it in
+        full from what the cache then holds.
+
+        The snapshot comes after the caller's ``get_modified_files``,
+        which is only a gate: a commit in between is one the body's
+        reads already see, so the check firing on it costs a full
+        refresh and nothing else. Files the body parses come from disk
+        or from an ``entry_store`` hit, which is stamped with the source
+        file's size and mtime (`entry_store.stamp_file`), so never older
+        than the disk, nor than a record another writer took from it.
 
         Needs a single connection for the scope, so it opens a
         ``batch()`` of its own when the caller has none. Nesting joins
@@ -1445,7 +1486,8 @@ def atomic_refresh(self) -> Generator[None, None, None]:
                     "transaction; a write on it bypassed write_transaction"
                 )
             self._release_leftover_spool(conn)
-            pending = _PendingRefresh(RefreshSpool(self.db_path))
+            records = self._project_file_records(conn)
+            pending = _PendingRefresh(RefreshSpool(self.db_path), records)
             try:
                 conn.execute("ATTACH DATABASE ? AS spool", (str(pending.spool.path),))
                 try:
@@ -1627,6 +1669,16 @@ def _merge_at_once(
             if file_ids:
                 self._reindex(conn, file_ids)
                 self._update_last_updated(conn)
+            if not pending.deferred:
+                return
+            # The spooled files' records are written above, in this
+            # transaction: only the others can have moved.
+            own: Dict[str, Optional[Tuple[Any, ...]]] = {
+                f.file_name: None for f in spooled
+            }
+            if self._moved_files(conn, pending.records, own):
+                self._mark_refresh_unfinished(conn)
+                return
             for write in pending.deferred:
                 write(conn)
 
@@ -1646,34 +1698,25 @@ def _merge_per_file(
         anywhere leaves every file whole and the project marked.
 
         The last transaction first checks that each file still carries
-        the record this merge wrote. Another writer may have rewritten
-        one in between, from a newer parse; derived rows from this
-        merge's parse would then read as current over those rows, so
-        they are not written, and the mark stays for the next run.
+        the record this merge wrote, and every other file the record it
+        had as the scope opened (`_moved_files`). Another writer may
+        have rewritten one in between, from a newer parse; derived rows
+        from this merge's view would then read as current over those
+        rows, so they are not written, and the mark stays for the next
+        run.
 
         With nothing deferred there are no derived rows to fall out of
         step, and no mark either.
         """
         if pending.deferred:
             with write_transaction(conn):
                 self._mark_refresh_unfinished(conn)
-        merged: Dict[int, Optional[Tuple[Any, ...]]] = {}
+        merged: Dict[str, Optional[Tuple[Any, ...]]] = {}
         for spooled_file in spooled:
-            file_id, record = self._merge_one_file(conn, spooled_file, pending)
-            merged[file_id] = record
+            _file_id, record = self._merge_one_file(conn, spooled_file, pending)
+            merged[spooled_file.file_name] = record
         with write_transaction(conn):
-            moved = [
-                file_id
-                for file_id, record in merged.items()
-                if self._file_record_identity(conn, file_id) != record
-            ]
-            if moved:
-                logger.info(
-                    "%d file(s) of %s rewritten by another writer during the"
-                    " merge; left for the next run to derive",
-                    len(moved),
-                    self.project_path,
-                )
+            if pending.deferred and self._moved_files(conn, pending.records, merged):
                 return
             if merged:
                 self._update_last_updated(conn)
@@ -1695,6 +1738,82 @@ def has_cached_files(self) -> bool:
             ).fetchall()
         return bool(rows)
 
+    @contextmanager
+    def cold_build(self) -> Generator[None, None, None]:
+        """The scope of a refresh without a spool: a project's first build.
+
+        Its files commit one by one as they are parsed, and the project's
+        total stays 0 until the aggregates, its last write. Another
+        refresh may commit in between (over the files already written,
+        so in an `atomic_refresh` scope of its own), and the session
+        rows this build derives would then describe an older view of a
+        file whose record reads current. So the scope reads the
+        project's file records as it opens, remembers the record it
+        leaves on each file it writes, and the aggregates are written
+        only if every file still carries the one or the other
+        (`_moved_files`); otherwise the total is written 0 again, and
+        the next run refreshes the project in full.
+
+        Nesting, a read-only instance or no project row: no scope.
+        """
+        if (
+            self._cold is not None
+            or self._refresh is not None
+            or self._read_only
+            or self._project_id is None
+        ):
+            yield
+            return
+        with self.batch(), self._get_connection() as conn:
+            self._cold = _ColdBuild(self._project_file_records(conn))
+            try:
+                yield
+            finally:
+                self._cold = None
+
+    def _project_file_records(self, conn: sqlite3.Connection) -> _FileRecords:
+        """Every file record of the project, by name, from the table
+        itself (not a scope's overlay)."""
+        rows = conn.execute(
+            "SELECT file_name, source_mtime, source_size, cached_mtime,"
+            " message_count FROM main.cached_files WHERE project_id = ?",
+            (self._project_id,),
+        ).fetchall()
+        return {row[0]: tuple(row[1:]) for row in rows}
+
+    def _moved_files(
+        self,
+        conn: sqlite3.Connection,
+        records: _FileRecords,
+        own: Dict[str, Optional[Tuple[Any, ...]]],
+    ) -> List[str]:
+        """The files whose record another writer has changed since a
+        refresh read ``records``, under the caller's write lock.
+
+        ``own`` holds the files the refresh wrote itself, with the
+        record it left (None: written in this transaction, so it cannot
+        have moved); every other file must still carry its record in
+        ``records``, and a file on one side only has moved.
+        """
+        current = self._project_file_records(conn)
+        moved = sorted(
+            name
+            for name in set(current) | set(records) | set(own)
+            if (
+                own[name] is not None and current.get(name) != own[name]
+                if name in own
+                else current.get(name) != records.get(name)
+            )
+        )
+        if moved:
+            logger.info(
+                "%d file(s) of %s rewritten by another writer during the"
+                " refresh; left for the next run to derive",
+                len(moved),
+                self.project_path,
+            )
+        return moved
+
     def _mark_refresh_unfinished(self, conn: sqlite3.Connection) -> None:
         conn.execute(
             "UPDATE projects SET total_message_count = 0 WHERE id = ?",
@@ -2258,6 +2377,10 @@ def save_cached_entries(
             file_id = self._replace_file_rows(
                 conn, jsonl_path, serialized_entries, source_stat, subagents_fp
             )
+            if self._cold is not None:
+                self._cold.written[jsonl_path.name] = self._file_record_identity(
+                    conn, file_id
+                )
 
             # Keep the full-archive search index in step with the rows we
             # just replaced, inside the same transaction. This is a no-op
@@ -2418,7 +2541,7 @@ def extend_cached_entries(
             # refused had it seen them. A refusal writes nothing, so the
             # commit it ends with is empty. Inside a `batch()` scope already
             # in a transaction, that one covers it.
-            return self._append_under_lock(
+            appended_ok = self._append_under_lock(
                 conn,
                 jsonl_path,
                 all_entries,
@@ -2427,6 +2550,11 @@ def extend_cached_entries(
                 source_stat,
                 subagents_fp,
             )
+            if appended_ok and self._cold is not None:
+                self._cold.written[jsonl_path.name] = self._file_record_identity(
+                    conn, self._file_id_under_lock(conn, jsonl_path.name)
+                )
+            return appended_ok
 
     def _append_under_lock(
         self,
@@ -2600,6 +2728,20 @@ def write(conn: sqlite3.Connection) -> None:
                 ),
             )
 
+        cold = self._cold
+        if cold is not None and self._refresh is None:
+            # A cold build's last write: if another writer moved a file
+            # since the build read or wrote it, the session rows already
+            # written may describe an older view; leave the project
+            # unfinished for the next run instead (`cold_build`).
+            def guarded(conn: sqlite3.Connection) -> None:
+                if self._moved_files(conn, cold.records, cold.written):
+                    self._mark_refresh_unfinished(conn)
+                else:
+                    write(conn)
+
+            self._write_or_defer(guarded)
+            return
         self._write_or_defer(write)
 
     def save_session_sidecar(self, sidecar: SessionSidecar) -> None:

claude_code_log/converter.py

@@ -4913,10 +4913,15 @@ def ensure_fresh_cache_detailed(
         # next run completes it. Its files commit one by one as they
         # are parsed, which keeps the lock hold of a large project's
         # first build to its largest file, and its disk to the cache.
+        #
+        # Either scope reads the project's file records as it opens,
+        # after the gate above and before any read the derived rows
+        # come from, and writes them only if no other refresh has moved
+        # a file since (`CacheManager._moved_files`).
         scope = (
             cache_manager.atomic_refresh()
             if cache_manager.has_cached_files()
-            else contextlib.nullcontext()
+            else cache_manager.cold_build()
         )
         with scope:
             # Streaming stage 4: when the update is driven purely by

dev-docs/application_model.md

@@ -384,9 +384,7 @@ the pass's in-pass retry, redoes it, re-parsing the modified files. Only
 a process killed mid-parse leaves a spool behind, and a pass removes the
 spools of dead processes before it starts, under its lock. A
 reader on another connection sees either the whole refresh or none of
-it, and two refreshes of one project overlapping commit two
-self-consistent snapshots: the later wins, and if it parsed older files
-its `cached_files` rows say so and the next run repairs it. Inside the
+it. Inside the
 scope, two TEMP views named `cached_files` and `messages` shadow the
 tables for every read on the scope's connection (the table minus the
 rows the spooled files replace, `UNION ALL` the spool's rows), because
@@ -415,21 +413,45 @@ cost. Past it, `_merge_per_file` commits per file between two more
 transactions: the first marks the project unfinished by zeroing its
 message total (016's heal: the next run refreshes it in full, from the
 cached entries), and the last writes the derived rows, whose aggregates
-lift the mark, but only if every file it merged still carries the
-record it wrote: a file another writer rewrote in between, from a newer
-parse, would otherwise read as current under derived rows from the
-older one, so the mark stays and the next run derives from what the
-cache holds. Either way the merge runs with a 64 MB page cache
+lift the mark, but only if no file moved (below). Either way the merge
+runs with a 64 MB page cache
 (`MERGE_CACHE_SIZE_KIB`; SQLite's default is 2 MB, which the indexes
 thrash: 6.1 s against 3.4 s for the copy above), set for the merge and
 given back, so no long-lived writing connection keeps it. A **cold
 build**, of a project with no cached file (a new project, or one whose
-rows a version change discarded), opens no scope at all: there are no
-derived rows to fall out of step with, and the project's total is 0
-until the aggregates, the refresh's last write, so a build that stops
-reads as unfinished and the next run completes it. Its files commit one
-by one as they are parsed, which keeps a large project's first build to
-its largest file's hold and adds no spool to its disk.
+rows a version change discarded), takes no spool (`CacheManager.cold_build`):
+there are no derived rows to fall out of step with, and the project's
+total is 0 until the aggregates, the refresh's last write, so a build
+that stops reads as unfinished and the next run completes it. Its files
+commit one by one as they are parsed, which keeps a large project's
+first build to its largest file's hold and adds no spool to its disk.
+
+**Two refreshes of one project overlapping.** Only a hierarchy pass
+holds the pass lock; a single-project convert, `watch <dir>`, a
+single-session file or the TUI's export refresh beside it. A file a
+refresh parsed it writes with the record of that parse, so if another
+refresh committed a newer one in between, the file reads modified and
+the next run reparses it. A file it took from the cache it does not
+write, and if the other rewrote that one before this refresh's derived
+rows landed, those rows (an incremental refresh's total included, which
+it computes as a delta over the total it read) would describe the rows
+the other replaced, under a record that reads current, and nothing
+would recompute them. So both scopes read every file record of the
+project as they open, after the modified-file gate (which only decides
+whether to refresh: a commit before the snapshot is one the scope's
+reads see) and before any read the derived rows come from, and write
+the derived rows only if every file still carries its record: the one
+the refresh left on the files it wrote (none to compare in the
+one-transaction merge, which writes them in the same transaction), the
+one in the snapshot on every other, a file on one side only counting as
+moved (`_moved_files`). Otherwise the derived rows are not written and
+the project's total is left at 0, so the next run refreshes it in full
+from what the cache then holds; the one-transaction merge still commits
+the rows it spooled. A commit after the snapshot that the refresh's
+reads did see trips the check as well; it costs that one full refresh. A
+file parsed from an `entry_store` hit is no exception: the store stamps
+its entries with the source file's size and mtime, so a hit is never
+older than the disk.
 
 **A run renders what its refresh committed.** Only the refresh writes
 entry rows; the render reads them and writes none. A session file can