.gitea/workflows/ci.yaml

@@ -373,3 +373,7 @@ jobs:
           AGENT_PG_TEST_DSN: postgres://ci_agent:ci_job_local_only@127.0.0.1:5432/agent_test?sslmode=disable
           AGENT_REDIS_TEST_ADDR: 127.0.0.1:6399
         run: make test-backend-pg
+      - name: Two-process HTTP acceptance over shared PostgreSQL (U7, required)
+        env:
+          AGENT_PG_TEST_DSN: postgres://ci_agent:ci_job_local_only@127.0.0.1:5432/agent_test?sslmode=disable
+        run: make test-multi-instance

Makefile

@@ -13,7 +13,7 @@ PLUGINS    := plugins
 COMBO      := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null)
 VERSION    := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO)
 
-.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-sync task-close task-finish facts-verify evidence version
+.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-sync task-close task-finish facts-verify evidence version
 
 version:
 	@echo "agent-system $(VERSION)"
@@ -112,6 +112,14 @@ test-backend-pg:
 	@test -n "$${AGENT_REDIS_TEST_ADDR}" || { echo "test-backend-pg: AGENT_REDIS_TEST_ADDR 未设置（真实 Redis 必需，不静默跳过）"; exit 1; }
 	cd $(BACKEND) && go test -count=1 ./...
 
+# 双进程 HTTP 级验收（协作存储方案 U7）：两个真实 serve 进程共享隔离 PG。
+# 缺环境时明确失败而非静默跳过；见 tests/integration/multi-instance/README.md。
+test-multi-instance:
+	@test -n "$${AGENT_PG_TEST_DSN}" || { echo "test-multi-instance: AGENT_PG_TEST_DSN 未设置（真实 PostgreSQL 必需，不静默跳过）"; exit 1; }
+	mkdir -p tests/integration/multi-instance/.build
+	cd $(BACKEND) && CGO_ENABLED=0 go build -o ../../tests/integration/multi-instance/.build/agent-system ./cmd/agent-system
+	cd tests/integration/multi-instance && AGENT_BIN=./.build/agent-system go test -count=1 -timeout 15m .
+
 eval-deliverables:
 	cd $(BACKEND) && go test ./eval/ ./internal/quality/ -count=1
 

docs/release-multi-instance.md

