feat(captain): queue copilot response jobs

internal/app/bootstrap.go

@@ -589,6 +589,10 @@ func Bootstrap(env string) (*App, error) {
 	captainScenarioService := service.NewCaptainScenarioService(captainScenarioRepo, captainAssistantRepo)
 	captainCustomToolService := service.NewCaptainCustomToolService(captainCustomToolRepo)
 	copilotService := service.NewCopilotService(copilotThreadRepo, copilotMessageRepo, copilotSuggestionRepo, llmProvider, captainAssistantRepo)
+	copilotService.SetWorkerPool(workerPool)
+	captainConversationService := service.NewCaptainConversationService(db, llmProvider)
+	captainConversationService.SetWorkerPool(workerPool)
+	_ = captainConversationService
 	copilotContextService := service.NewCopilotContextService(messageRepo, conversationRepo, contactRepo, llmProvider)
 	captainTaskService := service.NewCaptainTaskService(captainAssistantRepo, captainAssistantResponseRepo, captainCustomToolRepo, conversationRepo, messageRepo, llmProvider, copilotContextService, copilotSuggestionRepo)
 	conversationInsightService := service.NewConversationInsightService(conversationRepo, messageRepo, captainAssistantRepo, llmProvider)

internal/service/captain_conversation_service.go

@@ -0,0 +1,299 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gochat/gochat/internal/llm"
+	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
+	"gorm.io/datatypes"
+	"gorm.io/gorm"
+)
+
+const TaskTypeCaptainConversationResponseBuilder = "captain:conversation_response_builder"
+
+type CaptainConversationResponseBackend interface {
+	GenerateCaptainConversationResponse(ctx context.Context, req CaptainConversationResponseRequest) (*CaptainConversationResponse, error)
+}
+
+type CaptainConversationResponseRequest struct {
+	AccountID    uint
+	Conversation *model.Conversation
+	Assistant    *model.CaptainAssistant
+	Messages     []CaptainConversationMessage
+}
+
+type CaptainConversationMessage struct {
+	Role      string
+	Content   string
+	AgentName string
+}
+
+type CaptainConversationResponse struct {
+	Content           string
+	Action            string
+	AgentName         string
+	HandoffToolCalled bool
+}
+
+type CaptainConversationService struct {
+	db          *gorm.DB
+	llmProvider llm.Provider
+	backend     CaptainConversationResponseBackend
+	worker      *worker.WorkerPool
+}
+
+func NewCaptainConversationService(db *gorm.DB, llmProvider llm.Provider) *CaptainConversationService {
+	return &CaptainConversationService{db: db, llmProvider: llmProvider}
+}
+
+func (s *CaptainConversationService) SetResponseBackend(backend CaptainConversationResponseBackend) {
+	s.backend = backend
+}
+
+func (s *CaptainConversationService) SetWorkerPool(wp *worker.WorkerPool) {
+	s.worker = wp
+	RegisterCaptainConversationJobs(wp, s)
+}
+
+func (s *CaptainConversationService) BuildConversationResponseByAccount(ctx context.Context, accountID, conversationID, assistantID uint) (*model.Message, error) {
+	if s == nil || s.db == nil {
+		return nil, fmt.Errorf("captain conversation service database is required")
+	}
+	var conversation model.Conversation
+	if err := s.db.WithContext(ctx).Where("account_id = ? AND id = ?", accountID, conversationID).First(&conversation).Error; err != nil {
+		return nil, fmt.Errorf("conversation not found: %w", err)
+	}
+	if conversation.Status != string(model.ConversationStatusPending) {
+		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)
+	}
+
+	history, err := s.collectConversationMessages(ctx, accountID, conversation.ID)
+	if err != nil {
+		return nil, err
+	}
+	response, err := s.generateConversationResponse(ctx, accountID, &conversation, &assistant, history)
+	if err != nil {
+		return nil, err
+	}
+	if response == nil {
+		return nil, fmt.Errorf("empty captain conversation response")
+	}
+	if response.Action == "handoff" || response.Content == "conversation_handoff" || response.HandoffToolCalled {
+		return s.createCaptainHandoffMessage(ctx, &conversation, &assistant)
+	}
+	if strings.TrimSpace(response.Content) == "" {
+		return nil, fmt.Errorf("captain conversation response content is blank")
+	}
+	return s.createCaptainOutgoingMessage(ctx, &conversation, &assistant, response.Content, response.AgentName)
+}
+
+func (s *CaptainConversationService) collectConversationMessages(ctx context.Context, accountID, conversationID uint) ([]CaptainConversationMessage, error) {
+	var messages []model.Message
+	if err := s.db.WithContext(ctx).
+		Where("account_id = ? AND conversation_id = ? AND private = ? AND message_type IN ?", accountID, conversationID, false, []string{string(model.MessageTypeIncoming), string(model.MessageTypeOutgoing)}).
+		Order("created_at ASC, id ASC").
+		Find(&messages).Error; err != nil {
+		return nil, fmt.Errorf("load conversation messages: %w", err)
+	}
+	history := make([]CaptainConversationMessage, 0, len(messages))
+	for _, msg := range messages {
+		role := "assistant"
+		if msg.MessageType == string(model.MessageTypeIncoming) {
+			role = "user"
+		}
+		history = append(history, CaptainConversationMessage{Role: role, Content: msg.Content, AgentName: messageAgentName(msg.AdditionalAttributes)})
+	}
+	return history, nil
+}
+
+func (s *CaptainConversationService) generateConversationResponse(ctx context.Context, accountID uint, conversation *model.Conversation, assistant *model.CaptainAssistant, history []CaptainConversationMessage) (*CaptainConversationResponse, error) {
+	if s.backend != nil {
+		return s.backend.GenerateCaptainConversationResponse(ctx, CaptainConversationResponseRequest{AccountID: accountID, Conversation: conversation, Assistant: assistant, Messages: history})
+	}
+	if s.llmProvider == nil {
+		return nil, fmt.Errorf("captain conversation response generation disabled")
+	}
+	messages := []llm.ChatMessage{{Role: "system", Content: "You are Captain, a customer support assistant. Reply to the latest customer message. Return only the reply text."}}
+	for _, msg := range history {
+		if strings.TrimSpace(msg.Content) == "" {
+			continue
+		}
+		messages = append(messages, llm.ChatMessage{Role: msg.Role, Content: msg.Content})
+	}
+	resp, err := s.llmProvider.ChatCompletion(ctx, llm.ChatRequest{Model: "gpt-4", Messages: messages, Temperature: 0.7, MaxTokens: 1024})
+	if err != nil {
+		return nil, fmt.Errorf("generate captain conversation response: %w", err)
+	}
+	if len(resp.Choices) == 0 || strings.TrimSpace(resp.Choices[0].Message.Content) == "" {
+		return nil, fmt.Errorf("empty captain conversation response")
+	}
+	return &CaptainConversationResponse{Content: resp.Choices[0].Message.Content}, nil
+}
+
+func (s *CaptainConversationService) createCaptainOutgoingMessage(ctx context.Context, conversation *model.Conversation, assistant *model.CaptainAssistant, content, agentName string) (*model.Message, error) {
+	senderID := assistant.ID
+	attrs := datatypes.JSON([]byte(`{}`))
+	if strings.TrimSpace(agentName) != "" {
+		raw, _ := json.Marshal(map[string]any{"agent_name": strings.TrimSpace(agentName)})
+		attrs = datatypes.JSON(raw)
+	}
+	message := &model.Message{
+		AccountID:            conversation.AccountID,
+		ConversationID:       conversation.ID,
+		InboxID:              conversation.InboxID,
+		SenderID:             &senderID,
+		SenderType:           "CaptainAssistant",
+		Content:              content,
+		ContentType:          string(model.MessageContentTypeText),
+		MessageType:          string(model.MessageTypeOutgoing),
+		Status:               string(model.MessageStatusSent),
+		AdditionalAttributes: attrs,
+	}
+	if err := s.db.WithContext(ctx).Create(message).Error; err != nil {
+		return nil, fmt.Errorf("create captain outgoing message: %w", err)
+	}
+	if s.worker != nil {
+		if _, err := EnqueueSendReply(ctx, s.worker, message.ID); err != nil {
+			return message, err
+		}
+	}
+	return message, nil
+}
+
+func (s *CaptainConversationService) createCaptainHandoffMessage(ctx context.Context, conversation *model.Conversation, assistant *model.CaptainAssistant) (*model.Message, error) {
+	content := assistantConfigString(assistant.Config, "handoff_message")
+	if content == "" {
+		content = "Transferring to another agent for further assistance."
+	}
+	message, err := s.createCaptainOutgoingMessage(ctx, conversation, assistant, content, "")
+	if err != nil {
+		return nil, err
+	}
+	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)
+	}
+	return message, nil
+}
+
+func messageAgentName(raw datatypes.JSON) string {
+	if len(raw) == 0 {
+		return ""
+	}
+	var attrs map[string]any
+	if err := json.Unmarshal(raw, &attrs); err != nil {
+		return ""
+	}
+	if value, ok := attrs["agent_name"].(string); ok {
+		return value
+	}
+	return ""
+}
+
+func assistantConfigString(raw []byte, key string) string {
+	if len(raw) == 0 {
+		return ""
+	}
+	var cfg map[string]any
+	if err := json.Unmarshal(raw, &cfg); err != nil {
+		return ""
+	}
+	if value, ok := cfg[key].(string); ok {
+		return strings.TrimSpace(value)
+	}
+	return ""
+}
+
+type captainConversationResponseBuilderJob struct {
+	AccountID      uint `json:"account_id"`
+	ConversationID uint `json:"conversation_id"`
+	AssistantID    uint `json:"assistant_id"`
+	MessageID      uint `json:"message_id,omitempty"`
+}
+
+var captainConversationRegistrations sync.Map
+
+func RegisterCaptainConversationJobs(wp *worker.WorkerPool, svc *CaptainConversationService) {
+	if wp == nil || svc == nil {
+		return
+	}
+	if _, loaded := captainConversationRegistrations.LoadOrStore(wp, struct{}{}); loaded {
+		return
+	}
+	wp.Register(TaskTypeCaptainConversationResponseBuilder, svc.performConversationResponseBuilderJob)
+}
+
+func EnqueueCaptainConversationResponseForMessage(ctx context.Context, wp *worker.WorkerPool, db *gorm.DB, messageID uint) (*model.BackgroundJob, error) {
+	if wp == nil || db == nil || messageID == 0 || !db.Migrator().HasTable(&model.CaptainInbox{}) {
+		return nil, nil
+	}
+	var message model.Message
+	if err := db.WithContext(ctx).First(&message, messageID).Error; err != nil {
+		return nil, err
+	}
+	if message.MessageType != string(model.MessageTypeIncoming) || message.Private {
+		return nil, nil
+	}
+	var conversation model.Conversation
+	if err := db.WithContext(ctx).Where("account_id = ? AND id = ?", message.AccountID, message.ConversationID).First(&conversation).Error; err != nil {
+		return nil, err
+	}
+	if conversation.Status != string(model.ConversationStatusPending) {
+		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
+		}
+		return nil, err
+	}
+	options := []worker.EnqueueOption{
+		worker.WithMaxAttempts(3),
+		worker.WithIdempotencyKey(fmt.Sprintf("captain:conversation_response_builder:message:%d", message.ID)),
+	}
+	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...)
+}
+
+func captainConversationResponseScheduledAt(ctx context.Context, db *gorm.DB, messageID uint) time.Time {
+	if !db.Migrator().HasTable(&model.Attachment{}) {
+		return time.Time{}
+	}
+	var count int64
+	if err := db.WithContext(ctx).Model(&model.Attachment{}).Where("message_id = ?", messageID).Count(&count).Error; err != nil || count == 0 {
+		return time.Time{}
+	}
+	wait := time.Second + time.Duration(minInt64(count, 4))*time.Second
+	return time.Now().Add(wait)
+}
+
+func minInt64(a, b int64) int64 {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func (s *CaptainConversationService) performConversationResponseBuilderJob(ctx context.Context, job *model.BackgroundJob) error {
+	var payload captainConversationResponseBuilderJob
+	if err := json.Unmarshal(job.Payload, &payload); err != nil {
+		return fmt.Errorf("unmarshal captain conversation response builder job: %w", err)
+	}
+	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)
+	return err
+}

