fix: Reload the stale session in database_session_service when the update_time of storage is later than the current in memory session object

This changes the behavior of how we handle stale session, instead of reject the transaction entirely, we aggresively refresh the session.

Removed synchronous inspect(self).session calls within StorageSession.update_timestamp_tz and _dialect_name. This introspection was causing deadlocks/hangs when used with sqlalchemy.ext.asyncio in Python 3.13.

Closes issue: https://github.com/google/adk-python/issues/1733

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 861353058

tests/unittests/sessions/test_session_service.py

@@ -504,26 +504,65 @@ async def test_session_last_update_time_updates_on_event(session_service):
 
 
 @pytest.mark.asyncio
-async def test_get_session_with_config(session_service):
+async def test_append_event_to_stale_session():
+  session_service = get_session_service(
+      service_type=SessionServiceType.DATABASE
+  )
+
   app_name = 'my_app'
   user_id = 'user'
+  current_time = datetime.now().astimezone(timezone.utc).timestamp()
 
-  session = await session_service.create_session(
+  original_session = await session_service.create_session(
       app_name=app_name, user_id=user_id
   )
-  original_update_time = session.last_update_time
-
-  event = Event(invocation_id='invocation', author='user')
-  await session_service.append_event(session=session, event=event)
-
-  assert session.last_update_time >= event.timestamp
-
-  refreshed_session = await session_service.get_session(
-      app_name=app_name, user_id=user_id, session_id=session.id
+  event1 = Event(
+      invocation_id='inv1',
+      author='user',
+      timestamp=current_time + 1,
+      actions=EventActions(state_delta={'sk1': 'v1'}),
   )
-  assert refreshed_session is not None
-  assert refreshed_session.last_update_time >= event.timestamp
-  assert refreshed_session.last_update_time > original_update_time
+  await session_service.append_event(original_session, event1)
+
+  updated_session = await session_service.get_session(
+      app_name=app_name, user_id=user_id, session_id=original_session.id
+  )
+  event2 = Event(
+      invocation_id='inv2',
+      author='user',
+      timestamp=current_time + 2,
+      actions=EventActions(state_delta={'sk2': 'v2'}),
+  )
+  await session_service.append_event(updated_session, event2)
+
+  # original_session is now stale
+  assert original_session.last_update_time < updated_session.last_update_time
+  assert len(original_session.events) == 1
+  assert 'sk2' not in original_session.state
+
+  # Appending another event to stale original_session
+  event3 = Event(
+      invocation_id='inv3',
+      author='user',
+      timestamp=current_time + 3,
+      actions=EventActions(state_delta={'sk3': 'v3'}),
+  )
+  await session_service.append_event(original_session, event3)
+
+  # If we fetch session from DB, it should contain all 3 events and all state
+  # changes.
+  session_final = await session_service.get_session(
+      app_name=app_name, user_id=user_id, session_id=original_session.id
+  )
+  assert len(session_final.events) == 3
+  assert session_final.state.get('sk1') == 'v1'
+  assert session_final.state.get('sk2') == 'v2'
+  assert session_final.state.get('sk3') == 'v3'
+  assert [e.invocation_id for e in session_final.events] == [
+      'inv1',
+      'inv2',
+      'inv3',
+  ]
 
 
 @pytest.mark.asyncio