H-337: fix Web Channel availability and realtime delivery (#60)

* H-337: fix Web Channel availability and realtime delivery

* fix(widget): preserve realtime sender and activity contracts

* fix(widget): keep realtime sender payloads consistent

* fix(widget): make public message persistence atomic

* fix(inbox): keep availability projection out of schema

---------

Co-authored-by: Rogee <rogee@ipao.vip>

backend/internal/app/bootstrap.go

@@ -775,7 +775,7 @@ func Bootstrap(env string) (*App, error) {
 	widgetService := service.NewWidgetService(inboxRepo, contactRepo, contactInboxRepo, conversationRepo, messageRepo, widgetTypingAdapter, widgetThemeConfigRepo, preChatFormRepo, widgetFileUploadRepo, widgetOfflineMessageRepo, inboxMemberRepo, tagRepo, campaignRepo)
 	widgetService.SetWorkerPool(workerPool)
 	widgetService.SetDispatcher(channelDispatcher)
-	widgetHandler := widget.NewHandler(widgetService).WithEventPublisher(eventPublisher)
+	widgetHandler := widget.NewHandler(widgetService)
 
 	// Upload: DirectUpload repo + service + handler (account-level + widget direct uploads)
 	directUploadRepo := repository.NewDirectUploadRepo(db)

backend/internal/handler/api/v1/inbox_member_handler_test.go

@@ -35,7 +35,7 @@ func (s *InboxMemberHandlerTestSuite) SetupSuite() {
 		Logger: logger.Default.LogMode(logger.Silent),
 	})
 	s.Require().NoError(err)
-	s.Require().NoError(db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.User{}, &model.InboxMember{}))
+	s.Require().NoError(db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.User{}, &model.AccountUser{}, &model.InboxMember{}))
 	s.db = db
 
 	repo := repository.NewInboxMemberRepo(db)

backend/internal/handler/widget/coverage3_test.go

@@ -224,21 +224,6 @@ func TestWidgetHandler_Cov3_PublicGetContact_InvalidInbox_Cov3(t *testing.T) {
 	assert.True(t, w.Code == http.StatusNotFound || w.Code == http.StatusBadRequest, "expected 404 or 400, got %d: %s", w.Code, w.Body.String())
 }
 
-// ---------- uintFromPayload unit test ----------
-
-func TestWidgetHandler_Cov3_UintFromPayload_Cov3(t *testing.T) {
-	assert.Equal(t, uint(42), uintFromPayload(uint(42)))
-	assert.Equal(t, uint(42), uintFromPayload(int(42)))
-	assert.Equal(t, uint(0), uintFromPayload(int(-1)))
-	assert.Equal(t, uint(42), uintFromPayload(int64(42)))
-	assert.Equal(t, uint(0), uintFromPayload(int64(-1)))
-	assert.Equal(t, uint(42), uintFromPayload(float64(42)))
-	assert.Equal(t, uint(0), uintFromPayload(float64(-1)))
-	assert.Equal(t, uint(0), uintFromPayload("not a number"))
-	assert.Equal(t, uint(0), uintFromPayload(nil))
-	assert.Equal(t, uint(0), uintFromPayload(uint(0)))
-}
-
 // ---------- GetCableToken ----------
 
 func TestWidgetHandler_Cov3_GetCableToken_NoToken_Cov3(t *testing.T) {

backend/internal/handler/widget/coverage3b_test.go

@@ -311,37 +311,6 @@ func TestWidgetHandler_Cov3_GetConversations_ChatwootRoute_Cov3(t *testing.T) {
 	}
 }
 
-// ---------- SendMessage with event publisher ----------
-
-func TestWidgetHandler_Cov3_SendMessage_WithEventPublisher_Cov3(t *testing.T) {
-	db, router, handler := setupWidgetHandlerTest(t)
-	seedWidgetHandlerData(t, db)
-
-	// Set event publisher
-	publisher := &recordingWidgetEventPublisher{}
-	handler.WithEventPublisher(publisher)
-
-	// Config to get a widget token
-	wConfig := httptest.NewRecorder()
-	reqConfig, _ := http.NewRequest("POST", "/api/v1/widget/config?website_token=handler_ws_token_123", nil)
-	router.ServeHTTP(wConfig, reqConfig)
-	require.Equal(t, http.StatusOK, wConfig.Code)
-	var configResp map[string]interface{}
-	require.NoError(t, json.Unmarshal(wConfig.Body.Bytes(), &configResp))
-	authToken := configResp["contact"].(map[string]interface{})["pubsub_token"].(string)
-
-	// Send message
-	messageBody, _ := json.Marshal(map[string]interface{}{"message": map[string]interface{}{"content": "hello with publisher"}})
-	wMsg := httptest.NewRecorder()
-	reqMsg, _ := http.NewRequest("POST", "/api/v1/widget/messages?cw_conversation="+authToken, bytes.NewReader(messageBody))
-	reqMsg.Header.Set("Content-Type", "application/json")
-	router.ServeHTTP(wMsg, reqMsg)
-	require.Equal(t, http.StatusOK, wMsg.Code, wMsg.Body.String())
-
-	// The event publisher should have been called
-	assert.NotZero(t, publisher.accountID)
-}
-
 // ---------- SendTranscript with valid token (needs conversation) ----------
 
 func TestWidgetHandler_Cov3_SendTranscript_WithConversation_Cov3(t *testing.T) {

backend/internal/handler/widget/coverage4_test.go

@@ -86,17 +86,6 @@ func TestConfig_InvalidWebsiteToken_Cov4(t *testing.T) {
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
 
-func TestConfig_WithEventPublisher_Cov4(t *testing.T) {
-	db, router, handler := setupWidgetHandlerTest(t)
-	seedWidgetHandlerData(t, db)
-	pub := &recordingWidgetEventPublisher{}
-	handler.WithEventPublisher(pub)
-	w := httptest.NewRecorder()
-	req, _ := http.NewRequest("POST", "/api/v1/widget/config?website_token=handler_ws_token_123", nil)
-	router.ServeHTTP(w, req)
-	assert.Equal(t, http.StatusOK, w.Code)
-}
-
 // ============================================================
 // SendMessage — edge cases
 // ============================================================

backend/internal/handler/widget/widget_handler.go

@@ -12,20 +12,14 @@ import (
 	"github.com/gochat/gochat/internal/model"
 	"github.com/gochat/gochat/internal/service"
 	"github.com/gochat/gochat/internal/webhookutil"
-	wspkg "github.com/gochat/gochat/internal/ws"
 )
 
-type widgetEventPublisher interface {
-	PublishWidgetEvent(accountID uint, pubsubToken string, eventType string, payload interface{})
-}
-
 // WidgetHandler handles the public-facing widget API endpoints.
 // These endpoints are accessed by the embedded JS widget on customer websites
 // and do not require agent JWT authentication — they use a widget_token instead.
 // Reference: Chatwoot app/controllers/api/v1/widget_messages_controller.rb
 type WidgetHandler struct {
-	widgetService  *service.WidgetService
-	eventPublisher widgetEventPublisher
+	widgetService *service.WidgetService
 }
 
 // NewHandler creates a new WidgetHandler with the WidgetService dependency.
@@ -35,11 +29,6 @@ func NewHandler(widgetService *service.WidgetService) *WidgetHandler {
 	}
 }
 
-func (h *WidgetHandler) WithEventPublisher(publisher widgetEventPublisher) *WidgetHandler {
-	h.eventPublisher = publisher
-	return h
-}
-
 // Init handles widget initialization — authenticates/creates a contact
 // and returns a widget_token (pubsub_token) for subsequent requests.
 // POST /widget/init
@@ -212,7 +201,6 @@ func (h *WidgetHandler) SendMessage(c *gin.Context) {
 		if len(resp.Attachments) > 0 {
 			payload["attachments"] = widgetAttachmentPayloads(resp.Attachments)
 		}
-		h.publishWidgetMessageEvent(req.WidgetToken, payload)
 		c.JSON(http.StatusOK, resp)
 		return
 	}
@@ -220,7 +208,6 @@ func (h *WidgetHandler) SendMessage(c *gin.Context) {
 	if len(resp.Attachments) > 0 {
 		payload["attachments"] = widgetAttachmentPayloads(resp.Attachments)
 	}
-	h.publishWidgetMessageEvent(req.WidgetToken, payload)
 	c.JSON(http.StatusOK, payload)
 }
 
@@ -262,40 +249,6 @@ func (h *WidgetHandler) UpdateMessage(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{"contact": widgetContactFullPayload(contact)})
 }
 
