Add U7 two-process acceptance, k6 load harness and release checklist

Why: collaboration storage U7 (proposal sections 3/5/7) requires two-process HTTP-level acceptance over shared PG, a reproducible load-test entry honoring the load contract, and an auditable pre-release checklist; capacity windows and dual-host runs need user-scheduled resources, so the deliverable here is the verifiable entry, not capacity claims. Evidence: make test green (composite); make test-multi-instance green locally against an isolated PG (five scenarios incl. crash-takeover and restart invariants); run_load.sh --profile smoke end-to-end green (SSE probe p95~18ms, zero seq duplicates, stack auto-recycled); backend pinned to feature branch with the three cross-instance fixes. Impact: new tests/integration/multi-instance (CI test-pg job step), new tests/load/collaboration (k6 + SSE probe + run_load.sh), release checklist at docs/release-multi-instance.md; no runtime behavior change in Dev repo itself. Refs: backend PR #164; task card feature/u7-load-fault-release.
2026-10-12 02:31:35 +00:00
parent 65f72e961a
commit f705f99b8f
15 changed files with 1352 additions and 2 deletions
@@ -0,0 +1,620 @@
+// 协作存储 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"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+const (
+	aliceOwner = "mi_alice"
+	alicePass0 = "mi-alice-pass-0"
+	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_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
+	// 独立 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 (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= 兜底）。
+func (e *env) loginAlice(t *testing.T) *userClient {
+	t.Helper()
+	e.cli(t, "user", "create", "--dsn", e.dbDSN, "--owner", aliceOwner, "--name", "Alice", "--password", alicePass)
+	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 背压控制
+	}
+	// --mock 只写 OwnerDefault 的设置；alice 是独立租户，须显式开 mock
+	// （selectConversationRunner 按 EffectiveSettings 分流），否则 run 走
+	// LLMRunner 且无模型配置直接失败。
+	e.do(t, uc.api, "PUT", "/api/settings/user", map[string]any{"runtime": map[string]any{"mock": true}}, nil)
+	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)
+}