internal/service/captain_conversation_worker_test.go

@@ -0,0 +1,140 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/gochat/gochat/internal/channel"
+	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/repository"
+	"github.com/gochat/gochat/internal/worker"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+)
+
+func setupCaptainConversationWorkerTest(t *testing.T) (*gorm.DB, *CaptainConversationService, *MessageService, *model.Account, *model.Inbox, *model.Conversation, *model.CaptainAssistant) {
+	t.Helper()
+	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.CaptainAssistant{}, &model.CaptainInbox{}, &model.BackgroundJob{}))
+	t.Cleanup(func() {
+		sqlDB, _ := db.DB()
+		sqlDB.Close()
+	})
+	account := &model.Account{Name: "Captain Org", Active: true}
+	require.NoError(t, db.Create(account).Error)
+	inbox := &model.Inbox{AccountID: account.ID, Name: "Web", ChannelType: "web_widget", ChannelID: 1}
+	require.NoError(t, db.Create(inbox).Error)
+	contact := &model.Contact{AccountID: account.ID, Name: "Customer", Email: "customer@example.com"}
+	require.NoError(t, db.Create(contact).Error)
+	conversation := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusPending), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType}
+	require.NoError(t, db.Create(conversation).Error)
+	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)
+	conversationSvc := NewCaptainConversationService(db, nil)
+	messageSvc := NewMessageService(repository.NewMessageRepo(db), channel.NewDispatcher(), nil)
+	return db, conversationSvc, messageSvc, account, inbox, conversation, assistant
+}
+
+func TestCaptainConversationResponseJobQueuesFromIncomingMessage(t *testing.T) {
+	db, conversationSvc, messageSvc, account, _, conversation, assistant := setupCaptainConversationWorkerTest(t)
+	conversationSvc.SetResponseBackend(&fakeCaptainConversationBackend{response: &CaptainConversationResponse{Content: "Welcome to Captain", AgentName: "Fin"}})
+	wp := worker.NewWorkerPool(db)
+	conversationSvc.SetWorkerPool(wp)
+	messageSvc.SetWorkerPool(wp)
+
+	incoming, err := messageSvc.Create(context.Background(), account.ID, 99, CreateMessageRequest{ConversationID: conversation.ID, Content: "Hello", MessageType: string(model.MessageTypeIncoming), ContentType: string(model.MessageContentTypeText)})
+	require.NoError(t, err)
+	require.NotZero(t, incoming.ID)
+
+	var count int64
+	require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeCaptainConversationResponseBuilder, model.BackgroundJobStatusQueued).Count(&count).Error)
+	assert.Equal(t, int64(1), count)
+
+	processed, err := wp.ProcessOne(context.Background())
+	require.NoError(t, err)
+	assert.True(t, processed)
+
+	var outgoing model.Message
+	require.NoError(t, db.Where("conversation_id = ? AND message_type = ?", conversation.ID, model.MessageTypeOutgoing).First(&outgoing).Error)
+	assert.Equal(t, assistant.ID, *outgoing.SenderID)
+	assert.Equal(t, "CaptainAssistant", outgoing.SenderType)
+	assert.Equal(t, "Welcome to Captain", outgoing.Content)
+
+	require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeMessageSendReply).Count(&count).Error)
+	assert.Equal(t, int64(1), count)
+}
+
+func TestCaptainConversationResponseJobHandoffOpensConversation(t *testing.T) {
+	db, conversationSvc, _, account, _, conversation, _ := setupCaptainConversationWorkerTest(t)
+	conversationSvc.SetResponseBackend(&fakeCaptainConversationBackend{response: &CaptainConversationResponse{Action: "handoff"}})
+	wp := worker.NewWorkerPool(db)
+	conversationSvc.SetWorkerPool(wp)
+
+	_, err := wp.Enqueue(context.Background(), TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: account.ID, ConversationID: conversation.ID, AssistantID: 1, MessageID: 1}, worker.WithMaxAttempts(3))
+	require.NoError(t, err)
+	processed, err := wp.ProcessOne(context.Background())
+	require.NoError(t, err)
+	assert.True(t, processed)
+
+	var updated model.Conversation
+	require.NoError(t, db.First(&updated, conversation.ID).Error)
+	assert.Equal(t, string(model.ConversationStatusOpen), updated.Status)
+	var outgoing model.Message
+	require.NoError(t, db.Where("conversation_id = ? AND message_type = ?", conversation.ID, model.MessageTypeOutgoing).First(&outgoing).Error)
+	assert.Equal(t, "Let me connect you.", outgoing.Content)
+}
+
+func TestCaptainConversationResponseJobRetriesWhenProviderDisabled(t *testing.T) {
+	db, conversationSvc, _, account, _, conversation, _ := setupCaptainConversationWorkerTest(t)
+	now := time.Date(2026, 6, 6, 4, 30, 0, 0, time.UTC)
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute }))
+	conversationSvc.SetWorkerPool(wp)
+
+	_, err := wp.Enqueue(context.Background(), TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: account.ID, ConversationID: conversation.ID, AssistantID: 1, MessageID: 1}, worker.WithMaxAttempts(3))
+	require.NoError(t, err)
+	processed, err := wp.ProcessOne(context.Background())
+	require.Error(t, err)
+	assert.True(t, processed)
+
+	var job model.BackgroundJob
+	require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).First(&job).Error)
+	assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status)
+	assert.Contains(t, job.LastError, "captain conversation response generation disabled")
+}
+
+func TestCaptainConversationResponseSkipsNonPendingConversation(t *testing.T) {
+	db, conversationSvc, messageSvc, account, _, conversation, _ := setupCaptainConversationWorkerTest(t)
+	require.NoError(t, db.Model(&model.Conversation{}).Where("id = ?", conversation.ID).Update("status", string(model.ConversationStatusOpen)).Error)
+	wp := worker.NewWorkerPool(db)
+	conversationSvc.SetWorkerPool(wp)
+	messageSvc.SetWorkerPool(wp)
+
+	_, err := messageSvc.Create(context.Background(), account.ID, 99, CreateMessageRequest{ConversationID: conversation.ID, Content: "Hello", MessageType: string(model.MessageTypeIncoming), ContentType: string(model.MessageContentTypeText)})
+	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(0), count)
+}
+
+type fakeCaptainConversationBackend struct {
+	response *CaptainConversationResponse
+	err      error
+}
+
+func (b *fakeCaptainConversationBackend) GenerateCaptainConversationResponse(ctx context.Context, req CaptainConversationResponseRequest) (*CaptainConversationResponse, error) {
+	if b.err != nil {
+		return nil, b.err
+	}
+	if b.response == nil {
+		return nil, errors.New("missing response")
+	}
+	return b.response, nil
+}

