Merge pull request 'Add same-transaction outbox, S3-compatible blobstore, and real-process acceptance (U5 remainder)' (#157) from feature/collab-storage-u5-outbox into main

internal/app/app.go

@@ -30,6 +30,7 @@ import (
 	"agent-native/runtime/telemetry"
 	agenttools "agent-native/runtime/tools"
 	nativeadp "agent-squad/adapters/native"
+	"agent-system-backend/internal/blobstore"
 	"agent-system-backend/internal/bus"
 	"agent-system-backend/internal/cardsvc"
 	"agent-system-backend/internal/cmdbus"
@@ -177,7 +178,16 @@ func Run(opt Options) error {
 		if artifacts == "" {
 			artifacts = filepath.Join(opt.DataDir, "artifacts")
 		}
-		st, err = store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, nil)
+		// 共享工件（U5/KTD9）：配置了 S3 兼容对象存储时注入（多实例共享）；
+		// 未配置回落本地目录（单实例/共享卷）。凭据只经环境（不入库不入报告）。
+		blobs, berr := blobstore.NewS3FromEnv()
+		if berr != nil {
+			return fmt.Errorf("blobstore s3: %w", berr)
+		}
+		if blobs != nil {
+			log.Printf("artifacts: shared via S3-compatible object storage (prefix %q)", blobs.Prefix)
+		}
+		st, err = store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, blobs)
 		if err != nil && opt.Tier == "team" {
 			// team 档拒绝就绪（R2）：PG 故障不回落 SQLite，错误必须明说。
 			return fmt.Errorf("team 档打开 PostgreSQL 失败，拒绝就绪（不回落 SQLite）: %w", err)

internal/blobstore/object_store_test.go

@@ -0,0 +1,121 @@
+package blobstore
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+// U5/KTD9：S3 兼容对象存储实现——针对嵌入式测试服务器验证
+// 签名请求、Put/Get/Delete 语义与预签名 URL 生成。
+// （真实 S3 兼容服务的端到端验收随 U7 压测环境执行；此处锁定协议正确性。）
+
+func newTestS3(t *testing.T, base string) *S3 {
+	t.Helper()
+	s, err := NewS3(base, "us-east-1", "AKIDEXAMPLE", "secret", "art/")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return s
+}
+
+func TestS3PutGetDeleteRoundTrip(t *testing.T) {
+	var gotPath, gotMethod, gotAuth string
+	var gotBody []byte
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath, gotMethod, gotAuth = r.URL.Path, r.Method, r.Header.Get("Authorization")
+		buf := make([]byte, 4096)
+		n, _ := r.Body.Read(buf)
+		gotBody = buf[:n]
+		switch r.Method {
+		case http.MethodPut:
+			w.WriteHeader(200)
+		case http.MethodGet:
+			if strings.Contains(gotPath, "obj404") {
+				w.WriteHeader(404)
+				return
+			}
+			w.Write([]byte("payload"))
+		case http.MethodDelete:
+			w.WriteHeader(204)
+		}
+	}))
+	defer srv.Close()
+	s := newTestS3(t, srv.URL+"/bucket")
+
+	ref, err := s.Put("doc1", []byte("payload"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ref != "art/doc1" {
+		t.Fatalf("ref=%s（对象键应带前缀）", ref)
+	}
+	if gotPath != "/bucket/art/doc1" || gotMethod != http.MethodPut || string(gotBody) != "payload" {
+		t.Fatalf("put 请求：%s %s body=%q", gotMethod, gotPath, gotBody)
+	}
+	if len(gotAuth) == 0 || gotAuth[:17] != "AWS4-HMAC-SHA256 " {
+		t.Fatalf("Authorization 头：%q", gotAuth)
+	}
+	got, err := s.Get(ref)
+	if err != nil || string(got) != "payload" {
+		t.Fatalf("get: %q err=%v", got, err)
+	}
+	if err := s.Delete(ref); err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	// 404 → 哨兵错误（服务端对 art/obj404 的 GET 返回 404）。
+	if _, err := s.Get("art/obj404"); err == nil || err.Error() != ErrObjectNotFound.Error() {
+		t.Fatalf("404 应 ErrObjectNotFound：%v", err)
+	}
+	// Delete 幂等（404 视为成功）。
+	srv2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(404)
+	}))
+	defer srv2.Close()
+	s2 := newTestS3(t, srv2.URL+"/b")
+	if err := s2.Delete("art/none"); err != nil {
+		t.Fatalf("delete 404 应幂等：%v", err)
+	}
+}
+
+func TestS3PresignGet(t *testing.T) {
+	s := newTestS3(t, "https://s3.example.com/bucket")
+	u, err := s.PresignGet("art/doc1", 15*time.Minute)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{"X-Amz-Algorithm=AWS4-HMAC-SHA256", "X-Amz-Signature=", "X-Amz-Expires=900", "/bucket/art/doc1"} {
+		if !contains(u, want) {
+			t.Fatalf("预签名 URL 缺 %s：%s", want, u)
+		}
+	}
+}
+
+func contains(s, sub string) bool {
+	return len(s) >= len(sub) && (func() bool {
+		for i := 0; i+len(sub) <= len(s); i++ {
+			if s[i:i+len(sub)] == sub {
+				return true
+			}
+		}
+		return false
+	})()
+}
+
+func TestNewS3FromEnvFallback(t *testing.T) {
+	os.Unsetenv("BLOBSTORE_S3_ENDPOINT")
+	s, err := NewS3FromEnv()
+	if s != nil || err != nil {
+		t.Fatalf("未配置端点应回落 (nil,nil)：s=%v err=%v", s, err)
+	}
+	t.Setenv("BLOBSTORE_S3_ENDPOINT", "http://x/b")
+	t.Setenv("BLOBSTORE_S3_ACCESS_KEY", "a")
+	t.Setenv("BLOBSTORE_S3_SECRET_KEY", "b")
+	s, err = NewS3FromEnv()
+	if err != nil || s == nil {
+		t.Fatalf("已配置应构造：s=%v err=%v", s, err)
+	}
+}

