Agent-System/internal/server/taskstream_test.go

package server

import (
	"bufio"
	"bytes"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"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:          t.TempDir(),
		KbDbPort:         5432,
		AdminUsername:    "admin",
		AdminPassword:    "admin-test-pass",
		PiAgentRunnerURL: "http://127.0.0.1:1", // 死端口：connection refused → 网络类瞬态错误
		PiAgentModel:     "MiniMax-M3",
		PiAgentProvider:  "minimax",
		AgentTimeout:     5 * time.Second,
	}
	svc := bizdev.MustNewService(t, 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",
	})
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run/stream", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+testBearerToken(t, ts.URL, cfg))
	resp, err := http.DefaultClient.Do(req)
	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:          t.TempDir(),
		KbDbPort:         5432,
		AdminUsername:    "admin",
		AdminPassword:    "admin-test-pass",
		PiAgentRunnerURL: brokenRunner.URL,
		PiAgentModel:     "MiniMax-M3",
		PiAgentProvider:  "minimax",
		AgentTimeout:     10 * time.Second,
	}
	svc := bizdev.MustNewService(t, 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",
	})
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run/stream", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+testBearerToken(t, ts.URL, cfg))
	resp, err := http.DefaultClient.Do(req)
	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_MaterialsInjection 验证前端传入的 materials（Base64 编码的附件内容）
// 能被后端 HTTP handler 正确反序列化，透传给 ExecuteTask，并写入工作区的 materials_text/ 目录
func TestTaskStream_MaterialsInjection(t *testing.T) {
	cfg := &config.Config{
		Env:           "test",
		AgentBackend:  "stub",
		DataDir:       t.TempDir(),
		KbDbPort:      5432,
		AdminUsername: "admin",
		AdminPassword: "admin-test-pass",
	}
	svc := bizdev.MustNewService(t, cfg)
	srv := NewServer(cfg, svc)

	ts := httptest.NewServer(srv.Routes())
	defer ts.Close()

	// 模拟前端发来的包含 materials 字段的请求
	// Content 经过 Base64 编码："用户上传文档核心需求：需要支持 1080P 超声实时渲染。"
	const rawText = "用户上传文档核心需求：需要支持 1080P 超声实时渲染。"
	b64Content := "55So5oi35LiK5Lyg5paH5qGj5qC45b+D6ZyA5rGC77ya6ZyA6KaB5pSv5oyBIDEwODBQIOi2heWjsOWunuaXtua4suafk+OAgg=="

	body, _ := json.Marshal(map[string]any{
		"task_key":      "intake",
		"project_name":  "超声材料注入测试",
		"customer_name": "北京水木东方",
		"device_type":   "超声诊断设备",
		"backend":       "stub",
		"materials": []map[string]string{
			{
				"filename": "initial_requirements.txt",
				"content":  b64Content,
			},
		},
	})

	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run/stream", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+testBearerToken(t, ts.URL, cfg))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("post tasks run stream failed: %v", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("expected status 200, got %d", resp.StatusCode)
	}

	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 := 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)
	}
	if string(content) != rawText {
		t.Fatalf("material content mismatch: got %q, want %q", string(content), rawText)
	}
}