diff --git a/internal/chat/session.go b/internal/chat/session.go index 40a3be07..4b061838 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "chat-agent-system/internal/coordinator" @@ -49,6 +50,19 @@ type Session struct { 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 创建新对话会话 @@ -203,16 +217,21 @@ func (s *Session) syncRecalledMemories(ctx context.Context, userInput string, re // SendMessage 处理一轮用户输入 (标准阻塞模式) func (s *Session) SendMessage(ctx context.Context, userInput string) (*ChatResponse, error) { streamer := stream.FromContext(ctx) - return s.sendMessageInternal(ctx, userInput, streamer) + 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) { - ctxWithS, streamer := stream.WithListener(ctx, includeDetails, listener) - return s.sendMessageInternal(ctxWithS, userInput, streamer) + return s.SendMessageStreamTurn(ctx, userInput, listener, includeDetails, nil) } -func (s *Session) sendMessageInternal(ctx context.Context, userInput string, streamer *stream.Streamer) (*ChatResponse, error) { +// 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 { @@ -231,17 +250,32 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str s.Messages = append(s.Messages, userMsg) var userMsgID int64 + clientTurnID := "" + if turn != nil { + clientTurnID = turn.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, "", "", nil) + 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("", userMsgID, "processing", false) + streamer.EmitTurnAck(clientTurnID, userMsgID, "processing", false) } } @@ -398,6 +432,10 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str 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) @@ -427,7 +465,10 @@ 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), userMsgID, asstMsgID, evalCap.ContentRevision, evalCap.ContentSha256, evalCap) + 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) } } else { // 无 CtxMgr 时的降级 diff --git a/internal/server/chat_turn_api.go b/internal/server/chat_turn_api.go new file mode 100644 index 00000000..ac775705 --- /dev/null +++ b/internal/server/chat_turn_api.go @@ -0,0 +1,45 @@ +package server + +import ( + "encoding/json" + "errors" + "net/http" + + "chat-agent-system/internal/ctxmgr" +) + +// handleGetChatTurn 查询回合状态（断线续接） +func (s *Server) handleGetChatTurn(w http.ResponseWriter, r *http.Request) { + clientTurnID := r.PathValue("client_turn_id") + sessionID := r.URL.Query().Get("session_id") + if clientTurnID == "" || sessionID == "" { + http.Error(w, "client_turn_id and session_id required", http.StatusBadRequest) + return + } + + ctxMgr := s.svc.ContextManager() + if ctxMgr == nil { + http.Error(w, "context manager unavailable", http.StatusServiceUnavailable) + return + } + + rec, err := ctxMgr.GetChatTurn(r.Context(), sessionID, clientTurnID) + if errors.Is(err, ctxmgr.ErrChatTurnNotFound) { + http.Error(w, "turn not found", http.StatusNotFound) + return + } + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "session_id": rec.SessionID, + "client_turn_id": rec.ClientTurnID, + "status": rec.Status, + "user_message_id": rec.UserMessageID, + "assistant_message_id": rec.AssistantMessageID, + "attempt_count": rec.AttemptCount, + }) +} diff --git a/internal/server/server.go b/internal/server/server.go index b678fe96..3aad90fa 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -105,6 +105,7 @@ func (s *Server) Routes() http.Handler { // 4. 对话协同服务 (核心 SSE 流式长连接) mux.HandleFunc("POST /api/v1/chat/stream", s.handleChatStream) mux.HandleFunc("GET /api/v1/chat/stream", s.handleChatStream) + mux.HandleFunc("GET /api/v1/chat/turns/{client_turn_id}", s.handleGetChatTurn) // 5. 独立任务 / 阶段执行 mux.HandleFunc("POST /api/v1/tasks/run", s.handleRunTask) diff --git a/internal/server/sse.go b/internal/server/sse.go index ee817867..ad54e44e 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -2,6 +2,7 @@ package server import ( "encoding/json" + "errors" "fmt" "net/http" "strconv" @@ -9,6 +10,8 @@ import ( "time" "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/chat" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" ) @@ -36,6 +39,46 @@ type StreamChatRequest struct { UserRole string `json:"user_role,omitempty"` UserID int64 `json:"user_id,omitempty"` ProjectID int64 `json:"project_id,omitempty"` + ClientTurnID string `json:"client_turn_id,omitempty"` +} + +func writeSSE(w http.ResponseWriter, flusher http.Flusher, ev stream.StreamEvent) { + data, err := json.Marshal(ev) + if err != nil { + return + } + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + flusher.Flush() +} + +func replayExistingTurn(w http.ResponseWriter, flusher http.Flusher, rec *ctxmgr.ChatTurnRecord) { + status := string(rec.Status) + duplicate := true + if rec.UserMessageID > 0 { + writeSSE(w, flusher, stream.StreamEvent{ + Type: stream.EventTurnAck, + Timestamp: time.Now(), + Turn: &stream.TurnAckPayload{ + ClientTurnID: rec.ClientTurnID, + UserMessageID: rec.UserMessageID, + Status: status, + Duplicate: duplicate, + }, + }) + } + if rec.Status == ctxmgr.TurnStatusCompleted && rec.AssistantMessageID > 0 { + writeSSE(w, flusher, stream.StreamEvent{ + Type: stream.EventDone, + Timestamp: time.Now(), + Done: &stream.DonePayload{ + SessionID: rec.SessionID, + ClientTurnID: rec.ClientTurnID, + UserMessageID: rec.UserMessageID, + MessageID: rec.AssistantMessageID, + Status: "completed", + }, + }) + } } func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { @@ -134,8 +177,59 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { flusher.Flush() } - // 4. 发起流式执行 - _, err := sess.SendMessageStream(ctx, req.Message, listener, req.IncludeDetails) + clientTurnID := req.ClientTurnID + if clientTurnID == "" { + clientTurnID = r.Header.Get("Idempotency-Key") + } + + var turnOpts *chat.TurnOptions + if clientTurnID != "" { + turnOpts = &chat.TurnOptions{ClientTurnID: clientTurnID} + if sess.CtxMgr != nil { + bodyHash := ctxmgr.HashTurnBody(req.SessionID, req.Message) + acq, acqErr := sess.CtxMgr.AcquireChatTurn(ctx, req.SessionID, clientTurnID, bodyHash, requestID, sess.UserID, sess.ProjectID) + if errors.Is(acqErr, ctxmgr.ErrIdempotencyKeyReused) { + writeMu.Lock() + info := stream.ClassifyError(acqErr, "idempotency", requestID, sess.ID, "") + info.Code = "IDEMPOTENCY_KEY_REUSED" + writeSSE(w, flusher, stream.StreamEvent{ + Type: stream.EventError, + Timestamp: time.Now(), + Error: info.UserMessage, + ErrorInfo: &info, + }) + writeMu.Unlock() + return + } + if acqErr != nil { + writeMu.Lock() + info := stream.ClassifyError(acqErr, "turn_acquire", requestID, sess.ID, "") + writeSSE(w, flusher, stream.StreamEvent{ + Type: stream.EventError, + Timestamp: time.Now(), + Error: info.UserMessage, + ErrorInfo: &info, + }) + writeMu.Unlock() + return + } + if acq != nil && !acq.WonLease { + writeMu.Lock() + replayExistingTurn(w, flusher, acq.Record) + writeMu.Unlock() + return + } + } + } + + // 4. 同 session 串行执行后发起流式对话 + var err error + execErr := sess.WithExecLock(func() error { + var sendErr error + _, sendErr = sess.SendMessageStreamTurn(ctx, req.Message, listener, req.IncludeDetails, turnOpts) + return sendErr + }) + err = execErr if err != nil { writeMu.Lock() info := stream.ClassifyError(err, "chat_stream", requestID, sess.ID, "")