diff --git a/.doc/2026-09-12-remediation-gap-matrix.md b/.doc/2026-09-12-remediation-gap-matrix.md new file mode 100644 index 00000000..6bab68ae --- /dev/null +++ b/.doc/2026-09-12-remediation-gap-matrix.md @@ -0,0 +1,35 @@ +# 整改缺口复核矩阵 + +**日期**: 2026-09-12（更新） +**分支**: dev +**用途**: 审计文档 vs 代码现状；后续 Task 仅引用本矩阵 + +| ID | 审计描述 | 状态 | 证据 | 计划动作 | +|---|---|---|---|---| +| BUG-01 | 完成后 lifecycle 未更新 | PASS | `stage_remediation_test.go`；`scripts/fix-completed-instance-lifecycle.sql` | 已完成 | +| BUG-02 | 推进强制批文档 | PASS | `AdvanceStage` 无 doc UPDATE；`TestAdvanceStage_DoesNotModifyDocumentApprovalStatus` | 已完成 | +| BUG-03 | 无 skipped_disabled | PASS | memStore `skipped_disabled`；`TestDisabledDocumentPersistence` | 已完成 | +| BUG-04 | 对话走旧 bizdev | PASS | `engine.go` live 门控；`instance_advance_governance_test.go` | 已完成 | +| BUG-05 | 拒绝逻辑分散 | PASS | `document_access.go` + `stage_access_helpers.go`；HTTP 409 `DOC_TASK_DISABLED` | 已完成 | +| BUG-06 | token 明文/重放 | PASS | `ActivateInstanceWithToken` 同事务；`activation_token_test.go` | 已完成 | +| BUG-07 | 完成后下一流程 | PASS | 同 BUG-01 + settings `lifecycle_status=planning` E2E | 已完成 | +| BUG-08 | 测试库隔离 | PARTIAL | `db/testdb.go` Require/Skip；治理 PG 测试仍 Skip 无库时 | CI 必须设 `TEST_DATABASE_URL` | +| BUG-09 | 版本冲突自动重试 | PASS | `StageAdvanceBanner` reload；`advanceStage` 409 提示 | 已完成 | +| BUG-10 | 无审计视图 | PASS | `recent_workflow_events` in settings + UI | 已完成 | +| BUG-11 | 旧 API 未下线 | PARTIAL | 治理开启时 410 | 保持 + 文档标注 | +| BUG-12 | 前端名称硬编码 | PASS | `ProjectWorkflowSwitcher` catalog API | 已完成 | +| MISSING-01 | 上下文快照 0% | PARTIAL | `projectcontext/service.go` 五层源 + 私有对话排除计数 | PM/TE bizdev 传入 Dialogue 待加深 | +| MISSING-02 | 终止 0% | PASS | `GetTerminationPreview` + 取消在途 + 前端 | 已完成 | +| MISSING-04 | E2E 0% | PARTIAL | `e2e_workflow_governance_test.go` 单工作流完成→planning | 三正式工作流 + 真实 PI Agent 待 test 分支补跑 | + +**结论**: Phase A 全部 PASS；Phase B/C 核心项 PASS，E2E 与上下文快照仍有 PARTIAL。test 分支合并前需 `TEST_DATABASE_URL` + 真实 PI Agent 全链路验收。 + +## 发布门禁（test → main） + +``` +□ go test ./internal/projectinstance/... ./internal/coordinator/... ./internal/projectcontext/... 全绿 +□ TEST_DATABASE_URL 下 go test ./internal/server/... -run 'TestE2E_WorkflowGovernance|TestWorkflowGovernance' 全绿 +□ 真实 PI Agent（MiniMax-M3）三工作流治理 E2E 实录写入 test-reports +□ scripts/fix-completed-instance-lifecycle.sql 在目标库 dry-run 无意外 +□ 前端：版本冲突刷新、终止 preview、catalog 名称、审计事件列表人工抽检 +``` diff --git a/.doc/test-reports/2026-09-12-workflow-governance-remediation-test-report.md b/.doc/test-reports/2026-09-12-workflow-governance-remediation-test-report.md new file mode 100644 index 00000000..a4d39836 --- /dev/null +++ b/.doc/test-reports/2026-09-12-workflow-governance-remediation-test-report.md @@ -0,0 +1,62 @@ +# 三工作流治理整改 测试报告 + +## 一、 测试背景 (Background) + +2026-09-12 全面审计发现治理链路存在文档拒绝分散、对话双轨推进、终止 preview 缺失等问题。本报告记录 Phase A–C 整改后的单元/集成验证。 + +## 二、 测试目标 (Test Objectives) + +- Phase A：生命周期补测、推进不改文档、禁用文档、对话不切 bizdev、token 一次性、DocumentAccessDecider + HTTP 409 +- Phase B：终止 preview、E2E 完成→planning、上下文快照可复现（含私有对话排除计数） +- Phase C：catalog 驱动名称、版本冲突提示、settings 审计事件 + +## 三、 测试方案与环境 (Test Plan & Environment) + +- OS: Linux +- Go: 项目默认 toolchain +- 单元测试：内存 Store（`internal/projectinstance`） +- PG 集成：`TEST_DATABASE_URL` 设置时运行 `internal/server/e2e_workflow_governance_test.go` + +## 四、 预期结果 (Expected Results) + +- 末阶段推进后 `lifecycle_status=planning` +- live 实例对话推进零次 `bizdev.ExecuteStage` +- 禁用文档返回 `DOC_TASK_DISABLED`（task run HTTP 409） +- 终止后项目可重新规划 + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 5.1 单元测试（内存 Store） + +**命令**: `go test ./internal/projectinstance/... -count=1` + +**判定**: PASS（含 `TestAdvanceStage_CompletesProjectAfterLastStage`、`TestDisabledDocumentPersistence`、`TestActivationToken_OneTimeUse`、`TestTerminateInstance_UpdatesProjectLifecycle`、`TestCheckDocTaskAccess_DisabledDocumentRejected`） + +### 5.2 Coordinator 治理测试 + +**命令**: `go test ./internal/coordinator/... -run TestDialogueAdvanceStage_LiveInstanceDoesNotUseBizdev -count=1` + +**判定**: PASS — live 实例存在时 `ExecuteStage` 调用次数为 0 + +### 5.3 上下文快照 + +**命令**: `go test ./internal/projectcontext/... -count=1` + +**判定**: PASS — 同输入 hash 可复现；`[private]`/`[draft]` 对话计入 `ExcludedPrivateCount` 且不进入来源清单 + +### 5.4 PG E2E（需 TEST_DATABASE_URL） + +**命令**: `TEST_DATABASE_URL=... go test ./internal/server/... -run TestE2E_WorkflowGovernance_CompleteAndReplan -count=1 -v` + +**判定**: SKIP 或 PASS（视 CI 是否配置隔离库） + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +``` +go test ./internal/projectinstance/... ./internal/coordinator/... ./internal/projectcontext/... -count=1 +``` + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +- 整改代码路径已用单元测试锁定；PG 全链路需在 `test` 分支部署前用真实 PI Agent 补跑并更新本报告第五节 5.4 为 PASS 实录。 +- 建议每周运行 `scripts/fix-completed-instance-lifecycle.sql` dry-run 检查历史不一致。 diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 9bd43fe2..7442034e 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -680,11 +680,28 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u } func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent.CandidateIntent, userCtx *UserProjectContext) (*CoordinationResult, error) { - if e.instSvc != nil && e.instSvc.Available() && userCtx != nil && userCtx.Project.CreatorUserID > 0 { + if e.instSvc != nil && e.instSvc.Available() && userCtx != nil && userCtx.Project.ID != "" { projectID, parseErr := strconv.ParseInt(userCtx.Project.ID, 10, 64) userID, userErr := strconv.ParseInt(userCtx.User.ID, 10, 64) - if parseErr == nil && userErr == nil && projectID > 0 && userID == userCtx.Project.CreatorUserID { - if res, handled, advErr := e.tryAdvanceWorkflowInstance(ctx, projectID, userID); handled { + if parseErr == nil && userErr == nil && projectID > 0 && userID > 0 { + isCreator := userCtx.Project.CreatorUserID > 0 && userID == userCtx.Project.CreatorUserID + settings, settingsErr := e.instSvc.GetSettings(ctx, projectID, userID, isCreator) + hasLiveGovernance := settingsErr == nil && settings != nil && settings.LiveInstance != nil && + settings.LiveInstance.Status != projectinstance.InstanceCompleted && + settings.LiveInstance.Status != "terminated" + + if hasLiveGovernance { + if !isCreator { + return &CoordinationResult{ + Status: StatusRejected, + Intent: cand.Intent, + Branch: cand.Branch, + RejectReason: projectinstance.ErrForbiddenNotCreator.Error(), + NaturalReply: "只有项目创建者可以确认推进工作流阶段。", + UpdatedContext: userCtx, + }, projectinstance.ErrForbiddenNotCreator + } + res, _, advErr := e.tryAdvanceWorkflowInstance(ctx, projectID, userID) if advErr != nil { return &CoordinationResult{ Status: StatusRejected, @@ -697,6 +714,22 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent } return res, nil } + + if isCreator { + if res, handled, advErr := e.tryAdvanceWorkflowInstance(ctx, projectID, userID); handled { + if advErr != nil { + return &CoordinationResult{ + Status: StatusRejected, + Intent: cand.Intent, + Branch: cand.Branch, + RejectReason: advErr.Error(), + NaturalReply: fmt.Sprintf("阶段推进未通过门禁：%v", advErr), + UpdatedContext: userCtx, + }, advErr + } + return res, nil + } + } } } diff --git a/internal/coordinator/instance_advance_governance_test.go b/internal/coordinator/instance_advance_governance_test.go new file mode 100644 index 00000000..b2bd74d2 --- /dev/null +++ b/internal/coordinator/instance_advance_governance_test.go @@ -0,0 +1,83 @@ +package coordinator + +import ( + "context" + "testing" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/intent" + "chat-agent-system/internal/kb" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/squad" + "chat-agent-system/internal/storage" +) + +type countingBizdev struct { + stageCalls int +} + +func (c *countingBizdev) ListTasks() []bizdev.TaskMeta { return nil } +func (c *countingBizdev) ListStages() []bizdev.StageMeta { return nil } +func (c *countingBizdev) ExecuteTask(_ context.Context, _ bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) { + return &bizdev.TaskRunResult{}, nil +} +func (c *countingBizdev) ExecuteStage(_ context.Context, _ bizdev.StageRunRequest) (*bizdev.StageRunResult, error) { + c.stageCalls++ + return &bizdev.StageRunResult{StageTitle: "legacy"}, nil +} +func (c *countingBizdev) InspectWorkspace(_ context.Context, _ string) (*squad.WorkspaceState, error) { + return nil, nil +} +func (c *countingBizdev) ResumeOrModifyWorkspace(_ context.Context, _ squad.ModifyWorkflowRequest) (*squad.SquadRunResult, error) { + return nil, nil +} +func (c *countingBizdev) ContextManager() *ctxmgr.ContextManager { return nil } +func (c *countingBizdev) CAS() *storage.CAS { return nil } +func (c *countingBizdev) DocGov() *docgov.GovernanceEngine { return nil } +func (c *countingBizdev) KB() *kb.Client { return nil } + +func TestDialogueAdvanceStage_LiveInstanceDoesNotUseBizdev(t *testing.T) { + store := projectinstance.NewMemoryStore() + _ = projectinstance.SeedProjectMeta(store, 55, projectinstance.LifecyclePlanning, "", 1, 0) + instSvc := projectinstance.NewService(store, projectinstance.TestResolver(), nil, &projectinstance.StubDocChecker{States: map[string]string{"intake": "approved"}}) + + ctx := context.Background() + _, _ = instSvc.SaveWorkflowPlan(ctx, 55, 1, true, projectinstance.SavePlanInput{ + Plan: projectinstance.ValidTestPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + preview, _ := instSvc.ActivationPreview(ctx, 55, 1, true) + _, _ = instSvc.Activate(ctx, 55, 1, true, projectinstance.ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + + counter := &countingBizdev{} + engine := NewCoordinatorEngine(counter) + engine.SetInstanceService(instSvc) + + userCtx := &UserProjectContext{ + Project: ProjectInfo{ID: "55", CreatorUserID: 1, CurrentStage: bizdev.StageV01_Triage}, + User: UserInfo{ID: "1", ProjectRole: "owner"}, + } + recRes := &intent.IntentRecognitionResult{ + TopIntent: &intent.CandidateIntent{ + Intent: intent.IntentAdvanceStage, + FilledSlots: map[string]string{ + intent.SlotTargetStage: string(bizdev.StageV05_Feasibility), + }, + }, + } + + res, err := engine.ProcessIntent(ctx, recRes, userCtx) + if err != nil { + t.Fatalf("ProcessIntent: %v", err) + } + if res.Status != StatusAccepted { + t.Fatalf("expected accepted via instance service, got %s (%s)", res.Status, res.RejectReason) + } + if counter.stageCalls != 0 { + t.Fatalf("expected zero bizdev ExecuteStage calls with live instance, got %d", counter.stageCalls) + } +} diff --git a/internal/projectcontext/service.go b/internal/projectcontext/service.go index 562b5504..197c75a8 100644 --- a/internal/projectcontext/service.go +++ b/internal/projectcontext/service.go @@ -38,13 +38,15 @@ type Source struct { // Snapshot 不可变项目上下文快照。 type Snapshot struct { - ID int64 `json:"id"` - ProjectID int64 `json:"project_id"` - TaskKey string `json:"task_key"` - CutoffAt time.Time `json:"cutoff_at"` - AggregateHash string `json:"aggregate_hash"` - SourceCount int `json:"source_count"` - Sources []Source `json:"sources"` + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + TaskKey string `json:"task_key"` + CutoffAt time.Time `json:"cutoff_at"` + AggregateHash string `json:"aggregate_hash"` + SourceCount int `json:"source_count"` + ExcludedPrivateCount int `json:"excluded_private_count"` + ExcludedNoAccessCount int `json:"excluded_no_access_count"` + Sources []Source `json:"sources"` } // BuildInput 构建快照所需输入。 @@ -77,17 +79,19 @@ func (s *Service) Available() bool { // BuildSnapshot 建立来源清单并按任务检索相关片段（不 dump 全量原文）。 func (s *Service) BuildSnapshot(ctx context.Context, in BuildInput) (*Snapshot, error) { cutoff := time.Now() - sources := enumerateSources(in) + sources, excludedPrivate, excludedNoAccess := enumerateSources(in) sources = selectForTask(in.TaskKey, sources) aggHash := aggregateHash(sources) snap := &Snapshot{ - ProjectID: in.ProjectID, - TaskKey: in.TaskKey, - CutoffAt: cutoff, - AggregateHash: aggHash, - SourceCount: len(sources), - Sources: sources, + ProjectID: in.ProjectID, + TaskKey: in.TaskKey, + CutoffAt: cutoff, + AggregateHash: aggHash, + SourceCount: len(sources), + ExcludedPrivateCount: excludedPrivate, + ExcludedNoAccessCount: excludedNoAccess, + Sources: sources, } if !s.Available() || in.ProjectID <= 0 { return snap, nil @@ -137,8 +141,10 @@ func MaterialsForWorkspace(snap *Snapshot, maxItems int) []squad.MaterialItem { return out } -func enumerateSources(in BuildInput) []Source { +func enumerateSources(in BuildInput) ([]Source, int, int) { out := make([]Source, 0) + excludedPrivate := 0 + excludedNoAccess := 0 for k, v := range in.Facts { if strings.TrimSpace(v) == "" { continue @@ -177,12 +183,18 @@ func enumerateSources(in BuildInput) []Source { }) } for i, line := range in.Dialogue { - if strings.TrimSpace(line) == "" { + trimmed := strings.TrimSpace(line) + if trimmed == "" { + excludedPrivate++ + continue + } + if strings.HasPrefix(trimmed, "[private]") || strings.HasPrefix(trimmed, "[draft]") { + excludedPrivate++ continue } out = append(out, Source{ - Type: "dialogue", ID: fmt.Sprintf("turn_%d", i), ContentHash: hashText(line), - AuthorityLevel: AuthorityDialogue, PermissionScope: "project_shared", Snippet: truncate(line, 2000), + Type: "dialogue", ID: fmt.Sprintf("turn_%d", i), ContentHash: hashText(trimmed), + AuthorityLevel: AuthorityDialogue, PermissionScope: "project_shared", Snippet: truncate(trimmed, 2000), }) } for i, d := range in.Decisions { @@ -204,7 +216,7 @@ func enumerateSources(in BuildInput) []Source { sort.Slice(out, func(i, j int) bool { return authorityRank(out[i].AuthorityLevel) < authorityRank(out[j].AuthorityLevel) }) - return out + return out, excludedPrivate, excludedNoAccess } func selectForTask(taskKey string, sources []Source) []Source { diff --git a/internal/projectcontext/service_test.go b/internal/projectcontext/service_test.go new file mode 100644 index 00000000..37a409aa --- /dev/null +++ b/internal/projectcontext/service_test.go @@ -0,0 +1,62 @@ +package projectcontext + +import ( + "testing" + + "chat-agent-system/internal/squad" +) + +func TestBuildSnapshot_ConsistentForSameInput(t *testing.T) { + svc := NewService(nil) + in := BuildInput{ + ProjectID: 1, + TaskKey: "system_reqs", + UserID: 1, + Facts: map[string]string{"product_type": "超声设备"}, + Materials: []squad.MaterialItem{{Filename: "note.md", Content: []byte("材料说明")}}, + Dialogue: []string{"项目需要 CE 认证"}, + Decisions: []string{"目标市场：欧盟"}, + } + s1, err := svc.BuildSnapshot(t.Context(), in) + if err != nil { + t.Fatalf("BuildSnapshot: %v", err) + } + s2, err := svc.BuildSnapshot(t.Context(), in) + if err != nil { + t.Fatalf("BuildSnapshot second: %v", err) + } + if s1.AggregateHash != s2.AggregateHash { + t.Fatalf("expected reproducible hash, got %s vs %s", s1.AggregateHash, s2.AggregateHash) + } +} + +func TestBuildSnapshot_ExcludesEmptyPrivateDialogue(t *testing.T) { + svc := NewService(nil) + snap, err := svc.BuildSnapshot(t.Context(), BuildInput{ + ProjectID: 2, + TaskKey: "test_manager", + Dialogue: []string{"", " ", "[private] 不应纳入", "可共享的项目讨论"}, + }) + if err != nil { + t.Fatalf("BuildSnapshot: %v", err) + } + if snap.SourceCount != 1 { + t.Fatalf("expected 1 shared dialogue source, got %d", snap.SourceCount) + } + if snap.ExcludedPrivateCount < 3 { + t.Fatalf("expected excluded private count >= 3, got %d", snap.ExcludedPrivateCount) + } +} + +func TestMaterialsForWorkspace_RespectsAuthority(t *testing.T) { + snap := &Snapshot{ + Sources: []Source{ + {Type: "dialogue", ID: "turn_1", Referenced: false, Snippet: "ignored"}, + {Type: AuthorityApprovedDoc, ID: "srs", Referenced: true, Snippet: "SRS body"}, + }, + } + items := MaterialsForWorkspace(snap, 5) + if len(items) != 1 { + t.Fatalf("expected 1 referenced material, got %d", len(items)) + } +} diff --git a/internal/projectinstance/activation_token_test.go b/internal/projectinstance/activation_token_test.go new file mode 100644 index 00000000..151aeace --- /dev/null +++ b/internal/projectinstance/activation_token_test.go @@ -0,0 +1,80 @@ +package projectinstance + +import ( + "context" + "errors" + "sync" + "testing" +) + +func TestActivationToken_OneTimeUse(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 200, LifecyclePlanning, "", 1, 0) + svc := NewService(store, testResolver(), nil, nil) + + _, _ = svc.SaveWorkflowPlan(ctx, 200, 1, true, SavePlanInput{ + Plan: validPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + preview, err := svc.ActivationPreview(ctx, 200, 1, true) + if err != nil { + t.Fatalf("ActivationPreview: %v", err) + } + + _, err = svc.Activate(ctx, 200, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, + ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + if err != nil { + t.Fatalf("first activate: %v", err) + } + + _, err = svc.Activate(ctx, 200, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, + ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + if !errors.Is(err, ErrTokenInvalid) && !errors.Is(err, ErrPlanFrozen) { + t.Fatalf("expected ErrTokenInvalid or ErrPlanFrozen on reuse, got %v", err) + } +} + +func TestActivationToken_ConcurrentActivateOnlyOneSucceeds(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 201, LifecyclePlanning, "", 1, 0) + svc := NewService(store, testResolver(), nil, nil) + + _, _ = svc.SaveWorkflowPlan(ctx, 201, 1, true, SavePlanInput{ + Plan: validPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + preview, err := svc.ActivationPreview(ctx, 201, 1, true) + if err != nil { + t.Fatalf("ActivationPreview: %v", err) + } + + var wg sync.WaitGroup + successes := 0 + var mu sync.Mutex + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, actErr := svc.Activate(ctx, 201, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, + ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + if actErr == nil { + mu.Lock() + successes++ + mu.Unlock() + } + }() + } + wg.Wait() + if successes != 1 { + t.Fatalf("expected exactly 1 successful activation, got %d", successes) + } +} diff --git a/internal/projectinstance/document_access.go b/internal/projectinstance/document_access.go new file mode 100644 index 00000000..91d20782 --- /dev/null +++ b/internal/projectinstance/document_access.go @@ -0,0 +1,132 @@ +package projectinstance + +import ( + "context" +) + +const ( + DocAccessActionView = "view" + DocAccessActionEdit = "edit" + DocAccessActionSubmit = "submit" + DocAccessActionReview = "review" + DocAccessActionApprove = "approve" + DocAccessActionRun = "run" + + AccessCodeDocTaskDisabled = "DOC_TASK_DISABLED" + AccessCodeNotCurrentStage = "DOC_TASK_NOT_IN_ACTIVE_STAGE" + AccessCodeForbiddenDuty = "DOC_DUTY_FORBIDDEN" + AccessCodeInstanceClosed = "WORKFLOW_INSTANCE_CLOSED" +) + +// DocAccessDecision 统一文档任务访问判定结果。 +type DocAccessDecision struct { + Allowed bool `json:"allowed"` + Code string `json:"code,omitempty"` + Message string `json:"message,omitempty"` +} + +// CheckDocTaskAccess 对 live 治理实例下的文档任务做 fail-closed 判定。 +// 无 live 实例时不限制（legacy 兼容）。 +func (s *Service) CheckDocTaskAccess( + ctx context.Context, + projectID int64, + userID int64, + taskKey string, + action string, + isCreator bool, +) DocAccessDecision { + allow := DocAccessDecision{Allowed: true} + if s == nil || s.store == nil || projectID <= 0 || userID <= 0 || taskKey == "" { + return allow + } + + live, err := s.store.GetLiveInstance(ctx, projectID) + if err != nil || live == nil { + return allow + } + if live.Status == InstanceCompleted || live.Status == "terminated" { + return DocAccessDecision{ + Allowed: false, + Code: AccessCodeInstanceClosed, + Message: "工作流实例已结束", + } + } + + detail, err := s.store.GetInstanceByID(ctx, live.ID) + if err != nil || detail == nil { + return DocAccessDecision{Allowed: false, Code: AccessCodeInstanceClosed, Message: "工作流实例不可用"} + } + + var docInst *DocTaskInstance + var stageKey string + for _, st := range detail.Stages { + for i := range st.DocTasks { + if st.DocTasks[i].TaskKey == taskKey { + copyDoc := st.DocTasks[i] + docInst = ©Doc + stageKey = st.StageKey + break + } + } + if docInst != nil { + break + } + } + if docInst == nil { + return allow + } + + if docInst.Status == "skipped_disabled" { + return DocAccessDecision{ + Allowed: false, + Code: AccessCodeDocTaskDisabled, + Message: "该文档已被创建者禁用", + } + } + + if stageKey != "" && stageKey != detail.ActiveStageKey { + if action != DocAccessActionView { + return DocAccessDecision{ + Allowed: false, + Code: AccessCodeNotCurrentStage, + Message: "文档不在当前活动阶段", + } + } + } + + if action == DocAccessActionView { + return allow + } + + planTask := findDocTaskPlan(live.ConfigSnapshot, taskKey) + if planTask == nil || !planTask.Enabled { + return DocAccessDecision{ + Allowed: false, + Code: AccessCodeDocTaskDisabled, + Message: "该文档已被创建者禁用", + } + } + + if isCreator { + return allow + } + if docTaskDutyForUser(*planTask, userID) { + return allow + } + return DocAccessDecision{ + Allowed: false, + Code: AccessCodeForbiddenDuty, + Message: "当前用户不是该文档职责人", + } +} + +func findDocTaskPlan(plan WorkflowPlan, taskKey string) *DocTaskPlan { + for _, st := range plan.Stages { + for i := range st.DocTasks { + if st.DocTasks[i].TaskKey == taskKey { + return &st.DocTasks[i] + } + } + } + return nil +} diff --git a/internal/projectinstance/document_access_test.go b/internal/projectinstance/document_access_test.go new file mode 100644 index 00000000..b7a259bf --- /dev/null +++ b/internal/projectinstance/document_access_test.go @@ -0,0 +1,58 @@ +package projectinstance + +import ( + "context" + "testing" + + "chat-agent-system/internal/workflowdef" +) + +func TestCheckDocTaskAccess_DisabledDocumentRejected(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + _ = SeedProjectMeta(store, 400, LifecyclePlanning, "", 1, 0) + resolver := disabledDocResolver() + svc := NewService(store, resolver, nil, nil) + + version := resolver.versions[workflowdef.OfficialKeyMain] + plan := buildDefaultPlan(version) + for i := range plan.Stages[0].DocTasks { + if plan.Stages[0].DocTasks[i].TaskKey == "research" { + plan.Stages[0].DocTasks[i].Enabled = false + continue + } + plan.Stages[0].DocTasks[i].Duties = DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3} + } + _, _ = svc.SaveWorkflowPlan(ctx, 400, 1, true, SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1}) + preview, _ := svc.ActivationPreview(ctx, 400, 1, true) + _, _ = svc.Activate(ctx, 400, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + + decision := svc.CheckDocTaskAccess(ctx, 400, 1, "research", DocAccessActionRun, true) + if decision.Allowed || decision.Code != AccessCodeDocTaskDisabled { + t.Fatalf("expected DOC_TASK_DISABLED, got %+v", decision) + } +} + +func TestCheckDocTaskAccess_NonDutyUserRejected(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + _ = SeedProjectMeta(store, 401, LifecyclePlanning, "", 1, 0) + svc := NewService(store, TestResolver(), nil, nil) + + _, _ = svc.SaveWorkflowPlan(ctx, 401, 1, true, SavePlanInput{ + Plan: ValidTestPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + preview, _ := svc.ActivationPreview(ctx, 401, 1, true) + _, _ = svc.Activate(ctx, 401, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + + decision := svc.CheckDocTaskAccess(ctx, 401, 99, "intake", DocAccessActionRun, false) + if decision.Allowed || decision.Code != AccessCodeForbiddenDuty { + t.Fatalf("expected DOC_DUTY_FORBIDDEN, got %+v", decision) + } +} diff --git a/internal/projectinstance/service.go b/internal/projectinstance/service.go index 2fa513be..9d14ad6a 100644 --- a/internal/projectinstance/service.go +++ b/internal/projectinstance/service.go @@ -253,18 +253,6 @@ func (s *Service) Activate(ctx context.Context, projectID int64, userID int64, i return nil, fmt.Errorf("%w: no saved plan", ErrInvalidPlan) } - consumed, err := s.store.ConsumeActivationToken(ctx, in.ConfirmationToken, projectID, planInst.PlanHash) - if err != nil { - return nil, err - } - currentHash, err := computePlanHash(*consumed) - if err != nil { - return nil, err - } - if currentHash != planInst.PlanHash { - return nil, ErrPlanChangedReconfirm - } - version, err := s.resolvePublishedVersion(ctx, planInst.ConfigSnapshot) if err != nil { return nil, err @@ -274,7 +262,7 @@ func (s *Service) Activate(ctx context.Context, projectID int64, userID int64, i return nil, ErrValidationFailed } - activated, err := s.store.ActivateInstance(ctx, projectID, planInst, version, selVer, userID) + activated, err := s.store.ActivateInstanceWithToken(ctx, in.ConfirmationToken, projectID, planInst, version, selVer, userID, planInst.PlanHash) if err != nil { return nil, err } diff --git a/internal/projectinstance/service_test.go b/internal/projectinstance/service_test.go index 75e26a7e..c7ba5170 100644 --- a/internal/projectinstance/service_test.go +++ b/internal/projectinstance/service_test.go @@ -8,76 +8,6 @@ import ( "chat-agent-system/internal/workflowdef" ) -type stubVersionResolver struct { - versions map[string]*workflowdef.Version - defs map[string]*workflowdef.Definition -} - -func (s *stubVersionResolver) GetDefinitionByKey(_ context.Context, key string) (*workflowdef.Definition, error) { - d, ok := s.defs[key] - if !ok { - return nil, workflowdef.ErrNotFound - } - return d, nil -} - -func (s *stubVersionResolver) GetVersion(_ context.Context, workflowID int64, version int) (*workflowdef.Version, error) { - for _, v := range s.versions { - if v.WorkflowID == workflowID && v.Version == version { - copyV := *v - return ©V, nil - } - } - return nil, workflowdef.ErrNotFound -} - -func (s *stubVersionResolver) ListDefinitions(_ context.Context) ([]workflowdef.Definition, error) { - out := make([]workflowdef.Definition, 0, len(s.defs)) - for _, d := range s.defs { - out = append(out, *d) - } - return out, nil -} - -func testResolver() *stubVersionResolver { - key := workflowdef.OfficialKeyMain - ver := &workflowdef.Version{ - ID: 100, WorkflowID: 10, Version: 1, Status: workflowdef.StatusPublished, - WorkflowKey: key, WorkflowName: workflowdef.OfficialWorkflowName(key), - Definition: workflowdef.VersionDefinition{ - SchemaVersion: 2, - WorkflowKey: key, - Stages: []workflowdef.StageDef{{ - Key: "V0.1_triage", Name: "V0.1 需求摸底", Optional: false, Position: 10, - DocTasks: []workflowdef.DocTaskDef{{ - Key: "intake", Title: "用户初始需求收集表", CreatorConfigurable: true, BlockingWhenEnabled: true, DefaultEnabled: true, - OutputDocType: "intake", OutputDocTitle: "用户初始需求收集表", - }}, - }}, - }, - } - return &stubVersionResolver{ - defs: map[string]*workflowdef.Definition{ - key: {ID: 10, Key: key, Name: workflowdef.OfficialWorkflowName(key), LatestPublishedVersion: 1}, - }, - versions: map[string]*workflowdef.Version{key: ver}, - } -} - -func validPlan() WorkflowPlan { - return WorkflowPlan{ - WorkflowDefinitionKey: workflowdef.OfficialKeyMain, - WorkflowVersion: 1, - Stages: []StagePlan{{ - StageKey: "V0.1_triage", Enabled: true, - DocTasks: []DocTaskPlan{{ - TaskKey: "intake", Enabled: true, - Duties: DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3}, - }}, - }}, - } -} - func TestService_PlanPreviewActivate(t *testing.T) { ctx := context.Background() store := newMemStore() diff --git a/internal/projectinstance/stage_remediation_test.go b/internal/projectinstance/stage_remediation_test.go new file mode 100644 index 00000000..c1a5d53b --- /dev/null +++ b/internal/projectinstance/stage_remediation_test.go @@ -0,0 +1,209 @@ +package projectinstance + +import ( + "context" + "testing" + + "chat-agent-system/internal/workflowdef" +) + +func TestAdvanceStage_CompletesProjectAfterLastStage(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 100, LifecyclePlanning, "", 1, 0) + svc := NewService(store, testResolver(), nil, &stubDocChecker{states: map[string]string{"intake": "approved"}}) + + _, err := svc.SaveWorkflowPlan(ctx, 100, 1, true, SavePlanInput{ + Plan: validPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + if err != nil { + t.Fatalf("SaveWorkflowPlan: %v", err) + } + preview, err := svc.ActivationPreview(ctx, 100, 1, true) + if err != nil { + t.Fatalf("ActivationPreview: %v", err) + } + result, err := svc.Activate(ctx, 100, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, + ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + if err != nil { + t.Fatalf("Activate: %v", err) + } + + advanced, err := svc.AdvanceStage(ctx, result.Instance.ID, 1, true, AdvanceStageInput{ + StageKey: "V0.1_triage", ExpectedConfigRevision: result.Instance.ConfigRevision, + }) + if err != nil { + t.Fatalf("AdvanceStage: %v", err) + } + if !advanced.InstanceDone { + t.Fatal("expected instance done after last stage") + } + if advanced.Instance.Status != InstanceCompleted { + t.Fatalf("expected completed instance, got %q", advanced.Instance.Status) + } + + lifecycle, activeKey, _, _, err := store.GetProjectMeta(ctx, 100) + if err != nil { + t.Fatalf("GetProjectMeta: %v", err) + } + if lifecycle != LifecyclePlanning { + t.Fatalf("expected lifecycle planning, got %q", lifecycle) + } + if activeKey != "" { + t.Fatalf("expected empty active workflow key, got %q", activeKey) + } +} + +func TestAdvanceStage_DoesNotChangeProjectOnNonLastStage(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 101, LifecyclePlanning, "", 1, 0) + resolver := testResolverMultiStage() + svc := NewService(store, resolver, nil, &stubDocChecker{states: map[string]string{"intake": "approved"}}) + + plan := validPlan() + plan.Stages = append(plan.Stages, StagePlan{ + StageKey: "V0.5_feasibility", Enabled: true, + DocTasks: []DocTaskPlan{{ + TaskKey: "feasibility_reg", Enabled: true, + Duties: DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3}, + }}, + }) + _, err := svc.SaveWorkflowPlan(ctx, 101, 1, true, SavePlanInput{ + Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + if err != nil { + t.Fatalf("SaveWorkflowPlan: %v", err) + } + preview, _ := svc.ActivationPreview(ctx, 101, 1, true) + result, _ := svc.Activate(ctx, 101, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + + _, err = svc.AdvanceStage(ctx, result.Instance.ID, 1, true, AdvanceStageInput{ + StageKey: "V0.1_triage", ExpectedConfigRevision: result.Instance.ConfigRevision, + }) + if err != nil { + t.Fatalf("AdvanceStage: %v", err) + } + + lifecycle, _, _, _, err := store.GetProjectMeta(ctx, 101) + if err != nil { + t.Fatalf("GetProjectMeta: %v", err) + } + if lifecycle != LifecycleActive { + t.Fatalf("expected lifecycle active on non-last stage, got %q", lifecycle) + } +} + +func TestAdvanceStage_DoesNotModifyDocumentApprovalStatus(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 102, LifecyclePlanning, "", 1, 0) + svc := NewService(store, testResolver(), nil, &stubDocChecker{states: map[string]string{"intake": "approved"}}) + + _, _ = svc.SaveWorkflowPlan(ctx, 102, 1, true, SavePlanInput{ + Plan: validPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + preview, _ := svc.ActivationPreview(ctx, 102, 1, true) + result, _ := svc.Activate(ctx, 102, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + + before, err := store.GetInstanceByID(ctx, result.Instance.ID) + if err != nil { + t.Fatalf("GetInstanceByID: %v", err) + } + beforeStatus := before.Stages[0].DocTasks[0].Status + + _, err = svc.AdvanceStage(ctx, result.Instance.ID, 1, true, AdvanceStageInput{ + StageKey: "V0.1_triage", ExpectedConfigRevision: result.Instance.ConfigRevision, + }) + if err != nil { + t.Fatalf("AdvanceStage: %v", err) + } + + after, err := store.GetInstanceByID(ctx, result.Instance.ID) + if err != nil { + t.Fatalf("GetInstanceByID after: %v", err) + } + if after.Stages[0].DocTasks[0].Status != beforeStatus { + t.Fatalf("doc status changed from %q to %q after advance", beforeStatus, after.Stages[0].DocTasks[0].Status) + } +} + +func TestDisabledDocumentPersistence(t *testing.T) { + ctx := context.Background() + store := newMemStore() + _ = SeedProjectMeta(store, 103, LifecyclePlanning, "", 1, 0) + resolver := disabledDocResolver() + svc := NewService(store, resolver, nil, nil) + + version := resolver.versions[workflowdef.OfficialKeyMain] + plan := buildDefaultPlan(version) + for i := range plan.Stages[0].DocTasks { + if plan.Stages[0].DocTasks[i].TaskKey == "research" { + plan.Stages[0].DocTasks[i].Enabled = false + continue + } + plan.Stages[0].DocTasks[i].Duties = DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3} + } + _, err := svc.SaveWorkflowPlan(ctx, 103, 1, true, SavePlanInput{ + Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + if err != nil { + t.Fatalf("SaveWorkflowPlan: %v", err) + } + preview, _ := svc.ActivationPreview(ctx, 103, 1, true) + result, err := svc.Activate(ctx, 103, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + if err != nil { + t.Fatalf("Activate: %v", err) + } + + detail, err := store.GetInstanceByID(ctx, result.Instance.ID) + if err != nil { + t.Fatalf("GetInstanceByID: %v", err) + } + foundDisabled := false + for _, st := range detail.Stages { + for _, dt := range st.DocTasks { + if dt.TaskKey == "research" { + if dt.Status != "skipped_disabled" { + t.Fatalf("expected skipped_disabled, got %q", dt.Status) + } + foundDisabled = true + } + } + } + if !foundDisabled { + t.Fatal("disabled doc task instance not persisted") + } + + decision := svc.CheckDocTaskAccess(ctx, 103, 1, "research", DocAccessActionRun, true) + if decision.Allowed || decision.Code != AccessCodeDocTaskDisabled { + t.Fatalf("expected DOC_TASK_DISABLED, got %+v", decision) + } +} + +func testResolverMultiStage() *stubVersionResolver { + r := testResolver() + r.versions[workflowdef.OfficialKeyMain].Definition.Stages = append( + r.versions[workflowdef.OfficialKeyMain].Definition.Stages, + workflowdef.StageDef{ + Key: "V0.5_feasibility", Name: "V0.5 可行性", Optional: false, Position: 20, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "feasibility_reg", Title: "法规路径评估", CreatorConfigurable: true, BlockingWhenEnabled: true, DefaultEnabled: true, + OutputDocType: "feasibility", OutputDocTitle: "法规路径评估", + }}, + }, + ) + return r +} diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go index 942ce7c4..c9fa246a 100644 --- a/internal/projectinstance/store.go +++ b/internal/projectinstance/store.go @@ -42,6 +42,8 @@ type Store interface { HasLiveInstance(ctx context.Context, projectID int64) (bool, error) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) + ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) + CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error) TerminateInstance(ctx context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error) SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time) error @@ -292,16 +294,30 @@ func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst * } func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) { + return s.activateInstanceTx(ctx, s.pool, "", projectID, inst, version, expectedVersion, activatedBy, "") +} + +func (s *pgStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) { + return s.activateInstanceTx(ctx, s.pool, token, projectID, inst, version, expectedVersion, activatedBy, planHash) +} + +func (s *pgStore) activateInstanceTx(ctx context.Context, db pgxDB, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) { workflowKey := "" if version != nil { workflowKey = version.WorkflowKey } - tx, err := s.pool.Begin(ctx) + tx, err := db.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) + if token != "" { + if err := consumeActivationTokenTx(ctx, tx, token, projectID, planHash); err != nil { + return nil, err + } + } + tag, err := tx.Exec(ctx, ` UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2, workflow_selection_version = workflow_selection_version + 1, @@ -359,6 +375,55 @@ func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *W return activated, nil } +type pgxDB interface { + Begin(ctx context.Context) (pgx.Tx, error) +} + +func consumeActivationTokenTx(ctx context.Context, tx pgx.Tx, token string, projectID int64, planHash string) error { + tokenHash := hashActivationToken(token) + var storedHash string + var usedAt *time.Time + var expiresAt time.Time + err := tx.QueryRow(ctx, ` + SELECT plan_hash, used_at, expires_at + FROM workflow_activation_tokens WHERE token = $1 AND project_id = $2 + `, tokenHash, projectID).Scan(&storedHash, &usedAt, &expiresAt) + if errors.Is(err, pgx.ErrNoRows) { + return ErrTokenInvalid + } + if err != nil { + return err + } + if usedAt != nil || time.Now().After(expiresAt) || storedHash != planHash { + return ErrTokenInvalid + } + tag, err := tx.Exec(ctx, ` + UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL + `, tokenHash) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrTokenInvalid + } + return nil +} + +func (s *pgStore) CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error) { + if projectID <= 0 { + return 0, nil + } + var count int + err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM task_runs + WHERE project_id = $1 AND status IN ('pending', 'running') + `, projectID).Scan(&count) + if err != nil { + return 0, nil + } + return count, nil +} + func (s *pgStore) applyRuntimeMetadata(ctx context.Context, tx pgx.Tx, instanceID int64, def workflowdef.VersionDefinition) error { taskDefs := map[string]workflowdef.DocTaskDef{} for _, st := range def.Stages { @@ -502,6 +567,11 @@ func (s *pgStore) TerminateInstance(ctx context.Context, instanceID int64, proje return nil, err } + _, _ = tx.Exec(ctx, ` + UPDATE task_runs SET status = 'cancelled', updated_at = $2, error_message = 'workflow terminated' + WHERE project_id = $1 AND status IN ('pending', 'running') + `, projectID, now) + if err := tx.Commit(ctx); err != nil { return nil, err } @@ -587,7 +657,7 @@ func (m *memStore) rebuildStagesLocked(inst *WorkflowInstance, activate bool) { task := memDocTaskRecord{id: m.nextTaskID, taskKey: dt.TaskKey, enabled: dt.Enabled, status: "pending"} m.nextTaskID++ if !dt.Enabled { - task.status = "skipped" + task.status = "skipped_disabled" } rec.tasks = append(rec.tasks, task) } @@ -706,6 +776,17 @@ func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInsta return ©Inst, nil } +func (m *memStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) { + if _, err := m.ConsumeActivationToken(ctx, token, projectID, planHash); err != nil { + return nil, err + } + return m.ActivateInstance(ctx, projectID, inst, version, expectedVersion, activatedBy) +} + +func (m *memStore) CountActiveTaskRuns(_ context.Context, _ int64) (int, error) { + return 0, nil +} + func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) { workflowKey := "" if version != nil { diff --git a/internal/projectinstance/terminate.go b/internal/projectinstance/terminate.go index fffcaeb0..80db0d18 100644 --- a/internal/projectinstance/terminate.go +++ b/internal/projectinstance/terminate.go @@ -1,6 +1,9 @@ package projectinstance -import "context" +import ( + "context" + "fmt" +) // TerminateInput 受控终止输入。 type TerminateInput struct { @@ -15,6 +18,73 @@ type TerminateResult struct { Reason string `json:"reason"` } +// PendingDocSummary 终止预览中的待处理文档摘要。 +type PendingDocSummary struct { + TaskKey string `json:"task_key"` + Status string `json:"status"` + StageKey string `json:"stage_key,omitempty"` +} + +// TerminationPreview 受控终止影响预览。 +type TerminationPreview struct { + InstanceID int64 `json:"instance_id"` + PendingDocuments []PendingDocSummary `json:"pending_documents"` + InProgressTaskRuns int `json:"in_progress_task_runs"` + CanTerminate bool `json:"can_terminate"` + Impact string `json:"impact,omitempty"` +} + +func (s *Service) GetTerminationPreview(ctx context.Context, instanceID int64, projectID int64, userID int64, isCreator bool) (*TerminationPreview, error) { + if !isCreator { + return nil, ErrForbiddenNotCreator + } + if !s.Available() { + return nil, ErrNotAvailable + } + live, err := s.store.GetLiveInstance(ctx, projectID) + if err != nil || live == nil || live.ID != instanceID { + return nil, ErrInstanceNotFound + } + if live.Status == InstanceCompleted || live.Status == "terminated" { + return nil, ErrTerminationNotAllowed + } + + detail, err := s.store.GetInstanceByID(ctx, instanceID) + if err != nil { + return nil, err + } + + pending := make([]PendingDocSummary, 0) + for _, st := range detail.Stages { + if st.StageKey != detail.ActiveStageKey { + continue + } + for _, dt := range st.DocTasks { + if !dt.Enabled || dt.Status == "skipped_disabled" { + continue + } + if dt.Status != "approved" { + pending = append(pending, PendingDocSummary{ + TaskKey: dt.TaskKey, Status: dt.Status, StageKey: st.StageKey, + }) + } + } + } + + taskRuns, _ := s.store.CountActiveTaskRuns(ctx, projectID) + preview := &TerminationPreview{ + InstanceID: instanceID, + PendingDocuments: pending, + InProgressTaskRuns: taskRuns, + CanTerminate: true, + Impact: fmt.Sprintf( + "%d pending documents, %d in-progress task runs will be cancelled", + len(pending), taskRuns, + ), + } + return preview, nil +} + func (s *Service) TerminateInstance(ctx context.Context, instanceID int64, projectID int64, userID int64, isCreator bool, in TerminateInput) (*TerminateResult, error) { if !isCreator { return nil, ErrForbiddenNotCreator diff --git a/internal/projectinstance/terminate_test.go b/internal/projectinstance/terminate_test.go new file mode 100644 index 00000000..0b68f4fd --- /dev/null +++ b/internal/projectinstance/terminate_test.go @@ -0,0 +1,48 @@ +package projectinstance + +import ( + "context" + "testing" +) + +func TestTerminateInstance_UpdatesProjectLifecycle(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + _ = SeedProjectMeta(store, 300, LifecyclePlanning, "", 1, 0) + svc := NewService(store, TestResolver(), nil, &StubDocChecker{States: map[string]string{"intake": "approved"}}) + + _, _ = svc.SaveWorkflowPlan(ctx, 300, 1, true, SavePlanInput{ + Plan: ValidTestPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1, + }) + preview, _ := svc.ActivationPreview(ctx, 300, 1, true) + activated, _ := svc.Activate(ctx, 300, 1, true, ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + + pv, err := svc.GetTerminationPreview(ctx, activated.Instance.ID, 300, 1, true) + if err != nil { + t.Fatalf("GetTerminationPreview: %v", err) + } + if !pv.CanTerminate { + t.Fatal("expected can terminate") + } + + res, err := svc.TerminateInstance(ctx, activated.Instance.ID, 300, 1, true, TerminateInput{ + Reason: "测试终止", Confirm: true, ExpectedConfigRevision: activated.Instance.ConfigRevision, + }) + if err != nil { + t.Fatalf("TerminateInstance: %v", err) + } + if res.Instance.Status != "terminated" { + t.Fatalf("expected terminated, got %q", res.Instance.Status) + } + + lifecycle, activeKey, _, _, err := store.GetProjectMeta(ctx, 300) + if err != nil { + t.Fatalf("GetProjectMeta: %v", err) + } + if lifecycle != LifecyclePlanning || activeKey != "" { + t.Fatalf("expected planning with empty key, got lifecycle=%q key=%q", lifecycle, activeKey) + } +} diff --git a/internal/projectinstance/testutil.go b/internal/projectinstance/testutil.go new file mode 100644 index 00000000..9e6519d8 --- /dev/null +++ b/internal/projectinstance/testutil.go @@ -0,0 +1,116 @@ +package projectinstance + +import ( + "context" + + "chat-agent-system/internal/workflowdef" +) + +// NewMemoryStore 返回内存 Store，供跨包集成测试使用。 +func NewMemoryStore() Store { + return newMemStore() +} + +// TestResolver 返回最小正式商务工作流解析器。 +func TestResolver() VersionResolver { + return testResolver() +} + +// ValidTestPlan 返回单阶段 intake 测试规划。 +func ValidTestPlan() WorkflowPlan { + return validPlan() +} + +// StubDocChecker 内存文档批准状态探针。 +type StubDocChecker struct { + States map[string]string +} + +func (s *StubDocChecker) TaskApprovalStates(_ context.Context, _ int64) (map[string]string, error) { + if s == nil { + return map[string]string{}, nil + } + return s.States, nil +} + +type stubVersionResolver struct { + versions map[string]*workflowdef.Version + defs map[string]*workflowdef.Definition +} + +func (s *stubVersionResolver) GetDefinitionByKey(_ context.Context, key string) (*workflowdef.Definition, error) { + d, ok := s.defs[key] + if !ok { + return nil, workflowdef.ErrNotFound + } + return d, nil +} + +func (s *stubVersionResolver) GetVersion(_ context.Context, workflowID int64, version int) (*workflowdef.Version, error) { + for _, v := range s.versions { + if v.WorkflowID == workflowID && v.Version == version { + copyV := *v + return ©V, nil + } + } + return nil, workflowdef.ErrNotFound +} + +func (s *stubVersionResolver) ListDefinitions(_ context.Context) ([]workflowdef.Definition, error) { + out := make([]workflowdef.Definition, 0, len(s.defs)) + for _, d := range s.defs { + out = append(out, *d) + } + return out, nil +} + +func disabledDocResolver() *stubVersionResolver { + r := testResolver() + r.versions[workflowdef.OfficialKeyMain].Definition.Stages[0].DocTasks = append( + r.versions[workflowdef.OfficialKeyMain].Definition.Stages[0].DocTasks, + workflowdef.DocTaskDef{ + Key: "research", Title: "市场调研", CreatorConfigurable: true, BlockingWhenEnabled: false, DefaultEnabled: false, + OutputDocType: "research", OutputDocTitle: "市场调研", + }, + ) + return r +} + +func testResolver() *stubVersionResolver { + key := workflowdef.OfficialKeyMain + ver := &workflowdef.Version{ + ID: 100, WorkflowID: 10, Version: 1, Status: workflowdef.StatusPublished, + WorkflowKey: key, WorkflowName: workflowdef.OfficialWorkflowName(key), + Definition: workflowdef.VersionDefinition{ + SchemaVersion: 2, + WorkflowKey: key, + Stages: []workflowdef.StageDef{{ + Key: "V0.1_triage", Name: "V0.1 需求摸底", Optional: false, Position: 10, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "intake", Title: "用户初始需求收集表", CreatorConfigurable: true, BlockingWhenEnabled: true, DefaultEnabled: true, + OutputDocType: "intake", OutputDocTitle: "用户初始需求收集表", + }}, + }}, + }, + } + return &stubVersionResolver{ + defs: map[string]*workflowdef.Definition{ + key: {ID: 10, Key: key, Name: workflowdef.OfficialWorkflowName(key), LatestPublishedVersion: 1}, + }, + versions: map[string]*workflowdef.Version{key: ver}, + } +} + +func validPlan() WorkflowPlan { + return WorkflowPlan{ + WorkflowDefinitionKey: workflowdef.OfficialKeyMain, + WorkflowVersion: 1, + Stages: []StagePlan{{ + StageKey: "V0.1_triage", Enabled: true, + DocTasks: []DocTaskPlan{{ + TaskKey: "intake", Enabled: true, + Duties: DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3}, + }}, + }}, + } +} diff --git a/internal/server/e2e_workflow_governance_test.go b/internal/server/e2e_workflow_governance_test.go new file mode 100644 index 00000000..115100ef --- /dev/null +++ b/internal/server/e2e_workflow_governance_test.go @@ -0,0 +1,163 @@ +package server_test + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "chat-agent-system/internal/db" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/workflowdef" +) + +// TestE2E_WorkflowGovernance_CompleteAndReplan 验证末阶段完成后项目回到 planning 并可再次获取目录。 +func TestE2E_WorkflowGovernance_CompleteAndReplan(t *testing.T) { + dsn := db.SkipIfNoTestDatabaseURL(t) + _ = dsn + pool := openPGPoolWithMigrations(t) + if pool == nil { + return + } + defer pool.Close() + + ctx := context.Background() + handler, _ := setupPGGovernanceServer(t, pool) + + creatorToken := loginTestUser(t, handler, "zhanggong", "123456") + creatorID := fetchAuthUserID(t, handler, creatorToken) + + wfSvc := workflowdef.NewService(pool, nil) + wfKey := "e2e_gov_" + time.Now().Format("150405.000") + _, ver, err := wfSvc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{ + Key: wfKey, Name: "E2E Governance WF", + Definition: workflowdef.VersionDefinition{ + SchemaVersion: 1, + Stages: []workflowdef.StageDef{{ + Key: "only_stage", Name: "唯一阶段", Position: 0, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "intake", Required: true, Blocking: true, DefaultEnabled: true, + OutputDocType: "intake", OutputDocTitle: "Intake", + }}, + }}, + }, + }, 1) + if err != nil { + t.Fatalf("create workflow: %v", err) + } + if _, err := wfSvc.PublishVersion(ctx, ver.WorkflowID, ver.Version, 1); err != nil { + t.Fatalf("publish workflow: %v", err) + } + + var projectID int64 + code := "e2e-gov-" + time.Now().Format("150405.000") + err = pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) + VALUES ($1, 'E2E Governance', 'active', $2, 'planning', '', 1, 0) RETURNING id + `, code, creatorID).Scan(&projectID) + if err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + }) + + reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=1", projectID, wfKey), nil) + withBearerAuth(reqDefault, creatorToken) + recDefault := httptest.NewRecorder() + handler.ServeHTTP(recDefault, reqDefault) + if recDefault.Code != http.StatusOK { + t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String()) + } + var plan projectinstance.WorkflowPlan + _ = json.Unmarshal(recDefault.Body.Bytes(), &plan) + reviewerID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "lizhuguan", "123456")) + approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456")) + for i := range plan.Stages { + for j := range plan.Stages[i].DocTasks { + plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{ + AuthorUserID: creatorID, ReviewerUserID: reviewerID, ApproverUserID: approverID, + } + } + } + saveBody, _ := json.Marshal(projectinstance.SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1}) + reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", projectID), bytes.NewReader(saveBody)) + withBearerAuth(reqSave, creatorToken) + recSave := httptest.NewRecorder() + handler.ServeHTTP(recSave, reqSave) + if recSave.Code != http.StatusOK { + t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String()) + } + + reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil) + withBearerAuth(reqPreview, creatorToken) + recPreview := httptest.NewRecorder() + handler.ServeHTTP(recPreview, reqPreview) + var preview projectinstance.ActivationPreview + _ = json.Unmarshal(recPreview.Body.Bytes(), &preview) + + activateBody, _ := json.Marshal(projectinstance.ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody)) + withBearerAuth(reqActivate, creatorToken) + recActivate := httptest.NewRecorder() + handler.ServeHTTP(recActivate, reqActivate) + if recActivate.Code != http.StatusOK { + t.Fatalf("activate: %d %s", recActivate.Code, recActivate.Body.String()) + } + var activated projectinstance.ActivateResult + _ = json.Unmarshal(recActivate.Body.Bytes(), &activated) + + // 批准并推进至完成（复用 governance 测试中的 docgov 路径） + docID := fmt.Sprintf("doc_e2e_%d", projectID) + // 文档批准链在 workflow_governance_pg_test 已覆盖；此处直接标记实例文档 approved 并推进 + _, err = pool.Exec(ctx, ` + UPDATE project_doc_task_instances dti SET status = 'approved', updated_at = now() + FROM project_stage_instances psi + WHERE dti.stage_instance_id = psi.id AND psi.workflow_instance_id = $1 AND dti.task_key = 'intake' + `, activated.Instance.ID) + if err != nil { + t.Fatalf("mark approved: %v", err) + } + _ = docID + + advanceBody, _ := json.Marshal(projectinstance.AdvanceStageInput{ + StageKey: "only_stage", ExpectedConfigRevision: activated.Instance.ConfigRevision, + }) + reqAdvance := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/workflow-instances/%d/stages/only_stage/advance", activated.Instance.ID), bytes.NewReader(advanceBody)) + withBearerAuth(reqAdvance, creatorToken) + recAdvance := httptest.NewRecorder() + handler.ServeHTTP(recAdvance, reqAdvance) + if recAdvance.Code != http.StatusOK { + t.Fatalf("advance: %d %s", recAdvance.Code, recAdvance.Body.String()) + } + + reqSettings := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/settings", projectID), nil) + withBearerAuth(reqSettings, creatorToken) + recSettings := httptest.NewRecorder() + handler.ServeHTTP(recSettings, reqSettings) + if recSettings.Code != http.StatusOK { + t.Fatalf("settings: %d %s", recSettings.Code, recSettings.Body.String()) + } + var settings projectinstance.ProjectSettings + if err := json.Unmarshal(recSettings.Body.Bytes(), &settings); err != nil { + t.Fatalf("decode settings: %v", err) + } + if settings.LifecycleStatus != projectinstance.LifecyclePlanning { + t.Fatalf("expected planning after completion, got %q", settings.LifecycleStatus) + } + + reqCatalog := httptest.NewRequest(http.MethodGet, "/api/v1/workflows/catalog", nil) + withBearerAuth(reqCatalog, creatorToken) + recCatalog := httptest.NewRecorder() + handler.ServeHTTP(recCatalog, reqCatalog) + if recCatalog.Code != http.StatusOK { + t.Fatalf("catalog: %d %s", recCatalog.Code, recCatalog.Body.String()) + } +} diff --git a/internal/server/project_settings_handlers.go b/internal/server/project_settings_handlers.go index 7d97cb95..de370805 100644 --- a/internal/server/project_settings_handlers.go +++ b/internal/server/project_settings_handlers.go @@ -5,6 +5,7 @@ import ( "errors" "net/http" "strconv" + "time" "chat-agent-system/internal/projectinstance" ) @@ -52,7 +53,29 @@ func (s *Server) handleGetProjectSettings(w http.ResponseWriter, r *http.Request writeProjectInstanceError(w, err) return } - writeJSONResponse(w, http.StatusOK, settings) + type settingsResponse struct { + *projectinstance.ProjectSettings + RecentWorkflowEvents []workflowEventSummary `json:"recent_workflow_events,omitempty"` + } + out := settingsResponse{ProjectSettings: settings} + if s.notifySvc != nil && s.notifySvc.Available() { + items, listErr := s.notifySvc.ListProjectNotifications(r.Context(), user.ID, projectID, 0, 20) + if listErr == nil { + for _, n := range items { + out.RecentWorkflowEvents = append(out.RecentWorkflowEvents, workflowEventSummary{ + ID: n.ID, Type: n.Type, Title: n.Title, CreatedAt: n.CreatedAt.Format(time.RFC3339), + }) + } + } + } + writeJSONResponse(w, http.StatusOK, out) +} + +type workflowEventSummary struct { + ID int64 `json:"id"` + Type string `json:"type"` + Title string `json:"title"` + CreatedAt string `json:"created_at"` } func (s *Server) handleSaveWorkflowPlan(w http.ResponseWriter, r *http.Request) { diff --git a/internal/server/server.go b/internal/server/server.go index 5f5a258f..32fcff49 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -165,6 +165,7 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("GET /api/v1/workflow-instances/{id}", s.handleGetWorkflowInstance) mux.HandleFunc("GET /api/v1/workflow-instances/{id}/stages/{stage}/readiness", s.handleGetStageReadiness) mux.HandleFunc("POST /api/v1/workflow-instances/{id}/stages/{stage}/advance", s.handleAdvanceStage) + mux.HandleFunc("GET /api/v1/workflow-instances/{id}/termination-preview", s.handleTerminationPreview) mux.HandleFunc("POST /api/v1/workflow-instances/{id}/terminate", s.handleTerminateWorkflowInstance) mux.HandleFunc("GET /api/v1/workflows", s.handleListWorkflowDefinitions) mux.HandleFunc("GET /api/v1/workflows/catalog", s.handleWorkflowCatalog) diff --git a/internal/server/stage_access_helpers.go b/internal/server/stage_access_helpers.go index c0b1103a..5cf45916 100644 --- a/internal/server/stage_access_helpers.go +++ b/internal/server/stage_access_helpers.go @@ -20,18 +20,56 @@ func (s *Server) resolveStageAccess(ctx context.Context, projectID, userID int64 return s.instSvc.ResolveStageAccess(ctx, projectID, userID, isCreator) } -func (s *Server) allowWorkflowDocTaskAccess(ctx context.Context, projectID, userID int64, taskKey string) bool { +func (s *Server) checkWorkflowDocTaskAccess(ctx context.Context, projectID, userID int64, taskKey string) projectinstance.DocAccessDecision { if taskKey == "" { - return true + return projectinstance.DocAccessDecision{Allowed: true} + } + if s.instSvc != nil && s.instSvc.Available() && projectID > 0 && userID > 0 { + live, liveErr := s.instSvc.HasLiveInstance(ctx, projectID) + if liveErr == nil && live { + isCreator, _ := s.pwSvc.IsProjectCreator(ctx, projectID, userID) + return s.instSvc.CheckDocTaskAccess(ctx, projectID, userID, taskKey, projectinstance.DocAccessActionRun, isCreator) + } } access, err := s.resolveStageAccess(ctx, projectID, userID) if err != nil { - return !s.failClosedAuthz() + if s.failClosedAuthz() { + return projectinstance.DocAccessDecision{Allowed: false, Code: "WORKFLOW_ACCESS_DENIED", Message: "无权访问该阶段内容"} + } + return projectinstance.DocAccessDecision{Allowed: true} } - if access == nil || !access.Restricted { + if access == nil || !access.Restricted || access.TaskKeyAllowed(taskKey) { + return projectinstance.DocAccessDecision{Allowed: true} + } + return projectinstance.DocAccessDecision{Allowed: false, Code: "WORKFLOW_ACCESS_DENIED", Message: "无权访问该阶段内容"} +} + +func (s *Server) allowWorkflowDocTaskAccess(ctx context.Context, projectID, userID int64, taskKey string) bool { + return s.checkWorkflowDocTaskAccess(ctx, projectID, userID, taskKey).Allowed +} + +func (s *Server) enforceWorkflowDocTaskAccess(w http.ResponseWriter, ctx context.Context, projectID, userID int64, taskKey string) bool { + decision := s.checkWorkflowDocTaskAccess(ctx, projectID, userID, taskKey) + if decision.Allowed { return true } - return access.TaskKeyAllowed(taskKey) + s.writeDocAccessDecision(w, decision) + return false +} + +func (s *Server) writeDocAccessDecision(w http.ResponseWriter, decision projectinstance.DocAccessDecision) { + status := http.StatusForbidden + if decision.Code == projectinstance.AccessCodeDocTaskDisabled { + status = http.StatusConflict + } + msg := decision.Message + if msg == "" { + msg = "无权访问该文档任务" + } + payload, _ := json.Marshal(map[string]string{"error": decision.Code, "message": msg}) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write(payload) } func (s *Server) allowAnyStageAccess(ctx context.Context, projectID, userID int64) bool { diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 49282160..e4c5ad1d 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -68,8 +68,7 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { if req.TaskKey == "" && req.Stage == "" { req.TaskKey = "intake" } - if req.ProjectID > 0 && req.TaskKey != "" && !s.allowWorkflowDocTaskAccess(r.Context(), req.ProjectID, user.ID, req.TaskKey) { - s.writeWorkflowAccessDenied(w) + if req.ProjectID > 0 && req.TaskKey != "" && !s.enforceWorkflowDocTaskAccess(w, r.Context(), req.ProjectID, user.ID, req.TaskKey) { return } req.WorkDir = "" diff --git a/internal/server/workflow_terminate_handlers.go b/internal/server/workflow_terminate_handlers.go index 5cd592d1..e03bba1c 100644 --- a/internal/server/workflow_terminate_handlers.go +++ b/internal/server/workflow_terminate_handlers.go @@ -9,6 +9,35 @@ import ( "chat-agent-system/internal/projectinstance" ) +func (s *Server) handleTerminationPreview(w http.ResponseWriter, r *http.Request) { + if s.instSvc == nil || !s.instSvc.Available() { + http.Error(w, `{"error":"ERR_NOT_AVAILABLE"}`, http.StatusServiceUnavailable) + return + } + instanceID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || instanceID <= 0 { + http.Error(w, `{"error":"ERR_INVALID_PARAM"}`, http.StatusBadRequest) + return + } + user, err := s.authenticateUser(r) + if err != nil || user == nil { + http.Error(w, `{"error":"ERR_UNAUTHENTICATED"}`, http.StatusUnauthorized) + return + } + detail, err := s.instSvc.GetInstanceDetail(r.Context(), instanceID) + if err != nil { + http.Error(w, `{"error":"ERR_NOT_FOUND"}`, http.StatusNotFound) + return + } + isCreator, _ := s.pwSvc.IsProjectCreator(r.Context(), detail.Instance.ProjectID, user.ID) + preview, err := s.instSvc.GetTerminationPreview(r.Context(), instanceID, detail.Instance.ProjectID, user.ID, isCreator) + if err != nil { + writeProjectInstanceError(w, err) + return + } + writeJSONResponse(w, http.StatusOK, preview) +} + func (s *Server) handleTerminateWorkflowInstance(w http.ResponseWriter, r *http.Request) { if s.instSvc == nil || !s.instSvc.Available() { http.Error(w, `{"error":"ERR_NOT_AVAILABLE"}`, http.StatusServiceUnavailable) diff --git a/scripts/fix-completed-instance-lifecycle.sql b/scripts/fix-completed-instance-lifecycle.sql new file mode 100644 index 00000000..0d521952 --- /dev/null +++ b/scripts/fix-completed-instance-lifecycle.sql @@ -0,0 +1,18 @@ +-- 修复「实例 completed 但项目仍为 active」的历史不一致数据。 +-- 用法： +-- psql "$DATABASE_URL" -f scripts/fix-completed-instance-lifecycle.sql +-- Dry-run（仅统计）： +-- psql "$DATABASE_URL" -c "SELECT COUNT(*) FROM projects p JOIN project_workflow_instances i ON i.project_id = p.id AND i.status = 'completed' WHERE p.lifecycle_status <> 'planning';" + +BEGIN; + +UPDATE projects p +SET lifecycle_status = 'planning', + active_workflow_key = '', + updated_at = now() +FROM project_workflow_instances i +WHERE i.project_id = p.id + AND i.status = 'completed' + AND p.lifecycle_status <> 'planning'; + +COMMIT; diff --git a/web/src/api/projectSettings.ts b/web/src/api/projectSettings.ts index 0e3e4d30..0e7f2de9 100644 --- a/web/src/api/projectSettings.ts +++ b/web/src/api/projectSettings.ts @@ -6,6 +6,8 @@ import { StageReadiness, AdvanceStageResult, ProjectNotification, + WorkflowCatalogItem, + TerminationPreview, } from '../types/projectSettings'; import { getBearerAuthHeaders } from './authHeaders'; @@ -130,6 +132,44 @@ export async function fetchProjectNotifications( return Array.isArray(data.items) ? data.items : []; } +export async function fetchWorkflowCatalog(): Promise { + const res = await fetch(`${API_BASE}/api/v1/workflows/catalog`, { + headers: getBearerAuthHeaders(), + }); + if (!res.ok) { + return []; + } + const data = await res.json(); + return Array.isArray(data.workflows) ? data.workflows : []; +} + +export async function fetchTerminationPreview(instanceId: number): Promise { + const res = await fetch(`${API_BASE}/api/v1/workflow-instances/${instanceId}/termination-preview`, { + headers: getBearerAuthHeaders(), + }); + if (!res.ok) { + const text = await res.text(); + throw new Error(parseError(text) || `终止预览失败: ${res.status}`); + } + return res.json(); +} + +export async function terminateWorkflowInstance( + instanceId: number, + input: { reason: string; confirm: boolean; expected_config_revision: number } +): Promise<{ instance: { id: number; status: string }; reason: string }> { + const res = await fetch(`${API_BASE}/api/v1/workflow-instances/${instanceId}/terminate`, { + method: 'POST', + headers: getBearerAuthHeaders('application/json'), + body: JSON.stringify(input), + }); + if (!res.ok) { + const text = await res.text(); + throw new Error(parseError(text) || `终止工作流失败: ${res.status}`); + } + return res.json(); +} + export async function advanceStage( instanceId: number, stageKey: string, @@ -148,7 +188,11 @@ export async function advanceStage( ); if (!res.ok) { const text = await res.text(); - throw new Error(parseError(text) || `阶段推进失败: ${res.status}`); + const msg = parseError(text) || `阶段推进失败: ${res.status}`; + if (res.status === 409 || /revision|conflict|版本/i.test(msg)) { + throw new Error(`配置版本已变化，请刷新后重试：${msg}`); + } + throw new Error(msg); } return res.json(); } diff --git a/web/src/components/project/ProjectSettingsModal.tsx b/web/src/components/project/ProjectSettingsModal.tsx index 05d65cf9..4d5eefc0 100644 --- a/web/src/components/project/ProjectSettingsModal.tsx +++ b/web/src/components/project/ProjectSettingsModal.tsx @@ -6,6 +6,8 @@ import { saveWorkflowPlan, fetchActivationPreview, activateWorkflowInstance, + fetchTerminationPreview, + terminateWorkflowInstance, } from '../../api/projectSettings'; import { ProjectSettings, @@ -45,6 +47,8 @@ export const ProjectSettingsModal: React.FC = ({ const [plan, setPlan] = useState(null); const [preview, setPreview] = useState(null); const [selectedKey, setSelectedKey] = useState(''); + const [terminateReason, setTerminateReason] = useState(''); + const [terminatePreview, setTerminatePreview] = useState<{ impact?: string } | null>(null); const loadSettings = useCallback(async () => { setLoading(true); @@ -52,6 +56,16 @@ export const ProjectSettingsModal: React.FC = ({ try { const s = await fetchProjectSettings(projectId); setSettings(s); + if (s.live_instance && s.live_instance.status !== 'completed' && s.live_instance.status !== 'terminated') { + try { + const preview = await fetchTerminationPreview(s.live_instance.id); + setTerminatePreview(preview); + } catch { + setTerminatePreview(null); + } + } else { + setTerminatePreview(null); + } if (s.plan) { setPlan(s.plan); setSelectedKey(s.plan.workflow_definition_key); @@ -151,6 +165,24 @@ export const ProjectSettingsModal: React.FC = ({ } }; + const handleTerminate = async () => { + if (!settings?.live_instance || !terminateReason.trim()) return; + setLoading(true); + setError(null); + try { + await terminateWorkflowInstance(settings.live_instance.id, { + reason: terminateReason.trim(), + confirm: true, + expected_config_revision: settings.live_instance.config_revision, + }); + onClose(); + } catch (err: unknown) { + setError(err instanceof Error ? err.message : '终止工作流失败'); + } finally { + setLoading(false); + } + }; + const handleActivate = async () => { if (!preview || !settings) return; setLoading(true); @@ -260,6 +292,57 @@ export const ProjectSettingsModal: React.FC = ({ )} + {settings?.live_instance && settings.live_instance.status !== 'completed' && ( +
+
受控终止当前工作流
+ {terminatePreview?.impact && ( +
{terminatePreview.impact}
+ )} +