internal/service/copilot_response_worker.go

@@ -0,0 +1,46 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
+)
+
+const TaskTypeCaptainCopilotResponse = "captain:copilot_response"
+
+type captainCopilotResponseJob struct {
+	AccountID       uint   `json:"account_id"`
+	UserID          uint   `json:"user_id"`
+	ConversationID  uint   `json:"conversation_id,omitempty"`
+	CopilotThreadID uint   `json:"copilot_thread_id"`
+	MessageID       uint   `json:"message_id"`
+	Message         string `json:"message,omitempty"`
+}
+
+var copilotResponseRegistrations sync.Map
+
+func RegisterCopilotResponseJobs(wp *worker.WorkerPool, svc *CopilotService) {
+	if wp == nil || svc == nil {
+		return
+	}
+	if _, loaded := copilotResponseRegistrations.LoadOrStore(wp, struct{}{}); loaded {
+		return
+	}
+	wp.Register(TaskTypeCaptainCopilotResponse, svc.performCopilotResponseJob)
+}
+
+func (s *CopilotService) performCopilotResponseJob(ctx context.Context, job *model.BackgroundJob) error {
+	var payload captainCopilotResponseJob
+	if err := json.Unmarshal(job.Payload, &payload); err != nil {
+		return fmt.Errorf("unmarshal captain copilot response job: %w", err)
+	}
+	if payload.AccountID == 0 || payload.UserID == 0 || payload.CopilotThreadID == 0 || payload.MessageID == 0 {
+		return fmt.Errorf("invalid captain copilot response job payload: %#v", payload)
+	}
+	_, err := s.GenerateCopilotResponseByAccount(ctx, payload.AccountID, payload.UserID, payload.CopilotThreadID, payload.ConversationID, payload.MessageID, payload.Message)
+	return err
+}

