Files
Agent-System-Backend/internal/server/e2e_test.go
T
weichenandCursor 51b1ff6923
ci / vet-and-test (pull_request) Successful in 42s
Replay an open run from its first event
Why: Refreshing or switching back only knew the latest event cursor, so process that had already happened was missing until the run finished.
Evidence: GOWORK=off go test -count=1 -run TestConversationResumeSeq ./internal/server passed. While a hitl run waits, resumeSeq is the run's first seq minus one and is less than lastSeq; after the run completes the two cursors match.
Impact: Conversation detail gains resumeSeq beside lastSeq. No core contract change. Clients that ignore the field keep the old cursor.
Refs: Dev fix/live-run-visibility
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-03 15:21:29 +00:00

1136 lines
37 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 (
"archive/zip"
"bufio"
"bytes"
"context"
"encoding/json"
agentcommand "agent-command"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/cmdbus"
"io"
"mime/multipart"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
"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"
core "agent-system-core"
)
// newTestServer 组合根(测试版):mock runner 驱动全链路,不依赖模型。
func newTestServer(t *testing.T) (*Server, *store.Store) {
t.Helper()
dir := t.TempDir()
st, err := store.OpenSQLite(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.Close() })
_ = st.MergeSettingsScope(store.OwnerDefault, 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)
// 生产同款装配(app.go):卡片事件经 runmgr 事件缓冲持久化/发布——项目流
//(project:<id> 伪键,D50)依赖它;此前为 no-op emitter,事件不入库。
cardSvc := cardsvc.New(st, mgr.EmitCardEvent)
srv.Cards = cardSvc
mgr.SetCards(cardSvc) // run 内产卡(摄取/编排/submit_card)经 runmgr,与 app.go 装配一致
srv.Bus = cmdbus.New(st, srv.Cards, nil)
srv.WireBuiltinCommands()
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
}
// TestMockCardflowFullLoop 卡流全链路(D39/D35/D32):mock #cardflow → result 卡入收件箱
// → 助手消息含 cardId 引用部件 → finishRun 回填锚点(卡→消息互定位)。
func TestMockCardflowFullLoop(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#cardflow")
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("run status: %v", terminal["status"])
}
// 收件箱出现 result 卡。
_, cardsResp := doJSON(t, h, "GET", "/api/conversations/"+convID+"/cards", nil)
cards, _ := cardsResp["cards"].([]any)
var resultCard map[string]any
for _, ci := range cards {
c := ci.(map[string]any)
if c["kind"] == "result" && c["title"] == "会议纪要 · 第 12 次周会" {
resultCard = c
}
}
if resultCard == nil {
t.Fatalf("result card missing from inbox: %v", cards)
}
// 助手消息含 cardId 引用部件(卡片为产物权威,消息只存引用)。
_, detail := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
msgs := detail["messages"].([]any)
assistant := msgs[len(msgs)-1].(map[string]any)
cardID := resultCard["id"].(string)
refFound := false
for _, pi := range assistant["parts"].([]any) {
p := pi.(map[string]any)
if p["type"] != "card" {
continue
}
cp := p["card"].(map[string]any)
if cp["cardId"] == cardID {
refFound = true
}
}
if !refFound {
t.Errorf("assistant message missing card-part with cardId=%s", cardID)
}
// finishRun 回填锚点:卡→消息互定位(D32 方向)。
if anchor, _ := resultCard["anchorMessageId"].(string); anchor != assistant["id"].(string) {
t.Errorf("card anchorMessageId = %q, want assistant message %q", anchor, assistant["id"])
}
}
// TestMockCardflowStreamLoop 流式产卡全链路(CF-S/D44-D45):mock #cardstream →
// card_begin(active 卡入收件箱)→ card_append×N(card.updated 增量)→
// card_commit(done + cardId 引用部件 + finishRun 锚点回填),且正文逐段合并。
func TestMockCardflowStreamLoop(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#cardstream")
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("run status: %v", terminal["status"])
}
// 收件箱出现 result 卡(commit 后终态 done)。
_, cardsResp := doJSON(t, h, "GET", "/api/conversations/"+convID+"/cards", nil)
cards, _ := cardsResp["cards"].([]any)
var streamCard map[string]any
for _, ci := range cards {
c := ci.(map[string]any)
if c["kind"] == "result" && c["title"] == "项目周报 · 流式" {
streamCard = c
}
}
if streamCard == nil {
t.Fatalf("stream card missing from inbox: %v", cards)
}
if streamCard["status"] != "done" {
t.Errorf("stream card status = %v, want done", streamCard["status"])
}
// 正文逐段合并(begin 的 kv 摘要 + 三段 append)。
body, _ := streamCard["body"].([]any)
full := ""
for _, bi := range body {
n := bi.(map[string]any)
if n["type"] == "markdown" {
full += n["markdown"].(string)
}
}
for _, seg := range []string{"## 进展", "## 风险", "## 下周", "artifact 统一"} {
if !strings.Contains(full, seg) {
t.Errorf("stream body missing %q, got %q", seg, full)
}
}
// 助手消息含 cardId 引用部件 + 锚点回填。
_, detail := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
msgs := detail["messages"].([]any)
assistant := msgs[len(msgs)-1].(map[string]any)
cardID := streamCard["id"].(string)
refFound := false
for _, pi := range assistant["parts"].([]any) {
p := pi.(map[string]any)
if p["type"] != "card" {
continue
}
cp := p["card"].(map[string]any)
if cp["cardId"] == cardID {
refFound = true
}
}
if !refFound {
t.Errorf("assistant message missing card-part with cardId=%s", cardID)
}
if anchor, _ := streamCard["anchorMessageId"].(string); anchor != assistant["id"].(string) {
t.Errorf("stream card anchorMessageId = %q, want assistant message %q", anchor, assistant["id"])
}
}
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 TestConversationResumeSeq(t *testing.T) {
srv, st := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#hitl")
deadline := time.Now().Add(5 * time.Second)
var waiting map[string]any
for time.Now().Before(deadline) && waiting == nil {
code, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
if code == 200 {
if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 {
waiting = d
}
}
time.Sleep(30 * time.Millisecond)
}
if waiting == nil {
t.Fatal("no active hitl card")
}
lastSeq := int64(waiting["lastSeq"].(float64))
resumeSeq := int64(waiting["resumeSeq"].(float64))
var first int64
for _, e := range replayEvents(t, st, convID) {
if string(e.RunID) == runID {
first = e.Seq
break
}
}
if first == 0 {
t.Fatal("run has no events")
}
if resumeSeq != first-1 {
t.Fatalf("resumeSeq = %d, want first-1 = %d", resumeSeq, first-1)
}
if resumeSeq >= lastSeq {
t.Fatalf("resumeSeq %d should be before lastSeq %d while the run is open", resumeSeq, lastSeq)
}
cardID := waiting["activeHitl"].([]any)[0].(map[string]any)["id"].(string)
mustCode(t, h, "POST", "/api/hitl/"+cardID+"/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"])
}
_, settled := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
if int64(settled["resumeSeq"].(float64)) != int64(settled["lastSeq"].(float64)) {
t.Fatalf("resumeSeq = %v, lastSeq = %v, want them equal after the run ends", settled["resumeSeq"], settled["lastSeq"])
}
}
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)
// mock #doc 走 docgen:内容是 zip 容器,标题/表格排版在 word/document.xml 里。
body := rec.Body.Bytes()
zr, err := zip.NewReader(bytes.NewReader(body), int64(len(body)))
if err != nil {
t.Fatalf("artifact 不是合法 docx 容器: %v", err)
}
var docXML []byte
for _, f := range zr.File {
if f.Name == "word/document.xml" {
rc, oerr := f.Open()
if oerr == nil {
docXML, _ = io.ReadAll(rc)
rc.Close()
}
}
}
if !strings.Contains(string(docXML), "演示会议纪要") || !strings.Contains(string(docXML), "<w:tbl>") {
t.Error("docx 缺少标题/表格排版(仍是纯文本?)")
}
}
// TestArtifactFacadeCard 工件门面卡全链路(CF-A/D47):mock #doc → 工件落库 +
// artifact_ref 部件 → 门面卡入收件箱(摘要 kv + doc-ref)→ 消息含 card 引用 →
// finishRun 锚点回填。
func TestArtifactFacadeCard(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#doc")
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("run status: %v", terminal["status"])
}
_, cardsResp := doJSON(t, h, "GET", "/api/conversations/"+convID+"/cards", nil)
cards, _ := cardsResp["cards"].([]any)
var facade map[string]any
for _, ci := range cards {
c := ci.(map[string]any)
if c["kind"] == "result" && c["title"] == "工具大结果 · demo-会议纪要.docx" {
facade = c
}
}
if facade == nil {
t.Fatalf("facade card missing from inbox: %v", cards)
}
if facade["status"] != "done" {
t.Errorf("facade status = %v, want done", facade["status"])
}
hasDocRef := false
for _, bi := range facade["body"].([]any) {
n := bi.(map[string]any)
if n["type"] == "doc-ref" && n["docId"] != "" {
hasDocRef = true
}
}
if !hasDocRef {
t.Error("facade body missing doc-ref node")
}
_, detail := doJSON(t, h, "GET", "/api/conversations/"+convID, nil)
msgs := detail["messages"].([]any)
assistant := msgs[len(msgs)-1].(map[string]any)
cardID := facade["id"].(string)
hasRef := false
for _, pi := range assistant["parts"].([]any) {
p := pi.(map[string]any)
if p["type"] == "card" && p["card"].(map[string]any)["cardId"] == cardID {
hasRef = true
}
}
if !hasRef {
t.Errorf("assistant message missing facade card ref %s", cardID)
}
if anchor, _ := facade["anchorMessageId"].(string); anchor != assistant["id"].(string) {
t.Errorf("facade anchorMessageId = %q, want %q", anchor, assistant["id"])
}
}
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 TestDangerCommandConfirmFlowE2E(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
// 注册一条高危命令(模拟插件/扩展注册),API 入口 button 直发。
_ = srv.Bus.RegisterBuiltin(agentcommand.CommandSpec{
ID: "e2e.destroy", Title: "销毁演示", Danger: agentcommand.DangerHigh,
ParamsSchema: json.RawMessage(`{"type":"object","required":["what"],"properties":{"what":{"type":"string"}}}`),
}, "", func(ctx context.Context, req cmdbus.Request) (any, error) {
w, _ := req.Params["what"].(string)
return map[string]any{"destroyed": w}, nil
})
// 清单可见(数组契约,legacy {name,description} 兼容):原始请求解码。
req := httptest.NewRequest("GET", "/api/commands", nil)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != 200 {
t.Fatalf("list commands: %d", rec.Code)
}
var cl []map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &cl); err != nil {
t.Fatalf("list decode: %v", err)
}
found, legacySlash := false, false
for _, m := range cl {
if m["id"] == "e2e.destroy" && m["danger"] == "high" {
found = true
}
if m["name"] == "/help" {
legacySlash = true
}
}
if !found || !legacySlash {
t.Fatalf("listing wrong (found=%v legacy=%v): %v", found, legacySlash, cl)
}
// button 入口未确认 → 确认卡 + accepted 回执。
code, out := doJSON(t, h, "POST", "/api/commands/e2e.destroy", map[string]any{
"params": map[string]any{"what": "世界"}, "entrance": "button", "conversationId": convID})
if code != 200 {
t.Fatalf("dispatch: %d %v", code, out)
}
card, _ := out["card"].(map[string]any)
if card == nil {
t.Fatalf("confirm card missing: %v", out)
}
cardID, _ := card["id"].(string)
// 卡上确认 → 执行 committed + 审计可查。
code, out2 := doJSON(t, h, "POST", "/api/cards/"+cardID+"/actions/confirm", map[string]any{})
if code != 200 {
t.Fatalf("confirm action: %d %v", code, out2)
}
if rec, _ := out2["receipt"].(map[string]any); rec == nil || rec["state"] != "committed" {
t.Fatalf("confirm receipt: %v", out2)
}
code, audits := doJSON(t, h, "GET", "/api/commands/audit?conversationId="+convID, nil)
if code != 200 {
t.Fatalf("audit query: %d", code)
}
rows, _ := audits["audits"].([]any)
if len(rows) < 2 {
t.Fatalf("audit rows = %d, want >=2 (accepted+committed)", len(rows))
}
states := []string{}
for _, r := range rows {
m, _ := r.(map[string]any)
st, _ := m["state"].(string)
states = append(states, st)
}
if states[1] != "accepted" || states[0] != "committed" {
t.Fatalf("audit order wrong: %v (latest first)", states)
}
}
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(store.OwnerDefault, 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(store.OwnerDefault, 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)
}
}
// TestTaskListAndFlags 方案 A R1:/api/me 身份、跨项目任务列表与置顶/归档 PATCH。
func TestTaskListAndFlags(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
p1 := mustCode(t, h, "POST", "/api/projects", map[string]any{"name": "P1"})
p2 := mustCode(t, h, "POST", "/api/projects", map[string]any{"name": "P2"})
c1 := mustCode(t, h, "POST", "/api/projects/"+p1["id"].(string)+"/conversations", map[string]any{"title": "t1"})
c2 := mustCode(t, h, "POST", "/api/projects/"+p2["id"].(string)+"/conversations", map[string]any{"title": "t2"})
// 身份:嵌入式档(无认证中间件)→ 默认租户,auth=false。
code, me := doJSON(t, h, "GET", "/api/me", nil)
if code != 200 || me["owner"] != store.OwnerDefault || me["auth"] != false {
t.Fatalf("me: code=%d owner=%v auth=%v", code, me["owner"], me["auth"])
}
if s, _ := me["name"].(string); s == "" {
t.Fatalf("me.name must be non-empty, got %v", me["name"])
}
// 跨项目任务列表:两个项目的会话都在。
code, list := doAny(t, h, "GET", "/api/conversations", nil)
if code != 200 {
t.Fatalf("listTasks code=%d", code)
}
arr, _ := list.([]any)
if len(arr) != 2 {
t.Fatalf("task list must span projects, got %d", len(arr))
}
// 置顶 t2 → 列表首;归档 t1 → 默认隐藏、?archived=1 可见。
mustCode(t, h, "PATCH", "/api/conversations/"+c2["id"].(string), map[string]any{"pinned": true})
mustCode(t, h, "PATCH", "/api/conversations/"+c1["id"].(string), map[string]any{"archived": true})
_, list = doAny(t, h, "GET", "/api/conversations", nil)
arr, _ = list.([]any)
if len(arr) != 1 || arr[0].(map[string]any)["id"] != c2["id"] || arr[0].(map[string]any)["pinned"] != true {
t.Fatalf("pinned-first and archived-hidden expected, got %+v", list)
}
_, list = doAny(t, h, "GET", "/api/conversations?archived=1", nil)
arr, _ = list.([]any)
if len(arr) != 2 {
t.Fatalf("archived=1 must include archived, got %d", len(arr))
}
// 解除归档恢复可见。
mustCode(t, h, "PATCH", "/api/conversations/"+c1["id"].(string), map[string]any{"archived": false})
_, list = doAny(t, h, "GET", "/api/conversations", nil)
arr, _ = list.([]any)
if len(arr) != 2 {
t.Fatalf("unarchive must restore visibility, got %d", len(arr))
}
// 详情读取带标记(对账权威路径)。
det := mustCode(t, h, "GET", "/api/conversations/"+c2["id"].(string), nil)
if det["conversation"].(map[string]any)["pinned"] != true {
t.Fatalf("detail must carry pinned flag, got %+v", det["conversation"])
}
}
// TestProjectInbox 项目收件箱全链路(CF3/D48-D51):#cardflow 会话卡 → REST
// to_project(幂等)→ 项目聚合列表(副本可见、原卡 dismissed)→ 项目流持久化
// 事件(card.created,信封盖章 ProjectID)。
func TestProjectInbox(t *testing.T) {
srv, st := newTestServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := postMessage(t, h, convID, "#cardflow")
terminal := waitRunTerminal(t, h, runID, 5*time.Second)
if terminal["status"] != "completed" {
t.Fatalf("run status: %v", terminal["status"])
}
_, cardsResp := doJSON(t, h, "GET", "/api/conversations/"+convID+"/cards", nil)
var cardID string
for _, ci := range cardsResp["cards"].([]any) {
c := ci.(map[string]any)
if c["kind"] == "result" && c["title"] == "会议纪要 · 第 12 次周会" {
cardID = c["id"].(string)
}
}
if cardID == "" {
t.Fatal("result card missing")
}
// 留档到项目:副本项目级、原卡 dismissed。
code, raw := doAny(t, h, "POST", "/api/cards/"+cardID+"/to_project", nil)
if code != 200 {
t.Fatalf("to_project: %d %v", code, raw)
}
resp := raw.(map[string]any)
copyCard := resp["card"].(map[string]any)
if copyCard["conversationId"] != "" && copyCard["conversationId"] != nil {
t.Errorf("副本应为项目级(conversationId 空),got %v", copyCard["conversationId"])
}
if origin := resp["origin"].(map[string]any); origin["status"] != "dismissed" {
t.Errorf("原卡应 dismissed, got %v", origin["status"])
}
// 幂等:重复留档返回同一副本。
code2, raw2 := doAny(t, h, "POST", "/api/cards/"+cardID+"/to_project", nil)
if code2 != 200 || raw2.(map[string]any)["card"].(map[string]any)["id"] != copyCard["id"] {
t.Errorf("重复留档应幂等: %d %v", code2, raw2)
}
// 项目聚合列表含副本。
_, pc := doJSON(t, h, "GET", "/api/projects/"+projectID+"/cards", nil)
found := false
for _, ci := range pc["cards"].([]any) {
c := ci.(map[string]any)
if c["id"] == copyCard["id"] {
found = true
}
}
if !found {
t.Error("project cards missing archived copy")
}
// 项目流事件持久化 + 信封盖章 ProjectID(D50)。
evs, err := st.EventsAfter("project:"+projectID, 0, 100)
if err != nil {
t.Fatal(err)
}
created := false
for _, e := range evs {
if e.Type == core.EvCardCreated && e.ProjectID == core.ID(projectID) {
created = true
}
}
if !created {
t.Errorf("project stream missing card.created with ProjectID, got %d events", len(evs))
}
// scope 校验:未知项目 → NotFound(不泄露存在性)。
codeX, _ := doAny(t, h, "GET", "/api/events?scope=project&projectId=prj_nonexistent", nil)
if codeX != 404 && codeX != 500 {
t.Errorf("unknown project subscribe should not leak, got %d", codeX)
}
}