weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): gate chat stream on chat_turns with session serial execution

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
2026-09-11 13:18:16 +08:00
co-authored by Cursor
parent 9202ca7bf3
commit 1438af18f7
4 changed files with 190 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/chat/session.go
+48 -7
View File
@@ -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 时的降级
internal/server/chat_turn_api.go
+45
View File
@@ -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,
})
}
internal/server/server.go
+1
View File
@@ -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)
internal/server/sse.go
+96 -2
View File
@@ -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, "")