-func (h *WidgetHandler) publishWidgetMessageEvent(pubsubToken string, payload gin.H) {
-	if h.eventPublisher == nil || payload == nil {
-		return
-	}
-	accountID := uintFromPayload(payload["account_id"])
-	if accountID == 0 {
-		accountID = uintFromPayload(payload["accountId"])
-	}
-	if accountID == 0 {
-		return
-	}
-	h.eventPublisher.PublishWidgetEvent(accountID, pubsubToken, wspkg.EventMessageCreated, payload)
-}
-
-func uintFromPayload(value any) uint {
-	switch typed := value.(type) {
-	case uint:
-		return typed
-	case int:
-		if typed > 0 {
-			return uint(typed)
-		}
-	case int64:
-		if typed > 0 {
-			return uint(typed)
-		}
-	case float64:
-		if typed > 0 {
-			return uint(typed)
-		}
-	}
-	return 0
-}
-
 // GetLatestMessages implements Chatwoot's GET /api/v1/widget/messages endpoint.
 func (h *WidgetHandler) GetLatestMessages(c *gin.Context) {
 	widgetToken := widgetTokenFromRequest(c)

backend/internal/handler/widget/widget_handler_test.go

@@ -43,13 +43,6 @@ func (n *noopTypingIndicatorWidget) SetTypingOff(_ context.Context, _ uint, _ ui
 	return nil
 }
 
-type recordingWidgetEventPublisher struct {
-	accountID   uint
-	pubsubToken string
-	eventType   string
-	payload     interface{}
-}
-
 func TestPublicMessagePayloadSanitizesIdentity(t *testing.T) {
 	senderID := uint(9)
 	message := model.Message{
@@ -73,13 +66,6 @@ func TestPublicMessagePayloadSanitizesIdentity(t *testing.T) {
 	assert.Empty(t, publicMessagePayload(message, model.Conversation{})["content_attributes"])
 }
 
-func (p *recordingWidgetEventPublisher) PublishWidgetEvent(accountID uint, pubsubToken string, eventType string, payload interface{}) {
-	p.accountID = accountID
-	p.pubsubToken = pubsubToken
-	p.eventType = eventType
-	p.payload = payload
-}
-
 type recordingWidgetTranscriptDeliverer struct {
 	requests []automation.AutomationTranscriptRequest
 	err      error
@@ -779,42 +765,6 @@ func TestWidgetHandler_ChatwootMessages_AuthTokenAndNestedPayload(t *testing.T)
 	assertWidgetContactFixtureShape(t, contactResp)
 }
 
-func TestWidgetHandler_ChatwootMessagePublishesWidgetRealtimePayload(t *testing.T) {
-	db, router, handler := setupWidgetHandlerTest(t)
-	_, _ = seedWidgetHandlerData(t, db)
-	publisher := &recordingWidgetEventPublisher{}
-	handler.WithEventPublisher(publisher)
-
-	wConfig := httptest.NewRecorder()
-	reqConfig, _ := http.NewRequest("POST", "/api/v1/widget/config?website_token=handler_ws_token_123", nil)
-	router.ServeHTTP(wConfig, reqConfig)
-	require.Equal(t, http.StatusOK, wConfig.Code)
-
-	var configResp map[string]interface{}
-	require.NoError(t, json.Unmarshal(wConfig.Body.Bytes(), &configResp))
-	authToken := configResp["contact"].(map[string]interface{})["pubsub_token"].(string)
-
-	body, _ := json.Marshal(map[string]interface{}{
-		"message": map[string]interface{}{"content": "Realtime widget message"},
-	})
-	wMessage := httptest.NewRecorder()
-	reqMessage, _ := http.NewRequest("POST", "/api/v1/widget/messages", bytes.NewReader(body))
-	reqMessage.Header.Set("Content-Type", "application/json")
-	reqMessage.Header.Set("X-Auth-Token", authToken)
-	router.ServeHTTP(wMessage, reqMessage)
-	require.Equal(t, http.StatusOK, wMessage.Code)
-
-	assert.Equal(t, uint(1), publisher.accountID)
-	assert.Equal(t, authToken, publisher.pubsubToken)
-	assert.Equal(t, ws.EventMessageCreated, publisher.eventType)
-	payload, ok := publisher.payload.(gin.H)
-	require.True(t, ok, "expected widget realtime payload to use Chatwoot widget message shape")
-	assert.Equal(t, "Realtime widget message", payload["content"])
-	assert.Equal(t, uint(1), payload["account_id"])
-	assert.NotEmpty(t, payload["conversation_id"])
-	assertWidgetMessageFixtureShape(t, payload)
-}
-
 func TestWidgetHandler_ChatwootMessagesIndexFiltersInternalMessages(t *testing.T) {
 	db, router, _ := setupWidgetHandlerTest(t)
 	_, inbox := seedWidgetHandlerData(t, db)
@@ -1656,21 +1606,26 @@ func TestWidgetHandler_ChatwootInboxMembers_Success(t *testing.T) {
 	db, router, _ := setupWidgetHandlerTest(t)
 	_, inbox := seedWidgetHandlerData(t, db)
 
-	user := &model.User{
+	userA := &model.User{
 		AccountID: inbox.AccountID,
-		Name:      "Ada Agent",
+		Name:      "Agent A",
 		Email:     "ada@handler.test",
 		Password:  "secret",
 		AvatarURL: "https://example.test/ada.png",
 		Active:    true,
 	}
-	require.NoError(t, db.Create(user).Error)
+	userB := &model.User{AccountID: inbox.AccountID, Name: "Agent B", Email: "b@handler.test", Password: "secret", Active: true}
+	require.NoError(t, db.Create(userA).Error)
+	require.NoError(t, db.Create(userB).Error)
+	require.NoError(t, db.Create(&model.AccountUser{AccountID: inbox.AccountID, UserID: userA.ID, Role: "agent", Availability: "online"}).Error)
+	require.NoError(t, db.Create(&model.AccountUser{AccountID: inbox.AccountID, UserID: userB.ID, Role: "agent", Availability: "online"}).Error)
 	require.NoError(t, db.Create(&model.InboxMember{
 		InboxID:            inbox.ID,
-		UserID:             user.ID,
+		UserID:             userA.ID,
 		Role:               "agent",
-		AvailabilityStatus: "online",
+		AvailabilityStatus: "offline",
 	}).Error)
+	require.NoError(t, db.Create(&model.InboxMember{InboxID: inbox.ID, UserID: userB.ID, Role: "agent", AvailabilityStatus: "offline"}).Error)
 
 	w := httptest.NewRecorder()
 	req, _ := http.NewRequest("GET", "/api/v1/widget/inbox_members?website_token=handler_ws_token_123", nil)
@@ -1680,10 +1635,11 @@ func TestWidgetHandler_ChatwootInboxMembers_Success(t *testing.T) {
 	var resp map[string]interface{}
 	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp))
 	payload := resp["payload"].([]interface{})
-	require.Len(t, payload, 1)
-	member := payload[0].(map[string]interface{})
-	assert.Equal(t, "Ada Agent", member["name"])
-	assert.Equal(t, "online", member["availability_status"])
+	require.Len(t, payload, 2)
+	assert.Equal(t, "Agent A", payload[0].(map[string]interface{})["name"])
+	assert.Equal(t, "online", payload[0].(map[string]interface{})["availability_status"])
+	assert.Equal(t, "Agent B", payload[1].(map[string]interface{})["name"])
+	assert.Equal(t, "online", payload[1].(map[string]interface{})["availability_status"])
 }
 
 func TestWidgetHandler_ChatwootInboxMembers_InvalidWebsiteTokenReturnsNotFound(t *testing.T) {

backend/internal/model/inbox_member.go

@@ -4,13 +4,14 @@ package model
 // Reference: Chatwoot InboxMember model + P2B M2 spec
 type InboxMember struct {
 	Base
-	InboxID           uint   `gorm:"column:inbox_id;not null;index" json:"inbox_id"`
-	UserID            uint   `gorm:"column:user_id;not null;index" json:"user_id"`
-	Role              string `gorm:"column:role;size:50;default:agent" json:"role"` // agent, supervisor
-	AvailabilityStatus string `gorm:"column:availability_status;size:50;default:offline" json:"availability_status"` // online, offline, busy
+	InboxID             uint   `gorm:"column:inbox_id;not null;index" json:"inbox_id"`
+	UserID              uint   `gorm:"column:user_id;not null;index" json:"user_id"`
+	Role                string `gorm:"column:role;size:50;default:agent" json:"role"`                                 // agent, supervisor
+	AvailabilityStatus  string `gorm:"column:availability_status;size:50;default:offline" json:"availability_status"` // online, offline, busy
+	AccountAvailability string `gorm:"-" json:"-"`
 
 	Inbox Inbox `gorm:"foreignKey:InboxID" json:"inbox,omitempty"`
 	User  User  `gorm:"foreignKey:UserID" json:"user,omitempty"`
 }
 
-func (InboxMember) TableName() string { return "inbox_members" }
+func (InboxMember) TableName() string { return "inbox_members" }

backend/internal/repository/inbox_member_repo.go

@@ -41,10 +41,29 @@ func (r *InboxMemberRepo) FindByInboxAndUser(ctx context.Context, inboxID, userI
 
 // FindByInbox retrieves all members (agents) assigned to an inbox.
 func (r *InboxMemberRepo) FindByInbox(ctx context.Context, inboxID uint) ([]model.InboxMember, error) {
-	var ims []model.InboxMember
-	err := r.withAgentPreloads(r.db.WithContext(ctx)).Where("inbox_id = ?", inboxID).
-		Order("id ASC").Find(&ims).Error
-	return ims, err
+	type inboxMemberWithAvailability struct {
+		model.InboxMember   `gorm:"embedded"`
+		AccountAvailability string `gorm:"column:account_availability"`
+	}
+
+	var rows []inboxMemberWithAvailability
+	err := r.withAgentPreloads(r.db.WithContext(ctx)).
+		Model(&model.InboxMember{}).
+		Select("inbox_members.*, COALESCE(account_users.availability, 'offline') AS account_availability").
+		Joins("LEFT JOIN inboxes ON inboxes.id = inbox_members.inbox_id").
+		Joins("LEFT JOIN account_users ON account_users.account_id = inboxes.account_id AND account_users.user_id = inbox_members.user_id AND account_users.deleted_at IS NULL").
+		Where("inbox_members.inbox_id = ?", inboxID).
+		Order("inbox_members.id ASC").Find(&rows).Error
+	if err != nil {
+		return nil, err
+	}
+
+	ims := make([]model.InboxMember, len(rows))
+	for i := range rows {
+		ims[i] = rows[i].InboxMember
+		ims[i].AccountAvailability = rows[i].AccountAvailability
+	}
+	return ims, nil
 }
 
 // FindByUser retrieves all inboxes a user is assigned to.

backend/internal/repository/inbox_member_repo_test.go

@@ -103,11 +103,12 @@ func TestInboxMemberRepo_FindByInbox(t *testing.T) {
 	db := setupTestDB(t, &model.InboxMember{})
 	repo := NewInboxMemberRepo(db)
 
-	_, user1, inbox := createInboxMemberPrereqs(t, db)
+	account, user1, inbox := createInboxMemberPrereqs(t, db)
 
 	// Create a second user
 	user2 := &model.User{Name: "IMTestUser2", Email: "im2@example.com", Provider: "email"}
 	require.NoError(t, db.Create(user2).Error)
+	require.NoError(t, db.Create(&model.AccountUser{AccountID: account.ID, UserID: user1.ID, Role: "agent", Availability: "online"}).Error)
 
 	im1 := &model.InboxMember{InboxID: inbox.ID, UserID: user1.ID, Role: "agent"}
 	require.NoError(t, db.Create(im1).Error)
@@ -121,6 +122,8 @@ func TestInboxMemberRepo_FindByInbox(t *testing.T) {
 	// Verify User is preloaded
 	assert.Equal(t, "IMTestUser", members[0].User.Name)
 	assert.Equal(t, "IMTestUser2", members[1].User.Name)
+	assert.Equal(t, "online", members[0].AccountAvailability)
+	assert.Equal(t, "offline", members[1].AccountAvailability)
 
 	// Verify ordering by ID ASC
 	assert.Equal(t, im1.ID, members[0].ID)
@@ -240,4 +243,4 @@ func TestInboxMemberRepo_DeleteByInbox(t *testing.T) {
 	members, err := repo.FindByInbox(context.Background(), inbox.ID)
 	assert.NoError(t, err)
 	assert.Len(t, members, 0)
-}
+}

backend/internal/service/captain_conversation_service.go

@@ -393,63 +393,74 @@ func RegisterCaptainConversationJobs(wp *worker.WorkerPool, svc *CaptainConversa
 }
 
 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
+	job, created, err := enqueueCaptainConversationResponseForMessageInTransaction(ctx, wp, db, messageID)
+	if err != nil {
+		return nil, err
+	}
+	if created {
+		wp.Publish(ctx, job)
+	}
+	return job, nil
+}
+
+func enqueueCaptainConversationResponseForMessageInTransaction(ctx context.Context, wp *worker.WorkerPool, tx *gorm.DB, messageID uint) (*model.BackgroundJob, bool, error) {
+	if wp == nil || tx == nil || messageID == 0 || !tx.Migrator().HasTable(&model.CaptainInbox{}) {
+		return nil, false, nil
 	}
 	var message model.Message
-	if err := db.WithContext(ctx).First(&message, messageID).Error; err != nil {
-		return nil, err
+	if err := tx.WithContext(ctx).First(&message, messageID).Error; err != nil {
+		return nil, false, err
 	}
 	if message.MessageType != string(model.MessageTypeIncoming) || message.Private {
-		return nil, nil
+		return nil, false, 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 err := tx.WithContext(ctx).Where("account_id = ? AND id = ?", message.AccountID, message.ConversationID).First(&conversation).Error; err != nil {
+		return nil, false, err
 	}
 	if conversation.Status != string(model.ConversationStatusPending) {
-		return nil, nil
+		return nil, false, nil
 	}
 	var assistantID uint
 	var takeoverVersion uint
 	if conversation.AssigneeAgentBotID != 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 := tx.WithContext(ctx).Where("id = ? AND bot_type = ?", *conversation.AssigneeAgentBotID, "captain").First(&bot).Error; err != nil {
 			if err == gorm.ErrRecordNotFound {
-				return nil, nil
+				return nil, false, nil
 			}
-			return nil, err
+			return nil, false, 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 := tx.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, false, nil
 			}
-			return nil, err
+			return nil, false, 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 := tx.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, false, nil
 			}
-			return nil, err
+			return nil, false, err
 		}
 		assistantID = ci.AssistantID
 	}
 	if assistantID == 0 {
-		return nil, nil
+		return nil, false, nil
 	}
 	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() {
+	if scheduledAt := captainConversationResponseScheduledAt(ctx, tx, message.ID); !scheduledAt.IsZero() {
 		options = append(options, worker.WithScheduledAt(scheduledAt))
 	}
-	return wp.Enqueue(ctx, TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: conversation.AccountID, ConversationID: conversation.ID, AssistantID: assistantID, MessageID: message.ID, TakeoverVersion: takeoverVersion}, options...)
+	return wp.EnqueueInTransaction(ctx, tx, 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 {

backend/internal/service/message_service.go

@@ -76,12 +76,62 @@ func (s *MessageService) dispatchMessageEvent(ctx context.Context, eventType cha
 		event.UserID = *message.SenderID
 	}
 	event.Data["message"] = message
+	if eventType == channel.EventMessageCreated || eventType == channel.EventMessageUpdated || eventType == channel.EventMessageDeleted {
+		s.addWebWidgetEventContext(ctx, event, message)
+	}
 	applogger.L().Infof("dispatching event %s for message %d", eventType, message.ID)
 	if err := s.dispatcher.Dispatch(ctx, event); err != nil {
 		applogger.L().Errorf("failed to dispatch event %s for message %d: %v", eventType, message.ID, err)
 	}
 }
 
+func (s *MessageService) addWebWidgetEventContext(ctx context.Context, event *channel.ChannelEvent, message *model.Message) {
+	var inbox model.Inbox
+	if err := s.repo.DB().WithContext(ctx).First(&inbox, message.InboxID).Error; err != nil || inbox.ChannelType != string(channel.ChannelWebWidget) {
+		return
+	}
+	event.Channel = channel.ChannelWebWidget
+	event.Data["inbox"] = &inbox
+
+	var conversation model.Conversation
+	if err := s.repo.DB().WithContext(ctx).First(&conversation, message.ConversationID).Error; err != nil {
+		return
+	}
+	event.ContactID = conversation.ContactID
+	event.Data["conversation"] = &conversation
+
+	var contact model.Contact
+	if err := s.repo.DB().WithContext(ctx).First(&contact, conversation.ContactID).Error; err == nil {
+		event.Data["contact"] = &contact
+	}
+	if message.SenderID != nil {
+		switch strings.ToLower(strings.TrimSpace(message.SenderType)) {
+		case "user":
+			var sender model.User
+			if err := s.repo.DB().WithContext(ctx).First(&sender, *message.SenderID).Error; err == nil {
+				event.Data["sender"] = &sender
+			}
+		case "agentbot", "agent_bot":
+			var sender model.AgentBot
+			if err := s.repo.DB().WithContext(ctx).First(&sender, *message.SenderID).Error; err == nil {
+				event.Data["sender"] = &sender
+			}
+		case "captain::assistant", "captainassistant", "captain_assistant":
+			var sender model.CaptainAssistant
+			if err := s.repo.DB().WithContext(ctx).First(&sender, *message.SenderID).Error; err == nil {
+				event.Data["sender"] = &sender
+			}
+		}
+	}
+	if message.Private || message.MessageType == string(model.MessageTypeActivity) || conversation.ContactInboxID == nil {
+		return
+	}
+	var contactInbox model.ContactInbox
+	if err := s.repo.DB().WithContext(ctx).Select("pubsub_token").First(&contactInbox, *conversation.ContactInboxID).Error; err == nil {
+		event.Data["widget_token"] = contactInbox.PubsubToken
+	}
+}
+
 // ListByConversation retrieves all messages for a conversation.
 func (s *MessageService) ListByConversation(ctx context.Context, conversationID uint, offset, limit int) ([]model.Message, int64, error) {
 	return s.repo.FindByConversation(ctx, conversationID, offset, limit)

backend/internal/service/message_service_test.go

@@ -55,6 +55,17 @@ type mockRetryListener struct {
 	lastData      map[string]interface{}
 }
 
+type messageEventListener struct {
+	events []*channel.ChannelEvent
+}
+
+func (l *messageEventListener) Name() string { return "message-event-listener" }
+
+func (l *messageEventListener) OnEvent(_ context.Context, event *channel.ChannelEvent) error {
+	l.events = append(l.events, event)
+	return nil
+}
+
 func (l *mockRetryListener) Name() string {
 	return "mock-retry-listener"
 }
@@ -93,6 +104,86 @@ func setupMessageServiceWithDefaultLLM(t *testing.T) (*gorm.DB, *repository.Mess
 	return setupMessageService(t, mockLLM)
 }
 
+func TestMessageService_WebWidgetReplyCarriesRealtimeContext(t *testing.T) {
+	db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
+	account := createTestAccount(t, db)
+	agent := createTestUser(t, db, account.ID)
+	inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget))
+	contact := createTestContact(t, db, account.ID)
+	contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", PubsubToken: "visitor-token"}
+	require.NoError(t, db.Create(contactInbox).Error)
+	conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
+	require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error)
+
+	listener := &messageEventListener{}
+	dispatcher.Register(listener)
+	_, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
+		ConversationID: conversation.ID,
+		Content:        "agent reply",
+		ContentType:    "text",
+		MessageType:    "outgoing",
+	})
+	require.NoError(t, err)
+
+	var created *channel.ChannelEvent
+	for _, event := range listener.events {
+		if event.Type == channel.EventMessageCreated {
+			created = event
+			break
+		}
+	}
+	require.NotNil(t, created)
+	assert.Equal(t, channel.ChannelWebWidget, created.Channel)
+	assert.Equal(t, "visitor-token", created.Data["widget_token"])
+	assert.IsType(t, &model.Conversation{}, created.Data["conversation"])
+	assert.IsType(t, &model.Contact{}, created.Data["contact"])
+	assert.IsType(t, &model.User{}, created.Data["sender"])
+}
+
+func TestMessageService_WebWidgetReplyResolvesAutomatedSenders(t *testing.T) {
+	db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
+	require.NoError(t, db.AutoMigrate(&model.AgentBot{}, &model.CaptainAssistant{}))
+	account := createTestAccount(t, db)
+	inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget))
+	contact := createTestContact(t, db, account.ID)
+	contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", PubsubToken: "visitor-token"}
+	require.NoError(t, db.Create(contactInbox).Error)
+	conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
+	require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error)
+
+	bot := &model.AgentBot{AccountID: &account.ID, Name: "Reply Bot", BotType: "webhook"}
+	require.NoError(t, db.Create(bot).Error)
+	assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Captain", Config: json.RawMessage(`{}`), Status: model.AssistantStatusActive}
+	require.NoError(t, db.Create(assistant).Error)
+
+	tests := []struct {
+		name       string
+		senderType string
+		senderID   uint
+		expected   interface{}
+	}{
+		{name: "agent bot", senderType: "AgentBot", senderID: bot.ID, expected: &model.AgentBot{}},
+		{name: "captain namespaced", senderType: "Captain::Assistant", senderID: assistant.ID, expected: &model.CaptainAssistant{}},
+		{name: "captain alias", senderType: "CaptainAssistant", senderID: assistant.ID, expected: &model.CaptainAssistant{}},
+		{name: "unknown", senderType: "CustomSender", senderID: assistant.ID},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, account.ID, inbox.ID)
+			message := &model.Message{ConversationID: conversation.ID, InboxID: inbox.ID, SenderID: &tt.senderID, SenderType: tt.senderType}
+			svc.addWebWidgetEventContext(context.Background(), event, message)
+
+			assert.Equal(t, "visitor-token", event.Data["widget_token"])
+			assert.IsType(t, &model.Contact{}, event.Data["contact"])
+			if tt.expected == nil {
+				assert.NotContains(t, event.Data, "sender")
+			} else {
+				assert.IsType(t, tt.expected, event.Data["sender"])
+			}
+		})
+	}
+}
+
 // ========== ListByConversation 测试 ==========
 
 func TestMessageService_ListByConversation(t *testing.T) {

backend/internal/service/widget_service.go

@@ -366,6 +366,7 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe
 
 	// Resolve conversation
 	var conversation *model.Conversation
+	conversationCreated := false
 	if req.ConversationID != nil {
 		conversation, err = s.conversationRepo.FindByID(ctx, *req.ConversationID)
 		if err != nil {
@@ -387,6 +388,7 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe
 			if err != nil {
 				return nil, fmt.Errorf("failed to create conversation: %w", err)
 			}
+			conversationCreated = true
 		}
 	}
 
@@ -408,24 +410,38 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe
 		}
 	}
 
