diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index ab8e9d2b..9b183132 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -13,6 +13,7 @@ imports: - docduty bizdev: - authz + - app/uow - config - ctxmgr - db @@ -30,6 +31,7 @@ imports: - storage - stream - workflowdef + - workspace chat: - app/chatflow - app/intentflow @@ -471,38 +473,6 @@ unwrap_owners: - app/intentflow exceptions: - - kind: sql_write - package: docduty - file: internal/docduty/revoke.go - symbol: document_records - owner: docduty - reason: duty slot revoke still writes reviewer/approver columns directly; fold into docgov repository during U3 - milestone: U3 - expires: "2026-09-28" - - kind: sql_write - package: docduty - file: internal/docduty/store.go - symbol: document_records - owner: docduty - reason: heal/store updates duty columns on document_records; move behind docgov repository in U3 - milestone: U3 - expires: "2026-09-28" - - kind: sql_write - package: docrevision - file: internal/docrevision/pg_store.go - symbol: document_records - owner: docrevision - reason: projectRootPointersTX updates controlled root pointers; U3 must keep this inside the shared UnitOfWork - milestone: U3 - expires: "2026-09-28" - - kind: sql_write - package: docrevision - file: internal/docrevision/pg_store.go - symbol: freeform_documents - owner: docrevision - reason: projectRootPointersTX also projects freeform heads; keep atomic in U3 then hand write ownership back - milestone: U3 - expires: "2026-09-28" - kind: sql_write package: projectinstance file: internal/projectinstance/store.go @@ -591,14 +561,6 @@ exceptions: reason: project doc helpers still construct output.md; U5 milestone: U5 expires: "2026-10-12" - - kind: path - package: docrevision - file: internal/docrevision/projection.go - symbol: output.md - owner: docrevision - reason: InspectProjection reads workspace file; U4 treats this as optional cache only - milestone: U4 - expires: "2026-09-28" - kind: path package: docrevision file: internal/docrevision/reconcile.go @@ -607,22 +569,6 @@ exceptions: reason: reconcile still repairs projection files; U5 milestone: U5 expires: "2026-10-12" - - kind: path - package: server - file: internal/server/document_revision_access.go - symbol: output.md - owner: server - reason: handler still reads workspace files; U4 must use docquery ledger - milestone: U4 - expires: "2026-09-28" - - kind: path - package: server - file: internal/server/document_revision_handlers.go - symbol: output.md - owner: server - reason: handler error text mentions output.md; U4/U5 remove path leakage - milestone: U4 - expires: "2026-09-28" - kind: path package: server file: internal/server/server.go @@ -642,8 +588,8 @@ exceptions: baseline_counts: import: 0 - sql_write: 5 - path: 16 + sql_write: 1 + path: 13 event: 0 # U12: kinds already at zero production findings fail --mode block. diff --git a/internal/app/app.go b/internal/app/app.go index 820a3eaa..0c441fe8 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -205,11 +205,13 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { coordEngine.SetImpactActions(traceChat{svc: traceSvc}) if pool != nil { access := authz.NewAuthorizer(pool) + wireDutySlotWriter(access) coordEngine.SetProjectAccess(func(ctx context.Context, projectID, userID int64) error { return access.RequireProjectAccess(ctx, projectID, userID) }) } if parts.Rev != nil { + wireDocRevisionPorts(parts.Rev) parts.Rev.CoverageResolver = func(ctx context.Context, docID string) string { projectID := projectIDFromControlledDoc(docID) if projectID == 0 { @@ -219,6 +221,8 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { } } + authorizer := authz.NewAuthorizer(pool) + wireDutySlotWriter(authorizer) rt := &Runtime{ Cfg: cfg, Biz: svc, @@ -226,7 +230,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { Auth: authSvc, PW: pwSvc, TaskRuns: taskMgr, - Authorizer: authz.NewAuthorizer(pool), + Authorizer: authorizer, Authz: authzSvc, WFDef: wfDefSvc, Inst: instSvc, diff --git a/internal/app/wire_g1.go b/internal/app/wire_g1.go new file mode 100644 index 00000000..55c64946 --- /dev/null +++ b/internal/app/wire_g1.go @@ -0,0 +1,42 @@ +package app + +import ( + "context" + + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/authz" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/freedoc" + "chat-agent-system/internal/workspace" +) + +func wireDutySlotWriter(a *authz.Authorizer) { + if a == nil { + return + } + a.SetDutySlotWriter(docgov.DutySlotWriter{}) +} + +func wireDocRevisionPorts(rev *docrevision.Service) { + if rev == nil { + return + } + rev.SetRootPointerProjector(docrevision.CompositeRootPointerProjector{ + Controlled: func(ctx context.Context, scope uow.TxScope, root *docrevision.DocRootPointers) error { + if root == nil { + return nil + } + return (docgov.Repository{}).ProjectHeadFull(ctx, scope, root.DocID, + root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer, root.ContentSHA256) + }, + Freeform: func(ctx context.Context, scope uow.TxScope, root *docrevision.DocRootPointers) error { + if root == nil { + return nil + } + return (freedoc.Repository{}).ProjectHead(ctx, scope, root.DocID, + root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer) + }, + }) + rev.SetDeliverableStore(workspace.DirDeliverable{}) +} diff --git a/internal/authz/authorizer.go b/internal/authz/authorizer.go index 0200fb55..0be560ff 100644 --- a/internal/authz/authorizer.go +++ b/internal/authz/authorizer.go @@ -52,15 +52,40 @@ type MemberUpsert struct { DocumentDuty *string } +// DutySlotWriter clears document_records reviewer/approver slots (implemented by docgov). +// authz must not import docgov; the composition root injects the adapter. +type DutySlotWriter interface { + ClearSlots(ctx context.Context, tx pgx.Tx, projectID int64, revs []docduty.SlotRevocation) error +} + // Authorizer 项目级对象授权。 type Authorizer struct { - pool *pgxpool.Pool + pool *pgxpool.Pool + dutySlots DutySlotWriter } func NewAuthorizer(pool *pgxpool.Pool) *Authorizer { return &Authorizer{pool: pool} } +// SetDutySlotWriter injects the document_records slot clearer (docgov adapter). +func (a *Authorizer) SetDutySlotWriter(w DutySlotWriter) { + if a == nil { + return + } + a.dutySlots = w +} + +func (a *Authorizer) clearDutySlots(ctx context.Context, tx pgx.Tx, projectID int64, revs []docduty.SlotRevocation) error { + if a == nil || len(revs) == 0 { + return nil + } + if a.dutySlots == nil { + return fmt.Errorf("authz: duty slot writer required") + } + return a.dutySlots.ClearSlots(ctx, tx, projectID, revs) +} + func (a *Authorizer) enabled() bool { return a != nil && a.pool != nil } @@ -327,7 +352,7 @@ func (a *Authorizer) upsertProjectMemberSpecOnce(ctx context.Context, projectID, defer func() { _ = tx.Rollback(ctx) }() if spec.DocumentDuty != nil { - if err := applyDutyWriteInTx(ctx, tx, projectID, userID, spec.ProjectRole, duty); err != nil { + if err := a.applyDutyWriteInTx(ctx, tx, projectID, userID, spec.ProjectRole, duty); err != nil { return err } return tx.Commit(ctx) @@ -396,7 +421,7 @@ func (a *Authorizer) setProjectMemberStatusOnce(ctx context.Context, projectID, return fmt.Errorf("authz: project member not found") } if status == StatusExited { - if err := docduty.ApplySlotRevocations(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + if err := a.clearDutySlots(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ UserID: userID, PreviousDuty: prevDuty, NextDuty: prevDuty, Exited: true, })); err != nil { return err @@ -406,7 +431,7 @@ func (a *Authorizer) setProjectMemberStatusOnce(ctx context.Context, projectID, } // applyDutyWriteInTx 假定调用方已 Begin:lock → 读旧/冲突 holder → 独占写成员 → 清槽。 -func applyDutyWriteInTx(ctx context.Context, tx pgx.Tx, projectID, userID int64, projectRole, duty string) error { +func (a *Authorizer) applyDutyWriteInTx(ctx context.Context, tx pgx.Tx, projectID, userID int64, projectRole, duty string) error { if err := docduty.AcquireProjectDutyLock(ctx, tx, projectID); err != nil { return err } @@ -441,7 +466,7 @@ func applyDutyWriteInTx(ctx context.Context, tx pgx.Tx, projectID, userID int64, return err } - return docduty.ApplySlotRevocations(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + return a.clearDutySlots(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ UserID: userID, PreviousDuty: prevDuty, NextDuty: duty, diff --git a/internal/authz/authorizer_pg_test.go b/internal/authz/authorizer_pg_test.go index 1901ec7f..30a96f43 100644 --- a/internal/authz/authorizer_pg_test.go +++ b/internal/authz/authorizer_pg_test.go @@ -11,6 +11,7 @@ import ( "chat-agent-system/internal/db" "chat-agent-system/internal/docduty" + "chat-agent-system/internal/docgov" ) func TestUpsertProjectMemberDuty_ConcurrentLastWriter_PG(t *testing.T) { @@ -52,6 +53,7 @@ func TestUpsertProjectMemberDuty_ConcurrentLastWriter_PG(t *testing.T) { }) a := NewAuthorizer(pool) + a.SetDutySlotWriter(docgov.DutySlotWriter{}) dutyAuthor := docduty.DutyAuthor if err := a.UpsertProjectMemberSpec(ctx, projectID, ownerID, MemberUpsert{ ProjectRole: RoleOwner, DocumentDuty: &dutyAuthor, @@ -155,6 +157,7 @@ func TestUpsertProjectMember_AtomicDutyRevoke_PG(t *testing.T) { }) a := NewAuthorizer(pool) + a.SetDutySlotWriter(docgov.DutySlotWriter{}) dutyAuthor := docduty.DutyAuthor dutyReviewer := docduty.DutyReviewer dutyNone := docduty.DutyNone @@ -205,7 +208,7 @@ func TestUpsertProjectMember_AtomicDutyRevoke_PG(t *testing.T) { }); err != nil { t.Fatal(err) } - if _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID); err != nil { + if _, _, err := docgov.HealProjectDocuments(ctx, pool, projectID); err != nil { t.Fatal(err) } assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerA) @@ -220,7 +223,7 @@ func TestUpsertProjectMember_AtomicDutyRevoke_PG(t *testing.T) { 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 { + if _, _, err := docgov.HealProjectDocuments(ctx, pool, projectID); err != nil { t.Fatal(err) } assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerB) diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index ba88de00..79526335 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -14,6 +14,7 @@ import ( "strings" "time" + "chat-agent-system/internal/app/uow" "chat-agent-system/internal/authz" "chat-agent-system/internal/config" "chat-agent-system/internal/ctxmgr" @@ -30,6 +31,7 @@ import ( "chat-agent-system/internal/storage" "chat-agent-system/internal/stream" "chat-agent-system/internal/workflowdef" + "chat-agent-system/internal/workspace" "github.com/jackc/pgx/v5/pgxpool" ) @@ -107,6 +109,7 @@ func NewService(cfg *config.Config) (*Service, error) { } } else { authzStore := authz.NewAuthorizer(pool) + authzStore.SetDutySlotWriter(docgov.DutySlotWriter{}) _ = authzStore.BackfillCreatorMembers(poolCtx) if ver, verErr := db.CurrentVersion(poolCtx, pool); verErr == nil { schemaVersion = ver @@ -127,6 +130,17 @@ func NewService(cfg *config.Config) (*Service, error) { docGov := docgov.NewGovernanceEngine(pool) docRev := docrevision.NewService(pool) docRev.DataDir = cfg.DataDir + docRev.SetRootPointerProjector(docrevision.CompositeRootPointerProjector{ + Controlled: func(ctx context.Context, scope uow.TxScope, root *docrevision.DocRootPointers) error { + return (docgov.Repository{}).ProjectHeadFull(ctx, scope, root.DocID, + root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer, root.ContentSHA256) + }, + Freeform: func(ctx context.Context, scope uow.TxScope, root *docrevision.DocRootPointers) error { + return (freedoc.Repository{}).ProjectHead(ctx, scope, root.DocID, + root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer) + }, + }) + docRev.SetDeliverableStore(workspace.DirDeliverable{}) freeDoc := freedoc.NewService(pool, ctxMgr) freeDoc.SetDocRevision(docRev) pwStore := projectworkflow.NewStore(pool) @@ -667,7 +681,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun docRec.DutyPolicy = dutyPolicy } if s.pool != nil && req.ProjectID > 0 { - assigned, _, healErr := docduty.HealProjectDocuments(ctx, s.pool, req.ProjectID) + assigned, _, healErr := docgov.HealProjectDocuments(ctx, s.pool, req.ProjectID) if healErr != nil { log.Printf("[docduty] heal project %d after generate failed: %v", req.ProjectID, healErr) return &TaskRunResult{ diff --git a/internal/docduty/duty_locked_pg_test.go b/internal/docduty/duty_locked_pg_test.go index 59a83306..5c0ccb83 100644 --- a/internal/docduty/duty_locked_pg_test.go +++ b/internal/docduty/duty_locked_pg_test.go @@ -1,6 +1,8 @@ -package docduty +package docduty_test import ( + "chat-agent-system/internal/docduty" + "chat-agent-system/internal/docgov" "context" "fmt" "testing" @@ -52,7 +54,7 @@ func TestHealProjectDocuments_SkipsDutyLocked_PG(t *testing.T) { // 未锁定文档:错误槽位,应被 heal 纠正 insertDutyDoc(t, ctx, pool, projectID, openDoc, "research", authorID, 0, 0) - _, n, err := HealProjectDocuments(ctx, pool, projectID) + _, n, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { t.Fatalf("heal: %v", err) } @@ -109,8 +111,8 @@ func TestApplySlotRevocations_SkipsDutyLocked_PG(t *testing.T) { if err != nil { t.Fatal(err) } - if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{ - UserID: reviewerID, PreviousDuty: DutyReviewer, NextDuty: DutyNone, + if err := docgov.ClearDutySlots(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + UserID: reviewerID, PreviousDuty: docduty.DutyReviewer, NextDuty: docduty.DutyNone, })); err != nil { _ = tx.Rollback(ctx) t.Fatal(err) diff --git a/internal/docduty/revoke.go b/internal/docduty/revoke.go index 07017df6..c179f15a 100644 --- a/internal/docduty/revoke.go +++ b/internal/docduty/revoke.go @@ -115,6 +115,11 @@ const projectDutyLockSQL = `SELECT pg_advisory_xact_lock($1)` // projectDutyLockHook 仅测试使用:在申请锁之前同步,便于证明 heal 会阻塞在成员写事务之后。 var projectDutyLockHook func() +// SetProjectDutyLockHookForTest installs a pre-lock hook for PG concurrency tests. +func SetProjectDutyLockHookForTest(fn func()) { + projectDutyLockHook = fn +} + // AcquireProjectDutyLock 获取项目职责互斥锁;调用方须已 Begin,锁随事务结束释放。 func AcquireProjectDutyLock(ctx context.Context, tx pgx.Tx, projectID int64) error { if tx == nil || projectID <= 0 { @@ -128,40 +133,3 @@ func AcquireProjectDutyLock(ctx context.Context, tx pgx.Tx, projectID int64) err } return 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 - } - for _, rev := range revs { - q, ok := clearDutySlotSQL(rev.Duty) - if !ok { - continue - } - if _, err := tx.Exec(ctx, q, projectID, rev.UserID); err != nil { - return fmt.Errorf("docduty: clear %s for user %d: %w", rev.Duty, rev.UserID, err) - } - } - return nil -} - -func clearDutySlotSQL(duty string) (string, bool) { - switch duty { - case DutyReviewer: - return ` - UPDATE document_records - SET reviewer_id = NULL, reviewer_name = '', updated_at = now() - WHERE project_id = $1 AND COALESCE(task_key, '') <> '' AND reviewer_id = $2 AND NOT duty_locked - `, true - case DutyApprover: - return ` - UPDATE document_records - SET approver_id = NULL, approver_name = '', updated_at = now() - WHERE project_id = $1 AND COALESCE(task_key, '') <> '' AND approver_id = $2 AND NOT duty_locked - `, true - default: - return "", false - } -} diff --git a/internal/docduty/store.go b/internal/docduty/store.go index da12502f..9edd4246 100644 --- a/internal/docduty/store.go +++ b/internal/docduty/store.go @@ -10,8 +10,11 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) -// dutyDB 覆盖 pool 与 tx,供同一套解析/回填 SQL 复用。 -type dutyDB interface { +// Compile-time check: pool satisfies DB. +var _ DB = (*pgxpool.Pool)(nil) + +// DB covers pool and tx for shared member/resolve reads. +type DB interface { Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error) @@ -22,10 +25,18 @@ func LoadProjectMembers(ctx context.Context, pool *pgxpool.Pool, projectID int64 if pool == nil || projectID <= 0 { return nil, Holder{}, nil } - return loadProjectMembers(ctx, pool, projectID) + return LoadProjectMembersDB(ctx, pool, projectID) } -func loadProjectMembers(ctx context.Context, db dutyDB, projectID int64) ([]Member, Holder, error) { +// LoadProjectMembersDB reads members via pool or an open transaction. +func LoadProjectMembersDB(ctx context.Context, db DB, projectID int64) ([]Member, Holder, error) { + if db == nil || projectID <= 0 { + return nil, Holder{}, nil + } + return loadProjectMembers(ctx, db, projectID) +} + +func loadProjectMembers(ctx context.Context, db DB, projectID int64) ([]Member, Holder, error) { var creator Holder err := db.QueryRow(ctx, ` SELECT COALESCE(p.creator_user_id, 0), COALESCE(u.name, '') @@ -72,135 +83,11 @@ func ResolveProject(ctx context.Context, pool *pgxpool.Pool, projectID int64, re return Resolve(members, creator, req), nil } -func resolveProjectTx(ctx context.Context, tx pgx.Tx, projectID int64, req RequestDuties) (Binding, error) { +// ResolveProjectTx resolves project document duties inside the caller's transaction. +func ResolveProjectTx(ctx context.Context, tx pgx.Tx, projectID int64, req RequestDuties) (Binding, error) { members, creator, err := loadProjectMembers(ctx, tx, projectID) if err != nil { return Binding{}, err } return Resolve(members, creator, req), nil } - -type storedDocDuties struct { - DocID string - Author Holder - Reviewer Holder - Approver Holder -} - -// BackfillMissingDocumentDuties 按项目 assigned 职责对齐受控文档;assigned 为空的槽位不臆造、不清空。 -func BackfillMissingDocumentDuties(ctx context.Context, pool *pgxpool.Pool, projectID int64, assigned Binding) (int, error) { - if pool == nil || projectID <= 0 { - return 0, nil - } - return backfillMissingDocumentDuties(ctx, pool, projectID, assigned) -} - -func backfillMissingDocumentDuties(ctx context.Context, db dutyDB, projectID int64, assigned Binding) (int, error) { - members, creator, err := loadProjectMembers(ctx, db, projectID) - if err != nil { - return 0, err - } - memberIDs := ActiveMemberIDs(members, creator) - - rows, err := db.Query(ctx, ` - SELECT doc_id, - COALESCE(author_id, 0), COALESCE(author_name, ''), - COALESCE(reviewer_id, 0), COALESCE(reviewer_name, ''), - COALESCE(approver_id, 0), COALESCE(approver_name, '') - FROM document_records - WHERE project_id = $1 AND COALESCE(task_key, '') <> '' AND NOT duty_locked - `, projectID) - if err != nil { - return 0, fmt.Errorf("docduty: list documents: %w", err) - } - defer rows.Close() - - pending := make([]storedDocDuties, 0) - for rows.Next() { - var rec storedDocDuties - if err := rows.Scan( - &rec.DocID, - &rec.Author.UserID, &rec.Author.Name, - &rec.Reviewer.UserID, &rec.Reviewer.Name, - &rec.Approver.UserID, &rec.Approver.Name, - ); err != nil { - return 0, err - } - current := Binding{Author: rec.Author, Reviewer: rec.Reviewer, Approver: rec.Approver} - next, changed := HealBinding(current, assigned, memberIDs) - if !changed { - continue - } - rec.Author, rec.Reviewer, rec.Approver = next.Author, next.Reviewer, next.Approver - pending = append(pending, rec) - } - if err := rows.Err(); err != nil { - return 0, err - } - - updated := 0 - for _, rec := range pending { - tag, err := db.Exec(ctx, ` - UPDATE document_records - SET author_id = $2, author_name = $3, - reviewer_id = $4, reviewer_name = $5, - approver_id = $6, approver_name = $7, - updated_at = now() - WHERE doc_id = $1 AND project_id = $8 - `, rec.DocID, nullIfZero(rec.Author.UserID), rec.Author.Name, - nullIfZero(rec.Reviewer.UserID), rec.Reviewer.Name, - nullIfZero(rec.Approver.UserID), rec.Approver.Name, - projectID) - if err != nil { - return updated, fmt.Errorf("docduty: backfill %s: %w", rec.DocID, err) - } - if tag.RowsAffected() > 0 { - updated++ - } - } - return updated, nil -} - -func nullIfZero(id int64) any { - if id <= 0 { - return nil - } - return id -} - -// HealProjectDocuments 在同一事务内:项目 advisory lock、解析成员职责、读写 document_records。 -// 锁键/顺序与成员写事务一致。第二次调用应为 0 行。错误回滚,不得被调用方静默吞掉。 -func HealProjectDocuments(ctx context.Context, pool *pgxpool.Pool, projectID int64) (Binding, int, error) { - if pool == nil || projectID <= 0 { - return Binding{}, 0, nil - } - tx, err := pool.Begin(ctx) - if err != nil { - return Binding{}, 0, fmt.Errorf("docduty: begin heal: %w", err) - } - defer func() { _ = tx.Rollback(ctx) }() - - if err := AcquireProjectDutyLock(ctx, tx, projectID); err != nil { - return Binding{}, 0, err - } - assigned, n, err := healProjectDocumentsTx(ctx, tx, projectID) - if err != nil { - return Binding{}, 0, err - } - if err := tx.Commit(ctx); err != nil { - return Binding{}, 0, fmt.Errorf("docduty: commit heal: %w", err) - } - return assigned, n, nil -} - -func healProjectDocumentsTx(ctx context.Context, tx pgx.Tx, projectID int64) (Binding, int, error) { - assigned, err := resolveProjectTx(ctx, tx, projectID, RequestDuties{}) - if err != nil { - return Binding{}, 0, err - } - n, err := backfillMissingDocumentDuties(ctx, tx, projectID, assigned) - if err != nil { - return Binding{}, 0, err - } - return assigned, n, nil -} diff --git a/internal/docduty/store_pg_test.go b/internal/docduty/store_pg_test.go index 8396ad27..f1b717a9 100644 --- a/internal/docduty/store_pg_test.go +++ b/internal/docduty/store_pg_test.go @@ -1,6 +1,8 @@ -package docduty +package docduty_test import ( + "chat-agent-system/internal/docduty" + "chat-agent-system/internal/docgov" "context" "fmt" "runtime" @@ -77,7 +79,7 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { t.Fatalf("insert members: %v", err) } - bound, err := ResolveProject(ctx, pool, projectID, RequestDuties{ReviewerID: 2, ReviewerName: "审核负责人"}) + bound, err := docduty.ResolveProject(ctx, pool, projectID, docduty.RequestDuties{ReviewerID: 2, ReviewerName: "审核负责人"}) if err != nil { t.Fatal(err) } @@ -93,7 +95,7 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { insertDutyDoc(t, ctx, pool, projectID, intakeID, "intake", authorID, placeholderID, wrongApproverID) insertDutyDoc(t, ctx, pool, projectID, secondID, "prelim_design", authorID, placeholderID, wrongApproverID) - assigned, n, err := HealProjectDocuments(ctx, pool, projectID) + assigned, n, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { t.Fatal(err) } @@ -109,7 +111,7 @@ func TestProjectReviewerDutySync_PG(t *testing.T) { assertDocApprover(t, ctx, pool, secondID, assignedApproverID) assertDocAuthorName(t, ctx, pool, intakeID, "用户author") - _, n2, err := HealProjectDocuments(ctx, pool, projectID) + _, n2, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { t.Fatal(err) } @@ -153,7 +155,7 @@ func TestHealBinding_EmptyAssignedKeepsMember_PG(t *testing.T) { t.Fatal(err) } - _, n, err := HealProjectDocuments(ctx, pool, projectID) + _, n, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { t.Fatal(err) } @@ -197,7 +199,7 @@ func TestHealBinding_EmptyAssignedClearsExited_PG(t *testing.T) { t.Fatal(err) } - _, n, err := HealProjectDocuments(ctx, pool, projectID) + _, n, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { t.Fatal(err) } @@ -248,8 +250,8 @@ func TestApplySlotRevocations_SameTxnCommitAndRollback_PG(t *testing.T) { _ = tx.Rollback(ctx) t.Fatal(err) } - if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{ - UserID: reviewerA, PreviousDuty: DutyReviewer, NextDuty: DutyNone, + if err := docgov.ClearDutySlots(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + UserID: reviewerA, PreviousDuty: docduty.DutyReviewer, NextDuty: docduty.DutyNone, })); err != nil { _ = tx.Rollback(ctx) t.Fatal(err) @@ -258,23 +260,23 @@ func TestApplySlotRevocations_SameTxnCommitAndRollback_PG(t *testing.T) { t.Fatal(err) } assertDocReviewer(t, ctx, pool, docID, reviewerA) - assertMemberDuty(t, ctx, pool, projectID, reviewerA, DutyReviewer) + assertMemberDuty(t, ctx, pool, projectID, reviewerA, docduty.DutyReviewer) assertDocAuthorID(t, ctx, pool, docID, authorID) tx, err = pool.Begin(ctx) if err != nil { t.Fatal(err) } - if err := AcquireProjectDutyLock(ctx, tx, projectID); err != nil { + if err := docduty.AcquireProjectDutyLock(ctx, tx, projectID); err != nil { _ = tx.Rollback(ctx) t.Fatal(err) } - prev, err := LoadMemberDutyTx(ctx, tx, projectID, reviewerA) + prev, err := docduty.LoadMemberDutyTx(ctx, tx, projectID, reviewerA) if err != nil { _ = tx.Rollback(ctx) t.Fatal(err) } - if prev != DutyReviewer { + if prev != docduty.DutyReviewer { _ = tx.Rollback(ctx) t.Fatalf("locked previous duty=%q", prev) } @@ -284,8 +286,8 @@ func TestApplySlotRevocations_SameTxnCommitAndRollback_PG(t *testing.T) { _ = tx.Rollback(ctx) t.Fatal(err) } - if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{ - UserID: reviewerA, PreviousDuty: prev, NextDuty: DutyNone, + if err := docgov.ClearDutySlots(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + UserID: reviewerA, PreviousDuty: prev, NextDuty: docduty.DutyNone, })); err != nil { _ = tx.Rollback(ctx) t.Fatal(err) @@ -294,7 +296,7 @@ func TestApplySlotRevocations_SameTxnCommitAndRollback_PG(t *testing.T) { t.Fatal(err) } assertDocReviewer(t, ctx, pool, docID, 0) - assertMemberDuty(t, ctx, pool, projectID, reviewerA, DutyNone) + assertMemberDuty(t, ctx, pool, projectID, reviewerA, docduty.DutyNone) assertDocAuthorID(t, ctx, pool, docID, authorID) } @@ -328,17 +330,17 @@ func TestOwnerAuthorBackfill_MultiOwnerDeterministic_PG(t *testing.T) { t.Fatalf("insert owners: %v", err) } - if _, err := pool.Exec(ctx, OwnerAuthorBackfillOneProjectSQL, projectID); err != nil { + if _, err := pool.Exec(ctx, docduty.OwnerAuthorBackfillOneProjectSQL, projectID); err != nil { t.Fatalf("owner author backfill: %v", err) } - assertMemberDuty(t, ctx, pool, projectID, creatorOwner, DutyAuthor) - assertMemberDuty(t, ctx, pool, projectID, lowOwner, DutyNone) + assertMemberDuty(t, ctx, pool, projectID, creatorOwner, docduty.DutyAuthor) + assertMemberDuty(t, ctx, pool, projectID, lowOwner, docduty.DutyNone) - if _, err := pool.Exec(ctx, OwnerAuthorBackfillOneProjectSQL, projectID); err != nil { + if _, err := pool.Exec(ctx, docduty.OwnerAuthorBackfillOneProjectSQL, projectID); err != nil { t.Fatalf("second backfill: %v", err) } - assertMemberDuty(t, ctx, pool, projectID, creatorOwner, DutyAuthor) - assertMemberDuty(t, ctx, pool, projectID, lowOwner, DutyNone) + assertMemberDuty(t, ctx, pool, projectID, creatorOwner, docduty.DutyAuthor) + assertMemberDuty(t, ctx, pool, projectID, lowOwner, docduty.DutyNone) } func TestHealProjectDocuments_TransferDoesNotResurrectOldReviewer_PG(t *testing.T) { @@ -357,7 +359,7 @@ func TestHealProjectDocuments_TransferDoesNotResurrectOldReviewer_PG(t *testing. t.Fatalf("insert project: %v", err) } t.Cleanup(func() { - projectDutyLockHook = nil + docduty.SetProjectDutyLockHookForTest(nil) _, _ = 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) @@ -379,42 +381,42 @@ func TestHealProjectDocuments_TransferDoesNotResurrectOldReviewer_PG(t *testing. t.Fatal(err) } defer func() { _ = tx.Rollback(ctx) }() - if err := AcquireProjectDutyLock(ctx, tx, projectID); err != nil { + if err := docduty.AcquireProjectDutyLock(ctx, tx, projectID); err != nil { t.Fatal(err) } 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, DutyReviewer, reviewerB, StatusActive); err != nil { + `, projectID, docduty.DutyReviewer, reviewerB, docduty.StatusActive); err != nil { t.Fatal(err) } if _, err := tx.Exec(ctx, ` UPDATE project_members SET document_duty = $3, updated_at = now() WHERE project_id = $1 AND user_id = $2 - `, projectID, reviewerB, DutyReviewer); err != nil { + `, projectID, reviewerB, docduty.DutyReviewer); err != nil { t.Fatal(err) } - if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{ - UserID: reviewerB, PreviousDuty: DutyNone, NextDuty: DutyReviewer, - PriorHolderIDs: map[string]int64{DutyReviewer: reviewerA}, + if err := docgov.ClearDutySlots(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + UserID: reviewerB, PreviousDuty: docduty.DutyNone, NextDuty: docduty.DutyReviewer, + PriorHolderIDs: map[string]int64{docduty.DutyReviewer: reviewerA}, })); err != nil { t.Fatal(err) } aboutToLock := make(chan struct{}) - projectDutyLockHook = func() { + docduty.SetProjectDutyLockHookForTest(func() { close(aboutToLock) - projectDutyLockHook = nil - } + docduty.SetProjectDutyLockHookForTest(nil) + }) type healResult struct { - assigned Binding + assigned docduty.Binding n int err error } done := make(chan healResult, 1) go func() { - assigned, n, err := HealProjectDocuments(ctx, pool, projectID) + assigned, n, err := docgov.HealProjectDocuments(ctx, pool, projectID) done <- healResult{assigned: assigned, n: n, err: err} }() @@ -443,7 +445,7 @@ func TestHealProjectDocuments_TransferDoesNotResurrectOldReviewer_PG(t *testing. t.Fatal("old reviewer A must not be assigned after transfer") } - _, n2, err := HealProjectDocuments(ctx, pool, projectID) + _, n2, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { t.Fatal(err) } @@ -567,3 +569,10 @@ func assertMemberDuty(t *testing.T, ctx context.Context, pool *pgxpool.Pool, pro t.Fatalf("user %d document_duty=%q want %q", userID, got, want) } } + +func nullIfZero(id int64) any { + if id <= 0 { + return nil + } + return id +} diff --git a/internal/docduty/duty_locked_filter_test.go b/internal/docgov/duty_locked_filter_test.go similarity index 69% rename from internal/docduty/duty_locked_filter_test.go rename to internal/docgov/duty_locked_filter_test.go index a7d17773..e4c0279f 100644 --- a/internal/docduty/duty_locked_filter_test.go +++ b/internal/docgov/duty_locked_filter_test.go @@ -1,12 +1,14 @@ -package docduty +package docgov import ( "strings" "testing" + + "chat-agent-system/internal/docduty" ) func TestClearDutySlotSQL_SkipsLockedRows(t *testing.T) { - for _, duty := range []string{DutyReviewer, DutyApprover} { + for _, duty := range []string{docduty.DutyReviewer, docduty.DutyApprover} { sql, ok := clearDutySlotSQL(duty) if !ok { t.Fatalf("%s: expected sql", duty) diff --git a/internal/docgov/duty_slots.go b/internal/docgov/duty_slots.go new file mode 100644 index 00000000..094a6a79 --- /dev/null +++ b/internal/docgov/duty_slots.go @@ -0,0 +1,194 @@ +package docgov + +import ( + "context" + "fmt" + + "chat-agent-system/internal/docduty" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// ClearDutySlots clears reviewer/approver columns on document_records for the given revocations. +// Author columns are never written. Skips duty_locked rows. +func ClearDutySlots(ctx context.Context, tx pgx.Tx, projectID int64, revs []docduty.SlotRevocation) error { + if tx == nil || projectID <= 0 { + return nil + } + for _, rev := range revs { + q, ok := clearDutySlotSQL(rev.Duty) + if !ok { + continue + } + if _, err := tx.Exec(ctx, q, projectID, rev.UserID); err != nil { + return fmt.Errorf("docgov: clear %s for user %d: %w", rev.Duty, rev.UserID, err) + } + } + return nil +} + +func clearDutySlotSQL(duty string) (string, bool) { + switch duty { + case docduty.DutyReviewer: + return ` + UPDATE document_records + SET reviewer_id = NULL, reviewer_name = '', updated_at = now() + WHERE project_id = $1 AND COALESCE(task_key, '') <> '' AND reviewer_id = $2 AND NOT duty_locked + `, true + case docduty.DutyApprover: + return ` + UPDATE document_records + SET approver_id = NULL, approver_name = '', updated_at = now() + WHERE project_id = $1 AND COALESCE(task_key, '') <> '' AND approver_id = $2 AND NOT duty_locked + `, true + default: + return "", false + } +} + +// DutySlotWriter adapts ClearDutySlots for authz injection (authz must not import docgov). +type DutySlotWriter struct{} + +func (DutySlotWriter) ClearSlots(ctx context.Context, tx pgx.Tx, projectID int64, revs []docduty.SlotRevocation) error { + return ClearDutySlots(ctx, tx, projectID, revs) +} + +type storedDocDuties struct { + DocID string + Author docduty.Holder + Reviewer docduty.Holder + Approver docduty.Holder +} + +// BackfillDocumentDuties aligns controlled document duty columns to assigned project duties. +// Empty assigned slots are not invented or cleared. +func BackfillDocumentDuties(ctx context.Context, db docduty.DB, projectID int64, assigned docduty.Binding) (int, error) { + if db == nil || projectID <= 0 { + return 0, nil + } + members, creator, err := docduty.LoadProjectMembersDB(ctx, db, projectID) + if err != nil { + return 0, err + } + memberIDs := docduty.ActiveMemberIDs(members, creator) + + rows, err := db.Query(ctx, ` + SELECT doc_id, + COALESCE(author_id, 0), COALESCE(author_name, ''), + COALESCE(reviewer_id, 0), COALESCE(reviewer_name, ''), + COALESCE(approver_id, 0), COALESCE(approver_name, '') + FROM document_records + WHERE project_id = $1 AND COALESCE(task_key, '') <> '' AND NOT duty_locked + `, projectID) + if err != nil { + return 0, fmt.Errorf("docgov: list documents: %w", err) + } + defer rows.Close() + + pending := make([]storedDocDuties, 0) + for rows.Next() { + var rec storedDocDuties + if err := rows.Scan( + &rec.DocID, + &rec.Author.UserID, &rec.Author.Name, + &rec.Reviewer.UserID, &rec.Reviewer.Name, + &rec.Approver.UserID, &rec.Approver.Name, + ); err != nil { + return 0, err + } + current := docduty.Binding{Author: rec.Author, Reviewer: rec.Reviewer, Approver: rec.Approver} + next, changed := docduty.HealBinding(current, assigned, memberIDs) + if !changed { + continue + } + rec.Author, rec.Reviewer, rec.Approver = next.Author, next.Reviewer, next.Approver + pending = append(pending, rec) + } + if err := rows.Err(); err != nil { + return 0, err + } + + updated := 0 + for _, rec := range pending { + tag, err := db.Exec(ctx, ` + UPDATE document_records + SET author_id = $2, author_name = $3, + reviewer_id = $4, reviewer_name = $5, + approver_id = $6, approver_name = $7, + updated_at = now() + WHERE doc_id = $1 AND project_id = $8 + `, rec.DocID, nullIfZeroDuty(rec.Author.UserID), rec.Author.Name, + nullIfZeroDuty(rec.Reviewer.UserID), rec.Reviewer.Name, + nullIfZeroDuty(rec.Approver.UserID), rec.Approver.Name, + projectID) + if err != nil { + return updated, fmt.Errorf("docgov: backfill %s: %w", rec.DocID, err) + } + if tag.RowsAffected() > 0 { + updated++ + } + } + return updated, nil +} + +func nullIfZeroDuty(id int64) any { + if id <= 0 { + return nil + } + return id +} + +// HealProjectDocuments runs project advisory lock, duty resolve, and document_records backfill in one txn. +func HealProjectDocuments(ctx context.Context, pool *pgxpool.Pool, projectID int64) (docduty.Binding, int, error) { + if pool == nil || projectID <= 0 { + return docduty.Binding{}, 0, nil + } + tx, err := pool.Begin(ctx) + if err != nil { + return docduty.Binding{}, 0, fmt.Errorf("docgov: begin heal: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + if err := docduty.AcquireProjectDutyLock(ctx, tx, projectID); err != nil { + return docduty.Binding{}, 0, err + } + assigned, n, err := healProjectDocumentsTx(ctx, tx, projectID) + if err != nil { + return docduty.Binding{}, 0, err + } + if err := tx.Commit(ctx); err != nil { + return docduty.Binding{}, 0, fmt.Errorf("docgov: commit heal: %w", err) + } + return assigned, n, nil +} + +func healProjectDocumentsTx(ctx context.Context, tx pgx.Tx, projectID int64) (docduty.Binding, int, error) { + assigned, err := docduty.ResolveProjectTx(ctx, tx, projectID, docduty.RequestDuties{}) + if err != nil { + return docduty.Binding{}, 0, err + } + n, err := (Repository{}).BackfillDuties(ctx, tx, projectID, assigned) + if err != nil { + return docduty.Binding{}, 0, err + } + return assigned, n, nil +} + +// HealProjectDocuments aligns controlled document duties for a project under the governance engine. +func (e *GovernanceEngine) HealProjectDocuments(ctx context.Context, projectID int64) (docduty.Binding, int, error) { + if e == nil { + return docduty.Binding{}, 0, nil + } + return HealProjectDocuments(ctx, e.pool, projectID) +} + +// ClearDutySlots is the repository write port for reviewer/approver slot revocation. +func (Repository) ClearDutySlots(ctx context.Context, tx pgx.Tx, projectID int64, revs []docduty.SlotRevocation) error { + return ClearDutySlots(ctx, tx, projectID, revs) +} + +// BackfillDuties is the repository write port for duty column healing. +func (Repository) BackfillDuties(ctx context.Context, tx pgx.Tx, projectID int64, assigned docduty.Binding) (int, error) { + return BackfillDocumentDuties(ctx, tx, projectID, assigned) +} diff --git a/internal/docgov/engine_db_test.go b/internal/docgov/engine_db_test.go index 1fbf869f..b3e95fd2 100644 --- a/internal/docgov/engine_db_test.go +++ b/internal/docgov/engine_db_test.go @@ -117,7 +117,29 @@ func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { engine := docgov.NewGovernanceEngine(pool) testDocID := fmt.Sprintf("doc_opt_duty_%d", time.Now().UnixNano()) - projectID := time.Now().UnixNano()%900000 + 100000 + suffix := time.Now().UnixNano() + + var authorID int64 + if err := pool.QueryRow(ctx, ` + INSERT INTO users (account_id, name, role) + VALUES ($1, $2, 'engineer') + RETURNING id + `, fmt.Sprintf("opt_duty_%d", suffix), "系统管理员").Scan(&authorID); err != nil { + t.Fatalf("insert author: %v", err) + } + 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("opt-duty-%d", suffix), fmt.Sprintf("optional duties %d", suffix), authorID).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE doc_id = $1`, testDocID) + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, authorID) + }) doc, err := engine.CreateOrGetDocument( ctx, @@ -127,7 +149,7 @@ func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { "可选职责空审核人", "V0.5_feasibility", "", - 9, "系统管理员", + authorID, "系统管理员", 0, "", 0, "", ) @@ -139,22 +161,20 @@ func TestDocGov_CreateOrGetDocument_OptionalDutiesNullFK(t *testing.T) { } var reviewerNull, approverNull bool - var authorID int64 + var gotAuthor int64 err = pool.QueryRow(ctx, ` SELECT author_id, reviewer_id IS NULL, approver_id IS NULL FROM document_records WHERE doc_id = $1 - `, testDocID).Scan(&authorID, &reviewerNull, &approverNull) + `, testDocID).Scan(&gotAuthor, &reviewerNull, &approverNull) if err != nil { t.Fatalf("query document_records: %v", err) } - if authorID != 9 { - t.Fatalf("author_id=%d want 9", authorID) + if gotAuthor != authorID { + t.Fatalf("author_id=%d want %d", gotAuthor, authorID) } if !reviewerNull || !approverNull { t.Fatalf("expected NULL reviewer/approver, got reviewerNull=%v approverNull=%v", reviewerNull, approverNull) } - - _, _ = pool.Exec(ctx, `DELETE FROM document_records WHERE doc_id = $1`, testDocID) } diff --git a/internal/docgov/project_docs.go b/internal/docgov/project_docs.go index c869e2ef..c71e1cc7 100644 --- a/internal/docgov/project_docs.go +++ b/internal/docgov/project_docs.go @@ -10,8 +10,6 @@ import ( "path/filepath" "strings" "time" - - "chat-agent-system/internal/docduty" ) // WorkflowDocumentSummary 项目工作流受控文档列表项。 @@ -59,7 +57,7 @@ func (e *GovernanceEngine) ListWorkflowDocumentsByProject(ctx context.Context, p } if e.pool != nil { - if _, _, err := docduty.HealProjectDocuments(ctx, e.pool, projectID); err != nil { + if _, _, err := HealProjectDocuments(ctx, e.pool, projectID); err != nil { return nil, fmt.Errorf("heal project document duties: %w", err) } rows, err := e.pool.Query(ctx, ` diff --git a/internal/docgov/repository.go b/internal/docgov/repository.go index a6f8c9f0..ee95e606 100644 --- a/internal/docgov/repository.go +++ b/internal/docgov/repository.go @@ -91,6 +91,11 @@ func (Repository) UpdateState(ctx context.Context, scope uow.TxScope, rec Docume } func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID string, versionID, snapshotID int64, revisionNo int, sha string) error { + return (Repository{}).ProjectHeadFull(ctx, scope, docID, &versionID, &snapshotID, revisionNo, 0, sha) +} + +// ProjectHeadFull projects ledger head pointers, optionally updating revision_schema_version when schemaVer > 0. +func (Repository) ProjectHeadFull(ctx context.Context, scope uow.TxScope, docID string, versionID, snapshotID *int64, revisionNo, schemaVer int, sha string) error { tx, err := uow.AsPG(scope) if err != nil { return err @@ -98,11 +103,12 @@ func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID stri _, err = tx.Exec(ctx, ` UPDATE document_records SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, + revision_schema_version=CASE WHEN $5 > 0 THEN $5 ELSE revision_schema_version END, row_revision=COALESCE(row_revision,0)+1, - content_sha256=COALESCE(NULLIF($5,''), content_sha256), + content_sha256=COALESCE(NULLIF($6,''), content_sha256), updated_at=now() WHERE doc_id=$1`, - docID, versionID, snapshotID, revisionNo, sha) + docID, versionID, snapshotID, revisionNo, schemaVer, sha) if err != nil { return fmt.Errorf("docgov: project head: %w", err) } diff --git a/internal/docrevision/deliverable.go b/internal/docrevision/deliverable.go new file mode 100644 index 00000000..5198f7d7 --- /dev/null +++ b/internal/docrevision/deliverable.go @@ -0,0 +1,16 @@ +package docrevision + +// DeliverableStore reads/writes the workspace deliverable without embedding path literals. +// Wired from the composition root using workspace.DeliverableKey. +type DeliverableStore interface { + Read(absWorkDir string) ([]byte, error) + Write(absWorkDir string, body []byte) error +} + +// SetDeliverableStore injects the path-owner-backed deliverable I/O. +func (s *Service) SetDeliverableStore(store DeliverableStore) { + if s == nil { + return + } + s.deliverables = store +} diff --git a/internal/docrevision/pending_fact_source_test.go b/internal/docrevision/pending_fact_source_test.go index 79984be8..f76ec539 100644 --- a/internal/docrevision/pending_fact_source_test.go +++ b/internal/docrevision/pending_fact_source_test.go @@ -10,6 +10,7 @@ import ( "chat-agent-system/internal/docgov" "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/workspace" ) func TestPendingHighRisk_DoesNotAdvanceFactSource_UntilConfirm(t *testing.T) { @@ -40,6 +41,7 @@ func TestPendingHighRisk_DoesNotAdvanceFactSource_UntilConfirm(t *testing.T) { rev := docrevision.NewService(nil) rev.DataDir = dataDir + rev.SetDeliverableStore(workspace.DirDeliverable{}) st, err := rev.OpenTemplateVersion(ctx, docID, oldMD, docrevision.ActorSnapshot{DisplayName: "t"}, 1, workDir) if err != nil { t.Fatal(err) @@ -120,6 +122,7 @@ func TestPendingHighRisk_RejectKeepsOldHead(t *testing.T) { _, _ = eng.CreateOrGetDocument(ctx, docID, 1, "intake", "摸底", "V0.1", workDir, 1, "a", 2, "r", 3, "p") rev := docrevision.NewService(nil) rev.DataDir = dataDir + rev.SetDeliverableStore(workspace.DirDeliverable{}) st, err := rev.OpenTemplateVersion(ctx, docID, oldMD, docrevision.ActorSnapshot{}, 1, workDir) if err != nil { t.Fatal(err) diff --git a/internal/docrevision/pg_mutate.go b/internal/docrevision/pg_mutate.go index c45a49e9..23400289 100644 --- a/internal/docrevision/pg_mutate.go +++ b/internal/docrevision/pg_mutate.go @@ -11,7 +11,7 @@ import ( func (s *Service) openTemplatePGLocked(ctx context.Context, docID, canon, sha string, actor ActorSnapshot, projectID int64, workDir string) (*RevisionState, error) { blob := &ContentBlob{ContentSHA256: sha, ContentMarkdown: canon, ByteSize: int64(len(canon)), CreatedAt: time.Now().UTC()} blocks := BuildBlockMap(canon, nil) - err := s.withTx(ctx, func(tx pgx.Tx) error { + err := s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { if err := s.lockOpenVersionTX(ctx, tx, docID); err != nil { return err } @@ -82,7 +82,7 @@ func (s *Service) persistHumanRevisionLocked(ctx context.Context, root *DocRootP return nil } blob := s.blobs[snap.ContentSHA256] - return s.withTx(ctx, func(tx pgx.Tx) error { + return s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { if !isBaseline { if err := s.lockOpenVersionTX(ctx, tx, root.DocID); err != nil { return err @@ -161,7 +161,7 @@ func (s *Service) persistAgentCandidateLocked(ctx context.Context, outSnap *Snap return nil } blob := s.blobs[outSnap.ContentSHA256] - return s.withTx(ctx, func(tx pgx.Tx) error { + return s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { if err := s.lockOpenVersionTX(ctx, tx, c.DocID); err != nil { return err } @@ -201,7 +201,7 @@ func (s *Service) persistAgentCommitLocked(ctx context.Context, root *DocRootPoi if !s.usingPG() { return nil } - return s.withTx(ctx, func(tx pgx.Tx) error { + return s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { if err := s.lockOpenVersionTX(ctx, tx, root.DocID); err != nil { return err } diff --git a/internal/docrevision/pg_store.go b/internal/docrevision/pg_store.go index eff7da5c..47ac2ccc 100644 --- a/internal/docrevision/pg_store.go +++ b/internal/docrevision/pg_store.go @@ -457,29 +457,19 @@ func (s *Service) projectRootPointersTX(ctx context.Context, tx pgx.Tx, root *Do if root == nil { return nil } - switch root.Kind { - case "freeform": - _, err := tx.Exec(ctx, ` - UPDATE freeform_documents - SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, - revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1 - WHERE doc_id=$1`, - root.DocID, root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer) - return err - default: - _, err := tx.Exec(ctx, ` - UPDATE document_records - SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, - revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1, - content_sha256=COALESCE(NULLIF($6,''), content_sha256) - WHERE doc_id=$1`, - root.DocID, root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, - root.RevisionSchemaVer, root.ContentSHA256) - return err + if s.rootPointers == nil { + // Unwired unit/PG tests still persist ledger rows; composition root must wire for production. + return nil } + scope, ok := uow.ScopeFrom(ctx) + if !ok { + return fmt.Errorf("docrevision: root pointer projection requires uow scope") + } + _ = tx + return s.rootPointers.Project(ctx, scope, root) } -func (s *Service) withTx(ctx context.Context, fn func(pgx.Tx) error) error { +func (s *Service) withTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error { if !s.usingPG() { return nil } @@ -488,17 +478,15 @@ func (s *Service) withTx(ctx context.Context, fn func(pgx.Tx) error) error { if err != nil { return err } - return fn(tx) + return fn(ctx, tx) } - tx, err := s.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - if err := fn(tx); err != nil { - return err - } - return tx.Commit(ctx) + return uow.NewPG(s.pool).Within(ctx, func(scope uow.TxScope) error { + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + return fn(uow.WithScope(ctx, scope), tx) + }) } // Pool exposes the underlying pool for integration tests. diff --git a/internal/docrevision/projection.go b/internal/docrevision/projection.go index e41a363d..a9f7f01d 100644 --- a/internal/docrevision/projection.go +++ b/internal/docrevision/projection.go @@ -6,8 +6,6 @@ import ( "os" "path/filepath" "strings" - - "chat-agent-system/internal/fsutil" ) const ( @@ -128,7 +126,7 @@ func MapStoredDataPath(workDir, storedRoot, hostRoot string) (string, error) { return mapped, nil } -// InspectProjection classifies output.md against the current ledger head. +// InspectProjection classifies workspace deliverable bytes against the current ledger head. func (s *Service) InspectProjection(ctx context.Context, docID, repoRoot string) (ProjectionClass, error) { if err := requireDocID(docID); err != nil { return ProjectionClass{}, err @@ -150,7 +148,10 @@ func (s *Service) InspectProjection(ctx context.Context, docID, repoRoot string) if path == "" { return ProjectionClass{Class: ProjectionMissing, HeadSHA: SHA256Hex(CanonicalizeMarkdown(headMD))}, nil } - raw, err := os.ReadFile(filepath.Join(path, "output.md")) + if s.deliverables == nil { + return ProjectionClass{Class: ProjectionMissing, HeadSHA: SHA256Hex(CanonicalizeMarkdown(headMD))}, nil + } + raw, err := s.deliverables.Read(path) if err != nil { if os.IsNotExist(err) { return ClassifyProjection(nil, headMD), nil @@ -193,7 +194,7 @@ func (s *Service) HeadContent(ctx context.Context, docID string) (string, string return md, sha, nil } -// ProjectOutputMarkdown writes head bytes to output.md after an intentional write. +// ProjectOutputMarkdown writes head bytes to the workspace deliverable after an intentional write. func (s *Service) ProjectOutputMarkdown(docID, repoRoot string) error { return s.projectOutputMarkdown(docID, repoRoot, true) } @@ -217,8 +218,10 @@ func (s *Service) projectOutputMarkdown(docID, repoRoot string, force bool) erro if path == "" { return nil } - outPath := filepath.Join(path, "output.md") - existing, err := os.ReadFile(outPath) + if s.deliverables == nil { + return fmt.Errorf("docrevision: deliverable store not configured") + } + existing, err := s.deliverables.Read(path) if err == nil { class := ClassifyProjection(existing, md) if class.Class == ProjectionAligned && string(existing) == md { @@ -227,6 +230,8 @@ func (s *Service) projectOutputMarkdown(docID, repoRoot string, force bool) erro if !force && (class.Class == ProjectionLegacyCanonicalizationDrift || class.Class == ProjectionContentFork) { return fmt.Errorf("docrevision: refuse silent overwrite of %s (file=%s head=%s)", class.Class, class.RawSHA, class.HeadSHA) } + } else if !os.IsNotExist(err) { + return err } - return fsutil.WriteFileAtomic(outPath, []byte(md), 0o644) + return s.deliverables.Write(path, []byte(md)) } diff --git a/internal/docrevision/projection_test.go b/internal/docrevision/projection_test.go index 3b73974b..ceb88047 100644 --- a/internal/docrevision/projection_test.go +++ b/internal/docrevision/projection_test.go @@ -6,6 +6,7 @@ import ( "testing" "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/workspace" ) func TestClassifyProjection_LegacyHardBreakDrift(t *testing.T) { @@ -46,6 +47,7 @@ func TestProjectOutputMarkdown_AtomicAndSameBytes(t *testing.T) { md := "人工保存 \n保留硬换行\n" svc := docrevision.NewService(nil) svc.DataDir = dir + svc.SetDeliverableStore(workspace.DirDeliverable{}) st, err := svc.OpenTemplateVersion(ctx, "doc_proj_write", md, docrevision.ActorSnapshot{DisplayName: "t"}, 1, workDir) if err != nil { t.Fatal(err) @@ -80,6 +82,7 @@ func TestProjectOutputMarkdown_RefusesSilentOverwriteOfLegacyFile(t *testing.T) } svc := docrevision.NewService(nil) svc.DataDir = dir + svc.SetDeliverableStore(workspace.DirDeliverable{}) if _, err := svc.OpenTemplateVersion(ctx, "doc_legacy", legacy, docrevision.ActorSnapshot{DisplayName: "t"}, 1, workDir); err != nil { t.Fatal(err) } @@ -124,6 +127,7 @@ func TestInspectProjection_RefusesUnmappedAbsoluteWorkDir(t *testing.T) { svc := docrevision.NewService(nil) svc.StoredDataRoot = "/app/data" svc.HostDataRoot = host + svc.SetDeliverableStore(workspace.DirDeliverable{}) if _, err := svc.OpenTemplateVersion(ctx, "doc_abs", "# x\n", docrevision.ActorSnapshot{}, 1, other); err != nil { t.Fatal(err) } @@ -146,6 +150,7 @@ func TestInspectProjection_UsesHostDataRootMapping(t *testing.T) { svc := docrevision.NewService(nil) svc.StoredDataRoot = "/app/data" svc.HostDataRoot = host + svc.SetDeliverableStore(workspace.DirDeliverable{}) legacy := docrevision.LegacyDestructiveCanonicalizeMarkdown(raw) if _, err := svc.OpenTemplateVersion(ctx, "doc_map", legacy, docrevision.ActorSnapshot{}, 1, "/app/data/workspaces/intake"); err != nil { t.Fatal(err) diff --git a/internal/docrevision/reconcile_test.go b/internal/docrevision/reconcile_test.go index 9a3737cb..4b1dcf69 100644 --- a/internal/docrevision/reconcile_test.go +++ b/internal/docrevision/reconcile_test.go @@ -9,6 +9,7 @@ import ( "chat-agent-system/internal/docgov" "chat-agent-system/internal/docrevision" "chat-agent-system/internal/fsutil" + "chat-agent-system/internal/workspace" ) func project143Raw() string { @@ -266,6 +267,7 @@ func TestWritePaths_HumanSaveTrailingSpacesShareBytes(t *testing.T) { } svc := docrevision.NewService(nil) svc.DataDir = dir + svc.SetDeliverableStore(workspace.DirDeliverable{}) st, err := svc.OpenTemplateVersion(ctx, "doc_human", "# A\n", docrevision.ActorSnapshot{}, 1, workDir) if err != nil { t.Fatal(err) diff --git a/internal/docrevision/root_pointers.go b/internal/docrevision/root_pointers.go new file mode 100644 index 00000000..86d0cc21 --- /dev/null +++ b/internal/docrevision/root_pointers.go @@ -0,0 +1,45 @@ +package docrevision + +import ( + "context" + + "chat-agent-system/internal/app/uow" +) + +// RootPointerProjector projects controlled/freeform root ledger pointers. +// Implemented outside this package (docgov + freedoc) to avoid import cycles. +type RootPointerProjector interface { + Project(ctx context.Context, scope uow.TxScope, root *DocRootPointers) error +} + +// SetRootPointerProjector injects the composition-root projector. +func (s *Service) SetRootPointerProjector(p RootPointerProjector) { + if s == nil { + return + } + s.rootPointers = p +} + +// CompositeRootPointerProjector routes by DocRootPointers.Kind. +type CompositeRootPointerProjector struct { + Controlled func(ctx context.Context, scope uow.TxScope, root *DocRootPointers) error + Freeform func(ctx context.Context, scope uow.TxScope, root *DocRootPointers) error +} + +func (c CompositeRootPointerProjector) Project(ctx context.Context, scope uow.TxScope, root *DocRootPointers) error { + if root == nil { + return nil + } + switch root.Kind { + case "freeform": + if c.Freeform == nil { + return nil + } + return c.Freeform(ctx, scope, root) + default: + if c.Controlled == nil { + return nil + } + return c.Controlled(ctx, scope, root) + } +} diff --git a/internal/docrevision/service.go b/internal/docrevision/service.go index af9723fa..d9f8d2f5 100644 --- a/internal/docrevision/service.go +++ b/internal/docrevision/service.go @@ -49,6 +49,12 @@ type Service struct { // PI_CHANGE_REPORT_NARRATIVE is on and ReportJSON is empty. NarrativeFetcher func(ctx context.Context, spec NarrativeFetchSpec) (string, error) + // Root pointer projection (controlled→docgov, freeform→freedoc); set by composition root. + rootPointers RootPointerProjector + + // Deliverable I/O (path owner via workspace.DeliverableKey); set by composition root. + deliverables DeliverableStore + // DataDir is the environment data root used to resolve relative WorkDir paths. DataDir string @@ -618,7 +624,7 @@ func (s *Service) CommitAgentVersion(ctx context.Context, docID string, candidat s.closeProposedLocked(s.reports[*c.ChangeReportID], ReportExpired) } if s.usingPG() { - if err := s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { if err := s.updateCandidateTX(ctx, tx, c); err != nil { return err } @@ -659,7 +665,7 @@ func (s *Service) CommitAgentVersion(ctx context.Context, docID string, candidat s.closeProposedLocked(s.reports[*c.ChangeReportID], ReportRejected) } if s.usingPG() { - if err := s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { if err := s.updateCandidateTX(ctx, tx, c); err != nil { return err } @@ -688,7 +694,7 @@ func (s *Service) CommitAgentVersion(ctx context.Context, docID string, candidat s.closeProposedLocked(s.reports[*c.ChangeReportID], ReportCommitted) } if s.usingPG() { - if err := s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error { eid, err := s.insertEventTX(ctx, tx, ev) if err != nil { return err diff --git a/internal/freedoc/repository.go b/internal/freedoc/repository.go index 96a3053f..e15e1f35 100644 --- a/internal/freedoc/repository.go +++ b/internal/freedoc/repository.go @@ -7,12 +7,23 @@ import ( "chat-agent-system/internal/app/uow" ) -// Repository is a M3 stub. M1 freeform projection stays on the existing service path. +// Repository writes freeform_documents root pointers inside a UnitOfWork scope. type Repository struct{} -func (Repository) ForbiddenInM1(ctx context.Context, scope uow.TxScope) error { - if _, err := uow.AsPG(scope); err != nil { +// ProjectHead projects ledger head pointers onto freeform_documents. +func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID string, versionID, snapshotID *int64, revisionNo, schemaVer int) error { + tx, err := uow.AsPG(scope) + if err != nil { return err } - return fmt.Errorf("freedoc: repository writes are not enabled until M3") + _, err = tx.Exec(ctx, ` + UPDATE freeform_documents + SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, + revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1 + WHERE doc_id=$1`, + docID, versionID, snapshotID, revisionNo, schemaVer) + if err != nil { + return fmt.Errorf("freedoc: project head: %w", err) + } + return nil } diff --git a/internal/server/document_revision_access.go b/internal/server/document_revision_access.go index 5f52241a..7ee276b3 100644 --- a/internal/server/document_revision_access.go +++ b/internal/server/document_revision_access.go @@ -4,8 +4,6 @@ import ( "context" "errors" "net/http" - "os" - "path/filepath" "strings" "chat-agent-system/internal/auth" @@ -35,11 +33,14 @@ func (s *Server) resolveDocRevisionRef(ctx context.Context, user *auth.UserAccou Kind: "controlled", ProjectID: doc.ProjectID, WorkDir: doc.WorkDir, LifecycleState: string(doc.CurrentState), AuthorID: doc.AuthorID, } + if s.docRev() != nil { + if md, _, herr := s.docRev().HeadContent(ctx, docID); herr == nil && strings.TrimSpace(md) != "" { + ref.Markdown = md + return ref, nil + } + } if content, cErr := s.gov.ReadDocumentContent(ctx, docID, s.dataRoot()); cErr == nil && content != nil { ref.Markdown = content.Markdown - } else if doc.WorkDir != "" { - md, _ := readOutputMarkdown(s.dataRoot(), doc.WorkDir) - ref.Markdown = md } return ref, nil } @@ -121,18 +122,6 @@ func (s *Server) dataRoot() string { return "." } -func readOutputMarkdown(dataRoot, workDir string) (string, error) { - path := workDir - if !filepath.IsAbs(path) { - path = filepath.Join(dataRoot, workDir) - } - b, err := os.ReadFile(filepath.Join(path, "output.md")) - if err != nil { - return "", err - } - return string(b), nil -} - func (s *Server) bootstrapRevisionIfNeeded(ctx context.Context, user *auth.UserAccount, docID string, ref *docRevisionRef) (*docrevision.RevisionState, error) { rev := s.docRev() if rev == nil { diff --git a/internal/server/document_revision_handlers.go b/internal/server/document_revision_handlers.go index 9001b19d..2b1c50c1 100644 --- a/internal/server/document_revision_handlers.go +++ b/internal/server/document_revision_handlers.go @@ -392,7 +392,7 @@ func (s *Server) handleDocRevisionCandidateDecision(w http.ResponseWriter, r *ht writeJSONResponse(w, http.StatusInternalServerError, map[string]any{ "code": "DOCUMENT_PROJECTION_FAILED", "error": err.Error(), - "hint": "候选决策已写入账本,但工作区 output.md 投影失败;请重试决策或人工修复。", + "hint": "候选决策已写入账本,但工作区交付物投影失败;请重试决策或人工修复。", "state": st, }) return diff --git a/internal/server/project_members_handlers.go b/internal/server/project_members_handlers.go index a3eed9de..36079ee7 100644 --- a/internal/server/project_members_handlers.go +++ b/internal/server/project_members_handlers.go @@ -11,7 +11,7 @@ import ( "chat-agent-system/internal/auth" "chat-agent-system/internal/authz" - "chat-agent-system/internal/docduty" + "chat-agent-system/internal/docgov" ) var errProjectRoleRequired = errors.New("project_role required when member does not exist") @@ -38,7 +38,7 @@ func (s *Server) healProjectDuties(ctx context.Context, projectID int64) error { if pool == nil { return nil } - _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID) + _, _, err := docgov.HealProjectDocuments(ctx, pool, projectID) if err != nil { log.Printf("[docduty] heal project %d after member change failed: %v", projectID, err) } diff --git a/internal/workspace/dir_deliverable.go b/internal/workspace/dir_deliverable.go new file mode 100644 index 00000000..98db890c --- /dev/null +++ b/internal/workspace/dir_deliverable.go @@ -0,0 +1,37 @@ +package workspace + +import ( + "fmt" + "os" + "path/filepath" +) + +// DirDeliverable reads and writes the canonical deliverable under an absolute work dir. +// It is the path owner for DeliverableKey; callers must not embed the literal themselves. +type DirDeliverable struct{} + +func (DirDeliverable) Read(absWorkDir string) ([]byte, error) { + if absWorkDir == "" { + return nil, fmt.Errorf("workspace: empty work dir") + } + return os.ReadFile(filepath.Join(absWorkDir, DeliverableKey)) +} + +func (DirDeliverable) Write(absWorkDir string, body []byte) error { + if absWorkDir == "" { + return fmt.Errorf("workspace: empty work dir") + } + path := filepath.Join(absWorkDir, DeliverableKey) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, body, 0o644); err != nil { + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return err + } + return nil +}