refactor: enhance conversation handling and websocket integration with new team ID support

internal/pkg/enums/im.go

@@ -54,10 +54,10 @@ func GetIMConversationServiceModeLabel(mode IMConversationServiceMode) string {
 type IMSenderType string
 
 const (
-	IMSenderTypeAgent    IMSenderType = "agent"
-	IMSenderTypeCustomer IMSenderType = "customer"
-	IMSenderTypeAI       IMSenderType = "ai"
-	IMSenderTypeSystem   IMSenderType = "system"
+	IMSenderTypeAgent    IMSenderType = "agent"    // 客服
+	IMSenderTypeCustomer IMSenderType = "customer" // 客户
+	IMSenderTypeAI       IMSenderType = "ai"       // AI
+	IMSenderTypeSystem   IMSenderType = "system"   // 系统
 )
 
 var imSenderTypeLabelMap = map[IMSenderType]string{

internal/services/conversation_human_dispatch_realtime_test.go

@@ -0,0 +1,242 @@
+package services
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"cs-agent/internal/models"
+	"cs-agent/internal/pkg/enums"
+
+	"github.com/glebarez/sqlite"
+	"github.com/mlogclub/simple/sqls"
+	"gorm.io/gorm"
+	"gorm.io/gorm/schema"
+)
+
+func TestAIHandoffPublishesFinalAssignedConversationEvent(t *testing.T) {
+	db := setupHumanDispatchRealtimeTestDB(t)
+	WsService = newWsService()
+	session := captureHumanDispatchRealtimeSession(t, "admin:101", "admin:all")
+	aiAgent := createHumanDispatchRealtimeAIAgent(t, db, "1")
+	createHumanDispatchRealtimeTeam(t, db, 1)
+	createHumanDispatchRealtimeActiveSchedule(t, db, 1)
+	createHumanDispatchRealtimeAgentProfile(t, db, 101, 1)
+	conversation := createHumanDispatchRealtimeConversation(t, db, aiAgent.ID)
+
+	result, err := ConversationHumanDispatchService.HandoffByAI(conversation.ID, aiAgent, "用户要求转人工")
+	if err != nil {
+		t.Fatalf("HandoffByAI() error = %v", err)
+	}
+	if result == nil || result.Decision != HandoffDecisionAssigned {
+		t.Fatalf("expected assigned decision, got %+v", result)
+	}
+
+	event := findHumanDispatchRealtimeEvent(t, session, enums.IMRealtimeEventConversationAssigned)
+	if event.Data["conversationId"] != float64(conversation.ID) {
+		t.Fatalf("unexpected conversation id in event: %+v", event.Data)
+	}
+	if event.Data["status"] != float64(enums.IMConversationStatusActive) {
+		t.Fatalf("expected active status in assigned event, got %+v", event.Data["status"])
+	}
+	if event.Data["currentAssigneeId"] != float64(101) {
+		t.Fatalf("expected assignee 101 in assigned event, got %+v", event.Data["currentAssigneeId"])
+	}
+}
+
+func TestAIHandoffPublishesFinalTeamPoolConversationEvent(t *testing.T) {
+	db := setupHumanDispatchRealtimeTestDB(t)
+	WsService = newWsService()
+	session := captureHumanDispatchRealtimeSession(t, "admin:all")
+	aiAgent := createHumanDispatchRealtimeAIAgent(t, db, "1")
+	createHumanDispatchRealtimeTeam(t, db, 1)
+	createHumanDispatchRealtimeActiveSchedule(t, db, 1)
+	conversation := createHumanDispatchRealtimeConversation(t, db, aiAgent.ID)
+
+	result, err := ConversationHumanDispatchService.HandoffByAI(conversation.ID, aiAgent, "用户要求转人工")
+	if err != nil {
+		t.Fatalf("HandoffByAI() error = %v", err)
+	}
+	if result == nil || result.Decision != HandoffDecisionTeamPool {
+		t.Fatalf("expected team_pool decision, got %+v", result)
+	}
+
+	event := findHumanDispatchRealtimeEvent(t, session, enums.IMRealtimeEventConversationUpdated, func(event humanDispatchRealtimeEvent) bool {
+		return event.Data["currentTeamId"] == float64(1)
+	})
+	if event.Data["conversationId"] != float64(conversation.ID) {
+		t.Fatalf("unexpected conversation id in event: %+v", event.Data)
+	}
+	if event.Data["status"] != float64(enums.IMConversationStatusPending) {
+		t.Fatalf("expected pending status in updated event, got %+v", event.Data["status"])
+	}
+	if value, ok := event.Data["currentAssigneeId"]; ok && value != float64(0) {
+		t.Fatalf("expected no assignee in updated event, got %+v", event.Data["currentAssigneeId"])
+	}
+}
+
+type humanDispatchRealtimeEvent struct {
+	Type string         `json:"type"`
+	Data map[string]any `json:"data"`
+}
+
+func findHumanDispatchRealtimeEvent(t *testing.T, session *ClientSession, eventType string, matchers ...func(humanDispatchRealtimeEvent) bool) humanDispatchRealtimeEvent {
+	t.Helper()
+	timeout := time.After(500 * time.Millisecond)
+	for {
+		select {
+		case raw := <-session.Send:
+			var event humanDispatchRealtimeEvent
+			if err := json.Unmarshal(raw, &event); err != nil {
+				t.Fatalf("decode realtime event: %v", err)
+			}
+			if event.Type != eventType {
+				continue
+			}
+			matched := true
+			for _, matcher := range matchers {
+				if !matcher(event) {
+					matched = false
+					break
+				}
+			}
+			if matched {
+				return event
+			}
+		case <-timeout:
+			t.Fatalf("expected realtime event %q", eventType)
+		}
+	}
+}
+
+func captureHumanDispatchRealtimeSession(t *testing.T, topics ...string) *ClientSession {
+	t.Helper()
+	session := &ClientSession{
+		ID:     "test-session",
+		Role:   realtimeRoleAdmin,
+		Topics: map[string]struct{}{},
+		Send:   make(chan []byte, 32),
+	}
+	WsService.manager.Register(session, topics)
+	return session
+}
+
+func setupHumanDispatchRealtimeTestDB(t *testing.T) *gorm.DB {
+	t.Helper()
+	dbName := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name())
+	db, err := gorm.Open(sqlite.Open("file:"+dbName+"?mode=memory&cache=shared"), &gorm.Config{
+		NamingStrategy: schema.NamingStrategy{
+			TablePrefix:   "t_",
+			SingularTable: true,
+		},
+	})
+	if err != nil {
+		t.Fatalf("open sqlite error = %v", err)
+	}
+	t.Cleanup(func() {
+		sqlDB, err := db.DB()
+		if err == nil {
+			_ = sqlDB.Close()
+		}
+	})
+	if err := db.AutoMigrate(
+		&models.User{},
+		&models.Notification{},
+		&models.Customer{},
+		&models.CustomerIdentity{},
+		&models.Channel{},
+		&models.AIAgent{},
+		&models.AgentTeam{},
+		&models.AgentTeamSchedule{},
+		&models.AgentProfile{},
+		&models.Conversation{},
+		&models.ConversationParticipant{},
+		&models.ConversationAssignment{},
+		&models.ConversationEventLog{},
+		&models.ConversationReadState{},
+		&models.Message{},
+		&models.ChannelMessageOutbox{},
+	); err != nil {
+		t.Fatalf("auto migrate error = %v", err)
+	}
+	sqls.SetDB(db)
+	return db
+}
+
+func createHumanDispatchRealtimeAIAgent(t *testing.T, db *gorm.DB, teamIDs string) models.AIAgent {
+	t.Helper()
+	item := models.AIAgent{
+		Name:        "测试AI",
+		ServiceMode: enums.IMConversationServiceModeAIFirst,
+		TeamIDs:     teamIDs,
+		Status:      enums.StatusOk,
+	}
+	if err := db.Create(&item).Error; err != nil {
+		t.Fatalf("create ai agent error = %v", err)
+	}
+	return item
+}
+
+func createHumanDispatchRealtimeTeam(t *testing.T, db *gorm.DB, id int64) {
+	t.Helper()
+	if err := db.Create(&models.AgentTeam{ID: id, Name: "售后支持组", Status: enums.StatusOk}).Error; err != nil {
+		t.Fatalf("create team error = %v", err)
+	}
+}
+
+func createHumanDispatchRealtimeActiveSchedule(t *testing.T, db *gorm.DB, teamID int64) {
+	t.Helper()
+	now := time.Now()
+	if err := db.Create(&models.AgentTeamSchedule{
+		TeamID:  teamID,
+		StartAt: now.Add(-time.Hour),
+		EndAt:   now.Add(time.Hour),
+		Status:  enums.StatusOk,
+	}).Error; err != nil {
+		t.Fatalf("create schedule error = %v", err)
+	}
+}
+
+func createHumanDispatchRealtimeAgentProfile(t *testing.T, db *gorm.DB, userID, teamID int64) {
+	t.Helper()
+	if err := db.Create(&models.User{
+		ID:       userID,
+		Username: "agent",
+		Nickname: "客服",
+		Status:   enums.StatusOk,
+	}).Error; err != nil {
+		t.Fatalf("create user error = %v", err)
+	}
+	if err := db.Create(&models.AgentProfile{
+		UserID:             userID,
+		TeamID:             teamID,
+		AgentCode:          "A001",
+		DisplayName:        "客服",
+		ServiceStatus:      enums.ServiceStatusIdle,
+		MaxConcurrentCount: 3,
+		AutoAssignEnabled:  true,
+		Status:             enums.StatusOk,
+	}).Error; err != nil {
+		t.Fatalf("create profile error = %v", err)
+	}
+}
+
+func createHumanDispatchRealtimeConversation(t *testing.T, db *gorm.DB, aiAgentID int64) models.Conversation {
+	t.Helper()
+	now := time.Now()
+	item := models.Conversation{
+		AIAgentID:     aiAgentID,
+		ChannelID:     1,
+		CustomerID:    1,
+		CustomerName:  "测试访客",
+		Status:        enums.IMConversationStatusAIServing,
+		ServiceMode:   enums.IMConversationServiceModeAIFirst,
+		LastMessageAt: now,
+		LastActiveAt:  now,
+	}
+	if err := db.Create(&item).Error; err != nil {
+		t.Fatalf("create conversation error = %v", err)
+	}
+	return item
+}

