// 协作存储 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"
	aliceInitialPass = "MiAliceInitPass123!"
	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.Fatalf("event written via A not visible on B SSE within 8s: no event with seq > %d observed on %s", lastSeq, e.instanceOf(uc.streamB))
		}
	})

	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
		var statuses []string
		reconciled := false
		start := time.Now()
		deadline := start.Add(httpWait)
		for time.Now().Before(deadline) {
			e.do(t, uc.api, "GET", "/api/conversations/"+convID, nil, &detail)
			statuses = runStatuses(detail)
			if n := len(statuses); n > 0 && statuses[n-1] == "interrupted" {
				reconciled = true
				break
			}
			time.Sleep(300 * time.Millisecond)
		}
		if !reconciled {
			// 不能落到下面的 POST：未收口时它会得到看似无关的 409 in-flight。
			t.Fatalf("restarted %s never reported the crashed run as interrupted: waited %v (budget %v), last run status never became \"interrupted\"; run statuses %v; log:\n%s",
				e.instanceOf(uc.api), time.Since(start).Round(time.Millisecond), httpWait, statuses, e.a.log.String())
		}
		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://"))

		// 断言前先等重启实例给出完整的恢复视图：对账收口的 interrupted run 可见，
		// 且会话内无进行中 run。断言本身不依赖这次轮询的结果。
		var statuses []string
		recovered := false
		start := time.Now()
		readyBy := start.Add(httpWait)
		for time.Now().Before(readyBy) {
			var view map[string]any
			e.do(t, uc.apiB, "GET", "/api/conversations/"+convID, nil, &view)
			statuses = runStatuses(view)
			interrupted, busy := false, false
			for _, s := range statuses {
				switch s {
				case "interrupted":
					interrupted = true
				case "pending", "running", "waiting_hitl":
					busy = true
				}
			}
			if interrupted && !busy {
				recovered = true
				break
			}
			time.Sleep(300 * time.Millisecond)
		}
		if !recovered {
			t.Fatalf("restarted %s never showed the recovered view (an interrupted run and no pending/running/waiting_hitl run): waited %v (budget %v); run statuses %v; logA:\n%s\nlogB:\n%s",
				e.instanceOf(uc.apiB), time.Since(start).Round(time.Millisecond), httpWait, statuses, e.a.log.String(), e.b.log.String())
		}

		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: event seq %d never observed on %s SSE within 15s", seen, lastSeq, seen+1, e.instanceOf(uc.streamB))
			}
		}
		// 重启后写入仍可推进。
		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 = append(withoutBlobstoreS3(os.Environ()),
		"SUPER_ADMIN_OWNER=mi-root",
		"SUPER_ADMIN_PASSWORD=MiRootPass12345!",
	)
	// 独立 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()
	last := "no probe completed"
	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
			}
			last = "status " + strconv.Itoa(resp.StatusCode)
		} else {
			last = err.Error()
		}
		time.Sleep(300 * time.Millisecond)
	}
	t.Fatalf("instance %s (%s) not healthy in 60s: GET /api/health never returned 200/401 (last probe: %s); log:\n%s", inst.name, inst.base, last, 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()
	// 超级管理员 mi-root 已由 serve 启动时的 SUPER_ADMIN_* 建立；用 root key 走
	// admin API 创建 alice（带初始密码）。
	rootKey := e.cliOut(t, "apikey", "create", "--dsn", e.dbDSN, "--owner", "mi-root", "--name", "mi-acceptance-root-create-alice")
	root := &http.Client{Timeout: httpWait, Transport: bearerRT{key: rootKey}}
	e.do(t, root, "POST", "/api/admin/users", map[string]any{
		"owner": aliceOwner,
		"profile": map[string]any{
			"name": "Alice",
		},
		"password": aliceInitialPass,
	}, nil)
	jar, err := cookiejar.New(nil)
	if err != nil {
		t.Fatal(err)
	}
	sess := &http.Client{Timeout: httpWait, Jar: jar}
	// e.uc 尚未赋值 → baseFor 走 A；初始密码登录后会话仅允许改密。
	e.do(t, sess, "POST", "/api/auth/login", map[string]any{
		"owner":    aliceOwner,
		"password": aliceInitialPass,
	}, nil)
	e.do(t, sess, "POST", "/api/auth/password", map[string]any{
		"current": aliceInitialPass,
		"next":    alicePass,
	}, nil)

	apiKey := e.cliOut(t, "apikey", "create", "--dsn", e.dbDSN, "--owner", aliceOwner, "--name", "mi-acceptance")
	// 建项目需角色上的 project.create：由引导管理员经真实 API 建角色并绑定 alice。
	e.grantAliceProjectCreate(t)
	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 只由已启用账号所绑定的已启用角色上的单一布尔决定，管理员
// 身份或系统授权都不再隐含它。故新建角色、写入 project.create
// （allowed=capAllowed=true，按 roleVersion 乐观并发），再按 alice 的
// stateVersion 绑定该角色。mi-root 由 serve 启动时的 SUPER_ADMIN_* 建立，
// 可直接为其发 API key。
func (e *env) grantAliceProjectCreate(t *testing.T) {
	t.Helper()
	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"`
		StateVersion int64  `json:"stateVersion"`
	}
	e.do(t, root, "GET", "/api/admin/users", nil, &users)
	userID, stateVersion := "", int64(0)
	for _, u := range users {
		if u.Owner == aliceOwner {
			userID, stateVersion = u.UserID, u.StateVersion
		}
	}
	if userID == "" {
		t.Fatalf("user %s not found in /api/admin/users", aliceOwner)
	}
	var role struct {
		RoleID      string `json:"roleId"`
		RoleVersion int64  `json:"roleVersion"`
	}
	e.do(t, root, "POST", "/api/admin/roles", map[string]any{
		"name":        "mi-project-access",
		"description": "multi-instance acceptance: project.create",
	}, &role)
	if role.RoleID == "" {
		t.Fatal("POST /api/admin/roles returned no roleId")
	}
	e.do(t, root, "PUT", "/api/admin/roles/"+role.RoleID+"/permissions", map[string]any{
		"expectedRoleVersion": role.RoleVersion,
		"idempotencyKey":      "mi-role-project-create",
		"changes": []map[string]any{
			{"actionId": "project.create", "allowed": true, "capAllowed": true},
		},
	}, nil)
	e.do(t, root, "PUT", "/api/admin/users/"+userID+"/role", map[string]any{
		"roleId":               role.RoleID,
		"expectedStateVersion": stateVersion,
		"idempotencyKey":       "mi-alice-project-access-role",
	}, 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)
}

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

