fix(authz): clear reviewer/approver slots in the member-write transaction

Duty revoke must share the project advisory lock with reading old holders and writing members, so a post-commit service snapshot cannot miss a transfer. Heal still fills the new holder after commit, and sync failure stays ERR_DUTY_SYNC.

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

internal/authz/authorizer.go

@@ -282,6 +282,8 @@ func isRetryableDutyConflict(err error) bool {
 }
 
 // UpsertProjectMemberSpec 写入访问角色，并可选独占 document_duty。
+// 写入 document_duty 时，同一事务内：项目 advisory lock、读取旧/冲突 holder、
+// 成员更新、精确清空 document_records 的 reviewer/approver 槽；不改 author。
 // 并发抢同一职责时重试，收敛为 last-writer，并依赖部分唯一索引保证无双 reviewer。
 func (a *Authorizer) UpsertProjectMemberSpec(ctx context.Context, projectID, userID int64, spec MemberUpsert) error {
 	if projectID <= 0 || userID <= 0 {
@@ -320,41 +322,20 @@ func (a *Authorizer) upsertProjectMemberSpecOnce(ctx context.Context, projectID,
 	defer func() { _ = tx.Rollback(ctx) }()
 
 	if spec.DocumentDuty != nil {
-		if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil {
+		if err := applyDutyWriteInTx(ctx, tx, projectID, userID, spec.ProjectRole, duty); err != nil {
 			return err
 		}
+		return tx.Commit(ctx)
 	}
 
-	if spec.DocumentDuty != nil && duty != docduty.DutyNone {
-		if _, err := tx.Exec(ctx, `
-			UPDATE project_members
-			SET document_duty = '', updated_at = now()
-			WHERE project_id = $1 AND document_duty = $2 AND user_id <> $3 AND status = $4
-		`, projectID, duty, userID, StatusActive); err != nil {
-			return err
-		}
-	}
-
-	if spec.DocumentDuty != nil {
-		_, err = tx.Exec(ctx, `
-			INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at)
-			VALUES ($1, $2, $3, $4, $5, now(), now())
-			ON CONFLICT (project_id, user_id) DO UPDATE SET
-				project_role = EXCLUDED.project_role,
-				document_duty = EXCLUDED.document_duty,
-				status = EXCLUDED.status,
-				updated_at = now()
-		`, projectID, userID, spec.ProjectRole, duty, StatusActive)
-	} else {
-		_, err = tx.Exec(ctx, `
-			INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at)
-			VALUES ($1, $2, $3, $4, now(), now())
-			ON CONFLICT (project_id, user_id) DO UPDATE SET
-				project_role = EXCLUDED.project_role,
-				status = EXCLUDED.status,
-				updated_at = now()
-		`, projectID, userID, spec.ProjectRole, StatusActive)
-	}
+	_, err = tx.Exec(ctx, `
+		INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at)
+		VALUES ($1, $2, $3, $4, now(), now())
+		ON CONFLICT (project_id, user_id) DO UPDATE SET
+			project_role = EXCLUDED.project_role,
+			status = EXCLUDED.status,
+			updated_at = now()
+	`, projectID, userID, spec.ProjectRole, StatusActive)
 	if err != nil {
 		return err
 	}
@@ -362,6 +343,7 @@ func (a *Authorizer) upsertProjectMemberSpecOnce(ctx context.Context, projectID,
 }
 
 // SetProjectMemberStatus 更新成员状态（如 exited）。
+// 退出时与 upsert 相同：advisory lock、读取旧 duty、状态更新、精确清槽在同一事务。
 func (a *Authorizer) SetProjectMemberStatus(ctx context.Context, projectID, userID int64, status string) error {
 	if projectID <= 0 || userID <= 0 {
 		return fmt.Errorf("authz: invalid project or user id")
@@ -372,7 +354,32 @@ func (a *Authorizer) SetProjectMemberStatus(ctx context.Context, projectID, user
 	if !a.enabled() {
 		return nil
 	}
-	tag, err := a.pool.Exec(ctx, `
+	const maxAttempts = 4
+	var last error
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		last = a.setProjectMemberStatusOnce(ctx, projectID, userID, status)
+		if last == nil || !isRetryableDutyConflict(last) {
+			return last
+		}
+	}
+	return last
+}
+
+func (a *Authorizer) setProjectMemberStatusOnce(ctx context.Context, projectID, userID int64, status string) error {
+	tx, err := a.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil {
+		return err
+	}
+	prevDuty, err := docduty.LoadMemberDutyTx(ctx, tx, projectID, userID)
+	if err != nil {
+		return err
+	}
+	tag, err := tx.Exec(ctx, `
 		UPDATE project_members
 		SET status = $3, updated_at = now()
 		WHERE project_id = $1 AND user_id = $2
@@ -383,7 +390,58 @@ func (a *Authorizer) SetProjectMemberStatus(ctx context.Context, projectID, user
 	if tag.RowsAffected() == 0 {
 		return fmt.Errorf("authz: project member not found")
 	}
-	return nil
+	if status == StatusExited {
+		if err := docduty.ApplySlotRevocations(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{
+			UserID: userID, PreviousDuty: prevDuty, NextDuty: prevDuty, Exited: true,
+		})); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
+}
+
+// applyDutyWriteInTx 假定调用方已 Begin：lock → 读旧/冲突 holder → 独占写成员 → 清槽。
+func applyDutyWriteInTx(ctx context.Context, tx pgx.Tx, projectID, userID int64, projectRole, duty string) error {
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil {
+		return err
+	}
+	prevDuty, err := docduty.LoadMemberDutyTx(ctx, tx, projectID, userID)
+	if err != nil {
+		return err
+	}
+	holders, err := docduty.LoadActiveSlotHoldersTx(ctx, tx, projectID)
+	if err != nil {
+		return err
+	}
+
+	if duty != docduty.DutyNone {
+		if _, err := tx.Exec(ctx, `
+			UPDATE project_members
+			SET document_duty = '', updated_at = now()
+			WHERE project_id = $1 AND document_duty = $2 AND user_id <> $3 AND status = $4
+		`, projectID, duty, userID, StatusActive); err != nil {
+			return err
+		}
+	}
+
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at)
+		VALUES ($1, $2, $3, $4, $5, now(), now())
+		ON CONFLICT (project_id, user_id) DO UPDATE SET
+			project_role = EXCLUDED.project_role,
+			document_duty = EXCLUDED.document_duty,
+			status = EXCLUDED.status,
+			updated_at = now()
+	`, projectID, userID, projectRole, duty, StatusActive); err != nil {
+		return err
+	}
+
+	return docduty.ApplySlotRevocations(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{
+		UserID:         userID,
+		PreviousDuty:   prevDuty,
+		NextDuty:       duty,
+		PriorHolderIDs: holders,
+	}))
 }
 
 // TableExists 检查 project_members 表是否存在。

internal/authz/authorizer_pg_test.go

@@ -115,6 +115,162 @@ func TestUpsertProjectMemberDuty_ConcurrentLastWriter_PG(t *testing.T) {
 	}
 }
 
+func TestUpsertProjectMember_AtomicDutyRevoke_PG(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping PostgreSQL duty revoke atomicity in -short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+	defer cancel()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("postgres unavailable: %v", err)
+	}
+	defer pool.Close()
+	if err := pool.Ping(ctx); err != nil {
+		t.Fatalf("postgres ping failed: %v", err)
+	}
+	if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
+		t.Fatalf("migration failed: %v", err)
+	}
+
+	suffix := time.Now().UnixNano()
+	authorID := insertAuthzUser(t, ctx, pool, suffix, "author")
+	reviewerA := insertAuthzUser(t, ctx, pool, suffix, "ra")
+	reviewerB := insertAuthzUser(t, ctx, pool, suffix, "rb")
+
+	var projectID int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO projects (code, name, status, creator_user_id)
+		VALUES ($1, $2, 'active', $3)
+		RETURNING id
+	`, fmt.Sprintf("duty-atomic-%d", suffix), "atomic upsert revoke", authorID).Scan(&projectID); err != nil {
+		t.Fatalf("insert project: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3)`, authorID, reviewerA, reviewerB)
+	})
+
+	a := NewAuthorizer(pool)
+	dutyAuthor := docduty.DutyAuthor
+	dutyReviewer := docduty.DutyReviewer
+	dutyNone := docduty.DutyNone
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, authorID, MemberUpsert{
+		ProjectRole: RoleOwner, DocumentDuty: &dutyAuthor,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerA, MemberUpsert{
+		ProjectRole: RoleMember, DocumentDuty: &dutyReviewer,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerB, MemberUpsert{
+		ProjectRole: RoleMember, DocumentDuty: &dutyNone,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	docID := fmt.Sprintf("doc_proj%d_intake", projectID)
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO document_records (
+			doc_id, project_id, task_key, title, stage, version,
+			current_state, is_locked, is_template_only,
+			author_id, author_name, reviewer_id, reviewer_name, approver_id, approver_name,
+			created_at, updated_at
+		) VALUES (
+			$1, $2, 'intake', 'intake doc', 'V0.5', 1,
+			'drafting', false, false,
+			$3, '编制工程师', $4, '审核负责人', NULL, '',
+			now(), now()
+		)
+	`, docID, projectID, authorID, reviewerA); err != nil {
+		t.Fatalf("insert doc: %v", err)
+	}
+
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerA, MemberUpsert{
+		ProjectRole: RoleMember, DocumentDuty: &dutyNone,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0)
+	assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID)
+	assertAuthzMemberDuty(t, ctx, pool, projectID, reviewerA, docduty.DutyNone)
+
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerA, MemberUpsert{
+		ProjectRole: RoleMember, DocumentDuty: &dutyReviewer,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerA)
+
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerB, MemberUpsert{
+		ProjectRole: RoleMember, DocumentDuty: &dutyReviewer,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0)
+	assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID)
+	assertAuthzMemberDuty(t, ctx, pool, projectID, reviewerA, docduty.DutyNone)
+	assertAuthzMemberDuty(t, ctx, pool, projectID, reviewerB, docduty.DutyReviewer)
+
+	if _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerB)
+	assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID)
+
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerB, MemberUpsert{
+		ProjectRole: RoleMember, DocumentDuty: &dutyReviewer,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerB)
+
+	if err := a.SetProjectMemberStatus(ctx, projectID, reviewerB, StatusExited); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0)
+	assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID)
+
+	if err := a.SetProjectMemberStatus(ctx, projectID, reviewerB, StatusExited); err != nil {
+		t.Fatal(err)
+	}
+	assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0)
+	assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID)
+}
+
+func assertAuthzDocSlot(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, column string, want int64) {
+	t.Helper()
+	var got int64
+	q := fmt.Sprintf(`SELECT COALESCE(%s, 0) FROM document_records WHERE doc_id = $1`, column)
+	if err := pool.QueryRow(ctx, q, docID).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != want {
+		t.Fatalf("%s %s=%d want %d", docID, column, got, want)
+	}
+}
+
+func assertAuthzMemberDuty(t *testing.T, ctx context.Context, pool *pgxpool.Pool, projectID, userID int64, want string) {
+	t.Helper()
+	var got string
+	if err := pool.QueryRow(ctx, `
+		SELECT COALESCE(document_duty, '') FROM project_members WHERE project_id = $1 AND user_id = $2
+	`, projectID, userID).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != want {
+		t.Fatalf("user %d document_duty=%q want %q", userID, got, want)
+	}
+}
+
 func insertAuthzUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix int64, kind string) int64 {
 	t.Helper()
 	var id int64

internal/docduty/revoke.go

@@ -2,9 +2,10 @@ package docduty
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
-	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/jackc/pgx/v5"
 )
 
 // MemberDutyChange 成员职责/在任状态变更，供显式撤销文档槽位。
@@ -64,20 +65,56 @@ func PlanSlotRevocations(change MemberDutyChange) []SlotRevocation {
 	return out
 }
 
-// ReconcileMemberDuty 在项目级 advisory lock 下按更新前 duty 清空对应槽位，再 heal 新职责。
-func ReconcileMemberDuty(ctx context.Context, pool *pgxpool.Pool, projectID int64, change MemberDutyChange) error {
-	if pool == nil || projectID <= 0 {
+// LoadMemberDutyTx 在调用方事务内读取该成员当前 document_duty；无行则空串。
+func LoadMemberDutyTx(ctx context.Context, tx pgx.Tx, projectID, userID int64) (string, error) {
+	var duty string
+	err := tx.QueryRow(ctx, `
+		SELECT COALESCE(document_duty, '')
+		FROM project_members
+		WHERE project_id = $1 AND user_id = $2
+	`, projectID, userID).Scan(&duty)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("docduty: load member duty: %w", err)
+	}
+	return duty, nil
+}
+
+// LoadActiveSlotHoldersTx 在调用方事务内读取在任 reviewer/approver（不含 author）。
+func LoadActiveSlotHoldersTx(ctx context.Context, tx pgx.Tx, projectID int64) (map[string]int64, error) {
+	rows, err := tx.Query(ctx, `
+		SELECT user_id, document_duty
+		FROM project_members
+		WHERE project_id = $1 AND status = $2 AND document_duty IN ($3, $4)
+	`, projectID, StatusActive, DutyReviewer, DutyApprover)
+	if err != nil {
+		return nil, fmt.Errorf("docduty: load slot holders: %w", err)
+	}
+	defer rows.Close()
+	out := make(map[string]int64, 2)
+	for rows.Next() {
+		var userID int64
+		var duty string
+		if err := rows.Scan(&userID, &duty); err != nil {
+			return nil, fmt.Errorf("docduty: scan slot holder: %w", err)
+		}
+		out[duty] = userID
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("docduty: slot holders: %w", err)
+	}
+	return out, nil
+}
+
+// ApplySlotRevocations 在调用方已持有的事务内精确清空 reviewer/approver 槽。
+// 不开启新事务、不加锁、不 heal；author 列永不写入。
+func ApplySlotRevocations(ctx context.Context, tx pgx.Tx, projectID int64, revs []SlotRevocation) error {
+	if tx == nil || projectID <= 0 {
 		return nil
 	}
-	tx, err := pool.Begin(ctx)
-	if err != nil {
-		return fmt.Errorf("docduty: begin reconcile: %w", err)
-	}
-	defer func() { _ = tx.Rollback(ctx) }()
-	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil {
-		return fmt.Errorf("docduty: lock project %d: %w", projectID, err)
-	}
-	for _, rev := range PlanSlotRevocations(change) {
+	for _, rev := range revs {
 		q, ok := clearDutySlotSQL(rev.Duty)
 		if !ok {
 			continue
@@ -86,11 +123,7 @@ func ReconcileMemberDuty(ctx context.Context, pool *pgxpool.Pool, projectID int6
 			return fmt.Errorf("docduty: clear %s for user %d: %w", rev.Duty, rev.UserID, err)
 		}
 	}
-	if err := tx.Commit(ctx); err != nil {
-		return fmt.Errorf("docduty: commit reconcile: %w", err)
-	}
-	_, _, err = HealProjectDocuments(ctx, pool, projectID)
-	return err
+	return nil
 }
 
 func clearDutySlotSQL(duty string) (string, bool) {

internal/docduty/store_pg_test.go

@@ -207,104 +207,94 @@ func TestHealBinding_EmptyAssignedClearsExited_PG(t *testing.T) {
 	assertDocAuthorName(t, ctx, pool, docID, "用户ex_author")
 }
 
-func TestReconcileMemberDuty_ReviewerToViewerAndTransfer_PG(t *testing.T) {
+func TestApplySlotRevocations_SameTxnCommitAndRollback_PG(t *testing.T) {
 	ctx, _, pool := openDutyTestPool(t)
 	suffix := time.Now().UnixNano()
-	authorID := insertDutyUser(t, ctx, pool, suffix, "rec_author")
-	reviewerA := insertDutyUser(t, ctx, pool, suffix, "rec_ra")
-	reviewerB := insertDutyUser(t, ctx, pool, suffix, "rec_rb")
+	authorID := insertDutyUser(t, ctx, pool, suffix, "txn_author")
+	reviewerA := insertDutyUser(t, ctx, pool, suffix, "txn_ra")
 
 	var projectID int64
 	if err := pool.QueryRow(ctx, `
 		INSERT INTO projects (code, name, status, creator_user_id)
 		VALUES ($1, $2, 'active', $3)
 		RETURNING id
-	`, fmt.Sprintf("duty-rec-%d", suffix), "explicit reconcile", authorID).Scan(&projectID); err != nil {
+	`, fmt.Sprintf("duty-txn-%d", suffix), "atomic revoke txn", authorID).Scan(&projectID); err != nil {
 		t.Fatalf("insert project: %v", err)
 	}
 	t.Cleanup(func() {
 		_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id = $1`, projectID)
 		_, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID)
 		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
-		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3)`, authorID, reviewerA, reviewerB)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2)`, authorID, reviewerA)
 	})
 	if _, err := pool.Exec(ctx, `
 		INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at)
 		VALUES ($1, $2, 'owner', 'author', 'active', now(), now()),
-		       ($1, $3, 'member', 'reviewer', 'active', now(), now()),
-		       ($1, $4, 'member', '', 'active', now(), now())
-	`, projectID, authorID, reviewerA, reviewerB); err != nil {
+		       ($1, $3, 'member', 'reviewer', 'active', now(), now())
+	`, projectID, authorID, reviewerA); err != nil {
 		t.Fatalf("insert members: %v", err)
 	}
 	docID := fmt.Sprintf("doc_proj%d_intake", projectID)
 	insertDutyDoc(t, ctx, pool, projectID, docID, "intake", authorID, reviewerA, 0)
 
-	if _, err := pool.Exec(ctx, `
-		UPDATE project_members SET document_duty = '' WHERE project_id = $1 AND user_id = $2
-	`, projectID, reviewerA); err != nil {
+	tx, err := pool.Begin(ctx)
+	if err != nil {
 		t.Fatal(err)
 	}
-	if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{
+	if _, err := tx.Exec(ctx, `
+		UPDATE project_members SET document_duty = '' WHERE project_id = $1 AND user_id = $2
+	`, projectID, reviewerA); err != nil {
+		_ = tx.Rollback(ctx)
+		t.Fatal(err)
+	}
+	if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{
 		UserID: reviewerA, PreviousDuty: DutyReviewer, NextDuty: DutyNone,
-	}); err != nil {
+	})); err != nil {
+		_ = tx.Rollback(ctx)
 		t.Fatal(err)
 	}
-	assertDocReviewer(t, ctx, pool, docID, 0)
-	var authorGot int64
-	if err := pool.QueryRow(ctx, `SELECT COALESCE(author_id, 0) FROM document_records WHERE doc_id = $1`, docID).Scan(&authorGot); err != nil {
+	if err := tx.Rollback(ctx); err != nil {
 		t.Fatal(err)
 	}
-	if authorGot != authorID {
-		t.Fatalf("author must stay %d, got %d", authorID, authorGot)
-	}
+	assertDocReviewer(t, ctx, pool, docID, reviewerA)
+	assertMemberDuty(t, ctx, pool, projectID, reviewerA, DutyReviewer)
+	assertDocAuthorID(t, ctx, pool, docID, authorID)
 
-	if _, err := pool.Exec(ctx, `
+	tx, err = pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil {
+		_ = tx.Rollback(ctx)
+		t.Fatal(err)
+	}
+	prev, err := LoadMemberDutyTx(ctx, tx, projectID, reviewerA)
+	if err != nil {
+		_ = tx.Rollback(ctx)
+		t.Fatal(err)
+	}
+	if prev != DutyReviewer {
+		_ = tx.Rollback(ctx)
+		t.Fatalf("locked previous duty=%q", prev)
+	}
+	if _, err := tx.Exec(ctx, `
 		UPDATE project_members SET document_duty = '' WHERE project_id = $1 AND user_id = $2
 	`, projectID, reviewerA); err != nil {
+		_ = tx.Rollback(ctx)
 		t.Fatal(err)
 	}
-	if _, err := pool.Exec(ctx, `
-		UPDATE project_members SET document_duty = 'reviewer' WHERE project_id = $1 AND user_id = $2
-	`, projectID, reviewerB); err != nil {
+	if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{
+		UserID: reviewerA, PreviousDuty: prev, NextDuty: DutyNone,
+	})); err != nil {
+		_ = tx.Rollback(ctx)
 		t.Fatal(err)
 	}
-	if _, err := pool.Exec(ctx, `
-		UPDATE document_records SET reviewer_id = $2, reviewer_name = 'A' WHERE doc_id = $1
-	`, docID, reviewerA); err != nil {
-		t.Fatal(err)
-	}
-	if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{
-		UserID: reviewerB, PreviousDuty: DutyNone, NextDuty: DutyReviewer,
-		PriorHolderIDs: map[string]int64{DutyReviewer: reviewerA},
-	}); err != nil {
-		t.Fatal(err)
-	}
-	assertDocReviewer(t, ctx, pool, docID, reviewerB)
-
-	if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{
-		UserID: reviewerB, PreviousDuty: DutyReviewer, NextDuty: DutyReviewer,
-	}); err != nil {
-		t.Fatal(err)
-	}
-	assertDocReviewer(t, ctx, pool, docID, reviewerB)
-
-	if _, err := pool.Exec(ctx, `
-		UPDATE project_members SET status = 'exited', document_duty = 'reviewer' WHERE project_id = $1 AND user_id = $2
-	`, projectID, reviewerB); err != nil {
-		t.Fatal(err)
-	}
-	if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{
-		UserID: reviewerB, PreviousDuty: DutyReviewer, NextDuty: DutyReviewer, Exited: true,
-	}); err != nil {
-		t.Fatal(err)
-	}
-	assertDocReviewer(t, ctx, pool, docID, 0)
-	if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{
-		UserID: reviewerB, PreviousDuty: DutyReviewer, NextDuty: DutyReviewer, Exited: true,
-	}); err != nil {
+	if err := tx.Commit(ctx); err != nil {
 		t.Fatal(err)
 	}
 	assertDocReviewer(t, ctx, pool, docID, 0)
+	assertMemberDuty(t, ctx, pool, projectID, reviewerA, DutyNone)
+	assertDocAuthorID(t, ctx, pool, docID, authorID)
 }
 
 func TestOwnerAuthorBackfill_MultiOwnerDeterministic_PG(t *testing.T) {
@@ -407,6 +397,11 @@ func assertDocSlot(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID,
 	}
 }
 
+func assertDocAuthorID(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, want int64) {
+	t.Helper()
+	assertDocSlot(t, ctx, pool, docID, "author_id", want)
+}
+
 func assertDocAuthorName(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, want string) {
 	t.Helper()
 	var got string

internal/server/project_members_handlers.go

@@ -29,27 +29,7 @@ func writeDutySyncError(w http.ResponseWriter) {
 	http.Error(w, `{"error":"ERR_DUTY_SYNC","message":"成员已保存，但受控文档职责同步失败"}`, http.StatusInternalServerError)
 }
 
-func dutyChangeForRemove(existing *authz.ProjectMemberRecord, userID int64) docduty.MemberDutyChange {
-	change := docduty.MemberDutyChange{UserID: userID, Exited: true}
-	if existing != nil {
-		change.PreviousDuty = existing.DocumentDuty
-		change.NextDuty = existing.DocumentDuty
-	}
-	return change
-}
-
-func (s *Server) snapshotDutyHolders(ctx context.Context, projectID int64) map[string]int64 {
-	if s == nil || s.svc == nil || s.svc.Pool() == nil || projectID <= 0 {
-		return nil
-	}
-	members, _, err := docduty.LoadProjectMembers(ctx, s.svc.Pool(), projectID)
-	if err != nil {
-		return nil
-	}
-	return docduty.PriorDutyHolders(members)
-}
-
-func (s *Server) reconcileAndHealProjectDuties(ctx context.Context, projectID int64, change docduty.MemberDutyChange) error {
+func (s *Server) healProjectDuties(ctx context.Context, projectID int64) error {
 	if s == nil || s.svc == nil || projectID <= 0 {
 		return nil
 	}
@@ -57,9 +37,9 @@ func (s *Server) reconcileAndHealProjectDuties(ctx context.Context, projectID in
 	if pool == nil {
 		return nil
 	}
-	err := docduty.ReconcileMemberDuty(ctx, pool, projectID, change)
+	_, _, err := docduty.HealProjectDocuments(ctx, pool, projectID)
 	if err != nil {
-		log.Printf("[docduty] reconcile project %d after member change failed: %v", projectID, err)
+		log.Printf("[docduty] heal project %d after member change failed: %v", projectID, err)
 	}
 	return err
 }
@@ -137,7 +117,6 @@ func (s *Server) handleAddProjectMember(w http.ResponseWriter, r *http.Request,
 	if req.ProjectRole == "" {
 		req.ProjectRole = authz.RoleMember
 	}
-	priors := s.snapshotDutyHolders(r.Context(), projectID)
 	if err := s.authorizer.UpsertProjectMemberSpec(r.Context(), projectID, req.UserID, authz.MemberUpsert{
 		ProjectRole:  req.ProjectRole,
 		DocumentDuty: req.DocumentDuty,
@@ -145,13 +124,7 @@ func (s *Server) handleAddProjectMember(w http.ResponseWriter, r *http.Request,
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
-	nextDuty := ""
-	if req.DocumentDuty != nil {
-		nextDuty = *req.DocumentDuty
-	}
-	if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, docduty.MemberDutyChange{
-		UserID: req.UserID, NextDuty: nextDuty, PriorHolderIDs: priors,
-	}); err != nil {
+	if err := s.healProjectDuties(r.Context(), projectID); err != nil {
 		writeDutySyncError(w)
 		return
 	}
@@ -179,7 +152,7 @@ func (s *Server) handleUpdateProjectMember(w http.ResponseWriter, r *http.Reques
 		http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"project_role required with document_duty"}`, http.StatusBadRequest)
 		return
 	}
-	priors := s.snapshotDutyHolders(r.Context(), projectID)
+	wrote := false
 	if req.ProjectRole != "" || req.DocumentDuty != nil {
 		role, err := resolveUpdateProjectRole(req.ProjectRole, existing)
 		if err != nil {
@@ -193,28 +166,17 @@ func (s *Server) handleUpdateProjectMember(w http.ResponseWriter, r *http.Reques
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
+		wrote = true
 	}
 	if req.Status != "" {
 		if err := s.authorizer.SetProjectMemberStatus(r.Context(), projectID, req.UserID, req.Status); err != nil {
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
+		wrote = true
 	}
-	if req.ProjectRole != "" || req.DocumentDuty != nil || req.Status != "" {
-		nextDuty := ""
-		if existing != nil {
-			nextDuty = existing.DocumentDuty
-		}
-		if req.DocumentDuty != nil {
-			nextDuty = *req.DocumentDuty
-		}
-		if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, docduty.MemberDutyChange{
-			UserID:         req.UserID,
-			PreviousDuty:   existingDuty(existing),
-			NextDuty:       nextDuty,
-			Exited:         req.Status == authz.StatusExited,
-			PriorHolderIDs: priors,
-		}); err != nil {
+	if wrote {
+		if err := s.healProjectDuties(r.Context(), projectID); err != nil {
 			writeDutySyncError(w)
 			return
 		}
@@ -232,29 +194,13 @@ func (s *Server) handleRemoveProjectMember(w http.ResponseWriter, r *http.Reques
 		http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"user_id query required"}`, http.StatusBadRequest)
 		return
 	}
-	var existing *authz.ProjectMemberRecord
-	if s.authorizer != nil {
-		rec, err := s.authorizer.GetProjectMember(r.Context(), projectID, targetID)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
-		existing = rec
-	}
 	if err := s.authorizer.SetProjectMemberStatus(r.Context(), projectID, targetID, authz.StatusExited); err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, dutyChangeForRemove(existing, targetID)); err != nil {
+	if err := s.healProjectDuties(r.Context(), projectID); err != nil {
 		writeDutySyncError(w)
 		return
 	}
 	writeJSONResponse(w, http.StatusOK, map[string]bool{"success": true})
 }
-
-func existingDuty(existing *authz.ProjectMemberRecord) string {
-	if existing == nil {
-		return ""
-	}
-	return existing.DocumentDuty
-}

internal/server/project_members_role_test.go

@@ -29,13 +29,6 @@ func TestResolveUpdateProjectRole_RejectsMissingMember(t *testing.T) {
 	}
 }
 
-func TestDutyChangeForRemove_UsesExistingReviewerDuty(t *testing.T) {
-	got := dutyChangeForRemove(&authz.ProjectMemberRecord{DocumentDuty: "reviewer"}, 171)
-	if !got.Exited || got.UserID != 171 || got.PreviousDuty != "reviewer" || got.NextDuty != "reviewer" {
-		t.Fatalf("remove change: %+v", got)
-	}
-}
-
 func TestWriteDutySyncError_UsesExplicitCode(t *testing.T) {
 	rec := httptest.NewRecorder()
 	writeDutySyncError(rec)

scripts/run-isolated-duty-sync-pg-tests.sh

@@ -73,6 +73,6 @@ export TEST_DATABASE_URL="${TEST_URL}"
 echo "[info] running isolated duty-sync PostgreSQL tests"
 
 # 包串行，避免并行 goose 抢同一临时库。
-go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding|ReconcileMemberDuty'
-go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty'
-go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole|DutyChangeForRemove|WriteDutySyncError'
+go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding|ApplySlotRevocations'
+go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty|AtomicDutyRevoke|UpsertProjectMember'
+go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole|WriteDutySyncError'