feat(app): add UnitOfWork and owner-bound repositories

Give each document command one postgres transaction and only let table owners unwrap the scope, so cross-owner writes can roll back together.

Co-authored-by: Cursor <cursoragent@cursor.com>

.doc/system-design/module-boundaries.yaml

@@ -54,13 +54,18 @@ imports:
     - storage
   db: []
   docduty: []
+  app: []
+  app/uow: []
   docgov:
+    - app/uow
     - docduty
     - docrevision
     - fsutil
   docrevision:
+    - app/uow
     - fsutil
   freedoc:
+    - app/uow
     - ctxmgr
     - docrevision
     - replyexport
@@ -116,6 +121,7 @@ imports:
   storage: []
   stream: []
   taskrun:
+    - app/uow
     - bizdev
     - ctxmgr
     - projectworkflow
@@ -179,6 +185,12 @@ event_consumers:
     - server
     - notification
 
+unwrap_owners:
+  - docgov
+  - docrevision
+  - taskrun
+  - freedoc
+
 exceptions:
   - kind: sql_write
     package: docduty

internal/app/uow/pg_uow.go

@@ -0,0 +1,172 @@
+package uow
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type pgScope struct {
+	ctx  context.Context
+	tx   pgx.Tx
+	done bool
+}
+
+func (s *pgScope) Context() context.Context { return s.ctx }
+
+// AsPG returns the bound pgx.Tx for owner repositories only.
+// Application / docflow code must not call this; archtest enforces unwrap owners.
+func AsPG(scope TxScope) (pgx.Tx, error) {
+	s, ok := scope.(*pgScope)
+	if !ok || s == nil || s.tx == nil {
+		return nil, fmt.Errorf("uow: scope is not a postgres transaction")
+	}
+	if s.done {
+		return nil, ErrClosedScope
+	}
+	return s.tx, nil
+}
+
+type PG struct {
+	Pool        *pgxpool.Pool
+	Policy      RetryPolicy
+	Now         func() time.Time
+	Sleep       func(time.Duration)
+	Rand        *rand.Rand
+	BeforeBegin func(context.Context) error
+	AfterBegin  func(TxScope) error
+	BeforeCommit func(TxScope) error
+}
+
+func NewPG(pool *pgxpool.Pool) *PG {
+	return &PG{Pool: pool, Policy: DefaultRetryPolicy()}
+}
+
+func (p *PG) Within(ctx context.Context, fn func(TxScope) error) error {
+	if p == nil || p.Pool == nil {
+		return fmt.Errorf("uow: postgres pool required")
+	}
+	if fn == nil {
+		return fmt.Errorf("uow: command required")
+	}
+	policy := p.Policy
+	if policy.MaxAttempts < 1 {
+		policy.MaxAttempts = DefaultMaxAttempts
+	}
+	if policy.LockTimeout <= 0 {
+		policy.LockTimeout = DefaultLockTimeout
+	}
+	var last error
+	for attempt := 1; attempt <= policy.MaxAttempts; attempt++ {
+		last = p.one(ctx, policy.LockTimeout, fn)
+		if last == nil {
+			return nil
+		}
+		if !IsSerialization(last) || IsNonRetryable(last) {
+			return last
+		}
+		if attempt == policy.MaxAttempts {
+			return last
+		}
+		p.backoff(attempt)
+	}
+	return last
+}
+
+func (p *PG) one(ctx context.Context, lockWait time.Duration, fn func(TxScope) error) error {
+	if p.BeforeBegin != nil {
+		if err := p.BeforeBegin(ctx); err != nil {
+			return err
+		}
+	}
+	if dl, ok := ctx.Deadline(); ok {
+		remain := time.Until(dl)
+		if remain > 0 && lockWait > remain {
+			lockWait = remain
+		}
+	}
+	tx, err := p.Pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	scope := &pgScope{ctx: ctx, tx: tx}
+	defer func() {
+		scope.done = true
+		_ = tx.Rollback(ctx)
+	}()
+	if _, err := tx.Exec(ctx, fmt.Sprintf("SET LOCAL lock_timeout = '%dms'", lockWait.Milliseconds())); err != nil {
+		return err
+	}
+	if p.AfterBegin != nil {
+		if err := p.AfterBegin(scope); err != nil {
+			return err
+		}
+	}
+	if err := fn(scope); err != nil {
+		return classifyLock(err)
+	}
+	if p.BeforeCommit != nil {
+		if err := p.BeforeCommit(scope); err != nil {
+			return err
+		}
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return classifyLock(err)
+	}
+	scope.done = true
+	return nil
+}
+
+func (p *PG) backoff(attempt int) {
+	sleep := p.Sleep
+	if sleep == nil {
+		sleep = time.Sleep
+	}
+	n := 10
+	if p.Rand != nil {
+		n = 5 + p.Rand.Intn(20)
+	} else {
+		n = 5 + rand.Intn(20)
+	}
+	sleep(time.Duration(n*attempt) * time.Millisecond)
+}
+
+func IsSerialization(err error) bool {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		return pgErr.Code == "40001" || pgErr.Code == "40P01"
+	}
+	return false
+}
+
+func IsNonRetryable(err error) bool {
+	return errors.Is(err, ErrVersionConflict) ||
+		errors.Is(err, ErrPermissionDenied) ||
+		errors.Is(err, ErrDigestConflict)
+}
+
+func classifyLock(err error) error {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		switch pgErr.Code {
+		case "55P03": // lock_not_available
+			return ErrBusy
+		case "57014": // query_canceled / statement timeout
+			if strings.Contains(strings.ToLower(pgErr.Message), "lock") {
+				return ErrBusy
+			}
+		}
+	}
+	msg := strings.ToLower(err.Error())
+	if strings.Contains(msg, "lock timeout") || strings.Contains(msg, "canceling statement due to lock timeout") {
+		return ErrBusy
+	}
+	return err
+}