@@ -0,0 +1,59 @@
+# 双实例放行前检查单（协作存储 U7）
+
+协作存储方案 §5 U7 的放行出口要求：**单实例团队、双进程、双主机分别出具
+结果；CI 身份/链接/SHA 或独立 REPLAY 可核验才用于放行，本地 SELF 只作辅助。**
+本检查单列出每一放行条件、其证据来源与当前状态；未全部满足前，第二实例
+不得承接写请求、双主机部署不得开工。
+
+状态图例：✅ 已满足（证据可核验）｜⏳ 入口就绪、待执行窗口｜❌ 未开始。
+
+## A. 正确性与故障矩阵（代码与 CI 可覆盖）
+
+| # | 条件 | 证据（命令/测试） | 状态 |
+|---|---|---|---|
+| A1 | 单实例全链路（mock）：认证/项目/会话/SSE/检索/卡片 | backend `go test ./...`（CI job `test`，SQLite 档 35+ 包） | ✅ CI 绿 |
+| A2 | 真 PG/Redis 全量 | `make test-backend-pg`（CI job `test-pg`，AGENT_PG_TEST_DSN 必跑） | ✅ CI 绿 |
+| A3 | 磁盘满清楚拒绝、零残留、可恢复 | backend `TestUploadRejectsCleanlyWhenObjectDirUnwritable`、`TestArtifactOverwriteRejectsCleanlyWhenObjectDirUnwritable` | ✅ CI 绿 |
+| A4 | 恢复隔离（未知副作用不自动重跑） | backend `recovery_hold_test.go`（U3） | ✅ CI 绿 |
+| A5 | 库级多实例语义（争抢/围栏/命令/跨实例 resolve 不落卡） | backend `runmgr/multi_instance_test.go`（真库门控） | ✅ CI 绿 |
+| A6 | **双进程 HTTP 级验收**（跨实例读/SSE/409/HITL/取消/崩溃恢复/重启不变量） | `make test-multi-instance`（CI job `test-pg`；本地 `tests/integration/multi-instance/README.md`） | ✅ CI 绿 |
+| A7 | 权限矩阵重放（撤权零泄漏：请求/SSE/检索/下载/工具） | backend `resource_authz_test`、`shared_*_test`、`sse_gap_test`、`pgvector_test`（U3/U6） | ✅ CI 绿 |
+
+## B. 容量与负载合同（需要实测窗口，本地入口已就绪）
+
+| # | 条件 | 证据（命令） | 状态 |
+|---|---|---|---|
+| B1 | 冒烟压测通过（入口可复核、不变量零违反） | `tests/load/collaboration/run_load.sh --profile smoke` | ✅ 已跑通（SELF，2026-10-12：读 P95<1ms、SSE P95≈18ms、对账 0 违反） |
+| B2 | 阶梯负载 10→30→100 用户，读 P95≤300ms、P99≤1s、非预期失败≤0.1% | 同上 `--profile ladder`（k6 需安装：`brew install k6`） | ⏳ 待窗口 |
+| B3 | 60 分钟稳定负载 + 2 小时混合长稳 | `--profile sustained60m` / `--profile sustained2h` | ⏳ 待窗口（约 3h 机时） |
+| B4 | SSE 持久事件可见延迟 P95≤1s | 同上套件自动执行（sse_probe，断言即退出码） | ✅ 冒烟档已证机制；正式窗口随 B3 |
+| B5 | 数据量档（100 万消息/10 万知识分块/多项目交叉授权） | sustained 长跑自然积累 + seed 扩展（未实现 seed 工具，标注） | ❌ seed 工具未实施 |
+| B6 | 数据库基线（pgbench 只测 DB 能力，不替代业务验收） | `pgbench` 对隔离实例（只读参考） | ❌ 未执行 |
+| B7 | 容量数字绑定组件 SHA/配置/数据集归档 | results/<ts>/ 目录 + 报告引用 SHA | ⏳ 随 B2/B3 产出 |
+
+## C. 双进程与双主机放行（分布式出口）
+
+| # | 条件 | 证据 | 状态 |
+|---|---|---|---|
+| C1 | 双进程共享 PG：正确性矩阵全过（A6） | `make test-multi-instance` | ✅ |
+| C2 | 双进程**负载**下互相不拖累（热点会话/命令时延） | 双栈压测（run_load.sh 尚为单实例档，需扩展第二实例参数） | ❌ 入口未扩展 |
+| C3 | 双主机（跨机网络）验证：同套验收在两台主机执行 | A6 的 AGENT_BIN/DSN 指向远端；需用户指定第二主机 | ❌ 待资源 |
+| C4 | 故障演练：PG 断连/Redis 断连恢复（RTO≤15min） | backend ratelimit/recovery 测试 + 运维手册演练记录 | ⏳ 代码级 ✅；运维演练待窗口 |
+| C5 | 备份恢复演练（RPO≤5min/RTO≤60min，pg_dump+物理备份） | U4 迁移演练已过；定期备份恢复演练属运维，待排期 | ⏳ |
+
+## 放行结论（当前）
+
+- **单实例团队档**：A1–A7 全绿 + B1 冒烟通过。正式容量结论待 B2/B3 窗口；
+  在此之前按「已验收功能、未承诺容量」运营。
+- **双进程档**：正确性已证（A5/A6/C1），负载与运维演练（C2/C4）未过，
+  **不放行第二实例承接写请求**。
+- **双主机档**：未开始（C3），不放行。
+
+## 已知边界与诚实声明
+
+- U7 期间经双进程验收发现并修复三个跨实例缺陷（详见
+  `tests/integration/multi-instance/README.md`）：HITL resolve 落卡短路命令、
+  cancel 命令空会话 ID、CancelRun 假成功。修复均已入 CI 回归。
+- k6 首次冒烟使用临时二进制（v2.3.0）验证后删除；正式窗口前按
+  `brew install k6` 装固定位。
+- B5 大数据量 seed 与 C2 双实例压测入口是下一步实施项，本检查单不替代。

tests/integration/multi-instance/.gitignore

@@ -0,0 +1,2 @@
+/.build/
+/data/

tests/integration/multi-instance/README.md

@@ -0,0 +1,45 @@
+# 双进程 HTTP 级验收（协作存储 U7）
+
+两个**真实 backend serve 进程**共享同一 PostgreSQL 的黑盒验收，覆盖方案 §7
+最小验收矩阵的进程级行（库级语义已在 backend `internal/runmgr/multi_instance_test.go`
+覆盖；本目录验证 HTTP/SSE/进程生命周期层面的同一套不变量）：
+
+| 子场景 | 断言（矩阵行） |
+|---|---|
+| `cross_read` | A 建的项目/会话在 B 立即可见（共享权威） |
+| `cross_event_sse` | A 写入的事件在 B 的 SSE 可见（R5 跨实例传播） |
+| `same_session_conflict_then_cross_hitl` | 同会话并发提交明确 409 不误写；B 跨实例 resolve A 的 HITL 经命令总线到达执行者（R3/R4） |
+| `cross_cancel` | B 跨实例取消 A 的 run：命令持久、唯一消费、run 收口 cancelled（R4） |
+| `owner_crash_takeover` | 执行实例被 SIGKILL：租约期内 B 明确 409 不误抢；死实例不拖累其他会话；原实例重启对账收口遗留 run 后恢复推进（R4/R7/R8） |
+| `restart_invariants` | 实例重启后消息俱在、SSE 从 0 重放 seq 连续无洞、写入可推进（R5/R7） |
+
+## 运行
+
+```bash
+# 1) 隔离 PostgreSQL（示例：svc.sh 独立实例；已有实例则直接给 DSN）
+scripts/svc.sh db create mi7 --port 18890
+scripts/svc.sh mi7 start
+DSN="$(scripts/svc.sh db dsn mi7 --reveal)"
+
+# 2) 构建 + 运行（Make 目标自动构建二进制并注入）
+make test-multi-instance AGENT_PG_TEST_DSN="$DSN?sslmode=disable"
+
+# 3) 回收
+scripts/svc.sh mi7 stop && scripts/svc.sh db rm mi7
+```
+
+门控：`AGENT_PG_TEST_DSN`（真实 PostgreSQL，CI test-pg job 必跑不跳过）与
+`AGENT_BIN`（由 Make 目标构建注入）。测试自建临时数据库、自起进程并在
+退出时回收；不打任何生产服务。
+
+## 已实测发现并修复的缺陷（历史）
+
+本验收在 U7 落地时抓出三个跨实例真实缺陷（均已有回归测试）：
+
+1. **HITL resolve 落卡短路命令路径**：非执行实例先把卡标记 resolved，
+   命令 fallback 的 `!card.Resolved` 判定被跳过，执行者永远等不到答案
+   （backend `TestCrossInstanceResolveDoesNotSettleCard`）。
+2. **跨实例 cancel 命令 conversation_id 为空**：消费侧按执行权 JOIN
+   匹配，空串永远无法送达。
+3. **CancelRun 假成功**：非执行实例无句柄时 `CancelRun` 返回 nil，HTTP
+   层 200 受理但 run 永远停不下（命令路径被短路）。