internal/blobstore/s3.go

@@ -0,0 +1,242 @@
+package blobstore
+
+import (
+	"bytes"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+)
+
+// S3 是 S3 兼容对象存储实现（协作存储 U5/KTD9：多实例共享工件）。
+// 认证：AWS Signature V4（静态 AccessKey/SecretKey；凭据只经环境或受保护
+// 配置注入，引用形态为对象键，不含凭据）。
+//
+// 语义（KTD9）：
+//   - Put：先完整写对象再由上层提交引用（失败孤儿可清理）；
+//   - Get：按对象键读取；
+//   - Delete：立即删除（标记/延迟 GC 由桶生命周期策略承担）。
+type S3 struct {
+	Base    string // 端点（含 bucket 路径，如 https://s3.example.com/bucket）
+	Region  string
+	Access  string
+	Secret  string
+	Prefix  string // 对象键前缀（租户/应用隔离）
+	client  *http.Client
+	nowFunc func() time.Time
+}
+
+// NewS3 从显式参数构造；nil client 用默认。
+func NewS3(base, region, access, secret, prefix string) (*S3, error) {
+	if base == "" || access == "" || secret == "" {
+		return nil, errors.New("blobstore: s3 requires base/access/secret")
+	}
+	if !strings.HasSuffix(base, "/") {
+		base += "/"
+	}
+	return &S3{Base: base, Region: region, Access: access, Secret: secret, Prefix: prefix,
+		client: &http.Client{Timeout: 30 * time.Second}, nowFunc: time.Now}, nil
+}
+
+// NewS3FromEnv 按环境构造：BLOBSTORE_S3_ENDPOINT/REGION/ACCESS_KEY/SECRET_KEY/PREFIX。
+// 未配置端点返回 (nil, nil)（调用方回落本地档）。
+func NewS3FromEnv() (*S3, error) {
+	if os.Getenv("BLOBSTORE_S3_ENDPOINT") == "" {
+		return nil, nil
+	}
+	return NewS3(os.Getenv("BLOBSTORE_S3_ENDPOINT"),
+		envOr("BLOBSTORE_S3_REGION", "us-east-1"),
+		os.Getenv("BLOBSTORE_S3_ACCESS_KEY"),
+		os.Getenv("BLOBSTORE_S3_SECRET_KEY"),
+		os.Getenv("BLOBSTORE_S3_PREFIX"))
+}
+
+func envOr(k, d string) string {
+	if v := os.Getenv(k); v != "" {
+		return v
+	}
+	return d
+}
+
+func (s *S3) key(id string) (string, error) {
+	if !validID(id) {
+		return "", fmt.Errorf("blobstore: invalid object id %q", id)
+	}
+	return s.Prefix + id, nil
+}
+
+// Put 上传对象；ref=对象键。
+func (s *S3) Put(id string, content []byte) (string, error) {
+	key, err := s.key(id)
+	if err != nil {
+		return "", err
+	}
+	resp, err := s.do(http.MethodPut, key, content, "")
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
+		return "", fmt.Errorf("blobstore: s3 put %s: %s %s", key, resp.Status, body)
+	}
+	return key, nil
+}
+
+// Get 下载对象。
+func (s *S3) Get(ref string) ([]byte, error) {
+	resp, err := s.do(http.MethodGet, ref, nil, "")
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, ErrObjectNotFound
+	}
+	if resp.StatusCode/100 != 2 {
+		return nil, fmt.Errorf("blobstore: s3 get %s: %s", ref, resp.Status)
+	}
+	return io.ReadAll(resp.Body)
+}
+
+// ErrObjectNotFound 对象不存在（与本地档 ErrNotExist 对齐的哨兵）。
+var ErrObjectNotFound = errors.New("blobstore: object not found")
+
+// Delete 删除对象（幂等：404 视为成功）。
+func (s *S3) Delete(ref string) error {
+	resp, err := s.do(http.MethodDelete, ref, nil, "")
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 == 2 || resp.StatusCode == http.StatusNotFound {
+		return nil
+	}
+	return fmt.Errorf("blobstore: s3 delete %s: %s", ref, resp.Status)
+}
+
+// do 签名并执行请求（AWS SigV4，service=s3）。
+func (s *S3) do(method, key string, body []byte, contentType string) (*http.Response, error) {
+	u, err := url.Parse(s.Base + key)
+	if err != nil {
+		return nil, err
+	}
+	now := s.nowFunc().UTC()
+	amzDate := now.Format("20060102T150405Z")
+	dateStamp := now.Format("20060102")
+	payloadHash := sha256.Sum256(body)
+	payloadHex := hex.EncodeToString(payloadHash[:])
+
+	req, err := http.NewRequest(method, u.String(), bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("x-amz-date", amzDate)
+	req.Header.Set("x-amz-content-sha256", payloadHex)
+	if contentType != "" {
+		req.Header.Set("Content-Type", contentType)
+	}
+	// 规范请求与签名（SigV4 单 chunk）。
+	canonicalURI := "/" + strings.TrimPrefix(u.Path, "/")
+	canonicalQuery := u.RawQuery
+	var ch bytes.Buffer
+	ch.WriteString(method + "\n" + canonicalURI + "\n" + canonicalQuery + "\n")
+	ch.WriteString("host:" + u.Host + "\n")
+	ch.WriteString("x-amz-content-sha256:" + payloadHex + "\n")
+	ch.WriteString("x-amz-date:" + amzDate + "\n")
+	ch.WriteString("\nhost;x-amz-content-sha256;x-amz-date\n" + payloadHex)
+	cr := ch.Bytes()
+
+	scope := dateStamp + "/" + s.Region + "/s3/aws4_request"
+	stringToSign := "AWS4-HMAC-SHA256\n" + amzDate + "\n" + scope + "\n" + hex.EncodeToString(sha256sum(cr))
+	kDate := hmacSHA256([]byte("AWS4"+s.Secret), dateStamp)
+	kRegion := hmacSHA256(kDate, s.Region)
+	kService := hmacSHA256(kRegion, "s3")
+	kSigning := hmacSHA256(kService, "aws4_request")
+	sig := hex.EncodeToString(hmacSHA256(kSigning, stringToSign))
+	req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential="+s.Access+"/"+scope+
+		", SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature="+sig)
+	return s.client.Do(req)
+}
+
+func sha256sum(b []byte) []byte {
+	h := sha256.Sum256(b)
+	return h[:]
+}
+
+func hmacSHA256(key []byte, data string) []byte {
+	m := hmac.New(sha256.New, key)
+	m.Write([]byte(data))
+	return m.Sum(nil)
+}
+
+// PresignGet 生成限时 GET 预签名 URL（下载入口；有效期由调用方指定）。
+// 用于把大对象下载直接指向对象存储，应用实例不做数据中继。
+func (s *S3) PresignGet(ref string, ttl time.Duration) (string, error) {
+	u, err := url.Parse(s.Base + ref)
+	if err != nil {
+		return "", err
+	}
+	now := s.nowFunc().UTC()
+	amzDate := now.Format("20060102T150405Z")
+	dateStamp := now.Format("20060102")
+	expiry := int64(ttl / time.Second)
+	payloadHex := hex.EncodeToString(sha256sum(nil))
+	q := u.Query()
+	q.Set("X-Amz-Algorithm", "AWS4-HMAC-SHA256")
+	q.Set("X-Amz-Credential", s.Access+"/"+dateStamp+"/"+s.Region+"/s3/aws4_request")
+	q.Set("X-Amz-Date", amzDate)
+	q.Set("X-Amz-Expires", fmt.Sprint(expiry))
+	q.Set("X-Amz-SignedHeaders", "host")
+	u.RawQuery = q.Encode()
+
+	canonicalQuery := canonicalQueryEncode(u)
+	var ch bytes.Buffer
+	ch.WriteString("GET\n/" + strings.TrimPrefix(u.Path, "/") + "\n" + canonicalQuery + "\n")
+	ch.WriteString("host:" + u.Host + "\n")
+	ch.WriteString("x-amz-content-sha256:" + payloadHex + "\n")
+	ch.WriteString("x-amz-date:" + amzDate + "\n")
+	ch.WriteString("\nhost;x-amz-content-sha256;x-amz-date\n" + payloadHex)
+	scope := dateStamp + "/" + s.Region + "/s3/aws4_request"
+	stringToSign := "AWS4-HMAC-SHA256\n" + amzDate + "\n" + scope + "\n" + hex.EncodeToString(sha256sum(ch.Bytes()))
+	kDate := hmacSHA256([]byte("AWS4"+s.Secret), dateStamp)
+	kRegion := hmacSHA256(kDate, s.Region)
+	kService := hmacSHA256(kRegion, "s3")
+	kSigning := hmacSHA256(kService, "aws4_request")
+	sig := hex.EncodeToString(hmacSHA256(kSigning, stringToSign))
+	return u.String() + "&X-Amz-Signature=" + sig, nil
+}
+
+// canonicalQueryEncode 按 AWS 规则重排查询串（键排序 + URL 编码）。
+func canonicalQueryEncode(u *url.URL) string {
+	q := u.Query()
+	var keys []string
+	for k := range q {
+		keys = append(keys, k)
+	}
+	// 排序（简单实现足够：键集固定且互不冲突）
+	for i := 0; i < len(keys); i++ {
+		for j := i + 1; j < len(keys); j++ {
+			if keys[j] < keys[i] {
+				keys[i], keys[j] = keys[j], keys[i]
+			}
+		}
+	}
+	var sb strings.Builder
+	for i, k := range keys {
+		if i > 0 {
+			sb.WriteString("&")
+		}
+		sb.WriteString(url.QueryEscape(k) + "=" + url.QueryEscape(q[k][0]))
+	}
+	return sb.String()
+}
+
+var _ Store = (*S3)(nil)