// instanceOf 给出 client 实际轮询的实例（名 + 地址），供有界等待的失败信息定位。
func (e *env) instanceOf(c *http.Client) string {
	base := e.baseFor(c)
	if e.b != nil && base == e.b.base {
		return e.b.name + " (" + base + ")"
	}
	return e.a.name + " (" + base + ")"
}

// runStatuses 按顺序取会话详情里各 run 的状态，供等待条件与失败信息复用。
func runStatuses(detail map[string]any) []string {
	runs, _ := detail["runs"].([]any)
	out := make([]string, 0, len(runs))
	for _, r := range runs {
		m, _ := r.(map[string]any)
		s, _ := m["status"].(string)
		out = append(out, s)
	}
	return out
}

// ---- 场景辅助 ----

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()
	var statuses []string
	start := time.Now()
	deadline := start.Add(within)
	for time.Now().Before(deadline) {
		var detail map[string]any
		e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
		statuses = runStatuses(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 on %s: waited %v (budget %v) and never observed all runs out of pending/running/waiting_hitl; run statuses %v",
		convID, e.instanceOf(c), time.Since(start).Round(time.Millisecond), within, statuses)
}

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()
	var statuses []string
	cardCount := 0
	start := time.Now()
	deadline := start.Add(15 * time.Second)
	for time.Now().Before(deadline) {
		var detail map[string]any
		e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
		statuses = runStatuses(detail)
		cards, ok := detail["activeHitl"].([]any)
		cardCount = len(cards)
		if 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 with id and runId on %s for conversation %s: waited %v (budget 15s); last activeHitl count %d, run statuses %v; logA:\n%s",
		e.instanceOf(c), convID, time.Since(start).Round(time.Millisecond), cardCount, statuses, e.a.log.String())
	return hitlCard{}
}

func (e *env) waitRunStatus(t *testing.T, c *http.Client, runID, status string) {
	t.Helper()
	observed := "<none>"
	start := time.Now()
	deadline := start.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)
		s, _ := run["status"].(string)
		if s == status {
			return
		}
		observed = s
		time.Sleep(300 * time.Millisecond)
	}
	t.Fatalf("run %s not %s on %s: waited %v (budget %v), last observed status %q; logA:\n%s",
		runID, status, e.instanceOf(c), time.Since(start).Round(time.Millisecond), httpWait, observed, e.a.log.String())
}