From 69d8f2f50d529dd76d9681551a8952ebc29ae3ed Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Fri, 11 Sep 2026 13:16:42 +0800 Subject: [PATCH] feat(api): return user_message_id in SSE turn_ack and done Co-authored-by: Cursor --- internal/chat/session.go | 17 ++++++++++++++--- internal/stream/stream.go | 26 +++++++++++++++++++++++++- internal/stream/types.go | 13 +++++++++++++ 3 files changed, 52 insertions(+), 4 deletions(-) diff --git a/internal/chat/session.go b/internal/chat/session.go index 31afbdc0..40a3be07 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -230,8 +230,19 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str } s.Messages = append(s.Messages, userMsg) + var userMsgID int64 if s.CtxMgr != nil { - _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, s.UserID, "user", userInput, "", "", nil) + var recErr error + userMsgID, recErr = s.CtxMgr.RecordUserMessageWithID(ctx, s.ID, s.UserID, "user", userInput, "", "", nil) + if recErr != nil { + if streamer != nil { + streamer.EmitErrorInfo(stream.ClassifyError(recErr, "persist_user_message", stream.RequestIDFromContext(ctx), s.ID, "")) + } + return nil, fmt.Errorf("record user message: %w", recErr) + } + if streamer != nil && userMsgID > 0 { + streamer.EmitTurnAck("", userMsgID, "processing", false) + } } // 偏好画像同步 @@ -416,7 +427,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str if coordRes.AdvisorBrief != nil { streamer.EmitBriefing(coordRes.AdvisorBrief) } - streamer.EmitDoneWithExport(s.ID, time.Since(start), asstMsgID, evalCap.ContentRevision, evalCap.ContentSha256, evalCap) + streamer.EmitDoneWithExport(s.ID, time.Since(start), userMsgID, asstMsgID, evalCap.ContentRevision, evalCap.ContentSha256, evalCap) } } else { // 无 CtxMgr 时的降级 @@ -436,7 +447,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str if coordRes.AdvisorBrief != nil { streamer.EmitBriefing(coordRes.AdvisorBrief) } - streamer.EmitDoneWithExport(s.ID, time.Since(start), 0, evalCap.ContentRevision, evalCap.ContentSha256, evalCap) + streamer.EmitDoneWithExport(s.ID, time.Since(start), 0, 0, evalCap.ContentRevision, evalCap.ContentSha256, evalCap) } } diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 08a7bb52..27d1fd4d 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -242,8 +242,29 @@ func (s *Streamer) EmitDoneWithWorkspace(sessionID, workspaceID string, duration }) } +// EmitTurnAck 发送回合确认事件（用户消息落库后立即回传） +func (s *Streamer) EmitTurnAck(clientTurnID string, userMessageID int64, status string, duplicate bool) { + if s == nil { + return + } + s.Emit(StreamEvent{ + Type: EventTurnAck, + Turn: &TurnAckPayload{ + ClientTurnID: clientTurnID, + UserMessageID: userMessageID, + Status: status, + Duplicate: duplicate, + }, + }) +} + // EmitDoneWithExport 发送流结束事件并携带落库消息 ID 与权威导出能力 -func (s *Streamer) EmitDoneWithExport(sessionID string, duration time.Duration, messageID int64, revision int, sha256 string, capability any) { +func (s *Streamer) EmitDoneWithExport(sessionID string, duration time.Duration, userMessageID, messageID int64, revision int, sha256 string, capability any) { + s.EmitDoneWithTurnExport(sessionID, "", duration, userMessageID, messageID, revision, sha256, capability) +} + +// EmitDoneWithTurnExport 发送流结束事件并携带回合与用户/助手消息 ID +func (s *Streamer) EmitDoneWithTurnExport(sessionID, clientTurnID string, duration time.Duration, userMessageID, messageID int64, revision int, sha256 string, capability any) { if s == nil { return } @@ -251,9 +272,12 @@ func (s *Streamer) EmitDoneWithExport(sessionID string, duration time.Duration, Type: EventDone, Done: &DonePayload{ SessionID: sessionID, + ClientTurnID: clientTurnID, DurationMs: duration.Milliseconds(), TotalEvents: int(atomic.LoadInt64(&s.eventCount)), + UserMessageID: userMessageID, MessageID: messageID, + Status: "completed", ContentRevision: revision, ContentSha256: sha256, ExportCapability: capability, diff --git a/internal/stream/types.go b/internal/stream/types.go index 9aacae37..b92a0cf8 100644 --- a/internal/stream/types.go +++ b/internal/stream/types.go @@ -8,6 +8,7 @@ import ( type EventType string const ( + EventTurnAck EventType = "turn_ack" // 回合确认：用户消息已落库，携带 user_message_id EventPhase EventType = "phase" // 阶段性状态跃迁（组建小队、法规筛选、初稿撰写、门禁审核） EventStatus EventType = "status" // 意图识别结果、成熟度状态机更新 EventSummary EventType = "summary" // 提炼的核心进展摘要（低认知负荷） @@ -82,13 +83,24 @@ type DeliverablePayload struct { QualityScore float64 `json:"quality_score,omitempty"` } +// TurnAckPayload 回合确认载荷（用户消息落库后立即回传） +type TurnAckPayload struct { + ClientTurnID string `json:"client_turn_id,omitempty"` + UserMessageID int64 `json:"user_message_id,omitempty"` + Status string `json:"status,omitempty"` // processing | completed | duplicate + Duplicate bool `json:"duplicate,omitempty"` +} + // DonePayload 流结束载荷 type DonePayload struct { SessionID string `json:"session_id,omitempty"` + ClientTurnID string `json:"client_turn_id,omitempty"` DurationMs int64 `json:"duration_ms"` TotalEvents int `json:"total_events"` WorkspaceID string `json:"workspace_id,omitempty"` // P2: 任务工作区目录（供前端「继续任务」续跑） + UserMessageID int64 `json:"user_message_id,omitempty"` MessageID int64 `json:"message_id,omitempty"` + Status string `json:"status,omitempty"` ContentRevision int `json:"content_revision,omitempty"` ContentSha256 string `json:"content_sha256,omitempty"` ExportCapability any `json:"export_capability,omitempty"` @@ -131,5 +143,6 @@ type StreamEvent struct { Governance *GovernancePayload `json:"governance,omitempty"` Error string `json:"error,omitempty"` ErrorInfo *ErrorInfo `json:"error_info,omitempty"` + Turn *TurnAckPayload `json:"turn,omitempty"` Done *DonePayload `json:"done,omitempty"` }