internal/service/copilot_response_worker_test.go

@@ -0,0 +1,113 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/repository"
+	"github.com/gochat/gochat/internal/worker"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+)
+
+func setupCopilotResponseWorkerTest(t *testing.T) (*gorm.DB, *CopilotService, *model.Account, *model.User, *model.CaptainAssistant) {
+	t.Helper()
+	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.User{}, &model.CaptainAssistant{}, &model.CopilotThread{}, &model.CopilotMessage{}, &model.CopilotSuggestionMessage{}, &model.BackgroundJob{}))
+	t.Cleanup(func() {
+		sqlDB, _ := db.DB()
+		sqlDB.Close()
+	})
+	account := &model.Account{Name: "Copilot Org", Active: true}
+	require.NoError(t, db.Create(account).Error)
+	user := &model.User{AccountID: account.ID, Name: "Agent", Email: "agent@example.com", Password: "secret", Active: true}
+	require.NoError(t, db.Create(user).Error)
+	assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Helper", Config: json.RawMessage(`{}`), Status: model.AssistantStatusActive}
+	require.NoError(t, db.Create(assistant).Error)
+	svc := NewCopilotService(repository.NewCopilotThreadRepo(db), repository.NewCopilotMessageRepo(db), repository.NewCopilotSuggestionRepo(db), nil, repository.NewCaptainAssistantRepo(db))
+	return db, svc, account, user, assistant
+}
+
+func TestCopilotResponseJobQueuesAndPersistsAssistantReply(t *testing.T) {
+	db, svc, account, user, assistant := setupCopilotResponseWorkerTest(t)
+	svc.SetResponseBackend(&fakeCopilotResponseBackend{messages: []CopilotGeneratedMessage{{MessageType: model.CopilotMessageTypeAssistant, Message: map[string]any{"content": "Here is the answer"}}}})
+	wp := worker.NewWorkerPool(db)
+	svc.SetWorkerPool(wp)
+
+	thread, err := svc.CreateThread(context.Background(), account.ID, user.ID, &CreateThreadRequest{Message: "Need help", AssistantID: assistant.ID, ConversationID: 123})
+	require.NoError(t, err)
+	require.NotNil(t, thread)
+
+	var count int64
+	require.NoError(t, db.Model(&model.CopilotMessage{}).Where("copilot_thread_id = ?", thread.ID).Count(&count).Error)
+	assert.Equal(t, int64(1), count)
+	require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeCaptainCopilotResponse, model.BackgroundJobStatusQueued).Count(&count).Error)
+	assert.Equal(t, int64(1), count)
+
+	processed, err := wp.ProcessOne(context.Background())
+	require.NoError(t, err)
+	assert.True(t, processed)
+
+	var messages []model.CopilotMessage
+	require.NoError(t, db.Where("copilot_thread_id = ?", thread.ID).Order("id ASC").Find(&messages).Error)
+	require.Len(t, messages, 2)
+	assert.Equal(t, model.CopilotMessageTypeUser, messages[0].MessageType)
+	assert.Equal(t, model.CopilotMessageTypeAssistant, messages[1].MessageType)
+	assert.Equal(t, "Here is the answer", messages[1].GetMessageContent())
+}
+
+func TestCopilotResponseJobUsesDisabledFallbackWithoutProvider(t *testing.T) {
+	db, svc, account, user, assistant := setupCopilotResponseWorkerTest(t)
+	wp := worker.NewWorkerPool(db)
+	svc.SetWorkerPool(wp)
+
+	thread, err := svc.CreateThread(context.Background(), account.ID, user.ID, &CreateThreadRequest{Message: "Need help", AssistantID: assistant.ID})
+	require.NoError(t, err)
+	processed, err := wp.ProcessOne(context.Background())
+	require.NoError(t, err)
+	assert.True(t, processed)
+
+	var assistantMsg model.CopilotMessage
+	require.NoError(t, db.Where("copilot_thread_id = ? AND message_type = ?", thread.ID, model.CopilotMessageTypeAssistant).First(&assistantMsg).Error)
+	assert.Equal(t, CopilotUnavailableMessage, assistantMsg.GetMessageContent())
+}
+
+func TestCopilotResponseJobRetriesBackendFailure(t *testing.T) {
+	db, svc, account, user, assistant := setupCopilotResponseWorkerTest(t)
+	svc.SetResponseBackend(&fakeCopilotResponseBackend{err: errors.New("backend down")})
+	now := time.Date(2026, 6, 6, 4, 0, 0, 0, time.UTC)
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute }))
+	svc.SetWorkerPool(wp)
+
+	_, err := svc.CreateThread(context.Background(), account.ID, user.ID, &CreateThreadRequest{Message: "Need help", AssistantID: assistant.ID})
+	require.NoError(t, err)
+	processed, err := wp.ProcessOne(context.Background())
+	require.Error(t, err)
+	assert.True(t, processed)
+
+	var job model.BackgroundJob
+	require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainCopilotResponse).First(&job).Error)
+	assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status)
+	assert.Contains(t, job.LastError, "backend down")
+}
+
+type fakeCopilotResponseBackend struct {
+	messages []CopilotGeneratedMessage
+	err      error
+}
+
+func (b *fakeCopilotResponseBackend) GenerateCopilotResponse(ctx context.Context, req CopilotResponseRequest) ([]CopilotGeneratedMessage, error) {
+	if b.err != nil {
+		return nil, b.err
+	}
+	return b.messages, nil
+}