tests/integration/multi-instance/go.mod

@@ -0,0 +1,13 @@
+module agent-system-dev/tests/integration/multi-instance
+
+go 1.27
+
+require github.com/jackc/pgx/v5 v5.9.2
+
+require (
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	golang.org/x/sync v0.17.0 // indirect
+	golang.org/x/text v0.29.0 // indirect
+)

tests/integration/multi-instance/go.sum

@@ -0,0 +1,26 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
+github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
+golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
+golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

tests/integration/multi-instance/multi_instance_test.go

@@ -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)
+}

tests/load/collaboration/.gitignore

@@ -0,0 +1,2 @@
+/.build/
+/results/

tests/load/collaboration/README.md

@@ -0,0 +1,50 @@
+# 协作存储压测套件（U7）
+
+方案 §3 负载合同 / §5 U7 的业务压测入口。**只打隔离栈**（自建 PG 实例 +
+mock 模式 backend），绝不打生产服务；模拟模型（mock）用于容量基线，真实
+模型的全链路验证按方案另行小流量执行。
+
+## 组成
+
+| 文件 | 职责 |
+|---|---|
+| `run_load.sh` | 编排入口：起隔离栈 → 引导账号 → k6 阶梯 → SSE 探针 → 不变量对账 → 回收 |
+| `k6/collab_api.js` | 非模型普通 API 的混合读写负载（登录/项目/会话/卡片/设置/消息，requestId 幂等重放抽样；预期 409/429 单列计数） |
+| `sse_probe/` | SSE 持久事件可见延迟探针（Go）：k6 核心不支持 SSE 流式读取，P95≤1s 指标由此承担，含 seq 单调性断言 |
+| `results/<ts>/` | 每次运行的原始输出：`k6-summary.json`、`k6-output.txt`、`sse-probe.json`、`invariants.txt`、`app.log` |
+
+## 快速开始
+
+```bash
+# 冒烟（~2 分钟，无需 k6——缺失时自动跳过 k6 段并给出安装指引）
+tests/load/collaboration/run_load.sh --profile smoke
+
+# 阶梯（10→30→100 VU，~12 分钟；k6: brew install k6）
+tests/load/collaboration/run_load.sh --profile ladder --users 30
+
+# 长稳（60 分钟 / 2 小时混合，负载合同的两个正式窗口）
+tests/load/collaboration/run_load.sh --profile sustained60m
+tests/load/collaboration/run_load.sh --profile sustained2h
+
+# 复用已有隔离 PG；保留栈便于复查
+tests/load/collaboration/run_load.sh --dsn 'postgresql://…/agent' --keep
+```
+
+## 阈值与口径（负载合同，待实测校准）
+
+- 非模型普通 API：读路径 P95≤300ms、P99≤1s（k6 threshold 按 `read` 组单列）。
+- 非预期失败率≤0.1%（`http_req_failed<0.001`）；预期拥塞（会话忙 409 /
+  队列满 429）记入 `expected_contention` 计数器，**不**计入失败。
+- SSE 持久事件可见延迟 P95≤1s + seq 零乱序（探针退出码即断言）。
+- 数据不变量零违反：events 每会话 seq 无重复（SQL 对账，违反即 exit 1）；
+  各表计数与孤儿 run 报告落 `invariants.txt`。
+
+## 边界（诚实声明）
+
+- 首期为**单实例**压测（mock 模式、执行原语关闭）；分布式限流窗口与双实例
+  放行指标在双栈就位后扩展（`--dsn` 复用 + 第二实例待加）。
+- 容量数字（连接等待、CPU/内存/IO、pgbench 数据库基线）须绑定组件 SHA、
+  配置与数据集记录在结果目录；`pgbench` 只测数据库能力，不替代本套业务
+  验收（方案 §5 U7）。
+- 正式 60min/2h 窗口的执行与双主机验证需要用户指定资源与时间；本仓提供
+  的是可复核入口，不预填容量结论。

tests/load/collaboration/k6/collab_api.js

