fix(chat): add message scope columns and write-path resolution

Persist workflow_instance_id, stage_instance_id, scope_kind on chat
messages and task runs; resolve scope from live instance for chat and
document tasks so refresh can apply consistent stage authorization.

Co-authored-by: Cursor <cursoragent@cursor.com>

internal/ctxmgr/chat_window.go

@@ -2,7 +2,6 @@ package ctxmgr
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"sort"
@@ -118,8 +117,9 @@ func (m *ContextManager) getChatWindowPG(ctx context.Context, q ChatWindowQuery)
 		}
 
 		// 倒序取最多 MaxMsgPerDay 条
+		scoped := m.hasMessageScopeColumns(ctx)
 		msgRows, err := m.pool.Query(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+			SELECT `+m.chatMessageSelectColumns(ctx)+`
 			FROM chat_messages
 			WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL
 			ORDER BY id DESC
@@ -129,18 +129,7 @@ func (m *ContextManager) getChatWindowPG(ctx context.Context, q ChatWindowQuery)
 			return nil, fmt.Errorf("query messages for day %s failed: %w", dayStr, err)
 		}
 
-		var msgsDesc []ChatMessageRecord
-		for msgRows.Next() {
-			var r ChatMessageRecord
-			var metaBytes []byte
-			if err := msgRows.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt); err == nil {
-				if len(metaBytes) > 0 {
-					_ = json.Unmarshal(metaBytes, &r.Metadata)
-				}
-				enrichMessageExportCapability(&r)
-				msgsDesc = append(msgsDesc, r)
-			}
-		}
+		msgsDesc, _ := appendChatMessagesFromRows(msgRows, scoped)
 		msgRows.Close()
 
 		// 内存反转为正序 ASC