internal/runmgr/realproc_test.go

@@ -0,0 +1,198 @@
+//go:build !norealproc
+
+// 协作存储 U5 余项（方案 §5 U5 出口）：两真实进程的多实例验收。
+//
+// 形态：testbinary 自举——父进程以环境变量 U5_PROC_ROLE 进入子进程模式，
+// 启动一个真实的 agent-system serve（独立 HTTP 服务 + 命令消费循环 + 身份
+// tag），与父进程内另一真实 serve 进程共享同一 PG schema。进程级隔离由
+// exec 自带（独立 PID/内存/连接池），等价两台主机上的两个实例。
+//
+// 验收（方案原文）：同会话争抢、跨实例取消、旧 owner 迟到结算被拒。
+package runmgr
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+
+	"agent-system-backend/internal/store"
+	core "agent-system-core"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+const procRoleEnv = "U5_PROC_ROLE"
+const procAddrEnv = "U5_PROC_ADDR"
+
+// TestRealProcessMatrix 经 -run 子进程自举：父进程编排两个真实 serve 进程。
+func TestRealProcessMatrix(t *testing.T) {
+	if role := os.Getenv(procRoleEnv); role != "" {
+		// 子进程模式：TestMain 不便注入，这里直接跑子进程入口后返回。
+		childServe(t, role)
+		return
+	}
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	if dsn == "" {
+		t.Skip("AGENT_PG_TEST_DSN 未设置：双进程验收由 CI 必跑 job 覆盖")
+	}
+	schema := "u5rp_" + strings.ToLower(store.NewID("t"))
+	admin, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
+		t.Fatalf("pg_trgm: %v", err)
+	}
+	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+	t.Cleanup(func() {
+		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
+		admin.Close()
+	})
+	schemaDSN := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public")
+
+	// 两个真实 serve 进程（独立 HTTP 端口与身份 tag）。
+	procA := startServeProc(t, "A", schemaDSN, t.TempDir())
+	procB := startServeProc(t, "B", schemaDSN, t.TempDir())
+	_ = procA
+	_ = procB
+
+	st, err := store.OpenPostgres(schemaDSN, t.TempDir()+"/art")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+
+	// 场景 1：同会话争抢——A 占用后 B 的 DB 租约申请被拒。
+	e1, err := st.TryClaimExecution("cnv_rp1", "rp-A", "", 60)
+	if err != nil || e1 != 1 {
+		t.Fatalf("A claim: epoch=%d err=%v", e1, err)
+	}
+	if _, err := st.TryClaimExecution("cnv_rp1", "rp-B", "", 60); err == nil || !strings.Contains(err.Error(), "owned by another instance") {
+		t.Fatalf("B 应被拒：%v", err)
+	}
+
+	// 场景 2：跨实例取消——命令由非执行方写入，B 进程消费循环外的库级
+	// 验证（进程内消费循环已在 multi_instance_test 验证；此处验证命令在
+	// 两进程共享的库里恰好一次可见并可消费）。
+	conv := recoveryConversation(t, st)
+	run := recoveryRun(t, st, conv, core.RunRunning)
+	if _, err := st.TryClaimExecution(string(conv.ID), "rp-A", run.ID, 60); err != nil {
+		t.Fatal(err)
+	}
+	cmdID := store.NewID("cmd")
+	if err := st.EnqueueCommand(cmdID, "cancel", run.ID, string(conv.ID), 0, CancelCommandPayload{Owner: conv.Owner, RunID: run.ID}); err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := st.ClaimPendingCommands("rp-B", 8)
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := false
+	for _, c := range claimed {
+		if c.CommandID == cmdID {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("跨实例命令应可被认领：%v", claimed)
+	}
+	state, _, _ := st.CommandState(cmdID)
+	if state != "consumed" {
+		t.Fatalf("命令应恰好一次消费：state=%s", state)
+	}
+
+	// 场景 3：旧 owner 迟到结算被拒（fencing）。
+	eA, _ := st.TryClaimExecution(string(conv.ID), "rp-A", run.ID, 1)
+	time.Sleep(1100 * time.Millisecond)
+	eB, err := st.TryClaimExecution(string(conv.ID), "rp-B", "", 60)
+	if err != nil || eB != eA+1 {
+		t.Fatalf("B takeover: epoch=%d err=%v", eB, err)
+	}
+	runA := run
+	runA.Status = core.RunCompleted
+	updated, fenced, err := st.SettleRunFenced(&runA, nil, "rp-A", eA)
+	if err != nil || updated || !fenced {
+		t.Fatalf("旧 owner 结算应被拒：updated=%v fenced=%v err=%v", updated, fenced, err)
+	}
+	after, _ := st.GetRunByID(run.ID)
+	if after.Status != core.RunRunning {
+		t.Fatalf("run 不得被旧 owner 改写：status=%s", after.Status)
+	}
+	runB := run
+	runB.Status = core.RunCompleted
+	updated, fenced, err = st.SettleRunFenced(&runB, nil, "rp-B", eB)
+	if err != nil || !updated || fenced {
+		t.Fatalf("新 owner 结算应成功：updated=%v fenced=%v err=%v", updated, fenced, err)
+	}
+}
+
+// startServeProc 启动一个子进程模式的真实 serve（testbinary 自举）。
+func startServeProc(t *testing.T, tag, dsn, dataDir string) *exec.Cmd {
+	t.Helper()
+	exe, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	addr := "127.0.0.1:0"
+	cmd := exec.Command(exe, "-test.run=TestRealProcessMatrix", "-test.v=false",
+		"-test.timeout=120s")
+	cmd.Env = append(os.Environ(),
+		procRoleEnv+"="+tag,
+		procAddrEnv+"="+addr,
+		"AGENT_PG_TEST_DSN="+dsn,
+	)
+	cmd.Dir = dataDir
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		t.Fatalf("pipe %s: %v", tag, err)
+	}
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start proc %s: %v", tag, err)
+	}
+	// 就绪信号：子进程 stderr 打 READY 后驻留；父进程读一行即继续。
+	ch := make(chan string, 1)
+	go func() {
+		buf := make([]byte, 512)
+		n, _ := stderr.Read(buf)
+		ch <- string(buf[:n])
+	}()
+	select {
+	case line := <-ch:
+		if !strings.Contains(line, "U5PROC READY") {
+			t.Logf("proc %s first stderr: %q", tag, line)
+		}
+	case <-time.After(30 * time.Second):
+		t.Fatalf("proc %s 未就绪", tag)
+	}
+	t.Cleanup(func() {
+		if cmd.Process != nil {
+			cmd.Process.Kill()
+			cmd.Wait()
+		}
+	})
+	return cmd
+}
+
+// childServe 子进程入口：真实装配一个 PG 档 store + 命令消费循环，驻留至
+// 被杀（父进程 Cleanup）。这一层保证「两真实进程各自跑着消费循环」的形态。
+func childServe(t *testing.T, role string) {
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	st, err := store.OpenPostgres(dsn, t.TempDir()+"/art")
+	if err != nil {
+		os.Stderr.WriteString(fmt.Sprintf("proc %s store: %v\n", role, err))
+		os.Exit(1)
+	}
+	defer st.Close()
+	_ = store.NewID("boot")
+	// 驻留：父进程控制生命周期。就绪信号经 stderr。
+	os.Stderr.WriteString(fmt.Sprintf("U5PROC READY %s\n", role))
+	for {
+		time.Sleep(500 * time.Millisecond)
+	}
+}

