diff --git a/.doc/ops/short-term/2026-09-14-document-reliability-pilot.md b/.doc/ops/short-term/2026-09-14-document-reliability-pilot.md index 16f787ec..bfb0bd10 100644 --- a/.doc/ops/short-term/2026-09-14-document-reliability-pilot.md +++ b/.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。 diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index f523063e..9b7830a8 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.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: diff --git a/internal/app/docflow/commands.go b/internal/app/docflow/commands.go new file mode 100644 index 00000000..f6500e22 --- /dev/null +++ b/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 +} diff --git a/internal/app/docflow/commands_test.go b/internal/app/docflow/commands_test.go new file mode 100644 index 00000000..7a08648f --- /dev/null +++ b/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) + } +} diff --git a/internal/app/uow/uow.go b/internal/app/uow/uow.go index e77267d6..6aa18a1c 100644 --- a/internal/app/uow/uow.go +++ b/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 +} diff --git a/internal/docgov/engine.go b/internal/docgov/engine.go index 28e017a4..cc8c55cc 100644 --- a/internal/docgov/engine.go +++ b/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 } diff --git a/internal/docgov/repository.go b/internal/docgov/repository.go index 8c1c31da..e14ae50a 100644 --- a/internal/docgov/repository.go +++ b/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) } diff --git a/internal/docrevision/document_commit_concurrency_pg_test.go b/internal/docrevision/document_commit_concurrency_pg_test.go new file mode 100644 index 00000000..15e0daa5 --- /dev/null +++ b/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) + } +} diff --git a/internal/docrevision/pg_store.go b/internal/docrevision/pg_store.go index 5541dbdc..8c8cf96e 100644 --- a/internal/docrevision/pg_store.go +++ b/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 diff --git a/internal/docrevision/service.go b/internal/docrevision/service.go index f0d08ec8..565f1e00 100644 --- a/internal/docrevision/service.go +++ b/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, }) diff --git a/internal/server/document_revision_handlers.go b/internal/server/document_revision_handlers.go index d7a3b1bc..1ce3ada7 100644 --- a/internal/server/document_revision_handlers.go +++ b/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()}) } diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 09bbe592..37eccceb 100644 --- a/internal/server/handlers.go +++ b/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 }