diff --git a/python/packages/kagent-adk/src/kagent/adk/_session_service.py b/python/packages/kagent-adk/src/kagent/adk/_session_service.py index 5ed8f6885..e3d08e4f3 100644 --- a/python/packages/kagent-adk/src/kagent/adk/_session_service.py +++ b/python/packages/kagent-adk/src/kagent/adk/_session_service.py @@ -48,7 +48,7 @@ async def create_session( # Make API call to create session # Pass user_id as a query param so the controller's auth middleware - # (UnsecureAuthenticator) reads it consistently — matching the user_id + # (UnsecureAuthenticator) reads it consistently, matching the user_id # used by get_session, list_sessions, delete_session, and append_event. # Without this, unsecure-mode requests fall back to "admin@kagent.dev" # while all lookups use the A2A-derived user_id, causing SessionNotFoundError. @@ -78,22 +78,14 @@ async def get_session( config: Optional[GetSessionConfig] = None, ) -> Optional[Session]: try: - # ADK requires events to be chronological (especially for calculating deltas) - params: dict[str, str | int] = {"user_id": user_id, "order": "asc"} - if config: - if config.after_timestamp is not None: - params["after"] = datetime.fromtimestamp(config.after_timestamp, tz=timezone.utc).isoformat() - if config.num_recent_events is not None: - # Ascending order with a limit selects the oldest events, while ADK requests - # the most recent events in chronological order. Fetch newest-first, then reverse below. - # The API treats limit=0 as unlimited, so request one event and discard it below. - params["order"] = "desc" - params["limit"] = max(config.num_recent_events, 1) - else: - params["limit"] = -1 - else: - # return all - params["limit"] = -1 + # ADK requires events to be chronological (especially for calculating deltas). + # Always fetch the full history: state is built by replaying every event's + # state_delta below, so limiting the fetch here would silently drop state set + # by events outside the window. num_recent_events is applied after, by + # trimming session.events once state is already correct. + params: dict[str, str | int] = {"user_id": user_id, "order": "asc", "limit": -1} + if config and config.after_timestamp is not None: + params["after"] = datetime.fromtimestamp(config.after_timestamp, tz=timezone.utc).isoformat() # Make API call to get session response: httpx.Response = await self.client.get(f"/api/sessions/{session_id}", params=params) @@ -110,11 +102,6 @@ async def get_session( session_data = data["data"]["session"] events_data = data["data"]["events"] - if config and config.num_recent_events is not None: - if config.num_recent_events == 0: - events_data = [] - else: - events_data.reverse() events: list[Event] = [] for event_data in events_data: @@ -132,6 +119,13 @@ async def get_session( for event in events: await super().append_event(session, event) + if config and config.num_recent_events is not None: + # Trim only after every event has been replayed, so state is complete. + # num_recent_events == 0 means "no events", not "all events" ([-0:] would + # keep everything). + num_recent_events = config.num_recent_events + session.events = session.events[-num_recent_events:] if num_recent_events else [] + return session except httpx.HTTPStatusError as e: if e.response.status_code == 404: diff --git a/python/packages/kagent-adk/tests/unittests/test_session_service.py b/python/packages/kagent-adk/tests/unittests/test_session_service.py index fb302c4c5..6655a6dd5 100644 --- a/python/packages/kagent-adk/tests/unittests/test_session_service.py +++ b/python/packages/kagent-adk/tests/unittests/test_session_service.py @@ -75,7 +75,7 @@ async def test_create_session_passes_user_id_as_query_param(): the controller's UnsecureAuthenticator resolves identity from the query param (or X-User-Id header), not the JSON body. Without the query param the controller falls back to "admin@kagent.dev" for the session create, while - every subsequent GET uses the A2A-derived user_id — guaranteeing a 404. + every subsequent GET uses the A2A-derived user_id, guaranteeing a 404. Fixes: https://github.com/kagent-dev/kagent/issues/1882 """ mock_response = MagicMock(spec=httpx.Response) @@ -132,9 +132,9 @@ async def test_get_session_passes_after_timestamp_to_api(mock_client, session_re "/api/sessions/s1", params={ "user_id": "u1", - "order": "desc", + "order": "asc", + "limit": -1, "after": "2026-07-27T10:30:00+00:00", - "limit": 25, }, ) @@ -165,8 +165,11 @@ async def test_get_session_passes_epoch_timestamp_to_api(mock_client, session_re @pytest.mark.asyncio async def test_get_session_with_zero_recent_events_returns_no_events(make_event, session_response, mock_client): - """ADK defines a zero recent-event limit as returning session metadata without history.""" - client = mock_client(session_response([make_event("user")])) + """ADK defines a zero recent-event limit as returning session metadata without history. + + State still has to be complete: every event is replayed, only the events list is emptied. + """ + client = mock_client(session_response([make_event("user", state_delta={"key": "value"})])) svc = KAgentSessionService(client) session = await svc.get_session( @@ -178,18 +181,19 @@ async def test_get_session_with_zero_recent_events_returns_no_events(make_event, assert session is not None assert session.events == [] + assert session.state.get("key") == "value", "state must survive even when no events are returned" client.get.assert_awaited_once_with( "/api/sessions/s1", - params={"user_id": "u1", "order": "desc", "limit": 1}, + params={"user_id": "u1", "order": "asc", "limit": -1}, ) @pytest.mark.asyncio async def test_get_session_returns_recent_events_in_chronological_order(make_event, session_response, mock_client): - """Recent-event limits select newest rows while presenting them to ADK oldest-first.""" + """The recent-events window keeps the oldest-first order the API already returns.""" older_event = make_event("older") newer_event = make_event("newer") - client = mock_client(session_response([newer_event, older_event])) + client = mock_client(session_response([older_event, newer_event])) svc = KAgentSessionService(client) session = await svc.get_session( @@ -203,7 +207,7 @@ async def test_get_session_returns_recent_events_in_chronological_order(make_eve assert [event.id for event in session.events] == [older_event.id, newer_event.id] client.get.assert_awaited_once_with( "/api/sessions/s1", - params={"user_id": "u1", "order": "desc", "limit": 2}, + params={"user_id": "u1", "order": "asc", "limit": -1}, ) @@ -231,7 +235,7 @@ async def test_get_session_events_not_duplicated(make_event, session_response, s assert session is not None assert len(session.events) == len(events), ( - f"Expected {len(events)} events but got {len(session.events)} — possible event duplication in get_session" + f"Expected {len(events)} events but got {len(session.events)}, possible event duplication in get_session" ) @@ -271,11 +275,47 @@ async def test_get_session_state_delta_applied_once(make_event, session_response # so for an idempotent string the bug was silent; here we use a distinct value # and just verify the key is present with the correct value.) assert session.state.get("counter") == 7, ( - f"Expected state['counter'] == 7, got {session.state.get('counter')} — " + f"Expected state['counter'] == 7, got {session.state.get('counter')}, " "state_delta may have been applied more than once" ) +@pytest.mark.asyncio +async def test_get_session_state_kept_outside_recent_events_window(make_event, session_response): + """A state delta from an event outside the num_recent_events window must + still land in session.state, only session.events is trimmed to the window. + + This is what makes the test fail against the old code, which asked the + server for only num_recent_events and so never saw the older state_delta. + """ + all_events = [ + make_event("assistant", state_delta={"old_key": "old_value"}), + make_event("user"), + make_event("assistant"), + ] + + def get_side_effect(url, params=None): + assert params.get("limit") == -1, "get_session must always fetch full history to avoid losing state deltas" + mock_response = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = session_response(all_events) + mock_response.raise_for_status = MagicMock() + return mock_response + + client = MagicMock(spec=httpx.AsyncClient) + client.get = AsyncMock(side_effect=get_side_effect) + + session = await KAgentSessionService(client).get_session( + app_name="app", user_id="u1", session_id="s1", config=GetSessionConfig(num_recent_events=2) + ) + + assert session is not None + assert len(session.events) == 2 + assert session.state.get("old_key") == "old_value", ( + "state from the first event must still apply even though only the last 2 events are kept in session.events" + ) + + @pytest.mark.asyncio async def test_get_session_multiple_state_deltas_applied_once(make_event, session_response, service): """Multiple events each contributing a state key are each applied once."""