fix(sessions): list only the empty user's sessions in SqliteSessionService

src/google/adk/sessions/sqlite_session_service.py

@@ -352,7 +352,7 @@ async def list_sessions(
     sessions_list = []
     async with self._get_db_connection() as db:
       # Fetch sessions
-      if user_id:
+      if user_id is not None:
         session_rows = await db.execute_fetchall(
             "SELECT id, user_id, state, update_time FROM sessions WHERE"
             " app_name=? AND user_id=? ORDER BY update_time, user_id, id",
@@ -370,7 +370,7 @@ async def list_sessions(
 
       # Fetch user states
       user_states_map: dict[str, dict[str, Any]] = {}
-      if user_id:
+      if user_id is not None:
         user_state = await self._get_user_state(db, app_name, user_id)
         if user_state:
           user_states_map[user_id] = user_state

tests/unittests/sessions/_conformance.py

@@ -137,9 +137,6 @@ async def _make_per_agent_database(
     _Backend('in_memory_light_copy', _make_in_memory_light_copy),
     _Backend('database', _make_database),
     _Backend('sqlite', _make_sqlite),
-    # One more Redis divergence has no contract test to hang an xfail on yet:
-    # it builds its key scan pattern from a truthiness check on the user id, so
-    # an empty one lists every user's sessions.
     _Backend(
         'redis',
         _make_redis,

tests/unittests/sessions/test_session_service.py

@@ -571,6 +571,32 @@ async def test_list_sessions_all_users(session_service):
   assert sessions_all_map['session2a'].state == {'key': 'value2a'}
 
 
+@pytest.mark.asyncio
+async def test_list_sessions_with_empty_user_id_lists_only_that_user(
+    session_service,
+):
+  """An empty user id is a user id, not a request for every user."""
+  app_name = 'my_app'
+  await session_service.create_session(
+      app_name=app_name, user_id='', session_id='empty_user_session'
+  )
+  await session_service.create_session(
+      app_name=app_name,
+      user_id='other_user',
+      session_id='other_user_session',
+      state={'user:name': 'other'},
+  )
+
+  list_sessions_response = await session_service.list_sessions(
+      app_name=app_name, user_id=''
+  )
+
+  assert [s.id for s in list_sessions_response.sessions] == [
+      'empty_user_session'
+  ]
+  assert list_sessions_response.sessions[0].state == {}
+
+
 @pytest.mark.asyncio
 async def test_app_state_is_shared_by_all_users_of_app(session_service):
   app_name = 'my_app'