From 83a8ab315a5aac8efef70c31fb066ace3a8c0112 Mon Sep 17 00:00:00 2001 From: Rogee Date: Wed, 12 Aug 2026 22:57:06 +0800 Subject: [PATCH] H-16: align takeover with channel AI workflow (#2) * feat(conversations): complete manual AI takeover * fix(conversations): align AI takeover flow with channel AI * fix(conversations): close takeover review gaps --------- Co-authored-by: Rogee --- .../api/v1/assignable_agent_handler.go | 18 +-- .../api/v1/assignable_agent_handler_test.go | 12 +- .../handler/api/v1/conversation_handler.go | 104 ++++++++++----- .../api/v1/conversation_handler_crud_test.go | 49 +------ .../handler/api/v1/conversation_serializer.go | 17 ++- backend/internal/model/conversation.go | 1 + backend/internal/model/message.go | 1 + .../internal/repository/conversation_repo.go | 19 +++ backend/internal/router/router.go | 2 + .../internal/service/agent_bot_listener.go | 9 +- backend/internal/service/ai_takeover_test.go | 120 ++++++++++++++++++ .../service/captain_conversation_service.go | 90 +++++++++++-- .../captain_conversation_worker_test.go | 52 +++++++- .../internal/service/conversation_service.go | 70 ++++++++++ backend/internal/service/message_service.go | 80 ++++++++---- .../internal/service/message_service_test.go | 6 +- .../000077_add_ai_takeover_version.down.sql | 2 + .../000077_add_ai_takeover_version.up.sql | 2 + .../dashboard/api/assignableAgents.js | 2 +- .../dashboard/api/inbox/conversation.js | 15 ++- .../api/specs/assignableAgents.spec.js | 1 - .../api/specs/inbox/conversation.spec.js | 23 +++- .../widgets/conversation/ReplyBox.vue | 6 +- .../widgets/conversation/ReplyBoxBanner.vue | 36 ++++++ .../BulkAgentActions.vue | 4 +- .../composables/chatlist/useBulkActions.js | 26 +--- .../dashboard/helper/agentHelper.js | 2 +- .../i18n/locale/en/conversation.json | 7 + .../i18n/locale/zh_CN/conversation.json | 7 + .../conversation/ConversationAction.vue | 1 - .../store/modules/conversations/actions.js | 21 ++- 31 files changed, 604 insertions(+), 201 deletions(-) create mode 100644 backend/internal/service/ai_takeover_test.go create mode 100644 backend/migrations/000077_add_ai_takeover_version.down.sql create mode 100644 backend/migrations/000077_add_ai_takeover_version.up.sql diff --git a/backend/internal/handler/api/v1/assignable_agent_handler.go b/backend/internal/handler/api/v1/assignable_agent_handler.go index e99f10ea..c87c99ac 100644 --- a/backend/internal/handler/api/v1/assignable_agent_handler.go +++ b/backend/internal/handler/api/v1/assignable_agent_handler.go @@ -80,23 +80,7 @@ func (h *AssignableAgentHandler) List(c *gin.Context) { return } - includeAgentBots := c.Query("include_agent_bots") != "" - payload := serializeAssignableAgents(accountID, agents, includeAgentBots) - if includeAgentBots { - bots, err := h.svc.GetAssignableAgentBots(c.Request.Context(), accountID) - if err != nil { - response.AbortWithStatusError(c, http.StatusUnprocessableEntity, response.ErrInternal, "failed to list assignable agent bots") - return - } - for i := range bots { - bot := serializeAgentBotSlim(&bots[i]) - bot["assignee_type"] = "AgentBot" - bot["icon"] = "i-lucide-bot" - bot["availability_status"] = "offline" - bot["confirmed"] = true - payload = append(payload, bot) - } - } + payload := serializeAssignableAgents(accountID, agents) c.JSON(http.StatusOK, gin.H{"payload": payload}) } diff --git a/backend/internal/handler/api/v1/assignable_agent_handler_test.go b/backend/internal/handler/api/v1/assignable_agent_handler_test.go index 0d33c29d..c18b5177 100644 --- a/backend/internal/handler/api/v1/assignable_agent_handler_test.go +++ b/backend/internal/handler/api/v1/assignable_agent_handler_test.go @@ -306,7 +306,7 @@ func (s *AssignableAgentHandlerTestSuite) TestList_StandaloneResourceUsesFronten } } -func (s *AssignableAgentHandlerTestSuite) TestList_IncludeAgentBotsAddsTypedOwners() { +func (s *AssignableAgentHandlerTestSuite) TestList_IncludeAgentBotsDoesNotExposeBots() { bot := &model.AgentBot{AccountID: &s.account.ID, Name: "Triage bot", AvatarURL: "https://example.test/bot.png", BotType: "webhook"} s.Require().NoError(s.db.Create(bot).Error) s.T().Cleanup(func() { s.db.Delete(bot) }) @@ -315,18 +315,10 @@ func (s *AssignableAgentHandlerTestSuite) TestList_IncludeAgentBotsAddsTypedOwne s.router.ServeHTTP(w, req) s.Equal(http.StatusOK, w.Code, w.Body.String()) payload := s.decodeAssignablePayload(w) - foundBot := false for _, raw := range payload { owner := raw.(map[string]any) - s.Contains(owner, "assignee_type") - if owner["assignee_type"] == "AgentBot" { - foundBot = true - s.Equal(float64(bot.ID), owner["id"]) - s.Equal("i-lucide-bot", owner["icon"]) - s.Equal("offline", owner["availability_status"]) - } + s.NotEqual("AgentBot", owner["assignee_type"]) } - s.True(foundBot) } func (s *AssignableAgentHandlerTestSuite) TestList_MultipleInboxIDsQueryParams_NoIntersection() { diff --git a/backend/internal/handler/api/v1/conversation_handler.go b/backend/internal/handler/api/v1/conversation_handler.go index 0f87ddc8..dc756c1d 100644 --- a/backend/internal/handler/api/v1/conversation_handler.go +++ b/backend/internal/handler/api/v1/conversation_handler.go @@ -276,16 +276,7 @@ func (h *ConversationHandler) AssignAgent(c *gin.Context) { return } if req.AssigneeType == "AgentBot" { - _, bot, svcErr := h.conversationSvc.AssignAgentBot(c.Request.Context(), accountID, conversation.ID, req.AssigneeID) - if svcErr != nil { - handleServiceError(c, svcErr) - return - } - if bot == nil { - c.JSON(http.StatusOK, nil) - return - } - c.JSON(http.StatusOK, serializeAgentBotSlim(bot)) + response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrValidation, "AgentBot assignment is not supported; use AI takeover") return } @@ -1059,10 +1050,9 @@ func (h *ConversationHandler) AssignTeam(c *gin.Context) { } var req struct { - AgentID *uint `json:"agent_id"` - AssigneeID *uint `json:"assignee_id"` - AssigneeType string `json:"assignee_type"` - TeamID *uint `json:"team_id"` + AgentID *uint `json:"agent_id"` + AssigneeID *uint `json:"assignee_id"` + TeamID *uint `json:"team_id"` } if err := c.ShouldBindJSON(&req); err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrValidation, err.Error()) @@ -1077,23 +1067,6 @@ func (h *ConversationHandler) AssignTeam(c *gin.Context) { if !ok { return } - if req.AssigneeType == "AgentBot" { - if agentID == nil { - agentID = new(uint) - } - conversation, bot, svcErr := h.conversationSvc.AssignAgentBot(c.Request.Context(), accountID, conversation.ID, *agentID) - if svcErr != nil { - handleServiceError(c, svcErr) - return - } - recordAuditMutation(c, h.auditSvc, auditMutation{AccountID: accountID, AuditableType: "Conversation", AuditableID: conversation.ID, Action: "update", AuditedChanges: gin.H{"assignee_agent_bot_id": conversation.AssigneeAgentBotID}}) - if bot == nil { - c.JSON(http.StatusOK, nil) - return - } - c.JSON(http.StatusOK, serializeAgentBotSlim(bot)) - return - } conversation, svcErr := h.conversationSvc.AssignTeam(c.Request.Context(), accountID, conversation.ID, agentID, req.TeamID) if svcErr != nil { handleServiceError(c, svcErr) @@ -1111,6 +1084,75 @@ func (h *ConversationHandler) AssignTeam(c *gin.Context) { c.JSON(http.StatusOK, serializeConversation(h.requestContext(c), h.conversationSvc.DB(), conversation)) } +func (h *ConversationHandler) StartAITakeover(c *gin.Context) { + accountID, err := parseUintParam(c, "account_id") + if err != nil { + response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid account_id") + return + } + conversationID, err := parseUintParam(c, "conversation_id") + if err != nil { + response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid conversation_id") + return + } + conversation, ok := h.resolveConversationRoute(c, accountID, conversationID) + if !ok { + return + } + if !h.canManageConversationInbox(c.Request.Context(), accountID, conversation.InboxID, currentUserID(c)) { + response.AbortWithStatusError(c, http.StatusForbidden, response.ErrForbidden, "user does not have access to this inbox") + return + } + conversation, svcErr := h.conversationSvc.StartAITakeover(c.Request.Context(), accountID, conversation.ID) + if svcErr != nil { + handleServiceError(c, svcErr) + return + } + c.JSON(http.StatusOK, serializeConversation(h.requestContext(c), h.conversationSvc.DB(), conversation)) +} + +func (h *ConversationHandler) ExitAITakeover(c *gin.Context) { + accountID, err := parseUintParam(c, "account_id") + if err != nil { + response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid account_id") + return + } + conversationID, err := parseUintParam(c, "conversation_id") + if err != nil { + response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid conversation_id") + return + } + conversation, ok := h.resolveConversationRoute(c, accountID, conversationID) + if !ok { + return + } + if !h.canManageConversationInbox(c.Request.Context(), accountID, conversation.InboxID, currentUserID(c)) { + response.AbortWithStatusError(c, http.StatusForbidden, response.ErrForbidden, "user does not have access to this inbox") + return + } + conversation, svcErr := h.conversationSvc.ExitAITakeover(c.Request.Context(), accountID, conversation.ID) + if svcErr != nil { + handleServiceError(c, svcErr) + return + } + c.JSON(http.StatusOK, serializeConversation(h.requestContext(c), h.conversationSvc.DB(), conversation)) +} + +func (h *ConversationHandler) canManageConversationInbox(ctx context.Context, accountID, inboxID, userID uint) bool { + if userID == 0 || h.conversationSvc == nil || h.conversationSvc.DB() == nil { + return false + } + var accountUser model.AccountUser + if err := h.conversationSvc.DB().WithContext(ctx).Where("account_id = ? AND user_id = ?", accountID, userID).First(&accountUser).Error; err != nil { + return false + } + if accountUser.Role == "administrator" { + return true + } + var count int64 + return h.conversationSvc.DB().WithContext(ctx).Model(&model.InboxMember{}).Where("inbox_id = ? AND user_id = ?", inboxID, userID).Count(&count).Error == nil && count == 1 +} + // handleServiceError maps service-layer errors to appropriate HTTP responses. // GORM record-not-found → 404, validation/invalid/required → 400, other → 500. func handleServiceError(c *gin.Context, err error) { diff --git a/backend/internal/handler/api/v1/conversation_handler_crud_test.go b/backend/internal/handler/api/v1/conversation_handler_crud_test.go index b891ef82..c45346e8 100644 --- a/backend/internal/handler/api/v1/conversation_handler_crud_test.go +++ b/backend/internal/handler/api/v1/conversation_handler_crud_test.go @@ -727,62 +727,19 @@ func (s *ConversationCrudTestSuite) TestAssignAgent_Success() { assert.Equal(s.T(), float64(user.ID), resp["id"]) } -func (s *ConversationCrudTestSuite) TestAssignAgentBot_MutuallyExclusiveAndAccountScoped() { +func (s *ConversationCrudTestSuite) TestAssignAgentBot_IsRejectedByHumanAssignmentEndpoint() { bot := &model.AgentBot{AccountID: &s.testAccount.ID, Name: "Triage bot", BotType: "webhook", AccessToken: "triage-token", Secret: "triage-secret"} s.Require().NoError(s.db.Create(bot).Error) body, _ := json.Marshal(map[string]any{"assignee_id": bot.ID, "assignee_type": "AgentBot"}) w := httptest.NewRecorder() - req := httptest.NewRequest(http.MethodPost, s.convURL(s.testConv.ID)+"/assignments", bytes.NewReader(body)) + req := httptest.NewRequest(http.MethodPost, s.convURL(s.testConv.ID)+"/assign", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") s.router.ServeHTTP(w, req) - s.Equal(http.StatusOK, w.Code, w.Body.String()) + s.Equal(http.StatusBadRequest, w.Code, w.Body.String()) var assigned model.Conversation s.Require().NoError(s.db.First(&assigned, s.testConv.ID).Error) s.Nil(assigned.AssigneeID) - s.Require().NotNil(assigned.AssigneeAgentBotID) - s.Equal(bot.ID, *assigned.AssigneeAgentBotID) - s.Equal(string(model.ConversationStatusPending), assigned.Status) - var assignmentResponse map[string]any - s.Require().NoError(json.Unmarshal(w.Body.Bytes(), &assignmentResponse)) - s.Equal("AgentBot", assignmentResponse["assignee_type"]) - - // Repeating takeover is idempotent. - w = httptest.NewRecorder() - req = httptest.NewRequest(http.MethodPost, s.convURL(s.testConv.ID)+"/assignments", bytes.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - s.router.ServeHTTP(w, req) - s.Equal(http.StatusOK, w.Code, w.Body.String()) - s.Require().NoError(s.db.First(&assigned, s.testConv.ID).Error) - s.Require().NotNil(assigned.AssigneeAgentBotID) - s.Equal(bot.ID, *assigned.AssigneeAgentBotID) - s.Equal(string(model.ConversationStatusPending), assigned.Status) - - user := &model.User{Name: "Agent", Email: "bot-switch@example.com"} - s.Require().NoError(s.db.Create(user).Error) - s.Require().NoError(s.db.Create(&model.InboxMember{InboxID: s.testInbox.ID, UserID: user.ID}).Error) - s.Require().NoError(s.db.Create(&model.AccountUser{AccountID: s.testAccount.ID, UserID: user.ID, Role: "agent"}).Error) - body, _ = json.Marshal(map[string]any{"assignee_id": user.ID, "assignee_type": "User"}) - w = httptest.NewRecorder() - req = httptest.NewRequest(http.MethodPost, s.convURL(s.testConv.ID)+"/assign", bytes.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - s.router.ServeHTTP(w, req) - s.Equal(http.StatusOK, w.Code, w.Body.String()) - s.Require().NoError(s.db.First(&assigned, s.testConv.ID).Error) - s.Require().NotNil(assigned.AssigneeID) - s.Equal(user.ID, *assigned.AssigneeID) s.Nil(assigned.AssigneeAgentBotID) - s.Equal(string(model.ConversationStatusOpen), assigned.Status) - - other := &model.Account{Name: "Other"} - s.Require().NoError(s.db.Create(other).Error) - foreignBot := &model.AgentBot{AccountID: &other.ID, Name: "Foreign", BotType: "webhook", AccessToken: "foreign-token", Secret: "foreign-secret"} - s.Require().NoError(s.db.Create(foreignBot).Error) - body, _ = json.Marshal(map[string]any{"assignee_id": foreignBot.ID, "assignee_type": "AgentBot"}) - w = httptest.NewRecorder() - req = httptest.NewRequest(http.MethodPost, s.convURL(s.testConv.ID)+"/assign", bytes.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - s.router.ServeHTTP(w, req) - s.Equal(http.StatusNotFound, w.Code) } func (s *ConversationCrudTestSuite) TestAssignAgent_InvalidAccountID() { diff --git a/backend/internal/handler/api/v1/conversation_serializer.go b/backend/internal/handler/api/v1/conversation_serializer.go index 66212f7c..2d0ce8d5 100644 --- a/backend/internal/handler/api/v1/conversation_serializer.go +++ b/backend/internal/handler/api/v1/conversation_serializer.go @@ -48,6 +48,7 @@ type chatwootConversationPayload struct { Muted bool `json:"muted"` SnoozedUntil *int64 `json:"snoozed_until"` Status string `json:"status"` + AITakeoverActive bool `json:"ai_takeover_active"` CreatedAt int64 `json:"created_at"` UpdatedAt float64 `json:"updated_at"` Timestamp int64 `json:"timestamp"` @@ -132,6 +133,7 @@ type chatwootMessagePayload struct { ExternalSourceIDs map[string]any `json:"external_source_ids"` AdditionalAttrs map[string]any `json:"additional_attributes"` IdempotentReplay bool `json:"idempotent_replay,omitempty"` + AITakeoverExited bool `json:"ai_takeover_exited,omitempty"` Sender map[string]any `json:"sender,omitempty"` Attachments []any `json:"attachments,omitempty"` Call map[string]any `json:"call,omitempty"` @@ -260,6 +262,7 @@ func serializeConversation(ctx context.Context, db *gorm.DB, conversation *model Muted: conversationMuted(ctx, db, conversation), SnoozedUntil: conversation.SnoozedUntil, Status: conversation.Status, + AITakeoverActive: conversation.AssigneeAgentBotID != nil && conversation.Status == string(model.ConversationStatusPending), CreatedAt: conversation.CreatedAt.Unix(), UpdatedAt: float64(conversation.UpdatedAt.UnixNano()) / float64(time.Second), Timestamp: int64Value(conversation.LastActivityAt), @@ -446,6 +449,7 @@ func serializeMessage(ctx context.Context, db *gorm.DB, message *model.Message, ExternalSourceIDs: jsonObject(message.ExternalSourceIDs), AdditionalAttrs: jsonObject(message.AdditionalAttributes), IdempotentReplay: message.IdempotentReplay, + AITakeoverExited: message.AITakeoverExited, } if db != nil && message.SenderID != nil && *message.SenderID != 0 { senderType := normalizedSenderType(message.SenderType) @@ -767,13 +771,12 @@ func serializeCaptainAssistantSender(assistant *model.CaptainAssistant) map[stri func serializeAgentBotSlim(bot *model.AgentBot) map[string]any { return map[string]any{ - "id": bot.ID, - "name": bot.Name, - "description": bot.Description, - "thumbnail": bot.AvatarURL, - "outgoing_url": bot.OutgoingURL, - "bot_type": bot.BotType, - "assignee_type": "AgentBot", + "id": bot.ID, + "name": bot.Name, + "description": bot.Description, + "thumbnail": bot.AvatarURL, + "outgoing_url": bot.OutgoingURL, + "bot_type": bot.BotType, } } diff --git a/backend/internal/model/conversation.go b/backend/internal/model/conversation.go index ab5169cf..ef45d940 100644 --- a/backend/internal/model/conversation.go +++ b/backend/internal/model/conversation.go @@ -20,6 +20,7 @@ type Conversation struct { ContactInboxID *uint `gorm:"index" json:"contact_inbox_id,omitempty"` AssigneeID *uint `gorm:"index" json:"assignee_id"` AssigneeAgentBotID *uint `gorm:"index" json:"assignee_agent_bot_id,omitempty"` + AITakeoverVersion uint `gorm:"not null;default:0" json:"ai_takeover_version"` TeamID *uint `gorm:"index" json:"team_id,omitempty"` CampaignID *uint `gorm:"index" json:"campaign_id,omitempty"` SlaPolicyID *uint `gorm:"index" json:"sla_policy_id,omitempty"` diff --git a/backend/internal/model/message.go b/backend/internal/model/message.go index d341e807..444c339f 100644 --- a/backend/internal/model/message.go +++ b/backend/internal/model/message.go @@ -25,6 +25,7 @@ type Message struct { ExternalSourceIDs datatypes.JSON `gorm:"type:jsonb" json:"external_source_ids,omitempty"` // external platform IDs ExternalRequestHash string `gorm:"size:64" json:"-"` IdempotentReplay bool `gorm:"-" json:"idempotent_replay,omitempty"` + AITakeoverExited bool `gorm:"-" json:"ai_takeover_exited,omitempty"` Conversation *Conversation `gorm:"foreignKey:ConversationID" json:"conversation,omitempty"` Attachments []Attachment `gorm:"foreignKey:MessageID" json:"attachments,omitempty"` diff --git a/backend/internal/repository/conversation_repo.go b/backend/internal/repository/conversation_repo.go index d42be9ce..5d570402 100644 --- a/backend/internal/repository/conversation_repo.go +++ b/backend/internal/repository/conversation_repo.go @@ -333,6 +333,25 @@ func (r *ConversationRepo) AssignAgentBot(ctx context.Context, id, agentBotID ui Updates(map[string]any{"assignee_id": nil, "assignee_agent_bot_id": value, "status": status}).Error } +func (r *ConversationRepo) StartAITakeover(ctx context.Context, id, agentBotID uint) error { + return r.db.WithContext(ctx).Model(&model.Conversation{}).Where("id = ?", id). + Updates(map[string]any{ + "assignee_id": nil, "assignee_agent_bot_id": agentBotID, + "status": model.ConversationStatusPending, + "ai_takeover_version": gorm.Expr("ai_takeover_version + 1"), + }).Error +} + +func (r *ConversationRepo) ExitAITakeover(ctx context.Context, id uint) error { + return r.db.WithContext(ctx).Model(&model.Conversation{}). + Where("id = ? AND assignee_agent_bot_id IS NOT NULL", id). + Updates(map[string]any{ + "assignee_agent_bot_id": nil, + "status": gorm.Expr("CASE WHEN status = ? THEN ? ELSE status END", model.ConversationStatusPending, model.ConversationStatusOpen), + "ai_takeover_version": gorm.Expr("ai_takeover_version + 1"), + }).Error +} + // ToggleStatus toggles conversation between open/resolved. func (r *ConversationRepo) ToggleStatus(ctx context.Context, id uint, status model.ConversationStatus) error { return r.db.WithContext(ctx).Model(&model.Conversation{}).Where("id = ?", id). diff --git a/backend/internal/router/router.go b/backend/internal/router/router.go index 34fe7583..a3337d20 100644 --- a/backend/internal/router/router.go +++ b/backend/internal/router/router.go @@ -1073,6 +1073,8 @@ func registerV1Routes(g *gin.RouterGroup, h *Handlers) { conversations.POST("/:conversation_id/toggle_typing_status", h.Conversation.ToggleTyping) conversations.POST("/:conversation_id/update_last_seen", h.Conversation.UpdateLastSeen) conversations.POST("/:conversation_id/assignments", h.Conversation.AssignTeam) + conversations.POST("/:conversation_id/ai_takeover", h.Conversation.StartAITakeover) + conversations.DELETE("/:conversation_id/ai_takeover", h.Conversation.ExitAITakeover) // Participants nested under conversation participants := conversations.Group("/:conversation_id/participants") diff --git a/backend/internal/service/agent_bot_listener.go b/backend/internal/service/agent_bot_listener.go index f8c01501..d69807b2 100644 --- a/backend/internal/service/agent_bot_listener.go +++ b/backend/internal/service/agent_bot_listener.go @@ -179,13 +179,8 @@ func (l *AgentBotListener) HandleEvent(ctx context.Context, eventType string, ac // Captain bot type — route to CaptainConversationService instead of webhook if bot.BotType == "captain" { - if l.captainConvSvc == nil { - applogger.L().Warnf("AgentBotListener: captain bot %d but CaptainConversationService not configured", bot.ID) - continue - } - if err := l.handleCaptainBot(ctx, bot, eventType, accountID, inboxID, data); err != nil { - applogger.L().Errorf("AgentBotListener: captain bot %d response failed: %v", bot.ID, err) - } + // Captain replies are queued exactly once by MessageService using the + // incoming message ID as the durable idempotency key. continue } diff --git a/backend/internal/service/ai_takeover_test.go b/backend/internal/service/ai_takeover_test.go new file mode 100644 index 00000000..cff13ef0 --- /dev/null +++ b/backend/internal/service/ai_takeover_test.go @@ -0,0 +1,120 @@ +package service + +import ( + "context" + "fmt" + "testing" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func configureInboxAI(t *testing.T, svc *ConversationService, accountID, inboxID uint) (*model.AgentBot, *model.CaptainAssistant) { + t.Helper() + db := svc.DB() + require.NoError(t, db.AutoMigrate(&model.AgentBot{}, &model.AgentBotInbox{}, &model.CaptainAssistant{}, &model.CaptainInbox{})) + assistant := &model.CaptainAssistant{AccountID: accountID, Name: "Channel AI", Status: model.AssistantStatusActive, Config: []byte(`{}`)} + require.NoError(t, db.Create(assistant).Error) + require.NoError(t, db.Create(&model.CaptainInbox{AccountID: accountID, InboxID: inboxID, AssistantID: assistant.ID}).Error) + bot := &model.AgentBot{AccountID: &accountID, Name: "Channel AI", BotType: "captain", Secret: fmt.Sprintf("secret-%d", assistant.ID), AccessToken: fmt.Sprintf("token-%d", assistant.ID), Config: []byte(fmt.Sprintf(`{"assistant_id":%d}`, assistant.ID))} + require.NoError(t, db.Create(bot).Error) + require.NoError(t, db.Create(&model.AgentBotInbox{AgentBotID: bot.ID, InboxID: inboxID, Status: model.AgentBotInboxActive}).Error) + return bot, assistant +} + +func TestConversationServiceAITakeoverStartExitAndRestart(t *testing.T) { + svc, db := setupConversationService(t) + account := createConversationServiceTestAccount(t, db) + inbox := createConversationServiceTestInbox(t, db, account.ID) + contact := createConversationServiceTestContact(t, db, account.ID) + conversation := createConversationServiceTestConversation(t, db, account.ID, inbox.ID, contact.ID, string(model.ConversationStatusOpen)) + bot, _ := configureInboxAI(t, svc, account.ID, inbox.ID) + + started, err := svc.StartAITakeover(context.Background(), account.ID, conversation.ID) + require.NoError(t, err) + require.NotNil(t, started.AssigneeAgentBotID) + assert.Equal(t, bot.ID, *started.AssigneeAgentBotID) + assert.Equal(t, string(model.ConversationStatusPending), started.Status) + assert.Equal(t, uint(1), started.AITakeoverVersion) + + repeated, err := svc.StartAITakeover(context.Background(), account.ID, conversation.ID) + require.NoError(t, err) + assert.Equal(t, uint(1), repeated.AITakeoverVersion) + + exited, err := svc.ExitAITakeover(context.Background(), account.ID, conversation.ID) + require.NoError(t, err) + assert.Nil(t, exited.AssigneeAgentBotID) + assert.Equal(t, string(model.ConversationStatusOpen), exited.Status) + assert.Equal(t, uint(2), exited.AITakeoverVersion) + + restarted, err := svc.StartAITakeover(context.Background(), account.ID, conversation.ID) + require.NoError(t, err) + assert.Equal(t, uint(3), restarted.AITakeoverVersion) +} + +func TestConversationServiceAITakeoverRejectsMissingOrAmbiguousAI(t *testing.T) { + svc, db := setupConversationService(t) + account := createConversationServiceTestAccount(t, db) + inbox := createConversationServiceTestInbox(t, db, account.ID) + contact := createConversationServiceTestContact(t, db, account.ID) + conversation := createConversationServiceTestConversation(t, db, account.ID, inbox.ID, contact.ID, string(model.ConversationStatusOpen)) + require.NoError(t, db.AutoMigrate(&model.AgentBot{}, &model.AgentBotInbox{}, &model.CaptainAssistant{}, &model.CaptainInbox{})) + + _, err := svc.StartAITakeover(context.Background(), account.ID, conversation.ID) + assert.ErrorContains(t, err, "exactly one active AI") + + configureInboxAI(t, svc, account.ID, inbox.ID) + configureInboxAI(t, svc, account.ID, inbox.ID) + _, err = svc.StartAITakeover(context.Background(), account.ID, conversation.ID) + assert.ErrorContains(t, err, "exactly one active AI") +} + +func TestMessageServiceHumanOutgoingExitsAITakeoverAtomically(t *testing.T) { + db, _, _, svc := setupMessageServiceWithDefaultLLM(t) + account := createTestAccount(t, db) + user := createTestUser(t, db, account.ID) + inbox := createTestInbox(t, db, account.ID, "web_widget") + contact := createTestContact(t, db, account.ID) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + botID := uint(91) + require.NoError(t, db.Model(conversation).Updates(map[string]any{ + "assignee_agent_bot_id": botID, "status": model.ConversationStatusPending, "ai_takeover_version": 1, + }).Error) + + message, err := svc.Create(context.Background(), account.ID, user.ID, CreateMessageRequest{ + ConversationID: conversation.ID, Content: "human reply", MessageType: string(model.MessageTypeOutgoing), ContentType: string(model.MessageContentTypeText), + }) + require.NoError(t, err) + assert.True(t, message.AITakeoverExited) + + var updated model.Conversation + require.NoError(t, db.First(&updated, conversation.ID).Error) + assert.Nil(t, updated.AssigneeAgentBotID) + assert.Equal(t, string(model.ConversationStatusOpen), updated.Status) + assert.Equal(t, uint(2), updated.AITakeoverVersion) +} + +func TestCaptainConversationResponseQueuesOncePerIncomingMessageAndSupportsMultipleTurns(t *testing.T) { + db, conversationSvc, messageSvc, account, _, conversation, _ := setupCaptainConversationWorkerTest(t) + wp := worker.NewWorkerPool(db) + conversationSvc.SetWorkerPool(wp) + messageSvc.SetWorkerPool(wp) + + first, err := messageSvc.Create(context.Background(), account.ID, 99, CreateMessageRequest{ConversationID: conversation.ID, Content: "one", MessageType: string(model.MessageTypeIncoming), ContentType: string(model.MessageContentTypeText)}) + require.NoError(t, err) + second, err := messageSvc.Create(context.Background(), account.ID, 99, CreateMessageRequest{ConversationID: conversation.ID, Content: "two", MessageType: string(model.MessageTypeIncoming), ContentType: string(model.MessageContentTypeText)}) + require.NoError(t, err) + + _, err = EnqueueCaptainConversationResponseForMessage(context.Background(), wp, db, first.ID) + require.NoError(t, err) + var count int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).Count(&count).Error) + assert.Equal(t, int64(2), count) + + var jobs []model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).Order("id").Find(&jobs).Error) + assert.Contains(t, jobs[0].IdempotencyKey, fmt.Sprintf("message:%d", first.ID)) + assert.Contains(t, jobs[1].IdempotencyKey, fmt.Sprintf("message:%d", second.ID)) +} diff --git a/backend/internal/service/captain_conversation_service.go b/backend/internal/service/captain_conversation_service.go index f7b57efc..45b52b25 100644 --- a/backend/internal/service/captain_conversation_service.go +++ b/backend/internal/service/captain_conversation_service.go @@ -75,6 +75,10 @@ func (s *CaptainConversationService) SetWorkerPool(wp *worker.WorkerPool) { } func (s *CaptainConversationService) BuildConversationResponseByAccount(ctx context.Context, accountID, conversationID, assistantID uint) (*model.Message, error) { + return s.buildConversationResponseByAccount(ctx, accountID, conversationID, assistantID, 0) +} + +func (s *CaptainConversationService) buildConversationResponseByAccount(ctx context.Context, accountID, conversationID, assistantID, takeoverVersion uint) (*model.Message, error) { if s == nil || s.db == nil { return nil, fmt.Errorf("captain conversation service database is required") } @@ -85,6 +89,9 @@ func (s *CaptainConversationService) BuildConversationResponseByAccount(ctx cont if conversation.Status != string(model.ConversationStatusPending) { return nil, nil } + if takeoverVersion != 0 && (conversation.AssigneeAgentBotID == nil || conversation.AITakeoverVersion != takeoverVersion) { + return nil, nil + } var assistant model.CaptainAssistant if err := s.db.WithContext(ctx).Where("account_id = ? AND id = ?", accountID, assistantID).First(&assistant).Error; err != nil { return nil, fmt.Errorf("assistant not found: %w", err) @@ -101,6 +108,15 @@ func (s *CaptainConversationService) BuildConversationResponseByAccount(ctx cont if response == nil { return nil, fmt.Errorf("empty captain conversation response") } + if takeoverVersion != 0 { + active, err := s.aiTakeoverStillActive(ctx, accountID, conversationID, takeoverVersion) + if err != nil { + return nil, err + } + if !active { + return nil, nil + } + } if response.Action == "handoff" || response.Content == "conversation_handoff" || response.HandoffToolCalled { return s.createCaptainHandoffMessage(ctx, &conversation, &assistant) } @@ -110,6 +126,14 @@ func (s *CaptainConversationService) BuildConversationResponseByAccount(ctx cont return s.createCaptainOutgoingMessage(ctx, &conversation, &assistant, response.Content, response.AgentName) } +func (s *CaptainConversationService) aiTakeoverStillActive(ctx context.Context, accountID, conversationID, takeoverVersion uint) (bool, error) { + var active int64 + err := s.db.WithContext(ctx).Model(&model.Conversation{}). + Where("account_id = ? AND id = ? AND status = ? AND assignee_agent_bot_id IS NOT NULL AND ai_takeover_version = ?", accountID, conversationID, model.ConversationStatusPending, takeoverVersion). + Count(&active).Error + return active == 1, err +} + func (s *CaptainConversationService) collectConversationMessages(ctx context.Context, accountID, conversationID uint) ([]CaptainConversationMessage, error) { var messages []model.Message if err := s.db.WithContext(ctx). @@ -193,10 +217,15 @@ func (s *CaptainConversationService) createCaptainOutgoingMessage(ctx context.Co attrs = datatypes.JSON(raw) } if s.messageSvc != nil { + expectedVersion := uint(0) + if conversation.AssigneeAgentBotID != nil { + expectedVersion = conversation.AITakeoverVersion + } return s.messageSvc.Create(ctx, conversation.AccountID, assistant.ID, CreateMessageRequest{ ConversationID: conversation.ID, Content: content, ContentType: string(model.MessageContentTypeText), MessageType: string(model.MessageTypeOutgoing), SenderID: assistant.ID, SenderType: "Captain::Assistant", AdditionalAttributes: attrs, + ExpectedAITakeoverVersion: expectedVersion, }) } message := &model.Message{ @@ -231,6 +260,18 @@ func (s *CaptainConversationService) createCaptainHandoffMessage(ctx context.Con if err != nil { return nil, err } + if message == nil { + return nil, nil + } + if conversation.AssigneeAgentBotID != nil { + active, err := s.aiTakeoverStillActive(ctx, conversation.AccountID, conversation.ID, conversation.AITakeoverVersion) + if err != nil { + return message, err + } + if !active { + return message, nil + } + } if err := s.db.WithContext(ctx).Model(&model.Conversation{}).Where("account_id = ? AND id = ? AND status = ?", conversation.AccountID, conversation.ID, string(model.ConversationStatusPending)).Update("status", string(model.ConversationStatusOpen)).Error; err != nil { return message, fmt.Errorf("handoff captain conversation: %w", err) } @@ -266,10 +307,11 @@ func assistantConfigString(raw []byte, key string) string { } type captainConversationResponseBuilderJob struct { - AccountID uint `json:"account_id"` - ConversationID uint `json:"conversation_id"` - AssistantID uint `json:"assistant_id"` - MessageID uint `json:"message_id,omitempty"` + AccountID uint `json:"account_id"` + ConversationID uint `json:"conversation_id"` + AssistantID uint `json:"assistant_id"` + MessageID uint `json:"message_id,omitempty"` + TakeoverVersion uint `json:"takeover_version,omitempty"` } var captainConversationRegistrations sync.Map @@ -302,15 +344,37 @@ func EnqueueCaptainConversationResponseForMessage(ctx context.Context, wp *worke if conversation.Status != string(model.ConversationStatusPending) { return nil, nil } + var assistantID uint + var takeoverVersion uint if conversation.AssigneeAgentBotID != nil { - return nil, nil - } - var ci model.CaptainInbox - if err := db.WithContext(ctx).Where("account_id = ? AND inbox_id = ?", conversation.AccountID, conversation.InboxID).First(&ci).Error; err != nil { - if err == gorm.ErrRecordNotFound { - return nil, nil + var bot model.AgentBot + if err := db.WithContext(ctx).Where("id = ? AND bot_type = ?", *conversation.AssigneeAgentBotID, "captain").First(&bot).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return nil, err } - return nil, err + assistantID = extractAssistantIDFromBotConfig(bot.Config) + takeoverVersion = conversation.AITakeoverVersion + } else { + var preference model.CaptainPreference + if err := db.WithContext(ctx).Where("account_id = ? AND auto_reply_enabled = ?", conversation.AccountID, true).First(&preference).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return nil, err + } + var ci model.CaptainInbox + if err := db.WithContext(ctx).Where("account_id = ? AND inbox_id = ?", conversation.AccountID, conversation.InboxID).First(&ci).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return nil, err + } + assistantID = ci.AssistantID + } + if assistantID == 0 { + return nil, nil } options := []worker.EnqueueOption{ worker.WithMaxAttempts(3), @@ -319,7 +383,7 @@ func EnqueueCaptainConversationResponseForMessage(ctx context.Context, wp *worke if scheduledAt := captainConversationResponseScheduledAt(ctx, db, message.ID); !scheduledAt.IsZero() { options = append(options, worker.WithScheduledAt(scheduledAt)) } - return wp.Enqueue(ctx, TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: conversation.AccountID, ConversationID: conversation.ID, AssistantID: ci.AssistantID, MessageID: message.ID}, options...) + return wp.Enqueue(ctx, TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: conversation.AccountID, ConversationID: conversation.ID, AssistantID: assistantID, MessageID: message.ID, TakeoverVersion: takeoverVersion}, options...) } func captainConversationResponseScheduledAt(ctx context.Context, db *gorm.DB, messageID uint) time.Time { @@ -349,6 +413,6 @@ func (s *CaptainConversationService) performConversationResponseBuilderJob(ctx c if payload.AccountID == 0 || payload.ConversationID == 0 || payload.AssistantID == 0 { return fmt.Errorf("invalid captain conversation response builder job payload: %#v", payload) } - _, err := s.BuildConversationResponseByAccount(ctx, payload.AccountID, payload.ConversationID, payload.AssistantID) + _, err := s.buildConversationResponseByAccount(ctx, payload.AccountID, payload.ConversationID, payload.AssistantID, payload.TakeoverVersion) return err } diff --git a/backend/internal/service/captain_conversation_worker_test.go b/backend/internal/service/captain_conversation_worker_test.go index 1fe48e07..58143fcd 100644 --- a/backend/internal/service/captain_conversation_worker_test.go +++ b/backend/internal/service/captain_conversation_worker_test.go @@ -22,7 +22,7 @@ func setupCaptainConversationWorkerTest(t *testing.T) (*gorm.DB, *CaptainConvers dbName := fmt.Sprintf("file:%s?mode=memory&cache=private", t.Name()) db, err := gorm.Open(sqlite.Open(dbName), &gorm.Config{}) require.NoError(t, err) - require.NoError(t, db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.Contact{}, &model.Conversation{}, &model.Message{}, &model.Attachment{}, &model.AgentBot{}, &model.CaptainAssistant{}, &model.CaptainInbox{}, &model.BackgroundJob{})) + require.NoError(t, db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.Contact{}, &model.Conversation{}, &model.Message{}, &model.Attachment{}, &model.AgentBot{}, &model.AgentBotInbox{}, &model.CaptainAssistant{}, &model.CaptainInbox{}, &model.CaptainPreference{}, &model.BackgroundJob{})) t.Cleanup(func() { sqlDB, _ := db.DB() sqlDB.Close() @@ -38,6 +38,12 @@ func setupCaptainConversationWorkerTest(t *testing.T) (*gorm.DB, *CaptainConvers assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Fin", Config: []byte(`{"handoff_message":"Let me connect you."}`), Status: model.AssistantStatusActive} require.NoError(t, db.Create(assistant).Error) require.NoError(t, db.Create(&model.CaptainInbox{AccountID: account.ID, AssistantID: assistant.ID, InboxID: inbox.ID}).Error) + bot := &model.AgentBot{AccountID: &account.ID, Name: "Captain", BotType: "captain", Config: []byte(fmt.Sprintf(`{"assistant_id":%d}`, assistant.ID))} + require.NoError(t, db.Create(bot).Error) + require.NoError(t, db.Create(&model.AgentBotInbox{AgentBotID: bot.ID, InboxID: inbox.ID, Status: model.AgentBotInboxActive}).Error) + require.NoError(t, db.Model(conversation).Updates(map[string]any{"assignee_agent_bot_id": bot.ID, "ai_takeover_version": 1}).Error) + conversation.AssigneeAgentBotID = &bot.ID + conversation.AITakeoverVersion = 1 conversationSvc := NewCaptainConversationService(db, nil) messageSvc := NewMessageService(repository.NewMessageRepo(db), channel.NewDispatcher(), nil) return db, conversationSvc, messageSvc, account, inbox, conversation, assistant @@ -147,10 +153,9 @@ func TestCaptainConversationResponseSkipsNonPendingConversation(t *testing.T) { assert.Equal(t, int64(0), count) } -func TestCaptainInboxAutoResponseSkipsManuallyAssignedBot(t *testing.T) { +func TestCaptainInboxAutoResponseIsOffByDefault(t *testing.T) { db, _, messageSvc, account, _, conversation, _ := setupCaptainConversationWorkerTest(t) - botID := uint(42) - require.NoError(t, db.Model(&model.Conversation{}).Where("id = ?", conversation.ID).Update("assignee_agent_bot_id", botID).Error) + require.NoError(t, db.Model(&model.Conversation{}).Where("id = ?", conversation.ID).Update("assignee_agent_bot_id", nil).Error) wp := worker.NewWorkerPool(db) messageSvc.SetWorkerPool(wp) @@ -163,8 +168,9 @@ func TestCaptainInboxAutoResponseSkipsManuallyAssignedBot(t *testing.T) { } type fakeCaptainConversationBackend struct { - response *CaptainConversationResponse - err error + response *CaptainConversationResponse + err error + beforeReturn func() } func (b *fakeCaptainConversationBackend) GenerateCaptainConversationResponse(ctx context.Context, req CaptainConversationResponseRequest) (*CaptainConversationResponse, error) { @@ -174,5 +180,39 @@ func (b *fakeCaptainConversationBackend) GenerateCaptainConversationResponse(ctx if b.response == nil { return nil, errors.New("missing response") } + if b.beforeReturn != nil { + b.beforeReturn() + } return b.response, nil } + +func TestCaptainConversationResponseRejectsStaleJobAfterReentry(t *testing.T) { + db, conversationSvc, messageSvc, account, _, conversation, assistant := setupCaptainConversationWorkerTest(t) + conversationSvc.SetMessageService(messageSvc) + conversationSvc.SetResponseBackend(&fakeCaptainConversationBackend{response: &CaptainConversationResponse{Content: "stale"}}) + + require.NoError(t, db.Model(&model.Conversation{}).Where("id = ?", conversation.ID). + Updates(map[string]any{"ai_takeover_version": 3}).Error) + message, err := conversationSvc.buildConversationResponseByAccount(context.Background(), account.ID, conversation.ID, assistant.ID, 1) + require.NoError(t, err) + assert.Nil(t, message) + + var count int64 + require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ? AND message_type = ?", conversation.ID, model.MessageTypeOutgoing).Count(&count).Error) + assert.Zero(t, count) +} + +func TestCaptainConversationResponseRechecksTakeoverBeforeCreate(t *testing.T) { + db, conversationSvc, messageSvc, account, _, conversation, assistant := setupCaptainConversationWorkerTest(t) + conversationSvc.SetMessageService(messageSvc) + conversationSvc.SetResponseBackend(&fakeCaptainConversationBackend{ + response: &CaptainConversationResponse{Content: "late"}, + beforeReturn: func() { + require.NoError(t, db.Model(&model.Conversation{}).Where("id = ?", conversation.ID). + Updates(map[string]any{"assignee_agent_bot_id": nil, "status": model.ConversationStatusOpen, "ai_takeover_version": 2}).Error) + }, + }) + message, err := conversationSvc.buildConversationResponseByAccount(context.Background(), account.ID, conversation.ID, assistant.ID, 1) + require.NoError(t, err) + assert.Nil(t, message) +} diff --git a/backend/internal/service/conversation_service.go b/backend/internal/service/conversation_service.go index 475877a7..afe58394 100644 --- a/backend/internal/service/conversation_service.go +++ b/backend/internal/service/conversation_service.go @@ -537,6 +537,76 @@ func (s *ConversationService) AssignAgentBot(ctx context.Context, accountID, id, return conversation, &bot, nil } +// StartAITakeover activates the inbox's single configured Captain bot. The +// client deliberately supplies no bot ID: channel configuration is the source +// of truth for manual takeover. +func (s *ConversationService) StartAITakeover(ctx context.Context, accountID, id uint) (*model.Conversation, error) { + conversation, err := s.repo.FindByAccountAndID(ctx, accountID, id) + if err != nil { + return nil, err + } + var bindings []model.AgentBotInbox + if err := s.repo.DB().WithContext(ctx). + Where("inbox_id = ? AND status = ?", conversation.InboxID, model.AgentBotInboxActive). + Find(&bindings).Error; err != nil { + return nil, err + } + if len(bindings) != 1 { + return nil, errors.New("invalid inbox AI configuration: expected exactly one active AI") + } + var bot model.AgentBot + if err := s.repo.DB().WithContext(ctx). + Where("id = ? AND bot_type = ? AND (account_id IS NULL OR account_id = ?)", bindings[0].AgentBotID, "captain", accountID). + First(&bot).Error; err != nil { + return nil, errors.New("invalid inbox AI configuration: active AI is not a Captain bot") + } + assistantID := extractAssistantIDFromBotConfig(bot.Config) + if assistantID == 0 { + return nil, errors.New("active inbox AI has invalid assistant configuration") + } + var assistant model.CaptainAssistant + if err := s.repo.DB().WithContext(ctx). + Joins("JOIN captain_inboxes ON captain_inboxes.captain_assistant_id = captain_assistants.id"). + Where("captain_assistants.id = ? AND captain_assistants.account_id = ? AND captain_assistants.status = ? AND captain_inboxes.inbox_id = ?", assistantID, accountID, model.AssistantStatusActive, conversation.InboxID). + First(&assistant).Error; err != nil { + return nil, errors.New("invalid inbox AI configuration: assistant is disabled or not bound to this inbox") + } + if conversation.AssigneeAgentBotID != nil && *conversation.AssigneeAgentBotID == bot.ID && conversation.Status == string(model.ConversationStatusPending) { + return conversation, nil + } + if err := s.repo.StartAITakeover(ctx, conversation.ID, bot.ID); err != nil { + return nil, err + } + conversation.AssigneeID = nil + conversation.AssigneeAgentBotID = &bot.ID + conversation.Status = string(model.ConversationStatusPending) + conversation.AITakeoverVersion++ + s.dispatchConversationEvent(ctx, channel.EventConversationAssigned, conversation) + s.indexConversation(ctx, conversation) + return conversation, nil +} + +func (s *ConversationService) ExitAITakeover(ctx context.Context, accountID, id uint) (*model.Conversation, error) { + conversation, err := s.repo.FindByAccountAndID(ctx, accountID, id) + if err != nil { + return nil, err + } + if conversation.AssigneeAgentBotID == nil { + return conversation, nil + } + 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.AITakeoverVersion++ + s.dispatchConversationEvent(ctx, channel.EventConversationUnassigned, conversation) + s.indexConversation(ctx, conversation) + return conversation, nil +} + // UnassignAgent removes the agent assignment from a conversation. // Convenience wrapper around AssignAgent(id, 0) — sends EventConversationUnassigned. // Reference: Chatwoot conversations_controller.rb #unassign diff --git a/backend/internal/service/message_service.go b/backend/internal/service/message_service.go index 650cc9e8..c25d0c64 100644 --- a/backend/internal/service/message_service.go +++ b/backend/internal/service/message_service.go @@ -127,28 +127,29 @@ func (s *MessageService) Search(ctx context.Context, accountID uint, query strin // CreateMessageRequest is the DTO for creating a message. // Reference: Chatwoot app/controllers/api/v1/accounts/conversations/messages_controller.rb #create type CreateMessageRequest struct { - ConversationID uint `json:"conversation_id" validate:"required"` - Content string `json:"content"` - MessageType string `json:"message_type,omitempty"` - ContentType string `json:"content_type,omitempty"` - Private bool `json:"private,omitempty"` - SenderID uint `json:"sender_id,omitempty"` - SenderType string `json:"sender_type,omitempty"` - SourceID string `json:"source_id,omitempty"` - EchoID string `json:"echo_id,omitempty"` - ExternalCreatedAt string `json:"external_created_at,omitempty"` - External bool `json:"external,omitempty"` - ContentAttributes datatypes.JSON `json:"content_attributes,omitempty"` - AdditionalAttributes datatypes.JSON `json:"additional_attributes,omitempty"` - ExternalSourceIDs datatypes.JSON `json:"external_source_ids,omitempty"` - EmailHTMLContent string `json:"email_html_content,omitempty"` - CCEmails string `json:"cc_emails,omitempty"` - BCCEmails string `json:"bcc_emails,omitempty"` - ToEmails string `json:"to_emails,omitempty"` - CampaignID any `json:"campaign_id,omitempty"` - TemplateParams datatypes.JSON `json:"template_params,omitempty"` - IsVoiceMessage bool `json:"is_voice_message,omitempty"` - Attachments []MessageAttachmentInput `json:"-"` + ConversationID uint `json:"conversation_id" validate:"required"` + Content string `json:"content"` + MessageType string `json:"message_type,omitempty"` + ContentType string `json:"content_type,omitempty"` + Private bool `json:"private,omitempty"` + SenderID uint `json:"sender_id,omitempty"` + SenderType string `json:"sender_type,omitempty"` + SourceID string `json:"source_id,omitempty"` + EchoID string `json:"echo_id,omitempty"` + ExternalCreatedAt string `json:"external_created_at,omitempty"` + External bool `json:"external,omitempty"` + ContentAttributes datatypes.JSON `json:"content_attributes,omitempty"` + AdditionalAttributes datatypes.JSON `json:"additional_attributes,omitempty"` + ExternalSourceIDs datatypes.JSON `json:"external_source_ids,omitempty"` + EmailHTMLContent string `json:"email_html_content,omitempty"` + CCEmails string `json:"cc_emails,omitempty"` + BCCEmails string `json:"bcc_emails,omitempty"` + ToEmails string `json:"to_emails,omitempty"` + CampaignID any `json:"campaign_id,omitempty"` + TemplateParams datatypes.JSON `json:"template_params,omitempty"` + IsVoiceMessage bool `json:"is_voice_message,omitempty"` + Attachments []MessageAttachmentInput `json:"-"` + ExpectedAITakeoverVersion uint `json:"-"` } type MessageAttachmentInput struct { @@ -309,6 +310,36 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint var deliveryJob *model.BackgroundJob var deliveryCreated bool if err := s.repo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if req.ExpectedAITakeoverVersion != 0 { + var active model.Conversation + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id = ? AND account_id = ? AND status = ? AND assignee_agent_bot_id IS NOT NULL AND ai_takeover_version = ?", message.ConversationID, accountID, model.ConversationStatusPending, req.ExpectedAITakeoverVersion). + First(&active).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return errAITakeoverEnded + } + return err + } + if active.ID == 0 { + return errAITakeoverEnded + } + } + humanOutgoing := !req.External && !message.Private && + (message.MessageType == string(model.MessageTypeOutgoing) || message.MessageType == string(model.MessageTypeTemplate)) && + message.SenderType != string(model.SenderTypeAgentBot) && message.SenderType != "Captain::Assistant" + if humanOutgoing { + result := tx.Model(&model.Conversation{}). + Where("id = ? AND account_id = ? AND assignee_agent_bot_id IS NOT NULL", message.ConversationID, accountID). + Updates(map[string]any{ + "assignee_agent_bot_id": nil, + "status": gorm.Expr("CASE WHEN status = ? THEN ? ELSE status END", model.ConversationStatusPending, model.ConversationStatusOpen), + "ai_takeover_version": gorm.Expr("ai_takeover_version + 1"), + }) + if result.Error != nil { + return result.Error + } + message.AITakeoverExited = result.RowsAffected > 0 + } if err := tx.Create(message).Error; err != nil { return err } @@ -355,6 +386,9 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint } return nil }); err != nil { + if errors.Is(err, errAITakeoverEnded) { + return nil, nil + } if requestHash != "" { if existing, found, replayErr := s.findShangwutongMessageReplay(ctx, conversation.InboxID, req.SourceID, requestHash); replayErr == nil && found { return existing, nil @@ -394,6 +428,8 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint return message, nil } +var errAITakeoverEnded = errors.New("AI takeover ended") + func messageCreateAllowedIncoming(channelType string) bool { return channel.IsAPIInboxLike(channelType) } diff --git a/backend/internal/service/message_service_test.go b/backend/internal/service/message_service_test.go index 274821e2..ab82a1d7 100644 --- a/backend/internal/service/message_service_test.go +++ b/backend/internal/service/message_service_test.go @@ -450,6 +450,7 @@ func TestMessageService_Create(t *testing.T) { require.NoError(t, db.AutoMigrate(&model.AgentBot{})) agentBot := &model.AgentBot{AccountID: &account.ID, Name: "Reply Bot", BotType: "webhook"} require.NoError(t, db.Create(agentBot).Error) + require.NoError(t, db.Model(&conv).Update("assignee_agent_bot_id", agentBot.ID).Error) reqAgentBotSender := CreateMessageRequest{ ConversationID: conv.ID, Content: "Agent bot reply", @@ -470,9 +471,8 @@ func TestMessageService_Create(t *testing.T) { SenderID: agentBot.ID + 999, } createdFallbackMessage, errFallbackMessage := svc.Create(ctx, account.ID, user.ID, reqMissingAgentBotSender) - assert.NoError(t, errFallbackMessage) - assert.Equal(t, "user", createdFallbackMessage.SenderType) - assert.Equal(t, user.ID, *createdFallbackMessage.SenderID) + assert.Nil(t, createdFallbackMessage) + assert.EqualError(t, errFallbackMessage, "agent bot not found") // Chatwoot parity: is_voice_message tags only audio attachments reqVoiceMessage := CreateMessageRequest{ diff --git a/backend/migrations/000077_add_ai_takeover_version.down.sql b/backend/migrations/000077_add_ai_takeover_version.down.sql new file mode 100644 index 00000000..94e316d6 --- /dev/null +++ b/backend/migrations/000077_add_ai_takeover_version.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE conversations + DROP COLUMN IF EXISTS ai_takeover_version; diff --git a/backend/migrations/000077_add_ai_takeover_version.up.sql b/backend/migrations/000077_add_ai_takeover_version.up.sql new file mode 100644 index 00000000..7ad4869d --- /dev/null +++ b/backend/migrations/000077_add_ai_takeover_version.up.sql @@ -0,0 +1,2 @@ +ALTER TABLE conversations + ADD COLUMN IF NOT EXISTS ai_takeover_version BIGINT NOT NULL DEFAULT 0; diff --git a/frontend/app/javascript/dashboard/api/assignableAgents.js b/frontend/app/javascript/dashboard/api/assignableAgents.js index cad7afdd..5b999fac 100644 --- a/frontend/app/javascript/dashboard/api/assignableAgents.js +++ b/frontend/app/javascript/dashboard/api/assignableAgents.js @@ -8,7 +8,7 @@ class AssignableAgents extends ApiClient { get(inboxIds) { return axios.get(this.url, { - params: { inbox_ids: inboxIds, include_agent_bots: true }, + params: { inbox_ids: inboxIds }, }); } } diff --git a/frontend/app/javascript/dashboard/api/inbox/conversation.js b/frontend/app/javascript/dashboard/api/inbox/conversation.js index 08820aac..e0e393ec 100644 --- a/frontend/app/javascript/dashboard/api/inbox/conversation.js +++ b/frontend/app/javascript/dashboard/api/inbox/conversation.js @@ -62,13 +62,20 @@ class ConversationApi extends ApiClient { }); } - assignAgent({ conversationId, agentId, assigneeType }) { + assignAgent({ conversationId, agentId }) { return axios.post(`${this.url}/${conversationId}/assignments`, { assignee_id: agentId, - assignee_type: assigneeType, }); } + startAITakeover(conversationId) { + return axios.post(`${this.url}/${conversationId}/ai_takeover`); + } + + exitAITakeover(conversationId) { + return axios.delete(`${this.url}/${conversationId}/ai_takeover`); + } + assignTeam({ conversationId, teamId }) { const params = { team_id: teamId }; return axios.post(`${this.url}/${conversationId}/assignments`, params); @@ -111,7 +118,9 @@ class ConversationApi extends ApiClient { } sendEmailTranscript({ conversationId, email }) { - return axios.post(`${this.url}/${conversationId}/transcript`, { email }); + return axios.post(`${this.url}/${conversationId}/transcript`, { + email, + }); } updateCustomAttributes({ conversationId, customAttributes }) { diff --git a/frontend/app/javascript/dashboard/api/specs/assignableAgents.spec.js b/frontend/app/javascript/dashboard/api/specs/assignableAgents.spec.js index a9411597..d553d55c 100644 --- a/frontend/app/javascript/dashboard/api/specs/assignableAgents.spec.js +++ b/frontend/app/javascript/dashboard/api/specs/assignableAgents.spec.js @@ -23,7 +23,6 @@ describe('#AssignableAgentsAPI', () => { expect(axiosMock.get).toHaveBeenCalledWith('/api/v1/assignable_agents', { params: { inbox_ids: [1], - include_agent_bots: true, }, }); }); diff --git a/frontend/app/javascript/dashboard/api/specs/inbox/conversation.spec.js b/frontend/app/javascript/dashboard/api/specs/inbox/conversation.spec.js index ea0ef3e7..badc1086 100644 --- a/frontend/app/javascript/dashboard/api/specs/inbox/conversation.spec.js +++ b/frontend/app/javascript/dashboard/api/specs/inbox/conversation.spec.js @@ -12,6 +12,8 @@ describe('#ConversationAPI', () => { expect(conversationAPI).toHaveProperty('toggleStatus'); expect(conversationAPI).toHaveProperty('assignAgent'); expect(conversationAPI).toHaveProperty('assignTeam'); + expect(conversationAPI).toHaveProperty('startAITakeover'); + expect(conversationAPI).toHaveProperty('exitAITakeover'); expect(conversationAPI).toHaveProperty('markMessageRead'); expect(conversationAPI).toHaveProperty('toggleTyping'); expect(conversationAPI).toHaveProperty('mute'); @@ -79,7 +81,10 @@ describe('#ConversationAPI', () => { }); it('#toggleStatus', () => { - conversationAPI.toggleStatus({ conversationId: 12, status: 'online' }); + conversationAPI.toggleStatus({ + conversationId: 12, + status: 'online', + }); expect(axiosMock.post).toHaveBeenCalledWith( `/api/v1/conversations/12/toggle_status`, { @@ -93,17 +98,29 @@ describe('#ConversationAPI', () => { conversationAPI.assignAgent({ conversationId: 12, agentId: 34, - assigneeType: 'AgentBot', }); expect(axiosMock.post).toHaveBeenCalledWith( `/api/v1/conversations/12/assignments`, { assignee_id: 34, - assignee_type: 'AgentBot', } ); }); + it('#startAITakeover', () => { + conversationAPI.startAITakeover(12); + expect(axiosMock.post).toHaveBeenCalledWith( + '/api/v1/conversations/12/ai_takeover' + ); + }); + + it('#exitAITakeover', () => { + conversationAPI.exitAITakeover(12); + expect(axiosMock.delete).toHaveBeenCalledWith( + '/api/v1/conversations/12/ai_takeover' + ); + }); + it('#assignTeam', () => { conversationAPI.assignTeam({ conversationId: 12, teamId: 1 }); expect(axiosMock.post).toHaveBeenCalledWith( diff --git a/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue b/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue index 7f10190e..1fb96654 100644 --- a/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue +++ b/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBox.vue @@ -885,10 +885,14 @@ export default { copilotAcceptedMessage = '' ) { try { - await this.$store.dispatch( + const sentMessage = await this.$store.dispatch( 'createPendingMessageAndSend', messagePayload ); + 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); this.removeFromDraft(); diff --git a/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBoxBanner.vue b/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBoxBanner.vue index 8be715a4..f5d79546 100644 --- a/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBoxBanner.vue +++ b/frontend/app/javascript/dashboard/components/widgets/conversation/ReplyBoxBanner.vue @@ -62,6 +62,29 @@ const showBotHandoffBanner = computed( currentChat.value?.status === wootConstants.STATUS_TYPE.PENDING ); +const aiTakeoverActive = computed( + () => currentChat.value?.ai_takeover_active === true +); + +const toggleAITakeover = async () => { + try { + const wasActive = aiTakeoverActive.value; + const action = wasActive ? 'exitAITakeover' : 'startAITakeover'; + await store.dispatch(action, currentChat.value?.id); + useAlert( + t( + wasActive + ? 'CONVERSATION.AI_TAKEOVER.EXIT_SUCCESS' + : 'CONVERSATION.AI_TAKEOVER.START_SUCCESS' + ) + ); + } catch (error) { + useAlert( + error?.response?.data?.error || t('CONVERSATION.AI_TAKEOVER.ERROR') + ); + } +}; + const botHandoffActionLabel = computed(() => { return assignedAgent.value?.id === currentUser.value?.id ? t('CONVERSATION.BOT_HANDOFF_REOPEN_ACTION') @@ -109,6 +132,19 @@ const onClickBotHandoff = async () => {