feat(docflow): own document commands in one UnitOfWork

Lock the document row and re-authorize duty before edit, agent commit, approve, or reject share a single transaction. Characterization tests pin ETag races, one legal approve/reject terminal, and same-key digest conflicts.

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

.doc/ops/short-term/2026-09-14-document-reliability-pilot.md

@@ -35,7 +35,7 @@ mutex: docgov
 1. [x] 从最新 `dev`（`a8f9451`）创建实现分支，登记 CURRENT 并确认 `docgov` mutex 已释放。
 2. [x] U1：四类规则 arch-check + archtest + 可重放基线（新违规 block）。
 3. [x] 实现 U2，验证任意属主写入/commit 故障整体回滚。
-4. [ ] 实现 U3，删除提交后 outbox 正确性回调。
+4. [x] 实现 U3，删除提交后 outbox 正确性回调。
 5. [ ] 实现 U4，保证 GET 零写入和 projection 缺失不再伪装成 404。
 6. [ ] 执行多实例并发、PG、真实 PI、真实浏览器和 restart/rebuild 验收。
 7. [ ] 写 test report，更新验收清单、FEATURE-MATRIX、CURRENT 和 gap-matrix。

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

@@ -55,6 +55,10 @@ imports:
   db: []
   docduty: []
   app: []
+  app/docflow:
+    - app/uow
+    - docgov
+    - docrevision
   app/uow: []
   docgov:
     - app/uow
@@ -89,6 +93,8 @@ imports:
   replyexport: []
   server:
     - agent
+    - app/docflow
+    - app/uow
     - auth
     - authz
     - bizdev
@@ -171,10 +177,12 @@ path_owners:
 event_producers:
   outbox.notify:
     - docrevision
+    - app/docflow
   document.agent_candidate_ready:
     - docrevision
   document.agent_version_committed:
     - docrevision
+    - app/docflow
   document.downstream_impact_detected:
     - docrevision
   document.change_report_degraded:

internal/app/docflow/commands.go

@@ -0,0 +1,171 @@
+package docflow
+
+import (
+	"context"
+	"fmt"
+
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/docrevision"
+)
+
+type Action string
+
+const (
+	ActionEdit    Action = "document.edit"
+	ActionApprove Action = "document.approve"
+	ActionReject  Action = "document.reject"
+	ActionCommit  Action = "document.agent_commit"
+)
+
+type Principal struct {
+	UserID int64
+	Name   string
+	Role   docgov.UserRole
+}
+
+type HumanEditCommand struct {
+	DocID   string
+	Request docrevision.HumanRevisionRequest
+}
+
+type AgentCommitCommand struct {
+	DocID       string
+	CandidateID int64
+	Decision    docrevision.CommitAgentDecision
+}
+
+type DecisionCommand struct {
+	DocID        string
+	Comment      string
+	RejectReason string
+	RollbackTo   docgov.DocumentState
+}
+
+type Commands struct {
+	UoW  uow.UnitOfWork
+	Gov  *docgov.GovernanceEngine
+	Rev  *docrevision.Service
+	Auth Authorizer
+}
+
+type Authorizer interface {
+	Authorize(ctx context.Context, p Principal, rec *docgov.DocumentRecord, action Action) error
+}
+
+type dutyAuthorizer struct{}
+
+func DutyAuthorizer() Authorizer { return dutyAuthorizer{} }
+
+func (dutyAuthorizer) Authorize(_ context.Context, p Principal, rec *docgov.DocumentRecord, action Action) error {
+	if rec == nil {
+		return uow.ErrPermissionDenied
+	}
+	switch action {
+	case ActionEdit, ActionCommit:
+		if rec.AuthorID != 0 && p.UserID != 0 && rec.AuthorID != p.UserID {
+			return uow.ErrPermissionDenied
+		}
+	case ActionApprove, ActionReject:
+		if rec.ApproverID != 0 && p.UserID != 0 && rec.ApproverID != p.UserID {
+			return uow.ErrPermissionDenied
+		}
+	default:
+		return uow.ErrPermissionDenied
+	}
+	return nil
+}
+
+func (c *Commands) run(ctx context.Context, docID string, p Principal, action Action, fn func(context.Context, uow.TxScope) error) error {
+	if c == nil {
+		return fmt.Errorf("docflow: commands required")
+	}
+	if c.Auth == nil {
+		c.Auth = DutyAuthorizer()
+	}
+	if c.UoW == nil {
+		if err := c.authorizeUnlocked(ctx, docID, p, action); err != nil {
+			return err
+		}
+		return fn(ctx, nil)
+	}
+	return c.UoW.Within(ctx, func(scope uow.TxScope) error {
+		ctx = uow.WithScope(ctx, scope)
+		if docID != "" && c.Gov != nil && c.Gov.UsingPG() {
+			rec, err := (docgov.Repository{}).GetForUpdate(ctx, scope, docID)
+			if err != nil {
+				if action == ActionApprove || action == ActionReject {
+					return err
+				}
+			} else if err := c.Auth.Authorize(ctx, p, rec, action); err != nil {
+				return err
+			}
+		}
+		return fn(ctx, scope)
+	})
+}
+
+func (c *Commands) authorizeUnlocked(ctx context.Context, docID string, p Principal, action Action) error {
+	if docID == "" || c.Gov == nil {
+		return nil
+	}
+	rec, err := c.Gov.GetDocument(ctx, docID)
+	if err != nil || rec == nil {
+		if action == ActionApprove || action == ActionReject {
+			if err != nil {
+				return err
+			}
+			return uow.ErrPermissionDenied
+		}
+		return nil
+	}
+	return c.Auth.Authorize(ctx, p, rec, action)
+}
+
+func (c *Commands) HumanEdit(ctx context.Context, p Principal, cmd HumanEditCommand) (*docrevision.HumanRevisionResult, error) {
+	var out *docrevision.HumanRevisionResult
+	err := c.run(ctx, cmd.DocID, p, ActionEdit, func(ctx context.Context, _ uow.TxScope) error {
+		res, err := c.Rev.CommitHumanRevision(ctx, cmd.Request)
+		out = res
+		return err
+	})
+	return out, err
+}
+
+func (c *Commands) AgentCommit(ctx context.Context, p Principal, cmd AgentCommitCommand) (*docrevision.RevisionState, error) {
+	var out *docrevision.RevisionState
+	err := c.run(ctx, cmd.DocID, p, ActionCommit, func(ctx context.Context, _ uow.TxScope) error {
+		st, err := c.Rev.CommitAgentVersion(ctx, cmd.DocID, cmd.CandidateID, cmd.Decision)
+		out = st
+		return err
+	})
+	if err == nil && c.UoW != nil && c.Rev != nil && c.Rev.Notify && c.Rev.OnOutbox != nil && out != nil {
+		_ = c.Rev.OnOutbox(ctx, "document.agent_version_committed", map[string]any{
+			"doc_id": cmd.DocID, "version_no": out.CurrentVersionNo,
+		})
+	}
+	return out, err
+}
+
+func (c *Commands) Approve(ctx context.Context, p Principal, cmd DecisionCommand) (*docgov.DocumentRecord, error) {
+	var out *docgov.DocumentRecord
+	err := c.run(ctx, cmd.DocID, p, ActionApprove, func(ctx context.Context, _ uow.TxScope) error {
+		doc, err := c.Gov.ApproverApprove(ctx, cmd.DocID, p.UserID, p.Role, p.Name, cmd.Comment)
+		out = doc
+		return err
+	})
+	if err == nil && out != nil && c.UoW != nil && c.Gov != nil {
+		c.Gov.RunApprovedHooksAfterCommit(ctx, out)
+	}
+	return out, err
+}
+
+func (c *Commands) Reject(ctx context.Context, p Principal, cmd DecisionCommand) (*docgov.DocumentRecord, error) {
+	var out *docgov.DocumentRecord
+	err := c.run(ctx, cmd.DocID, p, ActionReject, func(ctx context.Context, _ uow.TxScope) error {
+		doc, err := c.Gov.ApproverReject(ctx, cmd.DocID, p.UserID, p.Role, p.Name, cmd.RejectReason, cmd.RollbackTo)
+		out = doc
+		return err
+	})
+	return out, err
+}

