feat(mcp,acp): Repository.WithTx + acp.Repository.InTx

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 <noreply@anthropic.com>

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 }
 

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

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} }

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 {

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 {