diff --git a/internal/chat/session.go b/internal/chat/session.go index 18ebc21e..b9085212 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -43,6 +43,8 @@ type ChatResponse struct { } // Session 多轮对话与协同会话管理器 +// mu 保护 Messages/UserCtx/LastAccessedAt/ProjectID/ActiveStageKey 的并发访问; +// execMu 仅用于同 session 串行执行 LLM 回合,两者正交。 type Session struct { ID string `json:"id"` UserID int64 `json:"user_id"` @@ -58,9 +60,54 @@ type Session struct { Messages []Message `json:"messages"` CreatedAt time.Time `json:"created_at"` LastAccessedAt time.Time `json:"last_accessed_at"` + mu sync.RWMutex `json:"-"` execMu sync.Mutex `json:"-"` } +// SnapshotMessages 返回 Messages 的只读快照(拷贝),调用方可在锁外安全遍历/编码。 +func (s *Session) SnapshotMessages() []Message { + s.mu.RLock() + defer s.mu.RUnlock() + out := make([]Message, len(s.Messages)) + copy(out, s.Messages) + return out +} + +// AppendMessage 在锁内向 Messages 追加一条消息。 +func (s *Session) AppendMessage(m Message) { + s.mu.Lock() + defer s.mu.Unlock() + s.Messages = append(s.Messages, m) +} + +// TouchLastAccessed 更新访问时间戳。 +func (s *Session) TouchLastAccessed() { + s.mu.Lock() + defer s.mu.Unlock() + s.LastAccessedAt = time.Now() +} + +// LastAccessed 读取访问时间戳。 +func (s *Session) LastAccessed() time.Time { + s.mu.RLock() + defer s.mu.RUnlock() + return s.LastAccessedAt +} + +// SetProjectID 设置当前会话绑定的项目 ID。 +func (s *Session) SetProjectID(pid int64) { + s.mu.Lock() + defer s.mu.Unlock() + s.ProjectID = pid +} + +// Project 读取当前会话绑定的项目 ID。 +func (s *Session) Project() int64 { + s.mu.RLock() + defer s.mu.RUnlock() + return s.ProjectID +} + // TurnOptions 客户端回合幂等选项 type TurnOptions struct { ClientTurnID string @@ -261,7 +308,7 @@ func (s *Session) SendMessageStreamTurn(ctx context.Context, userInput string, l func (s *Session) sendMessageInternal(ctx context.Context, userInput string, streamer *stream.Streamer, turn *TurnOptions) (*ChatResponse, error) { start := time.Now() - s.LastAccessedAt = time.Now() + s.TouchLastAccessed() if streamer != nil { streamer.EmitPhase("intent_recognizing", "正在研判您的意图与项目上下文...", "coordinator", 0, 0, 0) } @@ -275,7 +322,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str Content: userInput, Timestamp: time.Now(), } - s.Messages = append(s.Messages, userMsg) + s.AppendMessage(userMsg) var userMsgID int64 clientTurnID := "" @@ -472,7 +519,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str InteractiveCard: coordRes.InteractiveCard, Timestamp: time.Now(), } - s.Messages = append(s.Messages, asstMsg) + s.AppendMessage(asstMsg) if s.CtxMgr != nil { replySrc := coordRes.ReplySource diff --git a/internal/server/chat_document_reader.go b/internal/server/chat_document_reader.go index 36a7d959..059d9cb6 100644 --- a/internal/server/chat_document_reader.go +++ b/internal/server/chat_document_reader.go @@ -19,7 +19,7 @@ func (s *Server) bindSessionDocumentSummaryReader(sess *chat.Session, user *auth return } sess.UserCtx.DocumentSummaryReader = nil - projectID := sess.ProjectID + projectID := sess.Project() userID := int64(0) if user != nil { userID = user.ID diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 54757d08..3ba026c6 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -193,16 +193,16 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { return } s.hydrateVisibleHistory(r.Context(), sess, user.ID) - if req.ProjectID > 0 { - sess.ProjectID = req.ProjectID + if req.ProjectID > 0 && req.ProjectID != sess.Project() { + sess.SetProjectID(req.ProjectID) } - if sess.ProjectID > 0 { - if !s.enforceChatProjectAccess(w, r, user, sess.ProjectID) { + if sess.Project() > 0 { + if !s.enforceChatProjectAccess(w, r, user, sess.Project()) { return } } s.enrichSessionWorkflowStage(r.Context(), sess, user.ID) - if sess.ProjectID > 0 && !s.allowActiveStageChatAccess(r.Context(), sess.ProjectID, user.ID) { + if sess.Project() > 0 && !s.allowActiveStageChatAccess(r.Context(), sess.Project(), user.ID) { s.writeWorkflowAccessDenied(w) return } diff --git a/internal/server/server.go b/internal/server/server.go index bf569257..b2caff30 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -393,7 +393,7 @@ func (s *Server) cleanStaleSessionsLocked(ttl time.Duration) int { now := time.Now() cleaned := 0 for id, sess := range s.sessions { - if sess == nil || now.Sub(sess.LastAccessedAt) > ttl { + if sess == nil || now.Sub(sess.LastAccessed()) > ttl { delete(s.sessions, id) cleaned++ } @@ -432,7 +432,7 @@ func (s *Server) getOrCreateSession( if callerUserID > 0 && existing.UserID > 0 && existing.UserID != callerUserID { return nil, ctxmgr.ErrSessionOwnerMismatch } - existing.LastAccessedAt = time.Now() + existing.TouchLastAccessed() return existing, nil } @@ -443,7 +443,7 @@ func (s *Server) getOrCreateSession( if callerUserID > 0 && existing.UserID > 0 && existing.UserID != callerUserID { return nil, ctxmgr.ErrSessionOwnerMismatch } - existing.LastAccessedAt = time.Now() + existing.TouchLastAccessed() return existing, nil } @@ -540,7 +540,7 @@ func (s *Server) getOrCreateSession( sess.UserID = restored.UserID } if restored.ProjectID > 0 { - sess.ProjectID = restored.ProjectID + sess.SetProjectID(restored.ProjectID) } if len(restored.Facts) > 0 { _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ @@ -550,7 +550,7 @@ func (s *Server) getOrCreateSession( } if s.chatLog == nil { for _, rm := range restored.Messages { - sess.Messages = append(sess.Messages, chat.Message{ + sess.AppendMessage(chat.Message{ ID: fmt.Sprintf("msg_%d", rm.ID), Role: rm.Role, Content: rm.Content, @@ -575,7 +575,7 @@ func (s *Server) getOrCreateSession( // 新建会话时立即绑定调用者。NewSession 历史占位 UserID=1/ProjectID=1 不能当作已绑定归属。 if restored == nil { - sess.ProjectID = 0 + sess.SetProjectID(0) if callerUserID > 0 { sess.UserID = callerUserID } @@ -583,7 +583,7 @@ func (s *Server) getOrCreateSession( sess.UserID = callerUserID } - sess.LastAccessedAt = time.Now() + sess.TouchLastAccessed() s.sessions[sessionID] = sess return sess, nil } @@ -592,7 +592,7 @@ func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session, if s.chatLog == nil || sess == nil || viewerUserID <= 0 { return } - if len(sess.Messages) > 0 { + if len(sess.SnapshotMessages()) > 0 { return } restored, err := s.chatLog.Restore(ctx, chatflow.Principal{UserID: viewerUserID, Action: chatflow.ActionAssemble}, sess.ID) @@ -600,7 +600,7 @@ func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session, return } for _, rm := range restored.Messages { - sess.Messages = append(sess.Messages, chat.Message{ + sess.AppendMessage(chat.Message{ ID: fmt.Sprintf("msg_%d", rm.ID), Role: rm.Role, Content: rm.Content, diff --git a/internal/server/sse.go b/internal/server/sse.go index fbc03db0..ae720088 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -197,11 +197,11 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { return } s.hydrateVisibleHistory(r.Context(), sess, user.ID) - if req.ProjectID > 0 { - sess.ProjectID = req.ProjectID + if req.ProjectID > 0 && req.ProjectID != sess.Project() { + sess.SetProjectID(req.ProjectID) } - if sess.ProjectID > 0 { - if !s.enforceChatProjectAccess(w, r, user, sess.ProjectID) { + if sess.Project() > 0 { + if !s.enforceChatProjectAccess(w, r, user, sess.Project()) { return } } @@ -220,7 +220,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { ctx := stream.WithRequestID(r.Context(), requestID) s.enrichSessionWorkflowStage(ctx, sess, req.UserID) - if sess.ProjectID > 0 && !s.allowActiveStageChatAccess(ctx, sess.ProjectID, req.UserID) { + if sess.Project() > 0 && !s.allowActiveStageChatAccess(ctx, sess.Project(), req.UserID) { writeWorkflowAccessDeniedSSE(w, flusher, req.SessionID) return } diff --git a/internal/server/stage_access_helpers.go b/internal/server/stage_access_helpers.go index 3909f4ae..e435656b 100644 --- a/internal/server/stage_access_helpers.go +++ b/internal/server/stage_access_helpers.go @@ -140,13 +140,13 @@ func (s *Server) enrichSessionWorkflowStage(ctx context.Context, sess *chat.Sess if sess == nil { return } - if sess.ProjectID > 0 && s.instSvc != nil && s.instSvc.Available() { - settings, err := s.instSvc.GetSettings(ctx, sess.ProjectID, userID, false) + if sess.Project() > 0 && s.instSvc != nil && s.instSvc.Available() { + settings, err := s.instSvc.GetSettings(ctx, sess.Project(), userID, false) if err == nil && settings != nil { sess.ActiveStageKey = settings.ActiveStageKey } } - scope, err := s.resolveChatMessageScope(ctx, sess.ProjectID, sess.ActiveStageKey) + scope, err := s.resolveChatMessageScope(ctx, sess.Project(), sess.ActiveStageKey) if err == nil && scope != nil { sess.MessageScope = scope }