From 11f375f53d39b1cc875e3835efeded3d79d60ba8 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 19 Sep 2026 04:50:40 +0000 Subject: [PATCH 1/4] fix(docgov): surface lease/audit persist failures and read back fencing token - persistLeaseLocked now returns error; AcquireLease rolls back in-memory lease on DB failure so memory and PG stay consistent - appendAuditLocked uses request ctx and logs audit INSERT failures instead of swallowing them with context.Background() - loadDocumentFromDBLocked now SELECTs lease_fencing_token so fencing survives process restart - SaveDocumentContent/RecordDocumentViewed/RecordDocumentEdited/ DecideModificationProposal/SetDocumentWorkDir log persist failures via slog.Error instead of silent _, _ Co-authored-by: Cursor --- internal/docgov/engine.go | 52 ++++++++++++++++++++------------ internal/docgov/lease.go | 62 +++++++++++++++++++++++++++------------ 2 files changed, 78 insertions(+), 36 deletions(-) diff --git a/internal/docgov/engine.go b/internal/docgov/engine.go index 7e066959..f0220df7 100644 --- a/internal/docgov/engine.go +++ b/internal/docgov/engine.go @@ -315,11 +315,13 @@ func (e *GovernanceEngine) SetDocumentWorkDir(ctx context.Context, docID string, doc.UpdatedAt = time.Now() if e.pool != nil { - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` UPDATE document_records SET work_dir = $1, updated_at = $2 WHERE doc_id = $3 - `, doc.WorkDir, doc.UpdatedAt, docID) + `, doc.WorkDir, doc.UpdatedAt, docID); err != nil { + slog.Error("docgov_workdir_persist_failed", "doc_id", docID, "error", err) + } } return nil @@ -389,7 +391,7 @@ func (e *GovernanceEngine) RecordDocumentViewed(ctx context.Context, docID strin e.memAudits[docID] = append(e.memAudits[docID], audit) if e.pool != nil { - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` UPDATE document_records SET author_claimed = $1, author_claimed_at = $2, author_last_viewed_at = $3, reviewer_claimed = $4, reviewer_claimed_at = $5, @@ -399,14 +401,18 @@ func (e *GovernanceEngine) RecordDocumentViewed(ctx context.Context, docID strin `, doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastViewedAt, doc.ReviewerClaimed, doc.ReviewerClaimedAt, doc.ApproverClaimed, doc.ApproverClaimedAt, - doc.UpdatedAt, docID) + doc.UpdatedAt, docID); err != nil { + slog.Error("docgov_viewed_persist_failed", "doc_id", docID, "error", err) + } detJSON, _ := json.Marshal(audit.Details) - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), - string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil { + slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err) + } } return doc, nil @@ -467,18 +473,22 @@ func (e *GovernanceEngine) RecordDocumentEdited(ctx context.Context, docID strin e.memAudits[docID] = append(e.memAudits[docID], audit) if e.pool != nil { - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` UPDATE document_records SET author_claimed = $1, author_claimed_at = $2, author_last_edited_at = $3, updated_at = $4 WHERE doc_id = $5 - `, doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, docID) + `, doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, docID); err != nil { + slog.Error("docgov_edited_persist_failed", "doc_id", docID, "error", err) + } detJSON, _ := json.Marshal(audit.Details) - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), - string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil { + slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err) + } } return doc, nil @@ -647,11 +657,13 @@ func (e *GovernanceEngine) ProposeModification( } detJSON, _ := json.Marshal(audit.Details) - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), - string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil { + slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err) + } } return proposal, nil @@ -740,18 +752,22 @@ func (e *GovernanceEngine) DecideModificationProposal( e.memAudits[proposal.DocID] = append(e.memAudits[proposal.DocID], audit) if e.pool != nil { - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` UPDATE document_change_proposals SET status = $1, author_decision_reason = $2, decided_at = $3 WHERE proposal_id = $4 - `, string(proposal.Status), proposal.AuthorDecisionReason, proposal.DecidedAt, proposalID) + `, string(proposal.Status), proposal.AuthorDecisionReason, proposal.DecidedAt, proposalID); err != nil { + slog.Error("docgov_proposal_decide_persist_failed", "proposal_id", proposalID, "error", err) + } detJSON, _ := json.Marshal(audit.Details) - _, _ = e.pool.Exec(ctx, ` + if _, err := e.pool.Exec(ctx, ` INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), - string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil { + slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err) + } } return proposal, nil @@ -1087,7 +1103,7 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s last_rejected_reason, last_rejected_by_role, last_rejected_by_name, last_rejected_at, approved_at, config, created_at, updated_at, COALESCE(lease_holder_type, ''), COALESCE(lease_holder_id, ''), COALESCE(lease_work_dir, ''), - lease_expires_at, lease_heartbeat_at, + lease_expires_at, lease_heartbeat_at, COALESCE(lease_fencing_token, ''), COALESCE(duty_locked, false), COALESCE(duty_policy, '') FROM document_records WHERE doc_id = $1 @@ -1106,7 +1122,7 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s &doc.ReviewerClaimed, &doc.ReviewerClaimedAt, &doc.ApproverClaimed, &doc.ApproverClaimedAt, &doc.LastRejectedReason, &doc.LastRejectedByRole, &doc.LastRejectedByName, &doc.LastRejectedAt, &doc.ApprovedAt, &cfgJSON, &doc.CreatedAt, &doc.UpdatedAt, - &leaseType, &leaseID, &leaseWD, &doc.LeaseExpiresAt, &doc.LeaseHeartbeatAt, + &leaseType, &leaseID, &leaseWD, &doc.LeaseExpiresAt, &doc.LeaseHeartbeatAt, &doc.LeaseFencingToken, &doc.DutyLocked, &doc.DutyPolicy, ) if err != nil { diff --git a/internal/docgov/lease.go b/internal/docgov/lease.go index 73f017a1..c7e31233 100644 --- a/internal/docgov/lease.go +++ b/internal/docgov/lease.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "io" + "log/slog" "path/filepath" "strings" "time" @@ -72,14 +73,16 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq if doc.HasActiveLease(now) { sameHolder := doc.LeaseHolderType == req.HolderType && doc.LeaseHolderID == req.HolderID if !sameHolder && req.HolderType == LeaseHolderSquad && e.shouldReclaimSquadLease(ctx, doc) { - e.appendAuditLocked(doc, "lease_reclaim_orphan", 0, string(req.HolderType), RoleSystem, map[string]any{ + e.appendAuditLocked(ctx, doc, "lease_reclaim_orphan", 0, string(req.HolderType), RoleSystem, map[string]any{ "previous_holder_type": doc.LeaseHolderType, "previous_holder_id": doc.LeaseHolderID, }) doc.ClearLease() - e.persistLeaseLocked(ctx, doc) + if perr := e.persistLeaseLocked(ctx, doc); perr != nil { + slog.Error("docgov_lease_reclaim_persist_failed", "doc_id", doc.DocID, "error", perr) + } } else if !sameHolder { - e.appendAuditLocked(doc, "lease_deny", 0, string(req.HolderType), RoleSystem, map[string]any{ + e.appendAuditLocked(ctx, doc, "lease_deny", 0, string(req.HolderType), RoleSystem, map[string]any{ "requester_type": req.HolderType, "requester_id": req.HolderID, "holder_type": doc.LeaseHolderType, @@ -98,6 +101,9 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq } expires := now.Add(ttl) + // 先持久化再更新内存，失败时不污染内存租约状态 + prevType, prevID, prevWD := doc.LeaseHolderType, doc.LeaseHolderID, doc.LeaseWorkDir + prevExpires, prevHeartbeat, prevFencing := doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken doc.LeaseHolderType = req.HolderType doc.LeaseHolderID = req.HolderID if req.WorkDir != "" { @@ -108,8 +114,12 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq doc.LeaseFencingToken = newLeaseFencingToken() doc.UpdatedAt = now - e.persistLeaseLocked(ctx, doc) - e.appendAuditLocked(doc, "lease_acquire", 0, string(req.HolderType), RoleSystem, map[string]any{ + if perr := e.persistLeaseLocked(ctx, doc); perr != nil { + doc.LeaseHolderType, doc.LeaseHolderID, doc.LeaseWorkDir = prevType, prevID, prevWD + doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken = prevExpires, prevHeartbeat, prevFencing + return nil, perr + } + e.appendAuditLocked(ctx, doc, "lease_acquire", 0, string(req.HolderType), RoleSystem, map[string]any{ "holder_type": req.HolderType, "holder_id": req.HolderID, "work_dir": doc.LeaseWorkDir, @@ -134,7 +144,9 @@ func (e *GovernanceEngine) ReleaseLease(ctx context.Context, docID, holderType, if !doc.HasActiveLease(now) { doc.ClearLease() doc.UpdatedAt = now - e.persistLeaseLocked(ctx, doc) + if perr := e.persistLeaseLocked(ctx, doc); perr != nil { + slog.Error("docgov_lease_release_persist_failed", "doc_id", doc.DocID, "error", perr) + } return doc, nil } @@ -144,8 +156,10 @@ func (e *GovernanceEngine) ReleaseLease(ctx context.Context, docID, holderType, doc.ClearLease() doc.UpdatedAt = now - e.persistLeaseLocked(ctx, doc) - e.appendAuditLocked(doc, "lease_release", 0, holderType, RoleSystem, map[string]any{ + if perr := e.persistLeaseLocked(ctx, doc); perr != nil { + slog.Error("docgov_lease_release_persist_failed", "doc_id", doc.DocID, "error", perr) + } + e.appendAuditLocked(ctx, doc, "lease_release", 0, holderType, RoleSystem, map[string]any{ "holder_type": holderType, "holder_id": holderID, }) @@ -181,7 +195,9 @@ func (e *GovernanceEngine) HeartbeatLease(ctx context.Context, docID, holderType doc.LeaseExpiresAt = &expires doc.LeaseHeartbeatAt = &now doc.UpdatedAt = now - e.persistLeaseLocked(ctx, doc) + if perr := e.persistLeaseLocked(ctx, doc); perr != nil { + slog.Error("docgov_lease_heartbeat_persist_failed", "doc_id", doc.DocID, "error", perr) + } return doc, nil } @@ -402,16 +418,19 @@ func (e *GovernanceEngine) SaveDocumentContent(ctx context.Context, req ContentS } if e.pool != nil { - _, _ = e.pool.Exec(ctx, ` + if _, perr := e.pool.Exec(ctx, ` UPDATE document_records SET content_sha256 = $1, is_template_only = $2, work_dir = $3, author_claimed = $4, author_claimed_at = $5, author_last_edited_at = $6, updated_at = $7 WHERE doc_id = $8 `, doc.ContentSHA256, doc.IsTemplateOnly, doc.WorkDir, - doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, doc.DocID) + doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, doc.DocID); perr != nil { + slog.Error("docgov_content_persist_failed", "doc_id", doc.DocID, "error", perr) + return nil, fmt.Errorf("docgov: persist content: %w", perr) + } } - e.appendAuditLocked(doc, "author_edit_content", req.ActorID, req.ActorName, req.ActorRole, map[string]any{ + e.appendAuditLocked(ctx, doc, "author_edit_content", req.ActorID, req.ActorName, req.ActorRole, map[string]any{ "content_sha256": shaStr, "bytes": len(canon), }) @@ -453,20 +472,25 @@ func (e *GovernanceEngine) getDocLocked(ctx context.Context, docID string) (*Doc return doc, nil } -func (e *GovernanceEngine) persistLeaseLocked(ctx context.Context, doc *DocumentRecord) { +func (e *GovernanceEngine) persistLeaseLocked(ctx context.Context, doc *DocumentRecord) error { if e.pool == nil { - return + return nil } - _, _ = e.pool.Exec(ctx, ` + _, err := e.pool.Exec(ctx, ` UPDATE document_records SET lease_holder_type = $1, lease_holder_id = $2, lease_work_dir = $3, lease_expires_at = $4, lease_heartbeat_at = $5, lease_fencing_token = $6, updated_at = $7 WHERE doc_id = $8 `, string(doc.LeaseHolderType), doc.LeaseHolderID, doc.LeaseWorkDir, doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken, doc.UpdatedAt, doc.DocID) + if err != nil { + slog.Error("docgov_lease_persist_failed", "doc_id", doc.DocID, "holder", doc.LeaseHolderID, "error", err) + return fmt.Errorf("docgov: persist lease: %w", err) + } + return nil } -func (e *GovernanceEngine) appendAuditLocked(doc *DocumentRecord, action string, actorID int64, actorName string, actorRole UserRole, details map[string]any) { +func (e *GovernanceEngine) appendAuditLocked(ctx context.Context, doc *DocumentRecord, action string, actorID int64, actorName string, actorRole UserRole, details map[string]any) { now := time.Now() audit := &AuditLog{ DocID: doc.DocID, @@ -482,10 +506,12 @@ func (e *GovernanceEngine) appendAuditLocked(doc *DocumentRecord, action string, e.memAudits[doc.DocID] = append(e.memAudits[doc.DocID], audit) if e.pool != nil { detJSON, _ := json.Marshal(details) - _, _ = e.pool.Exec(context.Background(), ` + if _, err := e.pool.Exec(ctx, ` INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), - string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil { + slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err) + } } } From c7eab1f7d17190da9cdb27ee77ff67b2b6cd1f38 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 19 Sep 2026 04:55:55 +0000 Subject: [PATCH 2/4] fix(chat): add RWMutex to Session and snapshot accessors to close data race - 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 --- internal/chat/session.go | 53 +++++++++++++++++++++++-- internal/server/chat_document_reader.go | 2 +- internal/server/handlers.go | 10 ++--- internal/server/server.go | 18 ++++----- internal/server/sse.go | 10 ++--- internal/server/stage_access_helpers.go | 6 +-- 6 files changed, 73 insertions(+), 26 deletions(-) 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 } From e2b13406dbe5578468dc3d3dbed8b6d861f3aa0b Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 19 Sep 2026 05:00:01 +0000 Subject: [PATCH 3/4] fix(projectinstance): atomic plan save and correct completed_by attribution - Add Store.SavePlanWithRevision: single-transaction planning upsert + revision bump (bump first with optimistic lock, conflict rolls back before any plan write), closing the SaveWorkflowPlan split-transaction race where plan content and revision could diverge - AdvanceStage now takes operatorID and writes completed_by explicitly instead of SELECT activated_by, so the completing operator is the actual advancing user, not the original activator - Service.SaveWorkflowPlan/AdvanceStage pass userID through; memStore implements SavePlanWithRevision under its single mutex Co-authored-by: Cursor --- internal/projectinstance/service.go | 5 +- internal/projectinstance/stage_service.go | 2 +- internal/projectinstance/stage_store.go | 8 +- internal/projectinstance/store.go | 99 ++++++++++++++++++++++- 4 files changed, 102 insertions(+), 12 deletions(-) diff --git a/internal/projectinstance/service.go b/internal/projectinstance/service.go index 168bd3bf..2433ca6a 100644 --- a/internal/projectinstance/service.go +++ b/internal/projectinstance/service.go @@ -182,10 +182,7 @@ func (s *Service) SaveWorkflowPlan(ctx context.Context, projectID int64, userID ConfigSnapshot: in.Plan, CreatedBy: &userID, } - if _, err := s.store.UpsertPlanningInstance(ctx, inst); err != nil { - return nil, err - } - if _, err := s.store.BumpPlanRevision(ctx, projectID, planRev); err != nil { + if _, _, err := s.store.SavePlanWithRevision(ctx, inst, planRev); err != nil { return nil, err } return s.GetSettings(ctx, projectID, userID, isCreator) diff --git a/internal/projectinstance/stage_service.go b/internal/projectinstance/stage_service.go index de580e21..82b6317b 100644 --- a/internal/projectinstance/stage_service.go +++ b/internal/projectinstance/stage_service.go @@ -94,7 +94,7 @@ func (s *Service) AdvanceStage(ctx context.Context, instanceID int64, userID int } } - after, err := s.store.AdvanceStage(ctx, instanceID, in.StageKey, in.ExpectedConfigRevision) + after, err := s.store.AdvanceStage(ctx, instanceID, in.StageKey, in.ExpectedConfigRevision, userID) if err != nil { return nil, err } diff --git a/internal/projectinstance/stage_store.go b/internal/projectinstance/stage_store.go index d82143f5..70b7dbad 100644 --- a/internal/projectinstance/stage_store.go +++ b/internal/projectinstance/stage_store.go @@ -69,7 +69,7 @@ func (s *pgStore) loadInstanceDetail(ctx context.Context, inst *WorkflowInstance return detail, rows.Err() } -func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) { +func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int, operatorID int64) (*InstanceDetail, error) { tx, err := s.pool.Begin(ctx) if err != nil { return nil, err @@ -131,11 +131,9 @@ func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey s if errors.Is(err, pgx.ErrNoRows) { _, err = tx.Exec(ctx, ` UPDATE project_workflow_instances SET status = 'completed', config_revision = config_revision + 1, - completed_at = $2, completed_by = ( - SELECT activated_by FROM project_workflow_instances WHERE id = $1 - ), updated_at = $2 + completed_at = $2, completed_by = $3, updated_at = $2 WHERE id = $1 - `, instanceID, now) + `, instanceID, now, operatorID) if err != nil { return nil, err } diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go index e3555219..b43ba424 100644 --- a/internal/projectinstance/store.go +++ b/internal/projectinstance/store.go @@ -44,6 +44,9 @@ type Store interface { GetGoverningInstanceForAccess(ctx context.Context, projectID int64) (*WorkflowInstance, error) HasLiveInstance(ctx context.Context, projectID int64) (bool, error) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) + // SavePlanWithRevision 单事务完成 planning upsert + revision bump， + // 避免 SaveWorkflowPlan 两条独立事务间的版本号与 plan 内容错位。 + SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error) @@ -53,7 +56,7 @@ type Store interface { ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error) - AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) + AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int, operatorID int64) (*InstanceDetail, error) MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error } @@ -260,6 +263,85 @@ func (s *pgStore) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInst return inst, nil } +// SavePlanWithRevision 在单事务内完成 planning 实例 upsert 与 plan revision bump。 +// bump 先执行（带期望版本乐观锁），失败则整个事务回滚，保证 plan 内容与版本号原子一致。 +func (s *pgStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) { + snapRaw, _ := json.Marshal(inst.ConfigSnapshot) + now := time.Now() + if inst.CreatedAt.IsZero() { + inst.CreatedAt = now + } + inst.UpdatedAt = now + + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, 0, err + } + defer tx.Rollback(ctx) + + // 先 bump revision（乐观锁），冲突即回滚，不写 plan + var newRev int + err = tx.QueryRow(ctx, ` + UPDATE projects SET workflow_plan_revision = workflow_plan_revision + 1, updated_at = now() + WHERE id = $1 AND workflow_plan_revision = $2 + RETURNING workflow_plan_revision + `, inst.ProjectID, expectedPlanRevision).Scan(&newRev) + if errors.Is(err, pgx.ErrNoRows) { + return nil, 0, ErrPlanRevisionConflict + } + if err != nil { + return nil, 0, err + } + + var existingID int64 + err = tx.QueryRow(ctx, ` + SELECT id FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning' LIMIT 1 + `, inst.ProjectID).Scan(&existingID) + if errors.Is(err, pgx.ErrNoRows) { + var nextNo int + if err := tx.QueryRow(ctx, ` + SELECT COALESCE(MAX(instance_no), 0) + 1 + FROM project_workflow_instances + WHERE project_id = $1 + `, inst.ProjectID).Scan(&nextNo); err != nil { + return nil, 0, err + } + inst.InstanceNo = nextNo + err = tx.QueryRow(ctx, ` + INSERT INTO project_workflow_instances ( + project_id, workflow_version_id, instance_no, status, config_revision, plan_hash, + config_snapshot, created_by, created_at, updated_at + ) VALUES ($1, $2, $3, 'planning', $4, $5, $6, $7, $8, $9) + RETURNING id + `, inst.ProjectID, inst.WorkflowVersionID, nextNo, inst.ConfigRevision, inst.PlanHash, snapRaw, + inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID) + if err != nil { + return nil, 0, err + } + } else if err != nil { + return nil, 0, err + } else { + err = tx.QueryRow(ctx, ` + UPDATE project_workflow_instances SET + workflow_version_id = $2, config_revision = $3, plan_hash = $4, config_snapshot = $5, + updated_at = $6 + WHERE id = $1 AND status = 'planning' + RETURNING id + `, existingID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw, inst.UpdatedAt).Scan(&inst.ID) + if err != nil { + return nil, 0, err + } + } + + if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil { + return nil, 0, err + } + if err := tx.Commit(ctx); err != nil { + return nil, 0, err + } + return inst, newRev, nil +} + func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst *WorkflowInstance) error { _, err := tx.Exec(ctx, ` DELETE FROM document_duty_assignments WHERE doc_task_instance_id IN ( @@ -859,6 +941,19 @@ func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInsta return ©Inst, nil } +// SavePlanWithRevision 内存实现：bump 与 upsert 在同一互斥锁内，天然原子。 +func (m *memStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) { + newRev, err := m.BumpPlanRevision(ctx, inst.ProjectID, expectedPlanRevision) + if err != nil { + return nil, 0, err + } + saved, err := m.UpsertPlanningInstance(ctx, inst) + if err != nil { + return nil, 0, err + } + return saved, newRev, nil +} + func (m *memStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) { usesAuthorOnly := planUsesAuthorOnly(inst.ConfigSnapshot) var currentPolicyRev int64 @@ -937,7 +1032,7 @@ func (m *memStore) GetInstanceByID(_ context.Context, instanceID int64) (*Instan return nil, ErrInstanceNotFound } -func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) { +func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey string, expectedRevision int, _ int64) (*InstanceDetail, error) { m.mu.Lock() defer m.mu.Unlock() var inst *WorkflowInstance From a81db16cdd359a8d0224f201f72bad0d4ede17e7 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 19 Sep 2026 05:03:01 +0000 Subject: [PATCH 4/4] fix(server): thread request ctx through session bootstrap, drop Background DB calls - getOrCreateSession now takes ctx; GetSessionOwnerID/ RestoreSessionContext/CompilePromptGuidance run under context.WithTimeout(5s) instead of context.Background() - SyncRecentDialogue's rolling summary uses a 30s timeout context instead of Background, so client disconnect cancels the LLM call Co-authored-by: Cursor --- internal/chat/session.go | 4 +++- internal/server/handlers.go | 3 ++- internal/server/server.go | 16 +++++++++++++--- internal/server/sse.go | 1 + 4 files changed, 19 insertions(+), 5 deletions(-) diff --git a/internal/chat/session.go b/internal/chat/session.go index b9085212..af294a82 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -176,7 +176,9 @@ func (s *Session) SyncRecentDialogue() { // 若发生了截断且存在早期消息，同步计算或加载滚动摘要 if truncated && len(early) > 0 && s.CtxMgr != nil { - s.syncDialogueSummary(context.Background(), early) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + s.syncDialogueSummary(ctx, early) } } diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 3ba026c6..51f123c3 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -143,7 +143,7 @@ func (s *Server) handleGetProject(w http.ResponseWriter, r *http.Request) { return } - sess, err := s.getOrCreateSession(0, sessionID, "", "", "", "", "", "", "", "") + sess, err := s.getOrCreateSession(r.Context(), 0, sessionID, "", "", "", "", "", "", "", "") if err != nil || sess == nil || sess.UserCtx == nil { writeSafeError(w, r, http.StatusNotFound, "session or project context not found") return @@ -170,6 +170,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { req.WorkDir = "" sess, err := s.getOrCreateSession( + r.Context(), user.ID, req.SessionID, req.ProjectName, diff --git a/internal/server/server.go b/internal/server/server.go index b2caff30..c98f2976 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -403,6 +403,7 @@ func (s *Server) cleanStaleSessionsLocked(ttl time.Duration) int { // getOrCreateSession 获取或动态构建会话。callerUserID>0 时校验持久层归属，不匹配返回 ErrSessionOwnerMismatch。 func (s *Server) getOrCreateSession( + ctx context.Context, callerUserID int64, sessionID string, projectName string, @@ -420,7 +421,10 @@ func (s *Server) getOrCreateSession( // 归属校验必须先于任何缓存命中/持久层灌入 if callerUserID > 0 && s.ctxMgr != nil { - if owner, err := s.ctxMgr.GetSessionOwnerID(context.Background(), sessionID); err == nil && owner > 0 && owner != callerUserID { + dbCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + owner, err := s.ctxMgr.GetSessionOwnerID(dbCtx, sessionID) + cancel() + if err == nil && owner > 0 && owner != callerUserID { return nil, ctxmgr.ErrSessionOwnerMismatch } } @@ -455,7 +459,10 @@ func (s *Server) getOrCreateSession( // 尝试从持久层恢复会话上下文 (服务重启后跨进程保持 G3) var restored *ctxmgr.RestoredSessionContext if s.ctxMgr != nil { - if r, err := s.ctxMgr.RestoreSessionContext(context.Background(), sessionID); err == nil && r != nil && len(r.Messages) > 0 { + dbCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + r, err := s.ctxMgr.RestoreSessionContext(dbCtx, sessionID) + cancel() + if err == nil && r != nil && len(r.Messages) > 0 { restored = r } } @@ -563,7 +570,10 @@ func (s *Server) getOrCreateSession( } if sess.UserID > 0 && sess.CtxMgr != nil { - if guidance, err := sess.CtxMgr.CompilePromptGuidance(context.Background(), sess.UserID); err == nil && guidance != "" { + dbCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + guidance, err := sess.CtxMgr.CompilePromptGuidance(dbCtx, sess.UserID) + cancel() + if err == nil && guidance != "" { _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ ExpectedVersion: userCtx.Version, SetPersonalizedGuidance: true, diff --git a/internal/server/sse.go b/internal/server/sse.go index ae720088..84d066c9 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -174,6 +174,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { // 会话与项目绑定必须在 SSE headers/Flush 之前完成授权（与 tasks/run/stream 对齐） sess, err := s.getOrCreateSession( + r.Context(), user.ID, req.SessionID, req.ProjectName,