-	if err := s.messageRepo.Create(ctx, &msg); err != nil {
-		return nil, fmt.Errorf("failed to create message: %w", err)
-	}
-	if err := s.reopenWidgetConversationForIncomingMessage(ctx, conversation); err != nil {
-		return nil, err
-	}
-
-	attachments, err := s.attachWidgetUploads(ctx, &msg, req.AttachmentIDs)
+	attachments, err := s.createIncomingMessage(ctx, conversation, &msg, req.AttachmentIDs)
 	if err != nil {
 		return nil, err
 	}
-	if _, err := EnqueueCaptainConversationResponseForMessage(ctx, s.worker, s.messageRepo.DB(), msg.ID); err != nil {
-		return nil, err
-	}
 
-	// Dispatch message.created event to trigger auto-reply listener
+	// Dispatch the same complete lifecycle context Chatwoot gives ActionCable.
 	if s.dispatcher != nil {
 		inbox, _ := s.inboxRepo.FindByID(ctx, conversation.InboxID)
+		data := map[string]interface{}{
+			"inbox":        inbox,
+			"conversation": conversation,
+			"contact":      &contactInbox.Contact,
+			"widget_token": contactInbox.PubsubToken,
+			"channel_type": "web_widget",
+		}
+		if conversationCreated {
+			event := &channel.ChannelEvent{
+				Type: channel.EventConversationCreated, Channel: channel.ChannelWebWidget,
+				ConversationID: conversation.ID, InboxID: conversation.InboxID,
+				AccountID: conversation.AccountID, ContactID: conversation.ContactID,
+				Timestamp: time.Now().Unix(), Data: data,
+			}
+			if dispatchErr := s.dispatcher.Dispatch(ctx, event); dispatchErr != nil {
+				applogger.L().Warnf("widget conversation event dispatch failed: inbox=%d conv=%d err=%v",
+					conversation.InboxID, conversation.ID, dispatchErr)
+			}
+		}
+		messageData := make(map[string]interface{}, len(data)+1)
+		for key, value := range data {
+			messageData[key] = value
+		}
+		messageData["message"] = &msg
 		event := &channel.ChannelEvent{
 			Type:           channel.EventMessageCreated,
 			Channel:        channel.ChannelWebWidget,
@@ -433,13 +449,8 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe
 			InboxID:        conversation.InboxID,
 			AccountID:      conversation.AccountID,
 			Timestamp:      time.Now().Unix(),
-			Data: map[string]interface{}{
-				"inbox":        inbox,
-				"content":      msg.Content,
-				"sender_type":  "Contact",
-				"channel_type": "web_widget",
-				"source_id":    msg.SourceID,
-			},
+			ContactID:      conversation.ContactID,
+			Data:           messageData,
 		}
 		if dispatchErr := s.dispatcher.Dispatch(ctx, event); dispatchErr != nil {
 			applogger.L().Warnf("widget message event dispatch failed: inbox=%d conv=%d err=%v",
@@ -457,18 +468,51 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe
 	}, nil
 }
 
-func (s *WidgetService) reopenWidgetConversationForIncomingMessage(ctx context.Context, conversation *model.Conversation) error {
-	if conversation == nil || conversation.Muted {
-		return nil
+func (s *WidgetService) createIncomingMessage(ctx context.Context, conversation *model.Conversation, message *model.Message, attachmentIDs []string) ([]model.Attachment, error) {
+	var messageTimestamp int64
+	var attachments []model.Attachment
+	var captainJob *model.BackgroundJob
+	var captainJobCreated bool
+	reopen := conversation != nil && !conversation.Muted &&
+		(conversation.Status == string(model.ConversationStatusSnoozed) || conversation.Status == string(model.ConversationStatusResolved))
+	if err := s.messageRepo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := tx.Create(message).Error; err != nil {
+			return err
+		}
+		var err error
+		attachments, err = s.attachWidgetUploadsInTransaction(ctx, tx, message, attachmentIDs)
+		if err != nil {
+			return err
+		}
+		messageTimestamp = message.CreatedAt.Unix()
+		updates := map[string]any{
+			"last_activity_at": messageTimestamp,
+			"last_message_at":  messageTimestamp,
+		}
+		if reopen {
+			updates["status"] = string(model.ConversationStatusOpen)
+			updates["snoozed_until"] = nil
+		}
+		if err := tx.Model(&model.Conversation{}).
+			Where("id = ? AND account_id = ?", conversation.ID, conversation.AccountID).
+			Updates(updates).Error; err != nil {
+			return err
+		}
+		captainJob, captainJobCreated, err = enqueueCaptainConversationResponseForMessageInTransaction(ctx, s.worker, tx, message.ID)
+		return err
+	}); err != nil {
+		return nil, fmt.Errorf("failed to create message: %w", err)
 	}
-	if conversation.Status != string(model.ConversationStatusSnoozed) && conversation.Status != string(model.ConversationStatusResolved) {
-		return nil
+	conversation.LastActivityAt = &messageTimestamp
+	conversation.LastMessageAt = &messageTimestamp
+	if reopen {
+		conversation.Status = string(model.ConversationStatusOpen)
+		conversation.SnoozedUntil = nil
 	}
-	conversation.Status = string(model.ConversationStatusOpen)
-	conversation.SnoozedUntil = nil
-	return s.conversationRepo.DB().WithContext(ctx).Model(&model.Conversation{}).
-		Where("id = ?", conversation.ID).
-		Updates(map[string]any{"status": conversation.Status, "snoozed_until": nil}).Error
+	if captainJobCreated {
+		s.worker.Publish(ctx, captainJob)
+	}
+	return attachments, nil
 }
 
 // GetConversations returns conversations for a widget contact.
@@ -551,10 +595,13 @@ func (s *WidgetService) GetInboxMembersByWebsiteToken(ctx context.Context, websi
 		if name == "" {
 			name = member.User.Email
 		}
-		availability := member.AvailabilityStatus
+		availability := member.AccountAvailability
 		if availability == "" {
 			availability = "offline"
 		}
+		if !member.User.Active {
+			availability = "offline"
+		}
 		payload = append(payload, WidgetInboxMember{
 			ID:                 member.User.ID,
 			Name:               name,
@@ -993,6 +1040,10 @@ func (s *WidgetService) PublicCreateMessage(ctx context.Context, inboxIdentifier
 	if len([]rune(req.Content)) > widgetMessageContentLimit {
 		return nil, nil, nil, ErrWidgetMessageContentTooLong
 	}
+	_, contactInbox, err := s.resolvePublicContactInbox(ctx, inboxIdentifier, sourceID)
+	if err != nil {
+		return nil, nil, nil, err
+	}
 	conversation, err := s.PublicGetConversation(ctx, inboxIdentifier, sourceID, displayID)
 	if err != nil {
 		return nil, nil, nil, err
@@ -1009,33 +1060,29 @@ func (s *WidgetService) PublicCreateMessage(ctx context.Context, inboxIdentifier
 		Status:         "sent",
 		SourceID:       req.EchoID,
 	}
-	if err := s.messageRepo.Create(ctx, message); err != nil {
-		return nil, nil, nil, err
-	}
-	attachments, err := s.attachWidgetUploads(ctx, message, req.AttachmentIDs)
+	attachments, err := s.createIncomingMessage(ctx, conversation, message, req.AttachmentIDs)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	if _, err := EnqueueCaptainConversationResponseForMessage(ctx, s.worker, s.messageRepo.DB(), message.ID); err != nil {
-		return nil, nil, nil, err
-	}
 
-	// Dispatch message.created event to trigger auto-reply listener
+	// Public API messages use the same committed lifecycle payload as widget messages.
 	if s.dispatcher != nil {
 		inbox, _ := s.inboxRepo.FindByID(ctx, conversation.InboxID)
 		event := &channel.ChannelEvent{
 			Type:           channel.EventMessageCreated,
-			Channel:        channel.ChannelWebWidget,
+			Channel:        channel.ChannelAPI,
 			ConversationID: conversation.ID,
 			InboxID:        conversation.InboxID,
 			AccountID:      conversation.AccountID,
+			ContactID:      conversation.ContactID,
 			Timestamp:      time.Now().Unix(),
 			Data: map[string]interface{}{
 				"inbox":        inbox,
-				"content":      message.Content,
-				"sender_type":  "Contact",
-				"channel_type": "web_widget",
-				"source_id":    message.SourceID,
+				"conversation": conversation,
+				"contact":      &contactInbox.Contact,
+				"widget_token": contactInbox.PubsubToken,
+				"channel_type": "api",
+				"message":      message,
 			},
 		}
 		if dispatchErr := s.dispatcher.Dispatch(ctx, event); dispatchErr != nil {
@@ -2039,6 +2086,16 @@ func (s *WidgetService) findPublicContact(ctx context.Context, accountID uint, r
 }
 
 func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model.Message, signedIDs []string) ([]model.Attachment, error) {
+	var attachments []model.Attachment
+	err := s.messageRepo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var err error
+		attachments, err = s.attachWidgetUploadsInTransaction(ctx, tx, message, signedIDs)
+		return err
+	})
+	return attachments, err
+}
+
+func (s *WidgetService) attachWidgetUploadsInTransaction(ctx context.Context, tx *gorm.DB, message *model.Message, signedIDs []string) ([]model.Attachment, error) {
 	if len(signedIDs) == 0 {
 		return nil, nil
 	}
@@ -2049,7 +2106,7 @@ func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model.
 			continue
 		}
 		var upload model.DirectUpload
-		if err := s.messageRepo.DB().WithContext(ctx).Where("upload_uuid = ?", signedID).First(&upload).Error; err != nil {
+		if err := tx.WithContext(ctx).Where("upload_uuid = ?", signedID).First(&upload).Error; err != nil {
 			return nil, fmt.Errorf("direct upload not found: %w", err)
 		}
 		if upload.Source != model.DirectUploadSourceWidget {
@@ -2063,7 +2120,7 @@ func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model.
 		}
 		if time.Now().After(upload.ExpiresAt) {
 			upload.Status = model.DirectUploadStatusExpired
-			_ = s.messageRepo.DB().WithContext(ctx).Save(&upload).Error
+			_ = tx.WithContext(ctx).Save(&upload).Error
 			return nil, errors.New("direct upload has expired")
 		}
 		attachment := model.Attachment{
@@ -2076,12 +2133,12 @@ func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model.
 			FileName:  upload.OriginalName,
 			Metadata:  string(upload.Metadata),
 		}
-		if err := s.messageRepo.DB().WithContext(ctx).Create(&attachment).Error; err != nil {
+		if err := tx.WithContext(ctx).Create(&attachment).Error; err != nil {
 			return nil, fmt.Errorf("failed to create attachment: %w", err)
 		}
 		upload.Status = model.DirectUploadStatusCompleted
 		upload.AccountID = message.AccountID
-		if err := s.messageRepo.DB().WithContext(ctx).Save(&upload).Error; err != nil {
+		if err := tx.WithContext(ctx).Save(&upload).Error; err != nil {
 			return nil, fmt.Errorf("failed to mark direct upload completed: %w", err)
 		}
 		attachments = append(attachments, attachment)

backend/internal/service/widget_service_test.go

@@ -16,11 +16,24 @@ import (
 	"gorm.io/gorm"
 	"gorm.io/gorm/logger"
 
+	"github.com/gochat/gochat/internal/channel"
 	"github.com/gochat/gochat/internal/model"
+	channelmodel "github.com/gochat/gochat/internal/model/channel"
 	"github.com/gochat/gochat/internal/repository"
 	"github.com/gochat/gochat/internal/worker"
 )
 
+type widgetLifecycleListener struct {
+	events []*channel.ChannelEvent
+}
+
+func (l *widgetLifecycleListener) Name() string { return "widget-lifecycle-test" }
+
+func (l *widgetLifecycleListener) OnEvent(_ context.Context, event *channel.ChannelEvent) error {
+	l.events = append(l.events, event)
+	return nil
+}
+
 func TestIsGenericShangwutongName(t *testing.T) {
 	assert.True(t, isGenericShangwutongName("商务通访客"))
 	assert.True(t, isGenericShangwutongName("商务通访客·贵州贵阳"))
@@ -48,6 +61,8 @@ func setupWidgetServiceTest(t *testing.T) (*gorm.DB, *WidgetService) {
 		&model.ContactInbox{},
 		&model.Conversation{},
 		&model.Message{},
+		&model.Attachment{},
+		&model.DirectUpload{},
 		&model.WidgetThemeConfig{},
 		&model.PreChatForm{},
 		&model.WidgetFileUpload{},
@@ -58,6 +73,7 @@ func setupWidgetServiceTest(t *testing.T) (*gorm.DB, *WidgetService) {
 		&model.CaptainPreference{},
 		&model.AgentBot{},
 		&model.AgentBotInbox{},
+		&channelmodel.ChannelAPI{},
 	), "failed to auto-migrate models")
 
 	t.Cleanup(func() {
@@ -94,6 +110,271 @@ func setupWidgetServiceTest(t *testing.T) (*gorm.DB, *WidgetService) {
 	return db, svc
 }
 
+func TestWidgetService_SendMessageDispatchesCompleteLifecycle(t *testing.T) {
+	db, svc := setupWidgetServiceTest(t)
+	seedWidgetInbox(t, db)
+	initResp, err := svc.Init(context.Background(), WidgetInitRequest{WebsiteToken: "test_ws_token_123"})
+	require.NoError(t, err)
+
+	listener := &widgetLifecycleListener{}
+	dispatcher := channel.NewDispatcher()
+	dispatcher.Register(listener)
+	svc.SetDispatcher(dispatcher)
+
+	resp, err := svc.SendMessage(context.Background(), WidgetSendMessageRequest{
+		WidgetToken: initResp.WidgetToken,
+		Content:     "hello from widget",
+	})
+	require.NoError(t, err)
+	require.Len(t, listener.events, 2)
+	assert.Equal(t, channel.EventConversationCreated, listener.events[0].Type)
+	assert.Equal(t, channel.EventMessageCreated, listener.events[1].Type)
+	assert.Equal(t, resp.ConversationID, listener.events[1].ConversationID)
+	assert.Equal(t, initResp.WidgetToken, listener.events[1].Data["widget_token"])
+	assert.IsType(t, &model.Conversation{}, listener.events[1].Data["conversation"])
+	assert.IsType(t, &model.Contact{}, listener.events[1].Data["contact"])
+	assert.IsType(t, &model.Message{}, listener.events[1].Data["message"])
+}
+
+func TestWidgetService_SendMessageUpdatesExistingConversationActivityBeforeDispatch(t *testing.T) {
+	db, svc := setupWidgetServiceTest(t)
+	seedWidgetInbox(t, db)
+	initResp, err := svc.Init(context.Background(), WidgetInitRequest{WebsiteToken: "test_ws_token_123"})
+	require.NoError(t, err)
+
+	oldTimestamp := time.Now().Add(-time.Hour).Unix()
+	conversation := &model.Conversation{
+		AccountID: initResp.AccountID, InboxID: initResp.InboxID, ContactID: initResp.ContactID,
+		ContactInboxID: &initResp.ContactInboxID, Status: string(model.ConversationStatusOpen),
+		ChannelType: string(model.InboxChannelTypeWebWidget), Channel: string(model.InboxChannelTypeWebWidget),
+		LastActivityAt: &oldTimestamp, LastMessageAt: &oldTimestamp,
+	}
+	require.NoError(t, db.Create(conversation).Error)
+
+	listener := &widgetLifecycleListener{}
+	dispatcher := channel.NewDispatcher()
+	dispatcher.Register(listener)
+	svc.SetDispatcher(dispatcher)
+
+	resp, err := svc.SendMessage(context.Background(), WidgetSendMessageRequest{
+		WidgetToken:    initResp.WidgetToken,
+		Content:        "new visitor message",
+		ConversationID: &conversation.ID,
+	})
+	require.NoError(t, err)
+	require.Len(t, listener.events, 1)
+
+	wantTimestamp := resp.Message.CreatedAt.Unix()
+	var persisted model.Conversation
+	require.NoError(t, db.First(&persisted, conversation.ID).Error)
+	require.NotNil(t, persisted.LastActivityAt)
+	require.NotNil(t, persisted.LastMessageAt)
+	assert.Equal(t, wantTimestamp, *persisted.LastActivityAt)
+	assert.Equal(t, wantTimestamp, *persisted.LastMessageAt)
+
+	eventConversation, ok := listener.events[0].Data["conversation"].(*model.Conversation)
+	require.True(t, ok)
+	require.NotNil(t, eventConversation.LastActivityAt)
+	require.NotNil(t, eventConversation.LastMessageAt)
+	assert.Greater(t, *eventConversation.LastActivityAt, oldTimestamp)
+	assert.Equal(t, wantTimestamp, *eventConversation.LastActivityAt)
+	assert.Equal(t, wantTimestamp, *eventConversation.LastMessageAt)
+}
+
+func TestWidgetService_PublicCreateMessageUpdatesConversationBeforeDispatch(t *testing.T) {
+	db, svc := setupWidgetServiceTest(t)
+	_, _, _, _, conversation, displayID, _ := seedPublicMessageTest(t, db, model.ConversationStatusOpen)
+
+	listener := &widgetLifecycleListener{}
+	dispatcher := channel.NewDispatcher()
+	dispatcher.Register(listener)
+	svc.SetDispatcher(dispatcher)
+
+	message, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, PublicMessageRequest{Content: "hello"})
+	require.NoError(t, err)
+	require.Len(t, listener.events, 1)
+
+	wantTimestamp := message.CreatedAt.Unix()
+	var persisted model.Conversation
+	require.NoError(t, db.First(&persisted, conversation.ID).Error)
+	require.NotNil(t, persisted.LastActivityAt)
+	require.NotNil(t, persisted.LastMessageAt)
+	assert.Equal(t, wantTimestamp, *persisted.LastActivityAt)
+	assert.Equal(t, wantTimestamp, *persisted.LastMessageAt)
+
+	event := listener.events[0]
+	assert.Equal(t, channel.EventMessageCreated, event.Type)
+	assert.Equal(t, channel.ChannelAPI, event.Channel)
+	assert.Equal(t, "visitor-token", event.Data["widget_token"])
+	assert.Same(t, message, event.Data["message"])
+	assert.IsType(t, &model.Contact{}, event.Data["contact"])
+	eventConversation, ok := event.Data["conversation"].(*model.Conversation)
+	require.True(t, ok)
+	require.NotNil(t, eventConversation.LastActivityAt)
+	assert.Equal(t, wantTimestamp, *eventConversation.LastActivityAt)
+}
+
+func TestWidgetService_PublicCreateMessageRollsBackSecondAttachmentFailureAndRetries(t *testing.T) {
+	db, svc := setupWidgetServiceTest(t)
+	account, _, _, _, conversation, displayID, oldTimestamp := seedPublicMessageTest(t, db, model.ConversationStatusOpen)
+	firstUpload := seedWidgetDirectUpload(t, db, account.ID, "upload-one")
+
+	listener := &widgetLifecycleListener{}
+	dispatcher := channel.NewDispatcher()
+	dispatcher.Register(listener)
+	svc.SetDispatcher(dispatcher)
+	req := PublicMessageRequest{Content: "hello", AttachmentIDs: []string{firstUpload.UploadUUID, "upload-two"}}
+
+	_, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req)
+	require.Error(t, err)
+	assertPublicMessageRollback(t, db, conversation.ID, oldTimestamp, firstUpload.ID)
+	assert.Empty(t, listener.events)
+
+	secondUpload := seedWidgetDirectUpload(t, db, account.ID, "upload-two")
+	message, _, attachments, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req)
+	require.NoError(t, err)
+	require.Len(t, attachments, 2)
+	assert.Len(t, listener.events, 1)
+	var messageCount int64
+	require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error)
+	assert.Equal(t, int64(1), messageCount)
+	for _, uploadID := range []uint{firstUpload.ID, secondUpload.ID} {
+		var upload model.DirectUpload
+		require.NoError(t, db.First(&upload, uploadID).Error)
+		assert.Equal(t, model.DirectUploadStatusCompleted, upload.Status)
+	}
+	for _, attachment := range attachments {
+		assert.Equal(t, message.ID, attachment.MessageID)
+	}
+}
+
+func TestWidgetService_PublicCreateMessageRollsBackReopenFailureAndRetries(t *testing.T) {
+	db, svc := setupWidgetServiceTest(t)
+	_, _, _, _, conversation, displayID, oldTimestamp := seedPublicMessageTest(t, db, model.ConversationStatusResolved)
+	require.NoError(t, db.Exec(`CREATE TRIGGER fail_public_reopen BEFORE UPDATE OF status ON conversations
+		WHEN NEW.id = `+fmt.Sprint(conversation.ID)+` AND NEW.status = 'open'
+		BEGIN SELECT RAISE(FAIL, 'reopen failed'); END`).Error)
+
+	listener := &widgetLifecycleListener{}
+	dispatcher := channel.NewDispatcher()
+	dispatcher.Register(listener)
+	svc.SetDispatcher(dispatcher)
+	req := PublicMessageRequest{Content: "reopen me"}
+
+	_, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req)
+	require.Error(t, err)
+	assertPublicMessageRollback(t, db, conversation.ID, oldTimestamp, 0)
+	assert.Empty(t, listener.events)
+	var rolledBack model.Conversation
+	require.NoError(t, db.First(&rolledBack, conversation.ID).Error)
+	assert.Equal(t, string(model.ConversationStatusResolved), rolledBack.Status)
+
+	require.NoError(t, db.Exec("DROP TRIGGER fail_public_reopen").Error)
+	message, reopened, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req)
+	require.NoError(t, err)
+	assert.Equal(t, string(model.ConversationStatusOpen), reopened.Status)
+	assert.Equal(t, message.CreatedAt.Unix(), *reopened.LastActivityAt)
+	assert.Len(t, listener.events, 1)
+	var messageCount int64
+	require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error)
+	assert.Equal(t, int64(1), messageCount)
+}
+
+func TestWidgetService_PublicCreateMessageRollsBackCaptainJobFailureAndRetries(t *testing.T) {
+	db, svc := setupWidgetServiceTest(t)
+	account, inbox, _, _, conversation, displayID, oldTimestamp := seedPublicMessageTest(t, db, model.ConversationStatusPending)
+	assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Captain", Status: model.AssistantStatusActive, Config: json.RawMessage(`{}`)}
+	require.NoError(t, db.Create(assistant).Error)
+	require.NoError(t, db.Create(&model.CaptainInbox{AccountID: account.ID, InboxID: inbox.ID, AssistantID: assistant.ID}).Error)
+	require.NoError(t, db.Create(&model.CaptainPreference{AccountID: account.ID, AutoReplyEnabled: true}).Error)
+	svc.SetWorkerPool(worker.NewWorkerPool(db))
+	upload := seedWidgetDirectUpload(t, db, account.ID, "captain-upload")
+	require.NoError(t, db.Exec(`CREATE TRIGGER fail_captain_job BEFORE INSERT ON background_jobs
+		WHEN NEW.job_type = '`+TaskTypeCaptainConversationResponseBuilder+`'
+		BEGIN SELECT RAISE(FAIL, 'captain enqueue failed'); END`).Error)
+
+	listener := &widgetLifecycleListener{}
+	dispatcher := channel.NewDispatcher()
+	dispatcher.Register(listener)
+	svc.SetDispatcher(dispatcher)
+	req := PublicMessageRequest{Content: "ask captain", AttachmentIDs: []string{upload.UploadUUID}}
+
+	_, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req)
+	require.Error(t, err)
+	assertPublicMessageRollback(t, db, conversation.ID, oldTimestamp, upload.ID)
+	assert.Empty(t, listener.events)
+	var rolledBack model.Conversation
+	require.NoError(t, db.First(&rolledBack, conversation.ID).Error)
+	assert.Equal(t, string(model.ConversationStatusPending), rolledBack.Status)
+
+	require.NoError(t, db.Exec("DROP TRIGGER fail_captain_job").Error)
+	message, _, attachments, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req)
+	require.NoError(t, err)
+	require.Len(t, attachments, 1)
+	assert.Len(t, listener.events, 1)
+	var job model.BackgroundJob
+	require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).First(&job).Error)
+	assert.Contains(t, string(job.Payload), fmt.Sprintf(`"message_id":%d`, message.ID))
+	var messageCount int64
+	require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error)
+	assert.Equal(t, int64(1), messageCount)
+}
+
+func seedPublicMessageTest(t *testing.T, db *gorm.DB, status model.ConversationStatus) (*model.Account, *model.Inbox, *model.Contact, *model.ContactInbox, *model.Conversation, uint, int64) {
+	t.Helper()
+	account := &model.Account{Name: "Public API Account", Locale: "en", Status: "active"}
+	require.NoError(t, db.Create(account).Error)
+	inbox := &model.Inbox{AccountID: account.ID, Name: "API Inbox", ChannelType: "api", ChannelID: 1, Enabled: true}
+	require.NoError(t, db.Create(inbox).Error)
+	require.NoError(t, db.Create(&channelmodel.ChannelAPI{InboxID: inbox.ID, Identifier: "public-api"}).Error)
+	contact := &model.Contact{AccountID: account.ID, Name: "Public Visitor"}
+	require.NoError(t, db.Create(contact).Error)
+	contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor-source", PubsubToken: "visitor-token"}
+	require.NoError(t, db.Create(contactInbox).Error)
+	displayID := uint(42)
+	oldTimestamp := time.Now().Add(-time.Hour).Unix()
+	conversation := &model.Conversation{
+		AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, ContactInboxID: &contactInbox.ID,
+		DisplayID: &displayID, Status: string(status), ChannelType: "api", Channel: "api",
+		LastActivityAt: &oldTimestamp, LastMessageAt: &oldTimestamp,
+	}
+	require.NoError(t, db.Create(conversation).Error)
+	return account, inbox, contact, contactInbox, conversation, displayID, oldTimestamp
+}
+
+func seedWidgetDirectUpload(t *testing.T, db *gorm.DB, accountID uint, uploadUUID string) *model.DirectUpload {
+	t.Helper()
+	upload := &model.DirectUpload{
+		UploadUUID: uploadUUID, AccountID: accountID, Status: model.DirectUploadStatusPending,
+		Source: model.DirectUploadSourceWidget, OriginalName: uploadUUID + ".txt", FileType: "file",
+		FileSize: 10, FileURL: "/uploads/" + uploadUUID, ExpiresAt: time.Now().Add(time.Hour),
+	}
+	require.NoError(t, db.Create(upload).Error)
+	return upload
+}
+
+func assertPublicMessageRollback(t *testing.T, db *gorm.DB, conversationID uint, oldTimestamp int64, uploadID uint) {
+	t.Helper()
+	var messageCount, attachmentCount, jobCount int64
+	require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversationID).Count(&messageCount).Error)
+	require.NoError(t, db.Model(&model.Attachment{}).Count(&attachmentCount).Error)
+	require.NoError(t, db.Model(&model.BackgroundJob{}).Count(&jobCount).Error)
+	assert.Zero(t, messageCount)
+	assert.Zero(t, attachmentCount)
+	assert.Zero(t, jobCount)
+	var conversation model.Conversation
+	require.NoError(t, db.First(&conversation, conversationID).Error)
+	require.NotNil(t, conversation.LastActivityAt)
+	require.NotNil(t, conversation.LastMessageAt)
+	assert.Equal(t, oldTimestamp, *conversation.LastActivityAt)
+	assert.Equal(t, oldTimestamp, *conversation.LastMessageAt)
+	if uploadID != 0 {
+		var upload model.DirectUpload
+		require.NoError(t, db.First(&upload, uploadID).Error)
+		assert.Equal(t, model.DirectUploadStatusPending, upload.Status)
+	}
+}
+
 // seedWidgetInbox creates an Account + a web_widget Inbox with a known website_token
 // in its ChannelConfig JSON, suitable for Init/SendMessage tests.
 func seedWidgetInbox(t *testing.T, db *gorm.DB) (*model.Account, *model.Inbox) {

backend/internal/wsevent/bridge_listener.go

@@ -66,7 +66,11 @@ func (l *BridgeListener) OnEvent(ctx context.Context, event *channel.ChannelEven
 		payload["inbox_id"] = event.InboxID
 	}
 
-	l.publisher.PublishEvent(event.AccountID, eventType, payload)
+	if pubsubToken, _ := event.Data["widget_token"].(string); pubsubToken != "" {
+		l.publisher.PublishWidgetEvent(event.AccountID, pubsubToken, eventType, payload)
+	} else {
+		l.publisher.PublishEvent(event.AccountID, eventType, payload)
+	}
 
 	applogger.L().Debugf("ws_bridge: forwarded event %s for account %d", eventType, event.AccountID)
 	return nil
@@ -150,17 +154,72 @@ func messagePushPayload(message *model.Message, data map[string]interface{}) map
 			conversationPayload["last_activity_at"] = *conversation.LastActivityAt
 		}
 	}