internal/app/docflow/commands_test.go

@@ -0,0 +1,171 @@
+package docflow
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/docrevision"
+)
+
+func TestHumanEditCharacterizationSameETagOneWinner(t *testing.T) {
+	ctx := context.Background()
+	rev := docrevision.NewService(nil)
+	st, err := rev.OpenTemplateVersion(ctx, "doc_flow_etag", "# base\n", docrevision.ActorSnapshot{}, 1, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmds := &Commands{Rev: rev}
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	wins, losses := 0, 0
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			_, err := cmds.HumanEdit(ctx, Principal{UserID: int64(i + 1)}, HumanEditCommand{
+				DocID: "doc_flow_etag",
+				Request: docrevision.HumanRevisionRequest{
+					DocID: "doc_flow_etag", Markdown: fmt.Sprintf("# base\n\nedit %d\n", i),
+					ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo,
+					IdempotencyKey: fmt.Sprintf("flow-%d", i), ActorUserID: int64(i + 1), LifecycleState: "drafting",
+				},
+			})
+			mu.Lock()
+			defer mu.Unlock()
+			if err == nil {
+				wins++
+			} else if err == docrevision.ErrHeadConflict {
+				losses++
+			} else {
+				t.Errorf("unexpected %v", err)
+			}
+		}(i)
+	}
+	wg.Wait()
+	if wins != 1 || losses != 7 {
+		t.Fatalf("wins=%d losses=%d", wins, losses)
+	}
+	got, err := rev.GetRevisionState(ctx, "doc_flow_etag")
+	if err != nil || got.HeadRevisionNo != 1 {
+		t.Fatalf("head=%+v err=%v", got, err)
+	}
+}
+
+func TestApproveRejectCharacterizationOneTerminal(t *testing.T) {
+	ctx := context.Background()
+	gov := docgov.NewGovernanceEngine(nil)
+	doc, err := gov.CreateOrGetDocument(ctx, "doc_flow_appr", 1, "intake", "t", "V0.5_feasibility", "", 1, "A", 2, "R", 3, "P")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.SubmitForReview(ctx, doc.DocID, 1, docgov.RoleAuthor, "A"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.ReviewApprove(ctx, doc.DocID, 2, docgov.RoleReviewer, "R", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	cmds := &Commands{Gov: gov}
+	var wg sync.WaitGroup
+	results := make(chan string, 2)
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		doc, err := cmds.Approve(ctx, Principal{UserID: 3, Role: docgov.RoleApprover, Name: "P"}, DecisionCommand{DocID: doc.DocID})
+		if err == nil && doc != nil {
+			results <- string(doc.CurrentState)
+		} else {
+			results <- "err"
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		doc, err := cmds.Reject(ctx, Principal{UserID: 3, Role: docgov.RoleApprover, Name: "P"}, DecisionCommand{
+			DocID: doc.DocID, RejectReason: "no", RollbackTo: docgov.StateDrafting,
+		})
+		if err == nil && doc != nil {
+			results <- string(doc.CurrentState)
+		} else {
+			results <- "err"
+		}
+	}()
+	wg.Wait()
+	close(results)
+	var states []string
+	for s := range results {
+		states = append(states, s)
+	}
+	if len(states) != 2 {
+		t.Fatalf("%v", states)
+	}
+	approved, rejected := 0, 0
+	for _, s := range states {
+		switch s {
+		case string(docgov.StateApproved):
+			approved++
+		case string(docgov.StateDrafting), string(docgov.StateRejected):
+			rejected++
+		}
+	}
+	if approved+rejected != 1 {
+		t.Fatalf("expected one legal terminal, got %v", states)
+	}
+}
+
+func TestIdempotentHumanEditSameDigest(t *testing.T) {
+	ctx := context.Background()
+	rev := docrevision.NewService(nil)
+	st, err := rev.OpenTemplateVersion(ctx, "doc_flow_idemp", "# base\n", docrevision.ActorSnapshot{}, 1, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmds := &Commands{Rev: rev}
+	req := docrevision.HumanRevisionRequest{
+		DocID: "doc_flow_idemp", Markdown: "# base\n\nedit\n",
+		ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo,
+		IdempotencyKey: "same-key", ActorUserID: 1, LifecycleState: "drafting",
+	}
+	first, err := cmds.HumanEdit(ctx, Principal{UserID: 1}, HumanEditCommand{DocID: req.DocID, Request: req})
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := cmds.HumanEdit(ctx, Principal{UserID: 1}, HumanEditCommand{DocID: req.DocID, Request: req})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.State.HeadSnapshotID != second.State.HeadSnapshotID {
+		t.Fatal("same key+digest must replay")
+	}
+	req.Markdown = "# base\n\nother digest\n"
+	if _, err := cmds.HumanEdit(ctx, Principal{UserID: 1}, HumanEditCommand{DocID: req.DocID, Request: req}); err != docrevision.ErrIdempotencyConflict {
+		t.Fatalf("different digest: %v", err)
+	}
+}
+
+func TestReauthDeniesWrongAuthor(t *testing.T) {
+	ctx := context.Background()
+	gov := docgov.NewGovernanceEngine(nil)
+	if _, err := gov.CreateOrGetDocument(ctx, "doc_flow_authz", 1, "intake", "t", "V0.5_feasibility", "", 1, "A", 2, "R", 3, "P"); err != nil {
+		t.Fatal(err)
+	}
+	rev := docrevision.NewService(nil)
+	st, err := rev.OpenTemplateVersion(ctx, "doc_flow_authz", "# base\n", docrevision.ActorSnapshot{}, 1, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmds := &Commands{Gov: gov, Rev: rev}
+	_, err = cmds.HumanEdit(ctx, Principal{UserID: 99}, HumanEditCommand{
+		DocID: "doc_flow_authz",
+		Request: docrevision.HumanRevisionRequest{
+			DocID: "doc_flow_authz", Markdown: "# base\n\nnope\n",
+			ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo,
+			IdempotencyKey: "authz-deny", ActorUserID: 99, LifecycleState: "drafting",
+		},
+	})
+	if err != uow.ErrPermissionDenied {
+		t.Fatalf("want permission denied, got %v", err)
+	}
+}

internal/app/uow/uow.go

@@ -39,3 +39,21 @@ type RetryPolicy struct {
 func DefaultRetryPolicy() RetryPolicy {
 	return RetryPolicy{MaxAttempts: DefaultMaxAttempts, LockTimeout: DefaultLockTimeout}
 }
+
+type scopeCtxKey struct{}
+
+// WithScope lets owner repositories join the command UnitOfWork.
+func WithScope(ctx context.Context, scope TxScope) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return context.WithValue(ctx, scopeCtxKey{}, scope)
+}
+
+func ScopeFrom(ctx context.Context) (TxScope, bool) {
+	if ctx == nil {
+		return nil, false
+	}
+	s, ok := ctx.Value(scopeCtxKey{}).(TxScope)
+	return s, ok && s != nil
+}

internal/docgov/engine.go

@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	"chat-agent-system/internal/app/uow"
+
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
@@ -69,6 +71,14 @@ func (e *GovernanceEngine) runApprovedHooks(ctx context.Context, doc *DocumentRe
 	}
 }
 
+func (e *GovernanceEngine) UsingPG() bool {
+	return e != nil && e.pool != nil
+}
+
+func (e *GovernanceEngine) RunApprovedHooksAfterCommit(ctx context.Context, doc *DocumentRecord) {
+	e.runApprovedHooks(ctx, doc)
+}
+
 // NewGovernanceEngine 初始化治理引擎
 func NewGovernanceEngine(pool *pgxpool.Pool) *GovernanceEngine {
 	return &GovernanceEngine{
@@ -783,7 +793,9 @@ func (e *GovernanceEngine) SubmitForReview(ctx context.Context, docID string, ac
 	}
 
 	e.memAudits[docID] = append(e.memAudits[docID], audit)
-	e.syncDocAndAuditToDB(ctx, doc, audit)
+	if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil {
+		return nil, err
+	}
 	return doc, nil
 }
 
@@ -811,7 +823,9 @@ func (e *GovernanceEngine) ReviewApprove(ctx context.Context, docID string, acto
 	}
 
 	e.memAudits[docID] = append(e.memAudits[docID], audit)
-	e.syncDocAndAuditToDB(ctx, doc, audit)
+	if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil {
+		return nil, err
+	}
 	return doc, nil
 }
 
@@ -838,7 +852,9 @@ func (e *GovernanceEngine) ReviewReject(ctx context.Context, docID string, actor
 	_ = e.addCommentInternal(docID, actorID, actorName, actorRole, "", CommentRejection, rejectReason)
 
 	e.memAudits[docID] = append(e.memAudits[docID], audit)
-	e.syncDocAndAuditToDB(ctx, doc, audit)
+	if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil {
+		return nil, err
+	}
 	return doc, nil
 }
 
@@ -866,11 +882,15 @@ func (e *GovernanceEngine) ApproverApprove(ctx context.Context, docID string, ac
 	}
 
 	e.memAudits[docID] = append(e.memAudits[docID], audit)
-	e.syncDocAndAuditToDB(ctx, doc, audit)
+	if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil {
+		return nil, err
+	}
 
-	e.mu.Unlock()
-	e.runApprovedHooks(ctx, doc)
-	e.mu.Lock()
+	if _, joined := uow.ScopeFrom(ctx); !joined {
+		e.mu.Unlock()
+		e.runApprovedHooks(ctx, doc)
+		e.mu.Lock()
+	}
 
 	return doc, nil
 }
@@ -897,7 +917,9 @@ func (e *GovernanceEngine) ApproverReject(ctx context.Context, docID string, act
 	_ = e.addCommentInternal(docID, actorID, actorName, actorRole, "", CommentRejection, rejectReason)
 
 	e.memAudits[docID] = append(e.memAudits[docID], audit)
-	e.syncDocAndAuditToDB(ctx, doc, audit)
+	if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil {
+		return nil, err
+	}
 	return doc, nil
 }
 
@@ -1068,11 +1090,23 @@ func (e *GovernanceEngine) addCommentInternal(docID string, userID int64, userNa
 	return nil
 }
 
-func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *DocumentRecord, audit *AuditLog) {
+func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *DocumentRecord, audit *AuditLog) error {
 	if e.pool == nil {
-		return
+		return nil
 	}
-	_, _ = e.pool.Exec(ctx, `
+	exec := func(sql string, args ...any) error {
+		if scope, ok := uow.ScopeFrom(ctx); ok {
+			tx, err := uow.AsPG(scope)
+			if err != nil {
+				return err
+			}
+			_, err = tx.Exec(ctx, sql, args...)
+			return err
+		}
+		_, err := e.pool.Exec(ctx, sql, args...)
+		return err
+	}
+	if err := exec(`
 		UPDATE document_records
 		SET current_state = $1, is_locked = $2,
 		    last_rejected_reason = $3, last_rejected_by_role = $4, last_rejected_by_name = $5, last_rejected_at = $6,
@@ -1080,14 +1114,18 @@ func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *Documen
 		WHERE doc_id = $9
 	`, string(doc.CurrentState), doc.IsLocked,
 		doc.LastRejectedReason, doc.LastRejectedByRole, doc.LastRejectedByName, doc.LastRejectedAt,
-		doc.ApprovedAt, doc.UpdatedAt, doc.DocID)
-
+		doc.ApprovedAt, doc.UpdatedAt, doc.DocID); err != nil {
+		return fmt.Errorf("docgov: sync document: %w", err)
+	}
 	if audit != nil {
 		detJSON, _ := json.Marshal(audit.Details)
-		_, _ = e.pool.Exec(ctx, `
+		if err := exec(`
 			INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at)
 			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
 		`, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole),
-			string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt)
+			string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil {
+			return fmt.Errorf("docgov: sync audit: %w", err)
+		}
 	}
+	return nil
 }

