cmd/agent-system/main.go

@@ -12,6 +12,7 @@ import (
 
 	"agent-system-backend/internal/app"
 	"agent-system-backend/internal/config"
+	"agent-system-backend/internal/store"
 )
 
 var version = "0.1.0-dev"
@@ -48,6 +49,12 @@ func main() {
 	execBackend := fs.String("exec-backend", "", "执行后端：direct（默认，现状直启）| sandbox（bwrap 隔离档：断网、只读系统根、仅会话目录可写、进程数限额）")
 	execDepsDir := fs.String("exec-deps-dir", "", "sandbox 档只读依赖层目录（site-packages 等；空=不挂载）")
 	execWorkspace := fs.String("exec-workspace", "", "sandbox 档工作区模式：空=直写会话目录 | overlay=冻结基底+私有 upper+CAS 回收（S0）")
+	tier := fs.String("tier", "", "部署档：空=auto（DSN→PG，否则 SQLite）| local（SQLite，不允许 DSN）| team（PG fail-closed：缺 DSN/未认证拒绝启动，PG 故障拒绝就绪）")
+	pgMaxOpen := fs.Int("pg-max-open", 0, "PG 最大打开连接（0=默认 16）")
+	pgMaxIdle := fs.Int("pg-max-idle", 0, "PG 最大空闲连接（0=默认 8）")
+	pgIdleSec := fs.Int("pg-conn-max-idle-sec", 0, "PG 空闲连接回收秒数（0=默认 300）")
+	pgLifetimeSec := fs.Int("pg-conn-max-lifetime-sec", 0, "PG 连接最长生命周期秒数（0=不限）")
+	pgQueryTimeoutSec := fs.Int("pg-query-timeout-sec", 0, "PG statement_timeout 秒数（0=不限；覆盖全部业务查询）")
 	fs.Parse(os.Args[1:])
 	cliSet := visited(fs)
 
@@ -56,17 +63,31 @@ func main() {
 		StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance,
 		InsecureListen: *insecure, AllowedHostsRaw: *allowedHosts, DisableExecTools: *noExec,
 		ExecBackend: *execBackend, ExecDepsDir: *execDepsDir, ExecWorkspace: *execWorkspace,
+		Tier: *tier, PGMaxOpen: *pgMaxOpen, PGMaxIdle: *pgMaxIdle,
+		PGConnMaxIdleSec: *pgIdleSec, PGConnMaxLifetimeSec: *pgLifetimeSec, PGQueryTimeoutSec: *pgQueryTimeoutSec,
 	}, cliSet)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, "agent-system:", err)
 		os.Exit(1)
 	}
+	if msgs := resolved.TierProblems(); len(msgs) > 0 {
+		for _, msg := range msgs {
+			fmt.Fprintln(os.Stderr, "agent-system: tier guard:", msg)
+		}
+		os.Exit(1)
+	}
 	if err := app.Run(app.Options{
 		Addr: resolved.Addr, DataDir: resolved.DataDir, PluginsDir: resolved.PluginsDir, ForceMock: resolved.Mock,
 		StorageDSN: resolved.StorageDSN, ArtifactsDir: resolved.ArtifactsDir, AuthMode: resolved.AuthMode,
 		InstanceID: resolved.Instance, InsecureListen: resolved.InsecureListen, AllowedHosts: resolved.AllowedHosts(),
 		DisableExecTools: resolved.DisableExecTools, ExecBackend: resolved.ExecBackend, ExecDepsDir: resolved.ExecDepsDir,
-		ExecWorkspace:    resolved.ExecWorkspace,
+		ExecWorkspace: resolved.ExecWorkspace,
+		Tier:          resolved.Tier,
+		PGPool: store.PoolOptions{
+			MaxOpenConn: resolved.PGMaxOpen, MaxIdleConn: resolved.PGMaxIdle,
+			ConnMaxIdleTimeSec: resolved.PGConnMaxIdleSec, ConnMaxLifetimeSec: resolved.PGConnMaxLifetimeSec,
+			QueryTimeoutSec: resolved.PGQueryTimeoutSec,
+		},
 	}); err != nil {
 		fmt.Fprintln(os.Stderr, "agent-system:", err)
 		os.Exit(1)
@@ -85,6 +106,16 @@ func visited(fs *flag.FlagSet) map[string]bool {
 			name = "allowedHosts"
 		case "disable-exec-tools":
 			name = "disableExecTools"
+		case "pg-max-open":
+			name = "pgMaxOpen"
+		case "pg-max-idle":
+			name = "pgMaxIdle"
+		case "pg-conn-max-idle-sec":
+			name = "pgConnMaxIdleSec"
+		case "pg-conn-max-lifetime-sec":
+			name = "pgConnMaxLifetimeSec"
+		case "pg-query-timeout-sec":
+			name = "pgQueryTimeoutSec"
 		}
 		set[name] = true
 	})
@@ -113,12 +144,20 @@ func runConfig(args []string) int {
 	execBackend := fs.String("exec-backend", "", "")
 	execDepsDir := fs.String("exec-deps-dir", "", "")
 	execWorkspace := fs.String("exec-workspace", "", "")
+	tier := fs.String("tier", "", "")
+	pgMaxOpen := fs.Int("pg-max-open", 0, "")
+	pgMaxIdle := fs.Int("pg-max-idle", 0, "")
+	pgIdleSec := fs.Int("pg-conn-max-idle-sec", 0, "")
+	pgLifetimeSec := fs.Int("pg-conn-max-lifetime-sec", 0, "")
+	pgQueryTimeoutSec := fs.Int("pg-query-timeout-sec", 0, "")
 	fs.Parse(args[1:])
 	resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{
 		Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock,
 		StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance,
 		InsecureListen: *insecure, AllowedHostsRaw: *allowedHosts, DisableExecTools: *noExec,
 		ExecBackend: *execBackend, ExecDepsDir: *execDepsDir, ExecWorkspace: *execWorkspace,
+		Tier: *tier, PGMaxOpen: *pgMaxOpen, PGMaxIdle: *pgMaxIdle,
+		PGConnMaxIdleSec: *pgIdleSec, PGConnMaxLifetimeSec: *pgLifetimeSec, PGQueryTimeoutSec: *pgQueryTimeoutSec,
 	}, visited(fs))
 	if err != nil {
 		fmt.Fprintln(os.Stderr, "agent-system:", err)

internal/app/app.go

@@ -85,6 +85,42 @@ type Options struct {
 	// ExecWorkspace = "overlay" 时 sandbox 档启用 S0 工作区（冻结基底+私有
 	// upper+CAS 回收）；空 = EXT-16 现状直写会话目录。
 	ExecWorkspace string
+	// Tier 部署档（协作存储 KTD1）：空=auto（DSN→PG 否则 SQLite）；local=SQLite；
+	// team=PG fail-closed（缺 DSN/未认证拒绝启动；PG 故障拒绝就绪，不回落 SQLite）。
+	Tier string
+	// PGPool PG 连接预算（KTD12；零值走默认）。仅 PG 档生效。
+	PGPool store.PoolOptions
+}
+
+// tierName 输出部署档名（诊断统一拼写：空 = auto）。
+func tierName(opt Options) string {
+	if opt.Tier == "" {
+		return "auto"
+	}
+	return opt.Tier
+}
+
+// TierRefusal 部署档门禁（KTD1）：返回阻止启动的问题（空 = 放行）。
+// team 档 fail-closed：缺 DSN、未开认证拒绝；PG 故障的拒绝发生在其后的
+// OpenPostgres（Run 对 team 档的错误显式声明不回落 SQLite）。
+func TierRefusal(opt Options) []string {
+	return config.Resolved{Config: config.Config{
+		Addr: opt.Addr, AuthMode: opt.AuthMode, StorageDSN: opt.StorageDSN, Tier: opt.Tier,
+		PGMaxOpen: opt.PGPool.MaxOpenConn, PGMaxIdle: opt.PGPool.MaxIdleConn,
+		PGConnMaxIdleSec: opt.PGPool.ConnMaxIdleTimeSec, PGConnMaxLifetimeSec: opt.PGPool.ConnMaxLifetimeSec,
+		PGQueryTimeoutSec: opt.PGPool.QueryTimeoutSec,
+	}}.TierProblems()
+}
+
+// latestVersion 返回已应用迁移的最高版本（0 = 空账本）。
+func latestVersion(versions []int) int {
+	max := 0
+	for _, v := range versions {
+		if v > max {
+			max = v
+		}
+	}
+	return max
 }
 
 // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。
@@ -93,13 +129,25 @@ func Run(opt Options) error {
 		Service: "agent-system", Module: "app", Level: logsys.LevelInfo,
 		Sinks: []logsys.Sink{&sink.Console{Out: os.Stderr, Format: "json"}},
 	})
