feat(api): task stream request-id, resume mode and workspace-aware error classification
This commit is contained in:
WeiCHEN
+75
-8
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user