From be8d5e3a1b9de83b48d5e1e51a8d01debeabfb22 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Fri, 8 May 2026 12:40:54 +0800 Subject: [PATCH] feat(mcp,acp): Repository.WithTx + acp.Repository.InTx MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-module transactional support for spec §6.1: ACP CreateSession must atomically insert acp_sessions row + mcp_tokens row (FK NOT NULL). - mcp.Repository.WithTx(tx) -> Repository (tx-bound, no pool) - acp.Repository.InTx(ctx, fn) -> error (pgx.BeginFunc wrapper) - acp.Repository.WithTx(tx) -> Repository (tx-bound) - Update fakeRepos in test files to satisfy new interface methods Co-Authored-By: Claude Opus 4.6 --- internal/acp/agentkind_service_test.go | 9 +++++++++ internal/acp/relay_test.go | 9 +++++++++ internal/acp/repository.go | 20 ++++++++++++++++++++ internal/mcp/repository.go | 10 ++++++++++ internal/mcp/token_service_test.go | 5 +++++ 5 files changed, 53 insertions(+) diff --git a/internal/acp/agentkind_service_test.go b/internal/acp/agentkind_service_test.go index 1bfac4b..f12d6c3 100644 --- a/internal/acp/agentkind_service_test.go +++ b/internal/acp/agentkind_service_test.go @@ -6,6 +6,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -124,6 +125,14 @@ func (f *fakeAgentKindRepo) PurgeEventsBefore(context.Context, time.Time) (int64 panic("n/a") } +func (f *fakeAgentKindRepo) InTx(context.Context, func(context.Context, pgx.Tx) error) error { + panic("n/a") +} + +func (f *fakeAgentKindRepo) WithTx(pgx.Tx) acp.Repository { + return f +} + // agentKindRecordingRecorder：accumulate audit entries type agentKindRecordingRecorder struct{ entries []audit.Entry } diff --git a/internal/acp/relay_test.go b/internal/acp/relay_test.go index 8bd1f04..9c573f0 100644 --- a/internal/acp/relay_test.go +++ b/internal/acp/relay_test.go @@ -18,6 +18,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -112,6 +113,14 @@ func (f *fakeEventRepo) PurgeEventsBefore(context.Context, time.Time) (int64, er panic("n/a") } +func (f *fakeEventRepo) InTx(context.Context, func(context.Context, pgx.Tx) error) error { + panic("n/a") +} + +func (f *fakeEventRepo) WithTx(pgx.Tx) acp.Repository { + return f +} + // relayTestRig wires up a Relay with two io.Pipe pairs that simulate the agent // subprocess. Returns: // - r: the relay under test diff --git a/internal/acp/repository.go b/internal/acp/repository.go index ecb069b..8f117b8 100644 --- a/internal/acp/repository.go +++ b/internal/acp/repository.go @@ -47,6 +47,9 @@ type Repository interface { InsertEvent(ctx context.Context, e *Event) (*Event, error) ListEventsSince(ctx context.Context, sessionID uuid.UUID, sinceID int64, limit int32) ([]*Event, error) PurgeEventsBefore(ctx context.Context, before time.Time) (int64, error) + + InTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error + WithTx(tx pgx.Tx) Repository } // IsUniqueViolation reports whether err is a PG unique constraint violation @@ -74,6 +77,23 @@ func NewPostgresRepository(pool *pgxpool.Pool) Repository { return &pgRepo{pool: pool, q: acpsqlc.New(pool)} } +// InTx runs fn in a single transaction. Commits on nil error; rolls back +// otherwise. Used by SessionService.Create to atomically insert acp_sessions +// + mcp_tokens (spec §6.1). +func (r *pgRepo) InTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error { + if r.pool == nil { + return errs.New(errs.CodeInternal, "acp.Repository.InTx: nil pool (tx-bound instance)") + } + return pgx.BeginFunc(ctx, r.pool, func(tx pgx.Tx) error { + return fn(ctx, tx) + }) +} + +// WithTx 返回绑定到指定事务的 Repository 副本。 +func (r *pgRepo) WithTx(tx pgx.Tx) Repository { + return &pgRepo{pool: nil, q: r.q.WithTx(tx)} +} + // ===== 类型转换 helper ===== func toPgUUID(u uuid.UUID) pgtype.UUID { return pgtype.UUID{Bytes: u, Valid: true} } diff --git a/internal/mcp/repository.go b/internal/mcp/repository.go index dd52634..0a4b6a4 100644 --- a/internal/mcp/repository.go +++ b/internal/mcp/repository.go @@ -32,6 +32,9 @@ type Repository interface { // 测试辅助：直接物理删（生产不暴露） DeleteByID(ctx context.Context, id uuid.UUID) error + + // WithTx 返回绑定到指定事务的 Repository。详见 pgRepo.WithTx 文档。 + WithTx(tx pgx.Tx) Repository } // NewPostgresRepository 用 pgxpool 构造 Repository。 @@ -44,6 +47,13 @@ type pgRepo struct { q *mcpsqlc.Queries } +// WithTx 返回一个绑定到指定事务的 Repository 副本。Caller 在外部用 pgx.BeginFunc +// 或 acp.Repository.InTx 取得 tx；这里只负责生成一个把所有 sqlc 调用切到 tx 的轻 +// 实例。tx-bound 实例不持有 pool（防止误用同一实例混 pool/tx 两路）。 +func (r *pgRepo) WithTx(tx pgx.Tx) Repository { + return &pgRepo{pool: nil, q: r.q.WithTx(tx)} +} + // ===== 类型转换 helper ===== func toPgUUID(u uuid.UUID) pgtype.UUID { diff --git a/internal/mcp/token_service_test.go b/internal/mcp/token_service_test.go index 0105b2a..334e12b 100644 --- a/internal/mcp/token_service_test.go +++ b/internal/mcp/token_service_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -122,6 +123,10 @@ func (r *fakeRepo) DeleteByID(ctx context.Context, id uuid.UUID) error { return nil } +func (r *fakeRepo) WithTx(tx pgx.Tx) mcp.Repository { + return r +} + // ===== fake audit ===== type fakeAudit struct {