internal/docgov/repository.go

@@ -58,10 +58,12 @@ func (Repository) GetForUpdate(ctx context.Context, scope uow.TxScope, docID str
 	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
+		       is_locked, is_template_only, content_sha256, work_dir,
+		       COALESCE(author_id, 0), COALESCE(reviewer_id, 0), COALESCE(approver_id, 0)
 		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)
+		&rec.IsLocked, &rec.IsTemplateOnly, &rec.ContentSHA256, &rec.WorkDir,
+		&rec.AuthorID, &rec.ReviewerID, &rec.ApproverID)
 	if err != nil {
 		return nil, fmt.Errorf("docgov: lock document: %w", err)
 	}

internal/docrevision/document_commit_concurrency_pg_test.go

@@ -0,0 +1,100 @@
+package docrevision_test
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/docflow"
+	"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/pgxpool"
+)
+
+func TestDocumentCommitConcurrencyPG(t *testing.T) {
+	if testing.Short() {
+		t.Skip("postgresql concurrency is not part of -short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	if strings.Contains(dsn, "5433") || strings.Contains(dsn, "agent_system_dev") ||
+		strings.Contains(dsn, "5434") && strings.Contains(dsn, "agent_system_test") ||
+		strings.Contains(dsn, "5435") {
+		t.Fatal("TEST_DATABASE_URL points at a shared env port; refuse agent_system_dev/test/main")
+	}
+	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)
+	}
+
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	var projectID int64
+	if err := pool.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`,
+		"docflow-"+suffix, "docflow").Scan(&projectID); err != nil {
+		t.Fatalf("project: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_change_events WHERE doc_id LIKE $1`, "docflow_"+suffix+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_versions WHERE doc_id LIKE $1`, "docflow_"+suffix+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_snapshots WHERE doc_id LIKE $1`, "docflow_"+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)
+	})
+
+	docID := "docflow_" + suffix
+	rev := docrevision.NewService(pool)
+	gov := docgov.NewGovernanceEngine(pool)
+	if _, err := gov.CreateOrGetDocument(ctx, docID, projectID, "intake", "t", "V0.5_feasibility", "ws/"+docID, 1, "A", 2, "R", 3, "P"); err != nil {
+		t.Fatal(err)
+	}
+	st, err := rev.OpenTemplateVersion(ctx, docID, "# base\n", docrevision.ActorSnapshot{}, projectID, "ws/"+docID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmds := &docflow.Commands{UoW: uow.NewPG(pool), Gov: gov, Rev: rev}
+
+	var wg sync.WaitGroup
+	wins := 0
+	var mu sync.Mutex
+	for i := 0; i < 2; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			_, err := cmds.HumanEdit(ctx, docflow.Principal{UserID: 1}, docflow.HumanEditCommand{
+				DocID: docID,
+				Request: docrevision.HumanRevisionRequest{
+					DocID: docID, Markdown: fmt.Sprintf("# base\n\nuser %d\n", i),
+					ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo,
+					IdempotencyKey: fmt.Sprintf("pg-%s-%d", suffix, i), ActorUserID: 1,
+					LifecycleState: "drafting", ProjectID: projectID, WorkDir: "ws/" + docID,
+				},
+			})
+			if err == nil {
+				mu.Lock()
+				wins++
+				mu.Unlock()
+			}
+		}(i)
+	}
+	wg.Wait()
+	if wins != 1 {
+		t.Fatalf("same etag concurrent edits: wins=%d", wins)
+	}
+	got, err := rev.GetRevisionState(ctx, docID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.HeadRevisionNo != 1 {
+		t.Fatalf("head revision=%d", got.HeadRevisionNo)
+	}
+}

