From 32b88beee037881955b38652e9f420b0d6325236 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 12 Sep 2026 03:14:30 +0000 Subject: [PATCH] feat(projectinstance): refresh stage readiness after document approval. Sync doc task status on approve and evaluate whether to emit stage advance readiness; chain docgov hooks. Co-authored-by: Cursor --- internal/docgov/engine.go | 40 ++++++++-- internal/projectinstance/approval.go | 78 +++++++++++++++++++ internal/projectinstance/approval_test.go | 44 +++++++++++ .../projectinstance/pg_integration_test.go | 29 +++++++ internal/projectinstance/stage_store.go | 18 +++++ internal/projectinstance/store.go | 20 +++++ 6 files changed, 222 insertions(+), 7 deletions(-) create mode 100644 internal/projectinstance/approval.go create mode 100644 internal/projectinstance/approval_test.go diff --git a/internal/docgov/engine.go b/internal/docgov/engine.go index 125da1fe..0f54f28d 100644 --- a/internal/docgov/engine.go +++ b/internal/docgov/engine.go @@ -30,7 +30,8 @@ type GovernanceEngine struct { memComms map[string][]*DocumentComment memProps map[string]*ChangeProposal memAudits map[string][]*AuditLog - onApprovedHook func(ctx context.Context, doc *DocumentRecord) + onApprovedHook func(ctx context.Context, doc *DocumentRecord) + onApprovedHooks []func(ctx context.Context, doc *DocumentRecord) } // SetOnApprovedHook 注册文档终审批准时的回调钩子 @@ -40,6 +41,34 @@ func (e *GovernanceEngine) SetOnApprovedHook(hook func(ctx context.Context, doc e.onApprovedHook = hook } +// AddOnApprovedHook 追加文档终审批准回调（与 SetOnApprovedHook 并存）。 +func (e *GovernanceEngine) AddOnApprovedHook(hook func(ctx context.Context, doc *DocumentRecord)) { + if hook == nil { + return + } + e.mu.Lock() + defer e.mu.Unlock() + e.onApprovedHooks = append(e.onApprovedHooks, hook) +} + +func (e *GovernanceEngine) runApprovedHooks(ctx context.Context, doc *DocumentRecord) { + var primary func(context.Context, *DocumentRecord) + var extra []func(context.Context, *DocumentRecord) + e.mu.Lock() + primary = e.onApprovedHook + extra = append([]func(context.Context, *DocumentRecord){}, e.onApprovedHooks...) + e.mu.Unlock() + + if primary != nil { + primary(ctx, doc) + } + for _, hook := range extra { + if hook != nil { + hook(ctx, doc) + } + } +} + // NewGovernanceEngine 初始化治理引擎 func NewGovernanceEngine(pool *pgxpool.Pool) *GovernanceEngine { return &GovernanceEngine{ @@ -821,12 +850,9 @@ func (e *GovernanceEngine) ApproverApprove(ctx context.Context, docID string, ac e.memAudits[docID] = append(e.memAudits[docID], audit) e.syncDocAndAuditToDB(ctx, doc, audit) - if e.onApprovedHook != nil { - hook := e.onApprovedHook - e.mu.Unlock() - hook(ctx, doc) - e.mu.Lock() - } + e.mu.Unlock() + e.runApprovedHooks(ctx, doc) + e.mu.Lock() return doc, nil } diff --git a/internal/projectinstance/approval.go b/internal/projectinstance/approval.go new file mode 100644 index 00000000..d09000c3 --- /dev/null +++ b/internal/projectinstance/approval.go @@ -0,0 +1,78 @@ +package projectinstance + +import "context" + +// DocumentApprovedResult 文档批准后阶段就绪评估结果。 +type DocumentApprovedResult struct { + StageReady bool + InstanceID int64 + StageInstanceID int64 + StageKey string + StageName string + ConfigRevision int + RequiredDocCount int + ApprovedDocCount int +} + +// HandleDocumentApproved 同步文档任务状态并评估是否首次满足阶段推进条件。 +func (s *Service) HandleDocumentApproved(ctx context.Context, projectID int64, taskKey string) (*DocumentApprovedResult, error) { + out := &DocumentApprovedResult{} + if s == nil || !s.Available() || projectID <= 0 || taskKey == "" { + return out, nil + } + + if err := s.store.MarkDocTaskApproved(ctx, projectID, taskKey); err != nil { + return nil, err + } + + live, err := s.store.GetLiveInstance(ctx, projectID) + if err != nil || live == nil { + return out, err + } + + detail, err := s.store.GetInstanceByID(ctx, live.ID) + if err != nil || detail == nil || detail.ActiveStageKey == "" { + return out, err + } + + readiness, err := s.GetStageReadiness(ctx, live.ID, detail.ActiveStageKey, 0, true) + if err != nil { + return nil, err + } + if !readiness.Ready { + return out, nil + } + + var stageInstID int64 + for _, st := range detail.Stages { + if st.StageKey == detail.ActiveStageKey { + stageInstID = st.ID + break + } + } + if stageInstID <= 0 { + return out, nil + } + + required := 0 + approved := 0 + for _, dt := range readiness.DocTasks { + if !dt.Enabled || !dt.Required { + continue + } + required++ + if dt.Approved { + approved++ + } + } + + out.StageReady = true + out.InstanceID = live.ID + out.StageInstanceID = stageInstID + out.StageKey = readiness.StageKey + out.StageName = readiness.StageName + out.ConfigRevision = readiness.ConfigRevision + out.RequiredDocCount = required + out.ApprovedDocCount = approved + return out, nil +} diff --git a/internal/projectinstance/approval_test.go b/internal/projectinstance/approval_test.go new file mode 100644 index 00000000..034b6c27 --- /dev/null +++ b/internal/projectinstance/approval_test.go @@ -0,0 +1,44 @@ +package projectinstance + +import ( + "context" + "testing" +) + +func TestHandleDocumentApproved_StageReady(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 1, LifecycleActive, "medical_device_development", 1, 1) + + svc := NewService(store, testResolver(), nil, &stubDocChecker{states: map[string]string{"intake": "approved"}}) + plan := WorkflowPlan{ + WorkflowDefinitionKey: "medical_device_development", + WorkflowVersion: 1, + Stages: []StagePlan{{ + StageKey: "biz_intake", Enabled: true, + DocTasks: []DocTaskPlan{{ + TaskKey: "intake", Enabled: true, + Duties: DocDuty{AuthorUserID: 2, ReviewerUserID: 3, ApproverUserID: 4}, + }}, + }}, + } + inst := &WorkflowInstance{ + ID: 1, ProjectID: 1, WorkflowVersionID: 10, Status: InstanceActive, + ConfigRevision: 1, PlanHash: "h", ConfigSnapshot: plan, InstanceNo: 1, + } + activated, err := store.ActivateInstance(ctx, 1, inst, "medical_device_development", 1, 2) + if err != nil { + t.Fatalf("activate: %v", err) + } + + result, err := svc.HandleDocumentApproved(ctx, 1, "intake") + if err != nil { + t.Fatalf("handle approved: %v", err) + } + if result == nil || !result.StageReady { + t.Fatalf("expected stage ready, got %+v", result) + } + if result.InstanceID != activated.ID || result.StageKey != "biz_intake" { + t.Fatalf("unexpected result: %+v", result) + } +} diff --git a/internal/projectinstance/pg_integration_test.go b/internal/projectinstance/pg_integration_test.go index dd309331..b450cf1c 100644 --- a/internal/projectinstance/pg_integration_test.go +++ b/internal/projectinstance/pg_integration_test.go @@ -10,6 +10,7 @@ import ( "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" + "chat-agent-system/internal/notification" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/workflowdef" @@ -140,6 +141,34 @@ func TestPG_ActivationAndStageAdvance(t *testing.T) { t.Fatalf("approve doc: %v", err) } + approvedResult, err := svc.HandleDocumentApproved(ctx, projectID, "intake") + if err != nil { + t.Fatalf("handle document approved: %v", err) + } + if approvedResult == nil || !approvedResult.StageReady { + t.Fatalf("expected stage ready after approval, got %+v", approvedResult) + } + + notifySvc := notification.NewService(pool) + if notifySvc.Available() { + created, emitErr := notifySvc.EmitStageReadyToAdvance(ctx, 2, notification.StageReadyPayload{ + ProjectID: projectID, + WorkflowInstanceID: approvedResult.InstanceID, + StageInstanceID: approvedResult.StageInstanceID, + StageKey: approvedResult.StageKey, + StageName: approvedResult.StageName, + ConfigRevision: approvedResult.ConfigRevision, + RequiredDocCount: approvedResult.RequiredDocCount, + ApprovedDocCount: approvedResult.ApprovedDocCount, + }) + if emitErr != nil { + t.Fatalf("emit stage ready notification: %v", emitErr) + } + if !created { + t.Fatal("expected stage ready notification to be created") + } + } + readiness, err := svc.GetStageReadiness(ctx, activated.Instance.ID, "biz_intake", 2, true) if err != nil { t.Fatalf("readiness: %v", err) diff --git a/internal/projectinstance/stage_store.go b/internal/projectinstance/stage_store.go index 8d759e67..0d1af835 100644 --- a/internal/projectinstance/stage_store.go +++ b/internal/projectinstance/stage_store.go @@ -172,3 +172,21 @@ func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey s } return s.loadInstanceDetail(ctx, inst) } + +func (s *pgStore) MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error { + if projectID <= 0 || taskKey == "" { + return nil + } + _, err := s.pool.Exec(ctx, ` + UPDATE project_doc_task_instances dti + SET status = 'approved', updated_at = now() + FROM project_stage_instances psi + JOIN project_workflow_instances pwi ON pwi.id = psi.workflow_instance_id + WHERE dti.stage_instance_id = psi.id + AND pwi.project_id = $1 + AND dti.task_key = $2 + AND dti.enabled = true + AND dti.status <> 'skipped' + `, projectID, taskKey) + return err +} diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go index a1391af0..5e416fc6 100644 --- a/internal/projectinstance/store.go +++ b/internal/projectinstance/store.go @@ -44,6 +44,7 @@ type Store interface { GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error) AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) + MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error } // NewStore 构建 PG 或内存 Store。 @@ -672,6 +673,25 @@ func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey st return m.instanceDetailLocked(©Inst), nil } +func (m *memStore) MarkDocTaskApproved(_ context.Context, projectID int64, taskKey string) error { + m.mu.Lock() + defer m.mu.Unlock() + live := m.live[projectID] + if live == nil { + return nil + } + records := m.stages[live.ID] + for i := range records { + for j := range records[i].tasks { + if records[i].tasks[j].taskKey == taskKey && records[i].tasks[j].enabled { + records[i].tasks[j].status = "approved" + } + } + } + m.stages[live.ID] = records + return nil +} + func (m *memStore) SaveActivationToken(_ context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, _ int64, expiresAt time.Time) error { m.mu.Lock() defer m.mu.Unlock()