@@ -0,0 +1,137 @@
+// 协作存储 U7 压测（方案 §3 负载合同 / §5 U7）：k6 覆盖非模型普通 API
+// 的吞吐与延迟阶梯；SSE 可见延迟 P95 由 Go 探针（../sse_probe）承担——
+// k6 核心不支持 SSE 流式读取，两者分工记录在同一份结果目录。
+//
+// 目标实例：run_load.sh 起的隔离栈（mock 模式 backend + 隔离 PG），绝不打
+// 生产服务。预期限流/冲突（409/429）单列计数，不计入失败。
+//
+// 阶梯与环境变量（由 run_load.sh 注入）：
+//   BASE_URL   目标基址（默认 http://127.0.0.1:18980）
+//   ADMIN_*    引导管理员（run_load.sh 预建；setup 里登录后建压测用户）
+//   USERS      压测用户数（默认 30；每 VU 绑定一个用户）
+//   PROFILE    smoke | ladder | sustained60m | sustained2h（默认 smoke）
+//
+// 阈值（负载合同，待实测后校准）：
+//   读路径 p(95)<300ms、全量 p(99)<1s、非预期失败率<0.1%。
+
+import http from 'k6/http';
+import { check, group, sleep } from 'k6';
+import { Counter, Trend } from 'k6/metrics';
+import exec from 'k6/execution';
+
+const BASE = __ENV.BASE_URL || 'http://127.0.0.1:18980';
+const ADMIN_OWNER = __ENV.ADMIN_OWNER || 'load_admin';
+const ADMIN_PASS = __ENV.ADMIN_PASS || 'load-admin-pass';
+const USERS = parseInt(__ENV.USERS || '30', 10);
+const PROFILE = __ENV.PROFILE || 'smoke';
+
+// 预期拥塞（负载合同要求单列，不算失败）：409 会话忙 / 429 队列满。
+const expectedContention = new Counter('expected_contention');
+const msgLatency = new Trend('message_roundtrip', true);
+
+const stages = {
+  smoke: [{ duration: '30s', target: 10 }],
+  ladder: [
+    { duration: '2m', target: 10 },
+    { duration: '3m', target: 30 },
+    { duration: '5m', target: 100 },
+    { duration: '2m', target: 0 },
+  ],
+  sustained60m: [
+    { duration: '2m', target: 30 },
+    { duration: '56m', target: 30 },
+    { duration: '2m', target: 0 },
+  ],
+  sustained2h: [
+    { duration: '3m', target: 30 },
+    { duration: '114m', target: 30 },
+    { duration: '3m', target: 0 },
+  ],
+};
+
+export const options = {
+  stages: stages[PROFILE] || stages.smoke,
+  thresholds: {
+    // 非预期失败：2xx 之外的（409/429 已在 expected_contention 单列）。
+    http_req_failed: ['rate<0.001'],
+    'http_req_duration{group:::read}': ['p(95)<300', 'p(99)<1000'],
+    http_req_duration: ['p(99)<1000'],
+  },
+  summaryTrendStats: ['avg', 'p(50)', 'p(95)', 'p(99)', 'max'],
+};
+
+// ---- setup：管理员建压测用户，各自建项目与会话（数据集随用户数扩展） ----
+
+export function setup() {
+  // setup 上下文的 cookie jar 自动附着（k6 语义：jar 不作为请求器使用）。
+  const login = http.post(`${BASE}/api/auth/login`, JSON.stringify({ owner: ADMIN_OWNER, password: ADMIN_PASS }), { headers: { 'Content-Type': 'application/json' } });
+  if (login.status !== 200) throw new Error(`admin login: ${login.status} ${login.body}`);
+  // 首登强制改密由 run_load.sh 预处理（CLI 建号即带终态口令），此处直接可用。
+
+  const users = [];
+  for (let i = 0; i < USERS; i++) {
+    const owner = `load_u${i}`;
+    const password = `load-user-pass-${i}-Aa9`;
+    const created = http.post(`${BASE}/api/admin/users`, JSON.stringify({ owner, name: `压测用户${i}`, password, isAdmin: false }), { headers: { 'Content-Type': 'application/json' } });
+    // 409 = 已存在（重复压测同一栈），接受。
+    if (created.status !== 200 && created.status !== 409) throw new Error(`create user ${owner}: ${created.status} ${created.body}`);
+    users.push({ owner, password });
+  }
+  return { users };
+}
+
+// ---- default：每 VU 一个用户身份，混合读写 ----
+
+export default function (data) {
+  const u = data.users[exec.scenario.iterationInTest % data.users.length];
+  // VU 上下文的 cookie jar 自动附着；每 VU 绑定一个用户身份。
+  const login = http.post(`${BASE}/api/auth/login`, JSON.stringify({ owner: u.owner, password: u.password }), { headers: { 'Content-Type': 'application/json' } });
+  if (login.status !== 200) {
+    check(null, { 'user login': () => false });
+    return;
+  }
+  // 会话/项目绑定 VU（VU 内多次迭代复用同一身份的会话，形成热点会话）。
+  if (!u.__ctx) {
+    const proj = http.post(`${BASE}/api/projects`, JSON.stringify({ name: `压测-${u.owner}` }), { headers: { 'Content-Type': 'application/json' } });
+    const conv = http.post(`${BASE}/api/projects/${proj.json().id}/conversations`, JSON.stringify({ title: '压测会话' }), { headers: { 'Content-Type': 'application/json' } });
+    u.__ctx = { projectID: proj.json().id, convID: conv.json().id };
+  }
+  const { projectID, convID } = u.__ctx;
+
+  group('read', () => {
+    const projs = http.get(`${BASE}/api/projects`);
+    check(projs, { 'list projects 2xx': (r) => r.status === 200 });
+    const detail = http.get(`${BASE}/api/conversations/${convID}`);
+    check(detail, { 'conversation detail 2xx': (r) => r.status === 200 });
+    const cards = http.get(`${BASE}/api/conversations/${convID}/cards`);
+    check(cards, { 'cards 2xx': (r) => r.status === 200 });
+    const settings = http.get(`${BASE}/api/settings`);
+    check(settings, { 'settings 2xx': (r) => r.status === 200 });
+  });
+
+  group('write', () => {
+    const requestId = `req-${u.owner}-${exec.scenario.iterationInTest}`;
+    const body = JSON.stringify({ text: `压测消息 ${Date.now()}`, requestId });
+    const t0 = Date.now();
+    const sent = http.post(`${BASE}/api/conversations/${convID}/messages`, body, { headers: { 'Content-Type': 'application/json' } });
+    if (sent.status === 200) {
+      msgLatency.add(Date.now() - t0);
+      const id = sent.json().message && sent.json().message.id;
+      check(sent, { 'message has id': () => !!id });
+      // 幂等重放（矩阵行：同 requestId 返回同一逻辑结果）：抽样 10%。
+      if (exec.scenario.iterationInTest % 10 === 0) {
+        const replay = http.post(`${BASE}/api/conversations/${convID}/messages`, body, { headers: { 'Content-Type': 'application/json' } });
+        check(replay, {
+          'requestId replay same message': (r) => r.status === 200 && r.json().message && r.json().message.id === id,
+        });
+      }
+    } else if (sent.status === 409 || sent.status === 429) {
+      // 热点会话忙/队列满：预期拥塞，单列。
+      expectedContention.add(1);
+    } else {
+      check(sent, { 'message send 2xx': () => false });
+    }
+  });
+
+  sleep(1);
+}

