feat(arch): clear G1 U3/U4 sql_write and path exceptions

Move document_records duty-slot writes into docgov, project ledger
heads through docgov/freedoc ports, and workspace deliverable I/O
through DeliverableStore so arch-check baseline drops to sql_write:1
path:13.
This commit is contained in:
WeiCHEN
2026-09-22 21:40:12 +08:00
parent 364ed7f1a3
commit 9c7a5be209
29 changed files with 583 additions and 356 deletions
+4 -58
View File
@@ -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.
+5 -1
View File
@@ -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,
+42
View File
@@ -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{})
}
+30 -5
View File
@@ -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,
+5 -2
View File
@@ -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)
+15 -1
View File
@@ -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{
+6 -4
View File
@@ -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)
+5 -37
View File
@@ -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
}
}
+17 -130
View File
@@ -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
}
+43 -34
View File
@@ -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
}
@@ -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)
+194
View File
@@ -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)
}
+28 -8
View File
@@ -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)
}
+1 -3
View File
@@ -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, `
+8 -2
View File
@@ -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)
}
+16
View File
@@ -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
}
@@ -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)
+4 -4
View File
@@ -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
}
+18 -30
View File
@@ -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.
+13 -8
View File
@@ -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))
}
+5
View File
@@ -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)
+2
View File
@@ -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)
+45
View File
@@ -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)
}
}
+9 -3
View File
@@ -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
+15 -4
View File
@@ -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
}
+6 -17
View File
@@ -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 {
@@ -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
+2 -2
View File
@@ -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)
}
+37
View File
@@ -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
}