diff --git a/internal/server/sse.go b/internal/server/sse.go index fb167e8f..5804aa77 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -9,9 +9,18 @@ import ( "time" "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" ) +// taskStreamRequest 任务/阶段流式运行请求:在 handlers.go 的 TaskRunRequestDTO 基础上 +// 增量扩展 resume 能力(work_dir 非空 → 续跑既有工作区),不改动既有 DTO 定义(向后兼容)。 +type taskStreamRequest struct { + TaskRunRequestDTO + WorkDir string `json:"work_dir,omitempty"` + AddSteps []squad.StepSpec `json:"add_steps,omitempty"` +} + // StreamChatRequest SSE流式请求参数 type StreamChatRequest struct { SessionID string `json:"session_id"` @@ -136,7 +145,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { return } - var req TaskRunRequestDTO + var req taskStreamRequest includeDetails := false if r.Method == http.MethodPost { _ = json.NewDecoder(r.Body).Decode(&req) @@ -150,22 +159,33 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { req.IntendedUse = q.Get("intended_use") req.TargetMarket = q.Get("target_market") req.Backend = q.Get("backend") + req.WorkDir = q.Get("work_dir") req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair")) if inc := q.Get("include_details"); inc != "" { includeDetails, _ = strconv.ParseBool(inc) } } + // 请求追踪 ID:与 chat stream 对齐,注入 ctx 并回写响应头 + requestID := fmt.Sprintf("req_%d", time.Now().UnixNano()) + w.Header().Set("X-Request-Id", requestID) 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() 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 { return @@ -174,11 +194,40 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { flusher.Flush() } - ctxWithS, streamer := stream.WithListener(r.Context(), includeDetails, listener) + // 分类失败并发射结构化错误: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() + } + 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 + } + classifyAndEmitError(err, "task_resume", wsID) + } else { + streamer.EmitDoneWithWorkspace("task_resume", req.WorkDir, time.Since(start)) + } + return + } + if req.Stage != "" { - _, err := s.svc.ExecuteStage(ctxWithS, bizdev.StageRunRequest{ + res, err := s.svc.ExecuteStage(ctxWithS, bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(req.Stage), ProjectName: req.ProjectName, CustomerName: req.CustomerName, @@ -200,9 +249,17 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { ForceDirectEdit: req.ForceDirectEdit, }) if err != nil { - streamer.EmitError(err) + wsID := "" + if res != nil && res.WorkDir != "" { + wsID = res.WorkDir + } + classifyAndEmitError(err, "stage_run", wsID) } else { - streamer.EmitDone("stage_run", time.Since(start)) + wsID := "" + if res != nil { + wsID = res.WorkDir + } + streamer.EmitDoneWithWorkspace("stage_run", wsID, time.Since(start)) } return } @@ -211,7 +268,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { req.TaskKey = "intake" } - _, err := s.svc.ExecuteTask(ctxWithS, bizdev.TaskRunRequest{ + res, err := s.svc.ExecuteTask(ctxWithS, bizdev.TaskRunRequest{ TaskKey: req.TaskKey, ProjectName: req.ProjectName, CustomerName: req.CustomerName, @@ -233,8 +290,18 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { ForceDirectEdit: req.ForceDirectEdit, }) if err != nil { - streamer.EmitError(err) + // 软失败(执行器返回带 WorkDir 的 result,如 docgov 锁定 / squad 层任务中断)→ resumable; + // 硬失败(工作区未建立)→ workspaceID 为空,按超时/网络/默认规则分类 + wsID := "" + if res != nil && res.WorkDir != "" { + wsID = res.WorkDir + } + classifyAndEmitError(err, "", wsID) } else { - streamer.EmitDone("task_run", time.Since(start)) + wsID := "" + if res != nil { + wsID = res.WorkDir + } + streamer.EmitDoneWithWorkspace("task_run", wsID, time.Since(start)) } } diff --git a/internal/stream/stream.go b/internal/stream/stream.go index cd956d95..a9c495de 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -226,6 +226,22 @@ func (s *Streamer) EmitDone(sessionID string, duration time.Duration) { }) } +// EmitDoneWithWorkspace 发送流结束事件并携带任务工作区目录(P2:供前端「继续任务」续跑) +func (s *Streamer) EmitDoneWithWorkspace(sessionID, workspaceID string, duration time.Duration) { + if s == nil { + return + } + s.Emit(StreamEvent{ + Type: EventDone, + Done: &DonePayload{ + SessionID: sessionID, + WorkspaceID: workspaceID, + DurationMs: duration.Milliseconds(), + TotalEvents: int(atomic.LoadInt64(&s.eventCount)), + }, + }) +} + // EventCount 获取已发射事件总数 func (s *Streamer) EventCount() int { if s == nil { diff --git a/internal/stream/stream_test.go b/internal/stream/stream_test.go index f3c8fd1c..134a30d3 100644 --- a/internal/stream/stream_test.go +++ b/internal/stream/stream_test.go @@ -2,7 +2,9 @@ package stream import ( "context" + "encoding/json" "errors" + "strings" "sync" "testing" "time" @@ -165,3 +167,36 @@ func TestStreamer_Concurrency(t *testing.T) { t.Fatalf("expected %d events, got %d", expected, total) } } + +func TestStreamer_EmitDoneWithWorkspace(t *testing.T) { + var received StreamEvent + + streamer := NewStreamer(true, func(ev StreamEvent) { + received = ev + }) + + streamer.EmitDoneWithWorkspace("task_run", "/tmp/ws-demo", 1200*time.Millisecond) + + if received.Type != EventDone { + t.Fatalf("type=%s", received.Type) + } + if received.Done == nil { + t.Fatal("Done payload is nil") + } + if received.Done.WorkspaceID != "/tmp/ws-demo" { + t.Fatalf("WorkspaceID=%q want %q", received.Done.WorkspaceID, "/tmp/ws-demo") + } + if received.Done.DurationMs != 1200 { + t.Fatalf("DurationMs=%d want 1200", received.Done.DurationMs) + } + + // workspace_id 为空时 omitempty 不应出现在 JSON 中 + streamer.EmitDone("task_run", 100*time.Millisecond) + data, err := json.Marshal(received) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if strings.Contains(string(data), "workspace_id") { + t.Fatalf("expected workspace_id omitted, got %s", data) + } +} diff --git a/internal/stream/types.go b/internal/stream/types.go index 8b924556..e1894088 100644 --- a/internal/stream/types.go +++ b/internal/stream/types.go @@ -87,6 +87,7 @@ type DonePayload struct { SessionID string `json:"session_id,omitempty"` DurationMs int64 `json:"duration_ms"` TotalEvents int `json:"total_events"` + WorkspaceID string `json:"workspace_id,omitempty"` // P2: 任务工作区目录(供前端「继续任务」续跑) } type ErrorClass string