weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(api): sync removed members and fail tasks without a ledger

Browse Source 
Remove now reconciles document slots and returns ERR_DUTY_SYNC on
failure. CreateOrGetDocument errors abort the task. The frontend
refreshes workflow docs after add, update, restore, and remove.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-13 22:29:53 +00:00
co-authored by Cursor
parent 9a19bcc12c
commit 8b6bf2cca1
9 changed files with 241 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/bizdev/doc_id_test.go
+30
View File
@@ -2,11 +2,13 @@ package bizdev
import (
"context"
"errors"
"strings"
"testing"
"time"
"chat-agent-system/internal/config"
"chat-agent-system/internal/docgov"
)
func TestNormalizeDocID_LegacyPatterns(t *testing.T) {
@@ -209,6 +211,34 @@ func TestExecuteTask_LegacyDocIDDoesNotFailValidation(t *testing.T) {
}
}
func TestExecuteTask_CreateDocumentFailureIsNotSuccess(t *testing.T) {
svc := MustNewService(t, &config.Config{DataDir: t.TempDir(), AgentBackend: "stub"})
svc.createDocumentHook = func(context.Context, string, int64, string, string, string, string, int64, string, int64, string, int64, string) (*docgov.DocumentRecord, error) {
return nil, errors.New("ledger unavailable")
}
res, err := svc.ExecuteTask(context.Background(), TaskRunRequest{
TaskKey: "research",
ProjectID: 56,
ProjectName: "智能体演示项目",
CustomerName: "北京水木东方",
DeviceType: "医疗器械",
Backend: "stub",
Timeout: 10 * time.Second,
})
if err == nil || !strings.Contains(err.Error(), "ledger unavailable") {
t.Fatalf("want ledger error, got %v", err)
}
if res == nil || res.Success {
t.Fatalf("no ledger must not report success: %+v", res)
}
if !strings.Contains(res.ErrorReason, "create document ledger failed") {
t.Fatalf("ErrorReason=%q", res.ErrorReason)
}
if res.DocRecord != nil {
t.Fatalf("failed create must not return a ledger, got %#v", res.DocRecord)
}
}
func TestExecuteTask_EmptyDocIDSkipsDocRevision(t *testing.T) {
svc := MustNewService(t, &config.Config{DataDir: t.TempDir(), AgentBackend: "stub"})
ctx := context.Background()
internal/bizdev/service.go
+26 -5
View File
@@ -47,6 +47,8 @@ type Service struct {
stageGate StageAccessGate
// stageTaskHook 仅测试注入：覆盖 ExecuteStage 子任务执行，验证链中止边界。
stageTaskHook func(context.Context, TaskRunRequest) (*TaskRunResult, error)
// createDocumentHook 仅测试注入：覆盖账本创建失败路径。
createDocumentHook func(ctx context.Context, docID string, projectID int64, taskKey, title, stage, workDir string, authorID int64, authorName string, reviewerID int64, reviewerName string, approverID int64, approverName string) (*docgov.DocumentRecord, error)
}
// NewService 构造业务服务；生产/test 环境数据库不可用且未显式 ALLOW_IN_MEMORY 时返回 error。
@@ -475,11 +477,30 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
var appliedProposalID string
if s.docGov != nil && strings.TrimSpace(docID) != "" {
docRec, _ = s.docGov.CreateOrGetDocument(
ctx, docID, req.ProjectID, req.TaskKey, artifactTitle,
"V0.5_feasibility", req.WorkDir,
authorID, authorName, reviewerID, reviewerName, approverID, approverName,
)
var createErr error
if s.createDocumentHook != nil {
docRec, createErr = s.createDocumentHook(
ctx, docID, req.ProjectID, req.TaskKey, artifactTitle,
"V0.5_feasibility", req.WorkDir,
authorID, authorName, reviewerID, reviewerName, approverID, approverName,
)
} else {
docRec, createErr = s.docGov.CreateOrGetDocument(
ctx, docID, req.ProjectID, req.TaskKey, artifactTitle,
"V0.5_feasibility", req.WorkDir,
authorID, authorName, reviewerID, reviewerName, approverID, approverName,
)
}
if createErr != nil {
return &TaskRunResult{
TaskKey: req.TaskKey,
ArtifactTitle: artifactTitle,
WorkDir: req.WorkDir,
Success: false,
ErrorReason: "create document ledger failed: " + createErr.Error(),
DocRecord: docRec,
}, createErr
}
if s.pool != nil && req.ProjectID > 0 {
assigned, _, healErr := docduty.HealProjectDocuments(ctx, s.pool, req.ProjectID)
if healErr != nil {
internal/docgov/engine.go
+19 -21
View File
@@ -13,11 +13,11 @@ import (
// ModificationCheckResult 智能体或人员发起修改前的认领与授权判定结果
type ModificationCheckResult struct {
AllowedDirectEdit bool `json:"allowed_direct_edit"` // 是否允许直接执行修改
RequiresProposal bool `json:"requires_proposal"` // 是否需要生成变更提案待编制人拍板
AuthorClaimed bool `json:"author_claimed"` // 编制人是否已认领
PendingProposal *ChangeProposal `json:"pending_proposal,omitempty"`
Reason string `json:"reason"`
AllowedDirectEdit bool `json:"allowed_direct_edit"` // 是否允许直接执行修改
RequiresProposal bool `json:"requires_proposal"` // 是否需要生成变更提案待编制人拍板
AuthorClaimed bool `json:"author_claimed"` // 编制人是否已认领
PendingProposal *ChangeProposal `json:"pending_proposal,omitempty"`
Reason string `json:"reason"`
}
// GovernanceEngine 受控文档编制/审核/批准与认领治理引擎
@@ -25,11 +25,11 @@ type GovernanceEngine struct {
pool *pgxpool.Pool
sm *StateMachine
mu sync.RWMutex
memDocs map[string]*DocumentRecord
memComms map[string][]*DocumentComment
memProps map[string]*ChangeProposal
memAudits map[string][]*AuditLog
mu sync.RWMutex
memDocs map[string]*DocumentRecord
memComms map[string][]*DocumentComment
memProps map[string]*ChangeProposal
memAudits map[string][]*AuditLog
onApprovedHook func(ctx context.Context, doc *DocumentRecord)
onApprovedHooks []func(ctx context.Context, doc *DocumentRecord)
}
@@ -178,18 +178,16 @@ func (e *GovernanceEngine) CreateOrGetDocument(
doc.AuthorID, doc.AuthorName, doc.ReviewerID, doc.ReviewerName, doc.ApproverID, doc.ApproverName,
cfgJSON, doc.CreatedAt, doc.UpdatedAt,
).Scan(&id, &currState, &isLocked, &isTmpl, &authClaim, &revClaim, &appClaim)
if err == nil {
doc.ID = id
doc.CurrentState = DocumentState(currState)
doc.IsLocked = isLocked
doc.IsTemplateOnly = isTmpl
doc.AuthorClaimed = authClaim
doc.ReviewerClaimed = revClaim
doc.ApproverClaimed = appClaim
} else {
// 仅做诊断性日志输出
// fmt.Printf("[docgov] DB persist error: %v\n", err)
if err != nil {
return nil, fmt.Errorf("persist document ledger: %w", err)
}
doc.ID = id
doc.CurrentState = DocumentState(currState)
doc.IsLocked = isLocked
doc.IsTemplateOnly = isTmpl
doc.AuthorClaimed = authClaim
doc.ReviewerClaimed = revClaim
doc.ApproverClaimed = appClaim
}
e.memDocs[docID] = doc
internal/server/project_members_handlers.go
+84 -20
View File
@@ -25,7 +25,31 @@ func resolveUpdateProjectRole(requested string, existing *authz.ProjectMemberRec
return "", errProjectRoleRequired
}
func (s *Server) syncProjectDocumentDuties(ctx context.Context, projectID int64) error {
func writeDutySyncError(w http.ResponseWriter) {
http.Error(w, `{"error":"ERR_DUTY_SYNC","message":"成员已保存，但受控文档职责同步失败"}`, http.StatusInternalServerError)
}
func dutyChangeForRemove(existing *authz.ProjectMemberRecord, userID int64) docduty.MemberDutyChange {
change := docduty.MemberDutyChange{UserID: userID, Exited: true}
if existing != nil {
change.PreviousDuty = existing.DocumentDuty
change.NextDuty = existing.DocumentDuty
}
return change
}
func (s *Server) snapshotDutyHolders(ctx context.Context, projectID int64) map[string]int64 {
if s == nil || s.svc == nil || s.svc.Pool() == nil || projectID <= 0 {
return nil
}
members, _, err := docduty.LoadProjectMembers(ctx, s.svc.Pool(), projectID)
if err != nil {
return nil
}
return docduty.PriorDutyHolders(members)
}
func (s *Server) reconcileAndHealProjectDuties(ctx context.Context, projectID int64, change docduty.MemberDutyChange) error {
if s == nil || s.svc == nil || projectID <= 0 {
return nil
}
@@ -33,9 +57,9 @@ func (s *Server) syncProjectDocumentDuties(ctx context.Context, projectID int64)
if pool == nil {
return nil
}
_, _, err := docduty.HealProjectDocuments(ctx, pool, projectID)
err := docduty.ReconcileMemberDuty(ctx, pool, projectID, change)
if err != nil {
log.Printf("[docduty] sync project %d after member change failed: %v", projectID, err)
log.Printf("[docduty] reconcile project %d after member change failed: %v", projectID, err)
}
return err
}
@@ -113,6 +137,7 @@ func (s *Server) handleAddProjectMember(w http.ResponseWriter, r *http.Request,
if req.ProjectRole == "" {
req.ProjectRole = authz.RoleMember
}
priors := s.snapshotDutyHolders(r.Context(), projectID)
if err := s.authorizer.UpsertProjectMemberSpec(r.Context(), projectID, req.UserID, authz.MemberUpsert{
ProjectRole: req.ProjectRole,
DocumentDuty: req.DocumentDuty,
@@ -120,8 +145,14 @@ func (s *Server) handleAddProjectMember(w http.ResponseWriter, r *http.Request,
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err := s.syncProjectDocumentDuties(r.Context(), projectID); err != nil {
http.Error(w, `{"error":"ERR_DUTY_SYNC","message":"成员已保存，但受控文档职责同步失败"}`, http.StatusInternalServerError)
nextDuty := ""
if req.DocumentDuty != nil {
nextDuty = *req.DocumentDuty
}
if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, docduty.MemberDutyChange{
UserID: req.UserID, NextDuty: nextDuty, PriorHolderIDs: priors,
}); err != nil {
writeDutySyncError(w)
return
}
writeJSONResponse(w, http.StatusCreated, map[string]bool{"success": true})
@@ -136,20 +167,20 @@ func (s *Server) handleUpdateProjectMember(w http.ResponseWriter, r *http.Reques
http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"user_id required"}`, http.StatusBadRequest)
return
}
if req.ProjectRole != "" || req.DocumentDuty != nil {
var existing *authz.ProjectMemberRecord
if req.ProjectRole == "" {
if s.authorizer == nil {
http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"project_role required with document_duty"}`, http.StatusBadRequest)
return
}
rec, err := s.authorizer.GetProjectMember(r.Context(), projectID, req.UserID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
existing = rec
var existing *authz.ProjectMemberRecord
if s.authorizer != nil {
rec, err := s.authorizer.GetProjectMember(r.Context(), projectID, req.UserID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
existing = rec
} else if req.ProjectRole == "" && req.DocumentDuty != nil {
http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"project_role required with document_duty"}`, http.StatusBadRequest)
return
}
priors := s.snapshotDutyHolders(r.Context(), projectID)
if req.ProjectRole != "" || req.DocumentDuty != nil {
role, err := resolveUpdateProjectRole(req.ProjectRole, existing)
if err != nil {
http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"project_role required when member does not exist"}`, http.StatusBadRequest)
@@ -170,8 +201,21 @@ func (s *Server) handleUpdateProjectMember(w http.ResponseWriter, r *http.Reques
}
}
if req.ProjectRole != "" || req.DocumentDuty != nil || req.Status != "" {
if err := s.syncProjectDocumentDuties(r.Context(), projectID); err != nil {
http.Error(w, `{"error":"ERR_DUTY_SYNC","message":"成员已保存，但受控文档职责同步失败"}`, http.StatusInternalServerError)
nextDuty := ""
if existing != nil {
nextDuty = existing.DocumentDuty
}
if req.DocumentDuty != nil {
nextDuty = *req.DocumentDuty
}
if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, docduty.MemberDutyChange{
UserID: req.UserID,
PreviousDuty: existingDuty(existing),
NextDuty: nextDuty,
Exited: req.Status == authz.StatusExited,
PriorHolderIDs: priors,
}); err != nil {
writeDutySyncError(w)
return
}
}
@@ -188,9 +232,29 @@ func (s *Server) handleRemoveProjectMember(w http.ResponseWriter, r *http.Reques
http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"user_id query required"}`, http.StatusBadRequest)
return
}
var existing *authz.ProjectMemberRecord
if s.authorizer != nil {
rec, err := s.authorizer.GetProjectMember(r.Context(), projectID, targetID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
existing = rec
}
if err := s.authorizer.SetProjectMemberStatus(r.Context(), projectID, targetID, authz.StatusExited); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, dutyChangeForRemove(existing, targetID)); err != nil {
writeDutySyncError(w)
return
}
writeJSONResponse(w, http.StatusOK, map[string]bool{"success": true})
}
func existingDuty(existing *authz.ProjectMemberRecord) string {
if existing == nil {
return ""
}
return existing.DocumentDuty
}
internal/server/project_members_role_test.go
+21
View File
@@ -2,6 +2,9 @@ package server
import (
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
"chat-agent-system/internal/authz"
@@ -26,6 +29,24 @@ func TestResolveUpdateProjectRole_RejectsMissingMember(t *testing.T) {
}
}
func TestDutyChangeForRemove_UsesExistingReviewerDuty(t *testing.T) {
got := dutyChangeForRemove(&authz.ProjectMemberRecord{DocumentDuty: "reviewer"}, 171)
if !got.Exited || got.UserID != 171 || got.PreviousDuty != "reviewer" || got.NextDuty != "reviewer" {
t.Fatalf("remove change: %+v", got)
}
}
func TestWriteDutySyncError_UsesExplicitCode(t *testing.T) {
rec := httptest.NewRecorder()
writeDutySyncError(rec)
if rec.Code != http.StatusInternalServerError {
t.Fatalf("status=%d", rec.Code)
}
if !strings.Contains(rec.Body.String(), "ERR_DUTY_SYNC") {
t.Fatalf("body=%s", rec.Body.String())
}
}
func TestResolveUpdateProjectRole_ExplicitRoleWins(t *testing.T) {
role, err := resolveUpdateProjectRole(authz.RoleViewer, &authz.ProjectMemberRecord{ProjectRole: authz.RoleOwner})
if err != nil {
scripts/run-isolated-duty-sync-pg-tests.sh
+2 -2
View File
@@ -73,6 +73,6 @@ export TEST_DATABASE_URL="${TEST_URL}"
echo "[info] running isolated duty-sync PostgreSQL tests"
# 包串行，避免并行 goose 抢同一临时库。
go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding'
go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding|ReconcileMemberDuty'
go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty'
go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole'
go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole|DutyChangeForRemove|WriteDutySyncError'
web/src/App.tsx
+17 -7
View File
@@ -38,6 +38,7 @@ import {
buildManagedMemberUpsertPayload,
ensureCreatedProjectWithMembers,
} from './lib/createProjectMembersPersist';
import { afterProjectMemberPersisted } from './lib/afterProjectMemberPersisted';
import {
beginCreateProjectSubmit,
endCreateProjectSubmit,
@@ -3780,8 +3781,10 @@ export default function App() {
currentProject.creatorId,
),
);
await syncProjectMembersFromServer(currentProject.id, currentProject.creatorId);
setWorkflowDocsEpoch((n) => n + 1);
await afterProjectMemberPersisted({
syncMembers: () => syncProjectMembersFromServer(currentProject.id, currentProject.creatorId),
refreshWorkflowDocs: () => setWorkflowDocsEpoch((n) => n + 1),
});
const newSessId = buildPrivateSessionId(`p${currentProject.id}`, newMember.userId);
setSessions((prev) => {
@@ -3828,8 +3831,10 @@ export default function App() {
currentProject.id,
buildManagedMemberUpsertPayload({ userId, role: newRole }, currentProject.creatorId),
);
await syncProjectMembersFromServer(currentProject.id, currentProject.creatorId);
setWorkflowDocsEpoch((n) => n + 1);
await afterProjectMemberPersisted({
syncMembers: () => syncProjectMembersFromServer(currentProject.id, currentProject.creatorId),
refreshWorkflowDocs: () => setWorkflowDocsEpoch((n) => n + 1),
});
} catch (err) {
setProjectMembersError(err instanceof Error ? err.message : '更新成员角色失败');
}
@@ -3839,7 +3844,10 @@ export default function App() {
if (!currentProject) return;
try {
await removeProjectMember(currentProject.id, userId);
await syncProjectMembersFromServer(currentProject.id, currentProject.creatorId);
await afterProjectMemberPersisted({
syncMembers: () => syncProjectMembersFromServer(currentProject.id, currentProject.creatorId),
refreshWorkflowDocs: () => setWorkflowDocsEpoch((n) => n + 1),
});
} catch (err) {
setProjectMembersError(err instanceof Error ? err.message : '移出成员失败');
}
@@ -3856,8 +3864,10 @@ export default function App() {
currentProject.creatorId,
),
);
await syncProjectMembersFromServer(currentProject.id, currentProject.creatorId);
setWorkflowDocsEpoch((n) => n + 1);
await afterProjectMemberPersisted({
syncMembers: () => syncProjectMembersFromServer(currentProject.id, currentProject.creatorId),
refreshWorkflowDocs: () => setWorkflowDocsEpoch((n) => n + 1),
});
} catch (err) {
setProjectMembersError(err instanceof Error ? err.message : '恢复成员失败');
}
web/src/lib/afterProjectMemberPersisted.test.ts
+33
View File
@@ -0,0 +1,33 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { afterProjectMemberPersisted } from './afterProjectMemberPersisted';
test('afterProjectMemberPersisted syncs members then refreshes workflow docs', async () => {
const calls: string[] = [];
await afterProjectMemberPersisted({
syncMembers: async () => {
calls.push('sync');
},
refreshWorkflowDocs: () => {
calls.push('refresh');
},
});
assert.deepEqual(calls, ['sync', 'refresh']);
});
test('afterProjectMemberPersisted does not refresh when member sync fails', async () => {
let refreshed = false;
await assert.rejects(
() =>
afterProjectMemberPersisted({
syncMembers: async () => {
throw new Error('sync failed');
},
refreshWorkflowDocs: () => {
refreshed = true;
},
}),
/sync failed/,
);
assert.equal(refreshed, false);
});
web/src/lib/afterProjectMemberPersisted.ts
+9
View File
@@ -0,0 +1,9 @@
/** 成员增改/恢复/移出成功后同步成员并刷新受控文档矩阵。 */
export async function afterProjectMemberPersisted(opts: {
syncMembers: () => Promise<unknown>;
refreshWorkflowDocs: () => void;
}): Promise<void> {
await opts.syncMembers();
opts.refreshWorkflowDocs();
}