refactor(server): 会话/访客/设置业务逻辑下沉 internal/service 层，handler 瘦身

- handler 只保留请求绑定、权限检查与响应映射；业务校验顺序、状态码与文案逐字保留 - 新增 service.Error 语义错误，由 handler 统一映射 HTTP 响应 - session 875→361 行、widget 745→321 行、settings 540→73 行 - 自动分配/租户设置(工作时间、欢迎语)/消息校验/黑名单检查/联系方式提取随迁 service 包 - 相关单元测试随迁 service 包，go test ./... 全部通过
2026-09-05 14:39:34 +08:00
parent ba3b1534f8
commit 8b9f3d710f
24 changed files with 2158 additions and 1963 deletions
@@ -0,0 +1,651 @@
+package service
+
+import (
+	"sort"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"gorm.io/gorm"
+	"kefu-cloud/server/internal/model"
+)
+
+// Actor 请求发起者（由 handler 从鉴权中间件结果构造），service 层据此做数据范围判断。
+type Actor struct {
+	UserID     uint
+	SessionAll bool // 对 session 模块具有全量数据范围（主管/管理员）
+	HistoryAll bool // 对 chat_history 模块具有全量数据范围
+}
+
+// SessionListItem 会话列表项（含批量补全的关联信息）。
+type SessionListItem struct {
+	model.Session
+	UnreadCount   int        `json:"unread_count"`
+	LastMessage   string     `json:"last_message"`
+	LastMessageAt *time.Time `json:"last_message_at,omitempty"`
+	MessageCount  int        `json:"message_count"`
+	CustomerName  string     `json:"customer_name"`
+	AgentName     string     `json:"agent_name"`
+	ChannelName   string     `json:"channel_name"`
+	ChannelType   string     `json:"channel_type"`
+}
+
+// AgentItem 可转接/筛选坐席项。
+type AgentItem struct {
+	ID       uint   `json:"id"`
+	Nickname string `json:"nickname"`
+	Status   string `json:"status"` // online | busy | offline | disabled
+	Role     string `json:"role,omitempty"`
+}
+
+// SessionFilter 会话列表筛选。
+type SessionFilter struct {
+	Status    string
+	Priority  string
+	AgentID   string
+	ChannelID string
+	Search    string
+	From      string
+	To        string
+	Page      int
+	PageSize  int
+}
+
+// LoadTenantSession 加载本租户会话；跨租户访问返回 403。
+func LoadTenantSession(id string, tenantID uint) (*model.Session, error) {
+	var session model.Session
+	if err := model.DB.First(&session, id).Error; err != nil {
+		return nil, NotFound("会话不存在")
+	}
+	if session.TenantID != tenantID {
+		return nil, Forbidden("无权访问其他租户会话")
+	}
+	return &session, nil
+}
+
+// CanReadSession 按会话状态区分数据范围（进行中走 session，已结束走 chat_history）。
+func CanReadSession(actor Actor, session *model.Session) bool {
+	all := actor.SessionAll
+	if session.Status == "ended" || session.Status == "archived" {
+		all = actor.HistoryAll
+	}
+	if all {
+		return true
+	}
+	return session.Status == "waiting" || (session.AgentID != nil && *session.AgentID == actor.UserID)
+}
+
+// CanOperateSession 是否可操作（转接/结束/备注/发消息）该会话。
+func CanOperateSession(actor Actor, session *model.Session) bool {
+	if actor.SessionAll {
+		return true
+	}
+	return session.AgentID != nil && *session.AgentID == actor.UserID
+}
+
+// loadAssignableAgent 校验目标坐席属于本租户、未停用，且具备接待能力。
+func loadAssignableAgent(tenantID, agentID uint) error {
+	var agent model.User
+	if err := model.DB.First(&agent, agentID).Error; err != nil {
+		return err
+	}
+	if agent.TenantID != tenantID || agent.Status == "disabled" {
+		return gorm.ErrRecordNotFound
+	}
+	if agent.Role == "agent" {
+		return nil
+	}
+	var count int64
+	if err := model.DB.Table("roles r").
+		Joins("JOIN role_permissions rp ON rp.role_id = r.id").
+		Joins("JOIN permissions p ON p.id = rp.permission_id").
+		Where("r.tenant_id = ? AND r.code = ? AND r.type = ? AND p.code = ?", tenantID, agent.Role, "custom", "session.reply").
+		Count(&count).Error; err != nil || count == 0 {
+		return gorm.ErrRecordNotFound
+	}
+	return nil
+}
+
+// assignableRoleCodes 可被分配/转接的角色（内置 agent + 拥有 session.reply 的自定义角色）。
+func assignableRoleCodes(tenantID uint) []string {
+	codes := []string{"agent"}
+	var customCodes []string
+	model.DB.Table("roles r").
+		Joins("JOIN role_permissions rp ON rp.role_id = r.id").
+		Joins("JOIN permissions p ON p.id = rp.permission_id").
+		Where("r.tenant_id = ? AND r.type = ? AND p.code = ?", tenantID, "custom", "session.reply").
+		Distinct().Pluck("r.code", &customCodes)
+	return append(codes, customCodes...)
+}
+
+func unreadCount(session model.Session) int {
+	var count int64
+	model.DB.Model(&model.Message{}).
+		Where("session_id = ? AND sender_type = ? AND seq > ?", session.ID, "visitor", session.LastReadSeq).
+		Count(&count)
+	return int(count)
+}
+
+// ListSessions 会话列表：筛选、分页并批量补全客户/坐席/渠道/消息数。
+func ListSessions(actor Actor, tenantID uint, f SessionFilter) ([]SessionListItem, int64, error) {
+	var sessions []model.Session
+	var total int64
+
+	query := model.DB.Model(&model.Session{}).Where("sessions.tenant_id = ?", tenantID)
+	if !actor.SessionAll || !actor.HistoryAll {
+		query = query.Where("sessions.agent_id = ? OR sessions.status = ?", actor.UserID, "waiting")
+	}
+	if f.Status != "" {
+		query = query.Where("sessions.status = ?", f.Status)
+	}
+	if f.Priority != "" {
+		query = query.Where("sessions.priority = ?", f.Priority)
+	}
+	if f.AgentID != "" {
+		query = query.Where("sessions.agent_id = ?", f.AgentID)
+	}
+	if f.ChannelID != "" {
+		query = query.Where("sessions.channel_id = ?", f.ChannelID)
+	}
+	if f.From != "" {
+		if t, err := time.ParseInLocation("2006-01-02", f.From, time.Local); err == nil {
+			query = query.Where("sessions.created_at >= ?", t)
+		}
+	}
+	if f.To != "" {
+		if t, err := time.ParseInLocation("2006-01-02", f.To, time.Local); err == nil {
+			query = query.Where("sessions.created_at < ?", t.Add(24*time.Hour))
+		}
+	}
+	if f.Search != "" {
+		like := "%" + f.Search + "%"
+		query = query.Joins("LEFT JOIN customers ON customers.id = sessions.customer_id").
+			Where("customers.name LIKE ? OR CAST(sessions.id AS TEXT) LIKE ? OR COALESCE(sessions.end_reason, '') LIKE ?", like, like, like)
+	}
+
+	query.Count(&total)
+	if err := query.Order("sessions.created_at desc").Offset((f.Page - 1) * f.PageSize).Limit(f.PageSize).Find(&sessions).Error; err != nil {
+		return nil, 0, Internal("查询会话失败")
+	}
+
+	// 批量补全客户 / 客服 / 渠道 / 消息数
+	customerIDs := make([]uint, 0)
+	agentIDs := make([]uint, 0)
+	channelIDs := make([]uint, 0)
+	sessionIDs := make([]uint, 0, len(sessions))
+	for _, s := range sessions {
+		sessionIDs = append(sessionIDs, s.ID)
+		customerIDs = append(customerIDs, s.CustomerID)
+		channelIDs = append(channelIDs, s.ChannelID)
+		if s.AgentID != nil {
+			agentIDs = append(agentIDs, *s.AgentID)
+		}
+	}
+	customerMap := map[uint]model.Customer{}
+	if len(customerIDs) > 0 {
+		var customers []model.Customer
+		model.DB.Where("id IN ?", UniqueUint(customerIDs)).Find(&customers)
+		for _, cu := range customers {
+			customerMap[cu.ID] = cu
+		}
+	}
+	agentMap := map[uint]model.User{}
+	if len(agentIDs) > 0 {
+		var users []model.User
+		model.DB.Where("id IN ?", UniqueUint(agentIDs)).Find(&users)
+		for _, u := range users {
+			agentMap[u.ID] = u
+		}
+	}
+	channelMap := map[uint]model.Channel{}
+	if len(channelIDs) > 0 {
+		var channels []model.Channel
+		model.DB.Where("id IN ?", UniqueUint(channelIDs)).Find(&channels)
+		for _, ch := range channels {
+			channelMap[ch.ID] = ch
+		}
+	}
+	msgCountMap := map[uint]int{}
+	if len(sessionIDs) > 0 {
+		type row struct {
+			SessionID uint
+			Cnt       int
+		}
+		var rows []row
+		model.DB.Model(&model.Message{}).
+			Select("session_id, COUNT(*) as cnt").
+			Where("session_id IN ?", sessionIDs).
+			Group("session_id").
+			Scan(&rows)
+		for _, r := range rows {
+			msgCountMap[r.SessionID] = r.Cnt
+		}
+	}
+
+	items := make([]SessionListItem, 0, len(sessions))
+	for _, session := range sessions {
+		item := SessionListItem{Session: session, MessageCount: msgCountMap[session.ID]}
+		if !actor.SessionAll && session.AgentID != nil && *session.AgentID == actor.UserID {
+			item.UnreadCount = unreadCount(session)
+		}
+		if cu, ok := customerMap[session.CustomerID]; ok {
+			item.CustomerName = cu.Name
+		}
+		if session.AgentID != nil {
+			if u, ok := agentMap[*session.AgentID]; ok {
+				item.AgentName = u.Nickname
+			}
+		}
+		if ch, ok := channelMap[session.ChannelID]; ok {
+			item.ChannelName = ch.Name
+			item.ChannelType = ch.Type
+		}
+		var lastMsg model.Message
+		if err := model.DB.Where("session_id = ?", session.ID).Order("seq desc").First(&lastMsg).Error; err == nil {
+			if lastMsg.Type == "image" {
+				item.LastMessage = "[图片]"
+			} else {
+				item.LastMessage = lastMsg.Content
+				if utf8.RuneCountInString(item.LastMessage) > 40 {
+					runes := []rune(item.LastMessage)
+					item.LastMessage = string(runes[:40]) + "…"
+				}
+			}
+			t := lastMsg.SentAt
+			item.LastMessageAt = &t
+		}
+		items = append(items, item)
+	}
+
+	return items, total, nil
+}
+
+// SessionDetailResult 会话详情聚合结果。
+type SessionDetailResult struct {
+	Messages     []model.Message
+	Events       []model.SessionEvent
+	PageViews    []model.VisitorPageView
+	PendingCount int64
+	MaxSeq       int
+}
+
+// SessionDetail 会话详情：消息、事件、浏览轨迹与同客户排队数。
+func SessionDetail(session *model.Session) (*SessionDetailResult, error) {
+	var messages []model.Message
+	if err := model.DB.Where("session_id = ?", session.ID).Order("seq asc").Find(&messages).Error; err != nil {
+		return nil, Internal("查询消息失败")
+	}
+	var events []model.SessionEvent
+	model.DB.Where("session_id = ?", session.ID).Order("created_at asc").Find(&events)
+	var pageViews []model.VisitorPageView
+	// 最新在前，坐席侧无需滚到底
+	model.DB.Where("session_id = ?", session.ID).Order("entered_at desc").Limit(MaxPageViewsPerSession).Find(&pageViews)
+	var pendingCount int64
+	model.DB.Model(&model.Session{}).Where("customer_id = ? AND tenant_id = ? AND status = ?", session.CustomerID, session.TenantID, "waiting").Count(&pendingCount)
+
+	maxSeq := 0
+	for _, m := range messages {
+		if m.Seq > maxSeq {
+			maxSeq = m.Seq
+		}
+	}
+	return &SessionDetailResult{
+		Messages:     messages,
+		Events:       events,
+		PageViews:    pageViews,
+		PendingCount: pendingCount,
+		MaxSeq:       maxSeq,
+	}, nil
+}
+
+// MessagePage 按 seq 增量拉取的消息页。
+type MessagePage struct {
+	Messages []model.Message
+	AfterSeq int
+	MaxSeq   int
+	HasMore  bool
+}
+
+// SessionMessagePage 按 seq 增量拉取消息（重连 catch-up）。
+func SessionMessagePage(sessionID uint, afterSeq, limit int) (*MessagePage, error) {
+	query := model.DB.Where("session_id = ?", sessionID)
+	if afterSeq > 0 {
+		query = query.Where("seq > ?", afterSeq)
+	}
+
+	var messages []model.Message
+	if err := query.Order("seq asc").Limit(limit).Find(&messages).Error; err != nil {
+		return nil, Internal("查询消息失败")
+	}
+
+	var maxSeq int
+	if err := model.DB.Model(&model.Message{}).
+		Where("session_id = ?", sessionID).
+		Select("COALESCE(MAX(seq), 0)").
+		Scan(&maxSeq).Error; err != nil {
+		return nil, Internal("查询消息失败")
+	}
+
+	return &MessagePage{
+		Messages: messages,
+		AfterSeq: afterSeq,
+		MaxSeq:   maxSeq,
+		HasMore:  len(messages) >= limit && (len(messages) == 0 || messages[len(messages)-1].Seq < maxSeq),
+	}, nil
+}
+
+// MarkSessionRead 坐席标记已读；仅会话归属坐席生效，其余原样返回当前进度。
+func MarkSessionRead(session *model.Session, actor Actor) (int, error) {
+	if session.AgentID == nil || actor.SessionAll || *session.AgentID != actor.UserID {
+		return session.LastReadSeq, nil
+	}
+	var maxSeq int
+	if err := model.DB.Model(&model.Message{}).Where("session_id = ?", session.ID).Select("COALESCE(MAX(seq), 0)").Scan(&maxSeq).Error; err != nil {
+		return 0, Internal("标记已读失败")
+	}
+	if err := model.DB.Model(&model.Session{}).Where("id = ?", session.ID).Update("last_read_seq", maxSeq).Error; err != nil {
+		return 0, Internal("标记已读失败")
+	}
+	return maxSeq, nil
+}
+
+// AddSessionNote 添加内部备注（1-500 字）。
+func AddSessionNote(session *model.Session, operatorID uint, content string) (model.SessionEvent, error) {
+	content = strings.TrimSpace(content)
+	if content == "" || utf8.RuneCountInString(content) > 500 {
+		return model.SessionEvent{}, BadRequest("内部备注需为 1 至 500 字")
+	}
+	event := model.SessionEvent{SessionID: session.ID, OperatorID: operatorID, Action: "note", Detail: content}
+	if err := model.DB.Create(&event).Error; err != nil {
+		return model.SessionEvent{}, Internal("保存内部备注失败")
+	}
+	return event, nil
+}
+
+// ListAvailableAgents 可转接/筛选坐席列表；includeOffline 用于对话记录筛选。
+func ListAvailableAgents(tenantID uint, includeOffline bool) ([]AgentItem, error) {
+	q := model.DB.Where("tenant_id = ? AND role IN ?", tenantID, assignableRoleCodes(tenantID))
+	if !includeOffline {
+		q = q.Where("status IN ?", []string{"online", "busy"})
+	}
+	var users []model.User
+	if err := q.Order("nickname asc").Find(&users).Error; err != nil {
+		return nil, Internal("查询客服失败")
+	}
+	// 在线优先，其次忙碌，再按昵称（Order 已按昵称，稳定排序保持次序）
+	statusRank := map[string]int{"online": 0, "busy": 1, "offline": 2, "disabled": 3}
+	sort.SliceStable(users, func(i, j int) bool {
+		return statusRank[users[i].Status] < statusRank[users[j].Status]
+	})
+	items := make([]AgentItem, 0, len(users))
+	for _, user := range users {
+		items = append(items, AgentItem{
+			ID: user.ID, Nickname: user.Nickname, Status: user.Status, Role: user.Role,
+		})
+	}
+	return items, nil
+}
+
+// CreateSession 主管/管理员代客建会话。
+func CreateSession(tenantID, channelID, customerID uint, priority string) (model.Session, error) {
+	var channel model.Channel
+	if err := model.DB.Where("id = ? AND tenant_id = ? AND status = ?", channelID, tenantID, "enabled").First(&channel).Error; err != nil {
+		return model.Session{}, BadRequest("渠道不存在或未启用")
+	}
+	var customer model.Customer
+	if err := model.DB.Where("id = ? AND tenant_id = ?", customerID, tenantID).First(&customer).Error; err != nil {
+		return model.Session{}, BadRequest("客户不存在")
+	}
+
+	session := model.Session{
+		TenantID:   tenantID,
+		ChannelID:  channelID,
+		CustomerID: customerID,
+		Priority:   priority,
+		Status:     "waiting",
+	}
+
+	if session.Priority == "" {
+		session.Priority = "normal"
+	}
+
+	if err := model.DB.Create(&session).Error; err != nil {
+		return model.Session{}, Internal("创建会话失败")
+	}
+	model.DB.Model(&model.Customer{}).Where("id = ? AND tenant_id = ?", customer.ID, tenantID).
+		Updates(map[string]interface{}{"conversation_count": gorm.Expr("conversation_count + 1"), "last_contact_at": time.Now()})
+
+	return session, nil
+}
+
+// AssignSession 分配/领取会话（仅 waiting）。
+// 无转接权限时强制分配给本人；有权限的非管理员只能领取给自己。
+func AssignSession(session *model.Session, actor Actor, canTransfer bool, agentID uint) (string, error) {
+	if session.Status != "waiting" {
+		return "", Conflict("会话已被分配")
+	}
+	if !canTransfer {
+		agentID = actor.UserID
+	} else if !actor.SessionAll && agentID != actor.UserID {
+		return "", Forbidden("无权分配会话")
+	}
+	if agentID == 0 {
+		return "", BadRequest("请选择目标客服")
+	}
+	if err := loadAssignableAgent(session.TenantID, agentID); err != nil {
+		return "", BadRequest("目标客服不存在或不可用")
+	}
+
+	result := model.DB.Model(&model.Session{}).
+		Where("id = ? AND tenant_id = ? AND status = ?", session.ID, session.TenantID, "waiting").
+		Updates(map[string]interface{}{"agent_id": agentID, "status": "active", "last_read_seq": 0})
+
+	if result.Error != nil {
+		return "", Internal("分配失败")
+	}
+	if result.RowsAffected == 0 {
+		return "", Conflict("会话已被分配")
+	}
+
+	operatorID := actor.UserID
+	operatorName := UserDisplayName(operatorID)
+	targetName := UserDisplayName(agentID)
+	detail := "会话分配给 " + targetName
+	if operatorID == agentID {
+		detail = operatorName + " 领取了会话"
+	} else {
+		detail = operatorName + " 将会话分配给 " + targetName
+	}
+	model.DB.Create(&model.SessionEvent{
+		SessionID:  session.ID,
+		OperatorID: operatorID,
+		Action:     "assign",
+		Detail:     detail,
+	})
+	session.AgentID = &agentID
+	session.Status = "active"
+	BroadcastSessionUpdate(session)
+	return detail, nil
+}
+
+// TransferSession 转接进行中的会话。
+func TransferSession(session *model.Session, actor Actor, agentID uint) (string, error) {
+	if err := loadAssignableAgent(session.TenantID, agentID); err != nil {
+		return "", BadRequest("目标客服不存在或不可用")
+	}
+	if session.AgentID != nil && *session.AgentID == agentID {
+		return "", BadRequest("不能转接给当前接待坐席")
+	}
+
+	operatorID := actor.UserID
+	operatorName := UserDisplayName(operatorID)
+	targetName := UserDisplayName(agentID)
+	fromName := ""
+	if session.AgentID != nil {
+		fromName = UserDisplayName(*session.AgentID)
+	}
+
+	result := model.DB.Model(&model.Session{}).
+		Where("id = ? AND tenant_id = ? AND status = ?", session.ID, session.TenantID, "active").
+		Updates(map[string]interface{}{"agent_id": agentID, "last_read_seq": 0})
+	if result.Error != nil {
+		return "", Internal("转接失败")
+	}
+	if result.RowsAffected == 0 {
+		return "", Conflict("会话不可转接")
+	}
+
+	// 详情写明「谁 → 谁」，便于接手坐席在备注/时间线中一眼看清
+	detail := operatorName + " 转接给 " + targetName
+	if fromName != "" && fromName != operatorName {
+		detail = operatorName + " 将会话由 " + fromName + " 转接给 " + targetName
+	}
+
+	model.DB.Create(&model.SessionEvent{
+		SessionID:  session.ID,
+		OperatorID: operatorID,
+		Action:     "transfer",
+		Detail:     detail,
+	})
+	session.AgentID = &agentID
+	BroadcastSessionUpdate(session)
+
+	return detail, nil
+}
+
+var endReasons = map[string]bool{
+	"resolved": true, "no_response": true, "visitor_left": true, "transferred": true, "other": true,
+}
+
+// EndSession 结束会话（记录原因）。
+func EndSession(session *model.Session, operatorID uint, reason string) error {
+	if reason == "" {
+		return BadRequest("请填写结束原因")
+	}
+	if !endReasons[reason] {
+		return BadRequest("结束原因无效")
+	}
+	now := time.Now()
+
+	result := model.DB.Model(&model.Session{}).
+		Where("id = ? AND tenant_id = ? AND status <> ?", session.ID, session.TenantID, "ended").
+		Updates(map[string]interface{}{"status": "ended", "end_reason": reason, "ended_at": now})
+
+	if result.RowsAffected == 0 {
+		return NotFound("会话不存在")
+	}
+
+	model.DB.Create(&model.SessionEvent{
+		SessionID:  session.ID,
+		OperatorID: operatorID,
+		Action:     "end",
+		Detail:     "结束会话: " + reason,
+	})
+	session.Status = "ended"
+	session.EndedAt = &now
+	BroadcastSessionUpdate(session)
+
+	return nil
+}
+
+// UpdateSessionPriority 调整会话优先级（urgent / normal）。
+func UpdateSessionPriority(session *model.Session, actor Actor, priority string) error {
+	if priority != "urgent" && priority != "normal" {
+		return BadRequest("优先级无效")
+	}
+
+	result := model.DB.Model(&model.Session{}).
+		Where("id = ? AND tenant_id = ?", session.ID, session.TenantID).
+		Update("priority", priority)
+	if result.Error != nil || result.RowsAffected == 0 {
+		return Internal("更新失败")
+	}
+	session.Priority = priority
+	BroadcastSessionUpdate(session)
+
+	return nil
+}
+
+// ArchiveSession 将已结束会话归档。
+func ArchiveSession(session *model.Session, operatorID uint) (*model.Session, error) {
+	if session.Status != "ended" {
+		return nil, BadRequest("仅已结束的会话可归档")
+	}
+	result := model.DB.Model(&model.Session{}).
+		Where("id = ? AND tenant_id = ? AND status = ?", session.ID, session.TenantID, "ended").
+		Update("status", "archived")
+	if result.Error != nil || result.RowsAffected == 0 {
+		return nil, Internal("归档失败")
+	}
+	model.DB.Create(&model.SessionEvent{
+		SessionID:  session.ID,
+		OperatorID: operatorID,
+		Action:     "archive",
+		Detail:     "会话已归档",
+	})
+	session.Status = "archived"
+	BroadcastSessionUpdate(session)
+	return session, nil
+}
+
+// BatchArchiveSessions 批量归档已结束会话，返回归档条数。
+func BatchArchiveSessions(tenantID, operatorID uint, ids []uint) (int64, error) {
+	if len(ids) == 0 {
+		return 0, BadRequest("请选择要归档的会话")
+	}
+	if len(ids) > 100 {
+		return 0, BadRequest("单次最多归档 100 条")
+	}
+	result := model.DB.Model(&model.Session{}).
+		Where("tenant_id = ? AND status = ? AND id IN ?", tenantID, "ended", ids).
+		Update("status", "archived")
+	if result.Error != nil {
+		return 0, Internal("批量归档失败")
+	}
+	for _, id := range ids {
+		model.DB.Create(&model.SessionEvent{
+			SessionID:  id,
+			OperatorID: operatorID,
+			Action:     "archive",
+			Detail:     "会话已归档",
+		})
+	}
+	return result.RowsAffected, nil
+}
+
+// AgentSendMessage 坐席向会话发送消息并广播。
+func AgentSendMessage(session *model.Session, actor Actor, msgType, content string) (*model.Message, error) {
+	validated, err := ValidateMessageContent(msgType, content)
+	if err != nil {
+		return nil, BadRequest(err.Error())
+	}
+
+	if !CanOperateSession(actor, session) {
+		return nil, Forbidden("无权向该会话发送消息")
+	}
+	if session.Status == "ended" || session.Status == "archived" {
+		return nil, Conflict("会话已结束")
+	}
+
+	senderID := actor.UserID
+	msg := model.Message{
+		SessionID:  session.ID,
+		SenderType: "agent",
+		SenderID:   &senderID,
+		Content:    validated,
+		Type:       msgType,
+		SentAt:     time.Now(),
+	}
+
+	if err := model.CreateMessage(&msg); err != nil {
+		return nil, Internal("发送失败")
+	}
+	// 一线坐席发送后同步自己的已读进度；管理员/主管非会话归属，不推进
+	if !actor.SessionAll {
+		model.DB.Model(&model.Session{}).Where("id = ?", session.ID).Update("last_read_seq", msg.Seq)
+		session.LastReadSeq = msg.Seq
+	}
+	BroadcastSessionMessage(session, msg)
+
+	return &msg, nil
+}