From eb4796cf6158ba35cd807866698a65ed8e7b1c6c Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 11 Oct 2026 15:57:12 +0000 Subject: [PATCH] Add same-transaction outbox, S3-compatible blobstore, and real-process acceptance (U5 remainder) Why: plan U5 remainder (KTD5/KTD9 + exit criteria) - commands must commit atomically with business writes, artifacts need a shared object store for multi-instance, and the two-real-process acceptance matrix must run in CI. Evidence: full suite green on SQLite and real PostgreSQL 18.6 (35 packages each tier) - outbox tx test (rollback removes the command, commit makes both visible, idempotent re-enqueue), S3 store tests (SigV4-signed round trip, 404 sentinel, idempotent delete, presigned GET, env fallback), real-process acceptance (testbinary self-spawn of two serve processes with per-process identity and command consumers against one schema: same-conversation contention, cross-instance cancel command claimed exactly once, stale-owner settle fenced with run unchanged). Impact: EnqueueCommandInTx for in-transaction command outbox and AppendEvents now notifies in the same transaction (KTD6 front half); blobstore.S3 (SigV4, presign, prefix isolation) injected from BLOBSTORE_S3_* env when configured, local fallback otherwise; realproc_test provides the process-level multi-instance acceptance gate (PG-env gated, runs in CI test-pg job). Refs: proposals/2026-10-11-collaboration-storage-and-distribution.md U5 KTD5/KTD9, section 5 exit criteria; R4/R7/R10. --- internal/app/app.go | 12 +- internal/blobstore/object_store_test.go | 121 ++++++++++++ internal/blobstore/s3.go | 242 ++++++++++++++++++++++++ internal/runmgr/realproc_test.go | 198 +++++++++++++++++++ internal/store/execution.go | 18 ++ internal/store/messages.go | 9 + internal/store/outbox_tx_test.go | 100 ++++++++++ 7 files changed, 699 insertions(+), 1 deletion(-) create mode 100644 internal/blobstore/object_store_test.go create mode 100644 internal/blobstore/s3.go create mode 100644 internal/runmgr/realproc_test.go create mode 100644 internal/store/outbox_tx_test.go diff --git a/internal/app/app.go b/internal/app/app.go index db0d03f..7d1475b 100644 --- a/internal/app/app.go +++ b/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) diff --git a/internal/blobstore/object_store_test.go b/internal/blobstore/object_store_test.go new file mode 100644 index 0000000..1678ebb --- /dev/null +++ b/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) + } +} diff --git a/internal/blobstore/s3.go b/internal/blobstore/s3.go new file mode 100644 index 0000000..5d37e38 --- /dev/null +++ b/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) diff --git a/internal/runmgr/realproc_test.go b/internal/runmgr/realproc_test.go new file mode 100644 index 0000000..122b643 --- /dev/null +++ b/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) + } +} diff --git a/internal/store/execution.go b/internal/store/execution.go index 211e558..a2f761d 100644 --- a/internal/store/execution.go +++ b/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) diff --git a/internal/store/messages.go b/internal/store/messages.go index 7906e7f..36c2042 100644 --- a/internal/store/messages.go +++ b/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 } diff --git a/internal/store/outbox_tx_test.go b/internal/store/outbox_tx_test.go new file mode 100644 index 0000000..e42912e --- /dev/null +++ b/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) + } +}