merge: fix/audit-b1-data-consistency (audit batch 1: docgov persist errors, session race, workflow plan tx, ctx threading)

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
@@ -129,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)
 	}
 }
 
@@ -261,7 +310,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 +324,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 +521,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/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 {

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)
+		}
 	}
 }

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)

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
 	}

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
 		}

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 &copyInst, 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

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

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,
@@ -193,16 +194,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

@@ -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++
 		}
@@ -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
 		}
 	}
@@ -432,7 +436,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 +447,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
 	}
 
@@ -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
 		}
 	}
@@ -540,7 +547,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 +557,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,
@@ -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,
@@ -575,7 +585,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 +593,7 @@ func (s *Server) getOrCreateSession(
 		sess.UserID = callerUserID
 	}
 
-	sess.LastAccessedAt = time.Now()
+	sess.TouchLastAccessed()
 	s.sessions[sessionID] = sess
 	return sess, nil
 }
@@ -592,7 +602,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 +610,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

@@ -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,
@@ -197,11 +198,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 +221,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

@@ -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
 	}