From 88061f88d42453b1d42faf19c15c563e0c07b222 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Wed, 5 Aug 2026 18:10:44 +0800 Subject: [PATCH] fix: isolate historical evidence recall by message branch --- .../conversation/context_artifact.go | 32 +- .../conversation/context_artifact_test.go | 98 +++-- .../application/conversation/prompt_scope.go | 54 +-- .../conversation/prompt_scope_test.go | 32 +- .../service_generation_support.go | 12 +- .../service_message_completion.go | 5 +- .../conversation/service_message_send.go | 38 +- .../domain/conversation/context_artifact.go | 1 + .../internal/infra/persistence/models/chat.go | 2 +- .../postgres/conversation/repository.go | 63 ++- .../repository_context_artifact.go | 50 ++- .../repository_historical_scope.go | 58 +++ .../repository_postgres_integration_test.go | 130 +++++++ .../repository_sqlite_vector_test.go | 59 ++- .../postgres/conversation/repository_test.go | 367 ++++++++++++++++++ .../infra/persistence/schema/schema.go | 51 +++ .../infra/persistence/schema/schema_test.go | 164 ++++++++ .../conversation_context_artifact.go | 9 +- .../internal/repository/conversation_core.go | 24 +- 19 files changed, 1056 insertions(+), 193 deletions(-) create mode 100644 backend/internal/infra/persistence/postgres/conversation/repository_historical_scope.go create mode 100644 backend/internal/infra/persistence/postgres/conversation/repository_postgres_integration_test.go diff --git a/backend/internal/application/conversation/context_artifact.go b/backend/internal/application/conversation/context_artifact.go index e319d24f6..c6efda984 100644 --- a/backend/internal/application/conversation/context_artifact.go +++ b/backend/internal/application/conversation/context_artifact.go @@ -55,8 +55,6 @@ type snapshotContextArtifactInput struct { type historicalContextArtifactInput struct { CurrentMessageID uint HasCurrentSnapshot bool - CoveredUntilID uint - AllowedMessageIDs map[uint]struct{} Query string Candidates []domainconversation.ContextArtifact CurrentRAGChunks []domainconversation.RAGChunk @@ -167,17 +165,14 @@ func (s *Service) applyContextArtifactRetention(items []domainconversation.Conte // recallHistoricalContextArtifacts 读取近期上下文证据并按当前问题筛选。 func (s *Service) recallHistoricalContextArtifacts( ctx context.Context, - conversationID uint, - currentMessageID uint, + scope repository.HistoricalMessageScope, hasCurrentSnapshot bool, - coveredUntilID uint, - allowedMessageIDs map[uint]struct{}, query string, currentRAGChunks []domainconversation.RAGChunk, currentFallbacks []AttachmentInput, currentRecall []domainconversation.MessageChunk, ) []domainconversation.ContextArtifact { - if strings.TrimSpace(query) == "" { + if !scope.Valid() || strings.TrimSpace(query) == "" { return nil } kinds := []domainconversation.ContextArtifactKind{ @@ -190,22 +185,24 @@ func (s *Service) recallHistoricalContextArtifacts( if !hasCurrentSnapshot { kinds = append(kinds, domainconversation.ContextArtifactSummary) } - candidates, err := s.repo.ListRecentContextArtifacts(ctx, conversationID, kinds, historicalArtifactScanLimit) + candidates, err := s.repo.ListRecentContextArtifacts(ctx, repository.ContextArtifactListFilter{ + Scope: scope, + Kinds: kinds, + Limit: historicalArtifactScanLimit, + }) if err != nil { if s.logger != nil { s.logger.Warn("historical_context_artifact_recall_failed", zap.String("trace_id", traceid.FromContext(ctx)), - zap.Uint("conversation_id", conversationID), + zap.Uint("conversation_id", scope.ConversationID), zap.Error(err), ) } return nil } return selectHistoricalContextArtifacts(historicalContextArtifactInput{ - CurrentMessageID: currentMessageID, + CurrentMessageID: scope.LeafMessageID, HasCurrentSnapshot: hasCurrentSnapshot, - CoveredUntilID: coveredUntilID, - AllowedMessageIDs: allowedMessageIDs, Query: query, Candidates: candidates, CurrentRAGChunks: currentRAGChunks, @@ -441,17 +438,6 @@ func selectHistoricalContextArtifacts(input historicalContextArtifactInput) []do if input.HasCurrentSnapshot && item.Kind == domainconversation.ContextArtifactSummary { continue } - if input.CoveredUntilID > 0 && item.MessageID > 0 && item.MessageID <= input.CoveredUntilID { - continue - } - if len(input.AllowedMessageIDs) > 0 { - if item.MessageID == 0 { - continue - } - if _, ok := input.AllowedMessageIDs[item.MessageID]; !ok { - continue - } - } content := strings.TrimSpace(item.Content) if content == "" || item.MessageID == input.CurrentMessageID { continue diff --git a/backend/internal/application/conversation/context_artifact_test.go b/backend/internal/application/conversation/context_artifact_test.go index 480dc1beb..7328e1c09 100644 --- a/backend/internal/application/conversation/context_artifact_test.go +++ b/backend/internal/application/conversation/context_artifact_test.go @@ -1,6 +1,7 @@ package conversation import ( + "context" "encoding/json" "strings" "testing" @@ -8,8 +9,25 @@ import ( model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation" domainmemory "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/memory" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/config" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" ) +type toolArtifactCaptureRepository struct { + repository.ConversationRepository + toolCalls []model.ToolCall + artifacts []model.ContextArtifact +} + +func (r *toolArtifactCaptureRepository) CreateConversationToolCalls(_ context.Context, items []model.ToolCall) error { + r.toolCalls = append(r.toolCalls, items...) + return nil +} + +func (r *toolArtifactCaptureRepository) CreateContextArtifacts(_ context.Context, items []model.ContextArtifact) error { + r.artifacts = append(r.artifacts, items...) + return nil +} + func TestBuildPromptContextArtifactsRecordsRAGFallbackAndRecall(t *testing.T) { items := buildPromptContextArtifacts(promptContextArtifactInput{ ConversationID: 7, @@ -79,6 +97,33 @@ func TestBuildPromptContextArtifactsRecordsRAGFallbackAndRecall(t *testing.T) { } } +func TestPersistMessageToolCallsAnchorsEvidenceToAssistantMessage(t *testing.T) { + repo := &toolArtifactCaptureRepository{} + service := &Service{repo: repo} + + err := service.persistMessageToolCalls(context.Background(), persistMessageToolCallsInput{ + SendInput: SendMessageInput{ConversationID: 7, UserID: 11}, + AssistantMessageID: 22, + RunID: "run_tool", + Rows: []model.ToolCall{{ + ToolCallID: "call_1", + ToolType: "mcp", + ToolName: "lookup", + Status: "completed", + OutputJSON: `{"result":"ok"}`, + }}, + }) + if err != nil { + t.Fatalf("persistMessageToolCalls() error = %v", err) + } + if len(repo.toolCalls) != 1 || repo.toolCalls[0].MessageID != 22 { + t.Fatalf("expected tool call on assistant message, got %#v", repo.toolCalls) + } + if len(repo.artifacts) != 1 || repo.artifacts[0].MessageID != 22 { + t.Fatalf("expected tool evidence on assistant message, got %#v", repo.artifacts) + } +} + func hasContextArtifact(items []model.ContextArtifact, kind model.ContextArtifactKind, sourceID string) bool { for _, item := range items { if item.Kind == kind && item.SourceID == sourceID { @@ -301,59 +346,6 @@ func TestSelectHistoricalContextArtifactsSkipsSummaryWhenCurrentSnapshotExists(t } } -func TestSelectHistoricalContextArtifactsRespectsSnapshotScope(t *testing.T) { - items := selectHistoricalContextArtifacts(historicalContextArtifactInput{ - CurrentMessageID: 9, - HasCurrentSnapshot: true, - CoveredUntilID: 4, - AllowedMessageIDs: map[uint]struct{}{ - 6: {}, - }, - Query: "继续部署测试", - Candidates: []model.ContextArtifact{ - { - MessageID: 3, - Kind: model.ContextArtifactToolResult, - SourceTitle: "covered", - Content: "已被摘要覆盖的部署测试结果", - TokenEstimate: 10, - Score: 1, - }, - { - MessageID: 6, - Kind: model.ContextArtifactToolResult, - SourceTitle: "retained", - Content: "保留窗口内的部署测试结果", - TokenEstimate: 10, - Score: 1, - }, - { - MessageID: 8, - Kind: model.ContextArtifactToolResult, - SourceTitle: "sibling", - Content: "其他分支的部署测试结果", - TokenEstimate: 10, - Score: 1, - }, - { - MessageID: 0, - Kind: model.ContextArtifactToolResult, - SourceTitle: "unanchored", - Content: "没有消息锚点的部署测试结果", - TokenEstimate: 10, - Score: 1, - }, - }, - }) - - if len(items) != 1 { - t.Fatalf("expected one retained-scope artifact, got %#v", items) - } - if items[0].SourceTitle != "retained" { - t.Fatalf("expected retained artifact, got %#v", items[0]) - } -} - func TestSelectHistoricalContextArtifactsRequiresRelevanceWithoutFollowUp(t *testing.T) { items := selectHistoricalContextArtifacts(historicalContextArtifactInput{ Query: "部署 测试", diff --git a/backend/internal/application/conversation/prompt_scope.go b/backend/internal/application/conversation/prompt_scope.go index 02e34b81b..1583afc65 100644 --- a/backend/internal/application/conversation/prompt_scope.go +++ b/backend/internal/application/conversation/prompt_scope.go @@ -4,6 +4,7 @@ import ( appcompact "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/application/compact" model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/llm" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" ) type promptScope struct { @@ -12,7 +13,6 @@ type promptScope struct { RetainedMessages []model.Message Snapshot *model.ContextSnapshot CoveredUntilID uint - retainedMessageIDs map[uint]struct{} } func buildPromptScope(messages []model.Message, snapshot *model.ContextSnapshot, policy contextCompactionPolicy) promptScope { @@ -34,7 +34,6 @@ func buildPromptScope(messages []model.Message, snapshot *model.ContextSnapshot, scope.CoveredMessages = append([]model.Message(nil), messages[:boundaryIndex+1]...) scope.RetainedMessages = append([]model.Message(nil), messages[boundaryIndex+1:]...) scope.CoveredUntilID = snapshot.CoveredUntilMessageID - scope.retainedMessageIDs = messageIDSet(scope.RetainedMessages) return scope } @@ -45,48 +44,25 @@ func (s promptScope) activeMessages() []model.Message { return s.FullBranchMessages } -func (s promptScope) filterRecallChunks(chunks []model.MessageChunk) []model.MessageChunk { - if len(chunks) == 0 || s.CoveredUntilID == 0 { - return chunks +func (s promptScope) historicalMessageScope(conversationID uint, userID uint, currentMessageID uint) repository.HistoricalMessageScope { + if conversationID == 0 || userID == 0 || currentMessageID == 0 { + return repository.HistoricalMessageScope{} } - result := make([]model.MessageChunk, 0, len(chunks)) - for _, chunk := range chunks { - if chunk.MessageID > 0 && chunk.MessageID <= s.CoveredUntilID { - continue - } - if len(s.retainedMessageIDs) > 0 && chunk.MessageID > 0 { - if _, ok := s.retainedMessageIDs[chunk.MessageID]; !ok { - continue - } - } - result = append(result, chunk) - } - return result -} - -func (s promptScope) retainedMessageIDSet() map[uint]struct{} { - if len(s.retainedMessageIDs) == 0 { - return nil + messages := s.FullBranchMessages + if s.Snapshot != nil { + messages = s.RetainedMessages } - result := make(map[uint]struct{}, len(s.retainedMessageIDs)) - for id := range s.retainedMessageIDs { - result[id] = struct{}{} - } - return result -} - -func messageIDSet(messages []model.Message) map[uint]struct{} { - if len(messages) == 0 { - return nil - } - result := make(map[uint]struct{}, len(messages)) for _, message := range messages { - if message.ID == 0 { - continue + if message.ID > 0 && message.ID != currentMessageID { + return repository.HistoricalMessageScope{ + ConversationID: conversationID, + UserID: userID, + LeafMessageID: currentMessageID, + ExcludeThroughMessageID: s.CoveredUntilID, + } } - result[message.ID] = struct{}{} } - return result + return repository.HistoricalMessageScope{} } type historyMessageOptions struct { diff --git a/backend/internal/application/conversation/prompt_scope_test.go b/backend/internal/application/conversation/prompt_scope_test.go index 0ddde15f1..603a0cc61 100644 --- a/backend/internal/application/conversation/prompt_scope_test.go +++ b/backend/internal/application/conversation/prompt_scope_test.go @@ -55,22 +55,34 @@ func TestBuildPromptScopeReplacesCoveredPrefix(t *testing.T) { } } -func TestPromptScopeFilterRecallChunksDropsCoveredMessages(t *testing.T) { +func TestPromptScopeHistoricalMessageScopeUsesSnapshotBoundary(t *testing.T) { messages := promptScopeMessages() scope := buildPromptScope(messages, promptScopeSnapshot(messages[:2]), contextCompactionPolicy{AdminEnabled: true, UserEnabled: true}) - chunks := []model.MessageChunk{ - {MessageID: 1, Content: "covered"}, - {MessageID: 3, Content: "retained"}, - {MessageID: 99, Content: "sibling branch"}, + historicalScope := scope.historicalMessageScope(7, 11, 4) + if !historicalScope.Valid() { + t.Fatal("expected valid historical scope") } - filtered := scope.filterRecallChunks(chunks) + if historicalScope.ConversationID != 7 || historicalScope.UserID != 11 || historicalScope.LeafMessageID != 4 || historicalScope.ExcludeThroughMessageID != 2 { + t.Fatalf("unexpected historical scope: %#v", historicalScope) + } +} + +func TestPromptScopeHistoricalMessageScopeUsesFullBranchWithoutSnapshot(t *testing.T) { + messages := promptScopeMessages() + scope := buildPromptScope(messages, nil, contextCompactionPolicy{}) - if len(filtered) != 1 { - t.Fatalf("expected one retained recall chunk, got %d", len(filtered)) + historicalScope := scope.historicalMessageScope(7, 11, 4) + if !historicalScope.Valid() || historicalScope.ExcludeThroughMessageID != 0 { + t.Fatalf("unexpected historical scope: %#v", historicalScope) } - if filtered[0].MessageID != 3 { - t.Fatalf("expected retained chunk from message 3, got %d", filtered[0].MessageID) +} + +func TestPromptScopeHistoricalMessageScopeFailsClosedOnFirstTurn(t *testing.T) { + scope := buildPromptScope([]model.Message{{ID: 9, Role: "user"}}, nil, contextCompactionPolicy{}) + + if historicalScope := scope.historicalMessageScope(7, 11, 9); historicalScope.Valid() { + t.Fatalf("expected no historical scope, got %#v", historicalScope) } } diff --git a/backend/internal/application/conversation/service_generation_support.go b/backend/internal/application/conversation/service_generation_support.go index d9b403ae0..20abac90f 100644 --- a/backend/internal/application/conversation/service_generation_support.go +++ b/backend/internal/application/conversation/service_generation_support.go @@ -12,6 +12,7 @@ import ( domainbilling "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/billing" model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/llm" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" "github.com/google/uuid" "go.uber.org/zap" ) @@ -86,15 +87,20 @@ func reasoningPayload(delta *llm.ReasoningDelta) map[string]interface{} { } // recallSemanticContext 语义召回历史消息；无结果时返回空列表。 -func (s *Service) recallSemanticContext(ctx context.Context, conversationID uint, userID uint, query string) []model.MessageChunk { - if s.embeddingSvc == nil || strings.TrimSpace(query) == "" { +func (s *Service) recallSemanticContext(ctx context.Context, scope repository.HistoricalMessageScope, query string) []model.MessageChunk { + if s.embeddingSvc == nil || !scope.Valid() || strings.TrimSpace(query) == "" { return nil } embeddings, err := s.embeddingSvc.EmbedTexts(ctx, []string{query}) if err != nil || len(embeddings) == 0 { return nil } - chunks, err := s.repo.SearchMessageChunks(ctx, conversationID, userID, embeddings[0], 5, 0.75) + chunks, err := s.repo.SearchMessageChunks(ctx, repository.MessageChunkSearchInput{ + Scope: scope, + QueryEmbedding: embeddings[0], + TopK: 5, + MinSimilarity: 0.75, + }) if err != nil || len(chunks) == 0 { return nil } diff --git a/backend/internal/application/conversation/service_message_completion.go b/backend/internal/application/conversation/service_message_completion.go index 8fe0ac18a..2b3e2637a 100644 --- a/backend/internal/application/conversation/service_message_completion.go +++ b/backend/internal/application/conversation/service_message_completion.go @@ -78,7 +78,6 @@ const ( type persistMessageToolCallsInput struct { SendInput SendMessageInput - UserMessageID uint AssistantMessageID uint RunID string Rows []model.ToolCall @@ -278,7 +277,6 @@ func successfulMessageGenerationModelName(input persistMessageGenerationInput) s func (s *Service) finishSuccessfulMessageGeneration(ctx context.Context, input persistMessageGenerationInput) error { if err := s.persistMessageToolCalls(ctx, persistMessageToolCallsInput{ SendInput: input.SendInput, - UserMessageID: input.UserMessage.ID, AssistantMessageID: input.AssistantMessage.ID, RunID: input.AssistantMessage.RunID, Rows: input.ToolCallRows, @@ -371,7 +369,6 @@ func (s *Service) persistInterruptedMessageGeneration(ctx context.Context, input if err := s.persistMessageToolCalls(persistCtx, persistMessageToolCallsInput{ SendInput: input.SendInput, - UserMessageID: input.UserMessage.ID, AssistantMessageID: input.AssistantMessage.ID, RunID: input.AssistantMessage.RunID, Rows: input.ToolCallRows, @@ -582,7 +579,7 @@ func (s *Service) persistMessageToolCalls(ctx context.Context, input persistMess s.persistToolContextArtifacts(ctx, toolContextArtifactInput{ ConversationID: input.SendInput.ConversationID, UserID: input.SendInput.UserID, - MessageID: input.UserMessageID, + MessageID: input.AssistantMessageID, RunID: input.RunID, Rows: rows, }) diff --git a/backend/internal/application/conversation/service_message_send.go b/backend/internal/application/conversation/service_message_send.go index a2de7c761..44863ec66 100644 --- a/backend/internal/application/conversation/service_message_send.go +++ b/backend/internal/application/conversation/service_message_send.go @@ -397,19 +397,6 @@ func (s *Service) sendMessageInternal( prefetchCh <- r }() - // 异步语义召回：200ms 截止时限，不阻塞 LLM 关键路径。 - // 超时后优雅跳过；召回依赖 Embedding 服务。 - // 召回结果稍后作为用户上下文 XML 注入，避免把历史片段提升为 system 指令。 - var recallCh chan []model.MessageChunk - if cfg.EmbeddingEnabled && cfg.SemanticContextEnabled { - recallCh = make(chan []model.MessageChunk, 1) - go func() { - recallCtx, cancel := context.WithTimeout(ctx, semanticRecallDeadline) - defer cancel() - recallCh <- s.recallSemanticContext(recallCtx, input.ConversationID, input.UserID, input.Content) - }() - } - // 读取用户的文件处理模式偏好（auto / full_context / rag）。 fileMode := "auto" capability := s.resolveChatFileCapability(ctx) @@ -425,6 +412,19 @@ func (s *Service) sendMessageInternal( promptScope := buildPromptScope(contextMessages, prefetch.snapshot, compactPolicy) promptMessages := s.applyContextTokenBudget(promptScope.activeMessages(), route.UpstreamModel, route.ModelCapabilitiesJSON, reasoningContentPassback) ragQuery := buildRAGQuery(promptMessages, input.Content, cfg.RAGQueryHistoryTurns) + historicalScope := promptScope.historicalMessageScope(input.ConversationID, input.UserID, userMessage.ID) + + // 语义召回必须先限定到当前活跃分支，再由向量存储执行 Top-K，避免 sibling 分支占用名额。 + // 召回仍与附件和 RAG 处理并行，200ms 超时后按原行为优雅跳过。 + var recallCh chan []model.MessageChunk + if cfg.EmbeddingEnabled && cfg.SemanticContextEnabled && historicalScope.Valid() { + recallCh = make(chan []model.MessageChunk, 1) + go func() { + recallCtx, cancel := context.WithTimeout(ctx, semanticRecallDeadline) + defer cancel() + recallCh <- s.recallSemanticContext(recallCtx, historicalScope, input.Content) + }() + } conversationFileIDs := collectConversationFileIDs(promptMessages, input.FileIDs) conversationAttachments, err := s.resolveConversationFileContext(ctx, input.UserID, conversationFileIDs, input.FileIDs) @@ -612,7 +612,7 @@ func (s *Service) sendMessageInternal( userCtx.Attachments = imageAttachmentsForCurrentUser(stableFullContextAttachments) userCtx.RAGChunks = ragContextChunks // 语义召回注入：收集异步结果（与 RAG 解耦，独立运行）。 - // recallCh 为 nil 时（SemanticContextEnabled=false）直接跳过。 + // recallCh 为 nil 时（未启用语义召回或当前分支没有历史消息）直接跳过。 // // 必须阻塞等待（不用 select default），原因： // - 无附件时 hydrateAttachmentsForSend 几乎瞬间返回（~5ms）， @@ -621,16 +621,12 @@ func (s *Service) sendMessageInternal( // 因此 <-recallCh 最多阻塞 semanticRecallDeadline（200ms），不会死锁。 // - 有附件时 goroutine 早已完成（附件处理 >1s >> 200ms），等待开销为零。 if recallCh != nil { - recalled := <-recallCh // 阻塞等待，最多 semanticRecallDeadline（200ms） - userCtx.RecallChunks = promptScope.filterRecallChunks(recalled) + userCtx.RecallChunks = <-recallCh // 阻塞等待，最多 semanticRecallDeadline（200ms） } userCtx.HistoricalArtifacts = s.recallHistoricalContextArtifacts( ctx, - input.ConversationID, - userMessage.ID, + historicalScope, promptScope.Snapshot != nil, - promptScope.CoveredUntilID, - promptScope.retainedMessageIDSet(), input.Content, ragContextChunks, ragFallbackEvidenceAttachments(ragFallbacks), @@ -639,7 +635,7 @@ func (s *Service) sendMessageInternal( userCtx.CurrentArtifacts = s.persistPromptContextArtifacts(ctx, promptContextArtifactInput{ ConversationID: input.ConversationID, UserID: input.UserID, - MessageID: userMessage.ID, + MessageID: assistantMessage.ID, RunID: run.RunID, Query: ragQuery, RAGChunks: ragContextChunks, diff --git a/backend/internal/domain/conversation/context_artifact.go b/backend/internal/domain/conversation/context_artifact.go index 9518c01ea..cad207106 100644 --- a/backend/internal/domain/conversation/context_artifact.go +++ b/backend/internal/domain/conversation/context_artifact.go @@ -17,6 +17,7 @@ const ( ) // ContextArtifact 保存一次对话请求中被选入上下文规划的证据引用。 +// MessageID 指向产生该证据的助手消息节点，用于按不可变父链隔离分支。 type ContextArtifact struct { ID uint ConversationID uint diff --git a/backend/internal/infra/persistence/models/chat.go b/backend/internal/infra/persistence/models/chat.go index 6a2fa725c..e1899a935 100644 --- a/backend/internal/infra/persistence/models/chat.go +++ b/backend/internal/infra/persistence/models/chat.go @@ -335,7 +335,7 @@ type ChatContextRecord struct { BaseModel RecordType string `gorm:"size:32;not null;default:'';index:idx_chat_context_records_type;comment:记录类型(snapshot/artifact)"` ConversationID uint `gorm:"not null;default:0;index:idx_chat_context_records_conversation_id;index:idx_chat_context_records_conversation_message,priority:1;index:idx_chat_context_records_conversation_kind,priority:1;comment:会话ID"` - MessageID uint `gorm:"not null;default:0;index:idx_chat_context_records_message_id;index:idx_chat_context_records_conversation_message,priority:2;comment:触发消息ID"` + MessageID uint `gorm:"not null;default:0;index:idx_chat_context_records_message_id;index:idx_chat_context_records_conversation_message,priority:2;comment:证据归属助手消息ID"` UserID uint `gorm:"not null;default:0;index:idx_chat_context_records_user_id;comment:用户ID"` RunID string `gorm:"size:64;not null;default:'';index:idx_chat_context_records_run_id;comment:运行ID"` FromTurn int `gorm:"not null;default:0;comment:压缩快照起始轮次"` diff --git a/backend/internal/infra/persistence/postgres/conversation/repository.go b/backend/internal/infra/persistence/postgres/conversation/repository.go index f1453c757..07a312640 100644 --- a/backend/internal/infra/persistence/postgres/conversation/repository.go +++ b/backend/internal/infra/persistence/postgres/conversation/repository.go @@ -4316,12 +4316,12 @@ func insertSQLiteMessageChunkVectors(tx *gorm.DB, entities []models.MessageChunk return nil } -func (r *Repo) searchSQLiteMessageChunks(ctx context.Context, conversationID uint, userID uint, queryEmbedding []float32, topK int, minSimilarity float64) ([]domainconversation.MessageChunk, error) { - vector, err := sqlitevec.SerializeFloat32(queryEmbedding) +func (r *Repo) searchSQLiteMessageChunks(ctx context.Context, input repository.MessageChunkSearchInput) ([]domainconversation.MessageChunk, error) { + vector, err := sqlitevec.SerializeFloat32(input.QueryEmbedding) if err != nil { return nil, err } - query := fmt.Sprintf(` + query := historicalMessageScopeCTE + fmt.Sprintf(` SELECT chunks.id, chunks.conversation_id, chunks.message_id, chunks.user_id, chunks.role, chunks.chunk_index, chunks.content, chunks.token_count, chunks.created_at, (1.0 - vectors.distance) AS similarity @@ -4332,16 +4332,27 @@ func (r *Repo) searchSQLiteMessageChunks(ctx context.Context, conversationID uin AND vectors.k = ? AND vectors.user_id = ? AND vectors.conversation_id = ? + AND vectors.message_id IN ( + SELECT id + FROM valid_historical_message_scope + ) ORDER BY vectors.distance ASC`, sqlitevec.MessageChunkVectorTable, ) + args := historicalMessageScopeArgs(input.Scope) + args = append(args, + vector, + input.TopK, + input.Scope.UserID, + input.Scope.ConversationID, + ) var rows []messageChunkSearchRow - if err := r.db.WithContext(ctx).Raw(query, vector, topK, userID, conversationID).Scan(&rows).Error; err != nil { + if err := r.db.WithContext(ctx).Raw(query, args...).Scan(&rows).Error; err != nil { return nil, translateError(err) } results := make([]domainconversation.MessageChunk, 0, len(rows)) for _, row := range rows { - if row.Similarity < minSimilarity { + if row.Similarity < input.MinSimilarity { continue } results = append(results, domainconversation.MessageChunk{ @@ -4360,29 +4371,47 @@ func (r *Repo) searchSQLiteMessageChunks(ctx context.Context, conversationID uin return results, nil } -// SearchMessageChunks 按查询向量检索最相关的历史消息分片。 -func (r *Repo) SearchMessageChunks(ctx context.Context, conversationID uint, userID uint, queryEmbedding []float32, topK int, minSimilarity float64) ([]domainconversation.MessageChunk, error) { - if len(queryEmbedding) == 0 || topK <= 0 { +// SearchMessageChunks 在当前活跃分支内按查询向量检索最相关的历史消息分片。 +func (r *Repo) SearchMessageChunks(ctx context.Context, input repository.MessageChunkSearchInput) ([]domainconversation.MessageChunk, error) { + if !input.Scope.Valid() || len(input.QueryEmbedding) == 0 || input.TopK <= 0 { return nil, nil } if r.sqliteDialect() { - return r.searchSQLiteMessageChunks(ctx, conversationID, userID, queryEmbedding, topK, minSimilarity) - } - vec := float32SliceToPostgresVector(queryEmbedding) - query := ` - SELECT id, conversation_id, message_id, user_id, role, chunk_index, content, token_count, created_at, + return r.searchSQLiteMessageChunks(ctx, input) + } + vec := float32SliceToPostgresVector(input.QueryEmbedding) + // PostgreSQL 的 IVFFlat 会在近似索引扫描后应用普通过滤条件；直接 JOIN 分支范围可能让 sibling + // 候选先占满 Top-K。先物化当前分支分片，再执行精确距离排序，保证过滤严格发生在 Top-K 之前。 + query := historicalMessageScopeCTE + `, + branch_message_chunks AS MATERIALIZED ( + SELECT chunks.id, chunks.conversation_id, chunks.message_id, chunks.user_id, chunks.role, + chunks.chunk_index, chunks.content, chunks.token_count, chunks.created_at, chunks.embedding + FROM chat_message_chunks AS chunks + JOIN valid_historical_message_scope AS branch_scope ON branch_scope.id = chunks.message_id + WHERE chunks.conversation_id = ? + AND chunks.user_id = ? + AND chunks.embedding IS NOT NULL + ) + SELECT id, conversation_id, message_id, user_id, role, + chunk_index, content, token_count, created_at, (1 - (embedding <=> ?::vector)) AS similarity - FROM chat_message_chunks - WHERE conversation_id = ? AND user_id = ? AND embedding IS NOT NULL + FROM branch_message_chunks ORDER BY similarity DESC LIMIT ?` + args := historicalMessageScopeArgs(input.Scope) + args = append(args, + input.Scope.ConversationID, + input.Scope.UserID, + vec, + input.TopK, + ) var rows []messageChunkSearchRow - if err := r.db.WithContext(ctx).Raw(query, vec, conversationID, userID, topK).Scan(&rows).Error; err != nil { + if err := r.db.WithContext(ctx).Raw(query, args...).Scan(&rows).Error; err != nil { return nil, translateError(err) } results := make([]domainconversation.MessageChunk, 0, len(rows)) for _, row := range rows { - if row.Similarity < minSimilarity { + if row.Similarity < input.MinSimilarity { continue } results = append(results, domainconversation.MessageChunk{ diff --git a/backend/internal/infra/persistence/postgres/conversation/repository_context_artifact.go b/backend/internal/infra/persistence/postgres/conversation/repository_context_artifact.go index 96f71e504..c8d38c992 100644 --- a/backend/internal/infra/persistence/postgres/conversation/repository_context_artifact.go +++ b/backend/internal/infra/persistence/postgres/conversation/repository_context_artifact.go @@ -2,10 +2,12 @@ package conversation import ( "context" + "strings" "time" domainconversation "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation" models "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/persistence/models" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" ) // CreateContextArtifacts 批量写入本轮上下文证据。 @@ -15,6 +17,10 @@ func (r *Repo) CreateContextArtifacts(ctx context.Context, items []domainconvers } entities := make([]models.ChatContextRecord, 0, len(items)) for _, item := range items { + item.RunID = strings.TrimSpace(item.RunID) + if item.ConversationID == 0 || item.MessageID == 0 || item.UserID == 0 || item.RunID == "" { + return repository.ErrInvalidInput + } entities = append(entities, toContextArtifactModel(item)) } if err := r.db.WithContext(ctx).Create(&entities).Error; err != nil { @@ -39,7 +45,7 @@ func (r *Repo) GetContextArtifactByIDForUser(ctx context.Context, userID uint, a return &result, nil } -// ListContextArtifactsByMessage 查询单条用户消息对应的上下文证据。 +// ListContextArtifactsByMessage 查询单条消息对应的上下文证据。 func (r *Repo) ListContextArtifactsByMessage(ctx context.Context, conversationID uint, messageID uint) ([]domainconversation.ContextArtifact, error) { items := make([]models.ChatContextRecord, 0) if err := r.db.WithContext(ctx). @@ -52,28 +58,42 @@ func (r *Repo) ListContextArtifactsByMessage(ctx context.Context, conversationID return toContextArtifactDomains(items), nil } -// ListRecentContextArtifacts 按会话和类型查询最近的上下文证据。 -func (r *Repo) ListRecentContextArtifacts(ctx context.Context, conversationID uint, kinds []domainconversation.ContextArtifactKind, limit int) ([]domainconversation.ContextArtifact, error) { - if limit <= 0 { - limit = 20 +// ListRecentContextArtifacts 在当前活跃分支内按类型查询最近的上下文证据。 +func (r *Repo) ListRecentContextArtifacts(ctx context.Context, filter repository.ContextArtifactListFilter) ([]domainconversation.ContextArtifact, error) { + if !filter.Scope.Valid() { + return nil, nil + } + if filter.Limit <= 0 { + filter.Limit = 20 } - if limit > 200 { - limit = 200 + if filter.Limit > 200 { + filter.Limit = 200 } + scopeQuery := historicalMessageScopeSubquery(r.db.WithContext(ctx), filter.Scope) query := r.db.WithContext(ctx). - Where("record_type = ? AND conversation_id = ?", chatContextRecordArtifact, conversationID). - Where("expires_at IS NULL OR expires_at > ?", time.Now()). - Order("id DESC"). - Limit(limit) - if len(kinds) > 0 { - values := make([]string, 0, len(kinds)) - for _, kind := range kinds { + Model(&models.ChatContextRecord{}). + Select("chat_context_records.*"). + Joins(`JOIN chat_messages AS artifact_owner + ON artifact_owner.id = chat_context_records.message_id + AND artifact_owner.conversation_id = chat_context_records.conversation_id + AND artifact_owner.user_id = chat_context_records.user_id + AND artifact_owner.run_id = chat_context_records.run_id + AND artifact_owner.role = ? + AND artifact_owner.deleted_at IS NULL`, "assistant"). + Where("chat_context_records.record_type = ? AND chat_context_records.conversation_id = ? AND chat_context_records.user_id = ?", chatContextRecordArtifact, filter.Scope.ConversationID, filter.Scope.UserID). + Where("chat_context_records.message_id IN (?)", scopeQuery). + Where("chat_context_records.expires_at IS NULL OR chat_context_records.expires_at > ?", time.Now()). + Order("chat_context_records.id DESC"). + Limit(filter.Limit) + if len(filter.Kinds) > 0 { + values := make([]string, 0, len(filter.Kinds)) + for _, kind := range filter.Kinds { if kind != "" { values = append(values, string(kind)) } } if len(values) > 0 { - query = query.Where("kind IN ?", values) + query = query.Where("chat_context_records.kind IN ?", values) } } items := make([]models.ChatContextRecord, 0) diff --git a/backend/internal/infra/persistence/postgres/conversation/repository_historical_scope.go b/backend/internal/infra/persistence/postgres/conversation/repository_historical_scope.go new file mode 100644 index 000000000..6648e8654 --- /dev/null +++ b/backend/internal/infra/persistence/postgres/conversation/repository_historical_scope.go @@ -0,0 +1,58 @@ +package conversation + +import ( + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" + "gorm.io/gorm" +) + +// historicalMessageScopeCTE 从当前叶消息沿不可变 parent_message_id 向上解析分支。 +// UNION 在异常循环父指针下去重终止；ExcludeThroughMessageID 命中后停止继续向上遍历。 +const historicalMessageScopeCTE = ` +WITH RECURSIVE historical_message_scope(id, parent_message_id) AS ( + SELECT id, parent_message_id + FROM chat_messages + WHERE id = ? AND conversation_id = ? AND user_id = ? AND deleted_at IS NULL + UNION + SELECT messages.id, messages.parent_message_id + FROM chat_messages AS messages + INNER JOIN historical_message_scope AS scope ON messages.id = scope.parent_message_id + WHERE scope.id <> ? + AND messages.conversation_id = ? + AND messages.user_id = ? + AND messages.deleted_at IS NULL +), valid_historical_message_scope(id) AS ( + SELECT scope.id + FROM historical_message_scope AS scope + WHERE scope.id <> ? + AND scope.id <> ? + AND ( + ? = 0 OR EXISTS ( + SELECT 1 + FROM historical_message_scope AS boundary + WHERE boundary.id = ? + ) + ) +)` + +const historicalMessageScopeSubquerySQL = historicalMessageScopeCTE + ` +SELECT id +FROM valid_historical_message_scope` + +func historicalMessageScopeArgs(scope repository.HistoricalMessageScope) []interface{} { + return []interface{}{ + scope.LeafMessageID, + scope.ConversationID, + scope.UserID, + scope.ExcludeThroughMessageID, + scope.ConversationID, + scope.UserID, + scope.LeafMessageID, + scope.ExcludeThroughMessageID, + scope.ExcludeThroughMessageID, + scope.ExcludeThroughMessageID, + } +} + +func historicalMessageScopeSubquery(db *gorm.DB, scope repository.HistoricalMessageScope) *gorm.DB { + return db.Raw(historicalMessageScopeSubquerySQL, historicalMessageScopeArgs(scope)...) +} diff --git a/backend/internal/infra/persistence/postgres/conversation/repository_postgres_integration_test.go b/backend/internal/infra/persistence/postgres/conversation/repository_postgres_integration_test.go new file mode 100644 index 000000000..df87466f1 --- /dev/null +++ b/backend/internal/infra/persistence/postgres/conversation/repository_postgres_integration_test.go @@ -0,0 +1,130 @@ +package conversation + +import ( + "context" + "fmt" + "os" + "strings" + "testing" + "time" + + model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/persistence/models" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" + "gorm.io/driver/postgres" + "gorm.io/gorm" +) + +func TestSearchMessageChunksFiltersPostgresBranchBeforeTopK(t *testing.T) { + dsn := strings.TrimSpace(os.Getenv("DEEIX_TEST_DATABASE_DSN")) + if dsn == "" { + t.Skip("set DEEIX_TEST_DATABASE_DSN to run PostgreSQL branch-scoped vector integration test") + } + + db, cleanup := openConversationPostgresIntegrationDB(t, dsn) + t.Cleanup(cleanup) + if err := db.AutoMigrate(&model.Message{}, &model.MessageChunk{}); err != nil { + t.Fatalf("migrate conversation vector models: %v", err) + } + if err := db.Exec(`ALTER TABLE chat_message_chunks ADD COLUMN IF NOT EXISTS embedding vector(1536)`).Error; err != nil { + t.Fatalf("add message embedding column: %v", err) + } + + root := model.Message{ConversationID: 20, UserID: 1, PublicID: "msg_pg_vector_root", Role: "user", Status: "success"} + if err := db.Create(&root).Error; err != nil { + t.Fatalf("create root message: %v", err) + } + active := model.Message{ + ConversationID: 20, UserID: 1, PublicID: "msg_pg_vector_active", ParentMessageID: &root.ID, + Role: "assistant", Status: "success", + } + if err := db.Create(&active).Error; err != nil { + t.Fatalf("create active message: %v", err) + } + sibling := model.Message{ + ConversationID: 20, UserID: 1, PublicID: "msg_pg_vector_sibling", ParentMessageID: &root.ID, + Role: "assistant", BranchReason: "retry", Status: "success", + } + if err := db.Create(&sibling).Error; err != nil { + t.Fatalf("create sibling message: %v", err) + } + leaf := model.Message{ + ConversationID: 20, UserID: 1, PublicID: "msg_pg_vector_leaf", ParentMessageID: &active.ID, + Role: "user", Status: "pending", + } + if err := db.Create(&leaf).Error; err != nil { + t.Fatalf("create leaf message: %v", err) + } + + chunks := []model.MessageChunk{ + {ConversationID: 20, MessageID: active.ID, UserID: 1, Role: "assistant", Content: "active branch target"}, + {ConversationID: 20, MessageID: sibling.ID, UserID: 1, Role: "assistant", Content: "closer sibling target"}, + } + if err := db.Create(&chunks).Error; err != nil { + t.Fatalf("create message chunks: %v", err) + } + queryEmbedding := make([]float32, 1536) + queryEmbedding[0] = 1 + activeEmbedding := make([]float32, 1536) + activeEmbedding[0], activeEmbedding[1] = 0.8, 0.6 + if err := db.Exec(`UPDATE chat_message_chunks SET embedding = ?::vector WHERE id = ?`, float32SliceToPostgresVector(activeEmbedding), chunks[0].ID).Error; err != nil { + t.Fatalf("write active embedding: %v", err) + } + if err := db.Exec(`UPDATE chat_message_chunks SET embedding = ?::vector WHERE id = ?`, float32SliceToPostgresVector(queryEmbedding), chunks[1].ID).Error; err != nil { + t.Fatalf("write sibling embedding: %v", err) + } + if err := db.Exec(`CREATE INDEX message_chunks_embedding_test_idx ON chat_message_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists = 1)`).Error; err != nil { + t.Fatalf("create message embedding index: %v", err) + } + + results, err := NewRepo(db).SearchMessageChunks(context.Background(), repository.MessageChunkSearchInput{ + Scope: repository.HistoricalMessageScope{ + ConversationID: 20, + UserID: 1, + LeafMessageID: leaf.ID, + }, + QueryEmbedding: queryEmbedding, + TopK: 1, + }) + if err != nil { + t.Fatalf("SearchMessageChunks() error = %v", err) + } + if len(results) != 1 || results[0].MessageID != active.ID { + t.Fatalf("expected active branch result despite closer sibling, got %#v", results) + } +} + +func openConversationPostgresIntegrationDB(t *testing.T, dsn string) (*gorm.DB, func()) { + t.Helper() + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{}) + if err != nil { + t.Fatalf("open postgres: %v", err) + } + sqlDB, err := db.DB() + if err != nil { + t.Fatalf("resolve postgres db: %v", err) + } + sqlDB.SetMaxOpenConns(1) + var vectorAvailable bool + if err := db.Raw(`SELECT to_regtype('vector') IS NOT NULL`).Scan(&vectorAvailable).Error; err != nil { + _ = sqlDB.Close() + t.Fatalf("check pgvector extension: %v", err) + } + if !vectorAvailable { + _ = sqlDB.Close() + t.Skip("pgvector extension is required for PostgreSQL branch-scoped vector integration test") + } + schemaName := fmt.Sprintf("deeix_test_conversation_scope_%d", time.Now().UnixNano()) + if err := db.Exec(`CREATE SCHEMA ` + schemaName).Error; err != nil { + _ = sqlDB.Close() + t.Fatalf("create test schema: %v", err) + } + cleanup := func() { + _ = db.Exec(`DROP SCHEMA IF EXISTS ` + schemaName + ` CASCADE`).Error + _ = sqlDB.Close() + } + if err := db.Exec(`SET search_path TO ` + schemaName + `, public`).Error; err != nil { + cleanup() + t.Fatalf("set test search path: %v", err) + } + return db, cleanup +} diff --git a/backend/internal/infra/persistence/postgres/conversation/repository_sqlite_vector_test.go b/backend/internal/infra/persistence/postgres/conversation/repository_sqlite_vector_test.go index 9238f2e7f..0413d2d19 100644 --- a/backend/internal/infra/persistence/postgres/conversation/repository_sqlite_vector_test.go +++ b/backend/internal/infra/persistence/postgres/conversation/repository_sqlite_vector_test.go @@ -7,6 +7,7 @@ import ( domainconversation "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation" model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/persistence/models" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/persistence/sqlitevec" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" "gorm.io/driver/sqlite" "gorm.io/gorm" ) @@ -44,23 +45,71 @@ func TestSQLiteVectorStoreSearchesFileAndMessageChunks(t *testing.T) { } messageChunks := []domainconversation.MessageChunk{ - {ConversationID: 20, MessageID: 30, UserID: 1, Role: "user", ChunkIndex: 0, Content: "message target", TokenCount: 2}, + {ConversationID: 20, MessageID: 30, UserID: 1, Role: "assistant", ChunkIndex: 0, Content: "message target", TokenCount: 2}, {ConversationID: 20, MessageID: 31, UserID: 1, Role: "assistant", ChunkIndex: 0, Content: "message unrelated", TokenCount: 2}, } + rootMessageID := uint(29) + activeMessageID := uint(30) + branchMessages := []model.Message{ + { + BaseModel: model.BaseModel{ID: rootMessageID}, ConversationID: 20, UserID: 1, + PublicID: "msg_vector_root", Role: "user", Status: "success", + }, + { + BaseModel: model.BaseModel{ID: activeMessageID}, ConversationID: 20, UserID: 1, + PublicID: "msg_vector_active", ParentMessageID: &rootMessageID, Role: "assistant", Status: "success", + }, + { + BaseModel: model.BaseModel{ID: 31}, ConversationID: 20, UserID: 1, + PublicID: "msg_vector_sibling", ParentMessageID: &rootMessageID, Role: "assistant", Status: "success", + }, + { + BaseModel: model.BaseModel{ID: 32}, ConversationID: 20, UserID: 1, + PublicID: "msg_vector_leaf", ParentMessageID: &activeMessageID, Role: "user", Status: "pending", + }, + } + if err := db.Create(&branchMessages).Error; err != nil { + t.Fatalf("create message branch: %v", err) + } messageEmbeddings := [][]float32{ - {0, 0, 1}, - {0, 1, 0}, + {0.8, 0.6, 0}, + {1, 0, 0}, } if err := repo.UpsertMessageChunks(ctx, messageChunks, messageEmbeddings); err != nil { t.Fatalf("UpsertMessageChunks() error = %v", err) } - messageResults, err := repo.SearchMessageChunks(ctx, 20, 1, []float32{0, 0, 1}, 2, 0) + messageResults, err := repo.SearchMessageChunks(ctx, repository.MessageChunkSearchInput{ + Scope: repository.HistoricalMessageScope{ + ConversationID: 20, + UserID: 1, + LeafMessageID: 32, + }, + QueryEmbedding: []float32{1, 0, 0}, + TopK: 1, + }) if err != nil { t.Fatalf("SearchMessageChunks() error = %v", err) } if len(messageResults) == 0 || messageResults[0].Content != "message target" { t.Fatalf("expected nearest message chunk first, got %#v", messageResults) } + + coveredResults, err := repo.SearchMessageChunks(ctx, repository.MessageChunkSearchInput{ + Scope: repository.HistoricalMessageScope{ + ConversationID: 20, + UserID: 1, + LeafMessageID: 32, + ExcludeThroughMessageID: activeMessageID, + }, + QueryEmbedding: []float32{1, 0, 0}, + TopK: 1, + }) + if err != nil { + t.Fatalf("SearchMessageChunks(snapshot scope) error = %v", err) + } + if len(coveredResults) != 0 { + t.Fatalf("expected snapshot boundary to exclude covered chunk, got %#v", coveredResults) + } } func openConversationSQLiteVectorTestDB(t *testing.T) *gorm.DB { @@ -76,7 +125,7 @@ func openConversationSQLiteVectorTestDB(t *testing.T) *gorm.DB { _ = sqlDB.Close() } }) - if err := db.AutoMigrate(&model.FileChunk{}, &model.MessageChunk{}); err != nil { + if err := db.AutoMigrate(&model.FileChunk{}, &model.MessageChunk{}, &model.Message{}); err != nil { t.Fatalf("migrate models: %v", err) } if err := sqlitevec.Migrate(db); err != nil { diff --git a/backend/internal/infra/persistence/postgres/conversation/repository_test.go b/backend/internal/infra/persistence/postgres/conversation/repository_test.go index 923fe373f..515096689 100644 --- a/backend/internal/infra/persistence/postgres/conversation/repository_test.go +++ b/backend/internal/infra/persistence/postgres/conversation/repository_test.go @@ -22,6 +22,178 @@ func TestTranslateErrorAllowsNil(t *testing.T) { } } +func TestCreateContextArtifactsRejectsIncompleteOwnerScope(t *testing.T) { + repo := NewRepo(openConversationRepositoryTestDB(t)) + valid := domainconversation.ContextArtifact{ + ConversationID: 7, + MessageID: 11, + UserID: 1, + RunID: "run_1", + Kind: domainconversation.ContextArtifactToolResult, + Content: "evidence", + } + tests := map[string]func(*domainconversation.ContextArtifact){ + "conversation": func(item *domainconversation.ContextArtifact) { item.ConversationID = 0 }, + "message": func(item *domainconversation.ContextArtifact) { item.MessageID = 0 }, + "user": func(item *domainconversation.ContextArtifact) { item.UserID = 0 }, + "run": func(item *domainconversation.ContextArtifact) { item.RunID = " " }, + } + for name, invalidate := range tests { + t.Run(name, func(t *testing.T) { + item := valid + invalidate(&item) + err := repo.CreateContextArtifacts(context.Background(), []domainconversation.ContextArtifact{item}) + if !errors.Is(err, repository.ErrInvalidInput) { + t.Fatalf("CreateContextArtifacts() error = %v, want ErrInvalidInput", err) + } + }) + } +} + +func TestCreateContextArtifactsNormalizesRunOwner(t *testing.T) { + db := openConversationRepositoryTestDB(t) + if err := db.AutoMigrate(&model.ChatContextRecord{}); err != nil { + t.Fatalf("migrate context records: %v", err) + } + repo := NewRepo(db) + items := []domainconversation.ContextArtifact{{ + ConversationID: 7, + MessageID: 11, + UserID: 1, + RunID: " run_1 ", + Kind: domainconversation.ContextArtifactToolResult, + Content: "normalized evidence", + }} + + if err := repo.CreateContextArtifacts(context.Background(), items); err != nil { + t.Fatalf("CreateContextArtifacts() error = %v", err) + } + if items[0].RunID != "run_1" { + t.Fatalf("artifact run id = %q, want normalized run_1", items[0].RunID) + } +} + +func TestListRecentContextArtifactsFiltersBranchBeforeLimit(t *testing.T) { + db := openConversationRepositoryTestDB(t) + if err := db.AutoMigrate(&model.Message{}, &model.ChatContextRecord{}); err != nil { + t.Fatalf("migrate context records: %v", err) + } + repo := NewRepo(db) + ctx := context.Background() + rootMessageID := uint(1) + activeOwnerID := uint(10) + leafMessageID := uint(12) + branchMessages := []model.Message{ + { + BaseModel: model.BaseModel{ID: rootMessageID}, + ConversationID: 7, + UserID: 1, + PublicID: "msg_branch_root", + Role: "user", + Status: "success", + }, + { + BaseModel: model.BaseModel{ID: activeOwnerID}, + ConversationID: 7, + UserID: 1, + PublicID: "msg_artifact_owner", + ParentMessageID: &rootMessageID, + RunID: "run_active", + Role: "assistant", + Status: "success", + }, + { + BaseModel: model.BaseModel{ID: leafMessageID}, + ConversationID: 7, + UserID: 1, + PublicID: "msg_branch_leaf", + ParentMessageID: &activeOwnerID, + Role: "user", + Status: "pending", + }, + } + for index := 0; index < 31; index++ { + branchMessages = append(branchMessages, model.Message{ + BaseModel: model.BaseModel{ID: uint(100 + index)}, + ConversationID: 7, + UserID: 1, + PublicID: fmt.Sprintf("msg_sibling_%d", index), + ParentMessageID: &rootMessageID, + Role: "assistant", + Status: "success", + }) + } + if err := db.Create(&branchMessages).Error; err != nil { + t.Fatalf("create branch messages: %v", err) + } + + items := []model.ChatContextRecord{ + { + RecordType: chatContextRecordArtifact, + ConversationID: 7, + MessageID: activeOwnerID, + UserID: 1, + RunID: "run_active", + Kind: string(domainconversation.ContextArtifactToolResult), + SourceType: "tool_call", + SourceID: "active", + Content: "active branch evidence", + }, + { + RecordType: chatContextRecordArtifact, + ConversationID: 7, + MessageID: rootMessageID, + UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), + SourceType: "tool_call", + SourceID: "user-owned", + Content: "legacy evidence with ambiguous branch ownership", + }, + { + RecordType: chatContextRecordArtifact, + ConversationID: 7, + MessageID: activeOwnerID, + UserID: 1, + RunID: "run_wrong_owner", + Kind: string(domainconversation.ContextArtifactToolResult), + SourceType: "tool_call", + SourceID: "mismatched-run", + Content: "evidence must not borrow an unrelated assistant owner", + }, + } + for index := 0; index < 31; index++ { + items = append(items, model.ChatContextRecord{ + RecordType: chatContextRecordArtifact, + ConversationID: 7, + MessageID: uint(100 + index), + UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), + SourceType: "tool_call", + SourceID: fmt.Sprintf("sibling-%d", index), + Content: "sibling branch evidence", + }) + } + if err := db.Create(&items).Error; err != nil { + t.Fatalf("create context records: %v", err) + } + + artifacts, err := repo.ListRecentContextArtifacts(ctx, repository.ContextArtifactListFilter{ + Scope: repository.HistoricalMessageScope{ + ConversationID: 7, + UserID: 1, + LeafMessageID: leafMessageID, + }, + Kinds: []domainconversation.ContextArtifactKind{domainconversation.ContextArtifactToolResult}, + Limit: 1, + }) + if err != nil { + t.Fatalf("ListRecentContextArtifacts() error = %v", err) + } + if len(artifacts) != 1 || artifacts[0].MessageID != activeOwnerID { + t.Fatalf("expected active branch evidence before limit, got %#v", artifacts) + } +} + func TestConversationProjectDefaultsRoundTripAndDelete(t *testing.T) { db := openConversationRepositoryTestDB(t) repo := NewRepo(db) @@ -1222,3 +1394,198 @@ func TestListMessageAncestorsStopsAtConversationBoundary(t *testing.T) { t.Fatalf("expected only the in-conversation leaf, got %#v", got) } } + +func TestListRecentContextArtifactsUsesCTEForLongBranchAndSnapshotBoundary(t *testing.T) { + db := openConversationRepositoryTestDB(t) + if err := db.AutoMigrate(&model.Message{}, &model.ChatContextRecord{}); err != nil { + t.Fatalf("migrate context records: %v", err) + } + repo := NewRepo(db) + ctx := context.Background() + conversationID := uint(77) + + const branchLength = 1205 + var parentMessageID *uint + branchMessages := make([]model.Message, 0, branchLength) + branchMessageIDs := make([]uint, 0, branchLength) + for index := 0; index < branchLength; index++ { + messageID := uint(10_000 + index) + message := model.Message{ + BaseModel: model.BaseModel{ID: messageID}, + ConversationID: conversationID, + UserID: 1, + PublicID: fmt.Sprintf("msg_context_long_%d", index), + ParentMessageID: parentMessageID, + Role: []string{"user", "assistant"}[index%2], + ContentType: "text", + Content: fmt.Sprintf("message %d", index), + BranchReason: "default", + Status: "success", + } + branchMessages = append(branchMessages, message) + branchMessageIDs = append(branchMessageIDs, messageID) + parentMessageID = &messageID + } + if err := db.CreateInBatches(&branchMessages, 50).Error; err != nil { + t.Fatalf("create %d branch messages: %v", branchLength, err) + } + sibling := model.Message{ + ConversationID: conversationID, + UserID: 1, + PublicID: "msg_context_long_sibling", + ParentMessageID: &branchMessageIDs[10], + Role: "assistant", + ContentType: "text", + Content: "sibling", + BranchReason: "retry", + Status: "success", + } + if err := db.Create(&sibling).Error; err != nil { + t.Fatalf("create sibling: %v", err) + } + artifacts := []model.ChatContextRecord{ + { + RecordType: chatContextRecordArtifact, ConversationID: conversationID, MessageID: branchMessageIDs[1], UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), SourceType: "tool_call", SourceID: "covered", Content: "covered evidence", + }, + { + RecordType: chatContextRecordArtifact, ConversationID: conversationID, MessageID: branchMessageIDs[999], UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), SourceType: "tool_call", SourceID: "boundary", Content: "boundary evidence", + }, + { + RecordType: chatContextRecordArtifact, ConversationID: conversationID, MessageID: branchMessageIDs[branchLength-2], UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), SourceType: "tool_call", SourceID: "retained", Content: "retained evidence", + }, + { + RecordType: chatContextRecordArtifact, ConversationID: conversationID, MessageID: sibling.ID, UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), SourceType: "tool_call", SourceID: "sibling", Content: "sibling evidence", + }, + } + if err := db.Create(&artifacts).Error; err != nil { + t.Fatalf("create context artifacts: %v", err) + } + + items, err := repo.ListRecentContextArtifacts(ctx, repository.ContextArtifactListFilter{ + Scope: repository.HistoricalMessageScope{ + ConversationID: conversationID, + UserID: 1, + LeafMessageID: branchMessageIDs[branchLength-1], + ExcludeThroughMessageID: branchMessageIDs[999], + }, + Kinds: []domainconversation.ContextArtifactKind{domainconversation.ContextArtifactToolResult}, + Limit: 10, + }) + if err != nil { + t.Fatalf("ListRecentContextArtifacts() error = %v", err) + } + if len(items) != 1 || items[0].SourceID != "retained" { + t.Fatalf("expected only retained long-branch artifact, got %#v", items) + } + + items, err = repo.ListRecentContextArtifacts(ctx, repository.ContextArtifactListFilter{ + Scope: repository.HistoricalMessageScope{ + ConversationID: conversationID, + UserID: 1, + LeafMessageID: branchMessageIDs[branchLength-1], + ExcludeThroughMessageID: sibling.ID, + }, + Kinds: []domainconversation.ContextArtifactKind{domainconversation.ContextArtifactToolResult}, + Limit: 10, + }) + if err != nil { + t.Fatalf("ListRecentContextArtifacts(invalid boundary) error = %v", err) + } + if len(items) != 0 { + t.Fatalf("expected non-ancestor boundary to fail closed, got %#v", items) + } +} + +func TestListRecentContextArtifactsHistoricalScopeTerminatesCycle(t *testing.T) { + db := openConversationRepositoryTestDB(t) + if err := db.AutoMigrate(&model.Message{}, &model.ChatContextRecord{}); err != nil { + t.Fatalf("migrate context records: %v", err) + } + repo := NewRepo(db) + ctx := context.Background() + conversationID := uint(88) + first := model.Message{ + ConversationID: conversationID, UserID: 1, PublicID: "msg_scope_cycle_first", + Role: "assistant", ContentType: "text", Content: "first", BranchReason: "default", Status: "success", + } + if err := db.Create(&first).Error; err != nil { + t.Fatalf("create first message: %v", err) + } + second := model.Message{ + ConversationID: conversationID, UserID: 1, PublicID: "msg_scope_cycle_second", + ParentMessageID: &first.ID, + Role: "user", ContentType: "text", Content: "second", BranchReason: "default", Status: "success", + } + if err := db.Create(&second).Error; err != nil { + t.Fatalf("create second message: %v", err) + } + if err := db.Model(&first).Update("parent_message_id", second.ID).Error; err != nil { + t.Fatalf("create cycle: %v", err) + } + artifact := model.ChatContextRecord{ + RecordType: chatContextRecordArtifact, ConversationID: conversationID, MessageID: first.ID, UserID: 1, + Kind: string(domainconversation.ContextArtifactToolResult), SourceType: "tool_call", SourceID: "cycle", Content: "cycle evidence", + } + if err := db.Create(&artifact).Error; err != nil { + t.Fatalf("create cycle artifact: %v", err) + } + + items, err := repo.ListRecentContextArtifacts(ctx, repository.ContextArtifactListFilter{ + Scope: repository.HistoricalMessageScope{ConversationID: conversationID, UserID: 1, LeafMessageID: second.ID}, + Kinds: []domainconversation.ContextArtifactKind{domainconversation.ContextArtifactToolResult}, + Limit: 10, + }) + if err != nil { + t.Fatalf("ListRecentContextArtifacts() error = %v", err) + } + if len(items) != 1 || items[0].MessageID != first.ID { + t.Fatalf("expected cycle to terminate with one historical artifact, got %#v", items) + } +} + +func TestHistoricalMessageScopeStopsAtUserBoundary(t *testing.T) { + db := openConversationRepositoryTestDB(t) + if err := db.AutoMigrate(&model.Message{}); err != nil { + t.Fatalf("migrate messages: %v", err) + } + conversationID := uint(89) + ownerAncestor := model.Message{ + ConversationID: conversationID, UserID: 1, PublicID: "msg_scope_owner_ancestor", + Role: "assistant", ContentType: "text", Content: "owner ancestor", BranchReason: "default", Status: "success", + } + if err := db.Create(&ownerAncestor).Error; err != nil { + t.Fatalf("create owner ancestor: %v", err) + } + foreignParent := model.Message{ + ConversationID: conversationID, UserID: 2, PublicID: "msg_scope_foreign_parent", + ParentMessageID: &ownerAncestor.ID, + Role: "assistant", ContentType: "text", Content: "foreign parent", BranchReason: "default", Status: "success", + } + if err := db.Create(&foreignParent).Error; err != nil { + t.Fatalf("create foreign parent: %v", err) + } + leaf := model.Message{ + ConversationID: conversationID, UserID: 1, PublicID: "msg_scope_owner_leaf", + ParentMessageID: &foreignParent.ID, + Role: "user", ContentType: "text", Content: "owner leaf", BranchReason: "default", Status: "pending", + } + if err := db.Create(&leaf).Error; err != nil { + t.Fatalf("create owner leaf: %v", err) + } + + var messageIDs []uint + if err := historicalMessageScopeSubquery(db, repository.HistoricalMessageScope{ + ConversationID: conversationID, + UserID: 1, + LeafMessageID: leaf.ID, + }).Scan(&messageIDs).Error; err != nil { + t.Fatalf("query historical scope: %v", err) + } + if len(messageIDs) != 0 { + t.Fatalf("expected traversal to stop at foreign-user parent, got message ids %v", messageIDs) + } +} diff --git a/backend/internal/infra/persistence/schema/schema.go b/backend/internal/infra/persistence/schema/schema.go index 48c77924d..6a3156edb 100644 --- a/backend/internal/infra/persistence/schema/schema.go +++ b/backend/internal/infra/persistence/schema/schema.go @@ -81,9 +81,60 @@ func Migrate(db *gorm.DB) error { if err := db.AutoMigrate(Models()...); err != nil { return err } + if err := backfillContextArtifactMessageIDs(db); err != nil { + return err + } return backfillUsageLedgerBillingAt(db) } +// backfillContextArtifactMessageIDs 将旧证据统一迁移到产生该证据的助手运行节点。 +// 同一次生成的 user/assistant 消息共享 run_id；仅在唯一匹配助手消息时回填，异常重复 run 数据保持不变。 +func backfillContextArtifactMessageIDs(db *gorm.DB) error { + if !db.Migrator().HasTable(&model.ChatContextRecord{}) || !db.Migrator().HasTable(&model.Message{}) { + return nil + } + return db.Exec(` + WITH artifacts_to_backfill AS ( + SELECT records.id, records.run_id, records.conversation_id, records.user_id + FROM chat_context_records AS records + LEFT JOIN chat_messages AS current_owner + ON current_owner.id = records.message_id + AND current_owner.run_id = records.run_id + AND current_owner.conversation_id = records.conversation_id + AND current_owner.user_id = records.user_id + AND current_owner.role = 'assistant' + AND current_owner.deleted_at IS NULL + WHERE records.record_type = 'artifact' + AND records.run_id <> '' + AND records.deleted_at IS NULL + AND current_owner.id IS NULL + ), + unique_assistant_run_owners AS ( + SELECT artifacts.id AS record_id, MIN(messages.id) AS message_id + FROM artifacts_to_backfill AS artifacts + JOIN chat_messages AS messages + ON messages.run_id = artifacts.run_id + AND messages.conversation_id = artifacts.conversation_id + AND messages.user_id = artifacts.user_id + AND messages.role = 'assistant' + AND messages.deleted_at IS NULL + GROUP BY artifacts.id + HAVING COUNT(*) = 1 + ) + UPDATE chat_context_records + SET message_id = ( + SELECT owners.message_id + FROM unique_assistant_run_owners AS owners + WHERE owners.record_id = chat_context_records.id + ) + WHERE EXISTS ( + SELECT 1 + FROM unique_assistant_run_owners AS owners + WHERE owners.record_id = chat_context_records.id + ) + `).Error +} + func backfillUsageLedgerBillingAt(db *gorm.DB) error { if !db.Migrator().HasTable(&model.UsageLedger{}) || !db.Migrator().HasColumn(&model.UsageLedger{}, "billing_at") { return nil diff --git a/backend/internal/infra/persistence/schema/schema_test.go b/backend/internal/infra/persistence/schema/schema_test.go index 1dd106a2a..2ae3775a3 100644 --- a/backend/internal/infra/persistence/schema/schema_test.go +++ b/backend/internal/infra/persistence/schema/schema_test.go @@ -66,6 +66,170 @@ func TestMigrateLeavesLegacyMCPToolMetadataPendingConfirmation(t *testing.T) { } } +func TestBackfillContextArtifactMessageIDsUsesAssistantRunOwner(t *testing.T) { + dbName := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name()) + db, err := gorm.Open(sqlite.Open("file:"+dbName+"?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { + sqlDB, dbErr := db.DB() + if dbErr == nil { + _ = sqlDB.Close() + } + }) + if err = db.AutoMigrate(&model.Message{}, &model.ChatContextRecord{}); err != nil { + t.Fatalf("migrate context artifacts: %v", err) + } + userMessage := model.Message{ + ConversationID: 7, + UserID: 11, + PublicID: "msg_user", + RunID: "run_tool", + Role: "user", + Status: "success", + } + if err = db.Create(&userMessage).Error; err != nil { + t.Fatalf("create user message: %v", err) + } + assistantMessage := model.Message{ + ConversationID: 7, + UserID: 11, + PublicID: "msg_assistant", + ParentMessageID: &userMessage.ID, + RunID: "run_tool", + Role: "assistant", + Status: "success", + } + if err = db.Create(&assistantMessage).Error; err != nil { + t.Fatalf("create assistant message: %v", err) + } + ambiguousUser := model.Message{ + ConversationID: 8, + UserID: 13, + PublicID: "msg_ambiguous_user", + RunID: "run_ambiguous", + Role: "user", + Status: "success", + } + if err = db.Create(&ambiguousUser).Error; err != nil { + t.Fatalf("create ambiguous user message: %v", err) + } + ambiguousAssistants := []model.Message{ + { + ConversationID: 8, UserID: 13, PublicID: "msg_ambiguous_assistant_1", + ParentMessageID: &ambiguousUser.ID, RunID: "run_ambiguous", Role: "assistant", Status: "success", + }, + { + ConversationID: 8, UserID: 13, PublicID: "msg_ambiguous_assistant_2", + ParentMessageID: &ambiguousUser.ID, RunID: "run_ambiguous", Role: "assistant", Status: "success", + }, + } + if err = db.Create(&ambiguousAssistants).Error; err != nil { + t.Fatalf("create ambiguous assistant messages: %v", err) + } + artifacts := []model.ChatContextRecord{ + { + RecordType: "artifact", + ConversationID: 7, + MessageID: userMessage.ID, + UserID: 11, + RunID: "run_tool", + Kind: "tool_result", + SourceType: "tool_call", + SourceID: "call_1", + Content: "tool output", + }, + { + RecordType: "artifact", + ConversationID: 7, + MessageID: userMessage.ID, + UserID: 11, + RunID: "run_tool", + Kind: "file_rag_chunk", + SourceType: "file_chunk", + SourceID: "file_1:0", + Content: "file output", + }, + { + RecordType: "artifact", + ConversationID: 7, + MessageID: 99, + UserID: 12, + RunID: "run_tool", + Kind: "tool_result", + SourceType: "tool_call", + SourceID: "foreign_user_call", + Content: "foreign user output", + }, + { + RecordType: "artifact", + ConversationID: 8, + MessageID: ambiguousUser.ID, + UserID: 13, + RunID: "run_ambiguous", + Kind: "tool_result", + SourceType: "tool_call", + SourceID: "ambiguous_call", + Content: "ambiguous output", + }, + { + RecordType: "snapshot", + ConversationID: 7, + MessageID: userMessage.ID, + UserID: 11, + RunID: "run_tool", + SummaryText: "snapshot remains anchored by its own schema", + }, + } + if err = db.Create(&artifacts).Error; err != nil { + t.Fatalf("create artifacts: %v", err) + } + + if err = backfillContextArtifactMessageIDs(db); err != nil { + t.Fatalf("backfillContextArtifactMessageIDs() error = %v", err) + } + if err = backfillContextArtifactMessageIDs(db); err != nil { + t.Fatalf("backfillContextArtifactMessageIDs() second error = %v", err) + } + + var toolArtifact model.ChatContextRecord + if err = db.Where("source_id = ?", "call_1").First(&toolArtifact).Error; err != nil { + t.Fatalf("load tool artifact: %v", err) + } + if toolArtifact.MessageID != assistantMessage.ID { + t.Fatalf("tool artifact message id = %d, want %d", toolArtifact.MessageID, assistantMessage.ID) + } + var fileArtifact model.ChatContextRecord + if err = db.Where("source_type = ?", "file_chunk").First(&fileArtifact).Error; err != nil { + t.Fatalf("load file artifact: %v", err) + } + if fileArtifact.MessageID != assistantMessage.ID { + t.Fatalf("file artifact message id = %d, want %d", fileArtifact.MessageID, assistantMessage.ID) + } + var foreignUserArtifact model.ChatContextRecord + if err = db.Where("source_id = ?", "foreign_user_call").First(&foreignUserArtifact).Error; err != nil { + t.Fatalf("load foreign user artifact: %v", err) + } + if foreignUserArtifact.MessageID != 99 { + t.Fatalf("foreign user artifact message id = %d, want unchanged 99", foreignUserArtifact.MessageID) + } + var ambiguousArtifact model.ChatContextRecord + if err = db.Where("source_id = ?", "ambiguous_call").First(&ambiguousArtifact).Error; err != nil { + t.Fatalf("load ambiguous artifact: %v", err) + } + if ambiguousArtifact.MessageID != ambiguousUser.ID { + t.Fatalf("ambiguous artifact message id = %d, want unchanged %d", ambiguousArtifact.MessageID, ambiguousUser.ID) + } + var snapshot model.ChatContextRecord + if err = db.Where("record_type = ?", "snapshot").First(&snapshot).Error; err != nil { + t.Fatalf("load snapshot: %v", err) + } + if snapshot.MessageID != userMessage.ID { + t.Fatalf("snapshot message id = %d, want unchanged %d", snapshot.MessageID, userMessage.ID) + } +} + func TestSeedBillingCatalogBindsDefaultPermissionGroup(t *testing.T) { db := openSchemaTestDB(t) if err := SeedPermissionGroups(db); err != nil { diff --git a/backend/internal/repository/conversation_context_artifact.go b/backend/internal/repository/conversation_context_artifact.go index ad71fe957..d6939496d 100644 --- a/backend/internal/repository/conversation_context_artifact.go +++ b/backend/internal/repository/conversation_context_artifact.go @@ -12,6 +12,13 @@ type ContextArtifactRepository interface { CreateContextArtifacts(ctx context.Context, items []domainconversation.ContextArtifact) error GetContextArtifactByIDForUser(ctx context.Context, userID uint, artifactID uint) (*domainconversation.ContextArtifact, error) ListContextArtifactsByMessage(ctx context.Context, conversationID uint, messageID uint) ([]domainconversation.ContextArtifact, error) - ListRecentContextArtifacts(ctx context.Context, conversationID uint, kinds []domainconversation.ContextArtifactKind, limit int) ([]domainconversation.ContextArtifact, error) + ListRecentContextArtifacts(ctx context.Context, filter ContextArtifactListFilter) ([]domainconversation.ContextArtifact, error) DeleteExpiredContextArtifacts(ctx context.Context, before time.Time, limit int) (int64, error) } + +// ContextArtifactListFilter 描述当前分支内的历史证据召回范围。 +type ContextArtifactListFilter struct { + Scope HistoricalMessageScope + Kinds []domainconversation.ContextArtifactKind + Limit int +} diff --git a/backend/internal/repository/conversation_core.go b/backend/internal/repository/conversation_core.go index 54fc3eff2..d1f4e2d48 100644 --- a/backend/internal/repository/conversation_core.go +++ b/backend/internal/repository/conversation_core.go @@ -146,7 +146,29 @@ type ConversationEventLogListFilter struct { // MessageEmbeddingRepository 封装消息历史向量存储与检索能力。 type MessageEmbeddingRepository interface { UpsertMessageChunks(ctx context.Context, chunks []domainconversation.MessageChunk, embeddings [][]float32) error - SearchMessageChunks(ctx context.Context, conversationID uint, userID uint, queryEmbedding []float32, topK int, minSimilarity float64) ([]domainconversation.MessageChunk, error) + SearchMessageChunks(ctx context.Context, input MessageChunkSearchInput) ([]domainconversation.MessageChunk, error) +} + +// HistoricalMessageScope 描述当前消息所在分支中可参与历史召回的祖先范围。 +// ExcludeThroughMessageID 用于快照场景：排除该边界消息及其全部祖先。 +type HistoricalMessageScope struct { + ConversationID uint + UserID uint + LeafMessageID uint + ExcludeThroughMessageID uint +} + +// Valid 表示分支范围包含可验证的会话和叶消息锚点。 +func (scope HistoricalMessageScope) Valid() bool { + return scope.ConversationID > 0 && scope.UserID > 0 && scope.LeafMessageID > 0 +} + +// MessageChunkSearchInput 描述当前分支内的历史消息语义检索。 +type MessageChunkSearchInput struct { + Scope HistoricalMessageScope + QueryEmbedding []float32 + TopK int + MinSimilarity float64 } // CompactRepository 封装上下文压缩快照能力。