JeremyDev87 · JeremyDev87 · Apr 12, 2026 · Apr 12, 2026 · Apr 12, 2026
@@ -0,0 +1,52 @@
+"""Hook execution timing context manager (#1494).
+
+Records hook elapsed time via SessionStats.record_hook_timing(), flushing
+to disk so the Stop hook summary can surface the ⏱ timing report.
+"""
+import time
+from contextlib import contextmanager
+from typing import Optional
+
+
+@contextmanager
+def time_hook(
+    hook_name: str,
+    *,
+    session_id: Optional[str] = None,
+    data_dir: Optional[str] = None,
+):
+    """Context manager that records hook execution time.
+
+    Wraps hook logic with monotonic clock measurements and persists the
+    elapsed milliseconds to disk via SessionStats.  Recording failures
+    are silently swallowed so hook execution is never blocked.
+
+    Args:
+        hook_name: Claude Code event name (e.g. 'PostToolUse').
+        session_id: Explicit session ID.  If None, resolved via
+                    session_utils.get_session_id().
+        data_dir: Stats directory override (mainly for tests).
+
+    Usage::
+
+        with time_hook("PostToolUse"):
+            # ... hook logic ...
+    """
+    start = time.monotonic()
+    try:
+        yield
+    finally:
+        try:
+            elapsed_ms = (time.monotonic() - start) * 1000
+            if session_id is None:
+                from session_utils import get_session_id
+                session_id = get_session_id()
+            from stats import SessionStats
+            kwargs = {"session_id": session_id}
+            if data_dir is not None:
+                kwargs["data_dir"] = data_dir
+            stats = SessionStats(**kwargs)
+            stats.record_hook_timing(hook_name, elapsed_ms)
+            stats.flush()
+        except Exception:
+            pass  # Never block tool execution
@@ -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:

@@ -26,6 +26,13 @@ def handle_post_tool_use(data: dict):
     Records tool call stats (#825).
     Future: history tracking (#827).
     """
+    from hook_runtime import time_hook
+    with time_hook("PostToolUse"):
+        return _handle_post_tool_use(data)
+
+
+def _handle_post_tool_use(data: dict):
+    """Core PostToolUse logic, wrapped by time_hook."""
     try:
         from stats import SessionStats
 

@@ -289,7 +289,9 @@ def _handle(data: dict) -> Optional[dict]:
 @safe_main
 def handle_pre_tool_use(data: dict) -> Optional[dict]:
     """Entry point for PreToolUse hook."""
-    return _handle(data)
+    from hook_runtime import time_hook
+    with time_hook("PreToolUse"):
+        return _handle(data)
 
 
 if __name__ == "__main__":

@@ -887,6 +887,14 @@ def _check_briefing_recovery() -> None:
 
 def main():
     """Main entry point for the session start hook."""
+    _ensure_lib_path()
+    from hook_runtime import time_hook
+    with time_hook("SessionStart"):
+        return _main_inner()
+
+
+def _main_inner():
+    """Core session-start logic, wrapped by time_hook."""
     try:
         home = Path.home()
         hooks_dir = home / ".claude" / "hooks"

@@ -23,6 +23,13 @@ def handle_stop(data: dict):
 
     Finalizes session stats and returns a systemMessage summary.
     """
+    from hook_runtime import time_hook
+    with time_hook("Stop"):
+        return _handle_stop(data)
+
+
+def _handle_stop(data: dict):
+    """Core Stop logic, wrapped by time_hook."""
     try:
         from stats import SessionStats
 

@@ -51,6 +51,18 @@ def detect_mode(prompt: str) -> Optional[str]:
 
 def main():
     """Main entry point for the hook."""
+    # Ensure hooks/lib is importable for time_hook
+    _hooks_dir = os.path.dirname(os.path.abspath(__file__))
+    _lib_dir = os.path.join(_hooks_dir, "lib")
+    if _lib_dir not in sys.path:
+        sys.path.insert(0, _lib_dir)
+    from hook_runtime import time_hook
+    with time_hook("UserPromptSubmit"):
+        return _main_inner()
+
+
+def _main_inner():
+    """Core UserPromptSubmit logic, wrapped by time_hook."""
     try:
         # Read input from stdin
         input_data = json.load(sys.stdin)