internal/services/conversation_human_dispatch_service.go

@@ -121,6 +121,7 @@ func (s *conversationHumanDispatchService) DispatchPendingConversation(conversat
 			return nil, err
 		}
 		if dispatched != nil {
+			WsService.PublishConversationChanged(dispatched, enums.IMRealtimeEventConversationAssigned)
 			return &HandoffDecisionResult{
 				Decision:   HandoffDecisionAssigned,
 				TeamID:     dispatched.CurrentTeamID,
@@ -129,9 +130,13 @@ func (s *conversationHumanDispatchService) DispatchPendingConversation(conversat
 		}
 	}
 	teamID := activeTeamIDs[0]
-	if err := s.moveToTeamPool(conversationID, teamID, "手动触发自动分配"); err != nil {
+	teamPoolConversation, err := s.moveToTeamPool(conversationID, teamID, "手动触发自动分配")
+	if err != nil {
 		return nil, err
 	}
+	if teamPoolConversation != nil {
+		WsService.PublishConversationChanged(teamPoolConversation, enums.IMRealtimeEventConversationUpdated)
+	}
 	return &HandoffDecisionResult{Decision: HandoffDecisionTeamPool, TeamID: teamID}, nil
 }
 
@@ -150,6 +155,7 @@ func (s *conversationHumanDispatchService) dispatchAfterHandoff(conversationID,
 			return nil, err
 		}
 		if dispatched != nil {
+			WsService.PublishConversationChanged(dispatched, enums.IMRealtimeEventConversationAssigned)
 			if publishAssignEvent {
 				eventbus.PublishAsync(context.Background(), events.ConversationAssignedEvent{
 					ConversationID: dispatched.ID,
@@ -169,9 +175,13 @@ func (s *conversationHumanDispatchService) dispatchAfterHandoff(conversationID,
 	}
 
 	teamID := activeTeamIDs[0]
-	if err := s.moveToTeamPool(conversationID, teamID, reason); err != nil {
+	teamPoolConversation, err := s.moveToTeamPool(conversationID, teamID, reason)
+	if err != nil {
 		return nil, err
 	}
+	if teamPoolConversation != nil {
+		WsService.PublishConversationChanged(teamPoolConversation, enums.IMRealtimeEventConversationUpdated)
+	}
 	return &HandoffDecisionResult{Decision: HandoffDecisionTeamPool, TeamID: teamID, Message: HandoffWaitingMessage}, nil
 }
 
@@ -195,11 +205,12 @@ func (s *conversationHumanDispatchService) markHandoff(conversationID int64, aiA
 	})
 }
 
-func (s *conversationHumanDispatchService) moveToTeamPool(conversationID, teamID int64, reason string) error {
+func (s *conversationHumanDispatchService) moveToTeamPool(conversationID, teamID int64, reason string) (*models.Conversation, error) {
 	now := time.Now()
-	return sqls.WithTransaction(func(ctx *sqls.TxContext) error {
-		conversation := repositories.ConversationRepository.Get(ctx.Tx, conversationID)
-		if conversation == nil {
+	var conversation *models.Conversation
+	err := sqls.WithTransaction(func(ctx *sqls.TxContext) error {
+		current := repositories.ConversationRepository.Get(ctx.Tx, conversationID)
+		if current == nil {
 			return errorsx.InvalidParam("会话不存在")
 		}
 		if err := ConversationAssignmentService.FinishActiveAssignments(ctx, conversationID, now); err != nil {
@@ -215,16 +226,30 @@ func (s *conversationHumanDispatchService) moveToTeamPool(conversationID, teamID
 		}); err != nil {
 			return err
 		}
-		return ConversationEventLogService.CreateEvent(ctx, conversationID, enums.IMEventTypeTransfer, enums.IMSenderTypeSystem, 0, "会话进入客服组待接入", ConversationService.buildEventPayload(map[string]any{
-			"fromStatus":     conversation.Status,
+		if err := ConversationEventLogService.CreateEvent(ctx, conversationID, enums.IMEventTypeTransfer, enums.IMSenderTypeSystem, 0, "会话进入客服组待接入", ConversationService.buildEventPayload(map[string]any{
+			"fromStatus":     current.Status,
 			"toStatus":       enums.IMConversationStatusPending,
-			"fromAssigneeId": conversation.CurrentAssigneeID,
+			"fromAssigneeId": current.CurrentAssigneeID,
 			"toAssigneeId":   int64(0),
 			"toTeamId":       teamID,
 			"reason":         strings.TrimSpace(reason),
 			"decision":       string(HandoffDecisionTeamPool),
-		}))
+		})); err != nil {
+			return err
+		}
+		current.Status = enums.IMConversationStatusPending
+		current.CurrentTeamID = teamID
+		current.CurrentAssigneeID = 0
+		current.UpdateUserID = 0
+		current.UpdateUserName = "system"
+		current.UpdatedAt = now
+		conversation = current
+		return nil
 	})
+	if err != nil {
+		return nil, err
+	}
+	return conversation, nil
 }
 
 func (s *conversationHumanDispatchService) moveToGlobalPool(conversationID int64, operatorName string) error {

internal/services/conversation_read_state_service.go

@@ -172,24 +172,24 @@ func (s *conversationReadStateService) pickConversationReadStates(list []models.
 }
 
 // MarkAgentRead 在事务内更新/创建客服已读游标。
-func (s *conversationReadStateService) MarkAgentRead(ctx *sqls.TxContext, conversation *models.Conversation, operator *dto.AuthPrincipal, message *models.Message, now time.Time) (*models.ConversationReadState, error) {
+func (s *conversationReadStateService) MarkAgentRead(ctx *sqls.TxContext, conversation *models.Conversation, operator *dto.AuthPrincipal, message *models.Message) (*models.ConversationReadState, error) {
 	c, err := agentReaderCursor(operator)
 	if err != nil {
 		return nil, err
 	}
-	return s.markReadTxWithCursor(ctx, conversation, c, message, now)
+	return s.markReadTxWithCursor(ctx, conversation, c, message)
 }
 
 // MarkCustomerRead 在事务内更新/创建 IM 客户已读游标。
-func (s *conversationReadStateService) MarkCustomerRead(ctx *sqls.TxContext, conversation *models.Conversation, external *openidentity.ExternalUser, message *models.Message, now time.Time) (*models.ConversationReadState, error) {
+func (s *conversationReadStateService) MarkCustomerRead(ctx *sqls.TxContext, conversation *models.Conversation, external *openidentity.ExternalUser, message *models.Message) (*models.ConversationReadState, error) {
 	c, err := customerReaderCursor(external)
 	if err != nil {
 		return nil, err
 	}
-	return s.markReadTxWithCursor(ctx, conversation, c, message, now)
+	return s.markReadTxWithCursor(ctx, conversation, c, message)
 }
 
-func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, conversation *models.Conversation, c readerCursor, message *models.Message, now time.Time) (*models.ConversationReadState, error) {
+func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, conversation *models.Conversation, c readerCursor, message *models.Message) (*models.ConversationReadState, error) {
 	if ctx == nil || conversation == nil || message == nil {
 		return nil, nil
 	}
@@ -197,6 +197,8 @@ func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext,
 		return nil, errorsx.InvalidParam("不支持的已读操作类型")
 	}
 
+	now := time.Now()
+
 	item := &models.ConversationReadState{}
 	err := ctx.Tx.Where("conversation_id = ? AND reader_type = ? AND reader_id = ? AND external_reader_id = ?",
 		conversation.ID, c.readerType, c.readerID, c.externalReaderID,

internal/services/conversation_service.go

@@ -153,7 +153,7 @@ func (s *conversationService) Create(externalUser openidentity.ExternalUser, cha
 		if err := ConversationEventLogService.CreateEvent(ctx, conversation.ID, enums.IMEventTypeCreate, enums.IMSenderTypeCustomer, 0, "用户创建会话", ""); err != nil {
 			return err
 		}
-		welcomeMessage, err = MessageService.CreateAIWelcomeMessageTx(ctx, conversation, aiAgent, now)
+		welcomeMessage, err = MessageService.createAIWelcomeMessage(ctx, conversation, aiAgent, now)
 		return err
 	}); err != nil {
 		return nil, err
@@ -517,7 +517,7 @@ func displayExternalName(ext *openidentity.ExternalUser) string {
 type conversationReadActor interface {
 	isAgentSide() bool
 	getReadState(conversationID int64) *models.ConversationReadState
-	markReadTx(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message, now time.Time) error
+	markRead(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message) error
 	conversationUpdateAudit() (userID int64, userName string)
 }
 
@@ -531,8 +531,8 @@ func (a agentConversationReadActor) getReadState(conversationID int64) *models.C
 	return ConversationReadStateService.GetByAgentReader(conversationID, a.operator)
 }
 
-func (a agentConversationReadActor) markReadTx(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message, now time.Time) error {
-	_, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, a.operator, targetMessage, now)
+func (a agentConversationReadActor) markRead(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message) error {
+	_, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, a.operator, targetMessage)
 	return err
 }
 
@@ -553,8 +553,8 @@ func (a customerConversationReadActor) getReadState(conversationID int64) *model
 	return ConversationReadStateService.GetByCustomerReader(conversationID, a.external)
 }
 
-func (a customerConversationReadActor) markReadTx(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message, now time.Time) error {
-	_, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, a.external, targetMessage, now)
+func (a customerConversationReadActor) markRead(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message) error {
+	_, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, a.external, targetMessage)
 	return err
 }
 
@@ -607,8 +607,7 @@ func (s *conversationService) markConversationReadWithActor(conversation *models
 		if currentConversation == nil {
 			return errorsx.InvalidParam("会话不存在")
 		}
-		now := time.Now()
-		if err := actor.markReadTx(ctx, currentConversation, targetMessage, now); err != nil {
+		if err := actor.markRead(ctx, currentConversation, targetMessage); err != nil {
 			return err
 		}
 		agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, currentConversation.ID)
@@ -632,7 +631,7 @@ func (s *conversationService) markConversationReadWithActor(conversation *models
 			"customer_unread_count": customerUnreadCount,
 			"update_user_id":        updateUserID,
 			"update_user_name":      updateUserName,
-			"updated_at":            now,
+			"updated_at":            time.Now(),
 		})
 	})
 	if err != nil {

internal/services/message_service.go

@@ -258,7 +258,7 @@ func (s *messageService) SendAIServiceNotice(conversationID int64, aiAgentID int
 	}, nil)
 }
 
-func (s *messageService) CreateAIWelcomeMessageTx(ctx *sqls.TxContext, conversation *models.Conversation, aiAgent *models.AIAgent, now time.Time) (*models.Message, error) {
+func (s *messageService) createAIWelcomeMessage(ctx *sqls.TxContext, conversation *models.Conversation, aiAgent *models.AIAgent, now time.Time) (*models.Message, error) {
 	if ctx == nil || conversation == nil || aiAgent == nil || strings.TrimSpace(aiAgent.WelcomeMessage) == "" {
 		return nil, nil
 	}
@@ -300,7 +300,7 @@ func (s *messageService) CreateAIWelcomeMessageTx(ctx *sqls.TxContext, conversat
 		return nil, err
 	}
 
-	if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message, now); err != nil {
+	if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message); err != nil {
 		return nil, err
 	}
 	agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID)
@@ -450,56 +450,44 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation,
 		}
 
 		// 处理已读、维度
-		readStateType := senderType
-		if senderType == enums.IMSenderTypeAI {
-			readStateType = enums.IMSenderTypeAgent
-		}
-		if readStateType == enums.IMSenderTypeAgent {
-			if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message, now); err != nil {
-				return err
-			}
-		} else {
-			if _, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, external, message, now); err != nil {
-				return err
-			}
-		}
-		agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID)
-		agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer)
-		if err != nil {
-			return err
-		}
-		customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI)
+		agentUnreadCount, customerUnreadCount, err := s.handleReadState(ctx, senderType, conversation, operator, message, external)
 		if err != nil {
 			return err
 		}
 
