weichenandCursor
b97d6cd60b
Why: bootstrap-admin now refuses a database that has no agent_system_identity row, and serve does not insert one.
Evidence: Dev CI run 3266 job 5008 failed with system identity missing on 7eef723.
Impact: test setup only.
Refs: Agent-System-Dev pulls/558
Co-authored-by: Cursor <cursoragent@cursor.com>
708 lines
23 KiB
Go
708 lines
23 KiB
Go
// 协作存储 U7：双进程（真实 backend serve ×2 共享同一 PG）黑盒验收。
|
||
// 对应方案 §7 最小验收矩阵的进程级行：
|
||
// - 两实例同会话不同请求 → 仅一个主执行，另一个明确 409（不误写）
|
||
// - 在 B 提交 A 的 HITL / 取消 → 命令持久、唯一消费、到达执行者
|
||
// - 执行实例被杀 → 租约期内不误抢（明确 409），过期后另一实例受控接管
|
||
// - 实例重启 → 数据不变量保持（消息在、事件 seq 连续无洞）
|
||
// - 跨实例读一致与事件传播（A 写 → B 的 SSE 可见）
|
||
//
|
||
// 资源边界：测试自建临时数据库、自起进程、自回收（方案 §5 U7 的隔离测试
|
||
// 入口要求）。门控：AGENT_PG_TEST_DSN（真 PG；CI test-pg job 必跑）与
|
||
// AGENT_BIN（backend 二进制；由 make test-multi-instance 构建并传入）。
|
||
// 不打任何生产服务。
|
||
package multiinstance
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"mime/multipart"
|
||
"net"
|
||
"net/http"
|
||
"net/http/cookiejar"
|
||
"net/url"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"syscall"
|
||
"testing"
|
||
"time"
|
||
|
||
_ "github.com/jackc/pgx/v5/stdlib"
|
||
)
|
||
|
||
const (
|
||
aliceOwner = "mi_alice"
|
||
alicePass = "mi-alice-pass-1"
|
||
httpWait = 15 * time.Second
|
||
)
|
||
|
||
type instance struct {
|
||
name string
|
||
base string
|
||
cmd *exec.Cmd
|
||
log *lockedBuffer
|
||
}
|
||
|
||
type lockedBuffer struct {
|
||
mu sync.Mutex
|
||
buf bytes.Buffer
|
||
}
|
||
|
||
func (b *lockedBuffer) Write(p []byte) (int, error) {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
return b.buf.Write(p)
|
||
}
|
||
|
||
func (b *lockedBuffer) String() string {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
return b.buf.String()
|
||
}
|
||
|
||
type hitlCard struct{ id, runID string }
|
||
|
||
// userClient 持 alice 在两实例的会话（A：api；B：apiB/streamB）。
|
||
// 会话表在共享 PG，同账号可在两实例同时登录。
|
||
type userClient struct {
|
||
api *http.Client // alice@A 普通请求（带超时）
|
||
apiB *http.Client // alice@B 普通请求
|
||
streamB *http.Client // alice@B 流式（SSE：无整体超时，chan 背压控制）
|
||
}
|
||
|
||
type env struct {
|
||
t *testing.T
|
||
dbName string
|
||
dbDSN string
|
||
artsDir string
|
||
bin string
|
||
a, b *instance
|
||
uc *userClient
|
||
}
|
||
|
||
func TestMain(m *testing.M) {
|
||
if os.Getenv("AGENT_PG_TEST_DSN") == "" {
|
||
fmt.Println("SKIP: AGENT_PG_TEST_DSN 未设置（双进程验收由 CI test-pg job 必跑覆盖）")
|
||
return
|
||
}
|
||
if os.Getenv("AGENT_BIN") == "" {
|
||
fmt.Println("SKIP: AGENT_BIN 未设置（由 make test-multi-instance 构建并传入 agent-system 二进制路径）")
|
||
return
|
||
}
|
||
// 子进程用独立 cwd（避免 data/ 落测试目录），二进制路径必须先转绝对。
|
||
bin, err := filepath.Abs(os.Getenv("AGENT_BIN"))
|
||
if err != nil {
|
||
fmt.Println("AGENT_BIN abs:", err)
|
||
os.Exit(1)
|
||
}
|
||
os.Setenv("AGENT_BIN", bin)
|
||
os.Exit(m.Run())
|
||
}
|
||
|
||
func TestTwoInstancesOverSharedPG(t *testing.T) {
|
||
e := newEnv(t)
|
||
e.startPair()
|
||
|
||
uc := e.loginAlice(t)
|
||
e.uc = uc
|
||
projID, convID := e.setupConversation(t, uc)
|
||
|
||
t.Run("cross_read", func(t *testing.T) {
|
||
// A 建的资源在 B 立即可见（共享权威，无实例本地视图缺口）。
|
||
var detail map[string]any
|
||
e.do(t, uc.apiB, "GET", "/api/conversations/"+convID, nil, &detail)
|
||
if conv, _ := detail["conversation"].(map[string]any); conv == nil || conv["id"] != convID {
|
||
t.Fatalf("B cannot read conversation created via A: %v", detail)
|
||
}
|
||
})
|
||
|
||
t.Run("cross_artifact_roundtrip", func(t *testing.T) {
|
||
// 未配置对象存储时，A 上传的正文必须落在共享本地目录，B 原样取回。
|
||
payload := []byte("k4-cross-instance-artifact")
|
||
id := e.uploadArtifact(t, uc.api, projID, "k4.txt", payload)
|
||
code, raw := e.raw(t, uc.apiB, "GET", "/api/artifacts/"+id+"/content", nil)
|
||
if code != http.StatusOK {
|
||
t.Fatalf("B artifact content: status %d, body %.300s", code, raw)
|
||
}
|
||
if sha256.Sum256([]byte(raw)) != sha256.Sum256(payload) {
|
||
t.Fatalf("B downloaded different bytes")
|
||
}
|
||
})
|
||
|
||
t.Run("cross_event_sse", func(t *testing.T) {
|
||
lastSeq := e.lastSeq(t, uc.apiB, convID)
|
||
events := e.subscribe(t, uc.streamB, convID, lastSeq)
|
||
e.postMessage(t, uc.api, convID, "hello from A")
|
||
select {
|
||
case ev, ok := <-events:
|
||
if !ok {
|
||
t.Fatal("SSE stream closed before any event")
|
||
}
|
||
if ev.seq <= lastSeq {
|
||
t.Fatalf("stale event replayed: seq %d <= since %d", ev.seq, lastSeq)
|
||
}
|
||
case <-time.After(8 * time.Second):
|
||
t.Fatal("event written via A not visible on B SSE within 8s")
|
||
}
|
||
})
|
||
|
||
t.Run("same_session_conflict_then_cross_hitl", func(t *testing.T) {
|
||
// A 发 #hitl：mock runner 挂起 run（waiting_hitl），会话被 A 的执行租约占用。
|
||
e.postMessage(t, uc.api, convID, "#hitl")
|
||
card := e.waitActiveHITL(t, uc.apiB, convID)
|
||
|
||
// B 同会话并发提交 → 明确 409，而不是排队后误写或静默接受。
|
||
code, body := e.raw(t, uc.apiB, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "from B while A holds"})
|
||
if code != http.StatusConflict {
|
||
t.Fatalf("concurrent submit on busy session: status %d (want 409), body %.200s", code, body)
|
||
}
|
||
|
||
// B 跨实例 resolve A 的 HITL → 持久命令转交执行者（202 via command；
|
||
// 执行实例本地 resolve 才是 200），run 随后走完。
|
||
code, body = e.raw(t, uc.apiB, "POST", "/api/hitl/"+card.id+"/resolve", map[string]any{"choiceId": "md"})
|
||
if code != http.StatusOK && code != http.StatusAccepted {
|
||
t.Fatalf("cross-instance resolve: status %d, body %.200s", code, body)
|
||
}
|
||
e.waitRunStatus(t, uc.apiB, card.runID, "completed")
|
||
// run 收口后会话释放：B 可推进。
|
||
e.postMessage(t, uc.apiB, convID, "B proceeds after release")
|
||
})
|
||
|
||
t.Run("cross_cancel", func(t *testing.T) {
|
||
e.postMessage(t, uc.api, convID, "#hitl")
|
||
card := e.waitActiveHITL(t, uc.apiB, convID)
|
||
// 跨实例取消经命令总线受理（202 via command；执行实例本地取消为 200）。
|
||
code, body := e.raw(t, uc.apiB, "POST", "/api/runs/"+card.runID+"/cancel", nil)
|
||
if code != http.StatusOK && code != http.StatusAccepted {
|
||
t.Fatalf("cross-instance cancel: status %d, body %.200s", code, body)
|
||
}
|
||
e.waitRunStatus(t, uc.apiB, card.runID, "cancelled")
|
||
e.postMessage(t, uc.apiB, convID, "B proceeds after cancel")
|
||
})
|
||
|
||
t.Run("owner_crash_takeover", func(t *testing.T) {
|
||
e.postMessage(t, uc.api, convID, "#hitl")
|
||
_ = e.waitActiveHITL(t, uc.api, convID)
|
||
|
||
e.killNine(t, e.a)
|
||
|
||
// 会话执行租约（15 分钟档）期内不误抢：B 对该会话的提交得到明确
|
||
// 409（有界失败，不是写坏或挂死）；过期接管语义由库级围栏测试覆盖
|
||
// （backend runmgr multi_instance_test），进程级不空等 15 分钟。
|
||
code, body := e.raw(t, uc.apiB, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "too early"})
|
||
if code != http.StatusConflict {
|
||
t.Fatalf("submit inside dead owner's lease: status %d (want 409), body %.200s", code, body)
|
||
}
|
||
// 死实例不拖累其他会话：B 在新会话上正常推进（公平有界，R8）。
|
||
var conv2 map[string]any
|
||
e.do(t, uc.apiB, "POST", "/api/projects/"+projID+"/conversations", map[string]any{"title": "不受拖累"}, &conv2)
|
||
e.do(t, uc.apiB, "POST", "/api/conversations/"+conv2["id"].(string)+"/messages", map[string]any{"text": "B unaffected"}, nil)
|
||
// 整机失效恢复剧本：原实例以同一身份重启，启动对账把遗留 run 收口
|
||
// （interrupted），随后原实例重新占用会话并推进。
|
||
e.a = e.startInstance(t, e.a.name, strings.TrimPrefix(e.a.base, "http://"))
|
||
var detail map[string]any
|
||
deadline := time.Now().Add(httpWait)
|
||
for time.Now().Before(deadline) {
|
||
e.do(t, uc.api, "GET", "/api/conversations/"+convID, nil, &detail)
|
||
if runs, ok := detail["runs"].([]any); ok && len(runs) > 0 {
|
||
last, _ := runs[len(runs)-1].(map[string]any)
|
||
if last["status"] == "interrupted" {
|
||
break
|
||
}
|
||
}
|
||
time.Sleep(300 * time.Millisecond)
|
||
}
|
||
e.do(t, uc.api, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "A resumes after restart"}, nil)
|
||
})
|
||
|
||
t.Run("restart_invariants", func(t *testing.T) {
|
||
e.killNine(t, e.b)
|
||
e.b = e.startInstance(t, e.b.name, strings.TrimPrefix(e.b.base, "http://"))
|
||
|
||
var detail map[string]any
|
||
e.do(t, uc.apiB, "GET", "/api/conversations/"+convID, nil, &detail)
|
||
msgs, _ := detail["messages"].([]any)
|
||
if len(msgs) < 5 {
|
||
t.Fatalf("messages lost after restart: %d", len(msgs))
|
||
}
|
||
lastSeq := e.lastSeq(t, uc.apiB, convID)
|
||
if lastSeq == 0 {
|
||
t.Fatal("event seq lost after restart")
|
||
}
|
||
// SSE 从 0 全量重放：seq 严格连续，无洞（R5 不变量在进程失效后的视图）。
|
||
events := e.subscribe(t, uc.streamB, convID, 0)
|
||
seen := int64(0)
|
||
deadline := time.After(15 * time.Second)
|
||
for seen < lastSeq {
|
||
select {
|
||
case ev, ok := <-events:
|
||
if !ok {
|
||
t.Fatalf("replay stream closed early at seq %d (want up to %d)", seen, lastSeq)
|
||
}
|
||
if ev.seq != seen+1 {
|
||
t.Fatalf("replay gap: got seq %d after %d", ev.seq, seen)
|
||
}
|
||
seen = ev.seq
|
||
case <-deadline:
|
||
t.Fatalf("replay stalled at seq %d / %d", seen, lastSeq)
|
||
}
|
||
}
|
||
// 重启后写入仍可推进。
|
||
e.postMessage(t, uc.apiB, convID, "after restart")
|
||
})
|
||
}
|
||
|
||
// ---- 环境与进程 ----
|
||
|
||
func newEnv(t *testing.T) *env {
|
||
t.Helper()
|
||
dsn := os.Getenv("AGENT_PG_TEST_DSN")
|
||
admin, err := sql.Open("pgx", dsn)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
dbName := "mi_" + strconv.Itoa(os.Getpid()) + "_" + fmt.Sprint(time.Now().UnixNano()%1e6)
|
||
if _, err := admin.Exec("CREATE DATABASE " + dbName); err != nil {
|
||
admin.Close()
|
||
t.Fatalf("create test database: %v", err)
|
||
}
|
||
t.Cleanup(func() {
|
||
_, _ = admin.Exec("DROP DATABASE IF EXISTS " + dbName + " WITH (FORCE)")
|
||
admin.Close()
|
||
})
|
||
testDSN, err := swapDB(dsn, dbName)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
root := t.TempDir()
|
||
return &env{t: t, dbName: dbName, dbDSN: testDSN, artsDir: filepath.Join(root, "artifacts"), bin: os.Getenv("AGENT_BIN")}
|
||
}
|
||
|
||
// swapDB 替换 DSN 的库名，保留其余参数（sslmode 等）。
|
||
func swapDB(dsn, dbName string) (string, error) {
|
||
u, err := url.Parse(dsn)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
u.Path = "/" + dbName
|
||
return u.String(), nil
|
||
}
|
||
|
||
func freePort(t *testing.T) string {
|
||
t.Helper()
|
||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer l.Close()
|
||
_, port, _ := net.SplitHostPort(l.Addr().String())
|
||
return port
|
||
}
|
||
|
||
func (e *env) startPair() {
|
||
e.t.Helper()
|
||
e.a = e.startInstance(e.t, "miA", "127.0.0.1:"+freePort(e.t))
|
||
e.b = e.startInstance(e.t, "miB", "127.0.0.1:"+freePort(e.t))
|
||
}
|
||
|
||
func (e *env) startInstance(t *testing.T, name, addr string) *instance {
|
||
t.Helper()
|
||
cmd := exec.Command(e.bin, "serve",
|
||
"--dsn", e.dbDSN,
|
||
"--addr", addr,
|
||
"--tier", "team",
|
||
"--auth", "on",
|
||
"--mock",
|
||
"--instance", name,
|
||
"--artifacts", e.artsDir,
|
||
"--disable-exec-tools",
|
||
)
|
||
log := &lockedBuffer{}
|
||
cmd.Stdout, cmd.Stderr = log, log
|
||
// 演练环境可能带有空的对象存储变量。子进程必须按「未配置」启动，
|
||
// 否则测不到带类型空 *S3 回落本地目录的路径。
|
||
cmd.Env = withoutBlobstoreS3(os.Environ())
|
||
// 独立 cwd：serve 会向 ./data 写 skills/SQLite 等，不能落在测试目录。
|
||
cmd.Dir = t.TempDir()
|
||
if err := cmd.Start(); err != nil {
|
||
t.Fatalf("start %s: %v", name, err)
|
||
}
|
||
inst := &instance{name: name, base: "http://" + addr, cmd: cmd, log: log}
|
||
t.Cleanup(func() { e.stopInstance(inst) })
|
||
e.waitHealthy(t, inst)
|
||
return inst
|
||
}
|
||
|
||
func withoutBlobstoreS3(env []string) []string {
|
||
out := make([]string, 0, len(env))
|
||
for _, item := range env {
|
||
if strings.HasPrefix(item, "BLOBSTORE_S3_") {
|
||
continue
|
||
}
|
||
out = append(out, item)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (e *env) uploadArtifact(t *testing.T, c *http.Client, projectID, name string, payload []byte) string {
|
||
t.Helper()
|
||
var buf bytes.Buffer
|
||
writer := multipart.NewWriter(&buf)
|
||
part, err := writer.CreateFormFile("file", name)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := part.Write(payload); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := writer.Close(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
req, err := http.NewRequest(http.MethodPost, e.baseFor(c)+"/api/projects/"+projectID+"/uploads", &buf)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
req.Header.Set("Content-Type", writer.FormDataContentType())
|
||
resp, err := c.Do(req)
|
||
if err != nil {
|
||
t.Fatalf("upload: %v", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||
var info struct {
|
||
ArtifactID string `json:"artifactId"`
|
||
}
|
||
_ = json.Unmarshal(raw, &info)
|
||
if resp.StatusCode != http.StatusOK || info.ArtifactID == "" {
|
||
t.Fatalf("upload status %d id %q body %.300s", resp.StatusCode, info.ArtifactID, raw)
|
||
}
|
||
return info.ArtifactID
|
||
}
|
||
|
||
func (e *env) waitHealthy(t *testing.T, inst *instance) {
|
||
t.Helper()
|
||
deadline := time.Now().Add(60 * time.Second)
|
||
for time.Now().Before(deadline) {
|
||
resp, err := http.Get(inst.base + "/api/health")
|
||
if err == nil {
|
||
resp.Body.Close()
|
||
// auth=on 时 health 同样要求鉴权：401 = 服务已就绪并正确应答挑战。
|
||
if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusUnauthorized {
|
||
return
|
||
}
|
||
}
|
||
time.Sleep(300 * time.Millisecond)
|
||
}
|
||
t.Fatalf("instance %s not healthy in 60s; log:\n%s", inst.name, inst.log.String())
|
||
}
|
||
|
||
func (e *env) stopInstance(inst *instance) {
|
||
if inst.cmd.Process == nil || inst.cmd.ProcessState != nil {
|
||
return
|
||
}
|
||
_ = inst.cmd.Process.Signal(syscall.SIGTERM)
|
||
done := make(chan struct{})
|
||
go func() { _ = inst.cmd.Wait(); close(done) }()
|
||
select {
|
||
case <-done:
|
||
case <-time.After(10 * time.Second):
|
||
_ = inst.cmd.Process.Kill()
|
||
<-done
|
||
}
|
||
}
|
||
|
||
func (e *env) killNine(t *testing.T, inst *instance) {
|
||
t.Helper()
|
||
if inst.cmd.Process == nil {
|
||
t.Fatal("no process to kill")
|
||
}
|
||
if err := inst.cmd.Process.Signal(syscall.SIGKILL); err != nil {
|
||
t.Fatalf("SIGKILL %s: %v", inst.name, err)
|
||
}
|
||
_ = inst.cmd.Wait()
|
||
}
|
||
|
||
// ---- HTTP 客户端 ----
|
||
|
||
// loginAlice 引导 alice 账号并发放 API key（CLI 直连测试库；key 存共享 PG，
|
||
// 对两实例同时有效——程序化验收的首选凭证，SSE 亦可用 ?key= 兜底）。
|
||
// 当前基座拒绝带密码的 user create：空库先 bootstrap 一位开库管理员，再为
|
||
// alice 发放激活凭据。凭据兑换后设置 alicePass，然后才能发 API Key。
|
||
func (e *env) loginAlice(t *testing.T) *userClient {
|
||
t.Helper()
|
||
// 账户事务会锁库内服务身份；serve 不补写这行，须先显式建立。
|
||
e.cli(t, "identity", "ensure", "--dsn", e.dbDSN, "--instance-id", "mi-acceptance", "--dbref", "pg")
|
||
e.cli(t, "user", "bootstrap-admin", "--dsn", e.dbDSN, "--owner", "mi_setup", "--name", "Setup",
|
||
"--out-file", filepath.Join(t.TempDir(), "setup-grant"))
|
||
grantFile := filepath.Join(t.TempDir(), "alice-grant")
|
||
e.cli(t, "user", "create", "--dsn", e.dbDSN, "--owner", aliceOwner, "--name", "Alice", "--out-file", grantFile)
|
||
grantRaw, err := os.ReadFile(grantFile)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
jar, err := cookiejar.New(nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sess := &http.Client{Timeout: httpWait, Jar: jar}
|
||
// e.uc 尚未赋值 → baseFor 走 A；激活会话 cookie 落 jar 后设置密码。
|
||
e.do(t, sess, "POST", "/api/auth/login-key/redeem", map[string]any{
|
||
"key": strings.TrimSpace(string(grantRaw)),
|
||
}, nil)
|
||
e.do(t, sess, "POST", "/api/auth/password", map[string]any{
|
||
"next": alicePass,
|
||
}, nil)
|
||
|
||
key := e.cliOut(t, "apikey", "create", "--dsn", e.dbDSN, "--owner", aliceOwner, "--name", "mi-acceptance")
|
||
uc := &userClient{
|
||
api: &http.Client{Timeout: httpWait, Transport: bearerRT{key: key}},
|
||
apiB: &http.Client{Timeout: httpWait, Transport: bearerRT{key: key}},
|
||
streamB: &http.Client{Transport: bearerRT{key: key}}, // SSE 无整体超时；chan 背压控制
|
||
}
|
||
// runtime.mock 自系统受管设置体制（backend #188/#190）起由 --mock 播种进
|
||
// 系统层，经 Effective 链对所有租户生效（selectConversationRunner 按
|
||
// EffectiveSettings 分流）；alice 的用户层写入会被受管拦截，也不再必要。
|
||
return uc
|
||
}
|
||
|
||
// bearerRT 为所有请求附加 Authorization: Bearer <key>。
|
||
type bearerRT struct{ key string }
|
||
|
||
func (b bearerRT) RoundTrip(r *http.Request) (*http.Response, error) {
|
||
r2 := r.Clone(r.Context())
|
||
r2.Header.Set("Authorization", "Bearer "+b.key)
|
||
return http.DefaultTransport.RoundTrip(r2)
|
||
}
|
||
|
||
// cliOut 运行 CLI 并取 stdout 首行（apikey create 的明文仅此一次输出）。
|
||
func (e *env) cliOut(t *testing.T, args ...string) string {
|
||
t.Helper()
|
||
cmd := exec.Command(e.bin, args...)
|
||
cmd.Dir = t.TempDir()
|
||
var stdout, stderr bytes.Buffer
|
||
cmd.Stdout, cmd.Stderr = &stdout, &stderr
|
||
if err := cmd.Run(); err != nil {
|
||
t.Fatalf("cli %v…: %v\n%s", args[:2], err, stderr.String())
|
||
}
|
||
line := strings.TrimSpace(stdout.String())
|
||
if line == "" {
|
||
t.Fatalf("cli %v…: empty stdout", args[:2])
|
||
}
|
||
return strings.Fields(line)[0]
|
||
}
|
||
|
||
func (e *env) cli(t *testing.T, args ...string) {
|
||
t.Helper()
|
||
cmd := exec.Command(e.bin, args...)
|
||
cmd.Dir = t.TempDir()
|
||
out, err := cmd.CombinedOutput()
|
||
if err != nil {
|
||
t.Fatalf("cli %v…: %v\n%s", args[:2], err, out)
|
||
}
|
||
}
|
||
|
||
func (e *env) do(t *testing.T, c *http.Client, method, path string, body any, out any) {
|
||
t.Helper()
|
||
code, raw := e.raw(t, c, method, path, body)
|
||
if code != http.StatusOK {
|
||
t.Fatalf("%s %s: status %d, body %.300s", method, path, code, raw)
|
||
}
|
||
if out != nil {
|
||
if err := json.Unmarshal([]byte(raw), out); err != nil {
|
||
t.Fatalf("%s %s: decode %v", method, path, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// raw 以 JSON body 发起请求，返回状态码与响应体文本。请求地址取
|
||
// client 所绑定实例（由调用方选择 api / apiB / streamB 及对应 base）。
|
||
func (e *env) raw(t *testing.T, c *http.Client, method, path string, body any) (int, string) {
|
||
t.Helper()
|
||
return e.rawBase(t, c, e.baseFor(c), method, path, body)
|
||
}
|
||
|
||
func (e *env) rawBase(t *testing.T, c *http.Client, base, method, path string, body any) (int, string) {
|
||
t.Helper()
|
||
var rd io.Reader
|
||
if body != nil {
|
||
buf, _ := json.Marshal(body)
|
||
rd = bytes.NewReader(buf)
|
||
}
|
||
req, err := http.NewRequest(method, base+path, rd)
|
||
if err != nil {
|
||
t.Fatalf("%s %s: %v", method, path, err)
|
||
}
|
||
if body != nil {
|
||
req.Header.Set("Content-Type", "application/json")
|
||
}
|
||
resp, err := c.Do(req)
|
||
if err != nil {
|
||
t.Fatalf("%s %s: %v", method, path, err)
|
||
}
|
||
defer resp.Body.Close()
|
||
out, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
|
||
return resp.StatusCode, string(out)
|
||
}
|
||
|
||
// baseFor 按 client 身份路由实例：B 侧（apiB/streamB）→ B；其余 → A。
|
||
func (e *env) baseFor(c *http.Client) string {
|
||
if e.uc != nil && (c == e.uc.apiB || c == e.uc.streamB) {
|
||
return e.b.base
|
||
}
|
||
return e.a.base
|
||
}
|
||
|
||
// ---- 场景辅助 ----
|
||
|
||
type sseEvent struct {
|
||
seq int64
|
||
event string
|
||
data string
|
||
}
|
||
|
||
func (e *env) setupConversation(t *testing.T, uc *userClient) (string, string) {
|
||
t.Helper()
|
||
var proj map[string]any
|
||
e.do(t, uc.api, "POST", "/api/projects", map[string]any{"name": "mi-验收"}, &proj)
|
||
projID, _ := proj["id"].(string)
|
||
var conv map[string]any
|
||
e.do(t, uc.api, "POST", "/api/projects/"+projID+"/conversations", map[string]any{"title": "双进程"}, &conv)
|
||
convID, _ := conv["id"].(string)
|
||
return projID, convID
|
||
}
|
||
|
||
// postMessage 先等会话空闲（上一 run 终态），再提交消息——#hitl 需要一个
|
||
// 可占用的空闲会话；普通消息同样避免撞上上一 run 的收口窗口。
|
||
func (e *env) postMessage(t *testing.T, c *http.Client, convID, text string) {
|
||
t.Helper()
|
||
e.waitConversationIdle(t, c, convID, 20*time.Second)
|
||
e.do(t, c, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text}, nil)
|
||
}
|
||
|
||
func (e *env) waitConversationIdle(t *testing.T, c *http.Client, convID string, within time.Duration) {
|
||
t.Helper()
|
||
deadline := time.Now().Add(within)
|
||
for time.Now().Before(deadline) {
|
||
var detail map[string]any
|
||
e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
|
||
busy := false
|
||
if runs, ok := detail["runs"].([]any); ok {
|
||
for _, r := range runs {
|
||
m, _ := r.(map[string]any)
|
||
switch m["status"] {
|
||
case "pending", "running", "waiting_hitl":
|
||
busy = true
|
||
}
|
||
}
|
||
}
|
||
if !busy {
|
||
return
|
||
}
|
||
time.Sleep(300 * time.Millisecond)
|
||
}
|
||
t.Fatalf("conversation %s still busy after %v", convID, within)
|
||
}
|
||
|
||
func (e *env) lastSeq(t *testing.T, c *http.Client, convID string) int64 {
|
||
t.Helper()
|
||
var detail map[string]any
|
||
e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
|
||
n, _ := detail["lastSeq"].(float64)
|
||
return int64(n)
|
||
}
|
||
|
||
// subscribe 在 B 实例打开会话 SSE（since 起），帧 seq > since 才投递；
|
||
// 返回通道在流结束/出错时关闭。调用方通过关闭底层响应取消。
|
||
func (e *env) subscribe(t *testing.T, c *http.Client, convID string, since int64) <-chan sseEvent {
|
||
t.Helper()
|
||
u := e.b.base + "/api/events?conversationId=" + convID + "&since=" + strconv.FormatInt(since, 10)
|
||
req, err := http.NewRequest(http.MethodGet, u, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp, err := c.Do(req)
|
||
if err != nil {
|
||
t.Fatalf("subscribe SSE: %v", err)
|
||
}
|
||
if resp.StatusCode != http.StatusOK {
|
||
resp.Body.Close()
|
||
t.Fatalf("subscribe SSE: status %d", resp.StatusCode)
|
||
}
|
||
ch := make(chan sseEvent, 1024)
|
||
go func() {
|
||
defer close(ch)
|
||
defer resp.Body.Close()
|
||
sc := bufio.NewScanner(resp.Body)
|
||
sc.Buffer(make([]byte, 0, 64*1024), 4<<20)
|
||
var seq int64
|
||
var event, data string
|
||
flush := func() {
|
||
if seq > since {
|
||
// 阻塞发送：restart 场景依赖全量重放断言，不允许丢帧；
|
||
// 消费停滞时读取 goroutine 挂起（进程随测试退出回收）。
|
||
ch <- sseEvent{seq: seq, event: event, data: data}
|
||
}
|
||
seq, event, data = 0, "", ""
|
||
}
|
||
for sc.Scan() {
|
||
line := sc.Text()
|
||
switch {
|
||
case line == "":
|
||
if seq > 0 {
|
||
flush()
|
||
}
|
||
case strings.HasPrefix(line, "id: "):
|
||
seq, _ = strconv.ParseInt(strings.TrimPrefix(line, "id: "), 10, 64)
|
||
case strings.HasPrefix(line, "event: "):
|
||
event = strings.TrimPrefix(line, "event: ")
|
||
case strings.HasPrefix(line, "data: "):
|
||
data = strings.TrimPrefix(line, "data: ")
|
||
}
|
||
}
|
||
}()
|
||
return ch
|
||
}
|
||
|
||
func (e *env) waitActiveHITL(t *testing.T, c *http.Client, convID string) hitlCard {
|
||
t.Helper()
|
||
deadline := time.Now().Add(15 * time.Second)
|
||
for time.Now().Before(deadline) {
|
||
var detail map[string]any
|
||
e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
|
||
if cards, ok := detail["activeHitl"].([]any); ok && len(cards) > 0 {
|
||
m, _ := cards[0].(map[string]any)
|
||
id, _ := m["id"].(string)
|
||
runID, _ := m["runId"].(string)
|
||
if id != "" && runID != "" {
|
||
return hitlCard{id: id, runID: runID}
|
||
}
|
||
}
|
||
time.Sleep(300 * time.Millisecond)
|
||
}
|
||
t.Fatalf("no active HITL card in 15s; logA:\n%s", e.a.log.String())
|
||
return hitlCard{}
|
||
}
|
||
|
||
func (e *env) waitRunStatus(t *testing.T, c *http.Client, runID, status string) {
|
||
t.Helper()
|
||
deadline := time.Now().Add(httpWait)
|
||
for time.Now().Before(deadline) {
|
||
var out map[string]any
|
||
e.do(t, c, "GET", "/api/runs/"+runID, nil, &out)
|
||
run, _ := out["run"].(map[string]any)
|
||
if s, _ := run["status"].(string); s == status {
|
||
return
|
||
}
|
||
time.Sleep(300 * time.Millisecond)
|
||
}
|
||
t.Fatalf("run %s not %s within %v", runID, status, httpWait)
|
||
}
|