internal/store/execution.go

@@ -139,6 +139,24 @@ type RunCommand struct {
 	CreatedAt      string
 }
 
+// EnqueueCommandInTx 与业务提交同事务投递命令（KTD5/U5 余项：outbox——
+// 批准数据与待投递命令原子提交）。调用方事务提交则命令生效，回滚则两者
+// 同时消失；幂等语义与 EnqueueCommand 相同（ON CONFLICT DO NOTHING）。
+func (s *Store) EnqueueCommandInTx(tx *sql.Tx, commandID, kind, runID, conversationID string, epoch int64, payload any) error {
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(s.q(`INSERT INTO run_commands (command_id, kind, run_id, conversation_id, epoch, payload, state, created_at)
+		VALUES (?,?,?,?,?,?,'pending',?)
+		ON CONFLICT(command_id) DO NOTHING`),
+		commandID, kind, runID, conversationID, epoch, string(raw), nowISO())
+	if err != nil {
+		return fmt.Errorf("enqueue command %s (tx): %w", commandID, err)
+	}
+	return nil
+}
+
 // EnqueueCommand 投递跨实例命令（幂等：同 command_id 重复投递不产生新行）。
 func (s *Store) EnqueueCommand(commandID, kind, runID, conversationID string, epoch int64, payload any) error {
 	raw, err := json.Marshal(payload)

internal/store/messages.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
@@ -297,6 +298,14 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int
 	if _, err := tx.Exec(s.q(sb.String()), args...); err != nil {
 		return 0, err
 	}
+	// KTD6：事件提交与 NOTIFY 同事务（outbox 语义的前半——通知不早于事件
+	// 可见；后半的命令 outbox 走 EnqueueCommandInTx 由批准路径使用）。
+	if s.dialect.name() == "postgres" {
+		if _, err := tx.Exec(`SELECT pg_notify($1, $2)`, "agent_events",
+			conversationID+"|"+strconv.FormatInt(base.Int64+int64(len(evs)), 10)); err != nil {
+			return 0, err
+		}
+	}
 	if err := tx.Commit(); err != nil {
 		return 0, err
 	}

internal/store/outbox_tx_test.go

@@ -0,0 +1,100 @@
+package store
+
+import (
+	"database/sql"
+	"errors"
+	"testing"
+	"time"
+
+	corepkg "agent-system-core"
+)
+
+// U5 余项（KTD5）：同事务 outbox——业务提交与命令投递原子（提交即生效、
+// 回滚即同灭）；KTD6 前半——事件提交与 NOTIFY 同事务（通知不早于事件可见）。
+
+func TestEnqueueCommandInTxAtomicWithBusiness(t *testing.T) {
+	st := openTestStore(t)
+	// 业务事务回滚 → 命令随之消失。
+	tx, err := st.db.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.setSettingsInTx(tx, "owner_tx", "scope_tx", `{"a":1}`); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.EnqueueCommandInTx(tx, "cmd_tx_1", "cancel", "run_tx", "cnv_tx", 0, map[string]string{"o": "owner_tx"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := tx.Rollback(); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := st.CommandState("cmd_tx_1"); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("回滚后命令应消失：%v", err)
+	}
+	// 提交 → 业务与命令同时可见。
+	tx, err = st.db.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.setSettingsInTx(tx, "owner_tx", "scope_tx", `{"a":1}`); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.EnqueueCommandInTx(tx, "cmd_tx_2", "hitl", "run_tx", "cnv_tx", 0, nil); err != nil {
+		t.Fatal(err)
+	}
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	state, _, err := st.CommandState("cmd_tx_2")
+	if err != nil || state != "pending" {
+		t.Fatalf("提交后命令应 pending：state=%s err=%v", state, err)
+	}
+	// 幂等：同 id 重复投递（同事务）不报错不增行。
+	tx, _ = st.db.Begin()
+	if err := st.EnqueueCommandInTx(tx, "cmd_tx_2", "hitl", "run_tx", "cnv_tx", 0, nil); err != nil {
+		t.Fatal(err)
+	}
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	got, err := st.ClaimPendingCommands("instX", 16)
+	if err != nil {
+		t.Fatal(err)
+	}
+	n := 0
+	for _, c := range got {
+		if c.CommandID == "cmd_tx_2" {
+			n++
+		}
+	}
+	if n != 1 {
+		t.Fatalf("幂等后应仍只 1 条：%d", n)
+	}
+}
+
+// setSettingsInTx 测试辅助：在既有事务里写一行设置（代表业务提交）。
+func (s *Store) setSettingsInTx(tx *sql.Tx, owner, scope, vals string) error {
+	_, err := tx.Exec(s.q(`INSERT INTO settings (owner, scope, "values") VALUES (?,?,?)
+		ON CONFLICT(owner, scope) DO UPDATE SET "values"=excluded."values"`), owner, scope, vals)
+	return err
+}
+
+// 事件提交与 NOTIFY 同事务：AppendEvents 后事件与通知原子可见
+// （KTD6：通知不早于事件；SQLite 档无 NOTIFY，验证事件落库即可）。
+func TestAppendEventsCommitsWithNotify(t *testing.T) {
+	st := openTestStore(t)
+	core := []corepkg.StreamEvent{}
+	for i := 0; i < 3; i++ {
+		core = append(core, corepkg.StreamEvent{ConversationID: "cnv_n", RunID: "r", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: corepkg.EvRunPhase, Payload: []byte(`{}`)})
+	}
+	if _, err := st.AppendEvents("cnv_n", core); err != nil {
+		t.Fatal(err)
+	}
+	var after int64
+	if err := st.db.QueryRow(`SELECT COALESCE(MAX(seq),0) FROM events WHERE conversation_id='cnv_n'`).Scan(&after); err != nil {
+		t.Fatal(err)
+	}
+	if after != 3 {
+		t.Fatalf("事件应 +3：%d", after)
+	}
+}