diff --git a/internal/server/taskstream_test.go b/internal/server/taskstream_test.go new file mode 100644 index 00000000..90d72734 --- /dev/null +++ b/internal/server/taskstream_test.go @@ -0,0 +1,219 @@ +package server + +import ( + "bufio" + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/stream" +) + +// readSSEEvents 从 SSE 响应解析事件序列（测试断言辅助） +func readSSEEvents(t *testing.T, resp *http.Response) []stream.StreamEvent { + t.Helper() + scanner := bufio.NewScanner(resp.Body) + var events []stream.StreamEvent + var currentType string + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "event: ") { + currentType = strings.TrimPrefix(line, "event: ") + } else if strings.HasPrefix(line, "data: ") { + var ev stream.StreamEvent + if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err == nil { + if ev.Type == "" { + ev.Type = stream.EventType(currentType) + } + events = append(events, ev) + } + } + } + return events +} + +func findEvent(events []stream.StreamEvent, eventType stream.EventType) *stream.StreamEvent { + for i := range events { + if events[i].Type == eventType { + return &events[i] + } + } + return nil +} + +func hasAction(actions []string, want string) bool { + for _, a := range actions { + if a == want { + return true + } + } + return false +} + +// TestTaskStream_TransientInjection_DeadPiAgent 会话级 transient 故障注入： +// pi-agent 指向不可达端口 → dial 失败 → SSE 必须输出 event:error 且 error_info.class=transient +// （补齐 2026-09-09 测试报告遗留建议 #2） +func TestTaskStream_TransientInjection_DeadPiAgent(t *testing.T) { + cfg := &config.Config{ + Env: "test", + AgentBackend: "stub", + DataDir: "./data/test", + KbDbPort: 5432, + PiAgentRunnerURL: "http://127.0.0.1:1", // 死端口：connection refused → 网络类瞬态错误 + PiAgentModel: "MiniMax-M3", + PiAgentProvider: "minimax", + AgentTimeout: 5 * time.Second, + } + svc := bizdev.NewService(cfg) + srv := NewServer(cfg, svc) + + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + body, _ := json.Marshal(map[string]any{ + "task_key": "intake", + "project_name": "超声内镜系统", + "customer": "北京水木东方", + "backend": "pi-agent", + }) + resp, err := http.Post(ts.URL+"/api/v1/tasks/run/stream", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("post tasks run stream failed: %v", err) + } + defer resp.Body.Close() + + if rid := resp.Header.Get("X-Request-Id"); rid == "" { + t.Fatal("expected non-empty X-Request-Id header") + } + + events := readSSEEvents(t, resp) + errEv := findEvent(events, stream.EventError) + if errEv == nil { + t.Fatalf("expected event:error in stream, got %d events", len(events)) + } + if errEv.ErrorInfo == nil { + t.Fatal("error_info is nil") + } + if errEv.ErrorInfo.Class != stream.ErrorClassTransient { + t.Fatalf("class=%q want transient (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream) + } + if !hasAction(errEv.ErrorInfo.Actions, "retry") { + t.Fatalf("actions=%v want retry", errEv.ErrorInfo.Actions) + } + if errEv.ErrorInfo.RequestID == "" { + t.Fatal("error_info.request_id is empty") + } + if errEv.ErrorInfo.PhaseID == "" { + t.Fatal("error_info.phase_id is empty (expected last streamed phase)") + } +} + +// TestTaskStream_SoftFailure_ClassifiedResumable 软失败（pi-agent 上游业务失败而非网络/超时）： +// squad 层返回携带 WorkDir 的 result → class=resumable、actions=[continue restart]， +// workspace_id 供前端「继续任务」续跑 +func TestTaskStream_SoftFailure_ClassifiedResumable(t *testing.T) { + // 伪造 pi-agent runner：HTTP 500 + 非 JSON body → decode 失败 → 业务软失败（非 dial/timeout） + brokenRunner := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "simulated upstream business failure", http.StatusInternalServerError) + })) + defer brokenRunner.Close() + + cfg := &config.Config{ + Env: "test", + AgentBackend: "stub", + DataDir: "./data/test", + KbDbPort: 5432, + PiAgentRunnerURL: brokenRunner.URL, + PiAgentModel: "MiniMax-M3", + PiAgentProvider: "minimax", + AgentTimeout: 10 * time.Second, + } + svc := bizdev.NewService(cfg) + srv := NewServer(cfg, svc) + + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + body, _ := json.Marshal(map[string]any{ + "task_key": "intake", + "project_name": "超声内镜系统", + "customer": "北京水木东方", + "backend": "pi-agent", + }) + resp, err := http.Post(ts.URL+"/api/v1/tasks/run/stream", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("post tasks run stream failed: %v", err) + } + defer resp.Body.Close() + + events := readSSEEvents(t, resp) + errEv := findEvent(events, stream.EventError) + if errEv == nil { + t.Fatalf("expected event:error in stream, got %d events", len(events)) + } + if errEv.ErrorInfo == nil { + t.Fatal("error_info is nil") + } + if errEv.ErrorInfo.Class != stream.ErrorClassResumable { + t.Fatalf("class=%q want resumable (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream) + } + if errEv.ErrorInfo.WorkspaceID == "" { + t.Fatal("workspace_id is empty (expected squad soft-failure to carry work_dir)") + } + if !hasAction(errEv.ErrorInfo.Actions, "continue") || !hasAction(errEv.ErrorInfo.Actions, "restart") { + t.Fatalf("actions=%v want [continue restart]", errEv.ErrorInfo.Actions) + } +} + +// TestTaskStream_ResumeInvalidWorkspace_ClassifiedResumable resume 模式： +// work_dir 非空走 ResumeOrModifyWorkspace；目标工作区不存在 → 失败仍归类 resumable +// 且 workspace_id 等于请求的 work_dir（工作区可被修复后继续） +func TestTaskStream_ResumeInvalidWorkspace_ClassifiedResumable(t *testing.T) { + cfg := &config.Config{ + Env: "test", + AgentBackend: "stub", + DataDir: "./data/test", + KbDbPort: 5432, + AgentTimeout: 10 * time.Second, + } + svc := bizdev.NewService(cfg) + srv := NewServer(cfg, svc) + + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + const missingWS = "/tmp/p2-nonexistent-workspace-resume-test" + body, _ := json.Marshal(map[string]any{ + "work_dir": missingWS, + "backend": "stub", + }) + resp, err := http.Post(ts.URL+"/api/v1/tasks/run/stream", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("post tasks run stream failed: %v", err) + } + defer resp.Body.Close() + + events := readSSEEvents(t, resp) + errEv := findEvent(events, stream.EventError) + if errEv == nil { + t.Fatalf("expected event:error in stream, got %d events", len(events)) + } + if errEv.ErrorInfo == nil { + t.Fatal("error_info is nil") + } + if errEv.ErrorInfo.Class != stream.ErrorClassResumable { + t.Fatalf("class=%q want resumable (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream) + } + if errEv.ErrorInfo.WorkspaceID != missingWS { + t.Fatalf("workspace_id=%q want %q", errEv.ErrorInfo.WorkspaceID, missingWS) + } + if !hasAction(errEv.ErrorInfo.Actions, "continue") || !hasAction(errEv.ErrorInfo.Actions, "restart") { + t.Fatalf("actions=%v want [continue restart]", errEv.ErrorInfo.Actions) + } +}