fix(versioning): 修复 shallow 快照日志重复行导致的 diff 配对错误

根因：state.jsonl 采用只追加日志，track_edit 的中间半成品快照与
make_snapshot 的完整快照各自成行；加载时每行都被回放为独立快照，
_find_snapshot_pair 简单取下标-1 作为上一快照，导致 diff 基准落在
同一条消息自己的半成品行上——半成品中缺失的已跟踪文件被全部判为
added 并按全文件计新增（如 +1928 -0）。

彻底修复（非补丁）：
- 持久化改为原子全量重写（temp + os.replace），结构性保证磁盘上
  一条 message_id 永远只有一行
- _load_state 幂等重载并在加载时折叠历史重复行（保留每消息最后一行），
  旧脏数据无需迁移即可透明愈合
- track_edit / make_snapshot 在按 state 文件加锁的临界区内执行
  重载->变更->保存，避免短生命周期实例间丢行
- _find_snapshot_pair 防御性跳过同 message_id 行
- 新增 6 个回归测试（test/test_shallow_versioning.py）

验证：新测试 6/6 通过；test_server_refactor_smoke 6/6 通过；
用 conv_20260721_175016_796 真实数据只读复算，seq8 由 +1295/-0
修正为 +796/-7（3 文件），seq10 由 +1928/-0 修正为 +592/-3（4 文件），
seq31 与历史正确记录 +482/-1 完全一致。

modules/shallow_versioning.py

@@ -6,13 +6,34 @@ Mirrors Claude Code's file-history approach:
 - A snapshot is taken per user message, referencing the latest backup version
   of every tracked file at that point in time.
 - Rewind restores the tracked files to the state recorded for a target message.
