Files
Agent-System/internal/chat/session.go
T
WeiCHENandCursor 28faa4932d fix(agent): Chat Agent 全链路 P0-P5 整改
封堵鉴权旁路与导出 IDOR,DocGov 改服务端身份;对话层 PendingAction 接真实 docgov/taskrun;修复流式租约假死与会话级任务流;加固配置/doc_id/NULL;补齐 slog 与测试;前端 ErrorBoundary 与受控文档重试。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-11 18:22:45 +08:00

521 lines
18 KiB
Go

package chat
import (
"context"
"fmt"
"strings"
"sync"
"time"
"chat-agent-system/internal/coordinator"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/intent"
"chat-agent-system/internal/replyexport"
"chat-agent-system/internal/squad"
"chat-agent-system/internal/stream"
)
// Message 历史消息结构
type Message struct {
ID string `json:"id"`
Role string `json:"role"` // "user" | "assistant"
Content string `json:"content"`
Intent intent.IntentType `json:"intent,omitempty"`
Branch intent.ResolutionBranch `json:"branch,omitempty"`
AdvisorBrief *squad.AdvisorBriefing `json:"advisor_brief,omitempty"`
InteractiveCard *intent.InteractiveCardPayload `json:"interactive_card,omitempty"`
Timestamp time.Time `json:"timestamp"`
}
// ChatResponse 会话单轮响应
type ChatResponse struct {
ReplyText string `json:"reply_text"`
Branch intent.ResolutionBranch `json:"branch"`
IntentResult *intent.IntentRecognitionResult `json:"intent_result"`
CoordinationResult *coordinator.CoordinationResult `json:"coordination_result"`
AdvisorBrief *squad.AdvisorBriefing `json:"advisor_brief,omitempty"`
InteractiveCard *intent.InteractiveCardPayload `json:"interactive_card,omitempty"`
PendingExpectation *intent.PendingExpectation `json:"pending_expectation,omitempty"`
}
// Session 多轮对话与协同会话管理器
type Session struct {
ID string `json:"id"`
UserID int64 `json:"user_id"`
ProjectID int64 `json:"project_id"`
UserCtx *coordinator.UserProjectContext `json:"user_ctx"`
Recognizer intent.Recognizer `json:"-"`
Coordinator *coordinator.CoordinatorEngine `json:"-"`
CtxMgr *ctxmgr.ContextManager `json:"-"`
Actions *ActionServices `json:"-"`
Messages []Message `json:"messages"`
CreatedAt time.Time `json:"created_at"`
LastAccessedAt time.Time `json:"last_accessed_at"`
execMu sync.Mutex `json:"-"`
}
// TurnOptions 客户端回合幂等选项
type TurnOptions struct {
ClientTurnID string
}
// WithExecLock 同 session 内串行执行
func (s *Session) WithExecLock(fn func() error) error {
s.execMu.Lock()
defer s.execMu.Unlock()
return fn()
}
// NewSession 创建新对话会话
func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.Recognizer, coord *coordinator.CoordinatorEngine) *Session {
if rec == nil {
rec = intent.NewRecognizer()
}
if id == "" {
id = fmt.Sprintf("sess_%d", time.Now().Unix())
}
// 若上下文尚未挂起预期且当前已具有阶段,则初始化默认选项草稿
if userCtx != nil && userCtx.PendingExpectation == nil && userCtx.Project.CurrentStage != "" {
userCtx.PendingExpectation = coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
}
sess := &Session{
ID: id,
UserID: 1,
ProjectID: 1,
UserCtx: userCtx,
Recognizer: rec,
Coordinator: coord,
Messages: make([]Message, 0),
CreatedAt: time.Now(),
LastAccessedAt: time.Now(),
}
sess.SyncRecentDialogue()
return sess
}
// SyncRecentDialogue 使用双预算滑动窗口将消息转换为 DialogueTurn 同步到 UserCtx.RecentDialogue
func (s *Session) SyncRecentDialogue() {
if s.UserCtx == nil {
return
}
records := s.toChatRecords()
cfg := ctxmgr.DefaultWindowConfig()
window, early, truncated := ctxmgr.AssembleWindow(records, cfg)
turns := make([]coordinator.DialogueTurn, 0, len(window))
for _, m := range window {
if strings.TrimSpace(m.Content) == "" {
continue
}
turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content})
}
s.UserCtx.RecentDialogue = turns
// 若发生了截断且存在早期消息,同步计算或加载滚动摘要
if truncated && len(early) > 0 && s.CtxMgr != nil {
s.syncDialogueSummary(context.Background(), early)
}
}
func (s *Session) toChatRecords() []ctxmgr.ChatMessageRecord {
records := make([]ctxmgr.ChatMessageRecord, 0, len(s.Messages))
for i, m := range s.Messages {
records = append(records, ctxmgr.ChatMessageRecord{
ID: int64(i + 1),
SessionID: s.ID,
Role: m.Role,
Content: m.Content,
Intent: string(m.Intent),
Branch: string(m.Branch),
CreatedAt: m.Timestamp,
})
}
return records
}
func (s *Session) syncDialogueSummary(ctx context.Context, early []ctxmgr.ChatMessageRecord) {
if s.CtxMgr == nil || len(early) == 0 {
return
}
oldSum, _ := s.CtxMgr.GetChatSummary(ctx, s.ID)
if oldSum == nil || len(early) > oldSum.CoveredCount {
newSum, err := s.CtxMgr.GenerateRollingSummary(ctx, s.ID, oldSum, early)
if err == nil && newSum != nil && s.UserCtx != nil {
s.UserCtx.DialogueSummary = newSum.RenderPromptSnippet()
}
} else if oldSum != nil && s.UserCtx != nil {
s.UserCtx.DialogueSummary = oldSum.RenderPromptSnippet()
}
}
// syncPersonalizedGuidance 将用户偏好画像编译为 Prompt 指导语并注入 UserCtx
func (s *Session) syncPersonalizedGuidance(ctx context.Context, userInput string) {
if s.UserCtx == nil || s.CtxMgr == nil || s.UserID <= 0 {
return
}
// 首次注入,或用户显式表达定制要求("记住..." / "以后请..." / "默认都..."),或周期性刷新(每10条消息)
needsRefresh := s.UserCtx.PersonalizedGuidance == "" ||
strings.Contains(userInput, "记住") ||
strings.Contains(userInput, "以后请") ||
strings.Contains(userInput, "默认都") ||
len(s.Messages)%10 == 0
if !needsRefresh {
return
}
if guidance, err := s.CtxMgr.CompilePromptGuidance(ctx, s.UserID); err == nil && guidance != "" {
s.UserCtx.PersonalizedGuidance = guidance
}
}
// shouldRetrieveMemory 判定当轮意图与输入是否需要按需触发历史对话记忆语义召回
func (s *Session) shouldRetrieveMemory(userInput string, recRes *intent.IntentRecognitionResult) bool {
if recRes == nil {
return false
}
// 分支 3:模糊咨询/分忧引导,需要召回历史记忆辅助 LLM 理解语境
if recRes.Branch == intent.BranchAmbiguousFallback {
return true
}
// 用户包含显式历史追溯意图关键词,触发召回
keywords := []string{"历史", "之前", "上次", "讨论过", "记录", "查一下", "提过", "曾", "往期", "回顾"}
for _, kw := range keywords {
if strings.Contains(userInput, kw) {
return true
}
}
if recRes.TopIntent != nil {
switch recRes.TopIntent.Intent {
case intent.IntentQueryProgress, intent.IntentQuerySimilarCases, intent.IntentQueryRegulations:
return true
}
}
return false
}
// syncRecalledMemories 按需执行历史对话切片语义检索并格式化注入 UserCtx
func (s *Session) syncRecalledMemories(ctx context.Context, userInput string, recRes *intent.IntentRecognitionResult) {
if s.CtxMgr == nil || s.UserCtx == nil || !s.shouldRetrieveMemory(userInput, recRes) {
return
}
opts := ctxmgr.ChatMemorySearchOptions{
Query: userInput,
UserID: s.UserID,
ProjectID: s.ProjectID,
ExcludeSessionID: s.ID,
TopK: 3,
MinScore: 0.20,
}
items, err := s.CtxMgr.SearchChatMemory(ctx, opts)
if err == nil && len(items) > 0 {
s.UserCtx.RecalledMemories = items
s.UserCtx.RecalledMemorySnippet = ctxmgr.FormatRecalledMemorySnippet(items, s.ID)
}
}
// SendMessage 处理一轮用户输入 (标准阻塞模式)
func (s *Session) SendMessage(ctx context.Context, userInput string) (*ChatResponse, error) {
streamer := stream.FromContext(ctx)
return s.sendMessageInternal(ctx, userInput, streamer, nil)
}
// SendMessageStream 处理一轮用户输入并实时流式推送事件 (支持 SSE / 终端流式监听)
func (s *Session) SendMessageStream(ctx context.Context, userInput string, listener stream.Listener, includeDetails bool) (*ChatResponse, error) {
return s.SendMessageStreamTurn(ctx, userInput, listener, includeDetails, nil)
}
// SendMessageStreamTurn 带 client_turn_id 的流式对话
func (s *Session) SendMessageStreamTurn(ctx context.Context, userInput string, listener stream.Listener, includeDetails bool, turn *TurnOptions) (*ChatResponse, error) {
ctxWithS, streamer := stream.WithListener(ctx, includeDetails, listener)
return s.sendMessageInternal(ctxWithS, userInput, streamer, turn)
}
func (s *Session) sendMessageInternal(ctx context.Context, userInput string, streamer *stream.Streamer, turn *TurnOptions) (*ChatResponse, error) {
start := time.Now()
s.LastAccessedAt = time.Now()
if streamer != nil {
streamer.EmitPhase("intent_recognizing", "正在研判您的意图与项目上下文...", "coordinator", 0, 0, 0)
}
// 0. 同步已有前序对话到 UserCtx,供意图快照、路由评分与对话 LLM 感知历史轮次
s.SyncRecentDialogue()
userMsg := Message{
ID: fmt.Sprintf("msg_%d_u", time.Now().UnixNano()),
Role: "user",
Content: userInput,
Timestamp: time.Now(),
}
s.Messages = append(s.Messages, userMsg)
var userMsgID int64
clientTurnID := ""
if turn != nil {
clientTurnID = turn.ClientTurnID
}
turnCompleted := false
if clientTurnID != "" && s.CtxMgr != nil {
defer func() {
if !turnCompleted {
_ = s.CtxMgr.FailChatTurn(ctx, s.ID, clientTurnID)
}
}()
}
if s.CtxMgr != nil {
userMeta := map[string]any{}
if clientTurnID != "" {
userMeta["client_turn_id"] = clientTurnID
userMeta["client_id"] = clientTurnID + ":user"
}
var recErr error
userMsgID, recErr = s.CtxMgr.RecordUserMessageWithID(ctx, s.ID, s.UserID, "user", userInput, "", "", userMeta)
if recErr != nil {
if streamer != nil {
streamer.EmitErrorInfo(stream.ClassifyError(recErr, "persist_user_message", stream.RequestIDFromContext(ctx), s.ID, ""))
}
if clientTurnID != "" {
_ = s.CtxMgr.FailChatTurn(ctx, s.ID, clientTurnID)
}
return nil, fmt.Errorf("record user message: %w", recErr)
}
if clientTurnID != "" {
_ = s.CtxMgr.UpdateChatTurnUserMessage(ctx, s.ID, clientTurnID, userMsgID)
}
if streamer != nil && userMsgID > 0 {
streamer.EmitTurnAck(clientTurnID, userMsgID, "processing", false)
}
}
// 偏好画像同步
s.syncPersonalizedGuidance(ctx, userInput)
// 1. 意图识别 (L1 对话层,带非阻塞桌角备忘快照)
var snap *intent.ContextSnapshot
if s.UserCtx != nil {
snap = s.UserCtx.ToSnapshot()
}
recRes, err := s.Recognizer.Recognize(ctx, userInput, snap)
if err != nil {
if streamer != nil {
streamer.EmitErrorInfo(stream.ClassifyError(err, "intent_recognizing", stream.RequestIDFromContext(ctx), s.ID, ""))
}
return nil, fmt.Errorf("intent recognition failed: %w", err)
}
if streamer != nil && recRes.TopIntent != nil {
streamer.EmitStatus(
string(recRes.TopIntent.Intent),
string(recRes.Branch),
string(s.UserCtx.Project.CurrentStage),
fmt.Sprintf("识别分支: %s (置信度: %.0f%%)", recRes.Branch, recRes.TopIntent.Confidence*100),
recRes.TopIntent.FilledSlots,
)
}
// 1.5 规则查询改写 (P3 阶段:基于当轮输入前的历史状态机进行省略补全 + 代词消解 + 主题延续)
searchQuery := userInput
if s.UserCtx != nil {
rewriteRes := coordinator.RewriteQuery(
userInput,
s.UserCtx.TaskState,
s.UserCtx.RecentDialogue,
s.UserCtx.Project,
)
s.UserCtx.RewrittenQuery = rewriteRes.Rewritten
s.UserCtx.RewriteRules = rewriteRes.AppliedRules
if rewriteRes.SearchQuery != "" {
searchQuery = rewriteRes.SearchQuery
}
}
// 1.6 刷新会话级任务状态机 (DST,合入改写后完整问句所携带的实体与锚点)
if s.UserCtx != nil {
if s.UserCtx.TaskState == nil {
s.UserCtx.TaskState = ctxmgr.NewConversationTaskState()
}
intentName := ""
var filledSlots map[string]string
if recRes.TopIntent != nil {
intentName = string(recRes.TopIntent.Intent)
filledSlots = recRes.TopIntent.FilledSlots
}
effectiveInput := userInput
if s.UserCtx.RewrittenQuery != "" {
effectiveInput = s.UserCtx.RewrittenQuery
}
s.UserCtx.TaskState.UpdateFromTurn(
effectiveInput,
intentName,
filledSlots,
string(recRes.Branch),
string(s.UserCtx.Project.CurrentStage),
)
}
// 1.8 按需语义召回历史对话切片 (P2 阶段:优先使用改写增强后的 searchQuery)
s.syncRecalledMemories(ctx, searchQuery, recRes)
// 2. 协调调度与校验 (L2 协调层)
if streamer != nil {
streamer.EmitPhase("dialogue_generating", "正在整理回复...", "assistant", 0, 0, 40)
}
coordRes, err := s.Coordinator.ProcessIntent(ctx, recRes, s.UserCtx)
if err != nil {
if streamer != nil {
streamer.EmitErrorInfo(stream.ClassifyError(err, "dialogue_generating", stream.RequestIDFromContext(ctx), s.ID, ""))
}
return nil, fmt.Errorf("coordination execution failed: %w", err)
}
if coordRes.PendingAction != nil {
execResult, execErr := s.executeAction(ctx, coordRes.PendingAction, s.UserCtx)
coordRes.NaturalReply = renderActionReply(coordRes.PendingAction, execResult, execErr)
coordRes.PendingAction = nil
}
// 3. 更新上下文与挂起预期生命周期
if coordRes.UpdatedContext != nil {
if coordRes.UpdatedContext.TaskState == nil && s.UserCtx != nil {
coordRes.UpdatedContext.TaskState = s.UserCtx.TaskState
}
if coordRes.UpdatedContext.DialogueSummary == "" && s.UserCtx != nil {
coordRes.UpdatedContext.DialogueSummary = s.UserCtx.DialogueSummary
}
if len(coordRes.UpdatedContext.RecalledMemories) == 0 && s.UserCtx != nil {
coordRes.UpdatedContext.RecalledMemories = s.UserCtx.RecalledMemories
coordRes.UpdatedContext.RecalledMemorySnippet = s.UserCtx.RecalledMemorySnippet
}
if coordRes.UpdatedContext.RewrittenQuery == "" && s.UserCtx != nil {
coordRes.UpdatedContext.RewrittenQuery = s.UserCtx.RewrittenQuery
coordRes.UpdatedContext.RewriteRules = s.UserCtx.RewriteRules
}
s.UserCtx = coordRes.UpdatedContext
}
if coordRes.ClearExpectation {
s.UserCtx.PendingExpectation = nil
} else if coordRes.NextExpectation != nil {
s.UserCtx.PendingExpectation = coordRes.NextExpectation
} else if recRes.Branch == intent.BranchExpectationHit {
s.UserCtx.PendingExpectation = nil
}
// 若为 BranchNewIntentPreempt 且未产生新预期,原挂起预期在焦点底静默保留,不阻拦用户
asstMsg := Message{
ID: fmt.Sprintf("msg_%d_a", time.Now().UnixNano()),
Role: "assistant",
Content: coordRes.NaturalReply,
Intent: coordRes.Intent,
Branch: recRes.Branch,
AdvisorBrief: coordRes.AdvisorBrief,
InteractiveCard: coordRes.InteractiveCard,
Timestamp: time.Now(),
}
s.Messages = append(s.Messages, asstMsg)
if s.CtxMgr != nil {
replySrc := coordRes.ReplySource
if replySrc == "" {
replySrc = coordinator.ReplySourceTemplate
}
asstMeta := map[string]any{
"reply_source": replySrc,
}
if recRes.TopIntent != nil {
asstMeta["turn_intent"] = string(recRes.TopIntent.Intent)
}
isMilestone := false
if recRes.TopIntent != nil && len(recRes.TopIntent.FilledSlots) > 0 {
isMilestone = true
asstMeta["milestone"] = true
}
if coordRes.Intent == intent.IntentConfirmAction || coordRes.Intent == intent.IntentExecuteStageAdvance || coordRes.Intent == intent.IntentTriggerAdhocTask {
isMilestone = true
asstMeta["milestone"] = true
}
evalKind, evalCap := replyexport.EvaluatePolicy(replyexport.PolicyOptions{
Role: "assistant",
Status: "completed",
Content: coordRes.NaturalReply,
HasInteractiveCard: coordRes.InteractiveCard != nil,
Revision: 1,
})
asstMeta["content_kind"] = string(evalKind)
asstMeta["export_policy"] = string(evalCap.Policy)
asstMeta["content_revision"] = evalCap.ContentRevision
asstMeta["content_sha256"] = evalCap.ContentSha256
if clientTurnID != "" {
asstMeta["client_turn_id"] = clientTurnID
asstMeta["client_id"] = clientTurnID + ":assistant"
}
asstMsgID, _ := s.CtxMgr.RecordUserMessageWithID(ctx, s.ID, 0, "assistant", coordRes.NaturalReply, string(coordRes.Intent), string(recRes.Branch), asstMeta)
// 异步记录问答切片到向量记忆库 (P2 检索增强,不阻塞流式或同步答复)
sessID := s.ID
turnNo := len(s.Messages) / 2
uid := s.UserID
pid := s.ProjectID
q := userInput
ans := coordRes.NaturalReply
iName := string(coordRes.Intent)
bName := string(recRes.Branch)
go func() {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_, _ = s.CtxMgr.RecordTurnMemory(bgCtx, sessID, turnNo, uid, pid, q, ans, iName, bName, isMilestone)
}()
// 本轮全部处理完毕后,再次同步,保持 UserCtx 与 Messages 一致
s.SyncRecentDialogue()
// 4. 流式推送参谋自然语言回复与决策锦囊
if streamer != nil {
if coordRes.NaturalReply != "" {
streamer.EmitDelta(coordRes.NaturalReply)
}
if coordRes.AdvisorBrief != nil {
streamer.EmitBriefing(coordRes.AdvisorBrief)
}
streamer.EmitDoneWithTurnExport(s.ID, clientTurnID, time.Since(start), userMsgID, asstMsgID, evalCap.ContentRevision, evalCap.ContentSha256, evalCap)
}
if clientTurnID != "" {
_ = s.CtxMgr.CompleteChatTurn(ctx, s.ID, clientTurnID, asstMsgID)
turnCompleted = true
}
} else {
// 无 CtxMgr 时的降级
evalKind, evalCap := replyexport.EvaluatePolicy(replyexport.PolicyOptions{
Role: "assistant",
Status: "completed",
Content: coordRes.NaturalReply,
HasInteractiveCard: coordRes.InteractiveCard != nil,
Revision: 1,
})
_ = evalKind
s.SyncRecentDialogue()
if streamer != nil {
if coordRes.NaturalReply != "" {
streamer.EmitDelta(coordRes.NaturalReply)
}
if coordRes.AdvisorBrief != nil {
streamer.EmitBriefing(coordRes.AdvisorBrief)
}
streamer.EmitDoneWithExport(s.ID, time.Since(start), 0, 0, evalCap.ContentRevision, evalCap.ContentSha256, evalCap)
}
}
return &ChatResponse{
ReplyText: coordRes.NaturalReply,
Branch: recRes.Branch,
IntentResult: recRes,
CoordinationResult: coordRes,
AdvisorBrief: coordRes.AdvisorBrief,
InteractiveCard: coordRes.InteractiveCard,
PendingExpectation: s.UserCtx.PendingExpectation,
}, nil
}