internal/docrevision/pg_store.go

@@ -6,6 +6,8 @@ import (
 	"errors"
 	"time"
 
+	"chat-agent-system/internal/app/uow"
+
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -444,6 +446,13 @@ func (s *Service) withTx(ctx context.Context, fn func(pgx.Tx) error) error {
 	if !s.usingPG() {
 		return nil
 	}
+	if scope, ok := uow.ScopeFrom(ctx); ok {
+		tx, err := uow.AsPG(scope)
+		if err != nil {
+			return err
+		}
+		return fn(tx)
+	}
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {
 		return err

internal/docrevision/service.go

@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	"chat-agent-system/internal/app/uow"
+
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -203,6 +205,11 @@ func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequ
 
 	if eid, ok := s.idemp[req.IdempotencyKey]; ok {
 		ev := s.events[eid]
+		if ev != nil {
+			if snap := s.snapshots[ev.ToSnapshotID]; snap != nil && snap.ContentSHA256 != sha {
+				return nil, ErrIdempotencyConflict
+			}
+		}
 		st, _ := s.revisionStateLocked(req.DocID)
 		overlay := s.manualOverlayLocked(req.DocID)
 		return &HumanRevisionResult{State: *st, Event: *ev, Overlay: overlay}, nil
@@ -672,7 +679,7 @@ func (s *Service) CommitAgentVersion(ctx context.Context, docID string, candidat
 		return nil, err
 	}
 
-	if s.Notify && s.OnOutbox != nil {
+	if _, joined := uow.ScopeFrom(ctx); !joined && s.Notify && s.OnOutbox != nil {
 		_ = s.OnOutbox(ctx, "document.agent_version_committed", map[string]any{
 			"doc_id": c.DocID, "version_no": newVer.VersionNo, "risk": rep.RiskLevel, "candidate_id": c.ID,
 		})

internal/server/document_revision_handlers.go

@@ -9,11 +9,24 @@ import (
 	"strconv"
 	"strings"
 
+	"chat-agent-system/internal/app/docflow"
+	"chat-agent-system/internal/app/uow"
 	"chat-agent-system/internal/docgov"
 	"chat-agent-system/internal/docrevision"
 	"chat-agent-system/internal/freedoc"
 )
 
+func (s *Server) docflowCommands() *docflow.Commands {
+	if s == nil || s.svc == nil {
+		return nil
+	}
+	cmds := &docflow.Commands{Gov: s.svc.DocGov(), Rev: s.svc.DocRevision()}
+	if pool := s.svc.Pool(); pool != nil {
+		cmds.UoW = uow.NewPG(pool)
+	}
+	return cmds
+}
+
 func (s *Server) registerDocumentRevisionRoutes(mux *http.ServeMux) {
 	mux.HandleFunc("GET /api/v1/documents/{doc_id}/revision-state", s.handleDocRevisionState)
 	mux.HandleFunc("GET /api/v1/documents/{doc_id}/versions", s.handleDocRevisionListVersions)
@@ -235,12 +248,20 @@ func (s *Server) handleDocRevisionPatchContent(w http.ResponseWriter, r *http.Re
 			return
 		}
 	}
-	res, err := rev.CommitHumanRevision(r.Context(), docrevision.HumanRevisionRequest{
-		DocID: docID, Markdown: body.Markdown, ExpectedHeadSnapshot: expectedHead, ExpectedRevisionNo: expectedRev,
-		IdempotencyKey: idemp, EditSessionID: body.EditSessionID, ActorUserID: user.ID,
-		ActorSnapshot:  docrevision.ActorSnapshot{DisplayName: user.Name, Role: "author", Duty: "author"},
-		LifecycleState: lifecycle, WorkDir: workDir, ProjectID: projectID, AllowCreateBaseline: false,
-		ClientBlockMap: body.ClientBlockMap,
+	cmds := s.docflowCommands()
+	if cmds == nil || cmds.Rev == nil {
+		writeJSONResponse(w, http.StatusServiceUnavailable, map[string]any{"error": "docrevision unavailable"})
+		return
+	}
+	res, err := cmds.HumanEdit(r.Context(), docflow.Principal{UserID: user.ID, Name: user.Name, Role: docgov.RoleAuthor}, docflow.HumanEditCommand{
+		DocID: docID,
+		Request: docrevision.HumanRevisionRequest{
+			DocID: docID, Markdown: body.Markdown, ExpectedHeadSnapshot: expectedHead, ExpectedRevisionNo: expectedRev,
+			IdempotencyKey: idemp, EditSessionID: body.EditSessionID, ActorUserID: user.ID,
+			ActorSnapshot:  docrevision.ActorSnapshot{DisplayName: user.Name, Role: "author", Duty: "author"},
+			LifecycleState: lifecycle, WorkDir: workDir, ProjectID: projectID, AllowCreateBaseline: false,
+			ClientBlockMap: body.ClientBlockMap,
+		},
 	})
 	if errors.Is(err, docrevision.ErrHeadConflict) && res != nil && res.Conflict != nil {
 		writeJSONResponse(w, http.StatusConflict, map[string]any{
@@ -343,11 +364,19 @@ func (s *Server) handleDocRevisionCandidateDecision(w http.ResponseWriter, r *ht
 		return
 	}
 	cand, _ := rev.GetCandidate(r.Context(), docID, cid)
-	st, err := rev.CommitAgentVersion(r.Context(), docID, cid, docrevision.CommitAgentDecision{
-		Decision: body.Decision, Confirmed: body.Confirmed,
-		ExpectedInputSHA: body.ExpectedInputSHA, ExpectedOutputSHA: body.ExpectedOutputSHA,
-		Reason: body.Reason, ActorUserID: user.ID,
-		ActorSnapshot: docrevision.ActorSnapshot{DisplayName: user.Name},
+	cmds := s.docflowCommands()
+	if cmds == nil || cmds.Rev == nil {
+		writeJSONResponse(w, http.StatusServiceUnavailable, map[string]any{"error": "docrevision unavailable"})
+		return
+	}
+	st, err := cmds.AgentCommit(r.Context(), docflow.Principal{UserID: user.ID, Name: user.Name, Role: docgov.RoleAuthor}, docflow.AgentCommitCommand{
+		DocID: docID, CandidateID: cid,
+		Decision: docrevision.CommitAgentDecision{
+			Decision: body.Decision, Confirmed: body.Confirmed,
+			ExpectedInputSHA: body.ExpectedInputSHA, ExpectedOutputSHA: body.ExpectedOutputSHA,
+			Reason: body.Reason, ActorUserID: user.ID,
+			ActorSnapshot: docrevision.ActorSnapshot{DisplayName: user.Name},
+		},
 	})
 	if err != nil {
 		writeDocRevisionError(w, err)
@@ -457,8 +486,12 @@ func writeDocRevisionError(w http.ResponseWriter, err error) {
 		writeJSONResponse(w, http.StatusUnprocessableEntity, map[string]any{"code": docrevision.ErrCodeReportInvalid, "error": err.Error()})
 	case errors.Is(err, docrevision.ErrTooLarge):
 		writeJSONResponse(w, http.StatusRequestEntityTooLarge, map[string]any{"code": docrevision.ErrCodeTooLarge, "error": err.Error()})
-	case errors.Is(err, docrevision.ErrForbidden):
+	case errors.Is(err, docrevision.ErrForbidden), errors.Is(err, uow.ErrPermissionDenied):
 		writeJSONResponse(w, http.StatusForbidden, map[string]any{"code": docrevision.ErrCodeEditForbidden, "error": err.Error()})
+	case errors.Is(err, docrevision.ErrIdempotencyConflict), errors.Is(err, uow.ErrDigestConflict):
+		writeJSONResponse(w, http.StatusConflict, map[string]any{"code": "DOCUMENT_IDEMPOTENCY_CONFLICT", "error": err.Error()})
+	case errors.Is(err, uow.ErrBusy):
+		writeJSONResponse(w, http.StatusConflict, map[string]any{"code": "DOCUMENT_BUSY", "error": err.Error()})
 	default:
 		writeJSONResponse(w, http.StatusBadRequest, map[string]any{"error": err.Error()})
 	}

internal/server/handlers.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"time"
 
+	"chat-agent-system/internal/app/docflow"
+	"chat-agent-system/internal/app/uow"
 	"chat-agent-system/internal/bizdev"
 	"chat-agent-system/internal/ctxmgr"
 	"chat-agent-system/internal/docgov"
@@ -21,15 +23,15 @@ import (
 
 // HealthResponse 健康检查响应
 type HealthResponse struct {
-	Status        string `json:"status"`
-	Env           string `json:"env"`
-	AgentBackend  string `json:"agent_backend"`
-	Version       string `json:"version"`
-	SharedKBPort  int    `json:"shared_kb_port"`
-	DatabaseReady bool   `json:"database_ready"`
-	SchemaVersion int64  `json:"schema_version,omitempty"`
-	SchemaReady   bool   `json:"schema_ready,omitempty"`
-	TaskStoreReady bool  `json:"task_store_ready,omitempty"`
+	Status         string `json:"status"`
+	Env            string `json:"env"`
+	AgentBackend   string `json:"agent_backend"`
+	Version        string `json:"version"`
+	SharedKBPort   int    `json:"shared_kb_port"`
+	DatabaseReady  bool   `json:"database_ready"`
+	SchemaVersion  int64  `json:"schema_version,omitempty"`
+	SchemaReady    bool   `json:"schema_ready,omitempty"`
+	TaskStoreReady bool   `json:"task_store_ready,omitempty"`
 }
 
 // ChatRequest 对话请求
@@ -50,32 +52,32 @@ type ChatRequest struct {
 
 // TaskRunRequestDTO 独立任务运行请求
 type TaskRunRequestDTO struct {
-	TaskKey         string                `json:"task_key"`
-	Stage           string                `json:"stage,omitempty"` // 若指定 stage，则执行成熟度阶段跃迁
-	ProjectName     string                `json:"project_name"`
-	CustomerName    string                `json:"customer_name"`
-	DeviceType      string                `json:"device_type"`
-	IntendedUse     string                `json:"intended_use"`
-	TargetMarket    string                `json:"target_market"`
-	Fields          map[string]string     `json:"fields,omitempty"`
-	Backend         string                `json:"backend,omitempty"`
-	AutoRepair      bool                  `json:"auto_repair"`
-	UserID          int64                 `json:"user_id,omitempty"`
-	ProjectID       int64                 `json:"project_id,omitempty"`
-	DocID           string                `json:"doc_id,omitempty"`
-	AuthorID        int64                 `json:"author_id,omitempty"`
-	AuthorName      string                `json:"author_name,omitempty"`
-	ReviewerID      int64                 `json:"reviewer_id,omitempty"`
-	ReviewerName    string                `json:"reviewer_name,omitempty"`
-	ApproverID      int64                 `json:"approver_id,omitempty"`
-	ApproverName    string                `json:"approver_name,omitempty"`
-	ForceDirectEdit bool                  `json:"force_direct_edit,omitempty"`
-	CustomSteps     []squad.StepSpec      `json:"custom_steps,omitempty"`
-	WorkflowKey     string                `json:"workflow_key,omitempty"`
-	SessionID       string                `json:"session_id,omitempty"`
-	ClientTurnID    string                `json:"client_turn_id,omitempty"`
+	TaskKey         string            `json:"task_key"`
+	Stage           string            `json:"stage,omitempty"` // 若指定 stage，则执行成熟度阶段跃迁
+	ProjectName     string            `json:"project_name"`
+	CustomerName    string            `json:"customer_name"`
+	DeviceType      string            `json:"device_type"`
+	IntendedUse     string            `json:"intended_use"`
+	TargetMarket    string            `json:"target_market"`
+	Fields          map[string]string `json:"fields,omitempty"`
+	Backend         string            `json:"backend,omitempty"`
+	AutoRepair      bool              `json:"auto_repair"`
+	UserID          int64             `json:"user_id,omitempty"`
+	ProjectID       int64             `json:"project_id,omitempty"`
+	DocID           string            `json:"doc_id,omitempty"`
+	AuthorID        int64             `json:"author_id,omitempty"`
+	AuthorName      string            `json:"author_name,omitempty"`
+	ReviewerID      int64             `json:"reviewer_id,omitempty"`
+	ReviewerName    string            `json:"reviewer_name,omitempty"`
+	ApproverID      int64             `json:"approver_id,omitempty"`
+	ApproverName    string            `json:"approver_name,omitempty"`
+	ForceDirectEdit bool              `json:"force_direct_edit,omitempty"`
+	CustomSteps     []squad.StepSpec  `json:"custom_steps,omitempty"`
+	WorkflowKey     string            `json:"workflow_key,omitempty"`
+	SessionID       string            `json:"session_id,omitempty"`
+	ClientTurnID    string            `json:"client_turn_id,omitempty"`
 	// Materials 用户上传的参考文档材料，由前端 Base64 编码后传入；后端 json 反序列化自动解码 []byte
-	Materials       []bizdev.MaterialInput `json:"materials,omitempty"`
+	Materials []bizdev.MaterialInput `json:"materials,omitempty"`
 }
 
 func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
@@ -427,11 +429,11 @@ func (s *Server) handleGetUserContext(w http.ResponseWriter, r *http.Request) {
 
 // UpdateUserPreferenceRequest 更新用户工作偏好请求
 type UpdateUserPreferenceRequest struct {
-	UserID             int64                      `json:"user_id"`
-	CommunicationStyle ctxmgr.CommunicationStyle  `json:"communication_style"`
-	FocusAreas         []string                   `json:"focus_areas"`
-	OutputFormatPref   string                     `json:"output_format_pref"`
-	CustomInstructions string                     `json:"custom_instructions"`
+	UserID             int64                     `json:"user_id"`
+	CommunicationStyle ctxmgr.CommunicationStyle `json:"communication_style"`
+	FocusAreas         []string                  `json:"focus_areas"`
+	OutputFormatPref   string                    `json:"output_format_pref"`
+	CustomInstructions string                    `json:"custom_instructions"`
 }
 
 // handleUpdateUserPreference 定期整理或前端保存用户习惯偏好
@@ -528,10 +530,10 @@ func (s *Server) handleDocGovGetDocument(w http.ResponseWriter, r *http.Request)
 
 // DocGovClaimRequest 认领文档请求
 type DocGovClaimRequest struct {
-	DocID     string `json:"doc_id"`
-	UserID    int64  `json:"user_id"`
-	UserName  string `json:"user_name"`
-	UserRole  string `json:"user_role"` // author, reviewer, approver
+	DocID    string `json:"doc_id"`
+	UserID   int64  `json:"user_id"`
+	UserName string `json:"user_name"`
+	UserRole string `json:"user_role"` // author, reviewer, approver
 }
 
 // handleDocGovClaim 查阅并认领受控文档 (非模板状态查看方可认领)
@@ -609,7 +611,7 @@ type DocGovDecisionRequest struct {
 	UserName   string `json:"user_name"`
 	Action     string `json:"action"` // approve 或 reject
 	Comment    string `json:"comment,omitempty"`
-	Reason     string `json:"reason,omitempty"`     // 驳回时必填
+	Reason     string `json:"reason,omitempty"`      // 驳回时必填
 	RollbackTo string `json:"rollback_to,omitempty"` // 批准人驳回时可选: drafting 或 reviewing
 }
 
@@ -681,24 +683,39 @@ func (s *Server) handleDocGovApproverDecision(w http.ResponseWriter, r *http.Req
 	}
 
 	action := strings.ToLower(strings.TrimSpace(req.Action))
+	cmds := s.docflowCommands()
+	if cmds == nil || cmds.Gov == nil {
+		http.Error(w, "docgov engine not initialized", http.StatusServiceUnavailable)
+		return
+	}
+	p := docflow.Principal{UserID: user.ID, Name: user.Name, Role: docgov.RoleApprover}
 	var doc *docgov.DocumentRecord
 	var err error
 
 	if action == "approve" {
-		doc, err = s.svc.DocGov().ApproverApprove(r.Context(), req.DocID, user.ID, docgov.RoleApprover, user.Name, req.Comment)
+		doc, err = cmds.Approve(r.Context(), p, docflow.DecisionCommand{DocID: req.DocID, Comment: req.Comment})
 	} else if action == "reject" {
 		reason := req.Reason
 		if reason == "" {
 			reason = req.Comment
 		}
-		rollbackState := docgov.DocumentState(req.RollbackTo)
-		doc, err = s.svc.DocGov().ApproverReject(r.Context(), req.DocID, user.ID, docgov.RoleApprover, user.Name, reason, rollbackState)
+		doc, err = cmds.Reject(r.Context(), p, docflow.DecisionCommand{
+			DocID: req.DocID, RejectReason: reason, RollbackTo: docgov.DocumentState(req.RollbackTo),
+		})
 	} else {
 		http.Error(w, "invalid action: must be approve or reject", http.StatusBadRequest)
 		return
 	}
 
 	if err != nil {
+		if errors.Is(err, uow.ErrPermissionDenied) {
+			http.Error(w, "approver decision failed: "+err.Error(), http.StatusForbidden)
+			return
+		}
+		if errors.Is(err, uow.ErrBusy) {
+			http.Error(w, "approver decision failed: "+err.Error(), http.StatusConflict)
+			return
+		}
 		http.Error(w, "approver decision failed: "+err.Error(), http.StatusBadRequest)
 		return
 	}