-		updateUserID := int64(0)
-		updateUserName := ""
+		conversation.LastMessageID = message.ID
+		conversation.LastMessageAt = now
+		conversation.LastActiveAt = now
+		conversation.LastMessageSummary = limitText(summary, 255)
+		conversation.UpdateUserID = int64(0)
+		conversation.UpdateUserName = ""
 		if operator != nil {
-			updateUserID = operator.UserID
-			updateUserName = operator.Username
+			conversation.UpdateUserID = operator.UserID
+			conversation.UpdateUserName = operator.Username
 		}
 		if senderType == enums.IMSenderTypeCustomer && external != nil {
-			updateUserID = 0
-			updateUserName = displayExternalName(external)
+			conversation.UpdateUserID = 0
+			conversation.UpdateUserName = displayExternalName(external)
 		}
+		conversation.UpdatedAt = now
+		conversation.AgentUnreadCount = int(agentUnreadCount)
+		conversation.CustomerUnreadCount = int(customerUnreadCount)
 		if err := repositories.ConversationRepository.Updates(ctx.Tx, conversation.ID, map[string]any{
-			"last_message_id":       message.ID,
-			"last_message_at":       now,
-			"last_active_at":        now,
-			"last_message_summary":  limitText(summary, 255),
-			"update_user_id":        updateUserID,
-			"update_user_name":      updateUserName,
-			"updated_at":            now,
-			"agent_unread_count":    agentUnreadCount,
-			"customer_unread_count": customerUnreadCount,
+			"last_message_id":       conversation.LastMessageID,
+			"last_message_at":       conversation.LastMessageAt,
+			"last_active_at":        conversation.LastActiveAt,
+			"last_message_summary":  conversation.LastMessageSummary,
+			"update_user_id":        conversation.UpdateUserID,
+			"update_user_name":      conversation.UpdateUserName,
+			"updated_at":            conversation.UpdatedAt,
+			"agent_unread_count":    conversation.AgentUnreadCount,
+			"customer_unread_count": conversation.CustomerUnreadCount,
 		}); err != nil {
 			return err
 		}