internal/service/copilot_service.go

@@ -10,17 +10,38 @@ import (
 	"github.com/gochat/gochat/internal/llm"
 	"github.com/gochat/gochat/internal/model"
 	"github.com/gochat/gochat/internal/repository"
+	"github.com/gochat/gochat/internal/worker"
 	applogger "github.com/gochat/gochat/pkg/logger"
 )
 
 // CopilotService implements business logic for Copilot (conversation assistance) operations.
 // Reference: Chatwoot enterprise/app/controllers/api/v1/copilot_threads_controller.rb
 type CopilotService struct {
-	threadRepo     *repository.CopilotThreadRepo
-	messageRepo    *repository.CopilotMessageRepo
-	suggestionRepo *repository.CopilotSuggestionRepo
-	assistantRepo  *repository.CaptainAssistantRepo
-	llmProvider    llm.Provider
+	threadRepo      *repository.CopilotThreadRepo
+	messageRepo     *repository.CopilotMessageRepo
+	suggestionRepo  *repository.CopilotSuggestionRepo
+	assistantRepo   *repository.CaptainAssistantRepo
+	llmProvider     llm.Provider
+	responseBackend CopilotResponseBackend
+	worker          *worker.WorkerPool
+}
+
+type CopilotResponseBackend interface {
+	GenerateCopilotResponse(ctx context.Context, req CopilotResponseRequest) ([]CopilotGeneratedMessage, error)
+}
+
+type CopilotResponseRequest struct {
+	AccountID      uint
+	UserID         uint
+	ConversationID uint
+	Thread         *model.CopilotThread
+	Assistant      *model.CaptainAssistant
+	Message        string
+}
+
+type CopilotGeneratedMessage struct {
+	MessageType model.CopilotMessageType
+	Message     map[string]any
 }
 
 // NewCopilotService creates a new CopilotService.
