From 810eb26be3e662e63b2b86bcb944feebce0834be Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Tue, 8 Sep 2026 09:17:37 +0800 Subject: [PATCH] =?UTF-8?q?feat(api):=20=E5=AE=9E=E7=8E=B0HTTP=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E9=AA=A8=E6=9E=B6=E4=B8=8EServer-Sent=20Events?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E4=BA=A4=E4=BA=92=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- cmd/server/main.go | 78 +++++++++++++ internal/config/config.go | 20 ++++ internal/server/handlers.go | 170 +++++++++++++++++++++++++++ internal/server/server.go | 194 +++++++++++++++++++++++++++++++ internal/server/server_test.go | 150 ++++++++++++++++++++++++ internal/server/sse.go | 206 +++++++++++++++++++++++++++++++++ 6 files changed, 818 insertions(+) create mode 100644 cmd/server/main.go create mode 100644 internal/server/handlers.go create mode 100644 internal/server/server.go create mode 100644 internal/server/server_test.go create mode 100644 internal/server/sse.go diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 00000000..89aacd4c --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,78 @@ +package main + +import ( + "context" + "flag" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/server" +) + +func main() { + envPath := flag.String("env", ".env", "环境配置文件路径") + port := flag.Int("port", 0, "HTTP 服务端口 (默认读取配置文件或 8080)") + host := flag.String("host", "", "HTTP 监听主机 (默认读取配置文件或 127.0.0.1)") + backend := flag.String("backend", "", "智能体后端调度覆盖 (stub, pi-agent, cursor-cli)") + flag.Parse() + + cfg, err := config.Load(*envPath) + if err != nil { + fmt.Fprintf(os.Stderr, "加载配置失败: %v\n", err) + os.Exit(1) + } + + if *port > 0 { + cfg.HTTPPort = *port + } + if *host != "" { + cfg.HTTPHost = *host + } + if *backend != "" { + cfg.AgentBackend = *backend + } + + svc := bizdev.NewService(cfg) + srv := server.NewServer(cfg, svc) + + addr := fmt.Sprintf("%s:%d", cfg.HTTPHost, cfg.HTTPPort) + fmt.Println("==================================================================") + fmt.Println(" 🤖 医疗器械商务与研发智能参谋系统 - HTTP / SSE API 守护服务 ") + fmt.Println("==================================================================") + fmt.Printf("服务环境: %s | 智能体后端: %s\n", cfg.Env, cfg.AgentBackend) + fmt.Printf("监听地址: http://%s\n", addr) + fmt.Println("核心服务路由:") + fmt.Println(" • [GET] /api/v1/health - 服务健康检查与基础设施状态") + fmt.Println(" • [GET] /api/v1/projects - 查询项目当前上下文与黑板事实") + fmt.Println(" • [POST] /api/v1/chat - 标准阻塞式对话协同接口") + fmt.Println(" • [POST] /api/v1/chat/stream - 核心 SSE 流式长连接对话协议 (分层按需)") + fmt.Println(" • [GET] /api/v1/chat/stream - SSE 流式长连接对话调试接口 (EventSource)") + fmt.Println(" • [POST] /api/v1/tasks/run/stream- 独立任务与阶段执行 SSE 监控流") + fmt.Println("==================================================================") + + stop := make(chan os.Signal, 1) + signal.Notify(stop, os.Interrupt, syscall.SIGTERM) + + go func() { + if err := srv.Start(); err != nil && err != http.ErrServerClosed { + fmt.Fprintf(os.Stderr, "服务启动失败: %v\n", err) + os.Exit(1) + } + }() + + <-stop + fmt.Println("\n正在优雅停止服务...") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := srv.Shutdown(ctx); err != nil { + fmt.Fprintf(os.Stderr, "服务停止异常: %v\n", err) + } else { + fmt.Println("服务已安全退出。") + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 2f387c45..445dce72 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -11,6 +11,8 @@ import ( type Config struct { Env string `json:"env"` DataDir string `json:"data_dir"` + HTTPHost string `json:"http_host"` + HTTPPort int `json:"http_port"` AgentBackend string `json:"agent_backend"` AgentTimeout time.Duration `json:"agent_timeout"` CursorAPIKey string `json:"cursor_api_key"` @@ -45,6 +47,8 @@ func Load(path string) (*Config, error) { cfg := &Config{ Env: "dev", DataDir: "./data/dev", + HTTPHost: "127.0.0.1", + HTTPPort: 8080, AgentBackend: "pi-agent", AgentTimeout: 30 * time.Minute, CursorModel: "auto", @@ -102,6 +106,12 @@ func Load(path string) (*Config, error) { cfg.Env = val case "DATA_DIR": cfg.DataDir = val + case "APP_PUBLISH_HOST", "HTTP_HOST": + cfg.HTTPHost = val + case "HTTP_PORT": + if p, err := strconv.Atoi(val); err == nil { + cfg.HTTPPort = p + } case "AGENT_BACKEND": cfg.AgentBackend = val case "AGENT_TIMEOUT": @@ -165,6 +175,16 @@ func Load(path string) (*Config, error) { if v := os.Getenv("DATA_DIR"); v != "" { cfg.DataDir = v } + if v := os.Getenv("HTTP_HOST"); v != "" { + cfg.HTTPHost = v + } else if v := os.Getenv("APP_PUBLISH_HOST"); v != "" { + cfg.HTTPHost = v + } + if v := os.Getenv("HTTP_PORT"); v != "" { + if p, err := strconv.Atoi(v); err == nil { + cfg.HTTPPort = p + } + } if v := os.Getenv("AGENT_BACKEND"); v != "" { cfg.AgentBackend = v } diff --git a/internal/server/handlers.go b/internal/server/handlers.go new file mode 100644 index 00000000..7b0e35a6 --- /dev/null +++ b/internal/server/handlers.go @@ -0,0 +1,170 @@ +package server + +import ( + "encoding/json" + "net/http" + "path/filepath" + + "chat-agent-system/internal/bizdev" +) + +// HealthResponse 健康检查响应 +type HealthResponse struct { + Status string `json:"status"` + Env string `json:"env"` + AgentBackend string `json:"agent_backend"` + Version string `json:"version"` + SharedKBPort int `json:"shared_kb_port"` +} + +// ChatRequest 对话请求 +type ChatRequest struct { + SessionID string `json:"session_id"` + Message string `json:"message"` + ProjectName string `json:"project_name,omitempty"` + CustomerName string `json:"customer_name,omitempty"` + DeviceType string `json:"device_type,omitempty"` + IntendedUse string `json:"intended_use,omitempty"` + TargetMarket string `json:"target_market,omitempty"` + WorkDir string `json:"work_dir,omitempty"` + UserName string `json:"user_name,omitempty"` + UserRole string `json:"user_role,omitempty"` +} + +// TaskRunRequestDTO 独立任务运行请求 +type TaskRunRequestDTO struct { + TaskKey string `json:"task_key"` + Stage string `json:"stage,omitempty"` // 若指定 stage，则执行成熟度阶段跃迁 + ProjectName string `json:"project_name"` + CustomerName string `json:"customer_name"` + DeviceType string `json:"device_type"` + IntendedUse string `json:"intended_use"` + TargetMarket string `json:"target_market"` + Fields map[string]string `json:"fields,omitempty"` + Backend string `json:"backend,omitempty"` + AutoRepair bool `json:"auto_repair"` +} + +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + resp := HealthResponse{ + Status: "ok", + Env: s.cfg.Env, + AgentBackend: s.cfg.AgentBackend, + Version: "v1.0", + SharedKBPort: s.cfg.KbDbPort, + } + writeJSONResponse(w, http.StatusOK, resp) +} + +func (s *Server) handleGetProject(w http.ResponseWriter, r *http.Request) { + sessionID := r.URL.Query().Get("session_id") + if sessionID == "" { + http.Error(w, "missing session_id query param", http.StatusBadRequest) + return + } + + s.sessionsMu.RLock() + sess, exists := s.sessions[sessionID] + s.sessionsMu.RUnlock() + + if !exists || sess == nil || sess.UserCtx == nil { + http.Error(w, "session or project context not found", http.StatusNotFound) + return + } + + writeJSONResponse(w, http.StatusOK, sess.UserCtx) +} + +func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { + var req ChatRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest) + return + } + if req.Message == "" { + http.Error(w, "message cannot be empty", http.StatusBadRequest) + return + } + + sess := s.getOrCreateSession( + req.SessionID, + req.ProjectName, + req.CustomerName, + req.DeviceType, + req.IntendedUse, + req.TargetMarket, + req.WorkDir, + req.UserName, + req.UserRole, + ) + + resp, err := sess.SendMessage(r.Context(), req.Message) + if err != nil { + http.Error(w, "process chat message failed: "+err.Error(), http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusOK, resp) +} + +func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { + var req TaskRunRequestDTO + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest) + return + } + + // 1. 如果指定了 stage，则执行成熟度阶段编排 + if req.Stage != "" { + stageRes, err := s.svc.ExecuteStage(r.Context(), bizdev.StageRunRequest{ + Stage: bizdev.MaturityStage(req.Stage), + ProjectName: req.ProjectName, + CustomerName: req.CustomerName, + DeviceType: req.DeviceType, + IntendedUse: req.IntendedUse, + TargetMarket: req.TargetMarket, + Fields: req.Fields, + Backend: req.Backend, + AutoRepair: req.AutoRepair, + }) + if err != nil { + http.Error(w, "execute stage failed: "+err.Error(), http.StatusInternalServerError) + return + } + writeJSONResponse(w, http.StatusOK, stageRes) + return + } + + // 2. 执行单任务 + if req.TaskKey == "" { + req.TaskKey = "intake" + } + taskRes, err := s.svc.ExecuteTask(r.Context(), bizdev.TaskRunRequest{ + TaskKey: req.TaskKey, + ProjectName: req.ProjectName, + CustomerName: req.CustomerName, + DeviceType: req.DeviceType, + IntendedUse: req.IntendedUse, + TargetMarket: req.TargetMarket, + Fields: req.Fields, + Backend: req.Backend, + AutoRepair: req.AutoRepair, + }) + if err != nil { + http.Error(w, "execute task failed: "+err.Error(), http.StatusInternalServerError) + return + } + + // 隐藏不必要的中间大体积字节 + if taskRes.WorkDir != "" { + taskRes.WorkDir = filepath.Clean(taskRes.WorkDir) + } + + writeJSONResponse(w, http.StatusOK, taskRes) +} + +func writeJSONResponse(w http.ResponseWriter, status int, data any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(data) +} diff --git a/internal/server/server.go b/internal/server/server.go new file mode 100644 index 00000000..375b50ef --- /dev/null +++ b/internal/server/server.go @@ -0,0 +1,194 @@ +package server + +import ( + "context" + "fmt" + "net/http" + "sync" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/chat" + "chat-agent-system/internal/config" + "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/intent" +) + +// Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务 +type Server struct { + cfg *config.Config + svc *bizdev.Service + rec intent.Recognizer + engine *coordinator.CoordinatorEngine + sessions map[string]*chat.Session + sessionsMu sync.RWMutex + httpServer *http.Server +} + +// NewServer 构造函数 +func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { + if cfg == nil { + cfg, _ = config.Load("") + } + if svc == nil { + svc = bizdev.NewService(cfg) + } + + rec := intent.NewRecognizer() + coordEngine := coordinator.NewCoordinatorEngine(svc) + + s := &Server{ + cfg: cfg, + svc: svc, + rec: rec, + engine: coordEngine, + sessions: make(map[string]*chat.Session), + } + return s +} + +// Routes 构建 HTTP 路由分发器 +func (s *Server) Routes() http.Handler { + mux := http.NewServeMux() + + // 1. 健康检查与系统自检 + mux.HandleFunc("GET /health", s.handleHealth) + mux.HandleFunc("GET /api/v1/health", s.handleHealth) + + // 2. 项目状态与会话查询 + mux.HandleFunc("GET /api/v1/projects", s.handleGetProject) + + // 3. 对话协同服务 (标准 JSON 阻塞) + mux.HandleFunc("POST /api/v1/chat", s.handleChat) + + // 4. 对话协同服务 (核心 SSE 流式长连接) + mux.HandleFunc("POST /api/v1/chat/stream", s.handleChatStream) + mux.HandleFunc("GET /api/v1/chat/stream", s.handleChatStream) + + // 5. 独立任务 / 阶段执行 + mux.HandleFunc("POST /api/v1/tasks/run", s.handleRunTask) + mux.HandleFunc("POST /api/v1/tasks/run/stream", s.handleRunTaskStream) + mux.HandleFunc("GET /api/v1/tasks/run/stream", s.handleRunTaskStream) + + return s.corsMiddleware(mux) +} + +// Start 启动 HTTP 守护进程 +func (s *Server) Start() error { + addr := fmt.Sprintf("%s:%d", s.cfg.HTTPHost, s.cfg.HTTPPort) + s.httpServer = &http.Server{ + Addr: addr, + Handler: s.Routes(), + ReadHeaderTimeout: 10 * time.Second, + } + + return s.httpServer.ListenAndServe() +} + +// Shutdown 优雅停机 +func (s *Server) Shutdown(ctx context.Context) error { + if s.httpServer != nil { + return s.httpServer.Shutdown(ctx) + } + return nil +} + +// getOrCreateSession 获取或动态构建会话 +func (s *Server) getOrCreateSession( + sessionID string, + projectName string, + customerName string, + deviceType string, + intendedUse string, + targetMarket string, + workDir string, + userName string, + userRole string, +) *chat.Session { + if sessionID == "" { + sessionID = fmt.Sprintf("sess_%d", time.Now().UnixNano()) + } + + s.sessionsMu.RLock() + existing, exists := s.sessions[sessionID] + s.sessionsMu.RUnlock() + if exists && existing != nil { + return existing + } + + s.sessionsMu.Lock() + defer s.sessionsMu.Unlock() + // 双重检查 + if existing, exists := s.sessions[sessionID]; exists && existing != nil { + return existing + } + + if userName == "" { + userName = "用户" + } + if userRole == "" { + userRole = "engineer" + } + + u := coordinator.UserInfo{ + ID: "u_" + sessionID, + Name: userName, + Role: userRole, + ProjectRole: "composer", + } + + var userCtx *coordinator.UserProjectContext + if workDir != "" { + assembler := coordinator.NewContextAssembler() + userCtx, _ = assembler.AssembleFromWorkspace(workDir, u) + } + + if userCtx == nil { + if projectName == "" { + projectName = "医疗器械研发项目" + } + if customerName == "" { + customerName = "水木东方客户" + } + userCtx = &coordinator.UserProjectContext{ + SnapshotID: fmt.Sprintf("ctx_%d", time.Now().UnixNano()), + AssembledAt: time.Now(), + User: u, + WorkDir: workDir, + Project: coordinator.ProjectInfo{ + Name: projectName, + CustomerName: customerName, + DeviceType: deviceType, + IntendedUse: intendedUse, + TargetMarket: targetMarket, + CurrentWorkflow: "intake", + CurrentStage: bizdev.StageV01_Triage, + }, + BlackboardFacts: make(map[string]string), + AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, u.ProjectRole), + } + if targetMarket != "" { + userCtx.BlackboardFacts["target_market"] = targetMarket + } + if deviceType != "" { + userCtx.BlackboardFacts["device_type"] = deviceType + } + } + + sess := chat.NewSession(sessionID, userCtx, s.rec, s.engine) + s.sessions[sessionID] = sess + return sess +} + +func (s *Server) corsMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS, PUT, DELETE") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, Accept") + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + next.ServeHTTP(w, r) + }) +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go new file mode 100644 index 00000000..f0bdf20f --- /dev/null +++ b/internal/server/server_test.go @@ -0,0 +1,150 @@ +package server + +import ( + "bufio" + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/chat" + "chat-agent-system/internal/config" + "chat-agent-system/internal/stream" +) + +func newTestServer() *Server { + cfg := &config.Config{ + Env: "test", + AgentBackend: "stub", + DataDir: "./data/test", + KbDbPort: 5432, + } + svc := bizdev.NewService(cfg) + return NewServer(cfg, svc) +} + +func TestServer_Health(t *testing.T) { + srv := newTestServer() + req := httptest.NewRequest(http.MethodGet, "/api/v1/health", nil) + w := httptest.NewRecorder() + + srv.Routes().ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d", w.Code) + } + + var resp HealthResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode response failed: %v", err) + } + + if resp.Status != "ok" || resp.Env != "test" { + t.Errorf("unexpected health response: %+v", resp) + } +} + +func TestServer_ChatBlocking(t *testing.T) { + srv := newTestServer() + + chatReq := ChatRequest{ + SessionID: "test_sess_01", + Message: "查一下当前进展", + ProjectName: "内窥镜研发项目", + CustomerName: "水木东方", + } + body, _ := json.Marshal(chatReq) + req := httptest.NewRequest(http.MethodPost, "/api/v1/chat", bytes.NewReader(body)) + w := httptest.NewRecorder() + + srv.Routes().ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp chat.ChatResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode chat response failed: %v", err) + } + + if resp.ReplyText == "" { + t.Errorf("expected non-empty reply text") + } +} + +func TestServer_ChatSSEStream(t *testing.T) { + srv := newTestServer() + + chatReq := StreamChatRequest{ + SessionID: "test_sess_sse", + Message: "查一下当前进展", + IncludeDetails: false, + ProjectName: "内窥镜研发项目", + CustomerName: "水木东方", + } + body, _ := json.Marshal(chatReq) + req := httptest.NewRequest(http.MethodPost, "/api/v1/chat/stream", bytes.NewReader(body)) + w := httptest.NewRecorder() + + srv.Routes().ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d", w.Code) + } + + contentType := w.Header().Get("Content-Type") + if !strings.Contains(contentType, "text/event-stream") { + t.Fatalf("expected Content-Type text/event-stream, got %s", contentType) + } + + // 解析 SSE 事件流 + scanner := bufio.NewScanner(w.Body) + var currentEvent string + var eventsReceived []string + + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "event: ") { + currentEvent = strings.TrimPrefix(line, "event: ") + } else if strings.HasPrefix(line, "data: ") { + dataStr := strings.TrimPrefix(line, "data: ") + var ev stream.StreamEvent + if err := json.Unmarshal([]byte(dataStr), &ev); err == nil { + eventsReceived = append(eventsReceived, currentEvent) + } + } + } + + if len(eventsReceived) == 0 { + t.Fatalf("expected to receive SSE events, got 0") + } + + hasPhase := false + hasStatus := false + hasDone := false + + for _, et := range eventsReceived { + switch et { + case "phase": + hasPhase = true + case "status": + hasStatus = true + case "done": + hasDone = true + } + } + + if !hasPhase { + t.Errorf("expected phase event in SSE stream") + } + if !hasStatus { + t.Errorf("expected status event in SSE stream") + } + if !hasDone { + t.Errorf("expected done event in SSE stream") + } +} diff --git a/internal/server/sse.go b/internal/server/sse.go new file mode 100644 index 00000000..a5cdd32f --- /dev/null +++ b/internal/server/sse.go @@ -0,0 +1,206 @@ +package server + +import ( + "encoding/json" + "fmt" + "net/http" + "strconv" + "sync" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/stream" +) + +// StreamChatRequest SSE流式请求参数 +type StreamChatRequest struct { + SessionID string `json:"session_id"` + Message string `json:"message"` + IncludeDetails bool `json:"include_details"` // 是否推送底层长篇推演草稿 (按需展开，默认 false) + ProjectName string `json:"project_name,omitempty"` + CustomerName string `json:"customer_name,omitempty"` + DeviceType string `json:"device_type,omitempty"` + IntendedUse string `json:"intended_use,omitempty"` + TargetMarket string `json:"target_market,omitempty"` + WorkDir string `json:"work_dir,omitempty"` + UserName string `json:"user_name,omitempty"` + UserRole string `json:"user_role,omitempty"` +} + +func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported by underlying connection", http.StatusInternalServerError) + return + } + + var req StreamChatRequest + if r.Method == http.MethodPost { + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest) + return + } + } else { + // 允许 GET query params 方便 EventSource 及浏览器直接调试 + q := r.URL.Query() + req.SessionID = q.Get("session_id") + req.Message = q.Get("message") + req.ProjectName = q.Get("project_name") + req.CustomerName = q.Get("customer_name") + req.DeviceType = q.Get("device_type") + req.IntendedUse = q.Get("intended_use") + req.TargetMarket = q.Get("target_market") + req.WorkDir = q.Get("work_dir") + req.UserName = q.Get("user_name") + req.UserRole = q.Get("user_role") + if inc := q.Get("include_details"); inc != "" { + req.IncludeDetails, _ = strconv.ParseBool(inc) + } + } + + if req.Message == "" { + http.Error(w, "message cannot be empty", http.StatusBadRequest) + return + } + + // 1. 设置 SSE 响应头 + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲 + flusher.Flush() + + // 2. 准备会话 + sess := s.getOrCreateSession( + req.SessionID, + req.ProjectName, + req.CustomerName, + req.DeviceType, + req.IntendedUse, + req.TargetMarket, + req.WorkDir, + req.UserName, + req.UserRole, + ) + + // 3. 构建线程安全的 SSE 监听分发闭包 + var writeMu sync.Mutex + listener := func(ev stream.StreamEvent) { + writeMu.Lock() + defer writeMu.Unlock() + + data, err := json.Marshal(ev) + if err != nil { + return + } + // 按照标准 SSE 协议规范发射: event: \ndata: \n\n + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + flusher.Flush() + } + + // 4. 发起流式执行 + _, err := sess.SendMessageStream(r.Context(), req.Message, listener, req.IncludeDetails) + if err != nil { + writeMu.Lock() + errEv := stream.StreamEvent{ + Type: stream.EventError, + Timestamp: time.Now(), + Error: err.Error(), + } + data, _ := json.Marshal(errEv) + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", errEv.Type, string(data)) + flusher.Flush() + writeMu.Unlock() + } +} + +func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + + var req TaskRunRequestDTO + includeDetails := false + if r.Method == http.MethodPost { + _ = json.NewDecoder(r.Body).Decode(&req) + } else { + q := r.URL.Query() + req.TaskKey = q.Get("task_key") + req.Stage = q.Get("stage") + req.ProjectName = q.Get("project_name") + req.CustomerName = q.Get("customer_name") + req.DeviceType = q.Get("device_type") + req.IntendedUse = q.Get("intended_use") + req.TargetMarket = q.Get("target_market") + req.Backend = q.Get("backend") + req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair")) + if inc := q.Get("include_details"); inc != "" { + includeDetails, _ = strconv.ParseBool(inc) + } + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Access-Control-Allow-Origin", "*") + flusher.Flush() + + var writeMu sync.Mutex + listener := func(ev stream.StreamEvent) { + writeMu.Lock() + defer writeMu.Unlock() + data, err := json.Marshal(ev) + if err != nil { + return + } + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + flusher.Flush() + } + + ctxWithS, streamer := stream.WithListener(r.Context(), includeDetails, listener) + start := time.Now() + + if req.Stage != "" { + _, err := s.svc.ExecuteStage(ctxWithS, bizdev.StageRunRequest{ + Stage: bizdev.MaturityStage(req.Stage), + ProjectName: req.ProjectName, + CustomerName: req.CustomerName, + DeviceType: req.DeviceType, + IntendedUse: req.IntendedUse, + TargetMarket: req.TargetMarket, + Fields: req.Fields, + Backend: req.Backend, + AutoRepair: req.AutoRepair, + }) + if err != nil { + streamer.EmitError(err) + } else { + streamer.EmitDone("stage_run", time.Since(start)) + } + return + } + + if req.TaskKey == "" { + req.TaskKey = "intake" + } + + _, err := s.svc.ExecuteTask(ctxWithS, bizdev.TaskRunRequest{ + TaskKey: req.TaskKey, + ProjectName: req.ProjectName, + CustomerName: req.CustomerName, + DeviceType: req.DeviceType, + IntendedUse: req.IntendedUse, + TargetMarket: req.TargetMarket, + Fields: req.Fields, + Backend: req.Backend, + AutoRepair: req.AutoRepair, + }) + if err != nil { + streamer.EmitError(err) + } else { + streamer.EmitDone("task_run", time.Since(start)) + } +}