feat(api): return user_message_id in SSE turn_ack and done

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

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)
 		}
 	}
 

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,

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"`
 }