-	logger.Info("agent-system starting", "addr", opt.Addr, "data", opt.DataDir)
+	logger.Info("agent-system starting", "addr", opt.Addr, "data", opt.DataDir, "tier", tierName(opt))
 	resolved := config.Resolved{Config: config.Config{
 		Addr: opt.Addr, AuthMode: opt.AuthMode, StorageDSN: opt.StorageDSN, InsecureListen: opt.InsecureListen,
+		Tier: opt.Tier, PGMaxOpen: opt.PGPool.MaxOpenConn, PGMaxIdle: opt.PGPool.MaxIdleConn,
+		PGConnMaxIdleSec: opt.PGPool.ConnMaxIdleTimeSec, PGConnMaxLifetimeSec: opt.PGPool.ConnMaxLifetimeSec,
+		PGQueryTimeoutSec: opt.PGPool.QueryTimeoutSec,
 	}}
 	if msg := resolved.ListenRefusal(); msg != "" {
 		return fmt.Errorf("listen guard: %s", msg)
 	}
+	// 部署档门禁（KTD1 fail-closed）：team 档缺 DSN/未开认证直接拒绝启动；
+	// local 档带 DSN 是配置错误。此处与 config.TierProblems 同规则（纵深防御，
+	// 服务入口不依赖 CLI 是否校验）。
+	if msgs := TierRefusal(opt); len(msgs) > 0 {
+		return fmt.Errorf("tier guard: %s", strings.Join(msgs, "；"))
+	}
+	if opt.Tier == "team" && !opt.DisableExecTools && opt.ExecBackend == "" {
+		log.Printf("warning: team 档未配置执行隔离（disableExecTools 或 execBackend）；向真实多用户开放前必须补齐（方案 KTD1）")
+	}
 	if opt.InsecureListen && !config.IsLoopbackListen(opt.Addr) && !resolved.AuthEnabled() {
 		log.Printf("warning: listening on %s without authentication because insecureListen is set", opt.Addr)
 	}
@@ -110,7 +158,11 @@ func Run(opt Options) error {
 		if artifacts == "" {
 			artifacts = filepath.Join(opt.DataDir, "artifacts")
 		}
-		st, err = store.OpenPostgres(opt.StorageDSN, artifacts)
+		st, err = store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, nil)
+		if err != nil && opt.Tier == "team" {
+			// team 档拒绝就绪（R2）：PG 故障不回落 SQLite，错误必须明说。
+			return fmt.Errorf("team 档打开 PostgreSQL 失败，拒绝就绪（不回落 SQLite）: %w", err)
+		}
 	} else {
 		st, err = store.OpenSQLite(filepath.Join(opt.DataDir, "agent-system.db"))
 	}
@@ -118,7 +170,15 @@ func Run(opt Options) error {
 		return fmt.Errorf("open store: %w", err)
 	}
 	defer st.Close()
-	log.Printf("storage tier: %s", st.Dialect())
+	pool := st.Pool()
+	if versions, verr := st.MigrationStatus(); verr == nil {
+		log.Printf("storage tier: %s (dialect=%s pool=%d/%d idleSec=%d lifetimeSec=%d schema=v%d)",
+			tierName(opt), st.Dialect(), pool.MaxOpenConn, pool.MaxIdleConn, pool.ConnMaxIdleTimeSec,
+			pool.ConnMaxLifetimeSec, latestVersion(versions))
+	} else {
+		log.Printf("storage tier: %s (dialect=%s pool=%d/%d schema=? migration status unavailable: %v)",
+			tierName(opt), st.Dialect(), pool.MaxOpenConn, pool.MaxIdleConn, verr)
+	}
 
 	b := bus.New()
 	reg := tools.NewRegistry()

internal/app/tier_test.go

@@ -0,0 +1,31 @@
+package app
+
+import (
+	"strings"
+	"testing"
+)
+
+// 协作存储方案 U2（KTD1/R2）：服务入口的部署档门禁。
+// team 档缺 DSN / 未开认证必须拒绝启动——不创建 SQLite、不进入装配。
+func TestTierRefusalTeam(t *testing.T) {
+	msgs := TierRefusal(Options{Tier: "team", Addr: "127.0.0.1:8788"})
+	if len(msgs) != 2 {
+		t.Fatalf("team without dsn/auth: %v", msgs)
+	}
+	joined := strings.Join(msgs, "；")
+	if !strings.Contains(joined, "PostgreSQL DSN") || !strings.Contains(joined, "认证") {
+		t.Fatalf("refusal wording: %v", msgs)
+	}
+	msgs = TierRefusal(Options{Tier: "team", Addr: "127.0.0.1:8788", StorageDSN: "postgres://u:p@h/db", AuthMode: "on"})
+	if len(msgs) != 0 {
+		t.Fatalf("team with dsn+auth should pass: %v", msgs)
+	}
+	msgs = TierRefusal(Options{Tier: "local", Addr: "127.0.0.1:8788", StorageDSN: "postgres://u:p@h/db"})
+	if len(msgs) != 1 || !strings.Contains(msgs[0], "local 档不使用") {
+		t.Fatalf("local with dsn: %v", msgs)
+	}
+	msgs = TierRefusal(Options{Addr: "127.0.0.1:8788"})
+	if len(msgs) != 0 {
+		t.Fatalf("auto default: %v", msgs)
+	}
+}

internal/config/config.go

@@ -11,6 +11,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"strconv"
 	"strings"
 )
 
