fix(hooks): atomic read-modify-write in SessionStats.flush() (#1493)

Replace separate _locked_read() + _locked_write() in flush() with a
single _locked_modify() helper that holds LOCK_EX for the entire
read-modify-write window.  This prevents concurrent PostToolUse
processes from clobbering each other's updates.

Add TestConcurrentFlush regression test (8 procs × 100 calls = 800
expected) that fails without the fix.

Author: JeremyDev87

packages/claude-code-plugin/hooks/lib/stats.py

@@ -96,24 +96,38 @@ def record_tool_call(self, tool_name: str, success: bool = True) -> None:
             self.flush()
 
     def flush(self) -> None:
-        """Flush accumulated in-memory stats to disk."""
+        """Flush accumulated in-memory stats to disk.
+
+        Uses _locked_modify to perform atomic read-modify-write inside a
+        single LOCK_EX window, preventing lost updates from concurrent
+        processes (#1493).
+        """
         if self._pending_count == 0:
             return
-        data = self._locked_read()
-        data["tool_count"] = data.get("tool_count", 0) + self._mem_tool_count
-        data["error_count"] = data.get("error_count", 0) + self._mem_error_count
-        tool_names = data.get("tool_names", {})
-        for name, count in self._mem_tool_names.items():
-            tool_names[name] = tool_names.get(name, 0) + count
-        data["tool_names"] = tool_names
-        # Merge hook timings
-        hook_timings = data.get("hook_timings", {})
-        for name, times in self._mem_hook_timings.items():
-            if name not in hook_timings:
-                hook_timings[name] = []
-            hook_timings[name].extend(times)
-        data["hook_timings"] = hook_timings
-        self._locked_write(data)
+
+        # Capture deltas before entering critical section
+        delta_tool_count = self._mem_tool_count
+        delta_error_count = self._mem_error_count
+        delta_tool_names = dict(self._mem_tool_names)
+        delta_hook_timings = {k: list(v) for k, v in self._mem_hook_timings.items()}
+
+        def apply_deltas(data: Dict[str, Any]) -> Dict[str, Any]:
+            data["tool_count"] = data.get("tool_count", 0) + delta_tool_count
+            data["error_count"] = data.get("error_count", 0) + delta_error_count
+            tool_names = data.get("tool_names", {})
+            for name, count in delta_tool_names.items():
+                tool_names[name] = tool_names.get(name, 0) + count
+            data["tool_names"] = tool_names
+            hook_timings = data.get("hook_timings", {})
+            for name, times in delta_hook_timings.items():
+                if name not in hook_timings:
+                    hook_timings[name] = []
+                hook_timings[name].extend(times)
+            data["hook_timings"] = hook_timings
+            return data
+
+        self._locked_modify(apply_deltas)
+
         # Reset in-memory accumulators
         self._mem_tool_count = 0
         self._mem_error_count = 0
@@ -239,6 +253,47 @@ def cleanup_stale(data_dir: str, max_age_hours: int = 24) -> None:
                 except OSError:
                     pass
 
+    def _locked_modify(self, mutator: Any) -> None:
+        """Atomic read-modify-write inside a single LOCK_EX window (#1493).
+
+        Opens the stats file with exclusive lock, reads current data,
+        applies *mutator(data) -> data*, then writes back — all without
+        releasing the lock.  This prevents the lost-update race where
+        concurrent processes each read the same baseline.
+
+        Args:
+            mutator: Callable (Dict -> Dict) that transforms the data
+                     dict in place or returns the updated dict.
+
+        Note: When HAS_FCNTL is False (non-Unix platforms), locking is
+        skipped entirely.  Concurrent flushes on such platforms may lose
+        updates — this is a known limitation documented here for
+        visibility.
+        """
+        seed: Dict[str, Any] = {
+            "session_id": self.session_id,
+            "started_at": time.time(),
+            "tool_count": 0,
+            "error_count": 0,
+            "tool_names": {},
+            "hook_timings": {},
+        }
+        try:
+            fd = os.open(self.stats_file, os.O_RDWR | os.O_CREAT)
+            with os.fdopen(fd, "r+", encoding="utf-8") as f:
+                if HAS_FCNTL:
+                    fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+                raw = f.read()
+                data = json.loads(raw) if raw else dict(seed)
+                data = mutator(data)
+                f.seek(0)
+                f.truncate()
+                json.dump(data, f)
+        except (json.JSONDecodeError, OSError):
+            # File corrupted or missing — write seed with deltas applied
+            data = mutator(dict(seed))
+            self._locked_write(data)
+
     def _locked_read(self) -> Dict[str, Any]:
         """Read stats file with file locking."""
         try:

packages/claude-code-plugin/tests/test_stats.py

@@ -250,6 +250,55 @@ def test_format_summary_no_timing_when_empty(self, stats):
         assert "⏱" not in summary
 
 
+class TestConcurrentFlush:
+    """Regression test for race condition in flush() (#1493).
+
+    Multiple processes calling record_tool_call() + flush() against the
+    same session/data_dir must not lose updates.
+    """
+
+    @staticmethod
+    def _worker(data_dir: str, session_id: str, n: int) -> None:
+        """Worker that records n tool calls and flushes each one."""
+        import sys as _sys
+        _lib = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "hooks", "lib")
+        if _lib not in _sys.path:
+            _sys.path.insert(0, _lib)
+        from stats import SessionStats as _SS
+        s = _SS(session_id=session_id, data_dir=data_dir, flush_interval=10)
+        for _ in range(n):
+            s.record_tool_call("Bash")
+            s.flush()
+
+    def test_concurrent_flush_no_lost_updates(self, data_dir):
+        """8 processes x 100 calls = 800 total. Final disk count must be 800."""
+        import multiprocessing as mp
+
+        session_id = "race-test"
+        num_workers = 8
+        calls_per_worker = 100
+        expected = num_workers * calls_per_worker
+
+        # Seed the stats file
+        SessionStats(session_id=session_id, data_dir=data_dir)
+
+        procs = [
+            mp.Process(target=self._worker, args=(data_dir, session_id, calls_per_worker))
+            for _ in range(num_workers)
+        ]
+        for p in procs:
+            p.start()
+        for p in procs:
+            p.join()
+
+        s = SessionStats(session_id=session_id, data_dir=data_dir)
+        on_disk = s._locked_read()
+        assert on_disk["tool_count"] == expected, (
+            f"Expected {expected}, got {on_disk['tool_count']} — lost updates detected"
+        )
+        assert on_disk["tool_names"]["Bash"] == expected
+
+
 class TestCleanup:
     def test_cleanup_stale_removes_old_files(self, data_dir):
         # Create a stale file