@@ -343,6 +332,8 @@ func (m *ContextManager) getChatDayMessagesPG(ctx context.Context, sessionID, da
 
 	var msgs []ChatMessageRecord
 	var hasMoreBefore, hasMoreAfter bool
+	scoped := m.hasMessageScopeColumns(ctx)
+	cols := m.chatMessageSelectColumns(ctx)
 
 	if anchorID > 0 {
 		// 锚点上下文：前后各取 limit/2 条
@@ -353,25 +344,14 @@ func (m *ContextManager) getChatDayMessagesPG(ctx context.Context, sessionID, da
 
 		// before or equal to anchor
 		rowsBefore, err := m.pool.Query(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+			SELECT `+cols+`
 			FROM chat_messages
 			WHERE session_id = $1 AND day_key = $2::date AND id <= $3 AND deleted_at IS NULL
 			ORDER BY id DESC
 			LIMIT $4
 		`, sessionID, dayKey, anchorID, half)
 		if err == nil {
-			var beforeDesc []ChatMessageRecord
-			for rowsBefore.Next() {
-				var r ChatMessageRecord
-				var metaBytes []byte
-				if err := rowsBefore.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt); err == nil {
-					if len(metaBytes) > 0 {
-						_ = json.Unmarshal(metaBytes, &r.Metadata)
-					}
-					enrichMessageExportCapability(&r)
-					beforeDesc = append(beforeDesc, r)
-				}
-			}
+			beforeDesc, _ := appendChatMessagesFromRows(rowsBefore, scoped)
 			rowsBefore.Close()
 			for i := len(beforeDesc) - 1; i >= 0; i-- {
 				msgs = append(msgs, beforeDesc[i])
@@ -380,47 +360,27 @@ func (m *ContextManager) getChatDayMessagesPG(ctx context.Context, sessionID, da
 
 		// after anchor
 		rowsAfter, err := m.pool.Query(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+			SELECT `+cols+`
 			FROM chat_messages
 			WHERE session_id = $1 AND day_key = $2::date AND id > $3 AND deleted_at IS NULL
 			ORDER BY id ASC
 			LIMIT $4
 		`, sessionID, dayKey, anchorID, half)
 		if err == nil {
-			for rowsAfter.Next() {
-				var r ChatMessageRecord
-				var metaBytes []byte
-				if err := rowsAfter.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt); err == nil {
-					if len(metaBytes) > 0 {
-						_ = json.Unmarshal(metaBytes, &r.Metadata)
-					}
-					enrichMessageExportCapability(&r)
-					msgs = append(msgs, r)
-				}
-			}
+			afterMsgs, _ := appendChatMessagesFromRows(rowsAfter, scoped)
 			rowsAfter.Close()
+			msgs = append(msgs, afterMsgs...)
 		}
 	} else if beforeID > 0 {
 		rows, err := m.pool.Query(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+			SELECT `+cols+`
 			FROM chat_messages
 			WHERE session_id = $1 AND day_key = $2::date AND id < $3 AND deleted_at IS NULL
 			ORDER BY id DESC
 			LIMIT $4
 		`, sessionID, dayKey, beforeID, limit)
 		if err == nil {
-			var desc []ChatMessageRecord
-			for rows.Next() {
-				var r ChatMessageRecord
-				var metaBytes []byte
-				if err := rows.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt); err == nil {
-					if len(metaBytes) > 0 {
-						_ = json.Unmarshal(metaBytes, &r.Metadata)
-					}
-					enrichMessageExportCapability(&r)
-					desc = append(desc, r)
-				}
-			}
+			desc, _ := appendChatMessagesFromRows(rows, scoped)
 			rows.Close()
 			for i := len(desc) - 1; i >= 0; i-- {
 				msgs = append(msgs, desc[i])
@@ -428,48 +388,30 @@ func (m *ContextManager) getChatDayMessagesPG(ctx context.Context, sessionID, da
 		}
 	} else if afterID > 0 {
 		rows, err := m.pool.Query(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+			SELECT `+cols+`
 			FROM chat_messages
 			WHERE session_id = $1 AND day_key = $2::date AND id > $3 AND deleted_at IS NULL
 			ORDER BY id ASC
 			LIMIT $4
 		`, sessionID, dayKey, afterID, limit)
 		if err == nil {
-			for rows.Next() {
-				var r ChatMessageRecord
-				var metaBytes []byte
-				if err := rows.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt); err == nil {
-					if len(metaBytes) > 0 {
-						_ = json.Unmarshal(metaBytes, &r.Metadata)
-					}
-					enrichMessageExportCapability(&r)
-					msgs = append(msgs, r)
-				}
-			}
+			afterMsgs, _ := appendChatMessagesFromRows(rows, scoped)
 			rows.Close()
+			msgs = append(msgs, afterMsgs...)
 		}
 	} else {
 		// 默认从当日第一条起正序 limit 条
 		rows, err := m.pool.Query(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+			SELECT `+cols+`
 			FROM chat_messages
 			WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL
 			ORDER BY id ASC
 			LIMIT $3
 		`, sessionID, dayKey, limit)
 		if err == nil {
-			for rows.Next() {
-				var r ChatMessageRecord
-				var metaBytes []byte
-				if err := rows.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt); err == nil {
-					if len(metaBytes) > 0 {
-						_ = json.Unmarshal(metaBytes, &r.Metadata)
-					}
-					enrichMessageExportCapability(&r)
-					msgs = append(msgs, r)
-				}
-			}
+			defaultMsgs, _ := appendChatMessagesFromRows(rows, scoped)
 			rows.Close()
+			msgs = append(msgs, defaultMsgs...)
 		}
 	}
 

internal/ctxmgr/manager.go

@@ -199,12 +199,32 @@ func (m *ContextManager) RecordUserMessageWithID(ctx context.Context, sessionID
 				project_id = COALESCE($3, chat_sessions.project_id)
 		`, sessionID, uid, pid)
 
+		wfInstID, stInstID, stageKey, scopeKind := scopeFieldsFromMeta(meta)
+		rec.WorkflowInstanceID = wfInstID
+		rec.StageInstanceID = stInstID
+		rec.StageKey = stageKey
+		rec.ScopeKind = scopeKind
+
 		metaJSON, _ := json.Marshal(meta)
-		err := m.pool.QueryRow(ctx, `
-			INSERT INTO chat_messages (session_id, user_id, role, content, intent, branch, metadata, day_key, category, content_kind, export_policy, content_revision, content_sha256, created_at)
-			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
-			RETURNING id
-		`, sessionID, uid, role, content, intent, branch, metaJSON, dayKey, category, contentKind, exportPolicy, contentRevision, contentSha256, now).Scan(&messageID)
+		var err error
+		if m.hasMessageScopeColumns(ctx) {
+			err = m.pool.QueryRow(ctx, `
+				INSERT INTO chat_messages (
+					session_id, user_id, role, content, intent, branch, metadata, day_key, category,
+					content_kind, export_policy, content_revision, content_sha256,
+					workflow_instance_id, stage_instance_id, stage_key, scope_kind, created_at
+				)
+				VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)
+				RETURNING id
+			`, sessionID, uid, role, content, intent, branch, metaJSON, dayKey, category, contentKind, exportPolicy, contentRevision, contentSha256,
+				nullScopeInt64(wfInstID), nullScopeInt64(stInstID), nullScopeText(stageKey), scopeKind, now).Scan(&messageID)
+		} else {
+			err = m.pool.QueryRow(ctx, `
+				INSERT INTO chat_messages (session_id, user_id, role, content, intent, branch, metadata, day_key, category, content_kind, export_policy, content_revision, content_sha256, created_at)
+				VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
+				RETURNING id
+			`, sessionID, uid, role, content, intent, branch, metaJSON, dayKey, category, contentKind, exportPolicy, contentRevision, contentSha256, now).Scan(&messageID)
+		}
 		if err != nil {
 			return 0, fmt.Errorf("insert chat message failed: %w", err)
 		}
@@ -254,20 +274,15 @@ func (m *ContextManager) GetMessageByID(ctx context.Context, messageID int64) (*
 	}
 
 	if m.pool != nil {
-		var r ChatMessageRecord
-		var metaBytes []byte
-		err := m.pool.QueryRow(ctx, `
-			SELECT id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+		scoped := m.hasMessageScopeColumns(ctx)
+		r, err := scanChatMessageRow(m.pool.QueryRow(ctx, `
+			SELECT `+m.chatMessageSelectColumns(ctx)+`
 			FROM chat_messages
 			WHERE id = $1 AND deleted_at IS NULL
-		`, messageID).Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt)
+		`, messageID), scoped)
 		if err != nil {
 			return nil, err
 		}
-		if len(metaBytes) > 0 {
-			_ = json.Unmarshal(metaBytes, &r.Metadata)
-		}
-		enrichMessageExportCapability(&r)
 		return &r, nil
 	}
 