@@ -35,6 +36,17 @@ type Config struct {
 	ExecDepsDir string `json:"execDepsDir"`
 	// ExecWorkspace sandbox 档工作区模式：空=直写；overlay=冻结基底+upper。
 	ExecWorkspace string `json:"execWorkspace"`
+	// Tier 部署档（协作存储方案 KTD1）：空=auto（DSN→PG，否则 SQLite）；
+	// local=强制 SQLite（不允许 DSN）；team=强制 PG（缺 DSN/未开认证拒绝启动，
+	// PG 故障拒绝就绪，绝不回落 SQLite）。
+	Tier string `json:"tier"`
+	// PG 连接预算（KTD12）：零值走默认（16/8/300s/不限/不限）；负值非法。
+	// PGQueryTimeoutSec>0 时以 statement_timeout 附着到 PG DSN。
+	PGMaxOpen            int `json:"pgMaxOpen"`
+	PGMaxIdle            int `json:"pgMaxIdle"`
+	PGConnMaxIdleSec     int `json:"pgConnMaxIdleSec"`
+	PGConnMaxLifetimeSec int `json:"pgConnMaxLifetimeSec"`
+	PGQueryTimeoutSec    int `json:"pgQueryTimeoutSec"`
 }
 
 var defaults = Config{
@@ -46,20 +58,26 @@ var defaults = Config{
 
 // envKey 字段 → 环境变量名（AGENT_SYSTEM_ 前缀）。
 var envKey = map[string]string{
-	"addr":             "AGENT_SYSTEM_ADDR",
-	"data":             "AGENT_SYSTEM_DATA",
-	"plugins":          "AGENT_SYSTEM_PLUGINS",
-	"dsn":              "AGENT_SYSTEM_DSN",
-	"artifacts":        "AGENT_SYSTEM_ARTIFACTS",
-	"auth":             "AGENT_SYSTEM_AUTH",
-	"mock":             "AGENT_SYSTEM_MOCK",
-	"instance":         "AGENT_SYSTEM_INSTANCE",
-	"execBackend":      "AGENT_SYSTEM_EXEC_BACKEND",
-	"execDepsDir":      "AGENT_SYSTEM_EXEC_DEPS_DIR",
-	"execWorkspace":    "AGENT_SYSTEM_EXEC_WORKSPACE",
-	"insecureListen":   "AGENT_SYSTEM_INSECURE_LISTEN",
-	"allowedHosts":     "AGENT_SYSTEM_ALLOWED_HOSTS",
-	"disableExecTools": "AGENT_SYSTEM_DISABLE_EXEC_TOOLS",
+	"addr":                 "AGENT_SYSTEM_ADDR",
+	"data":                 "AGENT_SYSTEM_DATA",
+	"plugins":              "AGENT_SYSTEM_PLUGINS",
+	"dsn":                  "AGENT_SYSTEM_DSN",
+	"artifacts":            "AGENT_SYSTEM_ARTIFACTS",
+	"auth":                 "AGENT_SYSTEM_AUTH",
+	"mock":                 "AGENT_SYSTEM_MOCK",
+	"instance":             "AGENT_SYSTEM_INSTANCE",
+	"execBackend":          "AGENT_SYSTEM_EXEC_BACKEND",
+	"execDepsDir":          "AGENT_SYSTEM_EXEC_DEPS_DIR",
+	"execWorkspace":        "AGENT_SYSTEM_EXEC_WORKSPACE",
+	"insecureListen":       "AGENT_SYSTEM_INSECURE_LISTEN",
+	"allowedHosts":         "AGENT_SYSTEM_ALLOWED_HOSTS",
+	"disableExecTools":     "AGENT_SYSTEM_DISABLE_EXEC_TOOLS",
+	"tier":                 "AGENT_SYSTEM_TIER",
+	"pgMaxOpen":            "AGENT_SYSTEM_PG_MAX_OPEN",
+	"pgMaxIdle":            "AGENT_SYSTEM_PG_MAX_IDLE",
+	"pgConnMaxIdleSec":     "AGENT_SYSTEM_PG_CONN_MAX_IDLE_SEC",
+	"pgConnMaxLifetimeSec": "AGENT_SYSTEM_PG_CONN_MAX_LIFETIME_SEC",
+	"pgQueryTimeoutSec":    "AGENT_SYSTEM_PG_QUERY_TIMEOUT_SEC",
 }
 
 // Source 配置来源层级（数值越大优先级越高）。
@@ -109,11 +127,7 @@ func Load(file string, envLookup func(string) (string, bool), cli Config, cliSet
 		}
 		// 文件层按"键存在"覆盖（区分未出现与显式 false/空串；空值有效性由字段语义判定）。
 		for f := range present {
-			if isBoolField(f) {
-				setBool(&cfg, f, boolField(fc, f))
-			} else {
-				setString(&cfg, f, stringField(fc, f))
-			}
+			setField(&cfg, f, fc)
 			src[f] = SrcFile
 		}
 	}
@@ -126,6 +140,12 @@ func Load(file string, envLookup func(string) (string, bool), cli Config, cliSet
 					return Resolved{}, err
 				}
 				setBool(&cfg, f, b)
+			} else if isIntField(f) {
+				n, err := parseInt(key, v)
+				if err != nil {
+					return Resolved{}, err
+				}
+				setInt(&cfg, f, n)
 			} else {
 				setString(&cfg, f, v)
 			}
@@ -135,15 +155,22 @@ func Load(file string, envLookup func(string) (string, bool), cli Config, cliSet
 	// CLI 层：仅显式传入。
 	for f := range cliSet {
 		src[f] = SrcCLI
-		if isBoolField(f) {
-			setBool(&cfg, f, boolField(cli, f))
-		} else {
-			setString(&cfg, f, stringField(cli, f))
-		}
+		setField(&cfg, f, cli)
 	}
 	return Resolved{Config: cfg, sources: src}, nil
 }
 