+	senderType := senderTypeName(message.SenderType)
 	if contact, ok := eventContact(data); ok {
-		payload["sender"] = contactPushPayload(contact)
 		conversationPayload["contact_inbox"] = map[string]interface{}{
 			"source_id": contact.Identifier,
 		}
+		if senderType == "Contact" {
+			payload["sender"] = contactPushPayload(contact)
+		}
 	}
 	payload["conversation_id"] = conversationID
 	payload["conversation"] = conversationPayload
+	if senderType == "User" || senderType == "AgentBot" {
+		if sender, ok := eventSender(data); ok {
+			expectedType := "user"
+			if senderType == "AgentBot" {
+				expectedType = "agent_bot"
+			}
+			if sender["type"] == expectedType {
+				payload["sender"] = sender
+			}
+		}
+	}
+	payload["sender_type"] = senderType
 	return payload
 }
 
+func eventSender(data map[string]interface{}) (map[string]interface{}, bool) {
+	switch sender := data["sender"].(type) {
+	case *model.User:
+		if sender == nil {
+			return nil, false
+		}
+		availableName := sender.DisplayName
+		if availableName == "" {
+			availableName = sender.Name
+		}
+		return map[string]interface{}{
+			"id": sender.ID, "name": sender.Name, "available_name": availableName,
+			"avatar_url": sender.AvatarURL, "thumbnail": sender.AvatarURL,
+			"type": "user",
+		}, true
+	case model.User:
+		return eventSender(map[string]interface{}{"sender": &sender})
+	case *model.AgentBot:
+		if sender == nil {
+			return nil, false
+		}
+		return sender.PushEventData(), true
+	case model.AgentBot:
+		return eventSender(map[string]interface{}{"sender": &sender})
+	case *model.CaptainAssistant:
+		if sender == nil {
+			return nil, false
+		}
+		return map[string]interface{}{
+			"id": sender.ID, "name": sender.Name, "type": "agent_bot",
+		}, true
+	case model.CaptainAssistant:
+		return eventSender(map[string]interface{}{"sender": &sender})
+	case map[string]interface{}:
+		return sender, true
+	default:
+		return nil, false
+	}
+}
+
 func contactPushPayload(contact *model.Contact) map[string]interface{} {
 	payload := modelMap(contact)
 	payload["created_at"] = contact.CreatedAt.Unix()
@@ -246,6 +305,20 @@ func messageTypeValue(value string) int {
 	}
 }
 
