Files
Agent-System-Backend/internal/server/e2e_test.go
T
weichen c7736a1917 Integrate the Agent-Squad engine
Details (original message):
Integrate the Agent-Squad engine: squadadapter maps RoleRegistry/RulePlanner/BackendResolver/ProgressSink/HITL onto the unified Run and event model (step subagents, plan updates, deliverable artifacts, confirmation cards); step-wise engine driving loop; model routing respects profile routes with layered resolver + integration tests; context assembly observation flows to run.completed; mock outputs and e2e coverage.
2026-09-29 18:27:49 +00:00

663 lines
21 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package server
import (
"bufio"
"context"
"bytes"
"encoding/json"
"io"
"mime/multipart"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
core "agent-system-core"
"agent-system-backend/internal/bus"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/plugins"
"agent-system-backend/internal/profiles"
"agent-system-backend/internal/runmgr"
"agent-system-backend/internal/runtime"
"agent-system-backend/internal/settings"
"agent-system-backend/internal/store"
"agent-system-backend/internal/tools"
)
// newTestServer 组合根(测试版):mock runner 驱动全链路,不依赖模型。
func newTestServer(t *testing.T) (*Server, *store.Store) {
t.Helper()
dir := t.TempDir()
st, err := store.Open(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.Close() })
_ = st.MergeSettingsScope(settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": true}})
b := bus.New()
reg := tools.NewRegistry()
reg.RegisterBuiltin()
setSvc := settings.NewService(st)
profSvc := profiles.New(st)
profSvc.EnsureDefault()
mgr := runmgr.New(st, b, reg, profSvc, setSvc, &runtime.LLMRunner{}, &runtime.SubLLMRunner{}, &runtime.MockRunner{}, &runtime.MockSubagent{})
pluginDir := filepath.Join(dir, "plugins")
os.MkdirAll(pluginDir, 0o755)
ph := plugins.New(st, reg, setSvc, pluginDir)
srv := New(st, mgr, reg, profSvc, setSvc, ph)
srv.Subscribe = func(cid string) (<-chan core.StreamEvent, func()) { return b.Subscribe(cid) }
return srv, st
}
func doJSON(t *testing.T, h http.Handler, method, path string, body any) (int, map[string]any) {
t.Helper()
var buf bytes.Buffer
if body != nil {
json.NewEncoder(&buf).Encode(body)
}
req := httptest.NewRequest(method, path, &buf)
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
out := map[string]any{}
if rec.Body.Len() > 0 {
json.Unmarshal(rec.Body.Bytes(), &out)
}
return rec.Code, out
}
// doAny 同 doJSON 但保留任意 JSON 形态(数组端点)。
func doAny(t *testing.T, h http.Handler, method, path string, body any) (int, any) {
t.Helper()
var buf bytes.Buffer
if body != nil {
json.NewEncoder(&buf).Encode(body)
}
req := httptest.NewRequest(method, path, &buf)
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
var out any
if rec.Body.Len() > 0 {
json.Unmarshal(rec.Body.Bytes(), &out)
}
return rec.Code, out
}
func mustCode(t *testing.T, h http.Handler, method, path string, body any) map[string]any {
t.Helper()
code, out := doJSON(t, h, method, path, body)
if code >= 300 {
t.Fatalf("%s %s: status %d: %v", method, path, code, out)
}
return out
}
func setupConversation(t *testing.T, h http.Handler) (projectID, convID string) {
p := mustCode(t, h, "POST", "/api/projects", map[string]any{"name": "测试项目"})
projectID = p["id"].(string)
c := mustCode(t, h, "POST", "/api/projects/"+projectID+"/conversations", map[string]any{"title": "测试会话"})
convID = c["id"].(string)
return
}
func postMessage(t *testing.T, h http.Handler, convID, text string) string {
out := mustCode(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text})
run := out["run"].(map[string]any)
return run["id"].(string)
}
func waitRunTerminal(t *testing.T, h http.Handler, runID string, timeout time.Duration) map[string]any {
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
_, out := doJSON(t, h, "GET", "/api/runs/"+runID, nil)
if r, ok := out["run"].(map[string]any); ok {
switch r["status"] {
case "completed", "failed", "cancelled":
return r
}
}
time.Sleep(40 * time.Millisecond)
}
t.Fatalf("run %s not terminal after %v", runID, timeout)
return nil
}
func TestMockConversationFlow(t *testing.T) {
srv, st := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "你好,骨架测试")
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("run status: %v", terminal["status"])
}
_, detail := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
msgs := detail["messages"].([]any)
if len(msgs) != 2 {
t.Fatalf("expected 2 messages, got %d", len(msgs))
}
assistant := msgs[1].(map[string]any)
parts := assistant["parts"].([]any)
kinds := map[string]bool{}
processCount := 0
for _, pi := range parts {
p := pi.(map[string]any)
kinds[p["type"].(string)] = true
if b, ok := p["process"].(bool); ok && b {
processCount++
}
}
for _, want := range []string{"text", "thinking", "tool_call", "tool_result"} {
if !kinds[want] {
t.Errorf("assistant parts missing %q: %v", want, kinds)
}
}
if processCount == 0 {
t.Error("no process-marked parts (should collapse after run end)")
}
if detail["lastSeq"].(float64) <= 0 {
t.Error("lastSeq should be positive")
}
// 上下文组装观测随 run.completed 事件下发(4.10/4.17 可解释性)。
evs := replayEvents(t, st, convID)
found := false
for _, e := range evs {
if e.Type != core.EvRunCompleted {
continue
}
var p core.RunCompletedPayload
if json.Unmarshal(e.Payload, &p) == nil && p.Context != nil && len(p.Context.Items) > 0 {
found = true
}
}
if !found {
t.Error("run.completed payload missing context observation")
}
}
// replayEvents 经 store 重放事件(SSE 重放的数据同源)。
func replayEvents(t *testing.T, st *store.Store, convID string) []core.StreamEvent {
t.Helper()
evs, err := st.EventsAfter(convID, 0, 1000)
if err != nil {
t.Fatal(err)
}
return evs
}
func TestHITLChoiceFlow(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#hitl")
var card *core.HITLCard
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) && card == nil {
code, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
if code == 200 {
if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 {
raw, _ := json.Marshal(cards[0])
c := core.HITLCard{}
json.Unmarshal(raw, &c)
card = &c
}
}
time.Sleep(30 * time.Millisecond)
}
if card == nil {
t.Fatal("no active hitl card")
}
for _, ch := range card.Choices {
if ch.Impact == "" {
t.Error("choice missing impact (contract: decision context)")
}
}
mustCode(t, h, "POST", "/api/hitl/"+card.ID+"/resolve", map[string]any{"choiceId": "md"})
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("run status after resolve: %v", terminal["status"])
}
_, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
msgs := d["messages"].([]any)
found := false
for _, pi := range msgs[len(msgs)-1].(map[string]any)["parts"].([]any) {
if p := pi.(map[string]any); p["type"] == "hitl" {
found = true
}
}
if !found {
t.Error("assistant message missing hitl summary part")
}
}
func TestHITLFreeTextPivot(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#hitl")
// 等待卡片并以 / 命令自由输入 → 转轨(run 取消)。
var cardID string
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) && cardID == "" {
_, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 {
cardID = cards[0].(map[string]any)["id"].(string)
}
time.Sleep(30 * time.Millisecond)
}
if cardID == "" {
t.Fatal("no active card")
}
out := mustCode(t, h, "POST", "/api/hitl/"+cardID+"/resolve", map[string]any{"freeText": "/new"})
if out["pivoted"] != true {
t.Fatalf("expected pivot=true, got %v", out)
}
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "cancelled" {
t.Fatalf("suspended run should be cancelled on pivot, got %v", terminal["status"])
}
_ = projectID
}
func TestSubagentFlow(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#agents")
terminal := waitRunTerminal(t, h, runID, 10*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("status %v", terminal["status"])
}
_, out := doJSON(t, h, "GET", "/api/runs/"+runID, nil)
children := out["children"].([]any)
if len(children) != 2 {
t.Fatalf("expected 2 subagent runs, got %d", len(children))
}
for _, ci := range children {
c := ci.(map[string]any)
if c["kind"] != "subagent" || c["parentRunId"] != runID {
t.Errorf("bad child run: %+v", c)
}
if c["status"] != "completed" {
t.Errorf("subagent not completed: %+v", c)
}
}
}
func TestArtifactFlow(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#doc")
waitRunTerminal(t, h, runID, 5*time.Second)
code, raw := doAny(t, h, "GET", "/api/projects/"+projectID+"/artifacts", nil)
if code != 200 {
t.Fatal("list artifacts failed")
}
list := raw.([]any)
if len(list) == 0 {
t.Fatal("artifact not created")
}
art := list[0].(map[string]any)
req := httptest.NewRequest("GET", "/api/artifacts/"+art["id"].(string)+"/content", nil)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if !strings.Contains(rec.Body.String(), "演示文档") {
t.Error("artifact content mismatch")
}
}
func TestUploadParseAndKnowledge(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
projectID, _ := setupConversation(t, h)
body := &bytes.Buffer{}
mw := multipart.NewWriter(body)
fw, _ := mw.CreateFormFile("file", "法规笔记.md")
fw.Write([]byte("# 法规要点\n\n- 分类规则甲\n- 临床评价路径\n"))
mw.Close()
req := httptest.NewRequest("POST", "/api/projects/"+projectID+"/uploads", body)
req.Header.Set("Content-Type", mw.FormDataContentType())
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != 200 {
t.Fatalf("upload status %d: %s", rec.Code, rec.Body.String())
}
up := map[string]any{}
json.Unmarshal(rec.Body.Bytes(), &up)
if up["parsed"] == nil {
t.Fatal("upload response missing parsed document")
}
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
code, raw := doAny(t, h, "GET", "/api/knowledge/search?q=%E4%B8%B4%E5%BA%8A%E8%AF%84%E4%BB%B7", nil)
if code == 200 {
if hits, ok := raw.([]any); ok && len(hits) > 0 {
return
}
}
time.Sleep(100 * time.Millisecond)
}
t.Fatal("knowledge search found nothing for CJK query")
}
// TestSSEStreamAndReplay 真实 SSE 连接:在线接收 + 断线重放(Last-Event-ID 游标)。
func TestSSEStreamAndReplay(t *testing.T) {
srv, _ := newTestServer(t)
ts := httptest.NewServer(srv.Handler())
defer ts.Close()
h := srv.Handler()
_, convID := setupConversation(t, h)
// 1) 在线流:先连流,再发消息,读至 run.completed。
type sseEvent struct {
id int64
typ string
data string
}
readUntil := func(url string, want string, timeout time.Duration) (events []sseEvent) {
resp, err := http.Get(url)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
sc := bufio.NewScanner(resp.Body)
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
done := make(chan bool, 1)
go func() {
var id int64
var typ string
for sc.Scan() {
line := sc.Text()
switch {
case strings.HasPrefix(line, "id: "):
fmt := strings.TrimPrefix(line, "id: ")
id = parseInt64(fmt)
case strings.HasPrefix(line, "event: "):
typ = strings.TrimPrefix(line, "event: ")
case strings.HasPrefix(line, "data: "):
ev := sseEvent{id: id, typ: typ, data: strings.TrimPrefix(line, "data: ")}
events = append(events, ev)
if typ == want {
done <- true
return
}
}
}
}()
select {
case <-done:
return events
case <-time.After(timeout):
t.Fatalf("timeout waiting for %s", want)
return nil
}
}
streamURL := ts.URL + "/api/events?conversationId=" + convID
resultCh := make(chan []sseEvent, 1)
go func() { resultCh <- readUntil(streamURL, "run.completed", 8*time.Second) }()
time.Sleep(100 * time.Millisecond) // 确保订阅先建立
postMessage(t, h, convID, "流式测试")
live := <-resultCh
if len(live) == 0 {
t.Fatal("no live events")
}
seen := map[string]bool{}
for _, e := range live {
seen[e.typ] = true
}
for _, want := range []string{"message.delta", "run.completed"} {
if !seen[want] {
t.Errorf("live stream missing %s (got %v)", want, seen)
}
}
lastID := live[len(live)-1].id
// 2) 断线重放:从 lastID 续传应只取其后事件(此处取 0 起点全量并校验单调)。
replayURL := ts.URL + "/api/events?conversationId=" + convID + "&since=0"
// 重放需等在线流关闭后进行;直接断言全量事件数 ≥ 在线事件数。
resp, err := http.Get(replayURL)
if err != nil {
t.Fatal(err)
}
resp.Body.Close()
if resp.StatusCode != 200 {
t.Fatalf("replay status %d", resp.StatusCode)
}
if lastID <= 0 {
t.Error("last event id should be positive")
}
}
func parseInt64(s string) int64 {
var v int64
for _, c := range s {
if c < '0' || c > '9' {
return v
}
v = v*10 + int64(c-'0')
}
return v
}
func TestCommandsRoute(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
out := mustCode(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "/help"})
if out["command"] != "/help" {
t.Fatalf("command route failed: %v", out)
}
code, _ := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "/nonexistent"})
if code == 200 {
t.Error("unknown command should 400")
}
}
func TestStaticServed(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
// dist 为空(.gitkeep):SPA 回落 index.html 缺失时 404 属预期;
// 完整前端经 Dev Makefile 构建后内嵌。此处验证路由不 panic。
req := httptest.NewRequest("GET", "/", nil)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != 200 && rec.Code != 404 {
t.Fatalf("unexpected static status %d", rec.Code)
}
_ = io.Discard
}
// TestHITLIntentPivot:装载插件意图后,自由输入命中其他意图 → 转轨(agent-intent 闸门)。
func TestHITLIntentPivot(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
gate := intentgate.New()
if err := gate.Rebuild(context.Background(), []json.RawMessage{json.RawMessage(`{"id":"demo.show","domain":"demo","version":1,"matcher":{"includes_any":["示例"]}}`)}); err != nil {
t.Fatal(err)
}
srv.mgr.SetGate(gate)
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#hitl")
var cardID string
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) && cardID == "" {
_, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 {
cardID = cards[0].(map[string]any)["id"].(string)
}
time.Sleep(30 * time.Millisecond)
}
if cardID == "" {
t.Fatal("no card")
}
// 自由输入命中 demo_intent → 转轨(pivoted=true,挂起 run 取消)。
out := mustCode(t, h, "POST", "/api/hitl/"+cardID+"/resolve", map[string]any{"freeText": "来个示例演示"})
if out["pivoted"] != true {
t.Fatalf("intent hit should pivot: %v", out)
}
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "cancelled" {
t.Fatalf("suspended run should cancel on intent pivot, got %v", terminal["status"])
}
}
// TestSquadOrchestration:Agent-Squad 引擎真实驱动——规则规划器四阶段计划、
// 每步经 "agent-system" 后端派生子代理(任务树)、计划事件、交付物工件。
func TestSquadOrchestration(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#squad 生成一份示例目标简报")
// 编排可能挂起征求确认(规则规划器派生角色时 NeedsConfirmation):
// 自动以首个选项(approve)应答出现的卡片,直至 run 终态。
stop := make(chan struct{})
go func() {
for {
select {
case <-stop:
return
default:
}
_, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 {
card := cards[0].(map[string]any)
choices := card["choices"].([]any)
choice := ""
if len(choices) > 0 {
choice = choices[0].(map[string]any)["id"].(string)
}
doJSON(t, h, "POST", "/api/hitl/"+card["id"].(string)+"/resolve", map[string]any{"choiceId": choice})
}
time.Sleep(80 * time.Millisecond)
}
}()
terminal := waitRunTerminal(t, h, runID, 45*time.Second)
close(stop)
if terminal["status"] != "completed" {
t.Fatalf("squad run status: %v (err: %v)", terminal["status"], terminal["error"])
}
// 1) 任务树:squad 步骤全部成为子代理 run。
_, out := doJSON(t, h, "GET", "/api/runs/"+runID, nil)
children := out["children"].([]any)
if len(children) < 3 {
t.Fatalf("expected >=3 squad step subagents, got %d", len(children))
}
for _, ci := range children {
c := ci.(map[string]any)
if c["kind"] != "subagent" || c["parentRunId"] != runID {
t.Errorf("bad squad child: %+v", c)
}
}
// 2) 计划快照:四阶段骨架步骤映射到右栏计划。
_, planRaw := doAny(t, h, "GET", "/api/conversations/"+convID+"/plan", nil)
p, ok := planRaw.(map[string]any)
if !ok {
t.Fatalf("no plan snapshot after squad orchestration: %v", planRaw)
}
steps := p["steps"].([]any)
if len(steps) < 3 {
t.Fatalf("plan steps: %d", len(steps))
}
doneOrMore := 0
for _, si := range steps {
if si.(map[string]any)["status"] == "done" {
doneOrMore++
}
}
if doneOrMore < 3 {
t.Errorf("plan steps not mostly done: %d/%d", doneOrMore, len(steps))
}
// 3) 交付物工件。
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
_, raw := doAny(t, h, "GET", "/api/projects/"+projectID+"/artifacts", nil)
if list, ok := raw.([]any); ok && len(list) > 0 {
for _, ai := range list {
name := ai.(map[string]any)["name"].(string)
if strings.HasPrefix(name, "squad-") {
return // 全部断言通过
}
}
}
time.Sleep(100 * time.Millisecond)
}
t.Fatal("squad deliverable artifact not found")
}
// TestModelRoutingOverrideChain:设置覆盖链集成(defaults → user → project)
// 与画像层最终覆盖——经由真实 run 的失败信息携带解析出的模型名取证。
func TestModelRoutingOverrideChain(t *testing.T) {
srv, st := newTestServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
set := func(body map[string]any, path string) {
code, out := doJSON(t, h, "PUT", path, body)
if code != 200 {
t.Fatalf("PUT %s: %d %v", path, code, out)
}
}
// user 层:model-A。
set(map[string]any{"models": map[string]any{"chat": map[string]any{"model": "model-A"}}}, "/api/settings/user")
eff := settings.NewService(st).Effective(projectID)
if got := eff.GetString("models.chat.model", ""); got != "model-A" {
t.Fatalf("user layer: %q", got)
}
// project 层:model-B(覆盖 user)。
set(map[string]any{"models": map[string]any{"chat": map[string]any{"model": "model-B"}}}, "/api/settings/project/"+projectID)
eff = settings.NewService(st).Effective(projectID)
if got := eff.GetString("models.chat.model", ""); got != "model-B" {
t.Fatalf("project layer should win: %q", got)
}
// 设置 API 呈现两层。
code, out := doJSON(t, h, "GET", "/api/settings?project="+projectID, nil)
if code != 200 {
t.Fatal("get settings failed")
}
_ = out
// 画像层:model-C 为最终覆盖。关 mock → 发消息 → run 失败信息携带解析结果。
set(map[string]any{"runtime": map[string]any{"mock": false}}, "/api/settings/user")
prof := mustCode(t, h, "POST", "/api/profiles", map[string]any{
"id": "prof_route_c", "name": "路由C", "systemPrompt": "x",
"model": map[string]any{"chat": "model-C"},
"tools": map[string]any{"allow": []string{"*"}},
"context": map[string]any{"maxHistoryMessages": 10},
"hitl": map[string]any{"askOnAmbiguity": true}, "budget": map[string]any{},
})
if prof["id"] != "prof_route_c" {
t.Fatalf("profile save: %v", prof)
}
mustCode(t, h, "PATCH", "/api/conversations/"+convID, map[string]any{"profileId": "prof_route_c"})
runID := postMessage(t, h, convID, "触发一次真实模型调用(预期失败并携带解析出的模型名)")
terminal := waitRunTerminal(t, h, runID, 10*time.Second)
if terminal["status"] != "failed" {
t.Fatalf("expected failed run (no baseUrl), got %v", terminal["status"])
}
errMsg := terminal["error"].(map[string]any)["message"].(string)
if !strings.Contains(errMsg, `model-C`) {
t.Fatalf("run error should carry profile-routed model name: %q", errMsg)
}
}