test(server): task stream transient injection, classification and resume integration tests

2026-09-13 02:50:20 +00:00
parent 912eb61a02
commit 1df0a3f15f
1 changed files with 219 additions and 0 deletions
@@ -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)
+	}
+}