{session/inmemory,session/redis}: fix event filtering to ensure at least one user message (#737)

Author: nomand-zc

session/inmemory/service.go

@@ -723,7 +723,7 @@ func (s *SessionService) updateStoredSession(sess *session.Session, e *event.Eve
 		sess.EventMu.Lock()
 		sess.Events = append(sess.Events, *e)
 		if s.opts.sessionEventLimit > 0 && len(sess.Events) > s.opts.sessionEventLimit {
-			sess.Events = sess.Events[len(sess.Events)-s.opts.sessionEventLimit:]
+			sess.ApplyEventFiltering(session.WithEventNum(s.opts.sessionEventLimit))
 		}
 		sess.EventMu.Unlock()
 	}

session/redis/service.go

@@ -743,15 +743,7 @@ func (s *Service) getEventsList(
 ) ([][]event.Event, error) {
 	pipe := s.redisClient.Pipeline()
 	for _, key := range sessionKeys {
-		zrangeBy := &redis.ZRangeBy{
-			Min: fmt.Sprintf("%d", afterTime.UnixNano()),
-			Max: fmt.Sprintf("%d", time.Now().UnixNano()),
-		}
-		if limit > 0 {
-			zrangeBy.Offset = 0
-			zrangeBy.Count = int64(limit)
-		}
-		pipe.ZRevRangeByScore(ctx, getEventKey(key), zrangeBy)
+		pipe.ZRange(ctx, getEventKey(key), 0, -1)
 	}
 	cmds, err := pipe.Exec(ctx)
 	if err != nil && err != redis.Nil {
@@ -768,14 +760,14 @@ func (s *Service) getEventsList(
 		if err != nil {
 			return nil, fmt.Errorf("process event cmd failed: %w", err)
 		}
-
-		// reverse events to get chronological order (oldest first)
-		if len(events) > 1 {
-			for i, j := 0, len(events)-1; i < j; i, j = i+1, j-1 {
-				events[i], events[j] = events[j], events[i]
-			}
+		sess := &session.Session{
+			Events: events,
 		}
-		sessEventsList = append(sessEventsList, events)
+		if limit <= 0 {
+			limit = s.opts.sessionEventLimit
+		}
+		sess.ApplyEventFiltering(session.WithEventNum(limit), session.WithEventTime(afterTime))
+		sessEventsList = append(sessEventsList, sess.Events)
 	}
 	return sessEventsList, nil
 }
@@ -998,9 +990,6 @@ func (s *Service) addEvent(ctx context.Context, key session.Key, event *event.Ev
 			Score:  float64(event.Timestamp.UnixNano()),
 			Member: eventBytes,
 		})
-		if s.opts.sessionEventLimit > 0 {
-			txPipe.ZRemRangeByRank(ctx, getEventKey(key), 0, -(int64(s.opts.sessionEventLimit) + 1))
-		}
 		// Set TTL for session state and event list if configured
 		if s.sessionTTL > 0 {
 			txPipe.Expire(ctx, getEventKey(key), s.sessionTTL)

session/redis/service_test.go

@@ -1400,10 +1400,10 @@ func TestGetSession_EventFiltering_Integration(t *testing.T) {
 	require.NotNil(t, retrievedSess)
 
 	// Should have 2 events (from event3 onwards)
-	assert.Equal(t, 4, len(retrievedSess.Events), "Should filter out assistant events before first user event")
-	assert.Equal(t, "event1", retrievedSess.Events[0].ID, "First event should be the user event")
-	assert.Equal(t, model.RoleAssistant, retrievedSess.Events[0].Response.Choices[0].Message.Role)
-	assert.Equal(t, "event2", retrievedSess.Events[1].ID, "Second event should be the subsequent assistant event")
+	assert.Equal(t, 2, len(retrievedSess.Events), "Should filter out assistant events before first user event")
+	assert.Equal(t, "event3", retrievedSess.Events[0].ID, "First event should be the user event")
+	assert.Equal(t, model.RoleUser, retrievedSess.Events[0].Response.Choices[0].Message.Role)
+	assert.Equal(t, "event4", retrievedSess.Events[1].ID, "Second event should be the subsequent assistant event")
 
 	// Test ListSessions - should apply same filtering
 	sessionList, err := service.ListSessions(context.Background(), session.UserKey{
@@ -1414,9 +1414,9 @@ func TestGetSession_EventFiltering_Integration(t *testing.T) {
 	require.Len(t, sessionList, 1)
 
 	// Should have same filtering as GetSession
-	assert.Equal(t, 4, len(sessionList[0].Events), "ListSessions should also filter events")
-	assert.Equal(t, "event1", sessionList[0].Events[0].ID, "First event should be the user event")
-	assert.Equal(t, model.RoleAssistant, sessionList[0].Events[0].Response.Choices[0].Message.Role)
+	assert.Equal(t, 2, len(sessionList[0].Events), "ListSessions should also filter events")
+	assert.Equal(t, "event3", sessionList[0].Events[0].ID, "First event should be the user event")
+	assert.Equal(t, model.RoleUser, sessionList[0].Events[0].Response.Choices[0].Message.Role)
 }
 
 func TestGetSession_AllAssistantEvents_Integration(t *testing.T) {
@@ -1447,6 +1447,21 @@ func TestGetSession_AllAssistantEvents_Integration(t *testing.T) {
 				Choices: []model.Choice{
 					{
 						Index: 0,
+						Message: model.Message{
+							Role:    model.RoleUser,
+							Content: "user message 1",
+						},
+					},
+				},
+			},
+		},
+		{
+			ID:        "event1",
+			Timestamp: baseTime.Add(-3 * time.Hour),
+			Response: &model.Response{
+				Choices: []model.Choice{
+					{
+						Index: 1,
 						Message: model.Message{
 							Role:    model.RoleAssistant,
 							Content: "Assistant message 1",
@@ -1461,7 +1476,7 @@ func TestGetSession_AllAssistantEvents_Integration(t *testing.T) {
 			Response: &model.Response{
 				Choices: []model.Choice{
 					{
-						Index: 0,
+						Index: 2,
 						Message: model.Message{
 							Role:    model.RoleAssistant,
 							Content: "Assistant message 2",
@@ -1484,7 +1499,7 @@ func TestGetSession_AllAssistantEvents_Integration(t *testing.T) {
 	require.NotNil(t, retrievedSess)
 
 	// Should have no events since all are from assistant
-	assert.Equal(t, 2, len(retrievedSess.Events), "Should filter out all assistant events when no user events exist")
+	assert.Equal(t, 3, len(retrievedSess.Events), "Should filter out all assistant events when no user events exist")
 }
 
 func TestService_Close_MultipleTimes(t *testing.T) {