+func senderTypeName(value string) string {
+	trimmed := strings.TrimSpace(value)
+	switch strings.ToLower(trimmed) {
+	case "contact":
+		return "Contact"
+	case "agentbot", "agent_bot", "captain::assistant", "captainassistant", "captain_assistant":
+		return "AgentBot"
+	case "user":
+		return "User"
+	default:
+		return trimmed
+	}
+}
+
 func nonEmpty(value, fallback string) string {
 	if strings.TrimSpace(value) == "" {
 		return fallback

backend/internal/wsevent/bridge_listener_test.go

@@ -14,6 +14,8 @@ import (
 type captureHub struct {
 	accountID uint
 	data      []byte
+	room      string
+	roomData  []byte
 }
 
 func (h *captureHub) SendToAccount(accountID uint, data []byte) {
@@ -21,7 +23,89 @@ func (h *captureHub) SendToAccount(accountID uint, data []byte) {
 	h.data = append([]byte(nil), data...)
 }
 
-func (h *captureHub) SendToRoom(string, []byte) {}
+func (h *captureHub) SendToRoom(room string, data []byte) {
+	h.room = room
+	h.roomData = append([]byte(nil), data...)
+}
+
+func TestBridgeListenerRoutesWebWidgetEventsToDashboardAndVisitor(t *testing.T) {
+	event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, 1, 4)
+	event.Data["widget_token"] = "visitor-token"
+	event.Data["message"] = &model.Message{Base: model.Base{ID: 12}, AccountID: 1, InboxID: 4, ConversationID: 2, MessageType: "outgoing", SenderType: "User"}
+	event.Data["sender"] = &model.User{Base: model.Base{ID: 8}, Name: "Agent A"}
+
+	hub := &captureHub{}
+	listener := New(wspkg.NewEventPublisherLocal(hub, nil))
+	if err := listener.OnEvent(context.Background(), event); err != nil {
+		t.Fatalf("OnEvent failed: %v", err)
+	}
+	if hub.accountID != 1 || len(hub.data) == 0 {
+		t.Fatalf("dashboard account room did not receive event: account=%d", hub.accountID)
+	}
+	if hub.room != "pubsub_token_visitor-token" || len(hub.roomData) == 0 {
+		t.Fatalf("visitor room did not receive event: room=%q", hub.room)
+	}
+	var envelope map[string]interface{}
+	if err := json.Unmarshal(hub.roomData, &envelope); err != nil {
+		t.Fatalf("decode visitor event: %v", err)
+	}
+	payload := envelope["data"].(map[string]interface{})
+	if payload["sender_type"] != "User" || payload["sender"].(map[string]interface{})["name"] != "Agent A" {
+		t.Fatalf("visitor reply sender contract mismatch: %#v", payload)
+	}
+}
+
+func TestBridgeListenerPreservesWebWidgetSenderTypeContract(t *testing.T) {
+	tests := []struct {
+		senderType   string
+		expectedType string
+		sender       interface{}
+		expectedName string
+	}{
+		{senderType: "AgentBot", expectedType: "AgentBot", sender: &model.AgentBot{ID: 7, Name: "Reply Bot"}, expectedName: "Reply Bot"},
+		{senderType: "Captain::Assistant", expectedType: "AgentBot", sender: &model.CaptainAssistant{Base: model.Base{ID: 8}, Name: "Captain"}, expectedName: "Captain"},
+		{senderType: "CaptainAssistant", expectedType: "AgentBot", sender: &model.CaptainAssistant{Base: model.Base{ID: 8}, Name: "Captain"}, expectedName: "Captain"},
+		{senderType: "CustomSender", expectedType: "CustomSender"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.senderType, func(t *testing.T) {
+			event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, 1, 4)
+			event.Data["widget_token"] = "visitor-token"
+			event.Data["contact"] = &model.Contact{Base: model.Base{ID: 9}, Name: "Visitor", Identifier: "visitor-9"}
+			if tt.sender != nil {
+				event.Data["sender"] = tt.sender
+			}
+			event.Data["message"] = &model.Message{
+				Base: model.Base{ID: 12}, AccountID: 1, InboxID: 4, ConversationID: 2,
+				MessageType: "outgoing", SenderType: tt.senderType,
+			}
+
+			hub := &captureHub{}
+			listener := New(wspkg.NewEventPublisherLocal(hub, nil))
+			if err := listener.OnEvent(context.Background(), event); err != nil {
+				t.Fatalf("OnEvent failed: %v", err)
+			}
+			var envelope map[string]interface{}
+			if err := json.Unmarshal(hub.roomData, &envelope); err != nil {
+				t.Fatalf("decode visitor event: %v", err)
+			}
+			payload := envelope["data"].(map[string]interface{})
+			if payload["sender_type"] != tt.expectedType {
+				t.Fatalf("sender_type=%#v, want %q", payload["sender_type"], tt.expectedType)
+			}
+			if tt.expectedName == "" {
+				if _, exists := payload["sender"]; exists {
+					t.Fatalf("unknown sender must not reuse contact: %#v", payload["sender"])
+				}
+				return
+			}
+			sender, ok := payload["sender"].(map[string]interface{})
+			if !ok || sender["name"] != tt.expectedName || sender["type"] != "agent_bot" {
+				t.Fatalf("sender does not match sender_type: %#v", payload)
+			}
+		})
+	}
+}
 
 func TestBridgeListenerMessageCreatedUsesChatwootPushPayload(t *testing.T) {
 	now := time.Unix(1_783_834_149, 0)