-		if err := ConversationEventLogService.CreateEvent(ctx,
-			conversation.ID,
-			enums.IMEventTypeMessageSend,
-			senderType,
+
+		// 记录事件日志
+		if err := ConversationEventLogService.CreateEvent(ctx, conversation.ID, enums.IMEventTypeMessageSend, senderType,
 			func() int64 {
 				if operator != nil {
 					return operator.UserID
@@ -511,14 +499,18 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation,
 		); err != nil {
 			return err
 		}
+
 		return nil
 	})
 	if err != nil {
 		return nil, err
 	}
 
+	// 处理websocket消息
 	WsService.PublishMessageCreated(conversation, message)
 	WsService.PublishConversationChanged(conversation, enums.IMRealtimeEventConversationUpdated)
+
+	// 企业微信客服消息入队，异步发送
 	if enqueueErr := ChannelMessageOutboxService.EnqueueWxWorkKFMessage(conversation, message); enqueueErr != nil {
 		slog.Error("enqueue wxwork kf outbox failed",
 			"conversation_id", conversation.ID,
@@ -526,6 +518,8 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation,
 			"error", enqueueErr,
 		)
 	}
+
+	// 客户发送消息，触发AI回复
 	if senderType == enums.IMSenderTypeCustomer {
 		if TriggerAIReplyAsyncHook != nil {
 			TriggerAIReplyAsyncHook(*conversation, *message)
@@ -534,6 +528,31 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation,
 	return message, err
 }
 
+// handleReadState 根据发送者类型更新会话已读状态，并返回更新后的客服和客户未读消息数。
+func (s *messageService) handleReadState(ctx *sqls.TxContext, senderType enums.IMSenderType, conversation *models.Conversation, operator *dto.AuthPrincipal, message *models.Message, external *openidentity.ExternalUser) (agentUnreadCount int64, customerUnreadCount int64, err error) {
+	readStateType := senderType
+	if senderType == enums.IMSenderTypeAI {
+		readStateType = enums.IMSenderTypeAgent
+	}
+	if readStateType == enums.IMSenderTypeAgent {
+		if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message); err != nil {
+			return 0, 0, err
+		}
+	} else {
+		if _, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, external, message); err != nil {
+			return 0, 0, err
+		}
+	}
+	agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID)
+	if agentUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer); err != nil {
+		return 0, 0, err
+	}
+	if customerUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI); err != nil {
+		return 0, 0, err
+	}
+	return agentUnreadCount, customerUnreadCount, nil
+}
+
 func limitText(value string, maxLen int) string {
 	if maxLen <= 0 {
 		return ""

internal/services/ws_realtime_types.go

@@ -192,6 +192,7 @@ type RealtimeConversationChangedPayload struct {
 	Status                    enums.IMConversationStatus      `json:"status,omitempty"`
 	ServiceMode               enums.IMConversationServiceMode `json:"serviceMode,omitempty"`
 	CurrentAssigneeID         int64                           `json:"currentAssigneeId,omitempty"`
+	CurrentTeamID             int64                           `json:"currentTeamId,omitempty"`
 	LastMessageID             int64                           `json:"lastMessageId,omitempty"`
 	LastMessageAt             string                          `json:"lastMessageAt,omitempty"`
 	LastActiveAt              string                          `json:"lastActiveAt,omitempty"`

internal/services/ws_service.go

@@ -426,6 +426,7 @@ func (s *wsService) PublishConversationChanged(conversation *models.Conversation
 			Status:                    conversation.Status,
 			ServiceMode:               conversation.ServiceMode,
 			CurrentAssigneeID:         conversation.CurrentAssigneeID,
+			CurrentTeamID:             conversation.CurrentTeamID,
 			LastMessageID:             conversation.LastMessageID,
 			LastMessageAt:             formatWsTime(&conversation.LastMessageAt),
 			LastActiveAt:              formatWsTime(&conversation.LastActiveAt),

web/hooks/use-agent-conversation-realtime.ts

@@ -5,6 +5,7 @@ import { toast } from "sonner"
 
 import { createAdminWebSocketUrl } from "@/lib/api/admin"
 import { type AgentMessage } from "@/lib/api/agent"
+import { shouldReloadConversationListForRealtimePatch } from "@/lib/agent-conversation-realtime"
 import { readSession } from "@/lib/auth"
 import {
   normalizeRealtimeMessage,
@@ -131,6 +132,11 @@ export function useAgentConversationRealtime() {
 
           if (eventType.startsWith("conversation.") && payload) {
             store.applyRealtimeConversationChanged(payload)
+            if (shouldReloadConversationListForRealtimePatch(payload)) {
+              void store.resyncRealtimeData(conversationId).catch((error) => {
+                toast.error(error instanceof Error ? error.message : "同步会话列表失败")
+              })
+            }
           }
         } catch {
           // ignore invalid ws payload

web/lib/agent-conversation-realtime.test.mjs

@@ -0,0 +1,27 @@
+import assert from "node:assert/strict"
+import test from "node:test"
+
+import { shouldReloadConversationListForRealtimePatch } from "./agent-conversation-realtime.ts"
+
+test("reloads conversation list when realtime patch changes list membership fields", () => {
+  assert.equal(
+    shouldReloadConversationListForRealtimePatch({
+      conversationId: 1,
+      status: 3,
+      currentAssigneeId: 101,
+    }),
+    true
+  )
+})
+
+test("keeps local patching for message summary and unread-only changes", () => {
+  assert.equal(
+    shouldReloadConversationListForRealtimePatch({
+      conversationId: 1,
+      lastMessageId: 9,
+      lastMessageSummary: "hello",
+      agentUnreadCount: 1,
+    }),
+    false
+  )
+})

web/lib/agent-conversation-realtime.ts

@@ -0,0 +1,18 @@
+import type { RealtimeConversationPatch } from "@/lib/im-realtime-state"
+
+const listMembershipFields = new Set<keyof RealtimeConversationPatch>([
+  "status",
+  "currentAssigneeId",
+  "currentTeamId",
+])
+
+export function shouldReloadConversationListForRealtimePatch(
+  patch: RealtimeConversationPatch | null | undefined
+) {
+  if (!patch) {
+    return false
+  }
+  return Object.keys(patch).some((key) =>
+    listMembershipFields.has(key as keyof RealtimeConversationPatch)
+  )
+}

web/lib/api/im.ts

@@ -40,6 +40,7 @@ export type ImConversation = {
   priority: number
   currentAssigneeId: number
   currentAssigneeName?: string
+  currentTeamId?: number
   lastMessageId: number
   lastMessageAt?: string
   lastActiveAt?: string