diff --git a/backend/internal/ws/handler.go b/backend/internal/ws/handler.go index 4c77ff6..52f90e5 100644 --- a/backend/internal/ws/handler.go +++ b/backend/internal/ws/handler.go @@ -170,19 +170,43 @@ func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orche logger.Log.Infow("query received", "session", sessionID, "request", msg.RequestID) // 刷新会话 TTL - if err := sessionMgr.Touch(context.Background(), sessionID); err != nil { + if err := client.sessionMgr.Touch(context.Background(), sessionID); err != nil { logger.Log.Warnw("touch session failed", "session", sessionID, "error", err) } // 标记活跃请求 - if err := sessionMgr.SetActiveRequest(context.Background(), sessionID, msg.RequestID); err != nil { + if err := client.sessionMgr.SetActiveRequest(context.Background(), sessionID, msg.RequestID); err != nil { logger.Log.Warnw("set active request failed", "session", sessionID, "error", err) } - // 获取对话历史(供后续 Orchestrator 使用) - _, _ = sessionMgr.GetHistory(context.Background(), sessionID, 20) + // 获取对话历史 + history, _ := client.sessionMgr.GetHistory(context.Background(), sessionID, 20) - // TODO: 解码 audio Base64 → 启动 orchestrator.ProcessQuery goroutine + // 创建可取消的 context + ctx, cancel := context.WithCancel(context.Background()) + client.mu.Lock() + client.cancelFuncs[msg.RequestID] = cancel + client.mu.Unlock() + + // 创建 sender + sender := &WSClient{client: client, requestID: msg.RequestID} + + // 启动 orchestrator 处理 goroutine + go func() { + defer func() { + // 清理 cancel func + client.mu.Lock() + delete(client.cancelFuncs, msg.RequestID) + client.mu.Unlock() + cancel() + // 清除活跃请求 + _ = client.sessionMgr.ClearActiveRequest(context.Background(), sessionID) + }() + + if err := client.orchestrator.ProcessQuery(ctx, sessionID, msg, history, sender); err != nil { + logger.Log.Errorw("process query failed", "session", sessionID, "request", msg.RequestID, "error", err) + } + }() case "config": var msg models.WsConfig