tests/load/collaboration/run_load.sh

@@ -0,0 +1,143 @@
+#!/usr/bin/env bash
+# 协作存储 U7 压测入口（方案 §3 负载合同 / §5 U7）。
+#
+# 起一个隔离栈（backend mock 单实例 + 隔离 PG）→ 预建引导账号 → k6 阶梯
+# → SSE 可见延迟探针 → 数据不变量对账 → 结果落 results/<ts>/ → 回收。
+# 绝不打生产服务：默认自建 PG 实例（svc.sh）；--dsn 复用已有隔离实例。
+#
+# 用法：
+#   tests/load/collaboration/run_load.sh [--profile smoke|ladder|sustained60m|sustained2h]
+#                                        [--users N] [--dsn <PG DSN>] [--port <app port>]
+#                                        [--keep] [--skip-k6] [--skip-probe] [--help]
+# 依赖：bash4、go（构建 backend/探针）、psql（对账）、curl；
+#        k6 可选（缺失时跳过 k6 段并给出安装指引，探针与对账照跑）。
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
+HERE="$ROOT/tests/load/collaboration"
+PROFILE=smoke USERS=30 APP_PORT=18980 PG_PORT=18981 KEEP=0 SKIP_K6=0 SKIP_PROBE=0
+DSN=""
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --profile) PROFILE="$2"; shift 2 ;;
+    --users) USERS="$2"; shift 2 ;;
+    --dsn) DSN="$2"; shift 2 ;;
+    --port) APP_PORT="$2"; shift 2 ;;
+    --keep) KEEP=1; shift ;;
+    --skip-k6) SKIP_K6=1; shift ;;
+    --skip-probe) SKIP_PROBE=1; shift ;;
+    -h|--help) sed -n '2,16p' "$0"; exit 0 ;;
+    *) echo "unknown arg: $1" >&2; exit 2 ;;
+  esac
+done
+
+TS="$(date +%Y%m%d-%H%M%S)"
+RESULTS="$HERE/results/$TS"
+mkdir -p "$RESULTS"
+OWNED_PG=0
+APP_PID=""
+cleanup() {
+  [[ -n "$APP_PID" ]] && kill "$APP_PID" 2>/dev/null || true
+  if [[ $OWNED_PG -eq 1 && $KEEP -eq 0 ]]; then
+    "$ROOT/scripts/svc.sh" u7load stop >/dev/null 2>&1 || true
+    "$ROOT/scripts/svc.sh" rm u7load --purge >/dev/null 2>&1 || true
+  fi
+  if [[ $KEEP -eq 1 ]]; then
+    echo "[load] --keep：栈保留（PG=u7load:$PG_PORT，app=127.0.0.1:$APP_PORT，结果 $RESULTS）"
+  fi
+  return 0 2>/dev/null || true
+}
+trap cleanup EXIT
+
+# ---- 1) PG：自建隔离实例（或复用 --dsn） ----
+if [[ -z "$DSN" ]]; then
+  echo "[load] 创建隔离 PG 实例 u7load（端口 $PG_PORT）"
+  "$ROOT/scripts/svc.sh" db create u7load --port "$PG_PORT" >/dev/null
+  OWNED_PG=1
+  "$ROOT/scripts/svc.sh" u7load start >/dev/null
+  DSN="$("$ROOT/scripts/svc.sh" db dsn u7load --reveal)"
+fi
+echo "[load] PG DSN 已就绪（密码不回显）"
+
+# ---- 2) 构建 backend 与探针 ----
+echo "[load] 构建 backend（worktree 当前源码）与 SSE 探针"
+BIN="$HERE/.build/agent-system"
+mkdir -p "$HERE/.build"
+( cd "$ROOT/modules/backend" && CGO_ENABLED=0 go build -o "$BIN" ./cmd/agent-system )
+( cd "$HERE/sse_probe" && go build -o "$HERE/.build/sse_probe" . )
+
+# ---- 3) 压测库 + 引导账号（CLI 直连，终态口令避免首登改密分支） ----
+# DSN 规范化：库名换 u7load_app、query 补 sslmode=disable（密码不回显）。
+ADMIN_DSN="$(python3 - "$DSN" <<'PY'
+import sys
+from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
+u = urlsplit(sys.argv[1])
+q = dict(parse_qsl(u.query))
+q.setdefault("sslmode", "disable")
+print(urlunsplit((u.scheme, u.netloc, "/" + (u.path.lstrip("/") or "postgres"), urlencode(q), "")))
+PY
+)"
+psql "$ADMIN_DSN" -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS u7load_app WITH (FORCE)" -c "CREATE DATABASE u7load_app" >/dev/null
+APP_DSN="$(python3 - "$ADMIN_DSN" <<'PY'
+import sys
+from urllib.parse import urlsplit, urlunsplit
+u = urlsplit(sys.argv[1])
+print(urlunsplit((u.scheme, u.netloc, "/u7load_app", u.query, "")))
+PY
+)"
+"$BIN" user create --dsn "$APP_DSN" --admin --owner load_admin --password load-admin-pass-Aa1 >/dev/null
+"$BIN" user create --dsn "$APP_DSN" --owner probe_user --password probe-user-pass-Aa9 >/dev/null
+
+# ---- 4) 起被测实例（mock、单实例首期；执行原语关闭——压测不打执行面） ----
+echo "[load] 启动被测实例 127.0.0.1:$APP_PORT（mock / tier=team / auth=on）"
+APP_LOG="$RESULTS/app.log"
+"$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$APP_PORT" --tier team --auth on --mock \
+  --instance u7load --disable-exec-tools >"$APP_LOG" 2>&1 &
+APP_PID=$!
+# auth=on 时 /api/health 同样要求鉴权：200/401 都视为已就绪（服务在正确应答）。
+app_ready() { code=$(curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:$APP_PORT/api/health"); [[ "$code" == 200 || "$code" == 401 ]]; }
+for _ in $(seq 1 60); do app_ready && break; sleep 0.5; done
+app_ready || { echo "[load] app 未就绪，日志 $APP_LOG" >&2; exit 1; }
+
+BASE="http://127.0.0.1:$APP_PORT"
+
+# ---- 5) k6 阶梯 ----
+if [[ $SKIP_K6 -eq 1 ]]; then
+  echo "[load] --skip-k6：跳过 k6 段"
+elif ! command -v k6 >/dev/null 2>&1; then
+  cat >"$RESULTS/k6-skipped.txt" <<'EOF'
+k6 未安装：本段跳过（探针与对账照跑）。
+安装（本机 linuxbrew）：brew install k6
+安装后重跑：tests/load/collaboration/run_load.sh --profile <同参数>
+EOF
+  echo "[load] k6 未安装，跳过（见 $RESULTS/k6-skipped.txt）"
+else
+  echo "[load] k6 profile=$PROFILE users=$USERS"
+  ( cd "$HERE" && BASE_URL="$BASE" ADMIN_OWNER=load_admin ADMIN_PASS=load-admin-pass-Aa1 \
+      USERS="$USERS" PROFILE="$PROFILE" \
+      k6 run --summary-export "$RESULTS/k6-summary.json" k6/collab_api.js ) | tee "$RESULTS/k6-output.txt"
+fi
+
+# ---- 6) SSE 可见延迟探针 ----
+if [[ $SKIP_PROBE -eq 0 ]]; then
+  echo "[load] SSE 可见延迟探针（60s，10 会话）"
+  "$HERE/.build/sse_probe" -base "$BASE" -dur 60s -convs 10 -out "$RESULTS/sse-probe.json"
+fi
+
+# ---- 7) 数据不变量对账（方案 §3：零违反） ----
+echo "[load] 数据不变量对账"
+psql "$APP_DSN" -v ON_ERROR_STOP=1 \
+  -c "SELECT 'users' k, COUNT(*) v FROM users UNION ALL SELECT 'projects', COUNT(*) FROM projects UNION ALL SELECT 'conversations', COUNT(*) FROM conversations UNION ALL SELECT 'messages', COUNT(*) FROM messages UNION ALL SELECT 'events', COUNT(*) FROM events UNION ALL SELECT 'runs', COUNT(*) FROM runs" \
+  -c "SELECT 'dup_seq_per_conv' k, COUNT(*) v FROM (SELECT conversation_id, seq FROM events GROUP BY 1,2 HAVING COUNT(*)>1) d" \
+  -c "SELECT 'nonterminal_orphan_runs' k, COUNT(*) v FROM runs WHERE status IN ('pending','running','waiting_hitl')" \
+  | tee "$RESULTS/invariants.txt"
+
+DUP=$(psql "$APP_DSN" -tAc "SELECT COUNT(*) FROM (SELECT conversation_id, seq FROM events GROUP BY 1,2 HAVING COUNT(*)>1) d")
+if [[ "$DUP" != "0" ]]; then
+  echo "[load] 对账失败：events 每会话 seq 出现重复（$DUP 组）" >&2
+  exit 1
+fi
+echo "[load] 对账通过：无重复 seq；孤儿 run 报告见 invariants.txt（探针正常收口应为 0）"
+
+echo "[load] 完成。结果目录：$RESULTS"

tests/load/collaboration/sse_probe/go.mod

@@ -0,0 +1,3 @@
+module agent-system-dev/tests/load/collaboration/sse_probe
+
+go 1.27.1

tests/load/collaboration/sse_probe/main.go

@@ -0,0 +1,241 @@
+// 协作存储 U7：SSE 持久事件可见延迟探针（负载合同：P95≤1s）。
+// k6 核心不支持 SSE 流式读取，此项由本 Go 探针承担；与 k6 结果记录在
+// 同一目录（run_load.sh 编排）。
+//
+// 行为：并发 C 个会话各挂一个 SSE 订阅（共享登录 cookie）；写者按固定
+// 间隔轮转 POST 消息（mock 即时回合）；统计「POST 前基准 seq → 会话内
+// 新帧到达」延迟分布与 seq 单调性，输出 JSON 报告并以退出码执行
+// P95≤1s、零乱序断言（数据不变量零违反，方案 §3）。
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/cookiejar"
+	"os"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+type sample struct {
+	Conversation string  `json:"conversation"`
+	SendSeq      int64   `json:"sendSeq"`
+	LatencyMS    float64 `json:"latencyMs"`
+}
+
+type report struct {
+	Target      string   `json:"target"`
+	Duration    string   `json:"duration"`
+	Subscribers int      `json:"subscribers"`
+	Messages    int      `json:"messages"`
+	Samples     int      `json:"samples"`
+	P50MS       float64  `json:"p50Ms"`
+	P95MS       float64  `json:"p95Ms"`
+	P99MS       float64  `json:"p99Ms"`
+	MaxMS       float64  `json:"maxMs"`
+	OutOfOrder  int      `json:"outOfOrder"` // seq 乱序/回退（不变量违反）
+	ThresholdMS float64  `json:"thresholdMs"`
+	Pass        bool     `json:"pass"`
+	Detail      []sample `json:"samplesDetail"`
+}
+
+var (
+	jar, _       = cookiejar.New(nil)
+	client       = &http.Client{Timeout: 10 * time.Second, Jar: jar}
+	streamClient = &http.Client{Jar: jar} // SSE：无整体超时，stop 控制
+	pendingMu    sync.Mutex
+	pending      = map[string]pendingSend{}
+	outOfOrderMu sync.Mutex
+	outOfOrderN  int
+	samplesMu    sync.Mutex
+	samples      []sample
+)
+
+type pendingSend struct {
+	seq  int64
+	sent time.Time
+}
+
+func main() {
+	base := flag.String("base", "http://127.0.0.1:18980", "目标实例基址")
+	owner := flag.String("owner", "probe_user", "探针用户")
+	password := flag.String("password", "probe-user-pass-Aa9", "探针用户口令")
+	convs := flag.Int("convs", 10, "并发订阅会话数")
+	every := flag.Duration("every", 500*time.Millisecond, "写者间隔（轮转各会话）")
+	dur := flag.Duration("dur", 60*time.Second, "采样时长")
+	out := flag.String("out", "", "JSON 报告输出路径（空 = stdout）")
+	flag.Parse()
+
+	login(*base, *owner, *password)
+
+	projectID := apiJSON("POST", *base, "/api/projects", map[string]any{"name": "sse-probe"})["id"].(string)
+	convIDs := make([]string, *convs)
+	for i := range convIDs {
+		convIDs[i] = apiJSON("POST", *base, "/api/projects/"+projectID+"/conversations", map[string]any{"title": fmt.Sprintf("probe-%d", i)})["id"].(string)
+	}
+
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	for _, cid := range convIDs {
+		wg.Add(1)
+		go func(cid string) {
+			defer wg.Done()
+			subscribe(*base, cid, stop)
+		}(cid)
+	}
+	time.Sleep(500 * time.Millisecond) // 订阅就绪
+
+	deadline := time.Now().Add(*dur)
+	sent := 0
+	for i := 0; time.Now().Before(deadline); i++ {
+		cid := convIDs[i%len(convIDs)]
+		detail := apiJSON("GET", *base, "/api/conversations/"+cid, nil)
+		seq, _ := detail["lastSeq"].(float64)
+		pendingMu.Lock()
+		pending[cid] = pendingSend{seq: int64(seq), sent: time.Now()}
+		pendingMu.Unlock()
+		apiJSON("POST", *base, "/api/conversations/"+cid+"/messages", map[string]any{"text": fmt.Sprintf("probe %d", i)})
+		sent++
+		time.Sleep(*every)
+	}
+
+	time.Sleep(3 * time.Second) // 最后一批帧在途余量
+	close(stop)
+	wg.Wait()
+
+	r := buildReport(*base, *dur, *convs, sent)
+	data, _ := json.MarshalIndent(r, "", "  ")
+	if *out != "" {
+		if err := os.WriteFile(*out, data, 0o644); err != nil {
+			fmt.Fprintln(os.Stderr, "write report:", err)
+			os.Exit(2)
+		}
+		fmt.Println("report:", *out)
+	} else {
+		fmt.Println(string(data))
+	}
+	if !r.Pass {
+		fmt.Fprintf(os.Stderr, "sse probe FAIL: p95=%.1fms (≤%.0fms) samples=%d outOfOrder=%d\n", r.P95MS, r.ThresholdMS, r.Samples, r.OutOfOrder)
+		os.Exit(1)
+	}
+	fmt.Printf("sse probe PASS: p95=%.1fms samples=%d\n", r.P95MS, r.Samples)
+}
+
+func login(base, owner, pass string) {
+	body, _ := json.Marshal(map[string]string{"owner": owner, "password": pass})
+	resp, err := client.Post(base+"/api/auth/login", "application/json", bytes.NewReader(body))
+	if err != nil || resp.StatusCode != 200 {
+		fmt.Fprintf(os.Stderr, "probe login: err=%v status=%v\n", err, statusOf(resp))
+		os.Exit(2)
+	}
+	resp.Body.Close()
+}
+
+func statusOf(resp *http.Response) string {
+	if resp == nil {
+		return "<nil>"
+	}
+	return strconv.Itoa(resp.StatusCode)
+}
+
+func apiJSON(method, base, path string, body any) map[string]any {
+	var rd io.Reader
+	_ = rd
+	if body != nil {
+		buf, _ := json.Marshal(body)
+		rd = bytes.NewReader(buf)
+	}
+	req, _ := http.NewRequest(method, base+path, rd)
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "%s %s: %v\n", method, path, err)
+		os.Exit(2)
+	}
+	defer resp.Body.Close()
+	out := map[string]any{}
+	if resp.StatusCode != 200 {
+		fmt.Fprintf(os.Stderr, "%s %s: status %d\n", method, path, resp.StatusCode)
+		os.Exit(2)
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		fmt.Fprintf(os.Stderr, "%s %s: decode %v\n", method, path, err)
+		os.Exit(2)
+	}
+	return out
+}
+
+// subscribe 读会话 SSE：seq 严格递增校验；帧到达时与写者基准配对成延迟样本。
+func subscribe(base, cid string, stop <-chan struct{}) {
+	req, _ := http.NewRequest(http.MethodGet, base+"/api/events?conversationId="+cid, nil)
+	resp, err := streamClient.Do(req)
+	if err != nil {
+		return
+	}
+	defer resp.Body.Close()
+	sc := bufio.NewScanner(resp.Body)
+	sc.Buffer(make([]byte, 0, 64*1024), 4<<20)
+	last := int64(-1)
+	for sc.Scan() {
+		select {
+		case <-stop:
+			return
+		default:
+		}
+		line := sc.Text()
+		if !strings.HasPrefix(line, "id: ") {
+			continue
+		}
+		seq, _ := strconv.ParseInt(strings.TrimPrefix(line, "id: "), 10, 64)
+		arrived := time.Now()
+		if last != -1 && seq <= last {
+			outOfOrderMu.Lock()
+			outOfOrderN++
+			outOfOrderMu.Unlock()
+		}
+		last = seq
+		pendingMu.Lock()
+		p, ok := pending[cid]
+		pendingMu.Unlock()
+		if ok && seq > p.seq {
+			pendingMu.Lock()
+			delete(pending, cid)
+			pendingMu.Unlock()
+			samplesMu.Lock()
+			samples = append(samples, sample{Conversation: cid, SendSeq: seq, LatencyMS: float64(arrived.Sub(p.sent)) / float64(time.Millisecond)})
+			samplesMu.Unlock()
+		}
+	}
+}
+
+func buildReport(base string, dur time.Duration, subs, msgs int) report {
+	samplesMu.Lock()
+	detail := append([]sample(nil), samples...)
+	samplesMu.Unlock()
+	outOfOrderMu.Lock()
+	ooo := outOfOrderN
+	outOfOrderMu.Unlock()
+
+	r := report{Target: base, Duration: dur.String(), Subscribers: subs, Messages: msgs, Samples: len(detail), OutOfOrder: ooo, ThresholdMS: 1000, Detail: detail}
+	if len(detail) > 0 {
+		vals := make([]float64, len(detail))
+		for i, s := range detail {
+			vals[i] = s.LatencyMS
+		}
+		sort.Float64s(vals)
+		pct := func(p float64) float64 { return vals[int(float64(len(vals)-1)*p)] }
+		r.P50MS, r.P95MS, r.P99MS, r.MaxMS = pct(0.50), pct(0.95), pct(0.99), vals[len(vals)-1]
+	}
+	r.Pass = len(detail) > 0 && r.P95MS <= r.ThresholdMS && ooo == 0
+	return r
+}