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 <cursoragent@cursor.com>
2026-09-11 15:15:11 +08:00
parent 0c9c0928ec
commit 7d2f86bdc4
2 changed files with 65 additions and 128 deletions
@@ -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))
 	}
 }
@@ -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)