fix(sessions): skip unnecessary FOR UPDATE lock on app/user state rows

DatabaseSessionService.append_event() unconditionally acquires SELECT ...
FOR UPDATE on both app_states and user_states tables, even when the event
carries no state delta for those scopes.

Since app_states is keyed by app_name alone, all concurrent append_event
calls within the same app serialize on this single row lock, even when
they only carry session-scoped state (the vast majority of events).

Fix: pre-analyze the event's state_delta before acquiring locks and only
use FOR UPDATE when the corresponding scope actually has changes. This
also avoids a redundant call to extract_state_delta later in the method.

Fixes #4655

Author: giulio-leone

src/google/adk/sessions/database_session_service.py

@@ -550,11 +550,25 @@ async def append_event(self, session: Session, event: Event) -> Event:
         if storage_session is None:
           raise ValueError(f"Session {session.id} not found.")
 
+        # Pre-analyze state deltas to determine which scopes actually need
+        # write locks. Most events carry only session-scoped state (or no
+        # state at all), so acquiring FOR UPDATE on app_states / user_states
+        # unnecessarily serializes all concurrent append_event calls.
+        has_app_delta = False
+        has_user_delta = False
+        state_deltas = None
+        if event.actions and event.actions.state_delta:
+          state_deltas = _session_util.extract_state_delta(
+              event.actions.state_delta
+          )
+          has_app_delta = bool(state_deltas.get("app"))
+          has_user_delta = bool(state_deltas.get("user"))
+
         storage_app_state = await _select_required_state(
             sql_session=sql_session,
             state_model=schema.StorageAppState,
             predicates=(schema.StorageAppState.app_name == session.app_name,),
-            use_row_level_locking=use_row_level_locking,
+            use_row_level_locking=use_row_level_locking and has_app_delta,
             missing_message=(
                 "App state missing for app_name="
                 f"{session.app_name!r}. Session state tables should be "
@@ -568,7 +582,7 @@ async def append_event(self, session: Session, event: Event) -> Event:
                 schema.StorageUserState.app_name == session.app_name,
                 schema.StorageUserState.user_id == session.user_id,
             ),
-            use_row_level_locking=use_row_level_locking,
+            use_row_level_locking=use_row_level_locking and has_user_delta,
             missing_message=(
                 "User state missing for app_name="
                 f"{session.app_name!r}, user_id={session.user_id!r}. "
@@ -599,11 +613,8 @@ async def append_event(self, session: Session, event: Event) -> Event:
           storage_events = [e async for e in result]
           session.events = [e.to_event() for e in storage_events]
 
-        # Extract state delta
-        if event.actions and event.actions.state_delta:
-          state_deltas = _session_util.extract_state_delta(
-              event.actions.state_delta
-          )
+        # Apply state deltas (already extracted above for lock scoping)
+        if state_deltas is not None:
           app_state_delta = state_deltas["app"]
           user_state_delta = state_deltas["user"]
           session_state_delta = state_deltas["session"]