+// setField 从同类型快照覆盖单个字段（文件/CLI 层复用；字段类型由字段名决定）。
+func setField(dst *Config, f string, src Config) {
+	if isBoolField(f) {
+		setBool(dst, f, boolField(src, f))
+	} else if isIntField(f) {
+		setInt(dst, f, intField(src, f))
+	} else {
+		setString(dst, f, stringField(src, f))
+	}
+}
+
 func loadFile(path string) (Config, map[string]bool, error) {
 	raw, err := os.ReadFile(path)
 	if err != nil {
@@ -186,6 +213,18 @@ func loadFile(path string) (Config, map[string]bool, error) {
 			dst = &c.ExecDepsDir
 		case "execWorkspace":
 			dst = &c.ExecWorkspace
+		case "tier":
+			dst = &c.Tier
+		case "pgMaxOpen":
+			dst = &c.PGMaxOpen
+		case "pgMaxIdle":
+			dst = &c.PGMaxIdle
+		case "pgConnMaxIdleSec":
+			dst = &c.PGConnMaxIdleSec
+		case "pgConnMaxLifetimeSec":
+			dst = &c.PGConnMaxLifetimeSec
+		case "pgQueryTimeoutSec":
+			dst = &c.PGQueryTimeoutSec
 		default:
 			return Config{}, nil, fmt.Errorf("config file %s: unknown key %q", path, k)
 		}
@@ -221,6 +260,8 @@ func stringField(c Config, f string) string {
 		return c.ExecDepsDir
 	case "execWorkspace":
 		return c.ExecWorkspace
+	case "tier":
+		return c.Tier
 	}
 	return ""
 }
@@ -229,6 +270,45 @@ func isBoolField(f string) bool {
 	return f == "mock" || f == "insecureListen" || f == "disableExecTools"
 }
 
+func isIntField(f string) bool {
+	switch f {
+	case "pgMaxOpen", "pgMaxIdle", "pgConnMaxIdleSec", "pgConnMaxLifetimeSec", "pgQueryTimeoutSec":
+		return true
+	}
+	return false
+}
+
+func intField(c Config, f string) int {
+	switch f {
+	case "pgMaxOpen":
+		return c.PGMaxOpen
+	case "pgMaxIdle":
+		return c.PGMaxIdle
+	case "pgConnMaxIdleSec":
+		return c.PGConnMaxIdleSec
+	case "pgConnMaxLifetimeSec":
+		return c.PGConnMaxLifetimeSec
+	case "pgQueryTimeoutSec":
+		return c.PGQueryTimeoutSec
+	}
+	return 0
+}
+
+func setInt(c *Config, f string, v int) {
+	switch f {
+	case "pgMaxOpen":
+		c.PGMaxOpen = v
+	case "pgMaxIdle":
+		c.PGMaxIdle = v
+	case "pgConnMaxIdleSec":
+		c.PGConnMaxIdleSec = v
+	case "pgConnMaxLifetimeSec":
+		c.PGConnMaxLifetimeSec = v
+	case "pgQueryTimeoutSec":
+		c.PGQueryTimeoutSec = v
+	}
+}
+
 func boolField(c Config, f string) bool {
 	switch f {
 	case "insecureListen":
@@ -275,6 +355,8 @@ func setString(c *Config, f, v string) {
 		c.ExecDepsDir = v
 	case "execWorkspace":
 		c.ExecWorkspace = v
+	case "tier":
+		c.Tier = v
 	}
 }
 
@@ -288,6 +370,14 @@ func parseBool(name, v string) (bool, error) {
 	return false, fmt.Errorf("env %s: invalid bool %q", name, v)
 }
 
+func parseInt(name, v string) (int, error) {
+	n, err := strconv.Atoi(strings.TrimSpace(v))
+	if err != nil {
+		return 0, fmt.Errorf("env %s: invalid int %q", name, v)
+	}
+	return n, nil
+}
+
 // Explain 只读诊断：逐字段打印解析值与来源层级，DSN 脱敏。
 func (r Resolved) Explain(w io.Writer, file string) {
 	if file != "" {
@@ -296,16 +386,32 @@ func (r Resolved) Explain(w io.Writer, file string) {
 		fmt.Fprintln(w, "config file: (none)")
 	}
 	fmt.Fprintln(w, "field            value                                                    source")
-	for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock", "instance", "insecureListen", "allowedHosts", "disableExecTools", "execBackend"} {
-		v := stringField(r.Config, f)
+	for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock", "instance", "insecureListen", "allowedHosts", "disableExecTools", "execBackend", "tier", "pgMaxOpen", "pgMaxIdle", "pgConnMaxIdleSec", "pgConnMaxLifetimeSec", "pgQueryTimeoutSec"} {
+		var v string
 		if isBoolField(f) {
 			v = fmt.Sprintf("%t", boolField(r.Config, f))
+		} else if isIntField(f) {
+			n := intField(r.Config, f)
+			if d := intDefault(f); n == 0 && d != 0 {
+				v = fmt.Sprintf("%d (默认)", d)
+			} else {
+				v = fmt.Sprintf("%d", n)
+			}
+		} else {
+			v = stringField(r.Config, f)
 		}
 		if f == "dsn" {
 			v = redact(v)
 		}
 		fmt.Fprintf(w, "%-16s %-56s %s\n", f, v, r.sources[f])
 	}
+	if msgs := r.TierProblems(); len(msgs) > 0 {
+		for _, msg := range msgs {
+			fmt.Fprintf(w, "tier guard: 拒绝启动（%s）\n", msg)
+		}
+	} else {
+		fmt.Fprintln(w, "tier guard: 放行")
+	}
 	if msg := r.ListenRefusal(); msg != "" {
 		fmt.Fprintf(w, "listen guard: 拒绝启动（%s）\n", msg)
 	} else if r.InsecureListen && !IsLoopbackListen(r.Addr) && !r.AuthEnabled() {
@@ -320,6 +426,90 @@ func (r Resolved) AuthEnabled() bool {
 	return r.AuthMode == "on" || (r.AuthMode != "off" && r.StorageDSN != "")
 }
 
+// intDefault 返回 int 字段的默认值（0 表示零值合法、无默认替换）。
+func intDefault(f string) int {
+	switch f {
+	case "pgMaxOpen":
+		return DefaultPGMaxOpen
+	case "pgMaxIdle":
+		return DefaultPGMaxIdle
+	case "pgConnMaxIdleSec":
+		return DefaultPGConnMaxIdleSec
+	}
+	return 0
+}
+
+// 连接预算默认值（KTD12；与 store 层默认保持一致，两处同时修改时以测试对齐）。
+const (
+	DefaultPGMaxOpen        = 16
+	DefaultPGMaxIdle        = 8
+	DefaultPGConnMaxIdleSec = 300
+)
+
+// TierProblems 部署档语义校验：返回阻止启动的问题列表（空 = 放行）。
+// team 档 fail-closed：缺 DSN、未开认证直接拒绝；local 档不允许 DSN（KTD1）。
+// 执行隔离在 team 档是启动警告而非拒绝（本增量先可见、U3/U5 再硬化）。
+func (r Resolved) TierProblems() []string {
+	var msgs []string
+	switch r.Tier {
+	case "", "local", "team":
+	default:
+		msgs = append(msgs, fmt.Sprintf("未知部署档 %q（local | team）", r.Tier))
+	}
+	if r.Tier == "team" {
+		if r.StorageDSN == "" {
+			msgs = append(msgs, "team 档需要 PostgreSQL DSN（dsn / AGENT_SYSTEM_DSN）；不会回落 SQLite")
+		}
+		if !r.AuthEnabled() {
+			msgs = append(msgs, "team 档必须开启认证（auth=on）")
+		}
+	}
+	if r.Tier == "local" && r.StorageDSN != "" {
+		msgs = append(msgs, "local 档不使用 PostgreSQL DSN（改用 team 档，或去掉 dsn）")
+	}
+	if v := r.PGMaxOpen; v < 0 {
+		msgs = append(msgs, fmt.Sprintf("pgMaxOpen 不能为负（当前 %d）", v))
+	}
+	if v := r.PGMaxIdle; v < 0 {
+		msgs = append(msgs, fmt.Sprintf("pgMaxIdle 不能为负（当前 %d）", v))
+	}
+	if v := r.PGConnMaxIdleSec; v < 0 {
+		msgs = append(msgs, fmt.Sprintf("pgConnMaxIdleSec 不能为负（当前 %d）", v))
+	}
+	if v := r.PGConnMaxLifetimeSec; v < 0 {
+		msgs = append(msgs, fmt.Sprintf("pgConnMaxLifetimeSec 不能为负（当前 %d）", v))
+	}
+	if v := r.PGQueryTimeoutSec; v < 0 {
+		msgs = append(msgs, fmt.Sprintf("pgQueryTimeoutSec 不能为负（当前 %d）", v))
+	}
+	if idle, open := r.EffectivePGMaxIdle(), r.EffectivePGMaxOpen(); idle > open {
+		msgs = append(msgs, fmt.Sprintf("pgMaxIdle(%d) 大于 pgMaxOpen(%d)", idle, open))
+	}
+	return msgs
+}
+
+// EffectivePGMaxOpen/Idle/ConnMaxIdleSec 返回连接预算有效值（零值落默认，KTD12）。
+func (r Resolved) EffectivePGMaxOpen() int {
+	if r.PGMaxOpen > 0 {
+		return r.PGMaxOpen
+	}
+	return DefaultPGMaxOpen
+}
+
+func (r Resolved) EffectivePGMaxIdle() int {
+	if r.PGMaxIdle > 0 {
+		return r.PGMaxIdle
+	}
+	return DefaultPGMaxIdle
+}
+
+func (r Resolved) EffectivePGConnMaxIdleSec() int {
+	if r.PGConnMaxIdleSec > 0 {
+		return r.PGConnMaxIdleSec
+	}
+	return DefaultPGConnMaxIdleSec
+}
+
 // AllowedHosts 把逗号分隔的主机名拆开，忽略空段。
 func (r Resolved) AllowedHosts() []string {
 	var out []string

internal/config/config_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 )
 
@@ -220,3 +221,135 @@ func TestListenGuard(t *testing.T) {
 		t.Fatalf("auth on should pass: %+v %q", authed.Config, authed.ListenRefusal())
 	}
 }
+
+// ---- 协作存储方案 U2：部署档（KTD1）与连接预算（KTD12）----
+
+// team 档 fail-closed：缺 DSN、未开认证都拒绝；DSN+auth=on 放行。
+func TestTierTeamFailClosed(t *testing.T) {
+	r, err := Load("", noEnv, Config{Tier: "team"}, map[string]bool{"tier": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	msgs := r.TierProblems()
+	if len(msgs) != 2 {
+		t.Fatalf("team without dsn/auth should refuse with 2 problems, got %v", msgs)
+	}
+	joined := msgs[0] + msgs[1]
+	if !strings.Contains(joined, "PostgreSQL DSN") || !strings.Contains(joined, "认证") {
+		t.Fatalf("refusal messages wrong: %v", msgs)
+	}
+
+	// 补上 DSN（auto 认证随 DSN 开启）→ 放行。
+	r, err = Load("", noEnv, Config{Tier: "team", StorageDSN: "postgres://u:p@h/db"}, map[string]bool{"tier": true, "dsn": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msgs = r.TierProblems(); len(msgs) != 0 {
+		t.Fatalf("team with dsn should pass, got %v", msgs)
+	}
+
+	// 显式关认证仍然拒绝。
+	r, err = Load("", noEnv, Config{Tier: "team", StorageDSN: "postgres://u:p@h/db", AuthMode: "off"}, map[string]bool{"tier": true, "dsn": true, "auth": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msgs = r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "认证") {
+		t.Fatalf("team with auth=off should refuse, got %v", msgs)
+	}
+}
+
+// local 档不允许 DSN；未知档名拒绝。
+func TestTierLocalAndUnknown(t *testing.T) {
+	r, err := Load("", noEnv, Config{Tier: "local", StorageDSN: "postgres://u:p@h/db"}, map[string]bool{"tier": true, "dsn": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msgs := r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "local 档不使用") {
+		t.Fatalf("local with dsn should refuse, got %v", msgs)
+	}
+	r, err = Load("", noEnv, Config{Tier: "cluster"}, map[string]bool{"tier": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msgs := r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "未知部署档") {
+		t.Fatalf("unknown tier should refuse, got %v", msgs)
+	}
+	// auto（空）档维持现状语义：无 DSN 不拒绝。
+	r, err = Load("", noEnv, Config{}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msgs := r.TierProblems(); len(msgs) != 0 {
+		t.Fatalf("auto default should pass, got %v", msgs)
+	}
+}
+
+// 连接预算四源解析与环境变量整数校验；MaxIdle>MaxOpen 拒绝。
+func TestPGPoolConfigLayers(t *testing.T) {
+	env := func(k string) (string, bool) {
+		switch k {
+		case "AGENT_SYSTEM_PG_MAX_OPEN":
+			return "7", true
+		case "AGENT_SYSTEM_PG_MAX_IDLE":
+			return "9", true
+		}
+		return "", false
+	}
+	f := fileWith(t, `{"pgMaxOpen":5}`)
+	r, err := Load(f, env, Config{PGMaxOpen: 3, PGMaxIdle: 2}, map[string]bool{"pgMaxIdle": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	// 文件 5 被 env 7 覆盖（CLI 值 3 未显式传入不参与）；pgMaxIdle CLI 显式 2 胜。
+	if r.PGMaxOpen != 7 || r.SourceOf("pgMaxOpen") != SrcEnv {
+		t.Fatalf("pgMaxOpen = %d (%v), want env 7", r.PGMaxOpen, r.SourceOf("pgMaxOpen"))
+	}
+	if r.PGMaxIdle != 2 || r.SourceOf("pgMaxIdle") != SrcCLI {
+		t.Fatalf("pgMaxIdle = %d (%v), want CLI 2", r.PGMaxIdle, r.SourceOf("pgMaxIdle"))
+	}
+	if msgs := r.TierProblems(); len(msgs) != 0 {
+		t.Fatalf("2/2 pool valid, got %v", msgs)
+	}
+	// 有效值：零值落默认。
+	r, err = Load("", noEnv, Config{}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if r.EffectivePGMaxOpen() != DefaultPGMaxOpen || r.EffectivePGMaxIdle() != DefaultPGMaxIdle || r.EffectivePGConnMaxIdleSec() != DefaultPGConnMaxIdleSec {
+		t.Fatalf("effective defaults wrong: %+v", r.Config)
+	}
+	// idle > open 拒绝。
+	r, err = Load("", noEnv, Config{PGMaxOpen: 2, PGMaxIdle: 3}, map[string]bool{"pgMaxOpen": true, "pgMaxIdle": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msgs := r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "pgMaxIdle(3) 大于 pgMaxOpen(2)") {
+		t.Fatalf("idle>open should refuse, got %v", msgs)
+	}
+	// 非法整数环境变量报错定位到变量。
+	bad := func(k string) (string, bool) {
+		if k == "AGENT_SYSTEM_PG_MAX_OPEN" {
+			return "many", true
+		}
+		return "", false
+	}
+	if _, err := Load("", bad, Config{}, nil); err == nil || !strings.Contains(err.Error(), "AGENT_SYSTEM_PG_MAX_OPEN") {
+		t.Fatalf("invalid int env should fail with var name, got %v", err)
+	}
+}
+
+// explain 输出包含部署档与预算行，且 tier guard 结论可见（脱敏诊断）。
+func TestExplainTierAndPool(t *testing.T) {
+	r, err := Load("", noEnv, Config{Tier: "team"}, map[string]bool{"tier": true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var b bytes.Buffer
+	r.Explain(&b, "")
+	out := b.String()
+	for _, want := range []string{"tier", "pgMaxOpen", "pgMaxIdle", "pgConnMaxIdleSec", "pgConnMaxLifetimeSec", "pgQueryTimeoutSec", "tier guard: 拒绝启动"} {
+		if !strings.Contains(out, want) {
+			t.Fatalf("explain missing %q:\n%s", want, out)
+		}
+	}
+}

internal/pgbus/pgbus_test.go

@@ -29,6 +29,12 @@ func pgTestStore(t *testing.T) (*store.Store, string) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	// pg_trgm 预装 public（幂等），理由见 store 包 openTestStore 注释：
+	// 避免扩展落进用例 schema 后随 CASCADE 删除造成的并行竞态。
+	if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
+		admin.Close()
+		t.Fatalf("preinstall pg_trgm: %v", err)
+	}
 	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
 		admin.Close()
 		t.Fatal(err)
@@ -37,7 +43,7 @@ func pgTestStore(t *testing.T) (*store.Store, string) {
 		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
 		admin.Close()
 	})
-	full := dsn + "&search_path=" + schema
+	full := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public")
 	st, err := store.OpenPostgres(full, filepath.Join(t.TempDir(), "artifacts"))
 	if err != nil {
 		t.Fatal(err)

internal/runmgr/recovery_test.go

@@ -34,6 +34,12 @@ func openRecoveryStore(t *testing.T) *store.Store {
 	if err != nil {
 		t.Fatal(err)
 	}
+	// pg_trgm 预装 public（幂等）：避免 v4 扩展落进用例 schema 后随 CASCADE
+	// 删除造成的并行竞态（见 store 包 openTestStore 同名注释）。
+	if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
+		admin.Close()
+		t.Fatalf("preinstall pg_trgm: %v", err)
+	}
 	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
 		admin.Close()
 		t.Fatalf("create schema: %v", err)
@@ -42,7 +48,10 @@ func openRecoveryStore(t *testing.T) *store.Store {
 		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
 		admin.Close()
 	})
-	st, err := store.OpenPostgres(dsn+"&search_path="+schema, filepath.Join(t.TempDir(), "artifacts"))
+	// 共用 store 的 DSN 运行参数拼接（修复盲拼 "&search_path="：对无查询串的
+	// URL DSN 会把参数并进库名，连不上库——svc.sh 形态 DSN 的既有缺陷）。
+	schemaDSN := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public")
+	st, err := store.OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts"))
 	if err != nil {
 		t.Fatal(err)
 	}

internal/store/messages.go

@@ -40,7 +40,7 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro
 	m.CreatedAt = m.CreatedAt.UTC()
 	insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id)
 		SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?`
-	for attempt := 0; attempt < 8; attempt++ {
+	for attempt := 0; attempt < 32; attempt++ {
 		_, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID)
 		if err == nil {
 			s.TouchConversation(m.ConversationID)
@@ -56,8 +56,14 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro
 				return &DuplicateRequestError{Existing: existing}
 			}
 		}
-		// 并发同会话追加冲突（PG 多连接竞态）：短抖动退避后重算序号。
-		time.Sleep(time.Duration(attempt) * time.Millisecond)
+		// 并发同会话追加冲突（PG 多连接竞态）：平方退避（封顶 25ms）后重算
+		// 序号。真库 24 路并发实测：8 次 × 0–7ms 线性退避会耗尽重试（真实
+		// PG 门禁暴露的既有缺陷）；事务内会话锁的根治收口在 U3（KTD3/KTD7）。
+		backoff := attempt * attempt
+		if backoff > 25 {
+			backoff = 25
+		}
+		time.Sleep(time.Duration(backoff) * time.Millisecond)
 	}
 	return fmt.Errorf("store: message seq conflict retries exhausted (conversation %s)", m.ConversationID)
 }

internal/store/migrate.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"context"
 	"fmt"
 )
 
@@ -520,11 +521,51 @@ var v16Down = []string{
 }
 
 func (s *Store) migrate() error {
-	if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL, applied_at TEXT NOT NULL)`); err != nil {
+	if s.dialect.name() == "postgres" {
+		return s.migratePG()
+	}
+	// SQLite：_txlock=immediate（OpenSQLite DSN）保证迁移事务在 BEGIN 即取写锁，
+	// 跨进程迁移者按 busy_timeout 排队；后续版本读取发生在同一事务内，
+	// 前一个迁移者提交后本进程看到的已是完整账本（无部分版本可见）。
+	return s.migratePass()
+}
+
+// migratePG 以会话级咨询锁串行化并发迁移者（KTD8）：锁在专用连接上获取，
+// 等锁者在轮到自己后重新读取账本——前一个迁移者已完成时本轮为空操作。
+// 锁与迁移事务分离：迁移语句走连接池，锁只负责让 migrator 全局互斥。
+func (s *Store) migratePG() error {
+	ctx := context.Background()
+	conn, err := s.db.Conn(ctx)
+	if err != nil {
+		return fmt.Errorf("migrate lock: %w", err)
+	}
+	defer conn.Close()
+	const lockKey = "'agent-system:schema-migrate'"
+	if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock(hashtext(`+lockKey+`))`); err != nil {
+		return fmt.Errorf("migrate lock: %w", err)
+	}
+	defer conn.ExecContext(ctx, `SELECT pg_advisory_unlock(hashtext(`+lockKey+`))`)
+	return s.migratePass()
+}
+
+// migratePass 单事务执行全部待应用迁移：DDL 与版本登记同事务提交（KTD8），
+// 中途失败整体回滚——不存在"DDL 已生效而账本未记"的中间态；两步间崩溃
+// 的历史缺口（先提交 DDL 再单独 INSERT 版本）由此消除。
+func (s *Store) migratePass() error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return fmt.Errorf("migrate: %w", err)
+	}
+	defer tx.Rollback()
+	if _, err := tx.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL, applied_at TEXT NOT NULL)`); err != nil {
 		return fmt.Errorf("migrate: %w", err)
 	}
 	applied := map[int]bool{}
-	rows, err := s.db.Query(`SELECT version FROM schema_migrations`)
+	known := map[int]bool{}
+	for _, m := range migrations {
+		known[m.version] = true
+	}
+	rows, err := tx.Query(`SELECT version FROM schema_migrations`)
 	if err != nil {
 		return fmt.Errorf("migrate: %w", err)
 	}
@@ -534,6 +575,12 @@ func (s *Store) migrate() error {
 			rows.Close()
 			return fmt.Errorf("migrate: %w", err)
 		}
+		// 兼容门禁（KTD8 应用侧只核对兼容版本）：库里有本二进制不认识的版本
+		// 说明库由更新的构建迁移过——拒绝启动，不冒险在未知 schema 上服务。
+		if !known[v] {
+			rows.Close()
+			return fmt.Errorf("migrate: 数据库 schema 版本 v%d 超出本构建（最高 v%d）；拒绝启动，请升级二进制", v, latestMigrationVersion())
+		}
 		applied[v] = true
 	}
 	if err := rows.Err(); err != nil {
@@ -549,14 +596,28 @@ func (s *Store) migrate() error {
 		if len(stmts) == 0 {
 			return fmt.Errorf("migrate: v%d %s 缺少 %s 方言", m.version, m.name, s.dialect.name())
 		}
-		if err := s.execInTx(stmts); err != nil {
-			return fmt.Errorf("migrate v%d (%s): %w", m.version, m.name, err)
+		for _, q := range stmts {
+			if _, err := tx.Exec(s.q(q)); err != nil {
+				return fmt.Errorf("migrate v%d (%s): %w", m.version, m.name, err)
+			}
 		}
-		if _, err := s.db.Exec(s.q(`INSERT INTO schema_migrations(version,name,applied_at) VALUES(?,?,?)`), m.version, m.name, nowISO()); err != nil {
-			return fmt.Errorf("migrate: %w", err)
+		// 故障注入点：DDL 已执行、版本未登记——历史实现在此处崩溃会留下
+		// 「DDL 生效而账本缺失」的中间态；同事务设计下这里失败必须整体回滚。
+		if fault := s.takeMigrateFault(); fault != nil {
+			return fmt.Errorf("migrate v%d (%s) 注入故障: %w", m.version, m.name, fault)
+		}
+		if _, err := tx.Exec(s.q(`INSERT INTO schema_migrations(version,name,applied_at) VALUES(?,?,?)`), m.version, m.name, nowISO()); err != nil {
+			return fmt.Errorf("migrate v%d (%s) 版本登记: %w", m.version, m.name, err)
 		}
 	}
-	return nil
+	return tx.Commit()
+}
+
+func latestMigrationVersion() int {
+	if len(migrations) == 0 {
+		return 0
+	}
+	return migrations[len(migrations)-1].version
 }
 
 func (s *Store) execInTx(stmts []string) error {
@@ -592,6 +653,7 @@ func (s *Store) MigrationStatus() ([]int, error) {
 }
 
 // MigrateDown 回滚到指定版本（不含；运维入口，多租户数据折回限制见各 down 注释）。
+// 回滚语句与版本删除同事务（KTD8 原子性对称）。
 func (s *Store) MigrateDown(toVersion int) error {
 	applied, err := s.MigrationStatus()
 	if err != nil {
@@ -615,10 +677,21 @@ func (s *Store) MigrateDown(toVersion int) error {
 		if len(stmts) == 0 {
 			return fmt.Errorf("migrate-down: v%d 缺少 %s 回滚", v, s.dialect.name())
 		}
-		if err := s.execInTx(stmts); err != nil {
-			return fmt.Errorf("migrate-down v%d (%s): %w", v, m.name, err)
+		tx, err := s.db.Begin()
+		if err != nil {
+			return err
 		}
-		if _, err := s.db.Exec(s.q(`DELETE FROM schema_migrations WHERE version=?`), v); err != nil {
+		for _, q := range stmts {
+			if _, err := tx.Exec(s.q(q)); err != nil {
+				tx.Rollback()
+				return fmt.Errorf("migrate-down v%d (%s): %w", v, m.name, err)
+			}
+		}
+		if _, err := tx.Exec(s.q(`DELETE FROM schema_migrations WHERE version=?`), v); err != nil {
+			tx.Rollback()
+			return err
+		}
+		if err := tx.Commit(); err != nil {
 			return err
 		}
 	}

internal/store/migrate_atomic_test.go

@@ -0,0 +1,247 @@
+package store
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+)
+
+// 协作存储方案 U2（KTD8）：迁移原子性与并发正确性。
+//   - DDL 与版本登记同事务：注入「DDL 已执行、版本未登记」处故障 → 整体回滚；
+//   - 并发迁移者竞争（SQLite _txlock=immediate / PG 咨询锁）：账本恰好一套；
+//   - 旧 schema（仅部分版本）打开后自动升级到最新；
+//   - team 档 PG 不可达：打开失败、无任何回落（R2 fail-closed 在 store 层的基线）。
+
+// sqliteAt 用直接 SQL 预置一个「旧部署」库：只应用 v1 基线并登记账本。
+func sqliteAt(t *testing.T, path string) {
+	t.Helper()
+	st, err := OpenSQLite(path)
+	if err != nil {
+		t.Fatalf("seed open: %v", err)
+	}
+	if err := st.MigrateDown(1); //nolint:errcheck // 回滚到 v1 失败说明 down 链路破损，由下方状态断言兜住
+	err != nil {
+		t.Logf("seed migrate-down: %v", err)
+	}
+	if vers, err := st.MigrationStatus(); err != nil || len(vers) == 0 {
+		t.Fatalf("seed status: %v %v", vers, err)
+	} else if vers[len(vers)-1] != 1 {
+		t.Fatalf("seed should stop at v1, got %v", vers)
+	}
+	st.Close()
+}
+
+func openSQLiteAt(t *testing.T, path string) *Store {
+	t.Helper()
+	st, err := OpenSQLite(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return st
+}
+
+// 旧 schema 升级：只有 v1 账本的库重新打开后升级到最新（双档）。
+func TestMigrateUpgradesOldSchema(t *testing.T) {
+	if os.Getenv("AGENT_PG_TEST_DSN") != "" {
+		t.Run("postgres", func(t *testing.T) {
+			st := openTestStore(t)
+			if st.Dialect() != "postgres" {
+				t.Skip("非 PG 档")
+			}
+			if err := st.MigrateDown(1); err != nil {
+				t.Fatalf("migrate-down to v1: %v", err)
+			}
+			if err := st.migrate(); err != nil {
+				t.Fatalf("re-migrate: %v", err)
+			}
+			assertFullLedger(t, st)
+		})
+	}
+	t.Run("sqlite", func(t *testing.T) {
+		path := filepath.Join(t.TempDir(), "old.db")
+		sqliteAt(t, path)
+		st := openSQLiteAt(t, path)
+		assertFullLedger(t, st)
+	})
+}
+
+func assertFullLedger(t *testing.T, st *Store) {
+	t.Helper()
+	vers, err := st.MigrationStatus()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(vers) != len(migrations) {
+		t.Fatalf("ledger %v, want %d versions", vers, len(migrations))
+	}
+	for i, m := range migrations {
+		if vers[i] != m.version {
+			t.Fatalf("ledger[%d]=%d, want %d", i, vers[i], m.version)
+		}
+	}
+}
+
+// DDL 与账本同事务：在「DDL 已执行、版本未登记」处注入故障 → 全部回滚，
+// 不留下 v6 的表，也不留下 v6 的账本行（历史两步实现的中间态缺口）。
+func TestMigrateAtomicDDLAndLedger(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "atomic.db")
+	st := openSQLiteAt(t, path)
+	if err := st.MigrateDown(5); err != nil {
+		t.Fatalf("migrate-down to v5: %v", err)
+	}
+	vers, _ := st.MigrationStatus()
+	if vers[len(vers)-1] != 5 {
+		t.Fatalf("want v5 boundary, got %v", vers)
+	}
+	boom := errors.New("crash between DDL and ledger")
+	st.SetMigrateFault(boom)
+	err := st.migrate()
+	if err == nil || !strings.Contains(err.Error(), "注入故障") {
+		t.Fatalf("want injected fault, got %v", err)
+	}
+	// 账本不含 v6；v6 的表必须不存在（同事务回滚）。
+	if vers, _ = st.MigrationStatus(); vers[len(vers)-1] != 5 {
+		t.Fatalf("ledger advanced despite fault: %v", vers)
+	}
+	var n int
+	if err := st.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='activity_cards'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatal("activity_cards exists although the version insert faulted: DDL leaked outside the transaction")
+	}
+	// 恢复后重跑：升级完成。
+	if err := st.migrate(); err != nil {
+		t.Fatalf("retry migrate: %v", err)
+	}
+	assertFullLedger(t, st)
+}
+
+// 并发迁移者竞争同一 SQLite 文件：_txlock=immediate + busy_timeout 串行化，
+// 两边都成功，账本恰好一套（不因竞争出现重复或部分版本）。
+func TestMigrateRacingSQLite(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "race.db")
+	var wg sync.WaitGroup
+	errs := make([]error, 4)
+	for i := range errs {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			st, err := OpenSQLite(path)
+			if err != nil {
+				errs[i] = err
+				return
+			}
+			defer st.Close()
+			if vers, verr := st.MigrationStatus(); verr != nil {
+				errs[i] = verr
+			} else if len(vers) != len(migrations) {
+				errs[i] = fmt.Errorf("ledger=%v, want %d versions", vers, len(migrations))
+			}
+		}(i)
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err != nil {
+			t.Fatalf("migrator %d: %v", i, err)
+		}
+	}
+	st := openSQLiteAt(t, path)
+	assertFullLedger(t, st)
+}
+
+// 并发迁移者竞争同一 PG schema：会话级咨询锁串行化，账本恰好一套。
+func TestMigrateRacingPostgres(t *testing.T) {
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	if dsn == "" {
+		t.Skip("AGENT_PG_TEST_DSN 未设置：真实 PG 门禁由 CI 的必跑 job 覆盖")
+	}
+	schema := "t_race_" + strings.ToLower(newID("m")) + "_" + fmt.Sprint(os.Getpid())
+	admin, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
+		admin.Close()
+		t.Fatalf("create schema: %v", err)
+	}
+	t.Cleanup(func() {
+		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
+		admin.Close()
+	})
+	schemaDSN := withTestSearchPath(dsn, schema)
+	var wg sync.WaitGroup
+	errs := make([]error, 3)
+	for i := range errs {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			st, err := OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts"))
+			if err != nil {
+				errs[i] = err
+				return
+			}
+			defer st.Close()
+			if vers, verr := st.MigrationStatus(); verr != nil {
+				errs[i] = verr
+			} else if len(vers) != len(migrations) {
+				errs[i] = fmt.Errorf("ledger=%v, want %d versions", vers, len(migrations))
+			}
+		}(i)
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err != nil {
+			t.Fatalf("migrator %d: %v", i, err)
+		}
+	}
+	// 等锁者事后看到完整账本（不因竞争出现重复/部分版本）。
+	st, err := OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	assertFullLedger(t, st)
+}
+
+// PG 不可达：打开失败（team 档 R2 的 store 层基线——不存在任何回落路径）。
+func TestOpenPostgresUnreachableFails(t *testing.T) {
+	dsn := "postgres://u:p@127.0.0.1:1/none?sslmode=disable&connect_timeout=1"
+	if _, err := OpenPostgres(dsn, filepath.Join(t.TempDir(), "artifacts")); err == nil {
+		t.Fatal("unreachable PG must fail to open")
+	}
+}
+
+// 运行参数附着：URL 与键值两种 DSN 形式都正确追加 statement_timeout。
+func TestWithPGRuntimeParam(t *testing.T) {
+	url := withPGRuntimeParam("postgres://u:p@h:5432/db?sslmode=disable", "statement_timeout", "5000")
+	if want := "postgres://u:p@h:5432/db?sslmode=disable&statement_timeout=5000"; url != want {
+		t.Fatalf("url form: %q, want %q", url, want)
+	}
+	kv := withPGRuntimeParam("host=h port=5432 dbname=db", "statement_timeout", "5000")
+	if want := "host=h port=5432 dbname=db statement_timeout=5000"; kv != want {
+		t.Fatalf("kv form: %q, want %q", kv, want)
+	}
+	empty := withPGRuntimeParam("", "statement_timeout", "5000")
+	if want := "statement_timeout=5000"; empty != want {
+		t.Fatalf("empty dsn: %q, want %q", empty, want)
+	}
+}
+
+// 连接预算有效值：零值落默认，显式值生效。
+func TestPoolOptionsEffective(t *testing.T) {
+	got := PoolOptions{}.effective()
+	if got != DefaultPGPool {
+		t.Fatalf("zero pool should fall back to default, got %+v", got)
+	}
+	custom := PoolOptions{MaxOpenConn: 4, MaxIdleConn: 2, ConnMaxIdleTimeSec: 60, ConnMaxLifetimeSec: 3600, QueryTimeoutSec: 5}.effective()
+	if custom.MaxOpenConn != 4 || custom.MaxIdleConn != 2 || custom.ConnMaxIdleTimeSec != 60 || custom.ConnMaxLifetimeSec != 3600 || custom.QueryTimeoutSec != 5 {
+		t.Fatalf("custom pool mangled: %+v", custom)
+	}
+}

internal/store/store.go

@@ -16,6 +16,8 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -35,15 +37,46 @@ var ErrDuplicateRequest = errors.New("store: duplicate request id")
 // OwnerDefault 是单租户默认 owner（嵌入式档 / 未启用认证时的隐式身份）。
 const OwnerDefault = "local"
 
+// PoolOptions PG 连接预算（协作存储方案 KTD12）。零值走默认（DefaultPGPool）；
+// 预算应覆盖全部 API/Worker/LISTEN/migrator 连接，100 在线用户 ≠ 100 连接。
+type PoolOptions struct {
+	MaxOpenConn        int // 每进程最大打开连接；0=默认 16
+	MaxIdleConn        int // 每进程最大空闲连接；0=默认 8
+	ConnMaxIdleTimeSec int // 空闲连接回收时长；0=默认 300s
+	ConnMaxLifetimeSec int // 连接最长生命周期；0=不限
+	QueryTimeoutSec    int // PG statement_timeout（秒）；0=不限
+}
+
+// DefaultPGPool 与历史行为一致的默认预算（16/8/300s）。
+var DefaultPGPool = PoolOptions{MaxOpenConn: 16, MaxIdleConn: 8, ConnMaxIdleTimeSec: 300}
+
+func (p PoolOptions) effective() PoolOptions {
+	out := p
+	if out.MaxOpenConn <= 0 {
+		out.MaxOpenConn = DefaultPGPool.MaxOpenConn
+	}
+	if out.MaxIdleConn <= 0 {
+		out.MaxIdleConn = DefaultPGPool.MaxIdleConn
+	}
+	if out.ConnMaxIdleTimeSec <= 0 {
+		out.ConnMaxIdleTimeSec = DefaultPGPool.ConnMaxIdleTimeSec
+	}
+	return out
+}
+
 type Store struct {
 	db         *sql.DB
 	dialect    dialect
 	dataDir    string          // 工件对象目录（备份路径等诊断用）
 	blobs      blobstore.Store // 工件对象存储（本地档；多实例切共享卷/对象存储）
 	instanceID string          // 本进程标识；CreateRun 写入 runs.instance_id
+	pool       PoolOptions     // 生效的连接预算（诊断输出用）
 
 	settleMu    sync.Mutex
 	settleFault error // 非空时 SettleRun 在写入前失败，用于终态提交失败测试
+
+	migrateMu    sync.Mutex
+	migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败（原子性测试）
 }
 
 // OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。
@@ -51,43 +84,74 @@ func OpenSQLite(path string) (*Store, error) {
 	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
 		return nil, err
 	}
-	dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)", path)
+	// _txlock=immediate：写事务在 BEGIN 即取库写锁（而非首个写语句升级），
+	// 跨进程迁移/写路径依赖 busy_timeout 排队而不是中途 SQLITE_BUSY（KTD8）。
+	dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_txlock=immediate", path)
 	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, err
 	}
 	// SQLite 单写者：串行化连接，语句级原子 + 版本化迁移事务。
 	db.SetMaxOpenConns(1)
-	return finishOpen(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil)
+	return finishOpen(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil, PoolOptions{})
 }
 
-// OpenPostgres 打开 PostgreSQL 档存储并执行迁移。
+// OpenPostgres 打开 PostgreSQL 档存储并执行迁移（默认连接预算）。
 // DSN 例：postgres://user:pass@host:5432/agent_system?sslmode=disable
 // artifactsDir 为工件对象目录（多实例部署时指向共享卷；单实例默认本地）。
 func OpenPostgres(dsn, artifactsDir string) (*Store, error) {
-	db, err := sql.Open("pgx", dsn)
-	if err != nil {
-		return nil, err
-	}
-	db.SetMaxOpenConns(16)
-	db.SetMaxIdleConns(8)
-	db.SetConnMaxIdleTime(5 * time.Minute)
-	return finishOpen(db, postgresDialect{}, artifactsDir, nil)
+	return OpenPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, nil)
 }
 
 // OpenPostgresWithBlobs 是注入自定义对象存储的 PG 档入口（多实例共享卷/对象存储）。
 func OpenPostgresWithBlobs(dsn, artifactsDir string, blobs blobstore.Store) (*Store, error) {
+	return OpenPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, blobs)
+}
+
+// OpenPostgresWithOptions 按 KTD12 连接预算打开 PG 档；QueryTimeoutSec>0 时
+// 以 statement_timeout 运行参数附着到 DSN（对 URL 与键值两种形式均生效）。
+func OpenPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store) (*Store, error) {
+	pool = pool.effective()
+	if pool.QueryTimeoutSec > 0 {
+		dsn = withPGRuntimeParam(dsn, "statement_timeout", strconv.Itoa(pool.QueryTimeoutSec*1000))
+	}
 	db, err := sql.Open("pgx", dsn)
 	if err != nil {
 		return nil, err
 	}
-	db.SetMaxOpenConns(16)
-	db.SetMaxIdleConns(8)
-	db.SetConnMaxIdleTime(5 * time.Minute)
-	return finishOpen(db, postgresDialect{}, artifactsDir, blobs)
+	db.SetMaxOpenConns(pool.MaxOpenConn)
+	db.SetMaxIdleConns(pool.MaxIdleConn)
+	db.SetConnMaxIdleTime(time.Duration(pool.ConnMaxIdleTimeSec) * time.Second)
+	if pool.ConnMaxLifetimeSec > 0 {
+		db.SetConnMaxLifetime(time.Duration(pool.ConnMaxLifetimeSec) * time.Second)
+	}
+	return finishOpen(db, postgresDialect{}, artifactsDir, blobs, pool)
 }
 
-func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store) (*Store, error) {
+// withPGRuntimeParam 向 DSN 追加运行时参数（libpq 语义：URL 查询串或键值对）。
+// 不改写既有参数；值中不含分隔符时安全（statement_timeout/search_path 均满足）。
+func withPGRuntimeParam(dsn, key, value string) string {
+	return AppendPGRuntimeParam(dsn, key, value)
+}
+
+// AppendPGRuntimeParam 导出形态：测试与其他包（如 runmgr 的 schema 隔离）
+// 共用同一 DSN 拼接规则，避免各自盲拼 "&key=value" 造出非法库名。
+func AppendPGRuntimeParam(dsn, key, value string) string {
+	kv := key + "=" + value
+	if !strings.Contains(dsn, "://") {
+		if strings.TrimSpace(dsn) == "" {
+			return kv
+		}
+		return dsn + " " + kv
+	}
+	sep := "?"
+	if strings.Contains(dsn, "?") {
+		sep = "&"
+	}
+	return dsn + sep + kv
+}
+
+func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store, pool PoolOptions) (*Store, error) {
 	if blobs == nil {
 		var err error
 		blobs, err = blobstore.NewLocal(artifactsDir)
@@ -96,7 +160,7 @@ func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Stor
 			return nil, err
 		}
 	}
-	s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs}
+	s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs, pool: pool}
 	if err := s.migrate(); err != nil {
 		db.Close()
 		return nil, err
@@ -121,9 +185,28 @@ func (s *Store) SetSettleFault(err error) {
 	s.settleMu.Unlock()
 }
 
+// SetMigrateFault 让下一次 migratePass 在首个待应用迁移的「DDL 已执行、版本
+// 未登记」位置失败一次（nil 恢复）。测试专用：验证 DDL 与账本同事务回滚。
+func (s *Store) SetMigrateFault(err error) {
+	s.migrateMu.Lock()
+	s.migrateFault = err
+	s.migrateMu.Unlock()
+}
+
+func (s *Store) takeMigrateFault() error {
+	s.migrateMu.Lock()
+	defer s.migrateMu.Unlock()
+	err := s.migrateFault
+	s.migrateFault = nil
+	return err
+}
+
 // Dialect 返回当前存储档名（sqlite | postgres；诊断/健康检查用）。
 func (s *Store) Dialect() string { return s.dialect.name() }
 
+// Pool 返回生效的连接预算（启动诊断；脱敏，不含 DSN）。
+func (s *Store) Pool() PoolOptions { return s.pool }
+
 // DataDir 返回工件对象目录。
 func (s *Store) DataDir() string { return s.dataDir }
 

internal/store/store_test.go

@@ -31,6 +31,12 @@ func openTestStore(t *testing.T) *Store {
 	if err != nil {
 		t.Fatal(err)
 	}
+	// pg_trgm 预装到 public（幂等）：v4 迁移的 CREATE EXTENSION IF NOT EXISTS
+	// 会装进当时 search_path 的首个 schema（= 用例 schema），随 DROP SCHEMA
+	// CASCADE 被连带删除——并行用例间形成安装/删除竞态，gin_trgm_ops 随机
+	// 缺失。预装到 public 后 v4 的 CREATE EXTENSION 恒为 no-op，操作符经
+	// search_path 的 public 段稳定可见（与 svc.sh/CI 预装约定一致）。
+	ensurePGTrgm(t, admin)
 	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
 		admin.Close()
 		t.Fatalf("create schema: %v", err)
@@ -47,6 +53,19 @@ func openTestStore(t *testing.T) *Store {
 	return st
 }
 
+// ensurePGTrgm 在测试库 public 幂等安装 pg_trgm；并发首装的目录唯一冲突
+// 重试一次后放弃（下一个用例会再命中幂等分支）。
+func ensurePGTrgm(t *testing.T, admin *sql.DB) {
+	t.Helper()
+	for attempt := 0; attempt < 2; attempt++ {
+		if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err == nil {
+			return
+		} else if attempt == 1 {
+			t.Fatalf("preinstall pg_trgm: %v", err)
+		}
+	}
+}
+
 // withTestSearchPath 把用例 schema 放到 search_path 最前，并保留 public，
 // 以便看到已安装在 public 的 pg_trgm。svc.sh 给出的是无查询串的 URL；
 // 键值对 DSN（不含 ://）用空格追加。