internal/ctxmgr/message_scope_io.go

@@ -0,0 +1,161 @@
+package ctxmgr
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const chatMessageSelectColumnsLegacy = `
+	id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata,
+	COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'),
+	COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at
+`
+
+const chatMessageSelectColumnsScoped = `
+	id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata,
+	COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'),
+	COALESCE(content_revision, 1), COALESCE(content_sha256, ''),
+	COALESCE(workflow_instance_id, 0), COALESCE(stage_instance_id, 0),
+	COALESCE(stage_key, ''), COALESCE(scope_kind, 'legacy_unscoped'), created_at
+`
+
+func (m *ContextManager) hasMessageScopeColumns(ctx context.Context) bool {
+	if m == nil || m.pool == nil {
+		return false
+	}
+	var ok bool
+	_ = m.pool.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1 FROM information_schema.columns
+			WHERE table_schema = 'public' AND table_name = 'chat_messages' AND column_name = 'scope_kind'
+		)
+	`).Scan(&ok)
+	return ok
+}
+
+func (m *ContextManager) chatMessageSelectColumns(ctx context.Context) string {
+	if m.hasMessageScopeColumns(ctx) {
+		return chatMessageSelectColumnsScoped
+	}
+	return chatMessageSelectColumnsLegacy
+}
+
+func appendChatMessagesFromRows(rows pgx.Rows, scoped bool) ([]ChatMessageRecord, error) {
+	var out []ChatMessageRecord
+	for rows.Next() {
+		rec, err := scanChatMessageRow(rows, scoped)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, rec)
+	}
+	return out, rows.Err()
+}
+
+func scanChatMessageRow(row pgx.Row, scoped bool) (ChatMessageRecord, error) {
+	var r ChatMessageRecord
+	var metaBytes []byte
+	var err error
+	if scoped {
+		err = row.Scan(
+			&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes,
+			&r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256,
+			&r.WorkflowInstanceID, &r.StageInstanceID, &r.StageKey, &r.ScopeKind, &r.CreatedAt,
+		)
+	} else {
+		err = row.Scan(
+			&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes,
+			&r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256,
+			&r.CreatedAt,
+		)
+	}
+	if err != nil {
+		return r, err
+	}
+	if len(metaBytes) > 0 {
+		_ = json.Unmarshal(metaBytes, &r.Metadata)
+	}
+	if r.ScopeKind == "" {
+		_, _, _, r.ScopeKind = scopeFieldsFromMeta(r.Metadata)
+	}
+	if r.StageKey == "" {
+		if sk, ok := r.Metadata["stage_key"].(string); ok {
+			r.StageKey = sk
+		}
+	}
+	enrichMessageExportCapability(&r)
+	return r, nil
+}
+
+func nullScopeInt64(v int64) any {
+	if v <= 0 {
+		return nil
+	}
+	return v
+}
+
+func nullScopeText(v string) any {
+	if v == "" {
+		return nil
+	}
+	return v
+}
+
+func scopeFieldsFromMeta(meta map[string]any) (workflowInstanceID, stageInstanceID int64, stageKey, scopeKind string) {
+	if meta == nil {
+		return 0, 0, "", "legacy_unscoped"
+	}
+	if sk, ok := meta["scope_kind"].(string); ok && sk != "" {
+		scopeKind = sk
+	} else {
+		scopeKind = "legacy_unscoped"
+	}
+	if v, ok := meta["workflow_instance_id"].(float64); ok {
+		workflowInstanceID = int64(v)
+	} else if v, ok := meta["workflow_instance_id"].(int64); ok {
+		workflowInstanceID = v
+	}
+	if v, ok := meta["stage_instance_id"].(float64); ok {
+		stageInstanceID = int64(v)
+	} else if v, ok := meta["stage_instance_id"].(int64); ok {
+		stageInstanceID = v
+	}
+	if sk, ok := meta["stage_key"].(string); ok {
+		stageKey = sk
+	}
+	return workflowInstanceID, stageInstanceID, stageKey, scopeKind
+}
+
+// ListSessionMessagesForAuth 列出会话消息轻量字段（用于授权统计同集）。
+func (m *ContextManager) ListSessionMessagesForAuth(ctx context.Context, sessionID string) ([]ChatMessageRecord, error) {
+	if sessionID == "" {
+		return nil, nil
+	}
+	if m.pool == nil {
+		m.mu.RLock()
+		defer m.mu.RUnlock()
+		raw := m.messages[sessionID]
+		out := make([]ChatMessageRecord, 0, len(raw))
+		for _, msg := range raw {
+			if msg.DeletedAt != nil {
+				continue
+			}
+			out = append(out, msg)
+		}
+		return out, nil
+	}
+	scoped := m.hasMessageScopeColumns(ctx)
+	rows, err := m.pool.Query(ctx, `
+		SELECT `+m.chatMessageSelectColumns(ctx)+`
+		FROM chat_messages
+		WHERE session_id = $1 AND deleted_at IS NULL
+		ORDER BY id ASC
+	`, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	return appendChatMessagesFromRows(rows, scoped)
+}

internal/ctxmgr/types.go

@@ -102,23 +102,27 @@ type UserSummary struct {
 
 // ChatMessageRecord 结构化会话消息
 type ChatMessageRecord struct {
-	ID               int64          `json:"id"`
-	SessionID        string         `json:"session_id"`
-	UserID           int64          `json:"user_id"`
-	Role             string         `json:"role"`
-	Content          string         `json:"content"`
-	Intent           string         `json:"intent,omitempty"`
-	Branch           string         `json:"branch,omitempty"`
-	Metadata         map[string]any `json:"metadata,omitempty"`
-	DayKey           string         `json:"day_key"`
-	Category         string         `json:"category"`
-	ContentKind      string         `json:"content_kind,omitempty"`
-	ExportPolicy     string         `json:"export_policy,omitempty"`
-	ContentRevision  int            `json:"content_revision,omitempty"`
-	ContentSha256    string         `json:"content_sha256,omitempty"`
-	ExportCapability any            `json:"export_capability,omitempty"`
-	DeletedAt        *time.Time     `json:"deleted_at,omitempty"`
-	CreatedAt        time.Time      `json:"created_at"`
+	ID                 int64          `json:"id"`
+	SessionID          string         `json:"session_id"`
+	UserID             int64          `json:"user_id"`
+	Role               string         `json:"role"`
+	Content            string         `json:"content"`
+	Intent             string         `json:"intent,omitempty"`
+	Branch             string         `json:"branch,omitempty"`
+	Metadata           map[string]any `json:"metadata,omitempty"`
+	DayKey             string         `json:"day_key"`
+	Category           string         `json:"category"`
+	ContentKind        string         `json:"content_kind,omitempty"`
+	ExportPolicy       string         `json:"export_policy,omitempty"`
+	ContentRevision    int            `json:"content_revision,omitempty"`
+	ContentSha256      string         `json:"content_sha256,omitempty"`
+	WorkflowInstanceID int64          `json:"workflow_instance_id,omitempty"`
+	StageInstanceID    int64          `json:"stage_instance_id,omitempty"`
+	StageKey           string         `json:"stage_key,omitempty"`
+	ScopeKind          string         `json:"scope_kind,omitempty"`
+	ExportCapability   any            `json:"export_capability,omitempty"`
+	DeletedAt          *time.Time     `json:"deleted_at,omitempty"`
+	CreatedAt          time.Time      `json:"created_at"`
 }
 
 // ChatDayBucket 按日消息分片桶