@@ -43,6 +64,15 @@ func NewCopilotService(
 	return s
 }
 
+func (s *CopilotService) SetWorkerPool(wp *worker.WorkerPool) {
+	s.worker = wp
+	RegisterCopilotResponseJobs(wp, s)
+}
+
+func (s *CopilotService) SetResponseBackend(backend CopilotResponseBackend) {
+	s.responseBackend = backend
+}
+
 const CopilotUnavailableMessage = "You are out of Copilot credits. You can buy more credits from the billing section."
 
 // --- Thread CRUD ---
@@ -89,6 +119,12 @@ func (s *CopilotService) CreateThread(ctx context.Context, accountID, userID uin
 	if err != nil {
 		return nil, err
 	}
+	if s.worker != nil {
+		if err := s.enqueueCopilotResponse(ctx, accountID, userID, thread.ID, req.ConversationID, userMsg); err != nil {
+			return nil, err
+		}
+		return s.threadRepo.GetByAccountUserAndID(ctx, accountID, userID, thread.ID)
+	}
 	if _, err := s.createAssistantReply(ctx, accountID, thread.ID, userMsg, req.ConversationID); err != nil {
 		return nil, err
 	}
@@ -152,6 +188,13 @@ func (s *CopilotService) SendMessage(ctx context.Context, accountID, userID, thr
 		return nil, err
 	}
 
+	if s.worker != nil {
+		if err := s.enqueueCopilotResponse(ctx, accountID, userID, threadID, req.ConversationID, userMsg); err != nil {
+			return &SendMessageResult{UserMessage: userMsg}, err
+		}
+		return &SendMessageResult{UserMessage: userMsg}, nil
+	}
+
 	assistantMsg, err := s.createAssistantReply(ctx, accountID, threadID, userMsg, req.ConversationID)
 	if err != nil {
 		return &SendMessageResult{UserMessage: userMsg}, err
@@ -161,6 +204,88 @@ func (s *CopilotService) SendMessage(ctx context.Context, accountID, userID, thr
 	return &SendMessageResult{UserMessage: userMsg, AssistantMessage: assistantMsg}, nil
 }
 
+func (s *CopilotService) enqueueCopilotResponse(ctx context.Context, accountID, userID, threadID, conversationID uint, userMsg *model.CopilotMessage) error {
+	if s.worker == nil || userMsg == nil {
+		return nil
+	}
+	_, err := s.worker.Enqueue(ctx, TaskTypeCaptainCopilotResponse, captainCopilotResponseJob{
+		AccountID:       accountID,
+		UserID:          userID,
+		ConversationID:  conversationID,
+		CopilotThreadID: threadID,
+		MessageID:       userMsg.ID,
+		Message:         userMsg.GetMessageContent(),
+	},
+		worker.WithMaxAttempts(3),
+		worker.WithIdempotencyKey(fmt.Sprintf("captain:copilot_response:%d", userMsg.ID)),
+	)
+	return err
+}
+
+func (s *CopilotService) GenerateCopilotResponseByAccount(ctx context.Context, accountID, userID, threadID, conversationID, messageID uint, message string) ([]model.CopilotMessage, error) {
+	thread, err := s.threadRepo.GetByAccountUserAndID(ctx, accountID, userID, threadID)
+	if err != nil {
+		return nil, fmt.Errorf("thread not found: %w", err)
+	}
+	if messageID != 0 {
+		stored, err := s.messageRepo.GetByID(ctx, messageID)
+		if err != nil {
+			return nil, fmt.Errorf("message not found: %w", err)
+		}
+		if stored.AccountID != accountID || stored.CopilotThreadID != threadID || stored.MessageType != model.CopilotMessageTypeUser {
+			return nil, fmt.Errorf("message does not belong to copilot thread")
+		}
+		message = stored.GetMessageContent()
+	}
+	assistant := &thread.Assistant
+	if thread.AssistantID == nil || thread.Assistant.ID == 0 {
+		return nil, fmt.Errorf("assistant not found")
+	}
+
+	generated, err := s.generateCopilotMessages(ctx, accountID, userID, conversationID, thread, assistant, message)
+	if err != nil {
+		return nil, err
+	}
+	created := make([]model.CopilotMessage, 0, len(generated))
+	for _, item := range generated {
+		messageType := item.MessageType
+		if messageType == "" {
+			messageType = model.CopilotMessageTypeAssistant
+		}
+		raw, err := json.Marshal(item.Message)
+		if err != nil {
+			return created, fmt.Errorf("marshal copilot response message: %w", err)
+		}
+		msg := &model.CopilotMessage{AccountID: accountID, CopilotThreadID: threadID, MessageType: messageType, Message: raw}
+		if err := s.messageRepo.Create(ctx, msg); err != nil {
+			return created, fmt.Errorf("save copilot response message: %w", err)
+		}
+		reloaded, err := s.messageRepo.GetByID(ctx, msg.ID)
+		if err != nil {
+			return created, err
+		}
+		created = append(created, *reloaded)
+	}
+	return created, nil
+}
+
+func (s *CopilotService) generateCopilotMessages(ctx context.Context, accountID, userID, conversationID uint, thread *model.CopilotThread, assistant *model.CaptainAssistant, message string) ([]CopilotGeneratedMessage, error) {
+	if s.responseBackend != nil {
+		return s.responseBackend.GenerateCopilotResponse(ctx, CopilotResponseRequest{AccountID: accountID, UserID: userID, ConversationID: conversationID, Thread: thread, Assistant: assistant, Message: message})
+	}
+	content := CopilotUnavailableMessage
+	if s.llmProvider != nil {
+		generated, err := s.generateAssistantContent(ctx, thread, "")
+		if err != nil {
+			return nil, err
+		}
+		if strings.TrimSpace(generated) != "" {
+			content = generated
+		}
+	}
+	return []CopilotGeneratedMessage{{MessageType: model.CopilotMessageTypeAssistant, Message: map[string]any{"content": content}}}, nil
+}
+
 func (s *CopilotService) CreateThreadMessage(ctx context.Context, accountID, userID, threadID uint, req *SendMessageRequest) (*model.CopilotMessage, error) {
 	content := strings.TrimSpace(req.Message)
 	if content == "" {
@@ -231,10 +356,12 @@ func (s *CopilotService) generateAssistantContent(ctx context.Context, thread *m
 			Content: m.Content,
 		})
 	}
-	llmHistory = append(llmHistory, llm.ChatMessage{
-		Role:    "user",
-		Content: content,
-	})
+	if strings.TrimSpace(content) != "" {
+		llmHistory = append(llmHistory, llm.ChatMessage{
+			Role:    "user",
+			Content: content,
+		})
+	}
 
 	// Build system prompt
 	systemPrompt := "You are an AI assistant helping a customer support agent."

internal/service/message_service.go

@@ -212,6 +212,11 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint
 	// Dispatch additional event based on message type
 	if req.MessageType == "incoming" {
 		s.dispatchMessageEvent(ctx, channel.EventMessageIncoming, message)
+		if s.worker != nil {
+			if _, err := EnqueueCaptainConversationResponseForMessage(ctx, s.worker, s.repo.DB(), message.ID); err != nil {
+				return message, err
+			}
+		}
 	} else if req.MessageType == "outgoing" {
 		if s.worker != nil {
 			if _, err := EnqueueSendReply(ctx, s.worker, message.ID); err != nil {