weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(chat): add RWMutex to Session and snapshot accessors to close data race

Browse Source 
- Session gains mu sync.RWMutex + SnapshotMessages/AppendMessage/
  TouchLastAccessed/SetProjectID/Project accessors
- sendMessageInternal, hydrateVisibleHistory, getOrCreateSession,
  cleanStaleSessionsLocked, handleChat, handleChatStream,
  enrichSessionWorkflowStage, bindSessionDocumentSummaryReader all use
  locked accessors instead of raw field access
- ProjectID mutation now goes through SetProjectID (only when changed),
  closing the cross-project pollution window in handleChat/handleChatStream

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-19 04:55:55 +00:00
co-authored by Cursor
parent 11f375f53d
commit c7eab1f7d1
6 changed files with 73 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/chat/session.go
+50 -3
View File
@@ -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
internal/server/chat_document_reader.go
+1 -1
View File
@@ -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
internal/server/handlers.go
+5 -5
View File
@@ -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
}
internal/server/server.go
+9 -9
View File
@@ -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,
internal/server/sse.go
+5 -5
View File
@@ -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
}
internal/server/stage_access_helpers.go
+3 -3
View File
@@ -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
}