+
+Persistence model (state.jsonl)
+--------------------------------
+INVARIANT: the on-disk state file holds **exactly one row per message_id** —
+the most complete snapshot known for that message. While a message is being
+processed, its single row grows as new files get tracked (track_edit rewrites
+it); when the message finishes, make_snapshot replaces it with the final full
+snapshot. Every mutation rewrites the whole file atomically (temp + replace),
+so the invariant holds at rest, not only in memory.
+
+Legacy append-only logs may contain multiple rows per message_id (intermediate
+track_edit journals plus the final snapshot). _load_state collapses those
+duplicates keeping the last row, which heals old state files transparently;
+the next mutation rewrites the file clean permanently.
+
+Concurrency: mutations (track_edit / make_snapshot) run inside a per-state-file
+process lock as reload -> mutate -> save critical sections, so short-lived
+manager instances (one per edit / per task end) can never lose each other's
+rows. Readers only see whole files because writes go through os.replace.
 """
 
 from __future__ import annotations
 
 import hashlib
 import json
+import os
 import shutil
+import threading
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -37,6 +58,22 @@ class ShallowSnapshot:
     timestamp: str
 
 
+# Process-level locks keyed by state file path, so that separate manager
+# instances targeting the same conversation serialize their mutations.
+_STATE_LOCKS: Dict[str, threading.Lock] = {}
+_STATE_LOCKS_GUARD = threading.Lock()
+
+
+def _state_lock_for(state_file: Path) -> threading.Lock:
+    key = str(state_file)
+    with _STATE_LOCKS_GUARD:
+        lock = _STATE_LOCKS.get(key)
+        if lock is None:
+            lock = threading.Lock()
+            _STATE_LOCKS[key] = lock
+        return lock
+
+
 class ShallowVersioningManager:
     """Lightweight per-conversation backup for files edited by AI tools."""
 
@@ -50,6 +87,7 @@ class ShallowVersioningManager:
         self.save_root = (Path(data_dir).expanduser().resolve() / "save" / self.conversation_id).resolve()
         self.backup_dir = self.save_root / "shallow_backups"
         self.state_file = self.backup_dir / "state.jsonl"
+        self._state_lock = _state_lock_for(self.state_file)
         self._tracked_files: Set[str] = set()
         self._snapshots: List[ShallowSnapshot] = []
         self._snapshot_sequence = 0
@@ -59,10 +97,24 @@ class ShallowVersioningManager:
     # persistence
     # ----------------------------
     def _load_state(self) -> None:
+        """(Re)load snapshots from disk, collapsing duplicate rows per message_id.
+
+        Idempotent: resets in-memory state first, so it can be called inside
+        mutation critical sections to pick up rows written by other instances.
+        Legacy logs with multiple rows for the same message_id keep only the
+        last (most complete) row, ordered by last occurrence.
+        """
+        self._tracked_files = set()
+        self._snapshots = []
+        self._snapshot_sequence = 0
         if not self.state_file.exists():
             return
         try:
-            for raw in self.state_file.read_text(encoding="utf-8", errors="ignore").splitlines():
+            raw_lines = self.state_file.read_text(encoding="utf-8", errors="ignore").splitlines()
+        except OSError:
+            return
+        parsed: List[ShallowSnapshot] = []
+        for raw in raw_lines:
             line = raw.strip()
             if not line:
                 continue
@@ -74,17 +126,55 @@ class ShallowVersioningManager:
                 continue
             snapshot = self._deserialize_snapshot(row)
             if snapshot:
+                parsed.append(snapshot)
+        # Collapse duplicates: keep the last row per message_id.
+        last_index: Dict[str, int] = {}
+        for idx, snapshot in enumerate(parsed):
+            last_index[str(snapshot.message_id or "")] = idx
+        self._snapshots = [
+            snapshot for idx, snapshot in enumerate(parsed) if last_index[str(snapshot.message_id or "")] == idx
+        ]
+        self._snapshot_sequence = len(self._snapshots)
+        for snapshot in self._snapshots:
+            self._tracked_files.update(snapshot.tracked_file_backups.keys())
+
+    def _save_state(self) -> None:
+        """Persist the full snapshot list atomically (temp file + os.replace).
+
+        Must be called with self._state_lock held. Rewriting the whole file is
+        what enforces the one-row-per-message invariant on disk.
+        """
+        self.backup_dir.mkdir(parents=True, exist_ok=True)
+        tmp_path = self.state_file.parent / (self.state_file.name + ".tmp")
+        payload = "".join(
+            json.dumps(self._serialize_snapshot(snapshot), ensure_ascii=False) + "\n"
+            for snapshot in self._snapshots
+        )
+        tmp_path.write_text(payload, encoding="utf-8")
+        os.replace(tmp_path, self.state_file)
+
+    def _upsert_snapshot(self, snapshot: ShallowSnapshot) -> None:
+        """Insert or replace the snapshot for its message_id in memory.
+
+        Removes every existing entry with the same message_id and places the
+        new snapshot at the position of the first removed entry (appends when
+        no entry existed), preserving chronological order.
+        """
+        target_id = str(snapshot.message_id or "")
+        first_match = -1
+        for idx, existing in enumerate(self._snapshots):
+            if str(existing.message_id or "") == target_id:
+                first_match = idx
+                break
+        self._snapshots = [
+            existing for existing in self._snapshots if str(existing.message_id or "") != target_id
+        ]
+        if first_match < 0:
             self._snapshots.append(snapshot)
             self._snapshot_sequence += 1
-                    for path in snapshot.tracked_file_backups:
+        else:
-                        self._tracked_files.add(path)
+            insert_at = min(first_match, len(self._snapshots))
-        except OSError:
+            self._snapshots.insert(insert_at, snapshot)
-            pass
-
-    def _persist_snapshot(self, snapshot: ShallowSnapshot) -> None:
-        self.backup_dir.mkdir(parents=True, exist_ok=True)
-        with self.state_file.open("a", encoding="utf-8") as fh:
-            fh.write(json.dumps(self._serialize_snapshot(snapshot), ensure_ascii=False) + "\n")
 
     def _serialize_snapshot(self, snapshot: ShallowSnapshot) -> Dict[str, Any]:
         return {
@@ -204,9 +294,11 @@ class ShallowVersioningManager:
         perf_log("[ShallowVersioning] track_edit called", extra={
             "tracking_path": tracking_path,
             "message_id": message_id,
-            "already_tracked": tracking_path in self._tracked_files,
-            "snapshots_count": len(self._snapshots),
         })
+        # Critical section: reload so rows written by other (short-lived)
+        # instances are visible, then mutate and save atomically.
+        with self._state_lock:
+            self._load_state()
             if tracking_path in self._tracked_files:
                 return
 
@@ -214,31 +306,31 @@ class ShallowVersioningManager:
             backup = self._create_backup(resolved, 1)
             self._tracked_files.add(tracking_path)
 
-        # 归属到当前 message_id 对应的最新 snapshot；不存在则新建。
+            # 归属到当前 message_id 对应的 snapshot；不存在则新建。
             # 避免把本次输入的修改追加到上一个输入的 snapshot。
             now = datetime.now().isoformat()
             target_snapshot = self._find_latest_snapshot_by_message_id(message_id)
             if target_snapshot is not None:
                 target_snapshot.tracked_file_backups[tracking_path] = backup
-            self._persist_snapshot(target_snapshot)
             else:
-            snapshot = ShallowSnapshot(
+                target_snapshot = ShallowSnapshot(
                     message_id=message_id,
                     tracked_file_backups={tracking_path: backup},
                     timestamp=now,
                 )
-            self._snapshots.append(snapshot)
+            self._upsert_snapshot(target_snapshot)
-            self._snapshot_sequence += 1
+            self._save_state()
-            self._persist_snapshot(snapshot)
 
     def make_snapshot(self, message_id: str) -> Dict[str, Any]:
         """Create a snapshot for the given message, backing up changed tracked files."""
         from utils.perf_log import perf_log
         perf_log("[ShallowVersioning] make_snapshot called", extra={
             "message_id": message_id,
-            "tracked_files_count": len(self._tracked_files),
-            "snapshots_count": len(self._snapshots),
         })
+        # Critical section: reload so track_edit rows from other instances are
+        # visible before deciding what changed and which version comes next.
+        with self._state_lock:
+            self._load_state()
             tracked_file_backups: Dict[str, FileBackup] = {}
             now = datetime.now().isoformat()
 
@@ -253,35 +345,20 @@ class ShallowVersioningManager:
 
                 tracked_file_backups[tracking_path] = self._create_backup(resolved, next_version)
 
-        # Merge in backups from the latest snapshot for any files that may have been added
-        # by track_edit during the async window.
-        if self._snapshots:
-            for tracking_path, backup in self._snapshots[-1].tracked_file_backups.items():
-                if tracking_path not in tracked_file_backups:
-                    tracked_file_backups[tracking_path] = backup
-
             snapshot = ShallowSnapshot(
                 message_id=message_id,
                 tracked_file_backups=tracked_file_backups,
                 timestamp=now,
             )
-        # 如果当前 message_id 已存在 snapshot（由 track_edit 预创建），则替换它，
+            # 一条消息只保留一个快照：替换 track_edit 阶段留下的进行中的行。
-        # 避免同一个输入节点留下多个 snapshot 导致 diff 只计算最后两次之间的变化。
+            self._upsert_snapshot(snapshot)
-        existing_index = -1
-        for idx, s in enumerate(self._snapshots):
-            if str(s.message_id or "") == str(message_id or ""):
-                existing_index = idx
-        if existing_index >= 0:
-            self._snapshots[existing_index] = snapshot
-        else:
-            self._snapshots.append(snapshot)
-            self._snapshot_sequence += 1
-        self._persist_snapshot(snapshot)
 
             # Evict old snapshots while keeping the most recent MAX_SNAPSHOTS.
             if len(self._snapshots) > self.MAX_SNAPSHOTS:
                 self._snapshots = self._snapshots[-self.MAX_SNAPSHOTS :]
 
+            self._save_state()
+
         perf_log("[ShallowVersioning] make_snapshot done", extra={
             "message_id": message_id,
             "tracked_files_count": len(tracked_file_backups),
@@ -372,16 +449,26 @@ class ShallowVersioningManager:
         return insertions, deletions
 
     def _find_snapshot_pair(self, message_id: str):
-        """Return (target_snapshot, previous_snapshot) for the given message_id."""
+        """Return (target_snapshot, previous_snapshot) for the given message_id.
+
+        The previous snapshot is the nearest preceding snapshot belonging to a
+        *different* message_id, so duplicate rows of the same message can never
+        be mistaken for the diff base.
+        """
         target_snapshot: Optional[ShallowSnapshot] = None
         target_index = -1
+        target_id = str(message_id or "")
         for idx, snapshot in enumerate(self._snapshots):
-            if snapshot.message_id == message_id:
+            if str(snapshot.message_id or "") == target_id:
                 target_snapshot = snapshot
                 target_index = idx
         if not target_snapshot or target_index < 0:
             return None, None
-        previous_snapshot = self._snapshots[target_index - 1] if target_index > 0 else None
+        previous_snapshot: Optional[ShallowSnapshot] = None
+        for idx in range(target_index - 1, -1, -1):
+            if str(self._snapshots[idx].message_id or "") != target_id:
+                previous_snapshot = self._snapshots[idx]
+                break
         return target_snapshot, previous_snapshot
 
     def get_diff_stats(self, message_id: str) -> Optional[Dict[str, Any]]:
@@ -444,7 +531,6 @@ class ShallowVersioningManager:
 
             insertions, deletions = self._count_line_changes(previous_lines, target_lines)
 
-            resolved = self._resolve_file_path(tracking_path)
             status = "deleted" if target_backup.backup_file_name is None else ("added" if previous_backup is None or previous_backup.backup_file_name is None else "modified")
             result.append({
                 "path": tracking_path,

test/test_shallow_versioning.py

@@ -0,0 +1,205 @@
+"""Regression tests for shallow versioning snapshot pairing.
+
+Covers the bug where intermediate track_edit journal rows persisted to
+state.jsonl were replayed as independent snapshots, causing diffs to be
+computed against a partial snapshot of the *same* message (every tracked
+file showed as "added" with full insertions, e.g. +1928 -0).
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from modules.shallow_versioning import ShallowVersioningManager
+
+
+def _new_manager(project: Path, data: Path, conv: str = "conv_test") -> ShallowVersioningManager:
+    """Mirror production usage: a fresh short-lived instance per operation."""
+    return ShallowVersioningManager(project_path=project, data_dir=data, conversation_id=conv)
+
+
+def _read_state_rows(data: Path, conv: str = "conv_test"):
+    state_file = data / "save" / conv / "shallow_backups" / "state.jsonl"
+    if not state_file.exists():
+        return []
+    return [json.loads(line) for line in state_file.read_text(encoding="utf-8").splitlines() if line.strip()]
+
+
+class ShallowVersioningPairingTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp(prefix="shallow_ver_test_"))
+        self.project = self.tmp / "project"
+        self.data = self.tmp / "data"
+        self.project.mkdir(parents=True)
+        self.data.mkdir(parents=True)
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _write(self, rel: str, content: str) -> None:
+        path = self.project / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+
+    # ------------------------------------------------------------------
+    # realistic multi-instance flow (the original bug scenario)
+    # ------------------------------------------------------------------
+    def test_diff_pairs_with_previous_message_across_instances(self) -> None:
+        # message 1: AI creates two new files (each track_edit runs in its own instance)
+        self._write("a.txt", "a1\na2\n")
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_1")
+        self._write("b.txt", "b1\nb2\n")
+        _new_manager(self.project, self.data).track_edit("b.txt", "msg_1")
+        _new_manager(self.project, self.data).make_snapshot("msg_1")
+
+        # message 2: modify a.txt, create c.txt
+        self._write("a.txt", "a1\na2 changed\na3\n")
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_2")
+        self._write("c.txt", "c1\n")
+        _new_manager(self.project, self.data).track_edit("c.txt", "msg_2")
+        _new_manager(self.project, self.data).make_snapshot("msg_2")
+
+        # state file must hold exactly one row per message
+        rows = _read_state_rows(self.data)
+        msg_ids = [r["message_id"] for r in rows]
+        self.assertEqual(msg_ids.count("msg_1"), 1, f"duplicate msg_1 rows: {msg_ids}")
+        self.assertEqual(msg_ids.count("msg_2"), 1, f"duplicate msg_2 rows: {msg_ids}")
+
+        # lazy reader (fresh instance, like the HTTP endpoints)
+        reader = _new_manager(self.project, self.data)
+        stats = reader.get_diff_stats("msg_2") or {}
+        files = {f["path"]: f for f in reader.get_file_diff_stats("msg_2")}
+
+        # a.txt modified (+2 -1), c.txt added (+1 -0), b.txt untouched
+        self.assertEqual(set(files), {"a.txt", "c.txt"}, f"unexpected diff set: {files}")
+        self.assertEqual(files["a.txt"]["status"], "modified")
+        self.assertEqual((files["a.txt"]["insertions"], files["a.txt"]["deletions"]), (2, 1))
+        self.assertEqual(files["c.txt"]["status"], "added")
+        self.assertEqual((files["c.txt"]["insertions"], files["c.txt"]["deletions"]), (1, 0))
+        self.assertEqual(stats.get("insertions"), 3)
+        self.assertEqual(stats.get("deletions"), 1)
+
+    # ------------------------------------------------------------------
+    # legacy dirty logs: duplicates collapsed at load, healed on next save
+    # ------------------------------------------------------------------
+    def test_legacy_duplicate_rows_healed_on_load(self) -> None:
+        conv = "conv_legacy"
+        backup_dir = self.data / "save" / conv / "shallow_backups"
+        backup_dir.mkdir(parents=True)
+
+        # real backup files for a.txt v1/v2 and b.txt v1
+        (backup_dir / "a@v1").write_text("a1\na2\n", encoding="utf-8")
+        (backup_dir / "a@v2").write_text("a1\na2 changed\na3\n", encoding="utf-8")
+        (backup_dir / "b@v1").write_text("b1\n", encoding="utf-8")
+
+        def backup(name, version):
+            return {"backup_file_name": name, "version": version, "backup_time": "t"}
+
+        # hand-craft the legacy append-only pattern: partials + full per message
+        legacy_rows = [
+            {"message_id": "msg_1", "timestamp": "t", "tracked_file_backups": {"a.txt": backup("a@v1", 1)}},
+            {"message_id": "msg_1", "timestamp": "t", "tracked_file_backups": {"a.txt": backup("a@v1", 1), "b.txt": backup("b@v1", 1)}},
+            {"message_id": "msg_2", "timestamp": "t", "tracked_file_backups": {"a.txt": backup("a@v2", 2), "b.txt": backup("b@v1", 1)}},
+        ]
+        state_file = backup_dir / "state.jsonl"
+        state_file.write_text("".join(json.dumps(r) + "\n" for r in legacy_rows), encoding="utf-8")
+
+        manager = _new_manager(self.project, self.data, conv)
+        # load collapses duplicates: one snapshot per message
+        self.assertEqual(len(manager.list_snapshots()), 2)
+        # diff for msg_2 pairs against msg_1 full: only a.txt modified, NOT both added
+        files = {f["path"]: f for f in manager.get_file_diff_stats("msg_2")}
+        self.assertEqual(set(files), {"a.txt"}, f"unexpected diff set: {files}")
+        self.assertEqual(files["a.txt"]["status"], "modified")
+        self.assertEqual((files["a.txt"]["insertions"], files["a.txt"]["deletions"]), (2, 1))
+
+        # next mutation rewrites the file clean (one row per message)
+        self._write("a.txt", "a1\na2 changed\na3\na4\n")
+        _new_manager(self.project, self.data, conv).track_edit("a.txt", "msg_3")
+        _new_manager(self.project, self.data, conv).make_snapshot("msg_3")
+        rows = _read_state_rows(self.data, conv)
+        msg_ids = [r["message_id"] for r in rows]
+        self.assertEqual(len(msg_ids), len(set(msg_ids)), f"duplicates remain: {msg_ids}")
+
+    # ------------------------------------------------------------------
+    # message with exactly one newly tracked file (recorded stats were
+    # accidentally correct pre-fix; lazy view was not)
+    # ------------------------------------------------------------------
+    def test_single_new_file_message(self) -> None:
+        self._write("a.txt", "a1\n")
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_1")
+        _new_manager(self.project, self.data).make_snapshot("msg_1")
+
+        # msg_2 edits a.txt and tracks one new file b.txt
+        self._write("a.txt", "a1\na2\n")
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_2")
+        self._write("b.txt", "b1\nb2\nb3\n")
+        _new_manager(self.project, self.data).track_edit("b.txt", "msg_2")
+        _new_manager(self.project, self.data).make_snapshot("msg_2")
+
+        reader = _new_manager(self.project, self.data)
+        files = {f["path"]: f for f in reader.get_file_diff_stats("msg_2")}
+        self.assertEqual(set(files), {"a.txt", "b.txt"})
+        self.assertEqual(files["a.txt"]["status"], "modified")
+        self.assertEqual((files["a.txt"]["insertions"], files["a.txt"]["deletions"]), (1, 0))
+        self.assertEqual(files["b.txt"]["status"], "added")
+        self.assertEqual((files["b.txt"]["insertions"], files["b.txt"]["deletions"]), (3, 0))
+
+    # ------------------------------------------------------------------
+    # cross-instance visibility: mutation reloads state inside the lock
+    # ------------------------------------------------------------------
+    def test_mutation_reloads_state_from_other_instances(self) -> None:
+        self._write("a.txt", "a1\n")
+        stale = _new_manager(self.project, self.data)  # loads empty state
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_1")
+        _new_manager(self.project, self.data).make_snapshot("msg_1")
+
+        # stale instance mutates: must not wipe msg_1's row
+        self._write("b.txt", "b1\n")
+        stale.track_edit("b.txt", "msg_2")
+        _new_manager(self.project, self.data).make_snapshot("msg_2")
+
+        rows = _read_state_rows(self.data)
+        msg_ids = [r["message_id"] for r in rows]
+        self.assertIn("msg_1", msg_ids)
+        self.assertIn("msg_2", msg_ids)
+        self.assertEqual(len(msg_ids), 2)
+
+    # ------------------------------------------------------------------
+    # rewind still works
+    # ------------------------------------------------------------------
+    def test_rewind_restores_tracked_files(self) -> None:
+        self._write("a.txt", "original\n")
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_1")
+        _new_manager(self.project, self.data).make_snapshot("msg_1")
+
+        self._write("a.txt", "modified\n")
+        _new_manager(self.project, self.data).track_edit("a.txt", "msg_2")
+        _new_manager(self.project, self.data).make_snapshot("msg_2")
+
+        result = _new_manager(self.project, self.data).rewind("msg_1")
+        self.assertTrue(result.get("success"))
+        self.assertEqual((self.project / "a.txt").read_text(encoding="utf-8"), "original\n")
+
+    # ------------------------------------------------------------------
+    # get_snapshot_by_seq maps to user messages correctly
+    # ------------------------------------------------------------------
+    def test_get_snapshot_by_seq(self) -> None:
+        for i in (1, 2, 3):
+            self._write(f"f{i}.txt", f"content {i}\n")
+            _new_manager(self.project, self.data).track_edit(f"f{i}.txt", f"msg_{i}")
+            _new_manager(self.project, self.data).make_snapshot(f"msg_{i}")
+
+        reader = _new_manager(self.project, self.data)
+        self.assertEqual(reader.get_snapshot_by_seq(1).message_id, "msg_1")
+        self.assertEqual(reader.get_snapshot_by_seq(2).message_id, "msg_2")
+        self.assertEqual(reader.get_snapshot_by_seq(3).message_id, "msg_3")
+        self.assertIsNone(reader.get_snapshot_by_seq(4))
+
+
+if __name__ == "__main__":
+    unittest.main()