From 0555b296cc4b016728289afa20b3d2726f928157 Mon Sep 17 00:00:00 2001 From: Rogee Date: Thu, 13 Aug 2026 01:28:31 +0800 Subject: [PATCH] fix(conversations): sync AI takeover exit in realtime (#6) Co-authored-by: Rogee --- backend/internal/service/ai_takeover_test.go | 22 ++++++++++++++++- .../internal/service/conversation_service.go | 8 +++---- backend/internal/service/message_service.go | 15 ++++++++++++ backend/internal/wsevent/bridge_listener.go | 1 + backend/internal/wsevent/coverage3_test.go | 24 +++++++++++++++++++ .../widgets/conversation/ReplyBox.vue | 1 - .../helper/specs/actionCable.spec.js | 18 ++++++++++++++ 7 files changed, 83 insertions(+), 6 deletions(-) diff --git a/backend/internal/service/ai_takeover_test.go b/backend/internal/service/ai_takeover_test.go index cff13ef0..8883ecc9 100644 --- a/backend/internal/service/ai_takeover_test.go +++ b/backend/internal/service/ai_takeover_test.go @@ -5,6 +5,7 @@ import ( "fmt" "testing" + "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/worker" "github.com/stretchr/testify/assert" @@ -26,6 +27,8 @@ func configureInboxAI(t *testing.T, svc *ConversationService, accountID, inboxID func TestConversationServiceAITakeoverStartExitAndRestart(t *testing.T) { svc, db := setupConversationService(t) + capture := &captureConversationEventsListener{} + svc.dispatcher.Register(capture) account := createConversationServiceTestAccount(t, db) inbox := createConversationServiceTestInbox(t, db, account.ID) contact := createConversationServiceTestContact(t, db, account.ID) @@ -48,6 +51,15 @@ func TestConversationServiceAITakeoverStartExitAndRestart(t *testing.T) { assert.Nil(t, exited.AssigneeAgentBotID) assert.Equal(t, string(model.ConversationStatusOpen), exited.Status) assert.Equal(t, uint(2), exited.AITakeoverVersion) + require.GreaterOrEqual(t, len(capture.events), 2) + assert.Equal(t, channel.EventConversationUpdated, capture.events[len(capture.events)-2].Type) + assert.Equal(t, channel.EventConversationUnassigned, capture.events[len(capture.events)-1].Type) + + eventCount := len(capture.events) + repeatedExit, err := svc.ExitAITakeover(context.Background(), account.ID, conversation.ID) + require.NoError(t, err) + assert.Nil(t, repeatedExit.AssigneeAgentBotID) + assert.Equal(t, eventCount, len(capture.events)) restarted, err := svc.StartAITakeover(context.Background(), account.ID, conversation.ID) require.NoError(t, err) @@ -72,7 +84,9 @@ func TestConversationServiceAITakeoverRejectsMissingOrAmbiguousAI(t *testing.T) } func TestMessageServiceHumanOutgoingExitsAITakeoverAtomically(t *testing.T) { - db, _, _, svc := setupMessageServiceWithDefaultLLM(t) + db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t) + capture := &captureConversationEventsListener{} + dispatcher.Register(capture) account := createTestAccount(t, db) user := createTestUser(t, db, account.ID) inbox := createTestInbox(t, db, account.ID, "web_widget") @@ -94,6 +108,12 @@ func TestMessageServiceHumanOutgoingExitsAITakeoverAtomically(t *testing.T) { assert.Nil(t, updated.AssigneeAgentBotID) assert.Equal(t, string(model.ConversationStatusOpen), updated.Status) assert.Equal(t, uint(2), updated.AITakeoverVersion) + require.GreaterOrEqual(t, len(capture.events), 2) + assert.Equal(t, channel.EventConversationUpdated, capture.events[0].Type) + eventConversation := capture.events[0].Data["conversation"].(*model.Conversation) + assert.Nil(t, eventConversation.AssigneeAgentBotID) + assert.Equal(t, string(model.ConversationStatusOpen), eventConversation.Status) + assert.Equal(t, channel.EventMessageCreated, capture.events[1].Type) } func TestCaptainConversationResponseQueuesOncePerIncomingMessageAndSupportsMultipleTurns(t *testing.T) { diff --git a/backend/internal/service/conversation_service.go b/backend/internal/service/conversation_service.go index afe58394..a06db3f3 100644 --- a/backend/internal/service/conversation_service.go +++ b/backend/internal/service/conversation_service.go @@ -597,11 +597,11 @@ func (s *ConversationService) ExitAITakeover(ctx context.Context, accountID, id if err := s.repo.ExitAITakeover(ctx, conversation.ID); err != nil { return nil, err } - conversation.AssigneeAgentBotID = nil - if conversation.Status == string(model.ConversationStatusPending) { - conversation.Status = string(model.ConversationStatusOpen) + conversation, err = s.repo.FindByAccountAndID(ctx, accountID, conversation.ID) + if err != nil { + return nil, err } - conversation.AITakeoverVersion++ + s.dispatchConversationEvent(ctx, channel.EventConversationUpdated, conversation) s.dispatchConversationEvent(ctx, channel.EventConversationUnassigned, conversation) s.indexConversation(ctx, conversation) return conversation, nil diff --git a/backend/internal/service/message_service.go b/backend/internal/service/message_service.go index c25d0c64..49b97301 100644 --- a/backend/internal/service/message_service.go +++ b/backend/internal/service/message_service.go @@ -356,6 +356,11 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint Updates(conversationUpdates).Error; err != nil { return fmt.Errorf("update conversation message timestamps: %w", err) } + if message.AITakeoverExited { + if err := tx.Where("id = ? AND account_id = ?", message.ConversationID, accountID).First(&conversation).Error; err != nil { + return err + } + } for _, input := range req.Attachments { fileType := attachmentFileType(input.ContentType) attachment := &model.Attachment{ @@ -401,6 +406,16 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint s.worker.Publish(ctx, deliveryJob) } + if message.AITakeoverExited { + event := channel.NewChannelEvent(channel.EventConversationUpdated, channel.ChannelType(conversation.ChannelType), conversation.AccountID, conversation.InboxID) + event.ConversationID = conversation.ID + event.ContactID = conversation.ContactID + event.Data["conversation"] = &conversation + if err := s.dispatcher.Dispatch(ctx, event); err != nil { + applogger.L().Errorf("failed to dispatch event %s for conversation %d: %v", channel.EventConversationUpdated, conversation.ID, err) + } + } + // Dispatch EventMessageCreated s.dispatchMessageEvent(ctx, channel.EventMessageCreated, message) s.indexMessage(ctx, message) diff --git a/backend/internal/wsevent/bridge_listener.go b/backend/internal/wsevent/bridge_listener.go index c782b51c..40da9e51 100644 --- a/backend/internal/wsevent/bridge_listener.go +++ b/backend/internal/wsevent/bridge_listener.go @@ -107,6 +107,7 @@ func conversationPushPayload(conversation *model.Conversation, data map[string]i conversationID = *conversation.DisplayID } payload["id"] = conversationID + payload["ai_takeover_active"] = conversation.AssigneeAgentBotID != nil && conversation.Status == string(model.ConversationStatusPending) payload["created_at"] = conversation.CreatedAt.Unix() payload["updated_at"] = float64(conversation.UpdatedAt.UnixNano()) / 1e9 if conversation.LastActivityAt != nil { diff --git a/backend/internal/wsevent/coverage3_test.go b/backend/internal/wsevent/coverage3_test.go index 51609a1d..3f11d1b6 100644 --- a/backend/internal/wsevent/coverage3_test.go +++ b/backend/internal/wsevent/coverage3_test.go @@ -389,6 +389,12 @@ func TestConversationPushPayload_Cov3(t *testing.T) { assert.Equal(t, uint(200), result["id"]) assert.NotNil(t, result["meta"]) assert.NotNil(t, result["messages"]) + assert.Equal(t, false, result["ai_takeover_active"]) + + botID := uint(9) + conv.AssigneeAgentBotID = &botID + conv.Status = string(model.ConversationStatusPending) + assert.Equal(t, true, conversationPushPayload(conv, data)["ai_takeover_active"]) } // === messagePushPayload === @@ -507,6 +513,24 @@ func TestOnEvent_ConversationCreated_Cov3(t *testing.T) { assert.NotNil(t, hub.data) } +func TestOnEvent_ConversationUpdatedPublishesInactiveTakeover_Cov3(t *testing.T) { + hub := &captureHub_Cov3{} + l := New(wspkg.NewEventPublisherLocal(hub, nil)) + conv := makeConversation_Cov3(1) + event := &channel.ChannelEvent{ + Type: channel.EventConversationUpdated, + AccountID: 1, + ConversationID: 1, + Data: map[string]interface{}{"conversation": conv}, + } + + require.NoError(t, l.OnEvent(context.Background(), event)) + var envelope map[string]interface{} + require.NoError(t, json.Unmarshal(hub.data, &envelope)) + payload := envelope["data"].(map[string]interface{}) + assert.Equal(t, false, payload["ai_takeover_active"]) +} + func TestOnEvent_WithInboxID_Cov3(t *testing.T) { hub := &captureHub_Cov3{} l := New(wspkg.NewEventPublisherLocal(hub, nil)) diff --git a/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue b/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue index 1fb96654..774e6555 100644 --- a/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue +++ b/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue @@ -891,7 +891,6 @@ export default { ); if (sentMessage?.ai_takeover_exited) { useAlert(this.$t('CONVERSATION.AI_TAKEOVER.EXIT_SUCCESS')); - await this.$store.dispatch('getConversation', this.conversationId); } emitter.emit(BUS_EVENTS.SCROLL_TO_MESSAGE); emitter.emit(BUS_EVENTS.MESSAGE_SENT); diff --git a/frontend/app/javascript/dashboard/helper/specs/actionCable.spec.js b/frontend/app/javascript/dashboard/helper/specs/actionCable.spec.js index 8ba411a5..6c8ad5b8 100644 --- a/frontend/app/javascript/dashboard/helper/specs/actionCable.spec.js +++ b/frontend/app/javascript/dashboard/helper/specs/actionCable.spec.js @@ -160,4 +160,22 @@ describe('ActionCableConnector - Copilot Tests', () => { expect(mockDispatch).toHaveBeenCalledTimes(2); }); }); + + it('updates AI takeover state from conversation.updated', () => { + const conversation = { + id: 42, + account_id: 1, + ai_takeover_active: false, + }; + + actionCable.onReceived({ + event: 'conversation.updated', + data: conversation, + }); + + expect(mockDispatch).toHaveBeenCalledWith( + 'updateConversation', + conversation + ); + }); });