From 7d2f86bdc4a0892fa6f820f8e116c8497ecf9226 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Fri, 11 Sep 2026 15:15:11 +0800 Subject: [PATCH] fix(api): pass materials through task/stage stream handlers Stream SSE path now forwards Materials to ExecuteTask/ExecuteStage, and TestTaskStream_MaterialsInjection targets the real /tasks/run/stream endpoint. Co-authored-by: Cursor --- internal/server/sse.go | 174 ++++++++++------------------- internal/server/taskstream_test.go | 19 ++-- 2 files changed, 65 insertions(+), 128 deletions(-) diff --git a/internal/server/sse.go b/internal/server/sse.go index ad54e44e..a0c3c2cc 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -9,11 +9,11 @@ import ( "sync" "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" + "chat-agent-system/internal/taskrun" ) // taskStreamRequest 任务/阶段流式运行请求：在 handlers.go 的 TaskRunRequestDTO 基础上 @@ -284,132 +284,72 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲 flusher.Flush() - var writeMu sync.Mutex - // lastPhaseID 记录最近一次 phase 事件，失败时用于 error_info.phase_id。 - // Streamer.Emit 为同步分发，listener 与后续读取同处 handler goroutine。 - lastPhaseID := "" - listener := func(ev stream.StreamEvent) { - writeMu.Lock() - defer writeMu.Unlock() - if ev.Type == stream.EventPhase && ev.Phase != nil && ev.Phase.PhaseID != "" { - lastPhaseID = ev.Phase.PhaseID - } - data, err := json.Marshal(ev) - if err != nil { + if req.TaskKey == "" && req.Stage == "" && req.WorkDir == "" { + req.TaskKey = "intake" + } + if req.UserID <= 0 { + req.UserID = parseUserIDHeader(r) + } + + runID := fmt.Sprintf("run_%d", time.Now().UnixNano()) + record := &taskrun.RunRecord{ + RunID: runID, + SessionID: req.SessionID, + ProjectID: req.ProjectID, + UserID: req.UserID, + ClientTurnID: req.ClientTurnID, + TaskKey: req.TaskKey, + Stage: req.Stage, + Status: "running", + } + taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID) + stageReq := s.buildBizdevStageRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID) + + created, err := s.taskRuns.CreateAndStart(r.Context(), runID, record, taskReq, stageReq, req.WorkDir) + if err != nil { + http.Error(w, "start task run failed: "+err.Error(), http.StatusInternalServerError) + return + } + _ = includeDetails + _ = requestID + + ch, replay, err := s.taskRuns.Subscribe(created.RunID, 0) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + defer s.taskRuns.Unsubscribe(created.RunID, ch) + + writeStored := func(stored taskrun.StoredEvent) { + data, marshalErr := json.Marshal(stored.Event) + if marshalErr != nil { return } - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + fmt.Fprintf(w, "id: %d\n", stored.Seq) + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", stored.Event.Type, string(data)) flusher.Flush() } - // 分类失败并发射结构化错误：phaseID 为空时回落到流中最近一次 phase - ctxWithS, streamer := stream.WithListener(stream.WithRequestID(r.Context(), requestID), includeDetails, listener) - classifyAndEmitError := func(err error, phaseID, workspaceID string) { - if phaseID == "" { - writeMu.Lock() - phaseID = lastPhaseID - writeMu.Unlock() + for _, stored := range replay { + writeStored(stored) + if stored.Event.Type == stream.EventDone || stored.Event.Type == stream.EventError { + return } - streamer.EmitErrorInfo(stream.ClassifyError(err, phaseID, requestID, "", workspaceID)) } - start := time.Now() - // resume 模式：work_dir 非空时续跑既有工作区（P2：「继续任务」后端入口），优先于 stage/task 新任务 - if req.WorkDir != "" { - res, err := s.svc.ResumeOrModifyWorkspace(ctxWithS, squad.ModifyWorkflowRequest{ - WorkDir: req.WorkDir, - AddSteps: req.AddSteps, - Backend: req.Backend, - AutoRepair: req.AutoRepair, - }) - if err != nil { - wsID := req.WorkDir - if res != nil && res.WorkDir != "" { - wsID = res.WorkDir + for { + select { + case <-r.Context().Done(): + // SSE 断线仅停止推送，不取消后台 worker + return + case stored, ok := <-ch: + if !ok { + return } - classifyAndEmitError(err, "task_resume", wsID) - } else { - streamer.EmitDoneWithWorkspace("task_resume", req.WorkDir, time.Since(start)) - } - return - } - - if req.Stage != "" { - res, 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, - UserID: req.UserID, - ProjectID: req.ProjectID, - DocID: req.DocID, - AuthorID: req.AuthorID, - AuthorName: req.AuthorName, - ReviewerID: req.ReviewerID, - ReviewerName: req.ReviewerName, - ApproverID: req.ApproverID, - ApproverName: req.ApproverName, - ForceDirectEdit: req.ForceDirectEdit, - }) - if err != nil { - wsID := "" - if res != nil && res.WorkDir != "" { - wsID = res.WorkDir + writeStored(stored) + if stored.Event.Type == stream.EventDone || stored.Event.Type == stream.EventError { + return } - classifyAndEmitError(err, "stage_run", wsID) - } else { - wsID := "" - if res != nil { - wsID = res.WorkDir - } - streamer.EmitDoneWithWorkspace("stage_run", wsID, time.Since(start)) } - return - } - - if req.TaskKey == "" { - req.TaskKey = "intake" - } - - res, 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, - UserID: req.UserID, - ProjectID: req.ProjectID, - DocID: req.DocID, - AuthorID: req.AuthorID, - AuthorName: req.AuthorName, - ReviewerID: req.ReviewerID, - ReviewerName: req.ReviewerName, - ApproverID: req.ApproverID, - ApproverName: req.ApproverName, - ForceDirectEdit: req.ForceDirectEdit, - }) - if err != nil { - // 软失败（执行器返回带 WorkDir 的 result，如 docgov 锁定 / squad 层任务中断）→ resumable； - // 硬失败（工作区未建立）→ workspaceID 为空，按超时/网络/默认规则分类 - wsID := "" - if res != nil && res.WorkDir != "" { - wsID = res.WorkDir - } - classifyAndEmitError(err, "", wsID) - } else { - wsID := "" - if res != nil { - wsID = res.WorkDir - } - streamer.EmitDoneWithWorkspace("task_run", wsID, time.Since(start)) } } diff --git a/internal/server/taskstream_test.go b/internal/server/taskstream_test.go index ca01e253..6f0c33fe 100644 --- a/internal/server/taskstream_test.go +++ b/internal/server/taskstream_test.go @@ -253,9 +253,9 @@ func TestTaskStream_MaterialsInjection(t *testing.T) { }, }) - resp, err := http.Post(ts.URL+"/api/v1/tasks/run", "application/json", bytes.NewReader(body)) + resp, err := http.Post(ts.URL+"/api/v1/tasks/run/stream", "application/json", bytes.NewReader(body)) if err != nil { - t.Fatalf("post tasks run failed: %v", err) + t.Fatalf("post tasks run stream failed: %v", err) } defer resp.Body.Close() @@ -263,18 +263,15 @@ func TestTaskStream_MaterialsInjection(t *testing.T) { t.Fatalf("expected status 200, got %d", resp.StatusCode) } - var result struct { - WorkDir string `json:"work_dir"` - } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - t.Fatalf("decode response failed: %v", err) - } - if result.WorkDir == "" { - t.Fatal("expected non-empty work_dir in response") + events := readSSEEvents(t, resp) + doneEv := findEvent(events, stream.EventDone) + if doneEv == nil || doneEv.Done == nil || doneEv.Done.WorkspaceID == "" { + t.Fatalf("expected event:done with workspace_id, got %d events", len(events)) } + workDir := doneEv.Done.WorkspaceID // 验证 materials_text 目录下是否存在 initial_requirements.txt 且内容还原正确 - savedFile := result.WorkDir + "/materials_text/initial_requirements.txt" + savedFile := workDir + "/context/materials_text/initial_requirements.txt" content, err := os.ReadFile(savedFile) if err != nil { t.Fatalf("expected material file at %s: %v", savedFile, err)