From 0b0d5a3c6198161d66de7ba285c9ed73e0b6a35e Mon Sep 17 00:00:00 2001 From: weimch Date: Wed, 8 Jul 2026 15:27:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(sessions):=20list=5Fsessions=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=20user=5Fid=20=E4=B8=BA=20None=20=E6=97=B6=E8=8E=B7?= =?UTF-8?q?=E5=8F=96=20app=5Fname=20=E4=B8=8B=E5=85=A8=E9=83=A8=20session?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit list_sessions 的 user_id 参数改为可选，传 None 时返回指定 app 下 所有用户的会话列表（不含 events）。InMemory/SQL/Redis/EvalSessionService 实现统一调整，并补充中英文文档用法说明及单元测试。 --- docs/mkdocs/en/session.md | 12 ++- docs/mkdocs/zh/session.md | 12 ++- .../test_in_memory_session_service.py | 73 +++++++++++++++---- tests/sessions/test_redis_session_service.py | 70 +++++++++++------- tests/sessions/test_sql_session_service.py | 65 ++++++++++++----- trpc_agent_sdk/abc/_session_service.py | 15 +++- .../evaluation/_eval_session_service.py | 2 +- .../sessions/_in_memory_session_service.py | 36 +++++---- .../sessions/_redis_session_service.py | 26 ++++++- .../sessions/_sql_session_service.py | 11 ++- 10 files changed, 232 insertions(+), 90 deletions(-) diff --git a/docs/mkdocs/en/session.md b/docs/mkdocs/en/session.md index c75831da..899bf0cc 100644 --- a/docs/mkdocs/en/session.md +++ b/docs/mkdocs/en/session.md @@ -109,11 +109,18 @@ session = await session_service.get_session( **List Sessions**: ```python +# Specify user_id: returns all sessions for that user (without events) session_list = await session_service.list_sessions( app_name="my_app", user_id="user_001" ) -# Returns ListSessionsResponse, containing all sessions for the user (without events) + +# user_id=None: returns sessions across all users for the app +all_session_list = await session_service.list_sessions( + app_name="my_app", + user_id=None +) +# Returns ListSessionsResponse, containing matching sessions (without events) ``` **Delete Session**: @@ -128,7 +135,7 @@ await session_service.delete_session( **Implementation Logic** (`_base_session_service.py`): - `create_session`: Creates a session, separates and stores app/user/session state - `get_session`: Retrieves a session, merges app/user/session state, applies event filtering -- `list_sessions`: Lists sessions (excludes events to reduce data transfer) +- `list_sessions`: Lists sessions (excludes events to reduce data transfer); passing `user_id=None` returns sessions across all users for the app - `delete_session`: Deletes a session and its associated data --- @@ -611,6 +618,7 @@ session = await session_service.get_session( ) # List existing Sessions +# Specify user_id to return that user's sessions; user_id=None returns sessions across all users for the app session_list = await session_service.list_sessions( app_name=app_name, user_id=user_id diff --git a/docs/mkdocs/zh/session.md b/docs/mkdocs/zh/session.md index 3909973c..c13fc09a 100644 --- a/docs/mkdocs/zh/session.md +++ b/docs/mkdocs/zh/session.md @@ -109,11 +109,18 @@ session = await session_service.get_session( **列出会话**： ```python +# 指定 user_id：返回该用户的所有会话（不含 events） session_list = await session_service.list_sessions( app_name="my_app", user_id="user_001" ) -# 返回 ListSessionsResponse，包含该用户的所有会话（不含 events） + +# user_id 为 None：返回该 app 下所有用户的会话 +all_session_list = await session_service.list_sessions( + app_name="my_app", + user_id=None +) +# 返回 ListSessionsResponse，包含符合条件的所有会话（不含 events） ``` **删除会话**： @@ -128,7 +135,7 @@ await session_service.delete_session( **实现逻辑**（`_base_session_service.py`）： - `create_session`：创建会话，分离并存储 app/user/session 状态 - `get_session`：获取会话，合并 app/user/session 状态，应用事件过滤 -- `list_sessions`：列出会话列表（不包含 events，减少数据传输） +- `list_sessions`：列出会话列表（不包含 events，减少数据传输）；`user_id` 传 `None` 时返回该 app 下所有用户的会话 - `delete_session`：删除会话及其关联数据 --- @@ -611,6 +618,7 @@ session = await session_service.get_session( ) # 列出存在的 Session +# 指定 user_id 返回该用户的会话；user_id=None 返回该 app 下所有用户的会话 session_list = await session_service.list_sessions( app_name=app_name, user_id=user_id diff --git a/tests/sessions/test_in_memory_session_service.py b/tests/sessions/test_in_memory_session_service.py index 6b33bcd2..174daa31 100644 --- a/tests/sessions/test_in_memory_session_service.py +++ b/tests/sessions/test_in_memory_session_service.py @@ -34,8 +34,9 @@ def _make_session_config(ttl_seconds=0, cleanup_interval=0.0, enable_ttl=False, **kwargs): config = SessionServiceConfig(**kwargs) if enable_ttl: - config.ttl = SessionServiceConfig.create_ttl_config( - enable=True, ttl_seconds=ttl_seconds, cleanup_interval_seconds=cleanup_interval) + config.ttl = SessionServiceConfig.create_ttl_config(enable=True, + ttl_seconds=ttl_seconds, + cleanup_interval_seconds=cleanup_interval) else: config.clean_ttl_config() return config @@ -56,7 +57,9 @@ def _make_event(author="agent", text="hello", state_delta=None, partial=False): # SessionWithTTL # --------------------------------------------------------------------------- + class TestSessionWithTTL: + def test_update_and_get(self): session = Session(id="s1", app_name="app", user_id="user", save_key="k") wrapper = SessionWithTTL(session=session) @@ -85,7 +88,9 @@ def test_get_non_expired(self): # StateWithTTL # --------------------------------------------------------------------------- + class TestStateWithTTL: + def test_update(self): wrapper = StateWithTTL() result = wrapper.update({"key": "value"}) @@ -119,7 +124,9 @@ def test_update_expired_resets_then_updates(self): # InMemorySessionService — create_session # --------------------------------------------------------------------------- + class TestInMemoryCreateSession: + async def test_create_basic_session(self): svc = InMemorySessionService(session_config=_make_session_config()) session = await svc.create_session(app_name="app", user_id="user") @@ -143,13 +150,13 @@ async def test_create_with_whitespace_id_generates_uuid(self): async def test_create_with_state(self): svc = InMemorySessionService(session_config=_make_session_config()) - session = await svc.create_session( - app_name="app", user_id="user", - state={ - "session_key": "session_val", - f"{State.APP_PREFIX}app_key": "app_val", - f"{State.USER_PREFIX}user_key": "user_val", - }) + session = await svc.create_session(app_name="app", + user_id="user", + state={ + "session_key": "session_val", + f"{State.APP_PREFIX}app_key": "app_val", + f"{State.USER_PREFIX}user_key": "user_val", + }) assert session.state["session_key"] == "session_val" assert session.state[f"{State.APP_PREFIX}app_key"] == "app_val" assert session.state[f"{State.USER_PREFIX}user_key"] == "user_val" @@ -168,7 +175,9 @@ async def test_create_returns_deep_copy(self): # InMemorySessionService — get_session # --------------------------------------------------------------------------- + class TestInMemoryGetSession: + async def test_get_existing_session(self): svc = InMemorySessionService(session_config=_make_session_config()) created = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -185,13 +194,14 @@ async def test_get_nonexistent_session(self): async def test_get_returns_merged_state(self): svc = InMemorySessionService(session_config=_make_session_config()) - await svc.create_session( - app_name="app", user_id="user", session_id="s1", - state={ - "sk": "sv", - f"{State.APP_PREFIX}ak": "av", - f"{State.USER_PREFIX}uk": "uv", - }) + await svc.create_session(app_name="app", + user_id="user", + session_id="s1", + state={ + "sk": "sv", + f"{State.APP_PREFIX}ak": "av", + f"{State.USER_PREFIX}uk": "uv", + }) result = await svc.get_session(app_name="app", user_id="user", session_id="s1") assert result.state["sk"] == "sv" assert result.state[f"{State.APP_PREFIX}ak"] == "av" @@ -212,7 +222,9 @@ async def test_get_returns_deep_copy(self): # InMemorySessionService — list_sessions # --------------------------------------------------------------------------- + class TestInMemoryListSessions: + async def test_list_empty(self): svc = InMemorySessionService(session_config=_make_session_config()) result = await svc.list_sessions(app_name="app", user_id="user") @@ -251,12 +263,31 @@ async def test_list_nonexistent_user(self): assert result.sessions == [] await svc.close() + async def test_list_all_users_when_user_id_none(self): + svc = InMemorySessionService(session_config=_make_session_config()) + await svc.create_session(app_name="app", user_id="user1", session_id="s1") + await svc.create_session(app_name="app", user_id="user2", session_id="s2") + result = await svc.list_sessions(app_name="app", user_id=None) + ids = sorted(s.id for s in result.sessions) + assert ids == ["s1", "s2"] + await svc.close() + + async def test_list_all_users_filtered_by_app(self): + svc = InMemorySessionService(session_config=_make_session_config()) + await svc.create_session(app_name="app1", user_id="user1", session_id="s1") + await svc.create_session(app_name="app2", user_id="user1", session_id="s2") + result = await svc.list_sessions(app_name="app1", user_id=None) + assert [s.id for s in result.sessions] == ["s1"] + await svc.close() + # --------------------------------------------------------------------------- # InMemorySessionService — delete_session # --------------------------------------------------------------------------- + class TestInMemoryDeleteSession: + async def test_delete_existing(self): svc = InMemorySessionService(session_config=_make_session_config()) await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -275,7 +306,9 @@ async def test_delete_nonexistent(self): # InMemorySessionService — append_event # --------------------------------------------------------------------------- + class TestInMemoryAppendEvent: + async def test_append_basic(self): svc = InMemorySessionService(session_config=_make_session_config()) session = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -351,7 +384,9 @@ async def test_append_updates_conversation_count(self): # InMemorySessionService — update_session # --------------------------------------------------------------------------- + class TestInMemoryUpdateSession: + async def test_update_existing(self): svc = InMemorySessionService(session_config=_make_session_config()) session = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -384,7 +419,9 @@ async def test_update_nonexistent_session(self): # InMemorySessionService — cleanup # --------------------------------------------------------------------------- + class TestInMemoryCleanupExpired: + async def test_cleanup_removes_expired_sessions(self): config = _make_session_config(enable_ttl=True, ttl_seconds=1, cleanup_interval=3600.0) svc = InMemorySessionService(session_config=config) @@ -436,7 +473,9 @@ async def test_cleanup_nothing_expired(self): # InMemorySessionService — cleanup task lifecycle # --------------------------------------------------------------------------- + class TestInMemoryCleanupTask: + def test_no_cleanup_task_when_ttl_disabled(self): svc = InMemorySessionService(session_config=_make_session_config()) assert svc._InMemorySessionService__cleanup_task is None @@ -490,7 +529,9 @@ async def test_cleanup_loop_handles_error(self): # InMemorySessionService — internal helpers # --------------------------------------------------------------------------- + class TestInMemoryInternalHelpers: + async def test_get_app_state_nonexistent(self): svc = InMemorySessionService(session_config=_make_session_config()) assert svc._get_app_state("nonexistent") == {} diff --git a/tests/sessions/test_redis_session_service.py b/tests/sessions/test_redis_session_service.py index 963fe218..8269b186 100644 --- a/tests/sessions/test_redis_session_service.py +++ b/tests/sessions/test_redis_session_service.py @@ -27,15 +27,12 @@ from trpc_agent_sdk.types import Content, EventActions, Part, State -def _make_config(ttl_seconds=0, - cleanup_interval=0.0, - enable_ttl=False, - max_events=0, - store_historical_events=False): +def _make_config(ttl_seconds=0, cleanup_interval=0.0, enable_ttl=False, max_events=0, store_historical_events=False): config = SessionServiceConfig(max_events=max_events, store_historical_events=store_historical_events) if enable_ttl: - config.ttl = SessionServiceConfig.create_ttl_config( - enable=True, ttl_seconds=ttl_seconds, cleanup_interval_seconds=cleanup_interval) + config.ttl = SessionServiceConfig.create_ttl_config(enable=True, + ttl_seconds=ttl_seconds, + cleanup_interval_seconds=cleanup_interval) else: config.clean_ttl_config() return config @@ -117,6 +114,7 @@ def _create_service(config=None): class TestRedisCreateSession: + async def test_create_basic(self): svc = _create_service() session = await svc.create_session(app_name="app", user_id="user") @@ -133,13 +131,13 @@ async def test_create_with_custom_id(self): async def test_create_with_state(self): svc = _create_service() - session = await svc.create_session( - app_name="app", user_id="user", - state={ - "sk": "sv", - f"{State.APP_PREFIX}ak": "av", - f"{State.USER_PREFIX}uk": "uv", - }) + session = await svc.create_session(app_name="app", + user_id="user", + state={ + "sk": "sv", + f"{State.APP_PREFIX}ak": "av", + f"{State.USER_PREFIX}uk": "uv", + }) assert session.state["sk"] == "sv" assert session.state[f"{State.APP_PREFIX}ak"] == "av" assert session.state[f"{State.USER_PREFIX}uk"] == "uv" @@ -154,6 +152,7 @@ async def test_create_with_whitespace_id(self): class TestRedisGetSession: + async def test_get_existing(self): svc = _create_service() await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -170,13 +169,14 @@ async def test_get_nonexistent(self): async def test_get_with_merged_state(self): svc = _create_service() - await svc.create_session( - app_name="app", user_id="user", session_id="s1", - state={ - "sk": "sv", - f"{State.APP_PREFIX}ak": "av", - f"{State.USER_PREFIX}uk": "uv", - }) + await svc.create_session(app_name="app", + user_id="user", + session_id="s1", + state={ + "sk": "sv", + f"{State.APP_PREFIX}ak": "av", + f"{State.USER_PREFIX}uk": "uv", + }) result = await svc.get_session(app_name="app", user_id="user", session_id="s1") assert result.state["sk"] == "sv" assert result.state[f"{State.APP_PREFIX}ak"] == "av" @@ -185,6 +185,7 @@ async def test_get_with_merged_state(self): class TestRedisListSessions: + async def test_list_empty(self): svc = _create_service() result = await svc.list_sessions(app_name="app", user_id="user") @@ -210,8 +211,18 @@ async def test_list_sessions_have_no_events(self): assert s.historical_events == [] await svc.close() + async def test_list_all_users_when_user_id_none(self): + svc = _create_service() + await svc.create_session(app_name="app", user_id="user1", session_id="s1") + await svc.create_session(app_name="app", user_id="user2", session_id="s2") + result = await svc.list_sessions(app_name="app", user_id=None) + ids = sorted(s.id for s in result.sessions) + assert ids == ["s1", "s2"] + await svc.close() + class TestRedisDeleteSession: + async def test_delete_existing(self): svc = _create_service() await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -227,6 +238,7 @@ async def test_delete_nonexistent(self): class TestRedisAppendEvent: + async def test_append_basic(self): svc = _create_service() session = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -264,12 +276,13 @@ async def test_append_with_state_delta(self): async def test_append_does_not_persist_merged_or_temp_state_in_session_json(self): svc = _create_service() session = await svc.create_session(app_name="app", user_id="user", session_id="s1") - event = _make_event(state_delta={ - "session_key": "sv", - f"{State.APP_PREFIX}app_key": "av", - f"{State.USER_PREFIX}user_key": "uv", - f"{State.TEMP_PREFIX}temp_key": "tv", - }) + event = _make_event( + state_delta={ + "session_key": "sv", + f"{State.APP_PREFIX}app_key": "av", + f"{State.USER_PREFIX}user_key": "uv", + f"{State.TEMP_PREFIX}temp_key": "tv", + }) await svc.append_event(session, event) @@ -306,6 +319,7 @@ async def test_append_persists_filtered_active_and_historical_events(self): class TestRedisUpdateSession: + async def test_update_existing(self): svc = _create_service() session = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -323,6 +337,7 @@ async def test_update_nonexistent(self): class TestRedisRefreshTtl: + async def test_refresh_ttl_disabled(self): svc = _create_service(config=_make_config()) mock_redis = MagicMock() @@ -340,6 +355,7 @@ async def test_refresh_ttl_enabled(self): class TestRedisClose: + async def test_close(self): svc = _create_service() await svc.close() diff --git a/tests/sessions/test_sql_session_service.py b/tests/sessions/test_sql_session_service.py index 30b3df1e..e1730ec6 100644 --- a/tests/sessions/test_sql_session_service.py +++ b/tests/sessions/test_sql_session_service.py @@ -43,8 +43,9 @@ def _make_config(ttl_seconds=0, max_events=max_events, store_historical_events=store_historical_events) if enable_ttl: - config.ttl = SessionServiceConfig.create_ttl_config( - enable=True, ttl_seconds=ttl_seconds, cleanup_interval_seconds=cleanup_interval) + config.ttl = SessionServiceConfig.create_ttl_config(enable=True, + ttl_seconds=ttl_seconds, + cleanup_interval_seconds=cleanup_interval) else: config.clean_ttl_config() return config @@ -81,7 +82,9 @@ async def _create_service(config=None): # SessionStorageEvent — from_event / to_event # --------------------------------------------------------------------------- + class TestSessionStorageEvent: + def test_from_event_basic(self): session = Session(id="s1", app_name="app", user_id="user", save_key="k") event = _make_event(author="user", text="hello world") @@ -125,7 +128,10 @@ def test_to_event_drops_legacy_empty_parts(self): long_running_tool_ids=set(), timestamp=datetime.now(), model_flags=1, - content={"parts": [{}], "role": "model"}, + content={ + "parts": [{}], + "role": "model" + }, ) event = storage_event.to_event() assert event.content is None @@ -157,7 +163,9 @@ def test_long_running_tool_ids_empty(self): # SqlSessionService — create_session # --------------------------------------------------------------------------- + class TestSqlCreateSession: + async def test_create_basic(self): svc = await _create_service() session = await svc.create_session(app_name="app", user_id="user") @@ -174,13 +182,14 @@ async def test_create_with_custom_id(self): async def test_create_with_state(self): svc = await _create_service() - session = await svc.create_session( - app_name="app", user_id="user", session_id="s1", - state={ - "sk": "sv", - f"{State.APP_PREFIX}ak": "av", - f"{State.USER_PREFIX}uk": "uv", - }) + session = await svc.create_session(app_name="app", + user_id="user", + session_id="s1", + state={ + "sk": "sv", + f"{State.APP_PREFIX}ak": "av", + f"{State.USER_PREFIX}uk": "uv", + }) assert session.state["sk"] == "sv" assert session.state[f"{State.APP_PREFIX}ak"] == "av" assert session.state[f"{State.USER_PREFIX}uk"] == "uv" @@ -205,7 +214,9 @@ async def test_create_with_whitespace_id(self): # SqlSessionService — get_session # --------------------------------------------------------------------------- + class TestSqlGetSession: + async def test_get_existing(self): svc = await _create_service() await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -242,13 +253,14 @@ async def test_get_with_num_recent_events(self): async def test_get_with_merged_state(self): svc = await _create_service() - await svc.create_session( - app_name="app", user_id="user", session_id="s1", - state={ - "sk": "sv", - f"{State.APP_PREFIX}ak": "av", - f"{State.USER_PREFIX}uk": "uv", - }) + await svc.create_session(app_name="app", + user_id="user", + session_id="s1", + state={ + "sk": "sv", + f"{State.APP_PREFIX}ak": "av", + f"{State.USER_PREFIX}uk": "uv", + }) result = await svc.get_session(app_name="app", user_id="user", session_id="s1") assert result.state["sk"] == "sv" assert result.state[f"{State.APP_PREFIX}ak"] == "av" @@ -260,7 +272,9 @@ async def test_get_with_merged_state(self): # SqlSessionService — list_sessions # --------------------------------------------------------------------------- + class TestSqlListSessions: + async def test_list_empty(self): svc = await _create_service() result = await svc.list_sessions(app_name="app", user_id="user") @@ -286,12 +300,23 @@ async def test_list_sessions_have_no_events(self): assert s.historical_events == [] await svc.close() + async def test_list_all_users_when_user_id_none(self): + svc = await _create_service() + await svc.create_session(app_name="app", user_id="user1", session_id="s1") + await svc.create_session(app_name="app", user_id="user2", session_id="s2") + result = await svc.list_sessions(app_name="app", user_id=None) + ids = sorted(s.id for s in result.sessions) + assert ids == ["s1", "s2"] + await svc.close() + # --------------------------------------------------------------------------- # SqlSessionService — delete_session # --------------------------------------------------------------------------- + class TestSqlDeleteSession: + async def test_delete_existing(self): svc = await _create_service() await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -305,7 +330,9 @@ async def test_delete_existing(self): # SqlSessionService — append_event # --------------------------------------------------------------------------- + class TestSqlAppendEvent: + async def test_append_basic(self): svc = await _create_service() session = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -387,7 +414,9 @@ async def test_append_persists_historical_events(self): # SqlSessionService — update_session # --------------------------------------------------------------------------- + class TestSqlUpdateSession: + async def test_update_existing(self): svc = await _create_service() session = await svc.create_session(app_name="app", user_id="user", session_id="s1") @@ -410,7 +439,9 @@ async def test_update_nonexistent(self): # SqlSessionService — cleanup # --------------------------------------------------------------------------- + class TestSqlCleanup: + def test_no_cleanup_task_when_disabled(self): config = _make_config() with patch("trpc_agent_sdk.sessions._sql_session_service.SqlStorage"): diff --git a/trpc_agent_sdk/abc/_session_service.py b/trpc_agent_sdk/abc/_session_service.py index 15107b2e..419fac06 100644 --- a/trpc_agent_sdk/abc/_session_service.py +++ b/trpc_agent_sdk/abc/_session_service.py @@ -92,8 +92,19 @@ async def get_session( """Gets a session.""" @abstractmethod - async def list_sessions(self, *, app_name: str, user_id: str) -> ListSessionsResponse: - """Lists all the sessions.""" + async def list_sessions( + self, + *, + app_name: str, + user_id: Optional[str] = None, + ) -> ListSessionsResponse: + """Lists all the sessions. + + Args: + app_name: the name of the app. + user_id: the id of the user. When None, lists sessions across all + users for the given app. + """ @abstractmethod async def delete_session(self, *, app_name: str, user_id: str, session_id: str) -> None: diff --git a/trpc_agent_sdk/evaluation/_eval_session_service.py b/trpc_agent_sdk/evaluation/_eval_session_service.py index d8c97b82..d9e231db 100644 --- a/trpc_agent_sdk/evaluation/_eval_session_service.py +++ b/trpc_agent_sdk/evaluation/_eval_session_service.py @@ -71,7 +71,7 @@ async def get_session( ) @override - async def list_sessions(self, *, app_name: str, user_id: str): + async def list_sessions(self, *, app_name: str, user_id: Optional[str] = None): return await self._inner.list_sessions(app_name=app_name, user_id=user_id) @override diff --git a/trpc_agent_sdk/sessions/_in_memory_session_service.py b/trpc_agent_sdk/sessions/_in_memory_session_service.py index a2c542ed..567a52d1 100644 --- a/trpc_agent_sdk/sessions/_in_memory_session_service.py +++ b/trpc_agent_sdk/sessions/_in_memory_session_service.py @@ -183,26 +183,32 @@ async def get_session( return self._merge_state(app_state, user_state, copied_session) @override - async def list_sessions(self, *, app_name: str, user_id: str) -> ListSessionsResponse: + async def list_sessions(self, *, app_name: str, user_id: Optional[str] = None) -> ListSessionsResponse: empty_response = ListSessionsResponse() - if app_name not in self._sessions: + app_sessions = self._sessions.get(app_name) + if not app_sessions: return empty_response - if user_id not in self._sessions[app_name]: + + if user_id is None: + user_sessions = app_sessions + elif user_id in app_sessions: + user_sessions = {user_id: app_sessions[user_id]} + else: return empty_response + app_state = self._get_app_state(app_name) sessions_without_events = [] - for session_id in self._sessions[app_name][user_id].keys(): - session = self._get_session(app_name, user_id, session_id) - if session is None: - continue - - copied_session = copy.deepcopy(session) - copied_session.events = [] - copied_session.historical_events = [] - app_state = self._get_app_state(app_name) - user_state = self._get_user_state(app_name, user_id) - copied_session = self._merge_state(app_state, user_state, copied_session) - sessions_without_events.append(copied_session) + for uid, session_dict in user_sessions.items(): + user_state = self._get_user_state(app_name, uid) + for session_id in session_dict.keys(): + session = self._get_session(app_name, uid, session_id) + if session is None: + continue + + copied_session = copy.deepcopy(session) + copied_session.events = [] + copied_session.historical_events = [] + sessions_without_events.append(self._merge_state(app_state, user_state, copied_session)) return ListSessionsResponse(sessions=sessions_without_events) @override diff --git a/trpc_agent_sdk/sessions/_redis_session_service.py b/trpc_agent_sdk/sessions/_redis_session_service.py index 01f77182..8751a75f 100644 --- a/trpc_agent_sdk/sessions/_redis_session_service.py +++ b/trpc_agent_sdk/sessions/_redis_session_service.py @@ -36,6 +36,24 @@ from ._utils import user_state_key +def _session_key_prefix(app_name: str, user_id: Optional[str] = None) -> str: + """Generate a Redis key prefix for listing sessions. + + When user_id is None, the prefix matches sessions across all users for the + given app; otherwise it is scoped to the specific user. + + Args: + app_name: Application name + user_id: Optional user identifier + + Returns: + Formatted session key prefix with a trailing wildcard. + """ + if user_id is None: + return f"session:{app_name}:*" + return f"session:{app_name}:{user_id}:*" + + class RedisSessionService(BaseSessionService): """A Redis implementation of the session service. @@ -129,18 +147,17 @@ async def get_session( return self._merge_state(app_state, user_state, session) @override - async def list_sessions(self, *, app_name: str, user_id: str) -> ListSessionsResponse: + async def list_sessions(self, *, app_name: str, user_id: Optional[str] = None) -> ListSessionsResponse: async with self._redis_storage.create_db_session() as redis_session: - pattern = session_key(app_name, user_id, "*") + pattern = _session_key_prefix(app_name, user_id) command = RedisCommand(method='keys', args=(pattern, )) keys = await self._redis_storage.execute_command(redis_session, command) if not keys: return ListSessionsResponse() - # Get app and user state once for all sessions + # Get app state once for all sessions app_state = await self._get_app_state(redis_session, app_name) - user_state = await self._get_user_state(redis_session, app_name, user_id) sessions_without_events = [] for key in keys: @@ -150,6 +167,7 @@ async def list_sessions(self, *, app_name: str, user_id: str) -> ListSessionsRes storage_session.events = [] storage_session.historical_events = [] # Merge state + user_state = await self._get_user_state(redis_session, app_name, storage_session.user_id) storage_session = self._merge_state(app_state, user_state, storage_session) sessions_without_events.append(storage_session) diff --git a/trpc_agent_sdk/sessions/_sql_session_service.py b/trpc_agent_sdk/sessions/_sql_session_service.py index d5ad5a08..4333ffeb 100644 --- a/trpc_agent_sdk/sessions/_sql_session_service.py +++ b/trpc_agent_sdk/sessions/_sql_session_service.py @@ -489,15 +489,17 @@ async def get_session( return self.filter_events(session) @override - async def list_sessions(self, *, app_name: str, user_id: str) -> ListSessionsResponse: + async def list_sessions(self, *, app_name: str, user_id: Optional[str] = None) -> ListSessionsResponse: async with self._sql_storage.create_db_session() as sql_session: - filters = [StorageSession.app_name == app_name, StorageSession.user_id == user_id] + filters = [StorageSession.app_name == app_name] + if user_id is not None: + filters.append(StorageSession.user_id == user_id) conditions = SqlCondition(filters=filters) - session_key = SqlKey(key=(app_name, user_id), storage_cls=StorageSession) + session_key = SqlKey(key=(app_name, user_id) if user_id is not None else (app_name, ), + storage_cls=StorageSession) results: List[StorageSession] = await self._sql_storage.query(sql_session, session_key, conditions) app_state = await self._get_app_state(sql_session, app_name) - user_state = await self._get_user_state(sql_session, app_name, user_id) sessions = [] for storage_session in results: @@ -508,6 +510,7 @@ async def list_sessions(self, *, app_name: str, user_id: str) -> ListSessionsRes storage_session.events = [] + user_state = await self._get_user_state(sql_session, app_name, storage_session.user_id) merged_state = merge_state( StateStorageEntry(app_state_delta=app_state, user_state_delta=user_state,