// 协作存储 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"
	rootPass   = "mi-root-pass-1"
	httpWait   = 15 * time.Second
)

type instance struct {
	name    string
	base    string
	dataDir string // serve 的 cwd：本地实例身份（manifest）所在，账户 CLI 须以此身份运行
	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))
	// 库内服务身份是显式引导（svc init 同款），且为部署级单行：双进程共享
	// 同一身份，只对 A ensure 一次。账户 CLI 的审计与核验都依赖它。
	e.identityEnsure(e.t, e.a)
}

// identityEnsure 建立/对齐库内服务身份行（agent_system_identity）。
func (e *env) identityEnsure(t *testing.T, inst *instance) {
	t.Helper()
	cmd := exec.Command(e.bin, "identity", "ensure", "--instance-id", inst.name, "--data", inst.dataDir, "--dbref", "pg")
	cmd.Dir = inst.dataDir
	cmd.Env = append(os.Environ(), "AGENT_SYSTEM_DSN="+e.dbDSN)
	if out, err := cmd.CombinedOutput(); err != nil {
		t.Fatalf("identity ensure %s: %v\n%s", inst.name, err, out)
	}
}

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 等，不能落在测试目录。
	dataDir := t.TempDir()
	cmd.Dir = dataDir
	if err := cmd.Start(); err != nil {
		t.Fatalf("start %s: %v", name, err)
	}
	inst := &instance{name: name, base: "http://" + addr, dataDir: dataDir, 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= 兜底）。
// U3 后本地无密码创建：user create 只签发待激活账号的一次性激活 Key（凭据
// 独占落 JSON），redeem 会话进入 grant-setup 改密设 alicePass，完成后再发 Key。
func (e *env) loginAlice(t *testing.T) *userClient {
	t.Helper()
	// 空库先引导首个管理员（本地 pending 创建要求库内已有用户），再建 alice。
	// 两步都签发一次性凭据，独占落临时 JSON；mi-root 的凭据随后用于初始化其账号。
	rootCred := filepath.Join(t.TempDir(), "root-cred.json")
	e.cliIn(t, e.a.dataDir, "user", "bootstrap-admin", "--dsn", e.dbDSN, "--owner", "mi-root", "--out-file", rootCred)
	activation := filepath.Join(t.TempDir(), "activation.json")
	e.cliIn(t, e.a.dataDir, "user", "create", "--dsn", e.dbDSN, "--owner", aliceOwner, "--name", "Alice", "--out", activation)
	activateKey := activationSecret(t, activation)
	jar, err := cookiejar.New(nil)
	if err != nil {
		t.Fatal(err)
	}
	sess := &http.Client{Timeout: httpWait, Jar: jar}
	// e.uc 尚未赋值 → baseFor 走 A；redeem 会话（grant-setup）直接设 alicePass。
	e.do(t, sess, "POST", "/api/auth/login-key/redeem", map[string]any{
		"key": activateKey,
	}, nil)
	e.do(t, sess, "POST", "/api/auth/password", map[string]any{
		"next": alicePass,
	}, nil)

	apiKey := e.cliOut(t, "apikey", "create", "--dsn", e.dbDSN, "--owner", aliceOwner, "--name", "mi-acceptance")
	// IAM U4 把建项目改为系统动作 project.create：由引导管理员经真实 API 授予 alice。
	e.grantAliceProjectCreate(t, rootCred)
	uc := &userClient{
		api:     &http.Client{Timeout: httpWait, Transport: bearerRT{key: apiKey}},
		apiB:    &http.Client{Timeout: httpWait, Transport: bearerRT{key: apiKey}},
		streamB: &http.Client{Transport: bearerRT{key: apiKey}}, // SSE 无整体超时；chan 背压控制
	}
	// runtime.mock 自系统受管设置体制（backend #188/#190）起由 --mock 播种进
	// 系统层，经 Effective 链对所有租户生效（selectConversationRunner 按
	// EffectiveSettings 分流）；alice 的用户层写入会被受管拦截，也不再必要。
	return uc
}

// grantAliceProjectCreate 以 mi-root 身份经 admin API 为 alice 追加系统授权
// project.create（保留其已有授权；PUT 为整体替换，按 authorityEpoch 乐观并发）。
// bootstrap-admin 只签发一次性凭据，mi-root 未初始化前不能发 API key，故先照
// alice 的方式 redeem 并设 rootPass。
func (e *env) grantAliceProjectCreate(t *testing.T, rootCred string) {
	t.Helper()
	raw, err := os.ReadFile(rootCred)
	if err != nil {
		t.Fatal(err)
	}
	rootSecret := strings.TrimSpace(string(raw))
	if rootSecret == "" {
		t.Fatal("bootstrap-admin credential file is empty")
	}
	jar, err := cookiejar.New(nil)
	if err != nil {
		t.Fatal(err)
	}
	sess := &http.Client{Timeout: httpWait, Jar: jar}
	e.do(t, sess, "POST", "/api/auth/login-key/redeem", map[string]any{
		"key": rootSecret,
	}, nil)
	e.do(t, sess, "POST", "/api/auth/password", map[string]any{
		"next": rootPass,
	}, nil)

	rootKey := e.cliOut(t, "apikey", "create", "--dsn", e.dbDSN, "--owner", "mi-root", "--name", "mi-acceptance-root")
	root := &http.Client{Timeout: httpWait, Transport: bearerRT{key: rootKey}}
	var users []struct {
		UserID string `json:"userId"`
		Owner  string `json:"owner"`
	}
	e.do(t, root, "GET", "/api/admin/users", nil, &users)
	userID := ""
	for _, u := range users {
		if u.Owner == aliceOwner {
			userID = u.UserID
		}
	}
	if userID == "" {
		t.Fatalf("user %s not found in /api/admin/users", aliceOwner)
	}
	var grants struct {
		Actions        []string `json:"actions"`
		AuthorityEpoch int64    `json:"authorityEpoch"`
	}
	e.do(t, root, "GET", "/api/admin/users/"+userID+"/system-grants", nil, &grants)
	actions := grants.Actions
	has := false
	for _, a := range actions {
		has = has || a == "project.create"
	}
	if !has {
		actions = append(actions, "project.create")
	}
	e.do(t, root, "PUT", "/api/admin/users/"+userID+"/system-grants", map[string]any{
		"expectedAuthorityEpoch": grants.AuthorityEpoch,
		"idempotencyKey":         "mi-alice-project-create",
		"actions":                actions,
	}, nil)
}

// 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)
}

// activationSecret 读 CLI 凭据独占文件里的激活 Key（JSON {"kind","owner","secret"}）。
func activationSecret(t *testing.T, path string) string {
	t.Helper()
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var v struct {
		Secret string `json:"secret"`
	}
	if err := json.Unmarshal(b, &v); err != nil || v.Secret == "" {
		t.Fatalf("activation credential unreadable: %v", err)
	}
	return v.Secret
}

// 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()
	e.cliIn(t, t.TempDir(), args...)
}

// cliIn 以指定 cwd 运行 CLI：账户命令经 assertLocalAccountOwner 核验本地实例
// 身份，必须从持有该身份的 dataDir 运行。
func (e *env) cliIn(t *testing.T, dir string, args ...string) {
	t.Helper()
	cmd := exec.Command(e.bin, args...)
	cmd.Dir = dir
	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)
}