internal/app/uow/pg_uow_test.go

@@ -0,0 +1,267 @@
+package uow_test
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/docrevision"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func testPool(t *testing.T) (*pgxpool.Pool, context.Context) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("postgresql fault injection is not part of -short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	return pool, ctx
+}
+
+func init() {
+	// ResolveMigrationsDir is cwd-relative; tests in this package start under internal/app/uow.
+	if _, err := os.Stat("deploy/postgres/goose"); err == nil {
+		return
+	}
+	for _, cand := range []string{"../../..", "../../../.."} {
+		if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
+			_ = os.Chdir(cand)
+			return
+		}
+	}
+}
+
+func seedProject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix string) int64 {
+	t.Helper()
+	var projectID int64
+	err := pool.QueryRow(ctx, `
+		INSERT INTO projects (code, name, status)
+		VALUES ($1, $2, 'active')
+		RETURNING id
+	`, "uow-"+suffix, "uow "+suffix).Scan(&projectID)
+	if err != nil {
+		t.Fatalf("insert project: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_change_events WHERE doc_id LIKE $1`, "uow_"+suffix+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_versions WHERE doc_id LIKE $1`, "uow_"+suffix+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_snapshots WHERE doc_id LIKE $1`, "uow_"+suffix+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id=$1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id=$1`, projectID)
+	})
+	return projectID
+}
+
+func shaBody(body string) string {
+	sum := sha256.Sum256([]byte(body))
+	return hex.EncodeToString(sum[:])
+}
+
+func writeOwners(ctx context.Context, scope uow.TxScope, projectID int64, docID, body string) error {
+	if err := (docgov.Repository{}).InsertRoot(ctx, scope, docgov.DocumentRecord{
+		DocID: docID, ProjectID: projectID, TaskKey: "intake", Title: docID,
+		ContentSHA256: shaBody(body), WorkDir: "ws/" + docID,
+	}); err != nil {
+		return err
+	}
+	sha := shaBody(body)
+	if err := (docrevision.Repository{}).InsertBlob(ctx, scope, sha, body); err != nil {
+		return err
+	}
+	sid, err := (docrevision.Repository{}).InsertSnapshot(ctx, scope, docrevision.Snapshot{
+		DocID: docID, ContentSHA256: sha, SourceKind: docrevision.SourceTemplate,
+		ActorRef: "template", CreatedAt: time.Now().UTC(),
+	})
+	if err != nil {
+		return err
+	}
+	vid, err := (docrevision.Repository{}).InsertVersion(ctx, scope, docrevision.Version{
+		DocID: docID, VersionNo: 1, Status: docrevision.VersionOpen, OriginKind: docrevision.OriginTemplate,
+		BaselineSnapshotID: sid, HeadSnapshotID: sid, OpenedAt: time.Now().UTC(),
+	})
+	if err != nil {
+		return err
+	}
+	_, err = (docrevision.Repository{}).InsertEvent(ctx, scope, docrevision.ChangeEvent{
+		DocID: docID, VersionID: vid, EventNo: 1, ToSnapshotID: sid,
+		ChangeKind: docrevision.ChangeImport, ActorKind: docrevision.ActorTemplate,
+		ActorRef: "template", IdempotencyKey: "uow:" + docID, CreatedAt: time.Now().UTC(),
+	})
+	return err
+}
+
+func countDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string) (records, snaps, events int) {
+	t.Helper()
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM document_records WHERE doc_id=$1`, docID).Scan(&records); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM document_snapshots WHERE doc_id=$1`, docID).Scan(&snaps); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM document_change_events WHERE doc_id=$1`, docID).Scan(&events); err != nil {
+		t.Fatal(err)
+	}
+	return
+}
+
+func TestSecondOwnerFailureRollsBackFirst(t *testing.T) {
+	pool, ctx := testPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	projectID := seedProject(t, ctx, pool, suffix)
+	docID := "uow_" + suffix + "_rollback"
+	work := uow.NewPG(pool)
+	err := work.Within(ctx, func(scope uow.TxScope) error {
+		if err := (docgov.Repository{}).InsertRoot(ctx, scope, docgov.DocumentRecord{
+			DocID: docID, ProjectID: projectID, Title: docID, TaskKey: "intake",
+		}); err != nil {
+			return err
+		}
+		return errors.New("ledger write failed")
+	})
+	if err == nil {
+		t.Fatal("expected failure")
+	}
+	r, s, e := countDoc(t, ctx, pool, docID)
+	if r != 0 || s != 0 || e != 0 {
+		t.Fatalf("partial commit records=%d snaps=%d events=%d", r, s, e)
+	}
+}
+
+func TestStagingFailureLeavesNoLedgerOrRoot(t *testing.T) {
+	pool, ctx := testPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	projectID := seedProject(t, ctx, pool, suffix)
+	docID := "uow_" + suffix + "_staging"
+	work := uow.NewPG(pool)
+	err := work.Within(ctx, func(scope uow.TxScope) error {
+		if err := writeOwners(ctx, scope, projectID, docID, "# staging\n"); err != nil {
+			return err
+		}
+		return fmt.Errorf("outbox/staging write failed")
+	})
+	if err == nil {
+		t.Fatal("expected staging failure")
+	}
+	r, s, e := countDoc(t, ctx, pool, docID)
+	if r != 0 || s != 0 || e != 0 {
+		t.Fatalf("staging failure leaked rows records=%d snaps=%d events=%d", r, s, e)
+	}
+}
+
+func TestCommitFailureHasNoSideEffects(t *testing.T) {
+	pool, ctx := testPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	projectID := seedProject(t, ctx, pool, suffix)
+	docID := "uow_" + suffix + "_commit"
+	work := uow.NewPG(pool)
+	work.BeforeCommit = func(uow.TxScope) error { return errors.New("commit failed") }
+	sideEffect := false
+	err := work.Within(ctx, func(scope uow.TxScope) error {
+		return writeOwners(ctx, scope, projectID, docID, "# commit\n")
+	})
+	if err == nil {
+		sideEffect = true
+	}
+	if sideEffect {
+		t.Fatal("caller must see commit error and skip post-commit work")
+	}
+	r, s, e := countDoc(t, ctx, pool, docID)
+	if r != 0 || s != 0 || e != 0 {
+		t.Fatalf("commit failure leaked rows records=%d snaps=%d events=%d", r, s, e)
+	}
+}
+
+func TestSerializationRetriesWholeTransactionOnce(t *testing.T) {
+	pool, ctx := testPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	projectID := seedProject(t, ctx, pool, suffix)
+	docID := "uow_" + suffix + "_ser"
+	work := uow.NewPG(pool)
+	work.Sleep = func(time.Duration) {}
+	attempts := 0
+	err := work.Within(ctx, func(scope uow.TxScope) error {
+		attempts++
+		body := fmt.Sprintf("# ser %d\n", attempts)
+		if err := writeOwners(ctx, scope, projectID, docID, body); err != nil {
+			return err
+		}
+		if attempts == 1 {
+			return &pgconn.PgError{Code: "40001", Message: "could not serialize access"}
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if attempts != 2 {
+		t.Fatalf("attempts=%d want 2", attempts)
+	}
+	r, s, e := countDoc(t, ctx, pool, docID)
+	if r != 1 || s != 1 || e != 1 {
+		t.Fatalf("want one business result records=%d snaps=%d events=%d", r, s, e)
+	}
+}
+
+func TestNonRetryableErrorsAreNotRetried(t *testing.T) {
+	pool, ctx := testPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	projectID := seedProject(t, ctx, pool, suffix)
+	work := uow.NewPG(pool)
+	work.Sleep = func(time.Duration) {}
+	for _, want := range []error{uow.ErrVersionConflict, uow.ErrPermissionDenied, uow.ErrDigestConflict} {
+		attempts := 0
+		err := work.Within(ctx, func(scope uow.TxScope) error {
+			attempts++
+			_ = (docgov.Repository{}).InsertRoot(ctx, scope, docgov.DocumentRecord{
+				DocID: "uow_" + suffix + "_noretry", ProjectID: projectID, Title: "x", TaskKey: "intake",
+			})
+			return want
+		})
+		if !errors.Is(err, want) {
+			t.Fatalf("err=%v want %v", err, want)
+		}
+		if attempts != 1 {
+			t.Fatalf("%v retried attempts=%d", want, attempts)
+		}
+	}
+	r, _, _ := countDoc(t, ctx, pool, "uow_"+suffix+"_noretry")
+	if r != 0 {
+		t.Fatalf("non-retryable write leaked record=%d", r)
+	}
+}
+
+func TestHappyPathPersistsBothOwners(t *testing.T) {
+	pool, ctx := testPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	projectID := seedProject(t, ctx, pool, suffix)
+	docID := "uow_" + suffix + "_ok"
+	work := uow.NewPG(pool)
+	if err := work.Within(ctx, func(scope uow.TxScope) error {
+		return writeOwners(ctx, scope, projectID, docID, "# ok\n")
+	}); err != nil {
+		t.Fatal(err)
+	}
+	r, s, e := countDoc(t, ctx, pool, docID)
+	if r != 1 || s != 1 || e != 1 {
+		t.Fatalf("happy path records=%d snaps=%d events=%d", r, s, e)
+	}
+}

internal/app/uow/uow.go

@@ -0,0 +1,41 @@
+package uow
+
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+var (
+	ErrVersionConflict  = errors.New("uow: version conflict")
+	ErrPermissionDenied = errors.New("uow: permission denied")
+	ErrDigestConflict   = errors.New("uow: digest conflict")
+	ErrBusy             = errors.New("uow: lock busy")
+	ErrClosedScope      = errors.New("uow: transaction scope is closed")
+	ErrNestedBegin      = errors.New("uow: repository must not begin a nested transaction")
+	ErrPrematureCommit  = errors.New("uow: repository must not commit the unit of work")
+)
+
+// TxScope is an opaque transaction handle. Application code cannot Exec SQL through it.
+type TxScope interface {
+	Context() context.Context
+}
+
+// UnitOfWork runs one command in a single database transaction.
+type UnitOfWork interface {
+	Within(ctx context.Context, fn func(TxScope) error) error
+}
+
+const (
+	DefaultLockTimeout = 2 * time.Second
+	DefaultMaxAttempts = 3
+)
+
+type RetryPolicy struct {
+	MaxAttempts int
+	LockTimeout time.Duration
+}
+
+func DefaultRetryPolicy() RetryPolicy {
+	return RetryPolicy{MaxAttempts: DefaultMaxAttempts, LockTimeout: DefaultLockTimeout}
+}

internal/app/uow/uow_test.go

@@ -0,0 +1,48 @@
+package uow
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+func TestTxScopeDoesNotExposeExec(t *testing.T) {
+	typ := reflect.TypeOf((*TxScope)(nil)).Elem()
+	for _, name := range []string{"Exec", "Begin", "Commit", "Rollback"} {
+		if _, ok := typ.MethodByName(name); ok {
+			t.Fatalf("TxScope must not expose %s", name)
+		}
+	}
+	pg := reflect.TypeOf(&pgScope{})
+	if _, ok := pg.MethodByName("Exec"); ok {
+		t.Fatal("pgScope must not expose Exec")
+	}
+}
+
+type fakeScope struct{}
+
+func (fakeScope) Context() context.Context { return context.Background() }
+
+func TestAsPGRejectsOpaqueScope(t *testing.T) {
+	if _, err := AsPG(fakeScope{}); err == nil {
+		t.Fatal("expected unwrap failure")
+	}
+}
+
+func TestSerializationAndNonRetryable(t *testing.T) {
+	if !IsSerialization(&pgconn.PgError{Code: "40001"}) || !IsSerialization(&pgconn.PgError{Code: "40P01"}) {
+		t.Fatal("serialization/deadlock must retry")
+	}
+	if IsSerialization(errors.New("plain")) {
+		t.Fatal("plain error must not retry")
+	}
+	if !IsNonRetryable(ErrVersionConflict) || !IsNonRetryable(ErrPermissionDenied) || !IsNonRetryable(ErrDigestConflict) {
+		t.Fatal("typed conflicts must not enter retry")
+	}
+	if IsNonRetryable(&pgconn.PgError{Code: "40001"}) {
+		t.Fatal("serialization is retryable")
+	}
+}

internal/archtest/boundaries.go

@@ -24,6 +24,7 @@ type Boundaries struct {
 	PathOwners      map[string][]string `yaml:"path_owners"`
 	EventProducers  map[string][]string `yaml:"event_producers"`
 	EventConsumers  map[string][]string `yaml:"event_consumers"`
+	UnwrapOwners    []string            `yaml:"unwrap_owners"`
 	Exceptions      []Exception         `yaml:"exceptions"`
 	BaselineCounts  map[string]int      `yaml:"baseline_counts"`
 }
@@ -93,6 +94,9 @@ func LoadBoundaries(path string) (*Boundaries, error) {
 	if out.EventConsumers == nil {
 		out.EventConsumers = map[string][]string{}
 	}
+	if out.UnwrapOwners == nil {
+		out.UnwrapOwners = []string{}
+	}
 	if out.BaselineCounts == nil {
 		out.BaselineCounts = map[string]int{}
 	}

internal/archtest/scan.go

@@ -139,6 +139,7 @@ func scanProductionGo(fset *token.FileSet, repoRoot, path string, b *Boundaries)
 	out = append(out, scanSQLWrites(string(src), pkg, rel, b)...)
 	out = append(out, scanPaths(file, pkg, rel, b)...)
 	out = append(out, scanEvents(file, string(src), pkg, rel, b)...)
+	out = append(out, scanUnwrap(string(src), pkg, rel, b)...)
 	return out, nil
 }
 
@@ -409,6 +410,22 @@ func scanEvents(file *ast.File, src, pkg, rel string, b *Boundaries) []Finding {
 	return out
 }
 
+func scanUnwrap(src, pkg, rel string, b *Boundaries) []Finding {
+	if !strings.Contains(src, "uow.AsPG") {
+		return nil
+	}
+	if allowedHas(b.UnwrapOwners, pkg) {
+		return nil
+	}
+	return []Finding{{
+		Kind:    KindImport,
+		Package: pkg,
+		File:    rel,
+		Symbol:  "uow.AsPG",
+		Detail:  "application layer must not unwrap TxScope",
+	}}
+}
+
 func dedupe(in []Finding) []Finding {
 	seen := map[string]bool{}
 	var out []Finding

internal/docgov/repository.go

@@ -0,0 +1,106 @@
+package docgov
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+// Repository is the document_records writer bound to a UnitOfWork scope.
+type Repository struct{}
+
+func (Repository) InsertRoot(ctx context.Context, scope uow.TxScope, rec DocumentRecord) error {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return err
+	}
+	if rec.CreatedAt.IsZero() {
+		rec.CreatedAt = time.Now().UTC()
+	}
+	if rec.UpdatedAt.IsZero() {
+		rec.UpdatedAt = rec.CreatedAt
+	}
+	if rec.Stage == "" {
+		rec.Stage = "V0.5_feasibility"
+	}
+	if rec.CurrentState == "" {
+		rec.CurrentState = StateDrafting
+	}
+	if rec.Title == "" {
+		rec.Title = rec.DocID
+	}
+	if rec.TaskKey == "" {
+		rec.TaskKey = "intake"
+	}
+	_, err = tx.Exec(ctx, `
+		INSERT INTO document_records (
+			doc_id, project_id, task_key, title, stage, version,
+			current_state, is_locked, is_template_only, content_sha256, work_dir,
+			author_name, reviewer_name, approver_name, created_at, updated_at
+		) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)`,
+		rec.DocID, rec.ProjectID, rec.TaskKey, rec.Title, rec.Stage, rec.Version,
+		string(rec.CurrentState), rec.IsLocked, rec.IsTemplateOnly, rec.ContentSHA256, rec.WorkDir,
+		rec.AuthorName, rec.ReviewerName, rec.ApproverName, rec.CreatedAt, rec.UpdatedAt)
+	if err != nil {
+		return fmt.Errorf("docgov: insert root: %w", err)
+	}
+	return nil
+}
+
+func (Repository) GetForUpdate(ctx context.Context, scope uow.TxScope, docID string) (*DocumentRecord, error) {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return nil, err
+	}
+	var rec DocumentRecord
+	var state string
+	err = tx.QueryRow(ctx, `
+		SELECT doc_id, project_id, task_key, title, stage, version, current_state,
+		       is_locked, is_template_only, content_sha256, work_dir
+		FROM document_records WHERE doc_id=$1 FOR UPDATE`, docID).Scan(
+		&rec.DocID, &rec.ProjectID, &rec.TaskKey, &rec.Title, &rec.Stage, &rec.Version, &state,
+		&rec.IsLocked, &rec.IsTemplateOnly, &rec.ContentSHA256, &rec.WorkDir)
+	if err != nil {
+		return nil, fmt.Errorf("docgov: lock document: %w", err)
+	}
+	rec.CurrentState = DocumentState(state)
+	return &rec, nil
+}
+
+func (Repository) UpdateState(ctx context.Context, scope uow.TxScope, rec DocumentRecord) error {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `
+		UPDATE document_records
+		SET current_state=$2, is_locked=$3, content_sha256=COALESCE(NULLIF($4,''), content_sha256),
+		    updated_at=$5
+		WHERE doc_id=$1`,
+		rec.DocID, string(rec.CurrentState), rec.IsLocked, rec.ContentSHA256, time.Now().UTC())
+	if err != nil {
+		return fmt.Errorf("docgov: update state: %w", err)
+	}
+	return nil
+}
+
+func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID string, versionID, snapshotID int64, revisionNo int, sha string) error {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `
+		UPDATE document_records
+		SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4,
+		    row_revision=COALESCE(row_revision,0)+1,
+		    content_sha256=COALESCE(NULLIF($5,''), content_sha256),
+		    updated_at=now()
+		WHERE doc_id=$1`,
+		docID, versionID, snapshotID, revisionNo, sha)
+	if err != nil {
+		return fmt.Errorf("docgov: project head: %w", err)
+	}
+	return nil
+}

internal/docrevision/repository.go

@@ -0,0 +1,52 @@
+package docrevision
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+// Repository is the ledger writer bound to a UnitOfWork scope.
+type Repository struct{}
+
+func (Repository) InsertBlob(ctx context.Context, scope uow.TxScope, sha, markdown string) error {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `
+		INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size, created_at)
+		VALUES ($1,$2,$3,$4)
+		ON CONFLICT (content_sha256) DO NOTHING`,
+		sha, markdown, int64(len(markdown)), time.Now().UTC())
+	if err != nil {
+		return fmt.Errorf("docrevision: insert blob: %w", err)
+	}
+	return nil
+}
+
+func (Repository) InsertSnapshot(ctx context.Context, scope uow.TxScope, sn Snapshot) (int64, error) {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return 0, err
+	}
+	return (&Service{}).insertSnapshotTX(ctx, tx, &sn)
+}
+
+func (Repository) InsertVersion(ctx context.Context, scope uow.TxScope, v Version) (int64, error) {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return 0, err
+	}
+	return (&Service{}).insertVersionTX(ctx, tx, &v)
+}
+
+func (Repository) InsertEvent(ctx context.Context, scope uow.TxScope, ev ChangeEvent) (int64, error) {
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return 0, err
+	}
+	return (&Service{}).insertEventTX(ctx, tx, &ev)
+}

internal/freedoc/repository.go

@@ -0,0 +1,18 @@
+package freedoc
+
+import (
+	"context"
+	"fmt"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+// Repository is a M3 stub. M1 freeform projection stays on the existing service path.
+type Repository struct{}
+
+func (Repository) ForbiddenInM1(ctx context.Context, scope uow.TxScope) error {
+	if _, err := uow.AsPG(scope); err != nil {
+		return err
+	}
+	return fmt.Errorf("freedoc: repository writes are not enabled until M3")
+}

internal/taskrun/repository.go

@@ -0,0 +1,18 @@
+package taskrun
+
+import (
+	"context"
+	"fmt"
+
+	"chat-agent-system/internal/app/uow"
+)
+
+// Repository is a M3 stub. M1 must not write task_runs through this port.
+type Repository struct{}
+
+func (Repository) ForbiddenInM1(ctx context.Context, scope uow.TxScope) error {
+	if _, err := uow.AsPG(scope); err != nil {
+		return err
+	}
+	return fmt.Errorf("taskrun: repository writes are not enabled until M3")
+}