Fix race condition in RunUsage.incr() with thread-safe implementation

- Add threading.Lock property to prevent race conditions in concurrent tool calls
- Fix Pydantic serialization by making _lock a property (not dataclass field)
- Maintain thread safety while allowing TypeAdapter serialization
- Add proper pickle support with __getstate__/__setstate__ methods
- Fix import order and formatting issues
- Resolves PydanticSchemaGenerationError and Pyright type errors

This addresses the race condition where concurrent tool executions could
lead to undercounted tool_calls and corrupted details dictionary.

Author: certainly-param

pydantic_ai_slim/pydantic_ai/usage.py

@@ -1,6 +1,7 @@
 from __future__ import annotations as _annotations
 
 import dataclasses
+import threading
 from copy import copy
 from dataclasses import dataclass, fields
 from typing import Annotated, Any
@@ -189,26 +190,57 @@ class RunUsage(UsageBase):
     details: dict[str, int] = dataclasses.field(default_factory=dict)
     """Any extra details returned by the model."""
 
+    """Lock to prevent race conditions when incrementing usage from concurrent tool calls."""
+
+    @property
+    def _lock(self) -> threading.Lock:
+        """Get the lock, creating it if it doesn't exist."""
+        if not hasattr(self, '__lock'):
+            self.__lock = threading.Lock()
+        return self.__lock
+
     def incr(self, incr_usage: RunUsage | RequestUsage) -> None:
         """Increment the usage in place.
 
         Args:
             incr_usage: The usage to increment by.
         """
-        if isinstance(incr_usage, RunUsage):
-            self.requests += incr_usage.requests
-            self.tool_calls += incr_usage.tool_calls
-        return _incr_usage_tokens(self, incr_usage)
+        with self._lock:
+            if isinstance(incr_usage, RunUsage):
+                self.requests += incr_usage.requests
+                self.tool_calls += incr_usage.tool_calls
+            return _incr_usage_tokens(self, incr_usage)
 
     def __add__(self, other: RunUsage | RequestUsage) -> RunUsage:
         """Add two RunUsages together.
 
         This is provided so it's trivial to sum usage information from multiple runs.
         """
         new_usage = copy(self)
-        new_usage.incr(other)
+        # Note: We can't use await here since __add__ must be synchronous
+        # But __add__ creates a new object, so there's no race condition
+        # The race condition only happens when modifying the same object concurrently
+        # The new instance will get its own lock via the property
+
+        if isinstance(other, RunUsage):
+            new_usage.requests += other.requests
+            new_usage.tool_calls += other.tool_calls
+        _incr_usage_tokens(new_usage, other)
         return new_usage
 
+    def __getstate__(self) -> dict[str, Any]:
+        """Exclude the lock from pickling."""
+        state = self.__dict__.copy()
+        # Remove any lock-related attributes since they can't be pickled
+        state.pop('_lock', None)
+        state.pop('__lock', None)
+        return state
+
+    def __setstate__(self, state: dict[str, Any]) -> None:
+        """Restore state and create a new lock."""
+        self.__dict__.update(state)
+        # The lock will be created automatically via the property
+
 
 def _incr_usage_tokens(slf: RunUsage | RequestUsage, incr_usage: RunUsage | RequestUsage) -> None:
     """Increment the usage in place.