diff --git a/deploy/postgres/goose/00026_2026-09-13-chat-message-scope.sql b/deploy/postgres/goose/00026_2026-09-13-chat-message-scope.sql new file mode 100644 index 00000000..876ddb0c --- /dev/null +++ b/deploy/postgres/goose/00026_2026-09-13-chat-message-scope.sql @@ -0,0 +1,36 @@ +-- +goose Up +-- 消息工作流作用域：支持刷新后阶段授权与历史兼容 + +ALTER TABLE chat_messages + ADD COLUMN IF NOT EXISTS workflow_instance_id BIGINT REFERENCES project_workflow_instances(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS stage_instance_id BIGINT REFERENCES project_stage_instances(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS stage_key TEXT, + ADD COLUMN IF NOT EXISTS scope_kind TEXT NOT NULL DEFAULT 'legacy_unscoped'; + +CREATE INDEX IF NOT EXISTS idx_chat_messages_session_scope + ON chat_messages (session_id, scope_kind) + WHERE deleted_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_chat_messages_stage_key + ON chat_messages (session_id, stage_key) + WHERE deleted_at IS NULL AND stage_key IS NOT NULL; + +ALTER TABLE task_runs + ADD COLUMN IF NOT EXISTS workflow_instance_id BIGINT REFERENCES project_workflow_instances(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS stage_instance_id BIGINT REFERENCES project_stage_instances(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS workflow_stage_key TEXT; + +-- +goose Down +ALTER TABLE task_runs + DROP COLUMN IF EXISTS workflow_stage_key, + DROP COLUMN IF EXISTS stage_instance_id, + DROP COLUMN IF EXISTS workflow_instance_id; + +DROP INDEX IF EXISTS idx_chat_messages_stage_key; +DROP INDEX IF EXISTS idx_chat_messages_session_scope; + +ALTER TABLE chat_messages + DROP COLUMN IF EXISTS scope_kind, + DROP COLUMN IF EXISTS stage_key, + DROP COLUMN IF EXISTS stage_instance_id, + DROP COLUMN IF EXISTS workflow_instance_id; diff --git a/deploy/postgres/goose/00027_2026-09-13-chat-message-scope-backfill.sql b/deploy/postgres/goose/00027_2026-09-13-chat-message-scope-backfill.sql new file mode 100644 index 00000000..03358bac --- /dev/null +++ b/deploy/postgres/goose/00027_2026-09-13-chat-message-scope-backfill.sql @@ -0,0 +1,79 @@ +-- +goose Up +-- 受控回填历史消息作用域（零正文删除） + +-- 1. 任务消息：通过 task_runs 唯一匹配回填 workflow_stage +UPDATE chat_messages cm +SET + scope_kind = 'workflow_stage', + workflow_instance_id = tr.workflow_instance_id, + stage_instance_id = tr.stage_instance_id, + stage_key = tr.workflow_stage_key, + metadata = cm.metadata || jsonb_build_object( + 'scope_kind', 'workflow_stage', + 'workflow_instance_id', tr.workflow_instance_id, + 'stage_instance_id', tr.stage_instance_id, + 'stage_key', tr.workflow_stage_key + ) +FROM task_runs tr +WHERE cm.scope_kind = 'legacy_unscoped' + AND cm.deleted_at IS NULL + AND cm.metadata->>'task_key' IS NOT NULL + AND cm.metadata->>'task_key' = tr.task_key + AND cm.session_id = tr.session_id + AND tr.workflow_instance_id IS NOT NULL + AND tr.stage_instance_id IS NOT NULL + AND tr.workflow_stage_key IS NOT NULL; + +-- 2. 激活前 owner 私聊：项目有 live 实例且消息创建于激活之前 +UPDATE chat_messages cm +SET + scope_kind = 'pre_activation', + workflow_instance_id = pwi.id, + metadata = cm.metadata || jsonb_build_object( + 'scope_kind', 'pre_activation', + 'workflow_instance_id', pwi.id + ) +FROM chat_sessions cs +JOIN projects p ON p.id = cs.project_id +JOIN project_workflow_instances pwi ON pwi.project_id = p.id AND pwi.status = 'active' +WHERE cm.session_id = cs.session_id + AND cm.scope_kind = 'legacy_unscoped' + AND cm.deleted_at IS NULL + AND cm.metadata->>'task_key' IS NULL + AND pwi.activated_at IS NOT NULL + AND cm.created_at < pwi.activated_at + AND cs.user_id IS NOT NULL + AND cm.user_id = cs.user_id; + +-- 3. 已有 metadata.stage_key 但缺 scope_kind 的普通聊天（live 实例存在） +UPDATE chat_messages +SET + scope_kind = 'workflow_stage', + stage_key = chat_messages.metadata->>'stage_key', + workflow_instance_id = src.workflow_instance_id, + stage_instance_id = src.stage_instance_id, + metadata = chat_messages.metadata || jsonb_build_object( + 'scope_kind', 'workflow_stage', + 'workflow_instance_id', src.workflow_instance_id, + 'stage_instance_id', src.stage_instance_id + ) +FROM ( + SELECT cm.id AS message_id, pwi.id AS workflow_instance_id, psi.id AS stage_instance_id + FROM chat_messages cm + INNER JOIN chat_sessions cs ON cm.session_id = cs.session_id + INNER JOIN project_workflow_instances pwi ON pwi.project_id = cs.project_id AND pwi.status = 'active' + INNER JOIN project_stage_instances psi ON psi.workflow_instance_id = pwi.id + AND psi.stage_key = cm.metadata->>'stage_key' + WHERE cm.scope_kind = 'legacy_unscoped' + AND cm.deleted_at IS NULL + AND cm.metadata->>'stage_key' IS NOT NULL + AND cm.metadata->>'stage_key' <> '' + AND cm.metadata->>'task_key' IS NULL +) AS src +WHERE chat_messages.id = src.message_id; + +-- +goose Down +-- 回填不可逆；Down 仅重置为 legacy_unscoped（不含正文删除） +UPDATE chat_messages +SET scope_kind = 'legacy_unscoped' +WHERE scope_kind IN ('workflow_stage', 'pre_activation'); diff --git a/internal/chat/session.go b/internal/chat/session.go index f3e58a4f..aa854f1b 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -10,6 +10,7 @@ import ( "chat-agent-system/internal/coordinator" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" + "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/replyexport" "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" @@ -44,6 +45,7 @@ type Session struct { UserID int64 `json:"user_id"` ProjectID int64 `json:"project_id"` ActiveStageKey string `json:"active_stage_key,omitempty"` + MessageScope *projectinstance.MessageScope `json:"-"` UserCtx *coordinator.UserProjectContext `json:"user_ctx"` Recognizer intent.Recognizer `json:"-"` Coordinator *coordinator.CoordinatorEngine `json:"-"` @@ -269,9 +271,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str if s.ProjectID > 0 { userMeta["project_id"] = s.ProjectID } - if s.ActiveStageKey != "" { - userMeta["stage_key"] = s.ActiveStageKey - } + s.applyMessageScopeMeta(userMeta) if clientTurnID != "" { userMeta["client_turn_id"] = clientTurnID userMeta["client_id"] = clientTurnID + ":user" @@ -433,9 +433,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str if s.ProjectID > 0 { asstMeta["project_id"] = s.ProjectID } - if s.ActiveStageKey != "" { - asstMeta["stage_key"] = s.ActiveStageKey - } + s.applyMessageScopeMeta(asstMeta) if recRes.TopIntent != nil { asstMeta["turn_intent"] = string(recRes.TopIntent.Intent) } @@ -531,3 +529,16 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str PendingExpectation: s.UserCtx.PendingExpectation, }, nil } + +func (s *Session) applyMessageScopeMeta(meta map[string]any) { + if meta == nil { + return + } + if s.MessageScope != nil { + projectinstance.ApplyScopeToMetadata(meta, s.MessageScope) + return + } + if s.ActiveStageKey != "" { + meta["stage_key"] = s.ActiveStageKey + } +} diff --git a/internal/ctxmgr/chat_window.go b/internal/ctxmgr/chat_window.go index e1d420bd..22cf0420 100644 --- a/internal/ctxmgr/chat_window.go +++ b/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...) } } diff --git a/internal/ctxmgr/manager.go b/internal/ctxmgr/manager.go index 49065a38..ae72a5cf 100644 --- a/internal/ctxmgr/manager.go +++ b/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 } diff --git a/internal/ctxmgr/message_scope_io.go b/internal/ctxmgr/message_scope_io.go new file mode 100644 index 00000000..568d1624 --- /dev/null +++ b/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) +} diff --git a/internal/ctxmgr/types.go b/internal/ctxmgr/types.go index b9ec64bc..8216cd2f 100644 --- a/internal/ctxmgr/types.go +++ b/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 按日消息分片桶 diff --git a/internal/projectinstance/message_scope.go b/internal/projectinstance/message_scope.go new file mode 100644 index 00000000..56d29ed8 --- /dev/null +++ b/internal/projectinstance/message_scope.go @@ -0,0 +1,209 @@ +package projectinstance + +import ( + "context" + "errors" + "fmt" +) + +const ( + ScopeWorkflowStage = "workflow_stage" + ScopePreActivation = "pre_activation" + ScopeLegacyUnscoped = "legacy_unscoped" + ScopeNonProjectPrivate = "non_project_private" +) + +var ErrTaskScopeNotResolved = errors.New("无法唯一解析文档任务的工作流阶段作用域") + +// MessageScope 不可变消息工作流作用域（服务端解析，客户端不可伪造）。 +type MessageScope struct { + ScopeKind string `json:"scope_kind"` + WorkflowInstanceID int64 `json:"workflow_instance_id,omitempty"` + StageInstanceID int64 `json:"stage_instance_id,omitempty"` + StageKey string `json:"stage_key,omitempty"` +} + +// ApplyScopeToMetadata 将作用域写入 metadata（冗余，便于审计与兼容）。 +func ApplyScopeToMetadata(meta map[string]any, scope *MessageScope) { + if meta == nil || scope == nil { + return + } + meta["scope_kind"] = scope.ScopeKind + if scope.WorkflowInstanceID > 0 { + meta["workflow_instance_id"] = scope.WorkflowInstanceID + } + if scope.StageInstanceID > 0 { + meta["stage_instance_id"] = scope.StageInstanceID + } + if scope.StageKey != "" { + meta["stage_key"] = scope.StageKey + } +} + +// ScopeFromMetadata 从 metadata 提取作用域（历史兼容）。 +func ScopeFromMetadata(metadata map[string]any) MessageScope { + out := MessageScope{ScopeKind: ScopeLegacyUnscoped} + if metadata == nil { + return out + } + if sk, ok := metadata["scope_kind"].(string); ok && sk != "" { + out.ScopeKind = sk + } + if v, ok := metadata["workflow_instance_id"].(float64); ok && v > 0 { + out.WorkflowInstanceID = int64(v) + } else if v, ok := metadata["workflow_instance_id"].(int64); ok && v > 0 { + out.WorkflowInstanceID = v + } + if v, ok := metadata["stage_instance_id"].(float64); ok && v > 0 { + out.StageInstanceID = int64(v) + } else if v, ok := metadata["stage_instance_id"].(int64); ok && v > 0 { + out.StageInstanceID = v + } + out.StageKey = MessageStageKey(metadata) + return out +} + +// ResolveMessageScopeForChat 解析普通聊天消息作用域。 +func (s *Service) ResolveMessageScopeForChat(ctx context.Context, projectID int64, activeStageKey string) (*MessageScope, error) { + if projectID <= 0 { + return &MessageScope{ScopeKind: ScopeNonProjectPrivate}, nil + } + if s == nil || s.store == nil { + return &MessageScope{ScopeKind: ScopePreActivation}, nil + } + live, err := s.store.GetLiveInstance(ctx, projectID) + if err != nil || live == nil { + return &MessageScope{ScopeKind: ScopePreActivation}, nil + } + detail, err := s.store.GetInstanceByID(ctx, live.ID) + if err != nil || detail == nil { + return &MessageScope{ScopeKind: ScopePreActivation}, nil + } + stageKey := activeStageKey + if stageKey == "" { + stageKey = detail.ActiveStageKey + } + if stageKey == "" { + return &MessageScope{ + ScopeKind: ScopePreActivation, + WorkflowInstanceID: live.ID, + }, nil + } + for _, st := range detail.Stages { + if st.StageKey == stageKey { + return &MessageScope{ + ScopeKind: ScopeWorkflowStage, + WorkflowInstanceID: live.ID, + StageInstanceID: st.ID, + StageKey: st.StageKey, + }, nil + } + } + return &MessageScope{ + ScopeKind: ScopeWorkflowStage, + WorkflowInstanceID: live.ID, + StageKey: stageKey, + }, nil +} + +// ResolveMessageScopeForTask 解析文档任务消息作用域；无法唯一归属时拒绝。 +func (s *Service) ResolveMessageScopeForTask(ctx context.Context, projectID int64, taskKey string) (*MessageScope, error) { + if projectID <= 0 || taskKey == "" { + return nil, fmt.Errorf("%w: project_id=%d task_key=%q", ErrTaskScopeNotResolved, projectID, taskKey) + } + if s == nil || s.store == nil { + return nil, ErrTaskScopeNotResolved + } + live, err := s.store.GetLiveInstance(ctx, projectID) + if err != nil || live == nil { + return nil, fmt.Errorf("%w: no live instance for project %d", ErrTaskScopeNotResolved, projectID) + } + detail, err := s.store.GetInstanceByID(ctx, live.ID) + if err != nil || detail == nil { + return nil, ErrTaskScopeNotResolved + } + var matches []MessageScope + for _, st := range detail.Stages { + for _, dt := range st.DocTasks { + if dt.TaskKey == taskKey { + matches = append(matches, MessageScope{ + ScopeKind: ScopeWorkflowStage, + WorkflowInstanceID: live.ID, + StageInstanceID: st.ID, + StageKey: st.StageKey, + }) + } + } + } + if len(matches) == 0 { + stageKey := TaskKeyToStage(live.ConfigSnapshot, taskKey) + if stageKey == "" { + return nil, fmt.Errorf("%w: task_key %q not in instance snapshot", ErrTaskScopeNotResolved, taskKey) + } + for _, st := range detail.Stages { + if st.StageKey == stageKey { + return &MessageScope{ + ScopeKind: ScopeWorkflowStage, + WorkflowInstanceID: live.ID, + StageInstanceID: st.ID, + StageKey: st.StageKey, + }, nil + } + } + return nil, ErrTaskScopeNotResolved + } + if len(matches) > 1 { + first := matches[0] + for _, m := range matches[1:] { + if m.StageKey != first.StageKey { + return nil, fmt.Errorf("%w: ambiguous task_key %q", ErrTaskScopeNotResolved, taskKey) + } + } + } + scope := matches[0] + return &scope, nil +} + +// MessageVisible 判断消息对 viewer 是否可见（需会话 owner 与阶段授权上下文）。 +func MessageVisible(scope MessageScope, access *StageAccess, sessionOwnerID, viewerUserID int64) bool { + if viewerUserID <= 0 { + return false + } + switch scope.ScopeKind { + case ScopeNonProjectPrivate, ScopePreActivation: + return sessionOwnerID > 0 && viewerUserID == sessionOwnerID + case ScopeLegacyUnscoped: + if access == nil || !access.Restricted { + return sessionOwnerID <= 0 || viewerUserID == sessionOwnerID + } + return sessionOwnerID > 0 && viewerUserID == sessionOwnerID + case ScopeWorkflowStage, "": + if access == nil || !access.Restricted { + return true + } + stageKey := scope.StageKey + if stageKey == "" { + return sessionOwnerID > 0 && viewerUserID == sessionOwnerID + } + return access.StageKeyAllowed(stageKey) + default: + return false + } +} + +// MessageRecordVisible 从 ChatMessageRecord 字段判定可见性。 +func MessageRecordVisible( + scopeKind, stageKey string, + metadata map[string]any, + access *StageAccess, + sessionOwnerID, viewerUserID int64, +) bool { + scope := MessageScope{ScopeKind: scopeKind, StageKey: stageKey} + if scope.ScopeKind == "" { + scope = ScopeFromMetadata(metadata) + } + if scope.StageKey == "" { + scope.StageKey = MessageStageKey(metadata) + } + return MessageVisible(scope, access, sessionOwnerID, viewerUserID) +} diff --git a/internal/projectinstance/message_scope_test.go b/internal/projectinstance/message_scope_test.go new file mode 100644 index 00000000..1d325b53 --- /dev/null +++ b/internal/projectinstance/message_scope_test.go @@ -0,0 +1,54 @@ +package projectinstance + +import "testing" + +func TestMessageVisible_PreActivationOwnerOnly(t *testing.T) { + access := &StageAccess{Restricted: true, AllowedStageKeys: map[string]bool{"V0.1_triage": true}} + scope := MessageScope{ScopeKind: ScopePreActivation} + + if !MessageVisible(scope, access, 9, 9) { + t.Fatal("owner should see pre_activation messages") + } + if MessageVisible(scope, access, 9, 99) { + t.Fatal("non-owner must not see pre_activation messages") + } +} + +func TestMessageVisible_WorkflowStageRestricted(t *testing.T) { + access := &StageAccess{ + Restricted: true, + AllowedStageKeys: map[string]bool{"V0.1_triage": true}, + } + scope := MessageScope{ScopeKind: ScopeWorkflowStage, StageKey: "V0.1_triage"} + if !MessageVisible(scope, access, 9, 9) { + t.Fatal("allowed stage should be visible") + } + scope.StageKey = "V0.5_feasibility" + if MessageVisible(scope, access, 9, 9) { + t.Fatal("disallowed stage must be hidden") + } +} + +func TestMessageVisible_LegacyUnscopedOwnerWhenRestricted(t *testing.T) { + access := &StageAccess{Restricted: true} + scope := MessageScope{ScopeKind: ScopeLegacyUnscoped} + if !MessageVisible(scope, access, 9, 9) { + t.Fatal("owner should see legacy_unscoped when restricted") + } + if MessageVisible(scope, access, 9, 99) { + t.Fatal("non-owner must not see legacy_unscoped when restricted") + } +} + +func TestScopeFromMetadata_RoundTrip(t *testing.T) { + meta := map[string]any{ + "scope_kind": ScopeWorkflowStage, + "workflow_instance_id": float64(89), + "stage_instance_id": float64(201), + "stage_key": "V0.1_triage", + } + scope := ScopeFromMetadata(meta) + if scope.ScopeKind != ScopeWorkflowStage || scope.WorkflowInstanceID != 89 || scope.StageInstanceID != 201 || scope.StageKey != "V0.1_triage" { + t.Fatalf("unexpected scope: %+v", scope) + } +} diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index eb20ef92..21be8f7c 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -86,13 +86,15 @@ func (s *PGStore) InsertRun(ctx context.Context, rec *RunRecord) error { run_id, session_id, project_id, user_id, client_turn_id, task_key, stage, status, workspace_id, assistant_message_id, error_message, worker_id, lease_expires_at, lease_heartbeat_at, attempt, + workflow_instance_id, stage_instance_id, workflow_stage_key, created_at, updated_at - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20) ON CONFLICT (run_id) DO NOTHING `, rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID, rec.TaskKey, rec.Stage, rec.Status, rec.WorkspaceID, nullInt64(rec.AssistantMsgID), rec.ErrorMessage, rec.WorkerID, rec.LeaseExpiresAt, leaseHeartbeat, attempt, + nullInt64(rec.WorkflowInstanceID), nullInt64(rec.StageInstanceID), nullScopeText(rec.WorkflowStageKey), rec.CreatedAt, rec.UpdatedAt, ) return err @@ -101,13 +103,15 @@ func (s *PGStore) InsertRun(ctx context.Context, rec *RunRecord) error { INSERT INTO task_runs ( run_id, session_id, project_id, user_id, client_turn_id, task_key, stage, status, workspace_id, assistant_message_id, - error_message, created_at, updated_at - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) + error_message, workflow_instance_id, stage_instance_id, workflow_stage_key, + created_at, updated_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (run_id) DO NOTHING `, rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID, rec.TaskKey, rec.Stage, rec.Status, rec.WorkspaceID, nullInt64(rec.AssistantMsgID), - rec.ErrorMessage, rec.CreatedAt, rec.UpdatedAt, + rec.ErrorMessage, nullInt64(rec.WorkflowInstanceID), nullInt64(rec.StageInstanceID), nullScopeText(rec.WorkflowStageKey), + rec.CreatedAt, rec.UpdatedAt, ) return err } @@ -373,6 +377,13 @@ func nullInt64(v int64) *int64 { return &v } +func nullScopeText(v string) *string { + if v == "" { + return nil + } + return &v +} + // TablesReady 检查 task_runs 表是否已迁移。 func TablesReady(ctx context.Context, pool *pgxpool.Pool) bool { if pool == nil { diff --git a/internal/taskrun/types.go b/internal/taskrun/types.go index 74d0acbd..6546cb2a 100644 --- a/internal/taskrun/types.go +++ b/internal/taskrun/types.go @@ -25,22 +25,25 @@ type CreateRequest struct { // RunRecord 运行实例状态。 type RunRecord struct { - RunID string - SessionID string - ProjectID int64 - UserID int64 - ClientTurnID string - TaskKey string - Stage string - Status string // pending | running | completed | failed | cancelled | status_conflict - WorkspaceID string - AssistantMsgID int64 - WorkerID string - LeaseExpiresAt *time.Time - CreatedAt time.Time - UpdatedAt time.Time - ErrorMessage string - Attempt int + RunID string + SessionID string + ProjectID int64 + UserID int64 + ClientTurnID string + TaskKey string + Stage string + WorkflowInstanceID int64 + StageInstanceID int64 + WorkflowStageKey string + Status string // pending | running | completed | failed | cancelled | status_conflict + WorkspaceID string + AssistantMsgID int64 + WorkerID string + LeaseExpiresAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time + ErrorMessage string + Attempt int } // StoredEvent 持久化 SSE 事件（支持重连重放）。